Commit 7570d81c authored by Alex Kakiris's avatar Alex Kakiris
Browse files

extract CAPIF user guard into a decorator

parent 4dd88a2a

capif_backend/backoffice_service/controllers/me_controller.py

+8 −27
Changes for capif_backend/backoffice_service/controllers/me_controller.py: 8 added lines, 27 removed lines.
Original line number Diff line number Diff line
@@ -2,7 +2,7 @@ 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

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__)
@@ -10,15 +10,8 @@ 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")
@@ -27,20 +20,14 @@ def onboard_invoker(): 


    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", methods=["GET"])

@jwt_required()

def list_my_invokers():

    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



@capif_user_required

def list_my_invokers(user_uuid):

    try:

        invoker_ids = list_invokers(user_uuid)

    except VaultError as e:
@@ -51,14 +38,8 @@ def list_my_invokers(): 


@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



@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

capif_backend/backoffice_service/utils/utils.py

+16 −0
Changes for capif_backend/backoffice_service/utils/utils.py: 16 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -4,6 +4,7 @@ import uuid 
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)."""
@@ -45,6 +46,21 @@ def get_capif_uuid(): 
    """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 {