Adding new tests to start development of security context (aba73c90) · Commits · OCF / capif

tests/features/Helper/Interconnection API/interconnection.robot

+98 −0

Changes for tests/features/Helper/Interconnection API/interconnection.robot: 98 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -530,3 +530,101 @@ Publish API at CCF_B after CAPIFs Interconnection with capifProvDoms not present
		# ... username=${CCF_USERNAME}

		# Should Be Equal As Integers ${resp.status_code} 204



		Create a security context for an API invoker to access a published service API at other CCF (OAUTH)
		[Tags] interconnection-9 -all

		# Interconnection Request between CCF-A and CCF-B
		${INTERCONNECTION_CCF_HOSTNAME}= Set Variable capifcore-b:1443
		${INTERCONNECTION_CCF_REGISTER_HOSTNAME}= Set Variable register-b:8184

		Set Global Variable ${CAPIF_HTTPS_URL_B} https://${INTERCONNECTION_CCF_HOSTNAME}/
		Set Global Variable ${CAPIF_HTTPS_REGISTER_URL_B} https://${INTERCONNECTION_CCF_REGISTER_HOSTNAME}/

		${provider_username_b}= Set Variable ${provider_username}_B

		# Publish one api on CCF-A
		## Register APF
		${register_user_info_provider_b}= Provider Default Registration register_server=${CAPIF_HTTPS_REGISTER_URL_B} capif_server=${CAPIF_HTTPS_URL_B} provider_username=${provider_username_b}

		## Publish one api
		${service_api_description_published_b} ${resource_url} ${request_body}= Publish Service Api
		... ${register_user_info_provider_b}
		... is_shareable=${True}
		... server=${CAPIF_HTTPS_URL_B}

		# Request interconnection between CCF-A and CCF-B
		${resp}= Create Capif Interconnection server=${CAPIF_HTTPS_URL} ccf_hostname=${INTERCONNECTION_CCF_HOSTNAME} superadmin_username=${SUPERADMIN_USERNAME}
		${ccf_id_b}= Set Variable ${resp.json()['ccfId']}

		## Default Invoker Registration and Onboarding
		${register_user_info_invoker_a} ${url} ${request_body}= Invoker Default Onboarding

		## Test
		${discover_response}= Get Request Capif
		... ${DISCOVER_URL}${register_user_info_invoker_a['api_invoker_id']}&aef-id=${register_user_info_provider_b['aef_id']}
		... server=${CAPIF_HTTPS_URL}
		... verify=ca.crt
		... username=${INVOKER_USERNAME}

		Check Response Variable Type And Values ${discover_response} 200 DiscoveredAPIs

		# Check Results
		Log "API discovered by authorised API invoker at CCF-A"
		Should Not Be Empty ${discover_response.json()['serviceAPIDescriptions']}
		Length Should Be ${discover_response.json()['serviceAPIDescriptions']} 1
		# Should Not Be Empty ${discover_response.json()['serviceAPIDescriptions'][0]['pubApiPath']}
		# Length Should Be ${discover_response.json()['serviceAPIDescriptions'][0]['pubApiPath']['ccfIds']} 1
		# List Should Contain Value ${discover_response.json()['serviceAPIDescriptions'][0]['pubApiPath']['ccfIds']} ${ccf_id_b}

		# create Security Context
		${request_body} ${api_ids} ${aef_ids}= Create Service Security From Discover Response
		... ${NOTIFICATION_DESTINATION_URL}
		... ${discover_response}
		... legacy=${False}

		${security_response}= Put Request Capif
		... /capif-security/v1/trustedInvokers/${register_user_info_invoker_a['api_invoker_id']}
		... json=${request_body}
		... server=${CAPIF_HTTPS_URL}
		... verify=ca.crt
		... username=${INVOKER_USERNAME}

		# Check Results
		Check Response Variable Type And Values ${security_response} 201 ServiceSecurity
		${resource_url}= Check Location Header ${security_response} ${LOCATION_SECURITY_RESOURCE_REGEX}


		# Request creation of Security Context
		${service_name}= Get Service Name By Api Id ${discover_response} ${security_response.json()['securityInfo'][0]['apiId']}

		# Retrieve Token from CCF
		${scope}= Create Scope ${security_response.json()['securityInfo'][0]['aefId']} ${service_name}
		${request_body}= Create Access Token Req Body ${register_user_info_invoker_a['api_invoker_id']} ${scope}
		${resp}= Post Request Capif
		... /capif-security/v1/securities/${register_user_info_invoker_a['api_invoker_id']}/token
		... data=${request_body}
		... server=${CAPIF_HTTPS_URL}
		... verify=ca.crt
		... username=${INVOKER_USERNAME}

		# Check Results
		Check Response Variable Type And Values ${resp} 200 AccessTokenRsp
		... token_type=Bearer

		Should Not Be Empty ${resp.json()['access_token']}

		# Check JWT Token Validity
		${payload}= Validate JWT Token Flexible ${resp.json()['access_token']} ${CCF_USERNAME}.crt
		Should Not Be Empty ${payload}
		Log Valid JWT Token. Payload: ${payload}

		Retrieve CCF Certificates ${CCF_USERNAME}_B capif_url=${CAPIF_HTTPS_URL_B} vault_url=${CAPIF_HTTP_VAULT_URL} verify=ca.crt username=${SUPERADMIN_USERNAME}


		# Check JWT Token Validity
		${payload}= Validate JWT Token Flexible ${resp.json()['access_token']} ${CCF_USERNAME}_B.crt
		Should Not Be Empty ${payload}
		Log Valid JWT Token. Payload: ${payload}
		No newline at end of file

