Merge branch 'feat/invoker-onboarding' into 'dev'

This repository contains the OpenCAPIF Backoffice Backend service.

Its purpose is to manage OpenCAPIF resources (Invokers, Providers, APIs, and Users), including listing, filtering, and deletion operations, plus authentication and configuration management endpoints.

Its purpose is to manage OpenCAPIF resources (Invokers, Providers, APIs, and Users), including listing, filtering, and deletion operations, plus authentication and configuration management endpoints. It also lets regular users onboard their own API invokers.

## Prerequisites

- If you use the default local OpenCAPIF deployment, no value changes are usually needed.

- If any of those service names are changed in your OpenCAPIF deployment, you must update docker-compose.yml accordingly (for example CAPIF_HOSTNAME, REGISTER_HOSTNAME, VAULT_HOSTNAME, and extra_hosts). This will be improved in future versions.

- CCF_ONBOARDING_URL, CCF_API_ONBOARDING_URL, CCF_PUBLISH_URL, CCF_DISCOVER_URL, and CCF_SECURITY_URL are static per deployment and must match the ccf_* paths returned by CAPIF's /getauth response; a mismatch is logged as a WARNING on login.

- `CERT_ORGANIZATION` (default `Telefonica I+D`), `CERT_ORGANIZATIONAL_UNIT` (`Innovation`), `CERT_LOCALITY` (`Madrid`), `CERT_STATE` (`Madrid`), `CERT_COUNTRY` (`ES`, must be a 2-letter code) and `CERT_EMAIL` (`inno@tid.es`) set the subject of generated CSRs only. They do not appear in issued certificates: the Vault signing role `pki_int/roles/my-ca` has `use_csr_common_name=false` and `use_csr_sans=false`, so the signer sets the CN (e.g. `superadmin`, or the `apiInvokerId` for invokers) and ignores the CSR subject. They have defaults, so they don't need to be set in `docker-compose.yml`.

- `VAULT_ADDR` sets the Vault base URL (e.g. `https://vault.example:8200`). If unset, it defaults to `http://$VAULT_HOSTNAME:$VAULT_PORT`.

- `VAULT_CACERT` (optional) is the path to a CA bundle used to verify Vault's TLS certificate. If unset, default verification is used, so an `https://` Vault with a private CA requires it. Has no effect with an `http://` address.

Example /etc/hosts entries for a default local setup:

## Authentication

- Login endpoint currently validates fixed credentials (superadmin/admin).

- POST /api/login validates `superadmin` locally; all other users are authenticated against the CAPIF register service (`/getauth`). For those users, the CAPIF access token and the user's CAPIF uuid are embedded in the issued JWT (`capif_token`, `capif_uuid`).

- POST /api/login returns access_token and refresh_token.

- Most endpoints require a Bearer access token in the Authorization header.

- POST /api/refresh requires a valid refresh token.

- POST /api/refresh

- GET /api/auth/me

### Me (user-scoped)

Endpoints acting on behalf of the logged-in user. They require a valid access token but are not admin-only; only accounts with CAPIF credentials can use them (superadmin gets 403).

- POST /api/me/invoker

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

### Events and Monitoring

- GET /api/events/{id}

- Authentication with access and refresh JWT tokens.

- Resource management for Invokers, Providers, APIs, and Users.

- Configuration and Register configuration management endpoints.

- User-scoped invoker onboarding (`POST /api/me/invoker`), with credentials stored in Vault.

In progress:

- local_build.sh version automation and registry push flow.

import json

import requests

import os

from cryptography import x509

from cryptography.hazmat.primitives import hashes, serialization

from cryptography.hazmat.primitives.asymmetric import rsa

from cryptography.x509.oid import NameOID

from .controllers.default_controller import backoffice_routes

from .controllers.invokers_controller import invokers_routes

from .controllers.providers_controller import providers_routes

from .controllers.apis_controller import apis_routes

from .controllers.login import access_routes

