Merge branch 'feat/list-invokers' into 'dev'

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

- GET /api/me/invokers: lists the logged-in user's invoker ids (empty list if none).

### Events and Monitoring

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

from ..utils.vault import read_invoker, VaultError

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

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

@capif_user_required

def onboard_invoker(user_uuid):

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

    invoker_info = payload.get("apiInvokerInformation")

    notification_destination = payload.get("notificationDestination")

    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

        get_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"])

@me_routes.route("/invokers", 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

@capif_user_required

def list_my_invokers(user_uuid):

    try:

        invoker_ids = list_invokers(user_uuid)

    except VaultError as e:

        print(f"ERROR: {e}")

        return jsonify({"error": "Could not list invokers."}), 500

    user_uuid = get_capif_uuid()

    if not user_uuid:

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

    return jsonify({"invokers": invoker_ids}), 200

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

@jwt_required()

@capif_user_required

def discover_apis(api_invoker_id, user_uuid):

    # 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

import jwt

from flask import request, jsonify

from flask_jwt_extended import verify_jwt_in_request, get_jwt

from functools import wraps

def is_snake_case(value):

    """Verifica si una clave está en snake_case (solo letras minúsculas, números y guiones bajos)."""

    """Return the CAPIF user uuid from the current JWT, or None for superadmin."""

    return get_jwt().get("capif_uuid")

def capif_user_required(fn):

    """Decorator for routes guard for user-scoped endpoints.

    Must be below @jwt_required(), because it read claims from the JWT.

    """

    @wraps(fn)

    def wrapper(*args, **kwargs):

        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

        return fn(*args, user_uuid=user_uuid, **kwargs)

    return wrapper

def get_capif_urls():

    """CCF endpoints URLs. Static per deployment, so they are stored in environment variables."""

    return {

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

    except (ValueError, KeyError, TypeError) as e:

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

def list_invokers(user_uuid):

    """Return the apiInvokerIds stored for a user."""

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

    try:

        response = requests.get(

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

           params={"list": "true"},

           headers=vault_headers(),

           verify=vault_verify(),

           timeout=10,

        )

    except requests.exceptions.RequestException as e:

        raise VaultError(f"Vault list of invokers failed: {e}") from e

    # Vault returns 404 when the path has no keys: this user has no invokers yet.

    if response.status_code == 404:

        return []

    if not response.ok:

        raise VaultError(f"Vault list of invokers failed: {response.status_code}")

    try:

        keys = response.json()["data"]["keys"]

    except (ValueError, KeyError, TypeError) as e:

        raise VaultError("Unexpected Vault list response shape") from e

    # Keys ending in "/" would be sub-folders, not invokers

    return [key for key in keys if not key.endswith("/")]

 No newline at end of file