Loading services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py +0 −2 Changes for services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py: 0 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -190,7 +190,6 @@ class SecurityOperations(Resource): invoker = self.__retrieve_invoker_from_db(api_invoker_id) if invoker is None: current_app.logger.warning("Invoker not found on this CCF") # TODO -> Comprobar si el invoker está registrado en algún CCF remoto. remote_ccf = self.get_remote_invoker_ccf(api_invoker_id) if remote_ccf: current_app.logger.info(f"Invoker found in remote CCFs: {remote_ccf}") Loading Loading @@ -974,7 +973,6 @@ class SecurityOperations(Resource): def get_local_api_id(self, ccf_id, remote_api_id): """Retrieve the local API ID corresponding to the remote API ID for the given certificate common name.""" # TODO: Implement the mapping logic between remote API ID and local API ID interconnected_col = self.db.get_col_by_name(self.db.interconnected) interconnected_ccf = interconnected_col.find_one({"ccf_id": ccf_id}) Loading services/nginx/policies/security-mtls.conf +1 −1 Changes for services/nginx/policies/security-mtls.conf: 1 added line, 1 removed line. Original line number Diff line number Diff line map "$endpoint:$method:$role" $security_service_mtls_policy { default DENY; # Patrones específicos por endpoint/método con múltiples roles permitidos # Specific patterns for endpoint/method with multiple allowed roles ~^security_trusted_invokers_exact:(DELETE|GET):(aef|ccf)$ ALLOW; ~^security_trusted_invokers_exact:PUT:(invoker|ccf)$ ALLOW; ~^security_update:POST:(invoker|ccf)$ ALLOW; Loading tests/libraries/helpers.py +9 −9 Changes for tests/libraries/helpers.py: 9 added lines, 9 removed lines. Original line number Diff line number Diff line Loading @@ -268,7 +268,7 @@ def get_jwt_header_info(token): dict with the header information """ try: # Decodificar sin validar firma (solo para ver el header) # Decode without verifying the signature (just to see the header) header = jwt.get_unverified_header(token) print(f"✓ Header JWT: {header}") return header Loading @@ -291,26 +291,26 @@ def validate_jwt_token_flexible(token, ca_path, algorithms=None): dict with the token data if valid, None otherwise """ try: # Obtener header para ver qué algoritmo usa # Get the header to see which algorithm is used header = jwt.get_unverified_header(token) algo_usado = header.get('alg', 'desconocido') print(f"ℹ Algorithm detected in JWT: {algo_usado}") # Si no se especifican algoritmos, usar el detectado # If no algorithms are specified, use the detected one if algorithms is None: algorithms = [algo_usado] print(f"ℹ Allowed algorithms: {algorithms}") # Cargar el certificado de la CA # Load the CA certificate with open(ca_path, 'rb') as f: ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend()) # Extraer la clave pública # Extract the public key public_key = ca_cert.public_key() print(f"ℹ Public key extracted from the certificate") # Validar el token JWT # Validate the JWT token decoded = jwt.decode(token, public_key, algorithms=algorithms) print(f"✓ Token JWT validated successfully") return decoded Loading Loading @@ -346,16 +346,16 @@ def validate_jwt_with_public_key_file(token, public_key_path, algorithm="RS256") dict with the token data if valid, None otherwise """ try: # Obtener algoritmo usado en el token # Get the algorithm used in the token header = jwt.get_unverified_header(token) algo_jwt = header.get('alg', 'desconocido') print(f"ℹ Algorithm in JWT: {algo_jwt}, expected: {algorithm}") # Cargar la clave pública # Load the public key with open(public_key_path, 'rb') as f: public_key = f.read() # Validar # Validate decoded = jwt.decode(token, public_key, algorithms=[algorithm]) print(f"✓ Token JWT válido") return decoded Loading Loading
services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py +0 −2 Changes for services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py: 0 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -190,7 +190,6 @@ class SecurityOperations(Resource): invoker = self.__retrieve_invoker_from_db(api_invoker_id) if invoker is None: current_app.logger.warning("Invoker not found on this CCF") # TODO -> Comprobar si el invoker está registrado en algún CCF remoto. remote_ccf = self.get_remote_invoker_ccf(api_invoker_id) if remote_ccf: current_app.logger.info(f"Invoker found in remote CCFs: {remote_ccf}") Loading Loading @@ -974,7 +973,6 @@ class SecurityOperations(Resource): def get_local_api_id(self, ccf_id, remote_api_id): """Retrieve the local API ID corresponding to the remote API ID for the given certificate common name.""" # TODO: Implement the mapping logic between remote API ID and local API ID interconnected_col = self.db.get_col_by_name(self.db.interconnected) interconnected_ccf = interconnected_col.find_one({"ccf_id": ccf_id}) Loading
services/nginx/policies/security-mtls.conf +1 −1 Changes for services/nginx/policies/security-mtls.conf: 1 added line, 1 removed line. Original line number Diff line number Diff line map "$endpoint:$method:$role" $security_service_mtls_policy { default DENY; # Patrones específicos por endpoint/método con múltiples roles permitidos # Specific patterns for endpoint/method with multiple allowed roles ~^security_trusted_invokers_exact:(DELETE|GET):(aef|ccf)$ ALLOW; ~^security_trusted_invokers_exact:PUT:(invoker|ccf)$ ALLOW; ~^security_update:POST:(invoker|ccf)$ ALLOW; Loading
tests/libraries/helpers.py +9 −9 Changes for tests/libraries/helpers.py: 9 added lines, 9 removed lines. Original line number Diff line number Diff line Loading @@ -268,7 +268,7 @@ def get_jwt_header_info(token): dict with the header information """ try: # Decodificar sin validar firma (solo para ver el header) # Decode without verifying the signature (just to see the header) header = jwt.get_unverified_header(token) print(f"✓ Header JWT: {header}") return header Loading @@ -291,26 +291,26 @@ def validate_jwt_token_flexible(token, ca_path, algorithms=None): dict with the token data if valid, None otherwise """ try: # Obtener header para ver qué algoritmo usa # Get the header to see which algorithm is used header = jwt.get_unverified_header(token) algo_usado = header.get('alg', 'desconocido') print(f"ℹ Algorithm detected in JWT: {algo_usado}") # Si no se especifican algoritmos, usar el detectado # If no algorithms are specified, use the detected one if algorithms is None: algorithms = [algo_usado] print(f"ℹ Allowed algorithms: {algorithms}") # Cargar el certificado de la CA # Load the CA certificate with open(ca_path, 'rb') as f: ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend()) # Extraer la clave pública # Extract the public key public_key = ca_cert.public_key() print(f"ℹ Public key extracted from the certificate") # Validar el token JWT # Validate the JWT token decoded = jwt.decode(token, public_key, algorithms=algorithms) print(f"✓ Token JWT validated successfully") return decoded Loading Loading @@ -346,16 +346,16 @@ def validate_jwt_with_public_key_file(token, public_key_path, algorithm="RS256") dict with the token data if valid, None otherwise """ try: # Obtener algoritmo usado en el token # Get the algorithm used in the token header = jwt.get_unverified_header(token) algo_jwt = header.get('alg', 'desconocido') print(f"ℹ Algorithm in JWT: {algo_jwt}, expected: {algorithm}") # Cargar la clave pública # Load the public key with open(public_key_path, 'rb') as f: public_key = f.read() # Validar # Validate decoded = jwt.decode(token, public_key, algorithms=[algorithm]) print(f"✓ Token JWT válido") return decoded Loading