from .controllers.configuration_controller import configuration_routes

from .controllers.me_controller import me_routes

from flask_cors import CORS

from .utils.certs import generate_keypair_and_csr

from .utils.vault import vault_addr, vault_headers, vault_verify

# Create Flask app

app = Flask(__name__)

app.register_blueprint(apis_routes, url_prefix="/api/apis")

app.register_blueprint(users_routes, url_prefix="/api/users")

app.register_blueprint(configuration_routes, url_prefix="/api")

app.register_blueprint(me_routes, url_prefix="/api/me")

@app.route("/api/options", methods=["OPTIONS"])

def handle_options():

app.config["JWT_SECRET_KEY"] = "super-secret"

jwt = JWTManager(app)

# create public/private key

key = rsa.generate_private_key(public_exponent=65537, key_size=2048)

# Generate CSR

subject = x509.Name([

    x509.NameAttribute(NameOID.ORGANIZATION_NAME, 'Telefonica I+D'),

    x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, 'Innovation'),

    x509.NameAttribute(NameOID.LOCALITY_NAME, 'Madrid'),

    x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, 'Madrid'),

    x509.NameAttribute(NameOID.COUNTRY_NAME, 'ES'),

    x509.NameAttribute(NameOID.EMAIL_ADDRESS, 'inno@tid.es'),

])

csr_request = (

    x509.CertificateSigningRequestBuilder()

    .subject_name(subject)

    .sign(key, hashes.SHA256())

    .public_bytes(serialization.Encoding.PEM)

)

# Get private key

private_key = key.private_bytes(

    encoding=serialization.Encoding.PEM,

    format=serialization.PrivateFormat.PKCS8,

    encryption_algorithm=serialization.NoEncryption(),

)

csr_request, private_key = generate_keypair_and_csr()

# Read environment variable to determine whether to load from Vault or from environment variables

LOAD_FROM_VAULT = os.getenv('LOAD_FROM_VAULT', 'true').lower() == 'true'

    # Get ca.cert from Vault

    config = Config().get_config()  # get configuration from config.py

    url = 'http://{}:{}/v1/secret/data/ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))

    headers = {'X-Vault-Token': os.getenv('VAULT_TOKEN')}

    response = requests.request("GET", url, headers=headers, verify=False)

    url = f"{vault_addr()}/v1/secret/data/ca"

    headers = vault_headers()

    response = requests.request("GET", url, headers=headers, verify=vault_verify())

    # Print the full response content for inspection

    print("Response from Vault (ca.cert request):")

    ca_file.close()

    # Request Vault to sign the CSR

    url = 'http://{}:{}/v1/pki_int/sign/my-ca'.format(os.getenv('VAULT_HOSTNAME'), os.getenv('VAULT_PORT'))

    url = f"{vault_addr()}/v1/pki_int/sign/my-ca"

    data = {

        'format': 'pem_bundle',

        'ttl': '43000h',

        'csr': csr_request.decode('utf-8'),

        'common_name': "superadmin"

    }

    response = requests.request("POST", url, headers=headers, data=data, verify=False)

    response = requests.request("POST", url, headers=headers, data=data, verify=vault_verify())

    backoffice_cert = json.loads(response.text)['data']['certificate']

    # Save private_key

from datetime import timedelta

from ..core.operations import Operations

from ..utils.utils import get_capif_urls

from ..utils.utils import get_capif_urls, extract_uuid_from_capif_token

# create blueprint backoffice_routes

