nike/docs/PATTERN_Casdoor_FastAPI.md

# Casdoor SSO Authentication for FastAPI Pattern v1.0.0

Standardises how FastAPI applications integrate with Casdoor for Single Sign-On authentication, supporting both browser-based OIDC (via OAuth2-Proxy) and programmatic JWT access (via Casdoor Python SDK).

## 🐾 Red Panda Approval™

This pattern follows Red Panda Approval standards.

---

## Why a Pattern, Not a Shared Library

FastAPI applications have varied authentication needs that make a single shared auth library impractical:

- A **location manager** (Periplus) needs owner/guest roles with collection-level sharing
- A **task manager** (Kairos) needs Django-allauth with OIDC, not a FastAPI SDK
- A **CMS** (Angelia) needs Django-native OIDC callbacks
- An **MCP server** needs headless JWT-only auth without browser sessions

Instead, this pattern defines:

- **Required settings** — every FastAPI app must configure for Casdoor
- **Required auth dependency** — the `get_current_user` resolution chain
- **Required infrastructure** — OAuth2-Proxy sidecar + Casdoor app registration
- **Standard callback URL rules** — for interoperability
- **Anti-patterns** — hard-won lessons from production debugging

---

## Architecture

FastAPI applications use **dual-layer authentication**:

```
┌──────────┐     ┌───────────┐     ┌───────────────┐     ┌─────────────┐
│  Browser  │────▶│  HAProxy  │────▶│  OAuth2-Proxy │────▶│  FastAPI App │
│           │     │  (TLS)    │     │  (sidecar)    │     │             │
└──────────┘     └─────┬─────┘     └───────┬───────┘     └──────┬──────┘
                       │                   │                     │
                       │     ┌─────────────▼─────────────┐      │
                       └────▶│        Casdoor             │◀─────┘
                             │  (OIDC Provider)           │  JWT validation
                             └───────────────────────────┘
```

| Layer | Handles | Method |
|-------|---------|--------|
| **OAuth2-Proxy** (sidecar) | Browser sessions | OIDC cookie → `X-Forwarded-*` headers |
| **Casdoor SDK** (in-app) | API/MCP clients | `Authorization: Bearer <JWT>` |
| **Dev mode** | Local development | No auth required |

The FastAPI app is **never directly exposed** — it only accepts connections from OAuth2-Proxy within the Docker compose network.

---

## Required Settings

Every FastAPI app using Casdoor must define these settings via `pydantic-settings`. Use the app-specific `env_prefix` to namespace variables.

```python
# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # ── Casdoor SSO ──────────────────────────────────────────
    # Set APPNAME_CASDOOR_ENABLED=false for local dev (no auth required)
    casdoor_enabled: bool = False
    casdoor_endpoint: str = "https://sso.example.com"
    casdoor_client_id: str = ""
    casdoor_client_secret: str = ""
    # PEM-encoded X.509 certificate from Casdoor → Certs → your cert
    casdoor_certificate: str = ""
    casdoor_org_name: str = ""
    casdoor_app_name: str = ""

    # Owner identity (Casdoor username for the app owner)
    owner_name: str = ""

    model_config = {"env_prefix": "APPNAME_", "env_file": ".env"}


settings = Settings()
```

### Environment Variables (Production)

| Variable | Description | Example |
|----------|-------------|---------|
| `APPNAME_CASDOOR_ENABLED` | Enable/disable SSO | `true` |
| `APPNAME_CASDOOR_ENDPOINT` | Casdoor public URL | `https://id.d.helu.ca` |
| `APPNAME_CASDOOR_CLIENT_ID` | OAuth2 client ID | From OCI Vault |
| `APPNAME_CASDOOR_CLIENT_SECRET` | OAuth2 client secret | From OCI Vault |
| `APPNAME_CASDOOR_CERTIFICATE` | PEM X.509 cert for JWT validation | From Casdoor admin |
| `APPNAME_CASDOOR_ORG_NAME` | Casdoor organization | `heluca` |
| `APPNAME_CASDOOR_APP_NAME` | Casdoor application name | `periplus` |
| `APPNAME_OWNER_NAME` | Casdoor username for the app owner | `r@helu.ca` |

