diff --git a/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/auth_manager.py b/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/auth_manager.py index 0e0d7f126c4e965c7735362fc9b7689bbe339ebc..e331313de78d490dd4eeb2a2379be7d851715d6d 100644 --- a/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/auth_manager.py +++ b/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/auth_manager.py @@ -27,6 +27,10 @@ class AuthManager(Resource): auth_context = cert_col.find_one({"id":apf_id}) + # Service APIs published by an interconnected CCF have no certificate entry + if auth_context is None: + return + if "services" in auth_context["resources"]: if service_id in auth_context["resources"]["services"]: auth_context["resources"]["services"].remove(service_id) diff --git a/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/serviceapidescriptions.py b/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/serviceapidescriptions.py index 45486dfd49b3b123c910f86814664a9e17b8a4b4..ad148367ead77aa6b9c42498d69609a211acb0dc 100644 --- a/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/serviceapidescriptions.py +++ b/services/TS29222_CAPIF_Publish_Service_API/published_apis/core/serviceapidescriptions.py @@ -22,6 +22,11 @@ import encoder TOTAL_FEATURES = 10 SUPPORTED_FEATURES_HEX = "120" +# Seconds allowed for a single request towards a peer CCF +INTERCONNECTION_TIMEOUT = 30 +INTERCONNECTION_CERT = ('certs/server.crt', 'certs/server.key') +INTERCONNECTION_CA = 'certs/ca.crt' + publisher_ops = Publisher() @@ -53,6 +58,28 @@ def find_duplicate_service_by_api_name_and_aef( return collection.find_one(duplicate_query), aef_ids +def shared_capif_prov_doms(shareable_info): + """Return the CAPIF provider domains a service API is meant to be shared with.""" + shareable_info = shareable_info or {} + + if not shareable_info.get("is_shareable"): + return [] + + return shareable_info.get("capif_prov_doms") or [] + + +def interconnection_payload(service_api): + """Serialize a service API the way a peer CCF expects to receive it.""" + payload = { + key: value for key, value in service_api.items() + if key not in ("_id", "apf_id", "onboarding_date") + } + # The peer stores the service API as its own, it must not share it any further + payload["shareable_info"] = {"is_shareable": False} + + return json.dumps(clean_n_camel_case(payload), cls=encoder.CustomJSONEncoder) + + 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] @@ -188,46 +215,25 @@ class PublishServiceOperations(Resource): invalid_params=[{"param": "apiStatus", "reason": "defined but apiStatusMonitoring feature not active"}] ) - # 1. Check shareableInfo - if serviceapidescription.shareable_info.is_shareable: - interconnected_col = self.db.get_col_by_name(self.db.interconnected) - # 2. Iterate through capifProvDoms - for dom in serviceapidescription.shareable_info.capif_prov_doms: - # 3. For each domain, check if there is record in the "interconnected" table - interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": dom}) - # 4. If yes, make a publish request with CCF certificate - if interconnected_ccf: - config_col = self.db.get_col_by_name(self.db.capif_configuration) - config = config_col.find_one({}, {"_id": 0}) - ccf_id = config['ccf_id'] - serviceapidescription_dict['shareable_info'] = {"is_shareable": False} - - url = 'https://{}/published-apis/v1/{}/service-apis'.format(dom, ccf_id) - - headers = { - 'accept': 'application/json', - 'Content-Type': 'application/json' - } - - response = requests.request("POST", url, headers=headers, data=json.dumps(clean_n_camel_case(serviceapidescription_dict), cls=encoder.CustomJSONEncoder), cert=('certs/server.crt', 'certs/server.key'), verify='certs/ca.crt') - - if (response.status_code == 201 or response.status_code == 200): - if rec.get("pub_api_path") is None: - rec.update({"pub_api_path": {"ccf_ids": [interconnected_ccf.get("ccf_id")]}}) - else: - ccf_list = rec.get("pub_api_path").get("ccf_ids") - ccf_list.append(interconnected_ccf.get("ccf_id")) - rec.update({"pub_api_path": {"ccf_ids": ccf_list}}) - else: - rec['shareable_info']['capif_prov_doms'].remove(dom) + published, interconnection_error = self.publish_to_interconnected_ccfs( + rec, shared_capif_prov_doms(rec.get("shareable_info"))) + + if interconnection_error is not None: + # Nothing is stored here, so a copy left on a peer would be an orphan and + # would make every later attempt fail as a duplicate + self.withdraw_published_copies(rec.get("api_name"), published) + return interconnection_error + reached_ccf_ids = [ccf_id for _, ccf_id in published] + + # A publication coming from a peer CCF extends the path it travelled if "CCF" in apf_id: - if rec.get("pub_api_path") is None: - rec.update({"pub_api_path": {"ccf_ids": [apf_id]}}) - else: - ccf_list = rec.get("pub_api_path").get("ccf_ids") - ccf_list.append(apf_id) - rec.update({"pub_api_path": {"ccf_ids": ccf_list}}) + reached_ccf_ids.append(apf_id) + + if reached_ccf_ids: + ccf_ids = list((rec.get("pub_api_path") or {}).get("ccf_ids") or []) + ccf_ids.extend(ccf_id for ccf_id in reached_ccf_ids if ccf_id not in ccf_ids) + rec["pub_api_path"] = {"ccf_ids": ccf_ids} mycol.insert_one(rec) @@ -298,6 +304,279 @@ class PublishServiceOperations(Resource): current_app.logger.error(exception + "::" + str(e)) return internal_server_error(detail=exception, cause=str(e)) + def interconnection_service_apis_url(self, dom): + """Endpoint holding the service APIs this CCF published on the given domain.""" + config_col = self.db.get_col_by_name(self.db.capif_configuration) + config = config_col.find_one({}, {"_id": 0}) + + return 'https://{}/published-apis/v1/{}/service-apis'.format(dom, config['ccf_id']) + + def find_shared_service_api_id(self, dom, api_name, cause): + """Resolve the api_id a peer CCF assigned to its shared service API. + + A peer assigns its own api_id to the API it stores and that id is never kept + locally, so it is looked up by api_name out of the service APIs this CCF published + there. Returns None when the peer has not the specific API shared, plus an error response when the + peer could not be asked. + """ + url = self.interconnection_service_apis_url(dom) + + try: + response = requests.request("GET", url, headers={'accept': 'application/json'}, + cert=INTERCONNECTION_CERT, verify=INTERCONNECTION_CA, + timeout=INTERCONNECTION_TIMEOUT) + except requests.exceptions.RequestException as exc: + current_app.logger.error( + "Interconnection: listing service apis on {} failed: {}".format(dom, str(exc))) + return None, internal_server_error( + detail="Could not reach interconnected CCF {}".format(dom), cause=cause) + + try: + shared_apis = response.json() + except ValueError: + shared_apis = None + + if response.status_code != 200 or not isinstance(shared_apis, list): + current_app.logger.error( + "Interconnection: {} answered the service api listing with status {}".format( + dom, response.status_code)) + return None, internal_server_error( + detail="Interconnected CCF {} did not return its published service apis".format(dom), + cause=cause) + + # The listing is serialized in camel case by the peer + shared_api_id = next( + (shared_api.get("apiId") for shared_api in shared_apis + if isinstance(shared_api, dict) and shared_api.get("apiName") == api_name), + None) + + return shared_api_id, None + + def unpublish_from_interconnected_ccfs(self, api_name, capif_prov_doms): + """Unpublish a shared service API from every interconnected CCF holding a copy. + + Returns the ids of the CCFs that no longer hold a copy, and an error response if a + peer could not be reached or refused the removal. + """ + if not capif_prov_doms: + return [], None + + interconnected_col = self.db.get_col_by_name(self.db.interconnected) + cause = "Service API is still published on an interconnected CCF" + unpublished_ccf_ids = [] + + for dom in capif_prov_doms: + interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": dom}) + if interconnected_ccf is None: + continue + + shared_api_id, error = self.find_shared_service_api_id(dom, api_name, cause) + if error is not None: + return unpublished_ccf_ids, error + + if shared_api_id is None: + current_app.logger.warning( + "Interconnection: service api {} is not published on {}".format(api_name, dom)) + unpublished_ccf_ids.append(interconnected_ccf.get("ccf_id")) + continue + + url = '{}/{}'.format(self.interconnection_service_apis_url(dom), shared_api_id) + + try: + response = requests.request("DELETE", url, + headers={'accept': 'application/json'}, + cert=INTERCONNECTION_CERT, verify=INTERCONNECTION_CA, + timeout=INTERCONNECTION_TIMEOUT) + except requests.exceptions.RequestException as exc: + current_app.logger.error( + "Interconnection: unpublish from {} failed: {}".format(dom, str(exc))) + return unpublished_ccf_ids, internal_server_error( + detail="Could not reach interconnected CCF {}".format(dom), cause=cause) + + # 404 means the peer already dropped the specific API + if response.status_code not in (204, 404): + current_app.logger.error( + "Interconnection: {} refused to unpublish service api {} with status {}".format( + dom, shared_api_id, response.status_code)) + return unpublished_ccf_ids, internal_server_error( + detail="Interconnected CCF {} refused to unpublish the service api".format(dom), + cause=cause) + + unpublished_ccf_ids.append(interconnected_ccf.get("ccf_id")) + current_app.logger.debug( + "Interconnection: service api {} unpublished from {}".format(api_name, dom)) + + return unpublished_ccf_ids, None + + def update_on_interconnected_ccfs(self, shared_api_name, service_api, capif_prov_doms): + """Push the current definition of a service API to the CCFs already holding the shared API. + + The copies are looked up under the name they were shared with, which is the one + stored before this modification. Returns an error response if a peer could not be + reached or refused the update. + """ + if not capif_prov_doms: + return None + + interconnected_col = self.db.get_col_by_name(self.db.interconnected) + cause = "Service API could not be updated on an interconnected CCF" + headers = { + 'accept': 'application/json', + 'Content-Type': 'application/json' + } + body = interconnection_payload(service_api) + + for dom in capif_prov_doms: + if interconnected_col.find_one({"dst_prov_dom": dom}) is None: + continue + + shared_api_id, error = self.find_shared_service_api_id(dom, shared_api_name, cause) + if error is not None: + return error + + if shared_api_id is None: + current_app.logger.warning( + "Interconnection: service api {} is not published on {}, update skipped".format( + shared_api_name, dom)) + continue + + url = '{}/{}'.format(self.interconnection_service_apis_url(dom), shared_api_id) + + try: + response = requests.request("PUT", url, headers=headers, data=body, + cert=INTERCONNECTION_CERT, verify=INTERCONNECTION_CA, + timeout=INTERCONNECTION_TIMEOUT) + except requests.exceptions.RequestException as exc: + current_app.logger.error( + "Interconnection: update on {} failed: {}".format(dom, str(exc))) + return internal_server_error( + detail="Could not reach interconnected CCF {}".format(dom), cause=cause) + + if response.status_code not in (200, 204): + current_app.logger.error( + "Interconnection: {} refused to update service api {} with status {}".format( + dom, shared_api_id, response.status_code)) + return internal_server_error( + detail="Interconnected CCF {} refused to update the service api".format(dom), + cause=cause) + + current_app.logger.debug( + "Interconnection: service api {} updated on {}".format(shared_api_name, dom)) + + return None + + def publish_to_interconnected_ccfs(self, service_api, capif_prov_doms): + """Publish a shared service API to the interconnected CCF of every given domain. + + Returns the (domain, ccf_id) pairs that stored a copy, and an error response if a + peer could not be reached or refused the publication. The caller is expected to + withdraw the copies already accepted when an error comes back, as they would + otherwise be rejected as duplicates by a later attempt. + """ + if not capif_prov_doms: + return [], None + + interconnected_col = self.db.get_col_by_name(self.db.interconnected) + cause = "Service API could not be shared with an interconnected CCF" + headers = { + 'accept': 'application/json', + 'Content-Type': 'application/json' + } + body = interconnection_payload(service_api) + published = [] + + for dom in capif_prov_doms: + interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": dom}) + if interconnected_ccf is None: + current_app.logger.warning( + "Interconnection: {} is not interconnected, publication skipped".format(dom)) + continue + + url = self.interconnection_service_apis_url(dom) + + try: + response = requests.request("POST", url, headers=headers, data=body, + cert=INTERCONNECTION_CERT, verify=INTERCONNECTION_CA, + timeout=INTERCONNECTION_TIMEOUT) + except requests.exceptions.RequestException as exc: + current_app.logger.error( + "Interconnection: publish to {} failed: {}".format(dom, str(exc))) + return published, internal_server_error( + detail="Could not reach interconnected CCF {}".format(dom), cause=cause) + + if response.status_code not in (200, 201): + current_app.logger.error( + "Interconnection: {} rejected service api {} with status {}".format( + dom, service_api.get("api_name"), response.status_code)) + return published, internal_server_error( + detail="Interconnected CCF {} rejected the service api".format(dom), + cause=cause) + + published.append((dom, interconnected_ccf.get("ccf_id"))) + current_app.logger.debug( + "Interconnection: service api {} published on {}".format( + service_api.get("api_name"), dom)) + + return published, None + + def withdraw_published_copies(self, api_name, published): + """Undo the publications made before an interconnection procedure failed.""" + _, error = self.unpublish_from_interconnected_ccfs( + api_name, [dom for dom, _ in published]) + + if error is not None: + current_app.logger.error( + "Interconnection: service api {} could not be withdrawn from the CCFs it " + "reached before the failure".format(api_name)) + + def reconcile_interconnection_sharing(self, service_api, old_service_api): + """Align the interconnected CCFs holding the API with the modification requested. + + Domains dropped from the sharing list get the shared API withdrawn, which covers every + domain at once when sharing is turned off. Domains added to it receive the service + API, which covers every domain at once when sharing is turned on. Domains kept on + the list get the copy they already hold updated. Returns the publication path to + store and an error response if any peer could not be aligned, in which case the + caller must leave the service API untouched. + """ + old_doms = shared_capif_prov_doms(old_service_api.get("shareable_info")) + new_doms = shared_capif_prov_doms(service_api.get("shareable_info")) + pub_api_path = old_service_api.get("pub_api_path") + + if not old_doms and not new_doms: + return pub_api_path, None + + withdrawn_doms = [dom for dom in old_doms if dom not in new_doms] + kept_doms = [dom for dom in new_doms if dom in old_doms] + added_doms = [dom for dom in new_doms if dom not in old_doms] + + # Peers hold the sharedAPI under the name it carried when it was shared with them, so + # a modification renaming the service API must still be matched against the old one + shared_api_name = old_service_api.get("api_name") + + unpublished_ccf_ids, error = self.unpublish_from_interconnected_ccfs( + shared_api_name, withdrawn_doms) + if error is not None: + return pub_api_path, error + + error = self.update_on_interconnected_ccfs(shared_api_name, service_api, kept_doms) + if error is not None: + return pub_api_path, error + + # A fresh publication carries the new name, so that is what the copies are known + # as on the peers that just accepted them + published, error = self.publish_to_interconnected_ccfs(service_api, added_doms) + if error is not None: + self.withdraw_published_copies(service_api.get("api_name"), published) + return pub_api_path, error + + ccf_ids = [ccf_id for ccf_id in (pub_api_path or {}).get("ccf_ids") or [] + if ccf_id not in unpublished_ccf_ids] + ccf_ids.extend(ccf_id for _, ccf_id in published if ccf_id not in ccf_ids) + + # ccf_ids must hold at least one entry, so the whole attribute goes once emptied + return ({"ccf_ids": ccf_ids} if ccf_ids else None), None + def delete_serviceapidescription(self, service_api_id, apf_id): mycol = self.db.get_col_by_name(self.db.service_api_descriptions) @@ -327,9 +606,20 @@ class PublishServiceOperations(Resource): detail="Service API not existing", cause="Service API id not found") + # The shared APIs held by interconnected CCFs are removed first, so that when a peer cannot + # be reached, the service api remains published both on source and target CCF + # "_" contains the CCF ids where the API was successfully deleted + _, interconnection_error = self.unpublish_from_interconnected_ccfs( + serviceapidescription_dict.get("api_name"), + shared_capif_prov_doms(serviceapidescription_dict.get("shareable_info"))) + + if interconnection_error is not None: + return interconnection_error + mycol.delete_one(my_query) - self.auth_manager.remove_auth_service(service_api_id, apf_id) + if "APF" in apf_id: + self.auth_manager.remove_auth_service(service_api_id, apf_id) current_app.logger.info("Removed service from database") out = "The service matching api_id " + service_api_id + " was deleted." @@ -425,6 +715,20 @@ class PublishServiceOperations(Resource): invalid_params=[{"param": "apiStatus", "reason": "defined but apiStatusMoniroting feature not active"}] ) + # The publication path records where the CCFs holding the API live, so this CCF owns it and + # the one held in the request is discarded + service_api_description["pub_api_path"] = serviceapidescription_old.get("pub_api_path") + + # The peers are aligned before the local write, so that one that cannot be + # reached leaves the sharing settings as they were instead of drifting apart + pub_api_path, interconnection_error = self.reconcile_interconnection_sharing( + service_api_description, serviceapidescription_old) + + if interconnection_error is not None: + return interconnection_error + + service_api_description["pub_api_path"] = pub_api_path + result = mycol.find_one_and_replace( serviceapidescription_old, service_api_description, @@ -537,6 +841,20 @@ class PublishServiceOperations(Resource): invalid_params=[{"param": "apiStatus", "reason": "defined but apiStatusMoniroting feature not active"}] ) + # A $set merges the patch into the stored service API, so the peers are + # aligned against that same result. Attributes the patch leaves out keep their + # stored value, which for the sharing settings means no peer is touched + patched_service_api = dict(serviceapidescription_old) + patched_service_api.update(patch_service_api_description) + + pub_api_path, interconnection_error = self.reconcile_interconnection_sharing( + patched_service_api, serviceapidescription_old) + + if interconnection_error is not None: + return interconnection_error + + patch_service_api_description["pub_api_path"] = pub_api_path + result = mycol.find_one_and_update( my_query, {"$set": patch_service_api_description},