access_routes = Blueprint("access_routes", __name__)

    role = "admin"

    capif_token = None

    capif_uuid = None

    # Check if the username and password are correct

    # TODO: Implement proper authentication mechanism (e.g., check against a database)

    if username != "superadmin" or password != "admin":

            return jsonify({"msg": "Bad username or password"}), 401

        role = "user"

        capif_token = capif_auth.get("access_token")

        capif_uuid = extract_uuid_from_capif_token(capif_token)

        if not capif_uuid:

            print("WARNING: Could not extract UUID from CAPIF token sub claim")

        for key, configured in get_capif_urls().items():

            if configured != capif_auth.get(key):

                print(f"WARNING: {key} from /getauth ({capif_auth.get(key)}) differs from config ({configured})")

    expires_refresh = timedelta(minutes=240)

    # identity is a data that can be serialized in json -> username = superadmin

    claims = {"role": role, "capif_token": capif_token, "capif_uuid": capif_uuid}

    access_token = create_access_token(identity=username, expires_delta=expires,

                                    additional_claims={"role": role, "capif_token": capif_token})  # Include role and capif_token in the access token claims

                                    additional_claims=claims)

    refresh_token = create_refresh_token(identity=username, expires_delta=expires_refresh,

                                        additional_claims={"role": role, "capif_token": capif_token})  # Include role and capif_token in the refresh token claims

                                        additional_claims=claims)

    return jsonify(access_token=access_token, refresh_token=refresh_token, role=role)

    current_user = get_jwt_identity()

    role = get_jwt().get("role", "user")

    capif_token = get_jwt().get("capif_token")

    capif_uuid = get_jwt().get("capif_uuid")

    # Configure new access token expiration to 8 hours

    expires = timedelta(hours=8)

    new_access_token = create_access_token(identity=current_user, expires_delta=expires,

                                           additional_claims={"role": role, "capif_token": capif_token})

                                           additional_claims={"role": role, "capif_token": capif_token,

                                                              "capif_uuid": capif_uuid})

    return jsonify(access_token=new_access_token, user=current_user, role=role)

from flask import Blueprint, jsonify, request

from flask_jwt_extended import jwt_required

from ..core.operations import Operations

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

me_routes = Blueprint("me_routes", __name__)

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

@me_routes.route("/invoker", methods=["POST"])

@jwt_required()

def onboard_invoker():

    capif_token = get_capif_token()

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

    payload = request.get_json(silent=True) or {}

    invoker_info = payload.get("apiInvokerInformation")

    notification_destination = payload.get("notificationDestination")

    if not invoker_info or not notification_destination:

        return jsonify({"error": "apiInvokerInformation and notificationDestination are required."}), 400

    ccf_onboarding_url = get_capif_urls()["ccf_onboarding_url"]

    data, status = operations.onboard_invoker(

        capif_token, user_uuid, ccf_onboarding_url, invoker_info, notification_destination

    )

    return jsonify(data), status

 No newline at end of file

from flask import jsonify, make_response

import requests

# from ..configs.config import Config

from ..config import Config

import os

import json

import copy

from requests.auth import HTTPBasicAuth

from urllib.parse import urlencode

# from ..configs.config import Config

from ..config import Config

from ..utils.certs import generate_keypair_and_csr

from ..utils.vault import store_invoker

class Operations:

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when retrieving user list: {str(e)}"}), 500

    def find_user_by_uuid(self, uuid):

        try:

            # Call retrieve_users_pages_order to get all users

        except Exception as e:

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

    def get_number_of_users(self):

        def login2():

            try:

        except requests.exceptions.RequestException as e:

            return jsonify({"error": f"Error when creating user: {str(e)}"}), 500

    def get_invokers(self):

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

        except requests.exceptions.RequestException as e:

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

    def delete_invokers(self, invoker_ids):

        try:

            responses = []

                error_message += " | Response content: " + response.text

            raise Exception("Error deleting invokers: " + error_message)

    def get_num_invokers(self):

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

            # Handle request errors and return the error message

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

    def get_invokers_pages_order1(self, page, page_size, sort_order):

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

            print(f"Error: {str(e)}")  # Log de error

            return {"error": str(e)}

    def get_security_context_of_a_invoker(self, invoker_id):

        # Construir la URL con el invoker_id como query parameter

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

            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:

            return {"error": f"Could not reach CCF: {e}"}, 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