---

## Casdoor SDK Singleton

Initialise the `AsyncCasdoorSDK` once and reuse it across requests.

```python
# auth.py
from casdoor import AsyncCasdoorSDK

_sdk: AsyncCasdoorSDK | None = None


def _get_sdk() -> AsyncCasdoorSDK:
    global _sdk
    if _sdk is None:
        if not settings.casdoor_certificate:
            raise RuntimeError(
                "APPNAME_CASDOOR_CERTIFICATE is not configured. "
                "Set it to the PEM certificate from your Casdoor application."
            )
        _sdk = AsyncCasdoorSDK(
            endpoint=settings.casdoor_endpoint,
            client_id=settings.casdoor_client_id,
            client_secret=settings.casdoor_client_secret,
            certificate=settings.casdoor_certificate.encode()
            if isinstance(settings.casdoor_certificate, str)
            else settings.casdoor_certificate,
            org_name=settings.casdoor_org_name,
            application_name=settings.casdoor_app_name,
        )
    return _sdk


def get_sdk() -> AsyncCasdoorSDK:
    """FastAPI dependency — returns the configured Casdoor SDK."""
    return _get_sdk()
```

---

## Required Auth Dependency: `get_current_user`

The core dependency resolves the authenticated user by checking methods in priority order:

```python
from typing import Annotated
from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

_bearer = HTTPBearer(auto_error=False)


async def get_current_user(
    request: Request,
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)] = None,
    session: AsyncSession = Depends(get_session),
) -> User:
    if not settings.casdoor_enabled:
        return await _get_or_create_dev_owner(session)

    # Method 1: Bearer JWT (API/MCP clients)
    if credentials is not None:
        token = credentials.credentials
        try:
            sdk = _get_sdk()
            claims = sdk.parse_jwt_token(token)
        except Exception as exc:
            raise HTTPException(status_code=401, detail=f"Invalid token: {exc}") from exc

        sub = claims.get("sub") or claims.get("name") or ""
        if not sub:
            raise HTTPException(status_code=401, detail="Token missing subject claim")

        name = claims.get("displayName") or claims.get("name") or sub
        email = claims.get("email") or None
        return await _find_or_create_user(session, sub, name, email)

    # Method 2: OAuth2-Proxy forwarded headers (browser users)
    forwarded_email = request.headers.get("X-Forwarded-Email")
    forwarded_user = request.headers.get("X-Forwarded-User")
    forwarded_name = request.headers.get("X-Forwarded-Preferred-Username")

    if forwarded_email or forwarded_user:
        sub = forwarded_user or forwarded_email or ""
        name = forwarded_name or forwarded_user or forwarded_email or sub
        email = forwarded_email or None
        return await _find_or_create_user(session, sub, name, email)

    raise HTTPException(status_code=401, detail="Not authenticated")


async def _get_or_create_dev_owner(session: AsyncSession) -> User:
    """Return the dev-mode owner user row, creating it if it doesn't exist."""
    result = await session.execute(
        select(User).where(User.casdoor_sub == _DEV_OWNER_SUB)
    )
    user = result.scalar_one_or_none()
    if user is None:
        user = User(name="Owner", casdoor_sub=_DEV_OWNER_SUB)
        session.add(user)
        await session.commit()
        await session.refresh(user)
    return user


async def _find_or_create_user(
    session: AsyncSession,
    casdoor_sub: str,
    name: str,
    email: str | None,
) -> User:
    """Look up a user by casdoor_sub; create a new row if first login."""
    result = await session.execute(
        select(User).where(User.casdoor_sub == casdoor_sub)
    )
    user = result.scalar_one_or_none()
    if user is not None:
        return user

    # First-time login — provision a new user row
    user = User(name=name, email=email, casdoor_sub=casdoor_sub)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


# Type alias for use in route signatures
CurrentUser = Annotated[User, Depends(get_current_user)]
```

