Onboard API invokers for the logged-in user

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

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

def generate_keypair_and_csr():

    # 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, os.getenv('CERT_ORGANIZATION', 'Telefonica I+D')),

        x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, os.getenv('CERT_ORGANIZATIONAL_UNIT', 'Innovation')),

        x509.NameAttribute(NameOID.LOCALITY_NAME, os.getenv('CERT_LOCALITY', 'Madrid')),

        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, os.getenv('CERT_STATE', 'Madrid')),

        x509.NameAttribute(NameOID.COUNTRY_NAME, os.getenv('CERT_COUNTRY', 'ES')), # Country name must be exactly 2 characters

        x509.NameAttribute(NameOID.EMAIL_ADDRESS, os.getenv('CERT_EMAIL', '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(),

    )

    return csr_request, private_key

 No newline at end of file

import re

import os

import uuid

import jwt

from flask import request, jsonify

from flask_jwt_extended import verify_jwt_in_request, get_jwt

    """Return the CAPIF register token from the current JWT, or None for superadmin."""

    return get_jwt().get("capif_token")

def get_capif_uuid():

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

    return get_jwt().get("capif_uuid")

def get_capif_urls():

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

    return {

        "ccf_security_url": os.getenv("CCF_SECURITY_URL"),   

    }

def extract_uuid_from_capif_token(token):

    """Return the user's uuid from a CAPIF access token, or None if the token is missing or not in the expected shape. 

    Decoded without signature verification on purpose: the token is issued by the

    register service at login and is then sealed inside our own signed JWT, so this

    reads our own data rather than trusting client input.

    """

    if not token:

        return None

    try:

        claims = jwt.decode(token, options={"verify_signature": False})

    except jwt.PyJWTError:

        return None

    parts = str(claims.get("sub", "")).split()

    if len(parts) != 2:

        return None

    try:

        return str(uuid.UUID(parts[1]))

    except ValueError:

        return None

 No newline at end of file