Fixed cre4ate security context with aefId or InterfaceDescription (12c522d4) · Commits · OCF / capif

services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py

+66 −10

Original line number	Diff line number	Diff line
		@@ -274,17 +274,17 @@ class SecurityOperations(Resource):
		capif_service_col = self.db.get_col_by_name(
		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,
		"aef_profiles.interface_descriptions":{
		"$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(
		"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")
		@@ -306,28 +306,84 @@ 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.

		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", [])))

		current_app.logger.debug("Interface security methods: " + str(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")

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

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

		else:
		capif_service_col = self.db.get_col_by_name(
		self.db.capif_service_col)
		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))
		if services_security_object is None:
		current_app.logger.error(
		"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")

		# 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.")

		# 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)
		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
		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_methods) & set(pref_security_methods)