carry capif access token in backoffice jwt (8eb04627) · Commits · OCF / backoffice / backend

capif_backend/backoffice_service/controllers/login.py

+12 −5

Changes for capif_backend/backoffice_service/controllers/login.py: 12 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ from flask_cors import cross_origin
		from datetime import timedelta

		from ..core.operations import Operations
		from ..utils.utils import get_capif_urls

		# create blueprint backoffice_routes
		access_routes = Blueprint("access_routes", __name__)
		@@ -22,14 +23,19 @@ def login():
		print("username: ", username)

		role = "admin"
		capif_token = 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":
		_, status_code = operations.user_login(username, password)
		capif_auth, status_code = operations.user_login(username, password)
		if status_code != 200:
		print("Bad username or password")
		return jsonify({"msg": "Bad username or password"}), 401
		role = "user"
		capif_token = capif_auth.get("access_token")
		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})")

		# Configure token expiration to 24 hours
		# expires = timedelta(hours=8)
		@@ -42,9 +48,9 @@ def login():

		# identity is a data that can be serialized in json -> username = superadmin
		access_token = create_access_token(identity=username, expires_delta=expires,
		additional_claims={"role": role})
		additional_claims={"role": role, "capif_token": capif_token}) # Include role and capif_token in the access token claims
		refresh_token = create_refresh_token(identity=username, expires_delta=expires_refresh,
		additional_claims={"role": role})
		additional_claims={"role": role, "capif_token": capif_token}) # Include role and capif_token in the refresh token claims

		return jsonify(access_token=access_token, refresh_token=refresh_token, role=role)

		@@ -56,12 +62,13 @@ def refresh():
		return jsonify(message="CORS options OK"), 200

		current_user = get_jwt_identity()
		role = get_jwt().get("role", "user") # Get the role from the refresh token claims
		role = get_jwt().get("role", "user")
		capif_token = get_jwt().get("capif_token")

		# 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}) # New access token with the same role as the refresh token
		additional_claims={"role": role, "capif_token": capif_token})

		return jsonify(access_token=new_access_token, user=current_user, role=role)

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
		import re
		import os
		from flask import request, jsonify
		from flask_jwt_extended import verify_jwt_in_request, get_jwt

		@@ -34,3 +35,18 @@ def admin_only():

		return None

		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_urls():
		"""CCF endpoints URLs. Static per deployment, so they are stored in environment variables."""
		return {
		"ccf_onboarding_url": os.getenv("CCF_ONBOARDING_URL"),
		"ccf_api_onboarding_url": os.getenv("CCF_API_ONBOARDING_URL"),
		"ccf_publish_url": os.getenv("CCF_PUBLISH_URL"),
		"ccf_discover_url": os.getenv("CCF_DISCOVER_URL"),
		"ccf_security_url": os.getenv("CCF_SECURITY_URL"),
		}