Commit 7d516b8c authored by Jorge Moratinos's avatar Jorge Moratinos
Browse files

Merge branch... 

Merge branch 'OCF193-critical-authorization-bypass-vulnerability-in-capif-certificate-validation' into 'staging'

Resolve "Critical Authorization Bypass Vulnerability in CAPIF Certificate Validation"

See merge request !182
parents 12f89e33 16af8fab

services/TS29222_CAPIF_API_Invoker_Management_API/api_invoker_management/core/validate_user.py

+7 −7
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ from ..encoder import CustomJSONEncoder 
from ..models.problem_details import ProblemDetails

from ..util import serialize_clean_camel_case

from .resources import Resource

from .responses import internal_server_error

from .responses import forbidden_error, internal_server_error, not_found_error, unauthorized_error





class ControlAccess(Resource):
@@ -16,14 +16,14 @@ class ControlAccess(Resource): 
        cert_col = self.db.get_col_by_name(self.db.certs_col)



        try:

            my_query = {'invoker_id':api_invoker_id}

            my_query = {'id':api_invoker_id}

            cert_entry = cert_col.find_one(my_query)



            if cert_entry is not None:

            if cert_entry is None:

                return not_found_error(detail="Please provide an existing Network App ID", cause="Network App ID does not exist")

            

            if cert_entry["cert_signature"] != cert_signature:

                    prob = ProblemDetails(title="Unauthorized", detail="User not authorized", cause="You are not the owner of this resource")

                    prob = serialize_clean_camel_case(prob)

                    return Response(json.dumps(prob, cls=CustomJSONEncoder), status=401, mimetype="application/json")

                return forbidden_error(detail="User not authorized", cause="You are not the owner of this resource")



        except Exception as e:

            exception = "An exception occurred in validate invoker"

services/TS29222_CAPIF_API_Provider_Management_API/api_provider_management/core/validate_user.py

+10 −6
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ from ..encoder import CustomJSONEncoder 
from ..models.problem_details import ProblemDetails

from ..util import serialize_clean_camel_case

from .resources import Resource

from .responses import internal_server_error

from .responses import internal_server_error, not_found_error, forbidden_error





class ControlAccess(Resource):
@@ -19,11 +19,15 @@ class ControlAccess(Resource): 
            my_query = {'provider_id':api_provider_id, "role": "AMF"}

            cert_entry = cert_col.find_one(my_query)



            if cert_entry is not None:

            if cert_entry is None:

                provider_exists = cert_col.find_one({'provider_id': api_provider_id})

                if provider_exists is None:

                    return not_found_error(detail="Please provide an existing API Provider ID", cause="API Provider ID does not exist")

                else:

                    return forbidden_error(detail="AMF certificate required", cause="Only API Management Function (AMF) certificates can manage provider registrations")

            

            if cert_entry["cert_signature"] != cert_signature:

                    prob = ProblemDetails(title="Unauthorized", detail="User not authorized", cause="You are not the owner of this resource")

                    prob = serialize_clean_camel_case(prob)

                    return Response(json.dumps(prob, cls=CustomJSONEncoder), status=401, mimetype="application/json")

                return forbidden_error(detail="User not authorized", cause="You are not the owner of this resource")



        except Exception as e:

            exception = "An exception occurred in validate amf"

services/TS29222_CAPIF_Auditing_API/logs/core/validate_user.py

+5 −6
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ from ..encoder import CustomJSONEncoder 
from ..models.problem_details import ProblemDetails

from ..util import serialize_clean_camel_case

from .resources import Resource

from .responses import internal_server_error

from .responses import internal_server_error, not_found_error, forbidden_error





class ControlAccess(Resource):
@@ -19,11 +19,10 @@ class ControlAccess(Resource): 
            my_query = {'cert_signature': cert_signature}

            cert_entry = cert_col.find_one(my_query)



            if cert_entry is not None:

                if cert_entry["role"] != "AMF":

                    prob = ProblemDetails(title="Unauthorized", detail="User not authorized", cause="You are not the owner of this resource")

                    prob = serialize_clean_camel_case(prob)

                    return Response(json.dumps(prob, cls=CustomJSONEncoder), status=401, mimetype="application/json")

            if cert_entry is None:

                return not_found_error(detail="Certificate not found", cause="No certificate matches the provided signature")

            elif cert_entry["role"] != "AMF":

                return forbidden_error(detail="User not authorized", cause="You are not the owner of this resource")



        except Exception as e:

            exception = "An exception occurred in validate invoker"

services/TS29222_CAPIF_Discover_Service_API/service_apis/core/validate_user.py

+6 −6
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ from ..encoder import CustomJSONEncoder 
from ..models.problem_details import ProblemDetails

from ..util import serialize_clean_camel_case

from .resources import Resource

from .responses import internal_server_error

from .responses import internal_server_error, not_found_error, forbidden_error





class ControlAccess(Resource):
@@ -20,11 +20,11 @@ class ControlAccess(Resource): 
            my_query = {'id': api_invoker_id}

            cert_entry = cert_col.find_one(my_query)



            if cert_entry is not None:

            if cert_entry is None:

                return not_found_error(detail="Please provide an existing Network App ID", cause="Certificate not found for invoker")

            

            if cert_entry["cert_signature"] != cert_signature:

                    prob = ProblemDetails(title="Unauthorized", detail="User not authorized", cause="You are not the owner of this resource")

                    prob = serialize_clean_camel_case(prob)

                    return Response(json.dumps(prob, cls=CustomJSONEncoder), status=401, mimetype="application/json")

                return forbidden_error(detail="User not authorized", cause="You are not the owner of this resource")



        except Exception as e:

            exception = "An exception occurred in validate invoker"

services/TS29222_CAPIF_Events_API/capif_events/core/validate_user.py

+12 −10
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ from ..encoder import CustomJSONEncoder 
from ..models.problem_details import ProblemDetails

from ..util import serialize_clean_camel_case

from .resources import Resource

from .responses import internal_server_error

from .responses import internal_server_error, not_found_error, forbidden_error





class ControlAccess(Resource):
@@ -19,7 +19,9 @@ class ControlAccess(Resource): 
            my_query = {'id':subscriber_id}

            cert_entry = cert_col.find_one(my_query)



            if cert_entry is not None:

            if cert_entry is None:

                return not_found_error(detail="Please provide an existing Subscriber ID", cause="Certificate not found for Invoker or APF or AEF or AMF")

            

            if (event_id is None and cert_entry["cert_signature"] != cert_signature):

                prob = ProblemDetails(title="Unauthorized", detail="User not authorized", cause="You are not the owner of this resource")

                prob = serialize_clean_camel_case(prob)