tests/features/init.robot

+1 −10

Changes for tests/features/__init__.robot: 1 added line, 10 removed lines.
Original line number	Diff line number	Diff line
		@@ -70,14 +70,5 @@ Retrieve Superadmin Cert
		Retrieve CCF Cert
		[Documentation] This keyword retrieve ccf certificate from CAPIF and store it at ccf.crt in order to use at TLS communications

		${ccf_id}= Get Capif Ccf Id
		Log CCF ID obtained: ${ccf_id}
		Retrieve CCF Certificates ${CCF_USERNAME}


		${resp}= Obtain CCF Cert From Vault /v1/secret/data/capif/${ccf_id}/nginx ${CAPIF_HTTP_VAULT_URL}
		Status Should Be 200 ${resp}
		Log Dictionary ${resp.json()}
		Log ${resp.json()['data']['data']}
		Store In File ${CCF_USERNAME}.crt ${resp.json()['data']['data']['server_crt']}
		Store In File ${CCF_USERNAME}.key ${resp.json()['data']['data']['server_key']}
		Store In File ${CCF_USERNAME}_ca.crt ${resp.json()['data']['data']['ca']}

tests/libraries/api_logging_service/bodyRequests.py

+7 −0

Changes for tests/libraries/api_logging_service/bodyRequests.py: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -65,6 +65,13 @@ def get_api_ids_and_names_from_discover_response(discover_response):
		api_names.append(service_api_description['apiName'])
		return api_ids, api_names

		def get_service_name_by_api_id(discover_response, api_id):
		service_api_descriptions = discover_response.json()['serviceAPIDescriptions']
		for service_api_description in service_api_descriptions:
		if service_api_description['apiId'] == api_id:
		return service_api_description['apiName']
		return None


		def create_log(apiId, apiName, result, api_version='v1'):
		log= {

tests/libraries/helpers.py

+157 −0

Changes for tests/libraries/helpers.py: 157 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -4,6 +4,8 @@ from cryptography import x509
		from cryptography.hazmat.primitives import hashes, serialization
		from cryptography.hazmat.primitives.asymmetric import rsa
		from cryptography.x509.oid import NameOID
		from cryptography.hazmat.backends import default_backend
		import jwt
		import socket
		import copy
		import pickle
		@@ -212,3 +214,158 @@ def create_open_aef_profile_from_aef_profile(aef_profile):
		open_aef_profile[field] = aef_profile[field]

		return open_aef_profile


		def validate_certificate_against_ca(cert_path, ca_path):
		"""
		Validate a certificate against its CA.

		Args:
		cert_path: Path to the certificate file (.pem)
		ca_path: Path to the CA file (.pem)

		Returns:
		True if the certificate is valid, False otherwise

		Example (Robot Framework):
		${result}= Validate Certificate Against CA /path/to/cert.pem /path/to/ca.pem
		Should Be True ${result}
		"""
		try:
		# Load the certificate
		with open(cert_path, 'rb') as f:
		cert = x509.load_pem_x509_certificate(f.read(), default_backend())

		# Load the CA
		with open(ca_path, 'rb') as f:
		ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend())

		# Validate that the issuer of the certificate matches the subject of the CA
		is_valid = cert.issuer == ca_cert.subject

		if is_valid:
		print(f"✓ Certificate is valid. Issued by: {ca_cert.subject}")
		else:
		print(f"✗ Certificate is invalid. Issuer does not match CA")

		return is_valid

		except Exception as e:
		print(f"Error validating certificate: {e}")
		return False




		def get_jwt_header_info(token):
		"""
		Extracts information from the JWT header (algorithm, type, etc.)

		Args:
		token: The JWT token as a string

		Returns:
		dict with the header information
		"""
		try:
		# Decodificar sin validar firma (solo para ver el header)
		header = jwt.get_unverified_header(token)
		print(f"✓ Header JWT: {header}")
		return header
		except Exception as e:
		print(f"✗ Error reading JWT header: {e}")
		return None


		def validate_jwt_token_flexible(token, ca_path, algorithms=None):
		"""
		Validates JWT with support for multiple algorithms (enhanced debug).

		Args:
		token: The JWT token as a string
		ca_path: Path to the CA certificate file (.pem)
		algorithms: List of allowed algorithms (e.g., ["RS256", "RS512"])
		If None, it tries to detect automatically

		Returns:
		dict with the token data if valid, None otherwise
		"""
		try:
		# Obtener header para ver qué algoritmo usa
		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 algorithms is None:
		algorithms = [algo_usado]

		print(f"ℹ Allowed algorithms: {algorithms}")

		# Cargar el certificado de la CA
		with open(ca_path, 'rb') as f:
		ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend())

		# Extraer la clave pública
		public_key = ca_cert.public_key()
		print(f"ℹ Public key extracted from the certificate")

		# Validar el token JWT
		decoded = jwt.decode(token, public_key, algorithms=algorithms)
		print(f"✓ Token JWT validated successfully")
		return decoded

		except jwt.InvalidSignatureError:
		print(f"✗ INVALID SIGNATURE - The certificate used does NOT match the one that signed the token")
		print(f" Verify that you are using the correct certificate")
		return None
		except jwt.ExpiredSignatureError:
		print("✗ JWT token expired")
		return None
		except jwt.InvalidAlgorithmError as e:
		print(f"✗ Algorithm not allowed: {e}")
		return None
		except jwt.InvalidTokenError as e:
		print(f"✗ Invalid JWT token: {e}")
		return None
		except Exception as e:
		print(f"✗ Error validating token: {e}")
		return None


		def validate_jwt_with_public_key_file(token, public_key_path, algorithm="RS256"):
		"""
		Validates JWT using a public key in .pem format

		Args:
		token: The JWT token as a string
		public_key_path: Path to the file containing the public key (.pem)
		algorithm: Expected algorithm (RS256, RS512, HS256, etc.)

		Returns:
		dict with the token data if valid, None otherwise
		"""
		try:
		# Obtener algoritmo usado en el 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
		with open(public_key_path, 'rb') as f:
		public_key = f.read()

		# Validar
		decoded = jwt.decode(token, public_key, algorithms=[algorithm])
		print(f"✓ Token JWT válido")
		return decoded

		except jwt.InvalidSignatureError:
		print(f"✗ INVALID SIGNATURE - The public key does NOT match the one that signed the token")
		return None
		except jwt.ExpiredSignatureError:
		print("✗ JWT token expired")
		return None
		except Exception as e:
		print(f"✗ Error validating token: {e}")
		return None

tests/libraries/security_api/bodyRequests.py

+3 −1

Changes for tests/libraries/security_api/bodyRequests.py: 3 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -86,6 +86,7 @@ def create_service_security_from_discover_response(notification_destination, dis
		"requestTestNotification": True
		}
		api_ids = list()
		aef_ids = list()
		service_api_descriptions = discover_response.json()['serviceAPIDescriptions']
		for service_api_description in service_api_descriptions:
		for aef_profile in service_api_description['aefProfiles']:
		@@ -97,10 +98,11 @@ def create_service_security_from_discover_response(notification_destination, dis
		"apiId": service_api_description['apiId']
		})
		api_ids.append(service_api_description['apiId'])
		aef_ids.append(aef_profile['aefId'])
		if legacy:
		return data
		else:
		return data, api_ids
		return data, api_ids, aef_ids


		def update_service_security_with_discover_response(security_body, discover_response, legacy=True):