Authenticate Users

Validate JWTs, use service accounts, and propagate X-Project-ID from your solution’s services.

Authenticate Users

This page shows how a solution validates user identity and project context. The platform provides Keycloak-issued JWTs (multi-realm OIDC/PKCE); your service is responsible for validating the signature, extracting claims, and propagating context to downstream calls.

For the full platform-side authentication architecture (realm-per-org, JWKS rotation, SSO providers), see Platform › Authentication and Security › Authentication.

Request flow

Steps

1. Add JWT validation as a FastAPI dependency

   Use python-jose to verify the token against the realm’s JWKS. Cache the JWKS for ~10 minutes to avoid hammering Keycloak.

   packages/api/src/api/auth.py

   import os
   import time
   from typing import Annotated
   import httpx
   from fastapi import Depends, Header, HTTPException, status
   from jose import jwt, JWTError
   
   KEYCLOAK_ISSUER_URL = os.environ["KEYCLOAK_ISSUER_URL"]   # e.g. https://kc.example.com/realms/<org>
   KEYCLOAK_AUDIENCE   = os.environ["KEYCLOAK_AUDIENCE"]     # your service's client_id
   
   # JWKS TTL — Keycloak rotates signing keys (default ~12h). Lifetime cache
   # (e.g. lru_cache) WILL produce 401s for valid tokens after rotation.
   _JWKS_TTL_SECONDS = 600  # 10 minutes
   _jwks_cache: dict | None = None
   _jwks_cached_at: float = 0.0
   
   def _jwks() -> dict:
       global _jwks_cache, _jwks_cached_at
       now = time.monotonic()
       if _jwks_cache is None or now - _jwks_cached_at > _JWKS_TTL_SECONDS:
           url = f"{KEYCLOAK_ISSUER_URL.rstrip('/')}/protocol/openid-connect/certs"
           _jwks_cache = httpx.get(url, timeout=5.0).raise_for_status().json()
           _jwks_cached_at = now
       return _jwks_cache
   
   def current_user(authorization: Annotated[str | None, Header()] = None) -> dict:
       if not authorization or not authorization.lower().startswith("bearer "):
           raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing bearer token")
       token = authorization.split(" ", 1)[1]
       try:
           claims = jwt.decode(
               token,
               _jwks(),
               algorithms=["RS256"],
               audience=KEYCLOAK_AUDIENCE,
               issuer=KEYCLOAK_ISSUER_URL,
           )
       except JWTError as e:
           raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"Invalid token: {e}")
       return claims
   
   def project_id(x_project_id: Annotated[str | None, Header()] = None) -> str:
       if not x_project_id:
           raise HTTPException(status.HTTP_400_BAD_REQUEST, "Missing X-Project-ID header")
       return x_project_id
   
2. Protect your endpoints

   packages/api/src/api/main.py

   from fastapi import Depends, FastAPI
   from typing import Annotated
   from .auth import current_user, project_id
   
   app = FastAPI()
   
   @app.get("/healthz")
   async def healthz() -> dict[str, str]:
       return {"status": "ok"}
   
   @app.get("/echo")
   async def echo(
       msg: str,
       user: Annotated[dict, Depends(current_user)],
       proj: Annotated[str, Depends(project_id)],
   ) -> dict[str, str]:
       return {
           "echo": msg,
           "user_sub": user["sub"],
           "org_id": user.get("org_id", ""),
           "project_id": proj,
       }
   

   Routes that need auth declare current_user and (when project context matters) project_id as dependencies. Health probes stay unauthenticated.

3. Read identity, project, and roles from claims

   Standard claims you’ll use:

   Claim	Meaning	Notes
   sub	User UUID	Stable across sessions
   iss	Issuer URL	Realm path encodes org_id (last path segment)
   aud	Audience	Must equal your service’s client_id
   exp	Expiry	UNIX seconds
   org_id	Organization	Custom claim — also extractable from iss
   groups	Group memberships	Used by Governance for role mapping

   Helper to extract org_id from the issuer when the custom claim isn’t set:

   def org_id_from_issuer(iss: str) -> str:
       # iss like "https://kc.example.com/realms/acme-corp"
       return iss.rstrip("/").rsplit("/", 1)[-1]
   
