move invoker operations into core/invoker_ops.py (b0343e8f) · Commits · OCF / backoffice / backend

capif_backend/backoffice_service/controllers/me_controller.py

+3 −4

Changes for capif_backend/backoffice_service/controllers/me_controller.py: 3 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
		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

		@me_routes.route("/invoker", methods=["POST"])
		@jwt_required()
		@@ -19,7 +18,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 +54,7 @@ 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
		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