Login Is Not Permission: Building a Working Authorizer Demo for New Engineers

 Next.js web ─┐                                 ┌─ PostgreSQL (users)
 Flet app ────┼─► Authorizer (OIDC, RS256) ─────┤
              │             ▲ JWKS (public keys)└─ Redis (sessions)
              └─► FastAPI ──┘  verifies every token, then applies
                             role + division rules (datasets, RAG, tools)

authorizer-demo/
├── docker-compose.yml      Authorizer + PostgreSQL + Redis + API
├── .env.example            template; init.sh writes the real .env
├── Makefile
├── scripts/                init.sh, seed_users.py, e2e_check.py, _common.py
├── backend/                FastAPI: app/ and tests/
├── frontend/               Next.js "Access ledger"
└── flet_app/               Python client

# Authorizer v2 is configured ONLY through CLI flags (it ignores .env / OS env).
# We keep values in .env for Docker Compose, pass them into the container as
# environment variables, and let a small shell wrapper turn them into flags.
name: authorizer-demo
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: authorizer
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: authorizer
    volumes:
      - pg_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U authorizer -d authorizer"]
      interval: 5s
      timeout: 3s
      retries: 20
  redis:
    image: redis:7-alpine
    command: ["redis-server", "--appendonly", "yes"]
    volumes:
      - redis_data:/data
  authorizer:
    # Pin a tested tag in .env (AUTHORIZER_VERSION) before production.
    image: quay.io/authorizer/authorizer:${AUTHORIZER_VERSION:-latest}
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_started
    ports:
      - "8080:8080"
    environment:
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      AUTHORIZER_URL: ${AUTHORIZER_URL}
      CLIENT_ID: ${CLIENT_ID}
      CLIENT_SECRET: ${CLIENT_SECRET}
      ADMIN_SECRET: ${ADMIN_SECRET}
      ENCRYPTION_KEY: ${ENCRYPTION_KEY}
    volumes:
      # RSA key pair created by scripts/init.sh. Persisting it means tokens
      # stay valid across restarts and the JWKS endpoint stays stable.
      - ./keys:/keys:ro
    # The image ENTRYPOINT is ./authorizer; we swap in a shell so $$VARS expand
    # inside the container (the double $$ stops Compose expanding them first).
    entrypoint: ["/bin/sh", "-c"]
    command:
      - >-
        exec ./authorizer
        --database-type=postgres
        --database-url="postgres://authorizer:$${POSTGRES_PASSWORD}@postgres:5432/authorizer?sslmode=disable"
        --redis-url=redis://redis:6379
        --url="$${AUTHORIZER_URL}"
        --client-id="$${CLIENT_ID}"
        --client-secret="$${CLIENT_SECRET}"
        --admin-secret="$${ADMIN_SECRET}"
        --jwt-type=RS256
        --jwt-private-key="$$(cat /keys/jwt_private.pem)"
        --jwt-public-key="$$(cat /keys/jwt_public.pem)"
        --encryption-key="$${ENCRYPTION_KEY}"
        --jwt-role-claim=roles
        --roles=admin,viewer,div-finance,div-research
        --default-roles=viewer
        --protected-roles=admin,div-finance,div-research
        --allowed-origins=http://localhost:3000,http://localhost:8000,http://127.0.0.1:8765,http://localhost:8080,http://127.0.0.1:8080
        --enable-email-verification=false
        --disable-mfa
        --app-cookie-secure=false
        --admin-cookie-secure=false
        --organization-name="Authorizer Demo"
  backend:
    build: ./backend
    depends_on:
      - authorizer
    ports:
      - "8000:8000"
    environment:
      AUTHORIZER_URL: ${AUTHORIZER_URL}            # must equal the token's `iss`
      AUTHORIZER_INTERNAL_URL: http://authorizer:8080   # how the container reaches it
      CLIENT_ID: ${CLIENT_ID}                       # must equal the token's `aud`
      ROLE_CLAIM: allowed_roles                     # all roles assigned to the user
      ALLOWED_ORIGINS: http://localhost:3000
volumes:
  pg_data:
  redis_data:

# Copied to .env (with random secrets) by scripts/init.sh
AUTHORIZER_VERSION=latest
AUTHORIZER_URL=http://localhost:8080
CLIENT_ID=change-me
CLIENT_SECRET=change-me
ADMIN_SECRET=change-me
ENCRYPTION_KEY=change-me
POSTGRES_PASSWORD=change-me

