onboard provider

from flask import Blueprint, jsonify, request

from flask_jwt_extended import jwt_required

from ..core import invoker_ops

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__)

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

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

@jwt_required()

@capif_user_required

        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

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

    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.