### Authentication Priority Chain

> **Note on code execution order:** Dev mode is checked *first* in `get_current_user` (before JWT and headers) and short-circuits immediately. The table below shows conceptual priority for production use when `CASDOOR_ENABLED=true`.

| Priority | Method | Source | When Used |
|----------|--------|--------|-----------|
| 1 | Bearer JWT | `Authorization: Bearer <token>` | API clients, MCP |
| 2 | OAuth2-Proxy headers | `X-Forwarded-Email`, `X-Forwarded-User` | Browser sessions |
| — | Dev mode | No auth (short-circuits first) | `CASDOOR_ENABLED=false` |

---

## Auth Router Endpoints

Four standard endpoints for the direct OIDC flow (used by API/MCP clients that need tokens):

```python
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import JSONResponse, RedirectResponse

router = APIRouter(prefix="/auth", tags=["auth"])


@router.get("/login")
async def login(request: Request, redirect_uri: str = Query(None)):
    """Redirect the browser to the Casdoor authorization page."""
    if not settings.casdoor_enabled:
        raise HTTPException(400, "Casdoor SSO is not enabled")

    sdk = get_sdk()
    # ⚠️  CRITICAL: derive callback from request.base_url, NOT casdoor_endpoint
    callback = redirect_uri or f"{request.base_url}auth/callback"
    auth_url = await sdk.get_auth_link(
        redirect_uri=callback,
        response_type="code",
        scope="openid profile email",
    )
    return RedirectResponse(url=auth_url)


@router.get("/callback")
async def callback(
    code: str = Query(...),
    state: str = Query(None),
    redirect_uri: str = Query(None),
):
    """Exchange authorization code for tokens."""
    if not settings.casdoor_enabled:
        raise HTTPException(400, "Casdoor SSO is not enabled")

    sdk = get_sdk()
    try:
        # ⚠️  CRITICAL: method is get_oauth_token(), NOT async_get_oauth_token()
        token = await sdk.get_oauth_token(code=code)
    except Exception as exc:
        raise HTTPException(400, f"Token exchange failed: {exc}") from exc

    access_token = token.get("access_token", "")
    return RedirectResponse(url=f"/#token={access_token}")


@router.get("/me")
async def me(user: CurrentUser):
    """Return current authenticated user's profile."""
    return JSONResponse({
        "id": str(user.id),
        "name": user.name,
        "email": user.email,
        "is_owner": is_owner(user),
    })


@router.get("/logout")
async def logout():
    """Redirect to Casdoor logout."""
    if not settings.casdoor_enabled:
        return RedirectResponse(url="/")

    logout_url = (
        f"{settings.casdoor_endpoint}/login/oauth/logout"
        f"?client_id={settings.casdoor_client_id}"
        f"&post_logout_redirect_uri=/"
    )
    return RedirectResponse(url=logout_url)
```

---

## User Model

The user model stores three identity fields from Casdoor:

- **`name`** — The Casdoor username (stable, human-readable). Used for **owner matching** via `APPNAME_OWNER_NAME`.
- **`email`** — The user's email address.
- **`casdoor_sub`** — The Casdoor `sub` claim (an opaque UUID). Used as the lookup key for `_find_or_create_user()` to de-duplicate logins. **Not used for owner matching** — it is arbitrary and may change on Casdoor redeploy.

```python
from sqlalchemy import Column, DateTime, String, text
from sqlalchemy.dialects.postgresql import UUID

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
    name = Column(String, nullable=False)          # ← owner matching key
    email = Column(String, unique=True)
    casdoor_sub = Column(String, unique=True, nullable=True)  # Casdoor subject claim — used to map SSO identity to local user row
    created_at = Column(DateTime(timezone=True), server_default=text("now()"))
    updated_at = Column(DateTime(timezone=True), server_default=text("now()"))
```

Users are auto-provisioned on first login via `_find_or_create_user()`.

