Discover APIs for the user's invoker over mTLS

from ..core.operations import Operations

from ..utils.utils import get_capif_token, get_capif_urls, get_capif_uuid

from ..utils.vault import read_invoker, VaultError

me_routes = Blueprint("me_routes", __name__)

operations = Operations()   # create operations object -> operations.py

        capif_token, user_uuid, ccf_onboarding_url, invoker_info, notification_destination

    )

    return jsonify(data), status

@me_routes.route("/invokers/<api_invoker_id>/apis", methods=["GET"])

@jwt_required()

def discover_apis(api_invoker_id):

    if not get_capif_token():

        return jsonify({"error": "This account has no CAPIF credentials"}), 403

    user_uuid = get_capif_uuid()

    if not user_uuid:

        return jsonify({"error": "Could not determine the CAPIF user id for this account"}), 500

    # Authorization by path design: the Vault path is keyed by the uuid from the JWT,

    # never from the URL. If a user passes another user's apiInvokerId, the lookup runs

    # under their own uuid, finds nothing, and returns 404. Do not flatten the Vault path

    # (e.g. to .../invokers/{id}) without adding an explicit ownership check, or this

    # becomes an IDOR.

    try:

        creds = read_invoker(user_uuid, api_invoker_id)

    except VaultError as e:

        print(f"ERROR: {e}")

        return jsonify({"error": "Could not read invoker credentials."}), 500

    if creds is None:

        return jsonify({"error": "Invoker not found."}), 404

    ccf_discover_url = get_capif_urls()["ccf_discover_url"]

    data, status = operations.discover_apis(

        ccf_discover_url, api_invoker_id, creds["certificate"], creds["private_key"]

    )

    return jsonify(data), status

 No newline at end of file

# from ..configs.config import Config

from ..config import Config

from ..utils.certs import generate_keypair_and_csr

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:

    def __init__(self):

                timeout=30,

            )

        except requests.exceptions.RequestException as e:

            return {"error": f"Could not reach CCF: {e}"}, 502

            print(f"ERROR: could not reach CCF for invoker onboarding: {e}")

            return {"error": "Could not reach CCF."}, 502

        try:

            data = response.json()

            # Handle request error

            return jsonify({"error": str(e)}), 500

    def get_providers(self):

        url = "https://{}/helper/api/getProviders".format(os.getenv('CAPIF_HOSTNAME'))

            # Handle request errors and return the error message

            return jsonify({"Error when retrieving number of providers": str(e)}), 500

    def get_providers_list(self):

        url = "https://{}/helper/api/getProviders".format(os.getenv('CAPIF_HOSTNAME'))

            # Handle request error

            return jsonify({"error": str(e)}), 500

    def get_users_total(self):

        def login2():

            try:

        except Exception as e:

            return "Error when retrieving number of users: ", e

    def delete_providers(self, provider_ids):

        try:

            responses = []

        except Exception as e:

            raise Exception("Error deleting apis: " + str(e))

    def get_apis1(self):

        url = "https://{}/helper/api/getServices".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    def get_apis_list(self):

        url = "https://{}/helper/api/getServices".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    def get_num_apis(self):

        url = "https://{}/helper/api/getServices".format(os.getenv('CAPIF_HOSTNAME'))

            # Handle request errors and return the error message

            return jsonify({"Error when retrieving number of APIs": str(e)}), 500

    def get_configuration(self):

        # Build the request URL

        url = "https://{}/helper/configuration/getConfiguration".format(os.getenv('CAPIF_HOSTNAME'))

            # Handle errors in case of request failures

            return jsonify({"error": str(e)}), 500

    def update_configuration_param(self, param_path, new_value):

        """ Update a single configuration parameter """

        url = "https://{}/helper/configuration/updateConfigParam".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    def add_new_category(self, category_name, category_values):

        """Add a new parameter category in 'settings'."""

        url = "https://{}/helper/configuration/addNewConfiguration".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    def remove_configuration_param(self, param_path):

        """ Delete a parameter within a category in 'settings' """

        url = "https://{}/helper/configuration/removeConfigParam".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    def remove_configuration_category(self, category_name):

        """ Delete an entire category within 'settings' """

        url = "https://{}/helper/configuration/removeConfigCategory".format(os.getenv('CAPIF_HOSTNAME'))

        except requests.exceptions.RequestException as e:

            return jsonify({"error": str(e)}), 500

    ### REGISTER CONFIGURATION ###

    def get_register_configuration(self):

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when fetching register configuration: {str(e)}"}), 500

    def update_register_configuration_param(self, param_path, new_value):

        """ Update a single parameter in the register configuration """

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when updating register configuration: {str(e)}"}), 500

    def replace_register_configuration(self, new_config):

        """Replace the entire register configuration"""

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when replacing register configuration: {str(e)}"}), 500

    def add_new_config_register_category(self, category_name, category_values):

        """Add a new parameter category in 'settings' in register."""

        def login():

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when adding config setting: {str(e)}"}), 500

    def remove_register_configuration_param(self, param_path):

        """Delete a parameter within a category in 'settings' in register."""

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error deleting configuration parameter: {str(e)}"}), 500

    def remove_register_configuration_category(self, category_name):

        """Delete an entire category within 'settings' in register."""

        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

