Enforcing Client Identity Alongside Realm Roles

roles = claims.get("realm_access", {}).get("roles", [])
if "admin" not in roles:
    raise HTTPException(403)

"""FastAPI backend that authorizes on Keycloak realm role AND authorized party (azp).

A realm role alone is not enough: it is attached to the user, not to the client,
so any client in the realm that can obtain a token for a privileged user produces
a token carrying that same role. This dependency also checks which client the
token was issued for (azp) against an explicit allow-list before trusting the role.
"""
from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Callable

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient

REALM_ISSUER = os.environ.get(
    "OIDC_ISSUER", "https://auth.example.internal/realms/<your-realm>"
)
JWKS_URL = os.environ.get(
    "OIDC_JWKS_URL", f"{REALM_ISSUER}/protocol/openid-connect/certs"
)
# Requires an audience mapper on the client scope(s) used by trusted clients;
# Keycloak's default access token audience ("account") will not match this.
API_AUDIENCE = os.environ.get("OIDC_AUDIENCE", "admin-console-api")

# Clients whose tokens are trusted to carry privileged realm roles for this API.
# A client not on this list is rejected even if the user holds the required role.
TRUSTED_AZP = {"admin-console", "internal-batch-service"}
REQUIRED_ROLE = "admin"

bearer_scheme = HTTPBearer(auto_error=True)


@dataclass
class TokenVerifier:
    issuer: str
    audience: str
    trusted_azp: set[str]
    get_signing_key: Callable[[str], object]
    algorithms: tuple[str, ...] = ("RS256",)

    def verify(self, token: str) -> dict:
        try:
            signing_key = self.get_signing_key(token)
        except Exception as exc:  # JWKS lookup or key-id mismatch
            raise HTTPException(
                status.HTTP_401_UNAUTHORIZED, "Could not resolve signing key"
            ) from exc

        key = signing_key.key if hasattr(signing_key, "key") else signing_key

        try:
            claims = jwt.decode(
                token,
                key=key,
                algorithms=list(self.algorithms),
                issuer=self.issuer,
                audience=self.audience,
                options={"require": ["exp", "iat", "iss", "azp"]},
            )
        except jwt.PyJWTError as exc:
            raise HTTPException(
                status.HTTP_401_UNAUTHORIZED, f"Invalid token: {exc}"
            ) from exc

        azp = claims.get("azp")
        if azp not in self.trusted_azp:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                f"Client '{azp}' is not authorized to call this API",
            )

        return claims


def _default_signing_key(token: str):
    # PyJWKClient caches the JWKS response internally; no per-request fetch.
    jwks_client = PyJWKClient(JWKS_URL)
    return jwks_client.get_signing_key_from_jwt(token)


verifier = TokenVerifier(
    issuer=REALM_ISSUER,
    audience=API_AUDIENCE,
    trusted_azp=TRUSTED_AZP,
    get_signing_key=_default_signing_key,
)


def get_verifier() -> TokenVerifier:
    return verifier


def require_role(role: str):
    def dependency(
        credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
        token_verifier: TokenVerifier = Depends(get_verifier),
    ) -> dict:
        claims = token_verifier.verify(credentials.credentials)
        roles = claims.get("realm_access", {}).get("roles", [])
        if role not in roles:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, f"Missing required role '{role}'"
            )
        return claims

    return dependency


app = FastAPI(title="admin-console-api")


@app.get("/admin/ping")
def admin_ping(claims: dict = Depends(require_role(REQUIRED_ROLE))) -> dict:
    return {"status": "ok", "subject": claims["sub"], "azp": claims["azp"]}

"""Unit tests for the azp + realm-role authorization dependency.

No live Keycloak instance is required: a throwaway RSA key pair signs test
tokens locally, and the app's JWKS lookup is replaced with a dependency
override that hands back the matching public key directly.
"""
from __future__ import annotations

import time

import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi.testclient import TestClient

from main import API_AUDIENCE, REALM_ISSUER, TokenVerifier, app, get_verifier

_private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_public_pem = _private_key.public_key().public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo,
).decode()
_private_pem = _private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption(),
).decode()


def make_token(azp: str, roles: list[str]) -> str:
    now = int(time.time())
    claims = {
        "iss": REALM_ISSUER,
        "aud": API_AUDIENCE,
        "azp": azp,
        "sub": "3f1c2b7a-0000-0000-0000-000000000000",
        "iat": now,
        "exp": now + 300,
        "realm_access": {"roles": roles},
    }
    return jwt.encode(claims, _private_pem, algorithm="RS256")


@pytest.fixture()
def client():
    test_verifier = TokenVerifier(
        issuer=REALM_ISSUER,
        audience=API_AUDIENCE,
        trusted_azp={"admin-console", "internal-batch-service"},
        get_signing_key=lambda token: _public_pem,
    )
    app.dependency_overrides[get_verifier] = lambda: test_verifier
    yield TestClient(app)
    app.dependency_overrides.clear()


def test_right_role_wrong_azp_is_rejected(client: TestClient) -> None:
    token = make_token(azp="public-spa", roles=["admin"])
    response = client.get("/admin/ping", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403
    assert "not authorized" in response.json()["detail"]


def test_right_role_and_right_azp_is_accepted(client: TestClient) -> None:
    token = make_token(azp="admin-console", roles=["admin"])
    response = client.get("/admin/ping", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    body = response.json()
    assert body["azp"] == "admin-console"


def test_right_azp_wrong_role_is_rejected(client: TestClient) -> None:
    token = make_token(azp="admin-console", roles=["support"])
    response = client.get("/admin/ping", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403
    assert "Missing required role" in response.json()["detail"]