Add sync API, sync 2 CCFs after interconnection established (8a9bb7ac) · Commits · OCF / capif

services/helper/helper_service/services/interconnection/controllers/default_controller.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -81,7 +81,7 @@ def interconnect_sync(body): # noqa: E501
		if connexion.request.is_json:
		capif_domain_details = CapifDomainDetails.from_dict(connexion.request.get_json()) # noqa: E501

		res = capif_domain_operations.sync_capifdomain(capif_domain_details)
		res = ccf_operations.sync_ccfinstance(capif_domain_details)

		return res

services/helper/helper_service/services/interconnection/core/capifdomaindetails.py

+3 −4

Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@ from ..models.ccf_instance_details import CcfInstanceDetails
		from ..util import clean_empty, clean_n_camel_case, dict_to_camel_case, serialize_clean_camel_case
		# from ..vendor_specific import add_vend_spec_fields
		from .auth_manager import AuthManager
		from .ccfinstancedetails import CcfInstanceOperations
		from .publisher import Publisher
		from .redis_event import RedisEvent
		from .resources import Resource
		@@ -36,6 +37,7 @@ class CapifDomainOperations(Resource):
		Resource.__init__(self)
		self.auth_manager = AuthManager()
		self.db = get_mongo()
		self.ccf_operations = CcfInstanceOperations()

		def add_capifdomain(self, capifdomaindetails):
		current_app.logger.debug("Interconnection: Add domain")
		@@ -85,6 +87,7 @@ class CapifDomainOperations(Resource):
		if response.status_code in (200, 201):
		if isinstance(inter_ccf, dict):
		interconnected_col.insert_one(inter_ccf)
		self.ccf_operations.sync_with_ccf(capifdomaindetails.dst_prov_dom)
		ccfinstancedetails_new = CcfInstanceDetails().from_dict(dict_to_camel_case(inter_ccf))
		return make_response(object=ccfinstancedetails_new, status=response.status_code)

		@@ -113,10 +116,6 @@ class CapifDomainOperations(Resource):

		return result.modified_count

		def sync_capifdomain(self, capifdomaindetails):
		"""Sync interconnection state with a peer CCF."""
		pass

		def delete_capifdomain(self, ccf_id):
		"""Remove an interconnection request for a CCF."""
		current_app.logger.debug("Interconnection: Delete domain")

services/helper/helper_service/services/interconnection/core/ccfinstancedetails.py

+124 −2

