Fixed update security (bc8d4553) · Commits · OCF / capif

services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py

+101 −44

Original line number	Original line	Diff line number	Diff line
	@@ -422,14 +422,14 @@ 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}")

	if psk_interface:		if psk_interface:
	current_app.logger.debug("Deriving PSK")		current_app.logger.debug("Deriving PSK")
	psk = self.__derive_psk(Mkey, sesionId, psk_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)
	@@ -455,7 +455,7 @@ class SecurityOperations(Resource):
	projection={'_id': 0, 'api_invoker_id': 0}		projection={'_id': 0, 'api_invoker_id': 0}
	)		)

	res = make_response(object=dict_to_camel_case(security_context), 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)}"

	return res		return res
	@@ -581,54 +581,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.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")
	@@ -639,34 +710,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 service_instance.interface_details:
	current_app.logger.debug("Interface details found")
	interface = service_instance.interface_details.to_dict()

	else:		if psk_interface:
	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)