Ocf126 review creation of security contexts

Pelayo Torresrequested to merge
OCF126-review-creation-of-security-contexts into staging

Proposers

  • Pelayo Torres Alonso

Description

The creation and update of security contexts has been fixed, taking into account whether the aefId or interfaceDescription is used in the request. Added priority for security methods based on whether they are in the API published in the interfaces or the AEF profile. Fixed an issue with PSK creation when using domainName

Acknowledgements

This work is funded by the European Commission through the HORIZON-JU-SNS-2022 FIDAL project with Grant Agreement number 101096146.

Merge request reports

services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py

+196 −73
Changes for services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py: 196 added lines, 73 removed lines.
Original line number 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



        # Interface information

        if isinstance(interface, dict):

            host = None

            if 'fqdn' in interface:

                host = interface['fqdn']
@@ -119,6 +120,9 @@ class SecurityOperations(Resource): 
            if port and port != 443:

                interface_info += f":{port}"

            interface_info += api_prefix

        else:

            interface_info = interface



        

        # Normalize the strings to NFKC form

        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"}])



            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:
@@ -268,23 +272,26 @@ class SecurityOperations(Resource): 
            service_security.supported_features = negotiated["Final"]



            for service_instance in service_security.security_info:



                psk_interface = None



                if service_instance.interface_details is not None:



                    # We look for if the passed interface exists for the given apiId

                    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 +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.



                    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

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



                                # 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

                    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)
@@ -352,34 +426,19 @@ class SecurityOperations(Resource): 
                service_instance.sel_security_method = sorted_methods[0]



                if service_instance.sel_security_method == "PSK":

                    request.headers.get('X-TLS-Protocol', 'N/A')

                    sesionId = request.headers.get('X-TLS-Session-ID', '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}") 

                    tls_protocol = request.headers.get('X-TLS-Protocol', 'N/A')

                    session_id = request.headers.get('X-TLS-Session-ID', 'N/A')  

                    mkey = request.headers.get('X-TLS-MKey', 'N/A') 

                    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:

                        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:

                    if psk_interface:

                        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))



                        service_instance.authorization_info = str(psk)

                    else:

                        current_app.logger.error("No interface information available to derive PSK")

                        

                # Send service instance to ACL

                current_app.logger.debug("Sending message to create ACL")
@@ -388,10 +447,17 @@ class SecurityOperations(Resource): 
                current_app.logger.debug(

                    "Inserted security context in database")



            rec = dict()

            rec['api_invoker_id'] = api_invoker_id

            rec.update(service_security.to_dict())

            mycol.insert_one(rec)

            # We use update with $setOnInsert and $push with $each to add the security info array if the document is created

            on_insert = service_security.to_dict().copy()

            on_insert.pop('security_info', None) 



            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.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()

            for service_instance in service_security.security_info:



                psk_interface = None



                if service_instance.interface_details is not None:



                     # We look for if the passed interface exists for the given apiId

                    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))



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



                    # 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

                    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(

                        {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(

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

                    

                    # 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

                    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)

                    

                    

                if len(list(valid_security_method)) == 0:

                    current_app.logger.error(

                        "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})



                if service_instance.sel_security_method == "PSK":

                    request.headers.get('X-TLS-Protocol', 'N/A')

                    sesionId = request.headers.get('X-TLS-Session-ID', '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}") 

                    tls_protocol = request.headers.get('X-TLS-Protocol', 'N/A')

                    session_id = request.headers.get('X-TLS-Session-ID', 'N/A')  

                    mkey = request.headers.get('X-TLS-MKey', 'N/A') 

                    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:

                        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:

                    if psk_interface:

                        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))



                        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 = clean_empty(service_security)

services/nginx/nginx.conf

+1 −0
Changes for services/nginx/nginx.conf: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -70,6 +70,7 @@ http { 
              ssl_client_certificate  /etc/nginx/certs/ca.crt;

              ssl_verify_client       optional;

              ssl_verify_depth        2;

              ssl_session_tickets off;



              location / {

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