Original line number	Diff line number	Diff line
		import os
		import secrets
		from datetime import datetime
		import requests
		import json

		from flask import current_app
		from flask import current_app, Response
		# from pymongo import ReturnDocument

		from db.db import get_mongo
		from ..encoder import CustomJSONEncoder
		from ..models.capif_domain_details import CapifDomainDetails
		from ..models.ccf_instance_details import CcfInstanceDetails
		from ..util import clean_empty, clean_n_camel_case, dict_to_camel_case, serialize_clean
		from ..util import clean_empty, clean_n_camel_case, dict_to_camel_case, serialize_clean, serialize_clean_camel_case
		# from ..vendor_specific import add_vend_spec_fields
		from .auth_manager import AuthManager
		from .publisher import Publisher
		@@ -72,6 +76,124 @@ class CcfInstanceOperations(Resource):
		res = make_response(object=serialize_clean(ccfinstancedetails_new), status=status)
		return res

		def local_ccf_identity(self):
		"""Return the CCF identifier and the provider domain of this CAPIF instance."""
		config_col = self.db.get_col_by_name(self.db.capif_configuration)
		config = config_col.find_one({}, {"_id": 0})
		dst_prov_dom = "{}:{}".format(os.getenv("CAPIF_HOSTNAME"), os.getenv("CAPIF_HTTPS_PORT"))

		return config['ccf_id'], dst_prov_dom

		def add_ccf_to_pub_api_path(self, services_col, service_api, ccf_id):
		"""Record that a service API is from now on also published through the given CCF."""
		if service_api.get("pub_api_path") is None:
		update = {"$set": {"pub_api_path": {"ccf_ids": [ccf_id]}}}
		else:
		update = {"$addToSet": {"pub_api_path.ccf_ids": ccf_id}}

		services_col.update_one({"_id": service_api["_id"]}, update)

		def publish_shared_apis(self, peer_dom, peer_ccf_id, local_ccf_id):
		"""Publish to a peer CCF every local service API that is shareable with its domain."""
		services_col = self.db.get_col_by_name(self.db.services_col)
		shareable_apis = services_col.find({
		"shareable_info.is_shareable": True,
		"shareable_info.capif_prov_doms": peer_dom,
		"apf_id": {"$ne": peer_ccf_id},
		"pub_api_path.ccf_ids": {"$ne": peer_ccf_id}
		})

		url = "https://{}/published-apis/v1/{}/service-apis".format(peer_dom, local_ccf_id)
		headers = {
		'accept': 'application/json',
		'Content-Type': 'application/json'
		}

		published = 0
		for service_api in shareable_apis:
		payload = {
		key: value for key, value in service_api.items()
		if key not in ("_id", "apf_id", "onboarding_date")
		}
		# The peer stores the API as its own, it must not share it any further
		payload['shareable_info'] = {"is_shareable": False}

		try:
		response = requests.request("POST", url, headers=headers,
		data=json.dumps(clean_n_camel_case(payload), cls=CustomJSONEncoder),
		cert=('certs/server.crt', 'certs/server.key'),
		verify='certs/ca.crt')
		except requests.exceptions.RequestException as exc:
		current_app.logger.error("Interconnection: publish to {} failed: {}".format(peer_dom, str(exc)))
		continue

		if response.status_code not in (200, 201):
		current_app.logger.warning("Interconnection: {} rejected service api {} with status {}".format(
		peer_dom, service_api.get("api_name"), response.status_code))
		continue

		self.add_ccf_to_pub_api_path(services_col, service_api, peer_ccf_id)
		published += 1

		current_app.logger.debug("Interconnection: {} service apis shared with {}".format(published, peer_dom))

		return published

		def sync_ccfinstance(self, capifdomaindetails):
		"""Share the local service APIs with the peer CCF asking for a sync."""
		current_app.logger.debug("Interconnection: Sync instance")

		peer_dom = capifdomaindetails.dst_prov_dom

		interconnected_col = self.db.get_col_by_name(self.db.interconnected)
		interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": peer_dom})
		if interconnected_ccf is None:
		return not_found_error(
		detail="CAPIF domain {} is not interconnected".format(peer_dom),
		cause="NOT_FOUND")

		local_ccf_id, local_dom = self.local_ccf_identity()
		self.publish_shared_apis(peer_dom, interconnected_ccf.get("ccf_id"), local_ccf_id)

		capifdomaindetails_new = CapifDomainDetails(dst_prov_dom=local_dom)

		return make_response(object=serialize_clean_camel_case(capifdomaindetails_new), status=201)

		def sync_with_ccf(self, peer_dom):
		"""Ask a peer CCF for a sync and share the local service APIs back once it answers."""
		current_app.logger.debug("Interconnection: Sync with {}".format(peer_dom))

		interconnected_col = self.db.get_col_by_name(self.db.interconnected)
		interconnected_ccf = interconnected_col.find_one({"dst_prov_dom": peer_dom})
		if interconnected_ccf is None:
		current_app.logger.warning("Interconnection: {} is not interconnected, sync skipped".format(peer_dom))
		return 0

		local_ccf_id, local_dom = self.local_ccf_identity()

		url = "https://{}/helper/interconnection/sync".format(peer_dom)
		headers = {
		'accept': 'application/json',
		'Content-Type': 'application/json'
		}
		payload = json.dumps(serialize_clean_camel_case(CapifDomainDetails(dst_prov_dom=local_dom)))

		try:
		response = requests.request("POST", url, headers=headers, data=payload,
		cert=('certs/server.crt', 'certs/server.key'),
		verify='certs/ca.crt')
		except requests.exceptions.RequestException as exc:
		current_app.logger.error("Interconnection: sync request to {} failed: {}".format(peer_dom, str(exc)))
		return 0

		if response.status_code not in (200, 201):
		current_app.logger.warning("Interconnection: {} answered sync with status {}".format(
		peer_dom, response.status_code))
		return 0

		# The peer shared its own APIs while serving the request, now mirror the procedure
		return self.publish_shared_apis(peer_dom, interconnected_ccf.get("ccf_id"), local_ccf_id)

		def remove_ccf_from_pub_api_path(self, services_col, ccf_id):
		"""Drop a CCF from the publication path of every service API that references it."""