---

## OAuth2-Proxy Sidecar (Docker Compose)

The production `docker-compose.yml.j2` template for the OAuth2-Proxy sidecar:

```yaml
services:
  myapp:
    image: git.helu.ca/r/myapp:latest
    expose:
      - "8000"     # Internal only — NOT published to host
    restart: unless-stopped
    networks:
      - internal

  oauth2-proxy:
    image: quay.io/oauth2-proxy/oauth2-proxy:v7.6.0
    environment:
      OAUTH2_PROXY_PROVIDER: oidc
      OAUTH2_PROXY_OIDC_ISSUER_URL: "https://id.d.helu.ca"
      OAUTH2_PROXY_CLIENT_ID: "{{ app_casdoor_client_id }}"
      OAUTH2_PROXY_CLIENT_SECRET: "{{ app_casdoor_client_secret }}"
      OAUTH2_PROXY_COOKIE_SECRET: "{{ app_oauth2_cookie_secret }}"
      OAUTH2_PROXY_COOKIE_SECURE: "true"
      OAUTH2_PROXY_COOKIE_NAME: _myapp_session
      OAUTH2_PROXY_HTTP_ADDRESS: 0.0.0.0:4180
      # ⚠️  MUST use PLURAL form — see Anti-Patterns
      OAUTH2_PROXY_UPSTREAMS: http://myapp:8000
      OAUTH2_PROXY_REDIRECT_URL: "https://myapp.d.helu.ca/oauth2/callback"
      OAUTH2_PROXY_EMAIL_DOMAINS: "*"
      OAUTH2_PROXY_SKIP_PROVIDER_BUTTON: "true"
      OAUTH2_PROXY_SET_XAUTHREQUEST: "true"
      OAUTH2_PROXY_REVERSE_PROXY: "true"
      OAUTH2_PROXY_REAL_CLIENT_IP_HEADER: X-Forwarded-For
      OAUTH2_PROXY_SKIP_AUTH_ROUTES: "^/live/$,^/ready/$"
      OAUTH2_PROXY_INSECURE_OIDC_ALLOW_UNVERIFIED_EMAIL: "true"
    ports:
      - "{{ app_web_port }}:4180"
    depends_on:
      - myapp
    restart: unless-stopped
    networks:
      - internal

networks:
  internal:
```

---

## Casdoor Application Registration

Register the application in `ansible/casdoor/init_data.json.j2`:

```json
{
  "owner": "admin",
  "name": "myapp",
  "displayName": "My Application",
  "organization": "heluca",
  "cert": "cert-heluca",
  "clientId": "{{ myapp_casdoor_client_id }}",
  "clientSecret": "{{ myapp_casdoor_client_secret }}",
  "grantTypes": ["authorization_code", "refresh_token"],
  "redirectUris": [
    "https://myapp.d.helu.ca/oauth2/callback",
    "https://myapp.d.helu.ca/auth/callback"
  ],
  "tokenFormat": "JWT",
  "expireInHours": 168
}
```

### Redirect URIs (Both Required)

| URI | Purpose |
|-----|---------|
| `https://myapp.d.helu.ca/oauth2/callback` | OAuth2-Proxy browser flow |
| `https://myapp.d.helu.ca/auth/callback` | Direct SDK flow (API/MCP token acquisition) |

Both URIs must be registered in Casdoor's `redirectUris` array. Missing either one causes a 404 or Casdoor redirect error.

> **Note:** `init_data.json` only applies during **initial Casdoor setup**. For existing deployments, also add the redirect URI via the Casdoor admin UI: Applications → myapp → Redirect URIs.

---

## Uvicorn Proxy Headers

The Dockerfile **must** include `--proxy-headers` and `--forwarded-allow-ips=*` so that `request.base_url` resolves to the external URL (e.g., `https://myapp.d.helu.ca/`) rather than the internal Docker URL:

```dockerfile
CMD ["sh", "-c", "alembic upgrade head && uvicorn myapp.main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips=*"]
```

