> ## Documentation Index
> Fetch the complete documentation index at: https://docs.emergence.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# 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](/platform/authentication) and [Security › Authentication](/security/authentication).
## Request flow
```mermaid theme={null}
%%{init: {'theme': 'base', 'themeVariables': {'lineColor': '#555555', 'fontFamily': 'sans-serif', 'edgeLabelBackground': '#ffffff'}}}%%
sequenceDiagram
autonumber
participant U as User / Client
participant S as Your service
participant KC as Keycloak (JWKS)
participant G as Governance (OpenFGA)
participant DB as Your DB
U->>S: Request + Authorization: Bearer JWT + X-Project-ID
S->>KC: GET /realms/<org>/protocol/openid-connect/certs (cached)
KC-->>S: JWKS keys
S->>S: Verify signature, exp, aud, iss
S->>G: require_permission(user, action, resource)
G-->>S: allow / deny
alt Allowed
S->>DB: Query scoped by org_id + project_id
DB-->>S: Result
S-->>U: 200 OK
else Denied
S-->>U: 403 Forbidden
end
```
## Steps
Use `python-jose` to verify the token against the realm's JWKS. Cache the JWKS for \~10 minutes to avoid hammering Keycloak.
```python packages/api/src/api/auth.py theme={null}
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/
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
```
```python packages/api/src/api/main.py theme={null}
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.
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:
```python theme={null}
def org_id_from_issuer(iss: str) -> str:
# iss like "https://kc.example.com/realms/acme-corp"
return iss.rstrip("/").rsplit("/", 1)[-1]
```
Solutions never implement RBAC themselves — they delegate to OpenFGA via Governance. Forward the user's JWT and the resource ID; Governance returns allow/deny.
```python theme={null}
import httpx, os
GOVERNANCE_URL = os.environ.get("GOVERNANCE_URL", "http://em-runtime-governance.em-runtime:8000")
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](/sdks/python) instead of raw HTTP in real services — it handles error mapping, retries, and OpenAPI-validated request shapes.
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.
```python tests/conftest.py theme={null}
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`:
```bash theme={null}
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](/platform/service-accounts) for the management API.
```python theme={null}
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](/guides/solution-dev/troubleshooting#auth).
## Next steps
Move `KEYCLOAK_AUDIENCE` and DB creds out of `.env`.
Configure OpenFGA roles for your solution's resources.
Reference for all platform-API auth headers.
Use the auto-generated SDK instead of raw HTTP.