#!/usr/bin/env bash
# One-time setup: random secrets, RSA signing keys, frontend env file.
set -euo pipefail
cd "$(dirname "$0")/.."
rand() { openssl rand -hex 24; }
if [ ! -f .env ]; then
  cp .env.example .env
  sed -i.bak \
    -e "s/^CLIENT_ID=.*/CLIENT_ID=$(openssl rand -hex 12)/" \
    -e "s/^CLIENT_SECRET=.*/CLIENT_SECRET=$(rand)/" \
    -e "s/^ADMIN_SECRET=.*/ADMIN_SECRET=$(rand)/" \
    -e "s/^ENCRYPTION_KEY=.*/ENCRYPTION_KEY=$(rand)/" \
    -e "s/^POSTGRES_PASSWORD=.*/POSTGRES_PASSWORD=$(rand)/" .env
  rm -f .env.bak
  echo "created .env"
fi
mkdir -p keys
if [ ! -f keys/jwt_private.pem ]; then
  openssl genrsa -out keys/jwt_private.pem 2048 2>/dev/null
  openssl rsa -in keys/jwt_private.pem -pubout -out keys/jwt_public.pem 2>/dev/null
  # Demo only: the container runs as a non-root user and must read the key.
  chmod 644 keys/*.pem
  echo "created RSA signing keys in ./keys"
fi
set -a; . ./.env; set +a
cat > frontend/.env.local <<EOT
NEXT_PUBLIC_AUTHORIZER_URL=${AUTHORIZER_URL}
NEXT_PUBLIC_API_URL=http://localhost:8000
EOT
echo "created frontend/.env.local"
echo "Next: docker compose up -d --build && python3 scripts/seed_users.py"

.PHONY: init up down logs seed e2e api web flet test reset

init:  ; ./scripts/init.sh
up:    ; docker compose up -d --build
down:  ; docker compose down
logs:  ; docker compose logs -f authorizer backend
seed:  ; python3 scripts/seed_users.py
e2e:   ; python3 scripts/e2e_check.py
test:  ; cd backend && python3 -m pytest -q
api:   ; set -a && . ./.env && set +a && cd backend && uvicorn app.main:app --reload --port 8000
web:   ; cd frontend && npm install && npm run dev
reset: ; docker compose down -v && rm -f .env keys/*.pem
flet:  ; cd flet_app && pip install -r requirements.txt && flet run main.py

./scripts/init.sh
docker compose up -d --build

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # Reads real environment variables first, then a local/root .env file.
    model_config = SettingsConfigDict(env_file=(".env", "../.env"), extra="ignore")
    # Public Authorizer URL. Must equal the `iss` claim in the tokens.
    authorizer_url: str = "http://localhost:8080"
    # How THIS process reaches Authorizer (differs from the public URL in Docker).
    authorizer_internal_url: str | None = None
    # Must equal the `aud` claim (Authorizer uses its --client-id as audience).
    client_id: str = "demo-client-id"
    allowed_origins: str = "http://localhost:3000"
    jwks_cache_seconds: int = 300
    # Which claim carries the user's roles. Authorizer puts the roles ACTIVE in this session
    # in `roles` (just the default role unless the client asked for more at login) and ALL
    # roles assigned to the user in `allowed_roles`. The demo authorizes on `allowed_roles`.
    role_claim: str = "allowed_roles"
    @property
    def issuer(self) -> str:
        return self.authorizer_url.rstrip("/")
    @property
    def jwks_url(self) -> str:
        base = (self.authorizer_internal_url or self.authorizer_url).rstrip("/")
        return f"{base}/.well-known/jwks.json"
    @property
    def origins(self) -> list[str]:
        return [o.strip() for o in self.allowed_origins.split(",") if o.strip()]

@lru_cache
def get_settings() -> Settings:
    return Settings()

"""Token verification and authorization dependencies.Authentication  = "is this a valid token issued by Authorizer?"  -> current_user
Authorization   = "may THIS user do THIS thing to THIS resource?" -> require_roles
                  and the per-resource checks in main.py / data.py
Authorizer proves who the caller is. Every route still has to decide what that
caller may touch. That second half is never automatic.
"""
from dataclasses import dataclass
from functools import lru_cache
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient
from jwt.exceptions import PyJWKClientConnectionError, PyJWKClientError, PyJWTError
from .config import get_settings
bearer = HTTPBearer(auto_error=False)
DIVISION_PREFIX = "div-"

@dataclass(frozen=True)
class User:
    sub: str
    email: str | None
    roles: frozenset[str]
    claims: dict
    @property
    def is_admin(self) -> bool:
        return "admin" in self.roles
    @property
    def divisions(self) -> frozenset[str]:
        return frozenset(r.removeprefix(DIVISION_PREFIX) for r in self.roles if r.startswith(DIVISION_PREFIX))
    def can_access_division(self, division: str) -> bool:
        return self.is_admin or division in self.divisions

@lru_cache
def jwks_client() -> PyJWKClient:
    s = get_settings()
    # Keys are cached, so a request does not hit Authorizer every time.
    return PyJWKClient(s.jwks_url, cache_keys=True, lifespan=s.jwks_cache_seconds)

def _signing_key(token: str):
    client = jwks_client()
    kid = jwt.get_unverified_header(token).get("kid")
    if kid:
        return client.get_signing_key(kid).key
    keys = client.get_signing_keys()
    if len(keys) == 1:
        return keys[0].key
    raise PyJWTError("token has no kid and JWKS has several keys")

def _unauthorized(detail: str) -> HTTPException:
    return HTTPException(status.HTTP_401_UNAUTHORIZED, detail, headers={"WWW-Authenticate": "Bearer"})

def decode_token(token: str) -> dict:
    s = get_settings()
    try:
        claims = jwt.decode(
            token,
            _signing_key(token),
            algorithms=["RS256"],  # never accept "none" or HS256 here
            audience=s.client_id,
            issuer=s.issuer,
            options={"require": ["exp", "iat", "iss", "aud", "sub"]},
        )
    except PyJWKClientConnectionError:
        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "identity provider unreachable")
    except (PyJWKClientError, PyJWTError) as exc:
        raise _unauthorized(f"invalid token: {exc.__class__.__name__}")
    # An ID token is signed by the same key. It must not work as an API credential.
    if claims.get("token_type", "access_token") != "access_token":
        raise _unauthorized("not an access token")
    return claims

def _roles(claims: dict) -> frozenset[str]:
    """Read roles from the configured claim (then `roles`, `role`).
    Authorizer's `roles` claim is only what the session activated (usually the default role);
    `allowed_roles` is everything an admin assigned. Reading `roles` alone silently drops
    division roles, so the demo defaults to `allowed_roles`. For least-privilege sessions,
    have clients request specific roles at login and set ROLE_CLAIM=roles instead. Accepts a list, or a
    comma/space separated string, because identity providers differ."""
    for name in (get_settings().role_claim, "roles", "role"):
        raw = claims.get(name)
        if not raw:
            continue
        items = raw.replace(",", " ").split() if isinstance(raw, str) else [str(r) for r in raw]
        return frozenset(i.strip() for i in items if i.strip())
    return frozenset()

def current_user(creds: HTTPAuthorizationCredentials | None = Depends(bearer)) -> User:
    if creds is None or creds.scheme.lower() != "bearer":
        raise _unauthorized("missing bearer token")
    claims = decode_token(creds.credentials)
    return User(sub=claims["sub"], email=claims.get("email"), roles=_roles(claims), claims=claims)

def require_roles(*needed: str):
    """Allow the request if the user holds at least one of the roles."""
    def dependency(user: User = Depends(current_user)) -> User:
        if not user.roles.intersection(needed):
            raise HTTPException(status.HTTP_403_FORBIDDEN, f"requires one of: {', '.join(needed)}")
        return user
    return dependency

"""In-memory stand-ins for a database, a vector store, and an MCP tool registry.
The point for trainees: every record carries its owning division, and every read
path filters by the caller's divisions BEFORE anything is returned.
"""
from .security import User
DATASETS = [
    {"id": "fin-q3-revenue", "division": "finance", "title": "Q3 revenue ledger", "rows": 1240},
    {"id": "fin-forecast", "division": "finance", "title": "FY forecast model", "rows": 312},
    {"id": "res-trials-2026", "division": "research", "title": "Trial results 2026", "rows": 8850},
    {"id": "res-model-evals", "division": "research", "title": "Model evaluation runs", "rows": 4120},
]
DOCUMENTS = [
    {"id": "d1", "division": "finance", "title": "Revenue policy", "text": "Revenue is recognised when a signed order ships. Q3 revenue grew 12 percent."},
    {"id": "d2", "division": "finance", "title": "Budget memo", "text": "The forecast assumes flat headcount and a 4 percent cost increase."},
    {"id": "d3", "division": "research", "title": "Trial protocol", "text": "Trial 7 enrols 400 participants. Results are embargoed until publication."},
    {"id": "d4", "division": "research", "title": "Eval notes", "text": "Model evaluation shows a 3 point gain on the held-out benchmark."},
]
# Tool -> roles allowed to call it. Same idea an MCP server needs per tool call.
TOOLS = {
    "summarize_dataset": {"viewer", "admin"},
    "export_dataset": {"admin"},
}
AUDIT: list[dict] = []

def log(user: User, action: str, target: str, allowed: bool) -> None:
    AUDIT.append({"user": user.email or user.sub, "action": action, "target": target, "allowed": allowed})
    del AUDIT[:-200]

def visible_datasets(user: User) -> list[dict]:
    return [d for d in DATASETS if user.can_access_division(d["division"])]

def find_dataset(dataset_id: str) -> dict | None:
    return next((d for d in DATASETS if d["id"] == dataset_id), None)

def search_documents(user: User, query: str, limit: int = 3) -> list[dict]:
    allowed = [d for d in DOCUMENTS if user.can_access_division(d["division"])]  # filter first
    terms = {t for t in query.lower().split() if len(t) > 2}
    scored = []
    for doc in allowed:
        haystack = f"{doc['title']} {doc['text']}".lower()
        score = sum(t in haystack for t in terms)
        if score:
            scored.append((score, doc))
    scored.sort(key=lambda p: -p[0])
    return [{"id": d["id"], "title": d["title"], "division": d["division"], "snippet": d["text"]} for _, d in scored[:limit]]

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from . import data
from .config import get_settings
from .security import User, current_user, require_roles
app = FastAPI(title="Authorizer demo API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=get_settings().origins,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}  # public on purpose

@app.get("/api/me")
def me(user: User = Depends(current_user)) -> dict:
    return {
        "sub": user.sub,
        "email": user.email,
        "roles": sorted(user.roles),
        "divisions": sorted(user.divisions),
        "is_admin": user.is_admin,
    }

@app.get("/api/datasets")
def list_datasets(user: User = Depends(current_user)) -> dict:
    items = data.visible_datasets(user)
    data.log(user, "list", "datasets", True)
    return {"datasets": items}

@app.get("/api/datasets/{dataset_id}")
def get_dataset(dataset_id: str, user: User = Depends(current_user)) -> dict:
    ds = data.find_dataset(dataset_id)
    if ds is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "dataset not found")
    allowed = user.can_access_division(ds["division"])
    data.log(user, "read", dataset_id, allowed)
    if not allowed:
        # Resource-level check. Having a valid token is not enough.
        raise HTTPException(status.HTTP_403_FORBIDDEN, f"your account has no access to the {ds['division']} division")
    return ds

class Query(BaseModel):
    question: str

@app.post("/api/rag/query")
def rag_query(body: Query, user: User = Depends(current_user)) -> dict:
    # Permission-aware retrieval: filter by division BEFORE ranking or generating.
    hits = data.search_documents(user, body.question)
    data.log(user, "rag", body.question[:60], True)
    return {"question": body.question, "sources": hits}

class ToolCall(BaseModel):
    dataset_id: str

@app.post("/api/tools/{tool}/invoke")
def invoke_tool(tool: str, body: ToolCall, user: User = Depends(current_user)) -> dict:
    allowed_roles = data.TOOLS.get(tool)
    if allowed_roles is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "unknown tool")
    if not user.roles & allowed_roles:  # check 1: may this role call this tool?
        data.log(user, f"tool:{tool}", body.dataset_id, False)
        raise HTTPException(status.HTTP_403_FORBIDDEN, f"tool '{tool}' requires one of: {', '.join(sorted(allowed_roles))}")
    ds = data.find_dataset(body.dataset_id)
    if ds is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "dataset not found")
    if not user.can_access_division(ds["division"]):  # check 2: may they touch THIS data?
        data.log(user, f"tool:{tool}", body.dataset_id, False)
        raise HTTPException(status.HTTP_403_FORBIDDEN, f"no access to the {ds['division']} division")
    data.log(user, f"tool:{tool}", body.dataset_id, True)
    return {"tool": tool, "dataset": ds["id"], "result": f"{tool} completed on {ds['title']} ({ds['rows']} rows)"}

@app.get("/api/admin/audit")
def audit(user: User = Depends(require_roles("admin"))) -> dict:
    return {"entries": list(reversed(data.AUDIT))}

fastapi>=0.115
uvicorn[standard]>=0.30
pyjwt[crypto]>=2.9
pydantic-settings>=2.4
httpx>=0.27
pytest>=8.0
FROM python:3.12-slim
WORKDIR /srv
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
RUN useradd -r -u 10001 api && chown -R api /srv
USER api
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

"""Runs without Authorizer: we mint tokens with a throwaway RSA key and point the
backend's JWKS lookup at the matching public key. Each test mirrors a step of the
security checklist shown to trainees."""
import time
from types import SimpleNamespace
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi.testclient import TestClient
from app import security
from app.config import get_settings
from app.main import app
ISS = get_settings().issuer
AUD = get_settings().client_id
KID = "test-key"
_private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_other = rsa.generate_private_key(public_exponent=65537, key_size=2048)

class FakeJwks:
    def get_signing_key(self, kid):
        return SimpleNamespace(key=_private.public_key())
    def get_signing_keys(self):
        return [SimpleNamespace(key=_private.public_key())]

@pytest.fixture(autouse=True)
def fake_jwks(monkeypatch):
    monkeypatch.setattr(security, "jwks_client", lambda: FakeJwks())

client = TestClient(app)

def token(roles, *, extra=None, exp=300, aud=AUD, iss=ISS, key=None, token_type="access_token", email="u@demo.test"):
    now = int(time.time())
    claims = {"sub": "user-1", "email": email, "roles": roles, "aud": aud, "iss": iss,
              "iat": now, "exp": now + exp, "token_type": token_type, **(extra or {})}
    return jwt.encode(claims, key or _private, algorithm="RS256", headers={"kid": KID})

def get(path, tok=None):
    return client.get(path, headers={"Authorization": f"Bearer {tok}"} if tok else {})

# --- authentication: 401 ---------------------------------------------------
def test_no_token_is_401():
    assert get("/api/me").status_code == 401

def test_garbage_token_is_401():
    assert get("/api/me", "not.a.jwt").status_code == 401

def test_expired_token_is_401():
    assert get("/api/me", token(["viewer"], exp=-10)).status_code == 401

def test_wrong_audience_is_401():
    assert get("/api/me", token(["viewer"], aud="some-other-app")).status_code == 401

def test_wrong_issuer_is_401():
    assert get("/api/me", token(["viewer"], iss="http://evil.example")).status_code == 401

def test_token_signed_by_another_key_is_401():
    assert get("/api/me", token(["admin"], key=_other)).status_code == 401

def test_id_token_is_not_accepted_as_access_token():
    assert get("/api/me", token(["viewer"], token_type="id_token")).status_code == 401

def test_health_is_public():
    assert get("/health").status_code == 200

# --- authorization: 403 ----------------------------------------------------
def test_viewer_cannot_read_audit_log():
    assert get("/api/admin/audit", token(["viewer"])).status_code == 403

def test_admin_can_read_audit_log():
    assert get("/api/admin/audit", token(["admin", "viewer"])).status_code == 200

def test_division_user_sees_only_own_datasets():
    r = get("/api/datasets", token(["viewer", "div-finance"]))
    assert {d["division"] for d in r.json()["datasets"]} == {"finance"}

def test_user_without_division_sees_nothing():
    assert get("/api/datasets", token(["viewer"])).json()["datasets"] == []

def test_admin_sees_every_division():
    r = get("/api/datasets", token(["admin"]))
    assert {d["division"] for d in r.json()["datasets"]} == {"finance", "research"}

def test_cross_division_read_is_403():
    assert get("/api/datasets/res-trials-2026", token(["viewer", "div-finance"])).status_code == 403

def test_own_division_read_is_200():
    assert get("/api/datasets/fin-q3-revenue", token(["viewer", "div-finance"])).status_code == 200

# --- AI layer: retrieval and tools must apply the same rules ---------------
def test_rag_never_returns_other_divisions():
    r = client.post("/api/rag/query", json={"question": "trial results embargoed enrols"},
                    headers={"Authorization": f"Bearer {token(['viewer', 'div-finance'])}"})
    assert r.status_code == 200
    assert all(s["division"] == "finance" for s in r.json()["sources"])

def test_tool_role_and_division_are_both_checked():
    h = lambda roles: {"Authorization": f"Bearer {token(roles)}"}
    body = {"dataset_id": "fin-q3-revenue"}
    assert client.post("/api/tools/export_dataset/invoke", json=body, headers=h(["viewer", "div-finance"])).status_code == 403
    assert client.post("/api/tools/summarize_dataset/invoke", json={"dataset_id": "res-model-evals"}, headers=h(["viewer", "div-finance"])).status_code == 403
    assert client.post("/api/tools/summarize_dataset/invoke", json=body, headers=h(["viewer", "div-finance"])).status_code == 200

def test_roles_as_comma_separated_string_are_understood():
    r = get("/api/me", token("viewer,div-finance"))
    assert r.status_code == 200 and r.json()["divisions"] == ["finance"]

def test_real_authorizer_shape_active_roles_narrow_allowed_roles_wide():
    """Authorizer: `roles` = active (default) role only, `allowed_roles` = everything assigned."""
    tok = token(["viewer"], extra={"allowed_roles": ["viewer", "div-finance"]})
    assert get("/api/datasets/fin-q3-revenue", tok).status_code == 200
    assert get("/api/datasets/res-trials-2026", tok).status_code == 403
    admin = token(["viewer"], extra={"allowed_roles": ["viewer", "admin"]})
    assert get("/api/admin/audit", admin).status_code == 200

"""Shared helpers for the demo scripts (stdlib + httpx)."""
import base64
import json
import os
import pathlib
import httpx
ROOT = pathlib.Path(__file__).resolve().parent.parent
DEMO_USERS = [
    # email, password, roles the ADMIN grants after sign-up
    ("admin@demo.test", "Demo@12345", ["admin", "viewer"]),
    ("alice@demo.test", "Demo@12345", ["viewer", "div-finance"]),
    ("bob@demo.test", "Demo@12345", ["viewer", "div-research"]),
    ("carol@demo.test", "Demo@12345", ["viewer"]),
]

def load_env() -> dict:
    env = {}
    for line in (ROOT / ".env").read_text().splitlines():
        if line.strip() and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            env[k.strip()] = v.strip()
    return env

def q(value: str) -> str:
    """A JSON string is also a valid GraphQL string literal."""
    return json.dumps(value)

def gql(url: str, query: str, headers: dict | None = None) -> dict:
    # Authorizer >= 2.3.0 has a CSRF guard: state-changing requests need an Origin
    # header that appears in --allowed-origins. Browsers add it automatically; scripts
    # must send one. We borrow the web app's origin, which docker-compose.yml allows.
    merged = {"Origin": os.environ.get("AUTH_ORIGIN", "http://localhost:3000"), **(headers or {})}
    r = httpx.post(f"{url}/graphql", json={"query": query}, headers=merged, timeout=15)
    if r.status_code >= 400:
        raise RuntimeError(f"HTTP {r.status_code} from {url}/graphql: {r.text[:300]}")
    body = r.json()
    if body.get("errors"):
        raise RuntimeError(body["errors"][0]["message"])
    return body["data"]

def claims_of(token: str | None) -> dict:
    try:
        part = (token or "").split(".")[1]
        return json.loads(base64.urlsafe_b64decode(part + "=" * (-len(part) % 4)))
    except Exception:
        return {}

def login(url: str, email: str, password: str) -> dict:
    data = gql(url, f"mutation {{ login(params: {{email: {q(email)}, password: {q(password)}}}) "
                    "{ message access_token id_token expires_in user { id email roles } } }")["login"]
    if not data.get("access_token"):
        # No token means the server withheld it (MFA setup/verification, unverified email, ...).
        raise RuntimeError(f"login returned no access_token; server message: {data.get('message')!r}")
    # Some Authorizer versions return `user: null` here. The tokens always carry the facts.
    if not data.get("user"):
        tc = claims_of(data["access_token"])
        roles = tc.get("allowed_roles") or tc.get("roles") or claims_of(data.get("id_token")).get("roles") or []
        data["user"] = {"email": claims_of(data.get("id_token")).get("email", email), "roles": roles}
    return data

#!/usr/bin/env python3
"""Create demo users, then let the ADMIN assign roles.Users cannot give themselves admin or division roles: those are configured as
--protected-roles in docker-compose.yml. Only the admin API can grant them.
Dev only. In production start Authorizer with --disable-admin-header-auth=true
and manage roles from the dashboard instead of an admin secret in a script.
"""
from _common import DEMO_USERS, gql, load_env, q

def main() -> None:
    env = load_env()
    url = env["AUTHORIZER_URL"].rstrip("/")
    admin = {"x-authorizer-admin-secret": env["ADMIN_SECRET"]}
    for email, password, _ in DEMO_USERS:
        try:
            gql(url, f"mutation {{ signup(params: {{email: {q(email)}, password: {q(password)}, "
                     f"confirm_password: {q(password)}}}) {{ message }} }}")
            print(f"signed up   {email}")
        except RuntimeError as exc:
            print(f"skip signup {email}: {exc}")
    users = gql(url, "{ _users { users { id email roles } } }", admin)["_users"]["users"]
    by_email = {u["email"]: u for u in users}
    for email, _, roles in DEMO_USERS:
        user = by_email.get(email)
        if not user:
            print(f"!! {email} not found")
            continue
        role_list = ", ".join(q(r) for r in roles)
        gql(url, f"mutation {{ _update_user(params: {{id: {q(user['id'])}, roles: [{role_list}]}}) {{ id }} }}", admin)
        print(f"roles set   {email}: {roles}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Automates the trainee security checklist against the running stack."""
import sys
import httpx
from _common import claims_of, gql, load_env, login
API = "http://localhost:8000"
results: list[tuple[str, bool, str]] = []

def check(name: str, ok: bool, detail: str = "") -> None:
    results.append((name, ok, detail))
    print(f"{'PASS' if ok else 'FAIL'}  {name}  {detail}")

def api(path: str, token: str | None = None, show: bool = False) -> int:
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    r = httpx.get(f"{API}{path}", headers=headers, timeout=15)
    if show:
        print(f"      {path} -> {r.status_code} {r.text[:200]}")
    return r.status_code

def main() -> int:
    try:
        httpx.get(f"{API}/health", timeout=5).raise_for_status()
    except httpx.HTTPError as exc:
        print(f"API not reachable at {API} ({exc.__class__.__name__}).\n"
              "Check: docker compose ps   |   docker compose logs backend --tail 30\n"
              "Or run it locally: pip install -r backend/requirements.txt && make api")
        return 2
    env = load_env()
    url = env["AUTHORIZER_URL"].rstrip("/")
    alice = login(url, "alice@demo.test", "Demo@12345")
    check("login with valid credentials", bool(alice["access_token"]), f"roles={alice['user']['roles']}")
    try:
        login(url, "alice@demo.test", "Wrong@12345")
        check("login with wrong password is rejected", False)
    except RuntimeError as exc:
        check("login with wrong password is rejected", True, str(exc))
    tok = alice["access_token"]
    claims = claims_of(tok)
    print("      access-token role claims:", {k: v for k, v in claims.items() if "role" in k})
    print("      access-token iss/aud:", claims.get("iss"), claims.get("aud"))
    api("/api/me", tok, show=True)  # what the API believes about this token
    check("API without token -> 401", api("/api/me") == 401)
    check("API with tampered token -> 401", api("/api/me", tok[:-4] + "AAAA") == 401)
    check("API with valid token -> 200", api("/api/me", tok) == 200)
    check("viewer calling admin API -> 403", api("/api/admin/audit", tok) == 403)
    check("finance user, finance dataset -> 200", api("/api/datasets/fin-q3-revenue", tok, show=True) == 200)
    check("finance user, research dataset -> 403", api("/api/datasets/res-trials-2026", tok) == 403)
    admin = login(url, "admin@demo.test", "Demo@12345")
    check("admin calling admin API -> 200", api("/api/admin/audit", admin["access_token"], show=True) == 200)
    failed = [r for r in results if not r[1]]
    print(f"\n{len(results) - len(failed)}/{len(results)} passed")
    print("Manual: restart Authorizer (`docker compose restart authorizer`) and confirm the same token still works;"
          " then test logout/revocation from the web app.")
    return 1 if failed else 0

if __name__ == "__main__":
    sys.exit(main())