This works because HAProxy sets `X-Forwarded-Proto: https` and `X-Forwarded-Host`, and Uvicorn with `--proxy-headers` honours these.

---

## Database Migrations at Startup

Include `alembic upgrade head` in the container entrypoint so the database schema is always current:

```dockerfile
CMD ["sh", "-c", "alembic upgrade head && uvicorn myapp.main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips=*"]
```

Without this, a fresh database deployment fails with `relation "users" does not exist`.

---

## Role Helpers

Standard owner/guest role pattern:

```python
def is_owner(user: User) -> bool:
    """Check if the user is the owner."""
    if not settings.casdoor_enabled:
        return user.casdoor_sub == _DEV_OWNER_SUB
    return bool(settings.owner_name and user.name == settings.owner_name)


async def require_owner(user: CurrentUser) -> User:
    """Dependency — 403 unless the caller is the owner."""
    if not is_owner(user):
        raise HTTPException(status_code=403, detail="Owner access required")
    return user


OwnerUser = Annotated[User, Depends(require_owner)]
```

Use `CurrentUser` for any-authenticated-user routes, `OwnerUser` for owner-only routes.

---

## Anti-Patterns

### ❌ Don't match owner by `casdoor_sub`

```python
# WRONG — casdoor_sub is an opaque UUID that may change on Casdoor redeploy
return user.casdoor_sub == settings.owner_sub

# CORRECT — match on user.name (the stable Casdoor username)
return user.name == settings.owner_name
```

The `casdoor_sub` (OIDC `sub` claim) is an arbitrary UUID assigned by Casdoor. It can change if the Casdoor database is recreated. The `name` field is the Casdoor username — stable, human-readable, and what operators naturally configure.

### ❌ Don't use `casdoor_endpoint` for callback URLs

```python
# WRONG — redirects to Casdoor, which returns 404
callback = f"{settings.casdoor_endpoint}/auth/callback"

# CORRECT — redirects back to the application itself
callback = f"{request.base_url}auth/callback"
```

The callback URL must point to your **application**, not to Casdoor. After the user authenticates on Casdoor, Casdoor redirects the browser to the registered `redirect_uri` with an authorization code. That URI must be a route your app handles.

### ❌ Don't use `async_get_oauth_token()`

```python
# WRONG — method does not exist on AsyncCasdoorSDK
token = await sdk.async_get_oauth_token(code=code, redirect_uri=url)

# CORRECT — the async SDK uses the same method name as the sync SDK
token = await sdk.get_oauth_token(code=code)
```

The Casdoor Python SDK's `AsyncCasdoorSDK` uses `get_oauth_token()`, not a prefixed async variant. The SDK handles the async internally.

### ❌ Don't use singular oauth2-proxy env var names

```python
# WRONG — silently ignored, causes 404 or crash
OAUTH2_PROXY_EMAIL_DOMAIN: "*"     # ← singular
OAUTH2_PROXY_UPSTREAM: http://app  # ← singular

# CORRECT — must use plural form
OAUTH2_PROXY_EMAIL_DOMAINS: "*"    # ← plural
OAUTH2_PROXY_UPSTREAMS: http://app # ← plural
```

