add capif_uuid claim to issued tokens (f3193e49) · Commits · OCF / backoffice / backend

capif_backend/backoffice_service/controllers/login.py

+11 −4

Changes for capif_backend/backoffice_service/controllers/login.py: 11 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@ from flask_cors import cross_origin
		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__)
		@@ -24,6 +24,7 @@ def login():

		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":
		@@ -33,6 +34,9 @@ def login():
		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})")
		@@ -47,10 +51,11 @@ def login():
		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)

		@@ -64,11 +69,13 @@ def refresh():
		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)

capif_backend/backoffice_service/utils/utils.py

+27 −0

Changes for capif_backend/backoffice_service/utils/utils.py: 27 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import re
		import os
		import uuid
		import jwt
		from flask import request, jsonify
		from flask_jwt_extended import verify_jwt_in_request, get_jwt

		@@ -39,6 +41,10 @@ def get_capif_token():
		"""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 {
		@@ -49,4 +55,25 @@ def get_capif_urls():
		"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