Loading services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py +18 −0 Original line number Diff line number Diff line Loading @@ -24,6 +24,18 @@ security_context_not_found_detail = "Security context not found" api_invoker_no_context_cause = "API Invoker has no security context" TOTAL_FEATURES = 3 SUPPORTED_FEATURES_HEX = "4" def return_negotiated_supp_feat_dict(supp_feat): final_supp_feat = bin(int(supp_feat, 16) & int(SUPPORTED_FEATURES_HEX, 16))[2:].zfill(TOTAL_FEATURES)[::-1] return { "Notification_test_event": True if final_supp_feat[0] == "1" else False, "Notification_websocket": True if final_supp_feat[1] == "1" else False, "SecurityInfoPerAPI": True if final_supp_feat[2] == "1" else False, "Final": hex(int(final_supp_feat[::-1], 2))[2:] } class SecurityOperations(Resource): def __check_invoker(self, api_invoker_id): Loading Loading @@ -207,6 +219,9 @@ class SecurityOperations(Resource): "Already security context defined with same api invoker id") return forbidden_error(detail="Security method already defined", cause="Identical AEF Profile IDs") negotiated = return_negotiated_supp_feat_dict(service_security.supported_features) service_security.supported_features = negotiated["Final"] for service_instance in service_security.security_info: if service_instance.interface_details is not None: Loading Loading @@ -412,6 +427,9 @@ class SecurityOperations(Resource): mycol = self.db.get_col_by_name(self.db.security_info) try: negotiated_supported_features = return_negotiated_supp_feat_dict(service_security.supported_features) service_security.supported_features = negotiated_supported_features["Final"] current_app.logger.debug("Updating security context") result = self.__check_invoker(api_invoker_id) if result != None: Loading Loading
services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py +18 −0 Original line number Diff line number Diff line Loading @@ -24,6 +24,18 @@ security_context_not_found_detail = "Security context not found" api_invoker_no_context_cause = "API Invoker has no security context" TOTAL_FEATURES = 3 SUPPORTED_FEATURES_HEX = "4" def return_negotiated_supp_feat_dict(supp_feat): final_supp_feat = bin(int(supp_feat, 16) & int(SUPPORTED_FEATURES_HEX, 16))[2:].zfill(TOTAL_FEATURES)[::-1] return { "Notification_test_event": True if final_supp_feat[0] == "1" else False, "Notification_websocket": True if final_supp_feat[1] == "1" else False, "SecurityInfoPerAPI": True if final_supp_feat[2] == "1" else False, "Final": hex(int(final_supp_feat[::-1], 2))[2:] } class SecurityOperations(Resource): def __check_invoker(self, api_invoker_id): Loading Loading @@ -207,6 +219,9 @@ class SecurityOperations(Resource): "Already security context defined with same api invoker id") return forbidden_error(detail="Security method already defined", cause="Identical AEF Profile IDs") negotiated = return_negotiated_supp_feat_dict(service_security.supported_features) service_security.supported_features = negotiated["Final"] for service_instance in service_security.security_info: if service_instance.interface_details is not None: Loading Loading @@ -412,6 +427,9 @@ class SecurityOperations(Resource): mycol = self.db.get_col_by_name(self.db.security_info) try: negotiated_supported_features = return_negotiated_supp_feat_dict(service_security.supported_features) service_security.supported_features = negotiated_supported_features["Final"] current_app.logger.debug("Updating security context") result = self.__check_invoker(api_invoker_id) if result != None: Loading