import os

import contextlib

import tempfile

from cryptography import x509

from cryptography.hazmat.primitives import hashes, serialization

from cryptography.hazmat.primitives.asymmetric import rsa

    )

    return csr_request, private_key

# RAM-backed tmpfs. Private keys must never be written to the container's disk layer,

# which can be persisted, snapshotted or committed into an image.

SECURE_TMP_DIR = "/dev/shm"

class SecureTempUnavailable(Exception):

    """The RAM-backed temp directory is missing; we fail closed rather than fall back to disk."""

@contextlib.contextmanager

def mtls_files(cert_pem, key_pem):

    """Write a certificate bundle and private key to temp files for requests' `cert=`.

    Files are created with mkstemp (mode 0600) in /dev/shm and removed on exit,

    including when the request fails.

    """

    if not os.path.isdir(SECURE_TMP_DIR):

        raise SecureTempUnavailable(f"{SECURE_TMP_DIR} is not available")

    paths = []

    try:

        for content in (cert_pem, key_pem):

            fd, path = tempfile.mkstemp(dir=SECURE_TMP_DIR)

            paths.append(path)

            with os.fdopen(fd, "w") as f:

                f.write(content)

        yield paths[0], paths[1]

    finally:

        for path in paths:

            try:

                os.unlink(path)

            except FileNotFoundError:

                pass

 No newline at end of file

import os

import re

import requests

class VaultError(Exception):

    """Vault could not be reached or returned an unexpected response"""

# invoker_id comes from the URL and becomes part of the Vault path. Restricting it to

# [A-Za-z0-9_-] prevents path traversal (e.g. "..%2f..%2fca") out of the user's subtree.

# Do not relax this without another guard.

_INVOKER_ID_RE = re.compile(r"[A-Za-z0-9_-]+")

def vault_addr():

    """Base URL of Vault. VAULT_ADDR, otherwise falls back to http://VAULT_HOSTNAME:VAULT_PORT as before."""

    addr = os.getenv("VAULT_ADDR")

        return False

    return True

def read_invoker(user_uuid, invoker_id):

    """Read an invoker's stored credentials from Vault KV v2.

    Returns the stored dict (private_key, certificate, api_invoker_id), or None if

    no such invoker exists for this user. Raises VaultError on any other failure,

    so callers can tell "not found" (404) apart from "Vault is down" (5xx).

    """

    if not _INVOKER_ID_RE.fullmatch(invoker_id or ""):

        return None

    path = f"secret/data/backoffice/users/{user_uuid}/invokers/{invoker_id}"

    try:

        response = requests.get(

            f"{vault_addr()}/v1/{path}",

            headers=vault_headers(),

            verify=vault_verify(),

            timeout=10,

        )

    except requests.exceptions.RequestException as e:

        raise VaultError(f"Vault read failed for invoker {invoker_id}: {e}") from e

    if response.status_code == 404:

        return None

    if not response.ok:

        raise VaultError(f"Vault read failed for invoker {invoker_id}: {response.status_code}")

    try:

        return response.json()["data"]["data"]

    except (ValueError, KeyError, TypeError) as e:

        raise VaultError(f"Unexpected Vault response shape for invoker {invoker_id}") from e

 No newline at end of file

  - Onboards an API invoker in CAPIF for the logged-in user: generates a key pair and CSR, onboards via CCF using the user's CAPIF token, and stores the private key and certificate bundle in Vault under `secret/backoffice/users/{uuid}/invokers/{apiInvokerId}`.

  - Body: `{"apiInvokerInformation": "...", "notificationDestination": "..."}`

  - Responses: `201` `{"apiInvokerId": "..."}`; `400` missing fields; `403` account has no CAPIF credentials; `502` CCF could not be reached, rejected the onboarding, or returned an unexpected response; `500` credentials could not be stored in Vault. On `500` a rollback of the invoker in CCF is attempted; if it fails, the invoker remains in CCF and is logged for admin cleanup.

- GET /api/me/invokers/{apiInvokerId}/apis

  - Discovers the APIs visible to one of the user's invokers, calling CCF over mTLS with the invoker's certificate stored in Vault. Another user's invoker id returns `404`.

  - Requires `/dev/shm`: the key and certificate are written there only for the duration of the call. If it is missing, the call fails with `500` and never falls back to disk.

  - A CCF `404` with cause "No API Published accomplish filter conditions" is deliberately returned as `200` with an empty `serviceAPIDescriptions` list.

### Events and Monitoring