Implement interconneciton procedures for PUT and PATCH of Publish API (4eeb4471) · Commits · OCF / capif

services/TS29222_CAPIF_Publish_Service_API/published_apis/core/serviceapidescriptions.py

+309 −91

Original line number	Diff line number	Diff line
		@@ -24,6 +24,8 @@ 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()

		@@ -56,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]

		@@ -191,46 +215,25 @@ class PublishServiceOperations(Resource):
		invalid_params=[{"param": "apiStatus", "reason": "defined but apiStatusMonitoring feature not active"}]
		)

		# 1. Check shareableInfo
		if serviceapidescription.shareable_info and 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 or []:
		# 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'
		}
		published, interconnection_error = self.publish_to_interconnected_ccfs(
		rec, shared_capif_prov_doms(rec.get("shareable_info")))

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

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

		@@ -301,39 +304,32 @@ class PublishServiceOperations(Resource):
		current_app.logger.error(exception + "::" + str(e))
		return internal_server_error(detail=exception, cause=str(e))

		def unpublish_from_interconnected_ccfs(self, api_name, capif_prov_doms):
		"""Unpublish a shared service API from every interconnected CCF holding a copy.

		A peer assigns its own api_id to the copy it stores, and that id is never kept
		locally, so it is resolved by api_name out of the service APIs this CCF published
		there. Returns an error response if a peer could not be reached or refused the
		removal, None if every copy is gone.
		"""
		interconnected_col = self.db.get_col_by_name(self.db.interconnected)
		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})
		ccf_id = config['ccf_id']

		headers = {'accept': 'application/json'}
		certs = ('certs/server.crt', 'certs/server.key')
		return 'https://{}/published-apis/v1/{}/service-apis'.format(dom, config['ccf_id'])

		for dom in capif_prov_doms:
		interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": dom})
		if interconnected_ccf is None:
		continue
		def find_shared_service_api_id(self, dom, api_name, cause):
		"""Resolve the api_id a peer CCF assigned to its copy of a service API.

		url = 'https://{}/published-apis/v1/{}/service-apis'.format(dom, ccf_id)
		A peer assigns its own api_id to the copy 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 holds no copy, 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=headers, cert=certs,
		verify='certs/ca.crt',
		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 internal_server_error(
		detail="Could not reach interconnected CCF {}".format(dom),
		cause="Service API is still published on an interconnected CCF")
		return None, internal_server_error(
		detail="Could not reach interconnected CCF {}".format(dom), cause=cause)

		try:
		shared_apis = response.json()
		@@ -344,9 +340,9 @@ class PublishServiceOperations(Resource):
		current_app.logger.error(
		"Interconnection: {} answered the service api listing with status {}".format(
		dom, response.status_code))
		return internal_server_error(
		return None, internal_server_error(
		detail="Interconnected CCF {} did not return its published service apis".format(dom),
		cause="Service API is still published on an interconnected CCF")
		cause=cause)

		# The listing is serialized in camel case by the peer
		shared_api_id = next(
		@@ -354,37 +350,233 @@ class PublishServiceOperations(Resource):
		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", '{}/{}'.format(url, shared_api_id),
		headers=headers, cert=certs,
		verify='certs/ca.crt',
		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 internal_server_error(
		detail="Could not reach interconnected CCF {}".format(dom),
		cause="Service API is still published on an interconnected CCF")
		return unpublished_ccf_ids, internal_server_error(
		detail="Could not reach interconnected CCF {}".format(dom), cause=cause)

		# 404 means the peer already dropped its copy
		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 internal_server_error(
		return unpublished_ccf_ids, internal_server_error(
		detail="Interconnected CCF {} refused to unpublish the service api".format(dom),
		cause="Service API is still published on an interconnected CCF")
		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 a copy.

		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 a copy with the modification requested.

		Domains dropped from the sharing list get the copy 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 copy 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)
		@@ -416,11 +608,9 @@ class PublishServiceOperations(Resource):

		# The copies held by interconnected CCFs go first, so that a peer that cannot
		# be reached leaves the service api published here instead of orphaned there
		shareable_info = serviceapidescription_dict.get("shareable_info") or {}
		if shareable_info.get("is_shareable"):
		interconnection_error = self.unpublish_from_interconnected_ccfs(
		_, interconnection_error = self.unpublish_from_interconnected_ccfs(
		serviceapidescription_dict.get("api_name"),
		shareable_info.get("capif_prov_doms") or [])
		shared_capif_prov_doms(serviceapidescription_dict.get("shareable_info")))

		if interconnection_error is not None:
		return interconnection_error
		@@ -524,6 +714,20 @@ class PublishServiceOperations(Resource):
		invalid_params=[{"param": "apiStatus", "reason": "defined but apiStatusMoniroting feature not active"}]
		)

		# The publication path records where the copies 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,
		@@ -636,6 +840,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},