Onboard providers for the logged-in user (!7) · Merge requests · OCF / backoffice / backend

capif_backend/backoffice_service/controllers/me_controller.py

+20 −4

Changes for capif_backend/backoffice_service/controllers/me_controller.py: 20 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		from flask import Blueprint, jsonify, request
		from flask_jwt_extended import jwt_required

		from ..core.operations import Operations
		from ..core import invoker_ops, provider_ops
		from ..utils.utils import get_capif_token, get_capif_urls, capif_user_required
		from ..utils.vault import read_invoker, list_invokers, VaultError

		me_routes = Blueprint("me_routes", __name__)
		operations = Operations() # create operations object -> operations.py

		#----------------------------------------- INVOKER -----------------------------------------#

		@me_routes.route("/invoker", methods=["POST"])
		@jwt_required()
		@@ -19,7 +20,7 @@ def onboard_invoker(user_uuid):
		return jsonify({"error": "apiInvokerInformation and notificationDestination are required."}), 400

		ccf_onboarding_url = get_capif_urls()["ccf_onboarding_url"]
		data, status = operations.onboard_invoker(
		data, status = invoker_ops.onboard(
		get_capif_token(), user_uuid, ccf_onboarding_url, invoker_info, notification_destination
		)
		return jsonify(data), status
		@@ -55,7 +56,22 @@ def discover_apis(api_invoker_id, user_uuid):
		return jsonify({"error": "Invoker not found."}), 404

		ccf_discover_url = get_capif_urls()["ccf_discover_url"]
		data, status = operations.discover_apis(
		data, status = invoker_ops.discover(
		ccf_discover_url, api_invoker_id, creds["certificate"], creds["private_key"]
		)
		return jsonify(data), status

		#----------------------------------------- PROVIDER -----------------------------------------#

		@me_routes.route("/provider", methods=["POST"])
		@jwt_required()
		@capif_user_required
		def onboard_provider(user_uuid):
		payload = request.get_json(silent=True) or {}
		prov_dom_info = payload.get("apiProvDomInfo")
		if not prov_dom_info:
		return jsonify({"error": "apiProvDomInfo is required."}), 400

		ccf_provider_onboarding_url = get_capif_urls()["ccf_api_onboarding_url"]
		data, status = provider_ops.onboard(get_capif_token(), user_uuid, ccf_provider_onboarding_url, prov_dom_info)
		return jsonify(data), status
		No newline at end of file

capif_backend/backoffice_service/core/invoker_ops.py

0 → 100644

+120 −0

Changes for capif_backend/backoffice_service/core/invoker_ops.py: 120 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""User-scoped invoker operations against CCF: onboarding (with Vault storage and rollback) and discover."""
		import os

		import requests

		from ..utils.certs import generate_keypair_and_csr, mtls_files, SecureTempUnavailable
		from ..utils.vault import store_invoker

		# CCF (OpenCAPIF) answers discover with 404 when nothing matches, instead of 200 with an empty list.
		# We match its `cause` text deliberately: `detail` embeds the invoker id and would never match.
		# If CCF rewords this cause, the empty case surfaces as an error instead of silently hiding a
		# missing invoker; update the string here.
		_DISCOVER_NO_APIS_CAUSE = "No API Published accomplish filter conditions"

		def onboard(capif_token, user_uuid, onboarding_url, invoker_info, notification_destination):
		"""Onboard an API invoker in CCF on behalf of the logged-in user,
		and store its private key and certificate in Vault"""
		csr_request, private_key = generate_keypair_and_csr()

		body = {
		"apiInvokerInformation": invoker_info,
		"notificationDestination": notification_destination,
		"requestTestNotification": False,
		"onboardingInformation": {"apiInvokerPublicKey": csr_request.decode("utf-8")},
		"supportedFeatures": "fffffff", # TODO: placeholder from capif-client, needs real value
		"websockNotifConfig": {"requestWebsocketUri": True, "websocketUri": "websocketUri"}, # TODO: placeholder
		}

		url = "https://{}/{}".format(os.getenv('CAPIF_HOSTNAME'), onboarding_url)

		try:
		response = requests.post(
		url,
		json=body,
		headers={"Authorization": f"Bearer {capif_token}"},
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: could not reach CCF for invoker onboarding: {e}")
		return {"error": "Could not reach CCF."}, 502

		try:
		data = response.json()
		except ValueError:
		data = {"error": response.text}

		if response.status_code != 201:
		return {"error": "CCF onboarding failed.", "ccf_status": response.status_code, "ccf_response": data}, 502

		invoker_id = data.get("apiInvokerId")
		certificate = (data.get("onboardingInformation") or {}).get("apiInvokerCertificate")

		if not invoker_id or not certificate:
		print(f"ERROR: unexpected CCF onboarding response shape: {list(data.keys())}")
		if invoker_id:
		_rollback(invoker_id)
		return {"error": "CCF returned an unexpected onboarding response."}, 502

		if not store_invoker(user_uuid, invoker_id, private_key.decode("utf-8"), certificate):
		if _rollback(invoker_id):
		print(f"WARNING: invoker {invoker_id} rolled back in CCF after Vault write failure")
		else:
		print(f"ERROR: invoker {invoker_id} exists in CCF without stored credentials; admin cleanup needed")
		return {"error": "Onboarding failed: the invoker credentials could not be stored."}, 500

		return {"apiInvokerId": invoker_id}, 201

		def _rollback(invoker_id):
		"""Delete an invoker in CCF. Returns True on success."""
		url = "https://{}/api-invoker-management/v1/onboardedInvokers/{}".format(
		os.getenv('CAPIF_HOSTNAME'), invoker_id)
		try:
		response = requests.delete(
		url,
		cert=("newcerts/superadmin.crt", "newcerts/superadmin.key"),
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: rollback of invoker {invoker_id} failed: {e}")
		return False
		if response.status_code != 204:
		print(f"ERROR: rollback of invoker {invoker_id} failed: {response.status_code} {response.text}")
		return False
		return True

		def discover(discover_url, invoker_id, certificate, private_key):
		"""Discover APIs visible to an invoker. Authenticates to CCF with the invoker's own
		mTLS certificate only."""
		url = f"https://{os.getenv('CAPIF_HOSTNAME')}/{discover_url}{invoker_id}"

		try:
		with mtls_files(certificate, private_key) as (cert_path, key_path):
		response = requests.get(
		url,
		cert=(cert_path, key_path),
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except SecureTempUnavailable as e:
		print(f"ERROR: {e}")
		return {"error": "Secure temporary storage is unavailable."}, 500
		except requests.exceptions.RequestException as e:
		print(f"ERROR: could not reach CCF for discover (invoker {invoker_id}): {e}")
		return {"error": "Could not reach CCF."}, 502

		try:
		data = response.json()
		except ValueError:
		data = {"error": response.text}

		if (response.status_code == 404 and isinstance(data, dict) and data.get("cause") == _DISCOVER_NO_APIS_CAUSE):
		return {"serviceAPIDescriptions": []}, 200

		if response.status_code != 200:
		print(f"ERROR: CCF discover failed for invoker {invoker_id}: {response.status_code} {data}")
		return {"error": "CCF discover failed.", "ccf_status": response.status_code, "ccf_response": data}, 502

		return data, 200

capif_backend/backoffice_service/core/operations.py

+0 −111

Changes for capif_backend/backoffice_service/core/operations.py: 0 added lines, 111 removed lines.
Original line number	Diff line number	Diff line
		@@ -10,10 +10,7 @@ from urllib.parse import urlencode

		# from ..configs.config import Config
		from ..config import Config
		from ..utils.certs import generate_keypair_and_csr, mtls_files, SecureTempUnavailable
		from ..utils.vault import store_invoker

		_DISCOVER_NO_APIS_CAUSE = "No API Published accomplish filter conditions"

		class Operations:

		@@ -484,79 +481,6 @@ class Operations:
		print(f"Error: {str(e)}") # Log de error
		return {"error": str(e)}

		def onboard_invoker(self, capif_token, user_uuid, onboarding_url, invoker_info, notification_destination):
		"""Onboard an API invoker in CCF on behalf of the logged-in user,
		and store its private key and certificate in Vault"""
		csr_request, private_key = generate_keypair_and_csr()

		body = {
		"apiInvokerInformation": invoker_info,
		"notificationDestination": notification_destination,
		"requestTestNotification": False,
		"onboardingInformation": {"apiInvokerPublicKey": csr_request.decode("utf-8")},
		"supportedFeatures": "fffffff", # TODO: placeholder from capif-client, needs real value
		"websockNotifConfig": {"requestWebsocketUri": True, "websocketUri": "websocketUri"}, # TODO: placeholder
		}

		url = "https://{}/{}".format(os.getenv('CAPIF_HOSTNAME'), onboarding_url)

		try:
		response = requests.post(
		url,
		json=body,
		headers={"Authorization": f"Bearer {capif_token}"},
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: could not reach CCF for invoker onboarding: {e}")
		return {"error": "Could not reach CCF."}, 502

		try:
		data = response.json()
		except ValueError:
		data = {"error": response.text}

		if response.status_code != 201:
		return {"error": "CCF onboarding failed.", "ccf_status": response.status_code, "ccf_response": data}, 502

		invoker_id = data.get("apiInvokerId")
		certificate = (data.get("onboardingInformation") or {}).get("apiInvokerCertificate")

		if not invoker_id or not certificate:
		print(f"ERROR: unexpected CCF onboarding response shape: {list(data.keys())}")
		if invoker_id:
		self._rollback_invoker(invoker_id)
		return {"error": "CCF returned an unexpected onboarding response."}, 502

		if not store_invoker(user_uuid, invoker_id, private_key.decode("utf-8"), certificate):
		if self._rollback_invoker(invoker_id):
		print(f"WARNING: invoker {invoker_id} rolled back in CCF after Vault write failure")
		else:
		print(f"ERROR: invoker {invoker_id} exists in CCF without stored credentials; admin cleanup needed")
		return {"error": "Onboarding failed: the invoker credentials could not be stored."}, 500

		return {"apiInvokerId": invoker_id}, 201

		def _rollback_invoker(self, invoker_id):
		"""Delete an invoker in CCF. Returns True on success."""
		url = "https://{}/api-invoker-management/v1/onboardedInvokers/{}".format(
		os.getenv('CAPIF_HOSTNAME'), invoker_id)
		try:
		response = requests.delete(
		url,
		cert=("newcerts/superadmin.crt", "newcerts/superadmin.key"),
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: rollback of invoker {invoker_id} failed: {e}")
		return False
		if response.status_code != 204:
		print(f"ERROR: rollback of invoker {invoker_id} failed: {response.status_code} {response.text}")
		return False
		return True

		def get_invokers_of_a_user(self, uuid):
		# Construir la URL base
		base_url = "https://{}/helper/api/getInvokers".format(os.getenv('CAPIF_HOSTNAME'))
		@@ -1351,7 +1275,6 @@ class Operations:
		except requests.exceptions.RequestException as e:
		return jsonify({"error": f"Error when adding category: {str(e)}"}), 500


		def add_new_config_register_setting(self, param_path, new_value):
		"""Add a new parameter in a category within 'settings' in register."""
		def login():
		@@ -1499,37 +1422,3 @@ class Operations:

		except requests.exceptions.RequestException as e:
		return jsonify({"error": f"Error deleting configuration category: {str(e)}"}), 500

		def discover_apis(self, discover_url, invoker_id, certificate, private_key):
		"""Discover APIs visible to an invoker. Authenticates to CCF with the invoker's own
		mTLS certificate only."""
		url = f"https://{os.getenv('CAPIF_HOSTNAME')}/{discover_url}{invoker_id}"

		try:
		with mtls_files(certificate, private_key) as (cert_path, key_path):
		response = requests.get(
		url,
		cert=(cert_path, key_path),
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except SecureTempUnavailable as e:
		print(f"ERROR: {e}")
		return {"error": "Secure temporary storage is unavailable."}, 500
		except requests.exceptions.RequestException as e:
		print(f"ERROR: could not reach CCF for discover (invoker {invoker_id}): {e}")
		return {"error": "Could not reach CCF."}, 502

		try:
		data = response.json()
		except ValueError:
		data = {"error": response.text}

		if (response.status_code == 404 and isinstance(data, dict) and data.get("cause") == _DISCOVER_NO_APIS_CAUSE):
		return {"serviceAPIDescriptions": []}, 200

		if response.status_code != 200:
		print(f"ERROR: CCF discover failed for invoker {invoker_id}: {response.status_code} {data}")
		return {"error": "CCF discover failed.", "ccf_status": response.status_code, "ccf_response": data}, 502

		return data, 200
		No newline at end of file

capif_backend/backoffice_service/core/provider_ops.py

0 → 100644

+125 −0

Changes for capif_backend/backoffice_service/core/provider_ops.py: 125 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""User-scoped provider operations against CCF: onboarding (with Vault storage and rollback)."""
		import os

		import requests

		from ..utils.certs import generate_keypair_and_csr
		from ..utils.vault import store_provider

		PROVIDER_ROLES = ("AEF", "APF", "AMF")

		_SUPP_FEAT = "fff" # TODO: placeholder from capif-client, needs real value
		_FAIL_REASON = "string" # TODO: placeholder from capif-client, needs real value


		def onboard(capif_token, user_uuid, onboarding_url, prov_dom_info):
		"""Register an API provider in CCF on behalf of the logged-in user, with one key pair and
		CSR per function (AEF, APF, AMF), and store all three in a single Vault secret."""
		private_keys = {}
		api_prov_funcs = []
		for role in PROVIDER_ROLES:
		csr_request, private_key = generate_keypair_and_csr()
		private_keys[role] = private_key.decode("utf-8")
		api_prov_funcs.append({
		"apiProvFuncRole": role,
		"apiProvFuncInfo": "",
		"regInfo": {"apiProvPubKey": csr_request.decode("utf-8")},
		})

		body = {
		"apiProvFuncs": api_prov_funcs,
		"apiProvDomInfo": prov_dom_info,
		"suppFeat": _SUPP_FEAT,
		"failReason": _FAIL_REASON,
		"regSec": capif_token,
		}

		url = "https://{}/{}".format(os.getenv('CAPIF_HOSTNAME'), onboarding_url)

		try:
		response = requests.post(
		url,
		json=body,
		headers={"Authorization": f"Bearer {capif_token}"},
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: could not reach CCF for provider onboarding: {e}")
		return {"error": "Could not reach CCF."}, 502

		try:
		data = response.json()
		except ValueError:
		data = {"error": response.text}

		if response.status_code != 201:
		return {"error": "CCF provider onboarding failed.", "ccf_status": response.status_code, "ccf_response": data}, 502

		prov_dom_id = data.get("apiProvDomId")
		functions = _match_functions(data.get("apiProvFuncs"), private_keys)

		if not prov_dom_id or functions is None:
		print(f"ERROR: unexpected CCF provider onboarding response shape: {list(data.keys())}")
		if prov_dom_id:
		_rollback(prov_dom_id)
		return {"error": "CCF returned an unexpected provider onboarding response."}, 502

		if not store_provider(user_uuid, prov_dom_id, functions):
		if _rollback(prov_dom_id):
		print(f"WARNING: provider {prov_dom_id} rolled back in CCF after Vault write failure")
		else:
		print(f"ERROR: provider {prov_dom_id} exists in CCF without stored credentials; admin cleanup needed")
		return {"error": "Onboarding failed: the provider credentials could not be stored."}, 500

		return {"apiProvDomId": prov_dom_id}, 201


		def _match_functions(returned_funcs, private_keys):
		"""Pair each function CCF returned with the private key generated for its role.

		Matches on apiProvFuncRole, never on array order: a key paired with another role's
		certificate would fail only later, at the first mTLS call. Returns None unless every
		role comes back exactly once with an id and a certificate.
		"""
		if not isinstance(returned_funcs, list):
		return None

		functions = {}
		for func in returned_funcs:
		if not isinstance(func, dict):
		return None
		role = func.get("apiProvFuncRole")
		func_id = func.get("apiProvFuncId")
		certificate = (func.get("regInfo") or {}).get("apiProvCert")
		if role not in private_keys or role.lower() in functions or not func_id or not certificate:
		return None
		functions[role.lower()] = {
		"api_prov_func_id": func_id,
		"private_key": private_keys[role],
		"certificate": certificate,
		}

		if len(functions) != len(private_keys):
		return None
		return functions


		def _rollback(prov_dom_id):
		"""Best-effort delete of a provider in CCF (superadmin mTLS). Returns True on success."""
		url = "https://{}/api-provider-management/v1/registrations/{}".format(
		os.getenv('CAPIF_HOSTNAME'), prov_dom_id)
		try:
		response = requests.delete(
		url,
		cert=("newcerts/superadmin.crt", "newcerts/superadmin.key"),
		verify="newcerts/ca_root.crt",
		timeout=30,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: rollback of provider {prov_dom_id} failed: {e}")
		return False
		if response.status_code != 204:
		print(f"ERROR: rollback of provider {prov_dom_id} failed: {response.status_code} {response.text}")
		return False
		return True
		No newline at end of file

capif_backend/backoffice_service/utils/vault.py

+25 −0

Changes for capif_backend/backoffice_service/utils/vault.py: 25 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -51,6 +51,31 @@ def store_invoker(user_uuid, invoker_id, private_key, certificate):

		return True

		def store_provider(user_uuid, prov_dom_id, functions):
		"""Store a provider's three functions (aef, apf, amf), each with its id, private key and
		certificate, in ONE Vault KV v2 secret. One secret, not three, so a provider can never be
		half-stored. Returns True on success, False otherwise."""
		path = f"secret/data/backoffice/users/{user_uuid}/providers/{prov_dom_id}"
		payload = {"data": {"api_prov_dom_id": prov_dom_id, "functions": functions}}

		try:
		response = requests.post(
		f"{vault_addr()}/v1/{path}",
		headers=vault_headers(),
		json=payload,
		verify=vault_verify(),
		timeout=10,
		)
		except requests.exceptions.RequestException as e:
		print(f"ERROR: Vault write failed for provider {prov_dom_id}: {e}")
		return False

		if not response.ok:
		print(f"ERROR: Vault write failed for provider {prov_dom_id}: {response.status_code} {response.text}")
		return False

		return True

		def read_invoker(user_uuid, invoker_id):
		"""Read an invoker's stored credentials from Vault KV v2.

Onboard providers for the logged-in user

Merge request reports