This is a [known oauth2-proxy bug (#253)](https://github.com/oauth2-proxy/oauth2-proxy/issues/253). The singular form is silently ignored, causing `EMAIL_DOMAIN` to crash with "missing email validation" and `UPSTREAM` to return 404 for all requests.

### ❌ Don't skip database migrations in the container entrypoint

Without `alembic upgrade head` in the startup command, the first deployment to a fresh database will fail with `relation "users" does not exist`.

### ❌ Don't expose the app container port to the host

The FastAPI app should only use `expose` (Docker-internal), never `ports` (host-published). All external traffic must flow through OAuth2-Proxy.

### ❌ Don't hardcode `state` parameter values

If you see `state=appname` in the callback URL (instead of a random cryptographic nonce), your application code is likely constructing the authorization URL incorrectly. The `state` parameter should be generated by OAuth2-Proxy or the SDK, never hardcoded.

---

## Terraform Secrets

Standard secrets to provision in `terraform/oci/secrets.tf` for each new FastAPI app:

| Resource | Secret Name | Description |
|----------|-------------|-------------|
| `random_id` | `env-appname-casdoor-client-id` | OAuth2 client ID (hex) |
| `random_password` | `env-appname-casdoor-client-secret` | OAuth2 client secret |
| `random_password` | `env-appname-oauth2-cookie-secret` | Cookie encryption (32 chars) |
| `random_password` | `env-appname-db-password` | Database password |
| Placeholder | `env-appname-casdoor-certificate` | Updated manually after Casdoor deploy |
| Placeholder | `env-appname-owner-name` | Casdoor username of the app owner |

---

## Testing

### Dev Mode (No Auth)

```python
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_dev_mode_no_auth_required(client: AsyncClient):
    """When casdoor_enabled=false, requests succeed without authentication."""
    response = await client.get("/api/v1/collections")
    assert response.status_code == 200
```

### Bearer JWT Validation

```python
@pytest.mark.asyncio
async def test_invalid_bearer_returns_401(client: AsyncClient):
    """Invalid JWT token returns 401."""
    response = await client.get(
        "/api/v1/collections",
        headers={"Authorization": "Bearer invalid-token"},
    )
    assert response.status_code == 401
```

### OAuth2-Proxy Header Auth

```python
@pytest.mark.asyncio
async def test_proxy_headers_create_user(client: AsyncClient):
    """X-Forwarded-Email/User headers auto-provision a user."""
    response = await client.get(
        "/api/v1/collections",
        headers={
            "X-Forwarded-Email": "test@example.com",
            "X-Forwarded-User": "test-user-sub",
        },
    )
    assert response.status_code == 200
```

### Callback URL Derivation

```python
@pytest.mark.asyncio
async def test_login_callback_uses_app_url(client: AsyncClient):
    """The /auth/login endpoint must redirect with the app's URL, not Casdoor's."""
    response = await client.get("/auth/login", follow_redirects=False)
    assert response.status_code == 307
    location = response.headers["location"]
    # The redirect_uri parameter must contain the APP's domain
    assert "myapp.d.helu.ca/auth/callback" in location
    # Must NOT contain the Casdoor domain as callback
    assert "id.d.helu.ca/auth/callback" not in location
```

---

## Checklist for New FastAPI + Casdoor Application

- [ ] Add Terraform secrets (`client-id`, `client-secret`, `cookie-secret`, `certificate`, `owner-name`, `db-password`)
- [ ] Add application entry in `ansible/casdoor/init_data.json.j2` with both `redirectUris`
- [ ] Add `redirectUris` in Casdoor admin UI (for existing deployments)
- [ ] Create `config.py` with `pydantic-settings` and `APPNAME_` prefix
- [ ] Create `auth.py` with SDK singleton and `get_current_user` dependency
- [ ] Create `routers/auth.py` with `/login`, `/callback`, `/me`, `/logout`
- [ ] Create `User` model with `casdoor_sub` column
- [ ] Create Alembic migration for users table
- [ ] Configure Dockerfile with `alembic upgrade head &&` in CMD
- [ ] Configure `--proxy-headers --forwarded-allow-ips=*` in uvicorn
- [ ] Create `docker-compose.yml.j2` with OAuth2-Proxy sidecar (use **plural** env vars)
- [ ] Add HAProxy backend in `hippocamp.helu.ca.yml`
- [ ] Add Ansible deploy/remove playbooks
- [ ] Add OIDC skip-auth routes for health endpoints (`/live/`, `/ready/`)
- [ ] Test: dev mode works without auth
- [ ] Test: callback URL points to app, not Casdoor
- [ ] Update Casdoor certificate secret after first deploy
- [ ] Set `APPNAME_OWNER_NAME` to the Casdoor username of the app owner