Merge branch 'OCF126-review-creation-of-security-contexts' into 'staging' (0a6f0f85) · Commits · OCF / capif

services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py

+216 −93

Original line number	Original line	Diff line number	Diff line
	@@ -103,6 +103,7 @@ class SecurityOperations(Resource):
	## Derive the PSK using the provided master key, session ID, and interface information		## Derive the PSK using the provided master key, session ID, and interface information

	# Interface information		# Interface information
			if isinstance(interface, dict):
	host = None		host = None
	if 'fqdn' in interface:		if 'fqdn' in interface:
	host = interface['fqdn']		host = interface['fqdn']
	@@ -119,6 +120,9 @@ class SecurityOperations(Resource):
	if port and port != 443:		if port and port != 443:
	interface_info += f":{port}"		interface_info += f":{port}"
	interface_info += api_prefix		interface_info += api_prefix
			else:
			interface_info = interface


	# Normalize the strings to NFKC form		# Normalize the strings to NFKC form
	p0_string = unicodedata.normalize("NFKC", interface_info).encode("utf-8")		p0_string = unicodedata.normalize("NFKC", interface_info).encode("utf-8")
	@@ -256,7 +260,7 @@ class SecurityOperations(Resource):
	return bad_request_error(detail="Bad Param", cause="Detected Bad format of param", invalid_params=[{"param": "notificationDestination", "reason": "Not valid URL format"}])		return bad_request_error(detail="Bad Param", cause="Detected Bad format of param", invalid_params=[{"param": "notificationDestination", "reason": "Not valid URL format"}])

	services_security_object = mycol.find_one(		services_security_object = mycol.find_one(
	{"api_invoker_id": api_invoker_id})		{"api_invoker_id": api_invoker_id, "security_info.api_id": {"$in": [info.api_id for info in service_security.security_info]}}, {"_id": 0})

	if services_security_object is not None:		if services_security_object is not None:

	@@ -268,23 +272,26 @@ class SecurityOperations(Resource):
	service_security.supported_features = negotiated["Final"]		service_security.supported_features = negotiated["Final"]

	for service_instance in service_security.security_info:		for service_instance in service_security.security_info:

			psk_interface = None

	if service_instance.interface_details is not None:		if service_instance.interface_details is not None:

	# We look for if the passed interface exists for the given apiId		# We look for if the passed interface exists for the given apiId
	capif_service_col = self.db.get_col_by_name(		capif_service_col = self.db.get_col_by_name(
	self.db.capif_service_col)		self.db.capif_service_col)

	aef_profile = capif_service_col.find_one(		aef_profiles = capif_service_col.find_one(
	{"api_id": service_instance.api_id,		{"api_id": service_instance.api_id,
	"aef_profiles.interface_descriptions":{		"aef_profiles.interface_descriptions":{
	"$elemMatch": service_instance.interface_details.to_dict()		"$elemMatch": service_instance.interface_details.to_dict()
	}		}
	},		},
	{"aef_profiles.interface_descriptions.$": 1, "_id": 0})		{"_id": 0})

	current_app.logger.debug("Aef profile: " + str(aef_profile))		current_app.logger.debug("Aef profile: " + str(aef_profiles))

	if aef_profile is None:		if aef_profiles is None:
	current_app.logger.error(		current_app.logger.error(
	"Not found service with this interface description: " + json.dumps(clean_empty(service_instance.interface_details.to_dict())))		"Not found service with this interface description: " + json.dumps(clean_empty(service_instance.interface_details.to_dict())))
	return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")		return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")
	@@ -306,28 +313,95 @@ class SecurityOperations(Resource):
	#		#
	# To achieve this, we need to setup at config which domains or IPs are CAPIF-2e or CAPIF-2, and then we need to check if the domain or IP of the service is in the list.		# To achieve this, we need to setup at config which domains or IPs are CAPIF-2e or CAPIF-2, and then we need to check if the domain or IP of the service is in the list.

	security_methods = aef_profile["aef_profiles"][0]["interface_descriptions"][0]["security_methods"]		valid_security_methods = set()
			for aefProfile in aef_profiles.get("aef_profiles", []):
			current_app.logger.debug("AEF profile security methods: " + str(aefProfile.get("security_methods", [])))

			profile_methods = set(aefProfile.get("security_methods") or [])
			interfaces = aefProfile.get("interface_descriptions", [])

	current_app.logger.debug("Interface security methods: " + str(security_methods))		interface_methods = set()

			if interfaces and len(interfaces) > 0:
			for interface in interfaces:
			# If the interface has its own security methods, use them
			if interface == service_instance.interface_details.to_dict():
			if interface.get("security_methods"):
			interface_methods.update(interface["security_methods"])
			# If not, inherit the methods from the profile (if any)
			elif profile_methods:
			interface_methods.update(profile_methods)

			# After processing all interfaces, use the combined set
			valid_security_methods.update(interface_methods)
			else:
			current_app.logger.debug("No interfaces found in AEF profile.")
			return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")

			psk_interface = service_instance.interface_details.to_dict()

			current_app.logger.debug("Valid security methods: " + str(valid_security_methods))

	pref_security_methods = service_instance.pref_security_methods		pref_security_methods = service_instance.pref_security_methods
	valid_security_method = set(		valid_security_method = set(
	security_methods) & set(pref_security_methods)		valid_security_methods) & set(pref_security_methods)

	else:		else:
	capif_service_col = self.db.get_col_by_name(		capif_service_col = self.db.get_col_by_name(
	self.db.capif_service_col)		self.db.capif_service_col)
	services_security_object = capif_service_col.find_one(		services_security_object = capif_service_col.find_one(
	{"api_id": service_instance.api_id, self.filter_aef_id: service_instance.aef_id}, {"aef_profiles.security_methods.$": 1})		{"api_id": service_instance.api_id, self.filter_aef_id: service_instance.aef_id})

	current_app.logger.debug("Aef profile: " + str(services_security_object))		current_app.logger.debug("Aef profile: " + str(services_security_object))
	if services_security_object is None:		if services_security_object is None:
	current_app.logger.error(		current_app.logger.error(
	"Not found service with this aef id: " + service_instance.aef_id)		"Not found service with this aef id: " + service_instance.aef_id)
	return not_found_error(detail="Service with this aefId not found", cause="Not found Service")		return not_found_error(detail="Service with this aefId not found", cause="Not found Service")

			# We obtain all the security methods available for the given aef_id
			valid_security_methods = set()
			for aefProfile in services_security_object.get("aef_profiles", []):
			current_app.logger.debug("AEF profile security methods: " + str(aefProfile.get("security_methods", [])))

			profile_methods = set(aefProfile.get("security_methods") or [])
			interfaces = aefProfile.get("interface_descriptions", [])

			interface_methods = set()

			current_app.logger.debug(f"Interfaces: {interfaces}, Profile Methods: {profile_methods}")
			if interfaces and len(interfaces) > 0:
			for interface in interfaces:
			# If the interface has its own security methods, use them
			if interface.get("security_methods"):
			interface_methods.update(interface["security_methods"])
			# If not, inherit the methods from the profile (if any)
			elif profile_methods:
			interface_methods.update(profile_methods)
			else:
			current_app.logger.debug("Interface has no security methods and profile has none to inherit.")

			# Keep track if any interface supports PSK
			if psk_interface is None and "PSK" in interface_methods:
			psk_interface = interface

			# After processing all interfaces, use the combined set
			valid_security_methods.update(interface_methods)
			else:
			# No interfaces: use the profile's security methods directly
			if profile_methods:
			valid_security_methods.update(profile_methods)

			# Keep track if profile supports PSK
			if psk_interface is None and "PSK" in profile_methods:
			psk_interface = aefProfile.get("domain_name")

			else:
			current_app.logger.debug("AEF profile has no security methods defined (no interfaces either).")

			current_app.logger.debug("Valid security methods: " + str(valid_security_methods))

			# We intersect with preferred security methods
	pref_security_methods = service_instance.pref_security_methods		pref_security_methods = service_instance.pref_security_methods
	valid_security_methods = [security_method for array_methods in services_security_object["aef_profiles"]
	for security_method in array_methods["security_methods"]]
	valid_security_method = set(		valid_security_method = set(
	valid_security_methods) & set(pref_security_methods)		valid_security_methods) & set(pref_security_methods)

	@@ -352,34 +426,19 @@ class SecurityOperations(Resource):
	service_instance.sel_security_method = sorted_methods[0]		service_instance.sel_security_method = sorted_methods[0]

	if service_instance.sel_security_method == "PSK":		if service_instance.sel_security_method == "PSK":
	request.headers.get('X-TLS-Protocol', 'N/A')		tls_protocol = request.headers.get('X-TLS-Protocol', 'N/A')
	sesionId = request.headers.get('X-TLS-Session-ID', 'N/A')		session_id = request.headers.get('X-TLS-Session-ID', 'N/A')
	Mkey = request.headers.get('X-TLS-MKey', 'N/A')		mkey = request.headers.get('X-TLS-MKey', 'N/A')
	current_app.logger.info(f"TLS Protocol: {request.headers.get('X-TLS-Protocol', 'N/A')}, Session id: {sesionId}, Master Key: {Mkey}")		current_app.logger.info(f"TLS Protocol: {tls_protocol}, Session id: {session_id}, Master Key: {mkey}")

	interface = None		if psk_interface:
	if service_instance.interface_details:
	current_app.logger.debug("Interface details found")
	interface = service_instance.interface_details.to_dict()

	else:
	current_app.logger.error("Interface details not found")
	services_security_object = capif_service_col.find_one(
	{"api_id": service_instance.api_id}, {"aef_profiles": {"$elemMatch": {"aef_id": service_instance.aef_id}}, "_id": 0})
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]))
	if "interface_descriptions" in services_security_object["aef_profiles"][0]:
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]["interface_descriptions"]))
	interface = services_security_object["aef_profiles"][0]["interface_descriptions"][0]
	elif "domain_name" in services_security_object["aef_profiles"][0]:
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]["domain_name"]))
	interface = services_security_object["aef_profiles"][0]["domain_name"]

	if interface:
	current_app.logger.debug("Deriving PSK")		current_app.logger.debug("Deriving PSK")
	psk = self.__derive_psk(Mkey, sesionId, interface)		psk = self.__derive_psk(mkey, session_id, psk_interface)
	current_app.logger.debug("PSK derived : " + str(psk))		current_app.logger.debug("PSK derived : " + str(psk))

	service_instance.authorization_info = str(psk)		service_instance.authorization_info = str(psk)
			else:
			current_app.logger.error("No interface information available to derive PSK")

	# Send service instance to ACL		# Send service instance to ACL
	current_app.logger.debug("Sending message to create ACL")		current_app.logger.debug("Sending message to create ACL")
	@@ -388,10 +447,17 @@ class SecurityOperations(Resource):
	current_app.logger.debug(		current_app.logger.debug(
	"Inserted security context in database")		"Inserted security context in database")

	rec = dict()		# We use update with $setOnInsert and $push with $each to add the security info array if the document is created
	rec['api_invoker_id'] = api_invoker_id		on_insert = service_security.to_dict().copy()
	rec.update(service_security.to_dict())		on_insert.pop('security_info', None)
	mycol.insert_one(rec)
			security_context = mycol.find_one_and_update({'api_invoker_id': api_invoker_id},
			{"$setOnInsert": on_insert,
			"$push": {"security_info": {"$each": [sec.to_dict() for sec in service_security.security_info]}}},
			upsert=True ,
			return_document=ReturnDocument.AFTER,
			projection={'_id': 0, 'api_invoker_id': 0}
			)

	res = make_response(object=serialize_clean_camel_case(service_security), status=201)		res = make_response(object=serialize_clean_camel_case(service_security), status=201)
	res.headers['Location'] = f"https://{os.getenv("CAPIF_HOSTNAME")}/capif-security/v1/trustedInvokers/{str(api_invoker_id)}"		res.headers['Location'] = f"https://{os.getenv("CAPIF_HOSTNAME")}/capif-security/v1/trustedInvokers/{str(api_invoker_id)}"
	@@ -519,54 +585,125 @@ class SecurityOperations(Resource):

	update_acls=list()		update_acls=list()
	for service_instance in service_security.security_info:		for service_instance in service_security.security_info:

			psk_interface = None

	if service_instance.interface_details is not None:		if service_instance.interface_details is not None:

	# We look for if the passed interface exists for the given apiId		# We look for if the passed interface exists for the given apiId
	capif_service_col = self.db.get_col_by_name(		capif_service_col = self.db.get_col_by_name(
	self.db.capif_service_col)		self.db.capif_service_col)

	aef_profile = capif_service_col.find_one(		aef_profiles = capif_service_col.find_one(
	{"api_id": service_instance.api_id,		{"api_id": service_instance.api_id,
	"aef_profiles.interface_descriptions":{		"aef_profiles.interface_descriptions":{
	"$elemMatch": service_instance.interface_details.to_dict()		"$elemMatch": service_instance.interface_details.to_dict()
	}		}
	},		},
	{"aef_profiles.interface_descriptions.$": 1, "_id": 0})		{"_id": 0})

	current_app.logger.debug("Aef profile: " + str(aef_profile))		current_app.logger.debug("Aef profile: " + str(aef_profile))

	if aef_profile is None:		if aef_profiles is None:
	current_app.logger.error(		current_app.logger.error(
	"Not found service with this interface description: " + json.dumps(clean_empty(service_instance.interface_details.to_dict())))		"Not found service with this interface description: " + json.dumps(clean_empty(service_instance.interface_details.to_dict())))
	return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")		return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")

	# We obtain the interface security methods
	security_methods = aef_profile["aef_profiles"][0]["interface_descriptions"][0]["security_methods"]

	current_app.logger.debug("Interface security methods: " + str(security_methods))		valid_security_methods = set()
			for aefProfile in aef_profiles.get("aef_profiles", []):
			current_app.logger.debug("AEF profile security methods: " + str(aefProfile.get("security_methods", [])))

			profile_methods = set(aefProfile.get("security_methods") or [])
			interfaces = aefProfile.get("interface_descriptions", [])

			interface_methods = set()

			if interfaces and len(interfaces) > 0:
			for interface in interfaces:
			# If the interface has its own security methods, use them
			if interface == service_instance.interface_details.to_dict():
			if interface.get("security_methods"):
			interface_methods.update(interface["security_methods"])
			# If not, inherit the methods from the profile (if any)
			elif profile_methods:
			interface_methods.update(profile_methods)

			# After processing all interfaces, use the combined set
			valid_security_methods.update(interface_methods)
			else:
			current_app.logger.debug("No interfaces found in AEF profile.")
			return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")

			psk_interface = service_instance.interface_details.to_dict()

			current_app.logger.debug("Valid security methods: " + str(valid_security_methods))

	pref_security_methods = service_instance.pref_security_methods		pref_security_methods = service_instance.pref_security_methods
	valid_security_method = set(		valid_security_method = set(
	security_methods) & set(pref_security_methods)		valid_security_methods) & set(pref_security_methods)

	else:		else:


	capif_service_col = self.db.get_col_by_name(		capif_service_col = self.db.get_col_by_name(
	self.db.capif_service_col)		self.db.capif_service_col)
	services_security_object = capif_service_col.find_one(		services_security_object = capif_service_col.find_one(
	{self.filter_aef_id: service_instance.aef_id}, {"aef_profiles.security_methods.$": 1})		{"api_id": service_instance.api_id, self.filter_aef_id: service_instance.aef_id})

			current_app.logger.debug("Aef profile: " + str(services_security_object))
	if services_security_object is None:		if services_security_object is None:
	current_app.logger.error(		current_app.logger.error(
	"Service api with this aefId not found: " + service_instance.aef_id)		"Not found service with this aef id: " + service_instance.aef_id)
	return not_found_error(detail="Service with this aefId not found", cause="Not found Service")		return not_found_error(detail="Service with this aefId not found", cause="Not found Service")

			# We obtain all the security methods available for the given aef_id
			valid_security_methods = set()
			for aefProfile in services_security_object.get("aef_profiles", []):
			current_app.logger.debug("AEF profile security methods: " + str(aefProfile.get("security_methods", [])))

			profile_methods = set(aefProfile.get("security_methods") or [])
			interfaces = aefProfile.get("interface_descriptions", [])

			interface_methods = set()

			current_app.logger.debug(f"Interfaces: {interfaces}, Profile Methods: {profile_methods}")
			if interfaces and len(interfaces) > 0:
			for interface in interfaces:
			# If the interface has its own security methods, use them
			if interface.get("security_methods"):
			interface_methods.update(interface["security_methods"])
			# If not, inherit the methods from the profile (if any)
			elif profile_methods:
			interface_methods.update(profile_methods)
			else:
			current_app.logger.debug("Interface has no security methods and profile has none to inherit.")

			# Keep track if any interface supports PSK
			if psk_interface is None and "PSK" in interface_methods:
			psk_interface = interface

			# After processing all interfaces, use the combined set
			valid_security_methods.update(interface_methods)
			else:
			# No interfaces: use the profile's security methods directly
			if profile_methods:
			valid_security_methods.update(profile_methods)

			# Keep track if profile supports PSK
			if psk_interface is None and "PSK" in profile_methods:
			psk_interface = aefProfile.get("domain_name")

			else:
			current_app.logger.debug("AEF profile has no security methods defined (no interfaces either).")

			current_app.logger.debug("Valid security methods: " + str(valid_security_methods))

			# We intersect with preferred security methods
	pref_security_methods = service_instance.pref_security_methods		pref_security_methods = service_instance.pref_security_methods
	valid_security_methods = [security_method for array_methods in services_security_object["aef_profiles"]
	for security_method in array_methods["security_methods"]]
	valid_security_method = set(		valid_security_method = set(
	valid_security_methods) & set(pref_security_methods)		valid_security_methods) & set(pref_security_methods)


	if len(list(valid_security_method)) == 0:		if len(list(valid_security_method)) == 0:
	current_app.logger.error(		current_app.logger.error(
	"Not found comptaible security method with pref security method")		"Not found comptaible security method with pref security method")
	@@ -577,34 +714,20 @@ class SecurityOperations(Resource):
	update_acls.append({"api_id": service_instance.api_id, "aef_id": service_instance.aef_id})		update_acls.append({"api_id": service_instance.api_id, "aef_id": service_instance.aef_id})

	if service_instance.sel_security_method == "PSK":		if service_instance.sel_security_method == "PSK":
	request.headers.get('X-TLS-Protocol', 'N/A')		tls_protocol = request.headers.get('X-TLS-Protocol', 'N/A')
	sesionId = request.headers.get('X-TLS-Session-ID', 'N/A')		session_id = request.headers.get('X-TLS-Session-ID', 'N/A')
	Mkey = request.headers.get('X-TLS-MKey', 'N/A')		mkey = request.headers.get('X-TLS-MKey', 'N/A')
	current_app.logger.info(f"TLS Protocol: {request.headers.get('X-TLS-Protocol', 'N/A')}, Session id: {sesionId}, Master Key: {Mkey}")		current_app.logger.info(f"TLS Protocol: {tls_protocol}, Session id: {session_id}, Master Key: {mkey}")

	interface = None		if psk_interface:
	if service_instance.interface_details:
	current_app.logger.debug("Interface details found")
	interface = service_instance.interface_details.to_dict()

	else:
	current_app.logger.error("Interface details not found")
	services_security_object = capif_service_col.find_one(
	{"api_id": service_instance.api_id}, {"aef_profiles": {"$elemMatch": {"aef_id": service_instance.aef_id}}, "_id": 0})
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]))
	if "interface_descriptions" in services_security_object["aef_profiles"][0]:
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]["interface_descriptions"]))
	interface = services_security_object["aef_profiles"][0]["interface_descriptions"][0]
	elif "domain_name" in services_security_object["aef_profiles"][0]:
	current_app.logger.debug("Aef profile: " + str(services_security_object["aef_profiles"][0]["domain_name"]))
	interface = services_security_object["aef_profiles"][0]["domain_name"]

	if interface:
	current_app.logger.debug("Deriving PSK")		current_app.logger.debug("Deriving PSK")
	psk = self.__derive_psk(Mkey, sesionId, interface)		psk = self.__derive_psk(mkey, session_id, psk_interface)
	current_app.logger.debug("PSK derived : " + str(psk))		current_app.logger.debug("PSK derived : " + str(psk))

	service_instance.authorization_info = str(psk)		service_instance.authorization_info = str(psk)
			else:
			current_app.logger.error("No interface information available to derive PSK")

	service_security = service_security.to_dict()		service_security = service_security.to_dict()
	service_security = clean_empty(service_security)		service_security = clean_empty(service_security)

services/nginx/nginx.conf

+1 −0

Original line number	Original line	Diff line number	Diff line
	@@ -70,6 +70,7 @@ http {
	ssl_client_certificate /etc/nginx/certs/ca.crt;		ssl_client_certificate /etc/nginx/certs/ca.crt;
	ssl_verify_client optional;		ssl_verify_client optional;
	ssl_verify_depth 2;		ssl_verify_depth 2;
			ssl_session_tickets off;

	location / {		location / {
	proxy_pass $scheme://$http_host/api-invoker-management/v1/ui/;		proxy_pass $scheme://$http_host/api-invoker-management/v1/ui/;