4. Check permissions via Governance

   Solutions never implement RBAC themselves — they delegate to OpenFGA via Governance. Forward the user’s JWT and the resource ID; Governance returns allow/deny.

   import httpx, os
   
   GOVERNANCE_URL = os.environ.get("GOVERNANCE_URL", "http://em-runtime-governance.em-runtime:8001")
   
   async def require_permission(token: str, project_id: str, action: str, resource_uri: str) -> None:
       async with httpx.AsyncClient() as c:
           r = await c.post(
               f"{GOVERNANCE_URL}/governance/permissions/check",
               headers={"Authorization": f"Bearer {token}", "X-Project-ID": project_id},
               json={"action": action, "resource_uri": resource_uri},
               timeout=5.0,
           )
       r.raise_for_status()
       if not r.json()["allowed"]:
           raise HTTPException(status.HTTP_403_FORBIDDEN, "Not authorized")
   

   Use the auto-generated Python SDK instead of raw HTTP in real services — it handles error mapping, retries, and OpenAPI-validated request shapes.

5. Write a test fixture for a forged dev JWT

   Local tests should not call real Keycloak. Generate a signed token with a fixture private key and load the matching public key as a JWKS override.

   tests/conftest.py

   from datetime import datetime, timedelta, timezone
   from jose import jwt
   import pytest
   
   PRIV_KEY = open("tests/fixtures/dev-private.pem").read()
   PUB_KEY  = open("tests/fixtures/dev-public.pem").read()
   
   @pytest.fixture
   def dev_token():
       now = datetime.now(timezone.utc)
       return jwt.encode(
           {
               "sub": "user-123",
               "iss": "http://localhost:8080/realms/dev",
               "aud": "hello-solution-api",
               "org_id": "dev",
               "groups": ["developers"],
               "iat": int(now.timestamp()),
               "exp": int((now + timedelta(hours=1)).timestamp()),
           },
           PRIV_KEY,
           algorithm="RS256",
       )
   

   Generate the keypair once with openssl:

   openssl genrsa -out tests/fixtures/dev-private.pem 2048
   openssl rsa -in tests/fixtures/dev-private.pem -pubout -out tests/fixtures/dev-public.pem
   

   Configure your current_user dependency to load tests/fixtures/dev-public.pem as JWKS in test mode (e.g., when PYTEST_CURRENT_TEST is set).

Service-to-service auth

When your solution’s worker calls another service (or back into Governance/Assets), it can’t use a user’s JWT. Use a service account: a Keycloak client with client_credentials grant. See Platform › Service Accounts for the management API.

async def service_token() -> str:
    async with httpx.AsyncClient() as c:
        r = await c.post(
            f"{KEYCLOAK_ISSUER_URL}/protocol/openid-connect/token",
            data={
                "grant_type": "client_credentials",
                "client_id": os.environ["SVC_ACCOUNT_CLIENT_ID"],
                "client_secret": os.environ["SVC_ACCOUNT_SECRET"],
            },
            timeout=5.0,
        )
    return r.json()["access_token"]

The service-account client must be granted the same OpenFGA roles you’d grant a user. Cache tokens for expires_in - 60 seconds to avoid thundering herds at expiry.

Common errors

See the auth section of Troubleshooting.

Next steps

Manage secrets

Move KEYCLOAK_AUDIENCE and DB creds out of .env.

RBAC configuration

Configure OpenFGA roles for your solution’s resources.

API authentication

Reference for all platform-API auth headers.

SDKs › Python

Use the auto-generated SDK instead of raw HTTP.