admin-panel/app/oidc.py

"""OIDC client for Cloudron's OpenID Connect addon.

Implements the authorization-code flow against Cloudron's built-in OIDC
provider, so the app no longer depends on the proxyAuth header wall.

Cloudron exports these env vars (they change on every restart — read per-call,
never cache at import):

  CLOUDRON_OIDC_ISSUER           e.g. https://my.inference.coop/openid
  CLOUDRON_OIDC_AUTH_ENDPOINT    authorization endpoint
  CLOUDRON_OIDC_TOKEN_ENDPOINT   token endpoint
  CLOUDRON_OIDC_KEYS_ENDPOINT    JWKS endpoint (RS256/EdDSA keys)
  CLOUDRON_OIDC_PROFILE_ENDPOINT userinfo endpoint
  CLOUDRON_OIDC_CLIENT_ID        client id
  CLOUDRON_OIDC_CLIENT_SECRET    client secret
  CLOUDRON_APP_DOMAIN            the app's public domain

Identity: Cloudron's `sub` claim is the username (the unique user identifier).
The `email` scope adds `email`/`email_verified`; `profile` adds `name`,
`preferred_username`, etc. We use `sub` (username) as the identity, matching
the app's ADMIN_USERNAMES allowlist.
"""

import os
import secrets as _secrets
import logging
import urllib.parse

import httpx
from authlib.jose import JsonWebToken, JsonWebKey
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired

logger = logging.getLogger("admin-panel")

# Session cookie name + max age (8 hours, roughly a working day).
SESSION_COOKIE = "admin_oidc_session"
SESSION_MAX_AGE = 60 * 60 * 8

SCOPES = "openid profile email"


def _oidc_env(name: str) -> str:
    return os.environ.get(name, "").strip()


def is_oidc_configured() -> bool:
    """True when Cloudron has injected the OIDC addon env vars."""
    return bool(
        _oidc_env("CLOUDRON_OIDC_CLIENT_ID")
        and _oidc_env("CLOUDRON_OIDC_CLIENT_SECRET")
        and _oidc_env("CLOUDRON_OIDC_AUTH_ENDPOINT")
    )


def _session_serializer() -> URLSafeTimedSerializer:
    # Derive a stable, secret signing key from the OIDC client secret (already
    # a secret the app holds, and stable across restarts).
    secret = _oidc_env("CLOUDRON_OIDC_CLIENT_SECRET") or _secrets.token_urlsafe(32)
    return URLSafeTimedSerializer(secret, salt="admin-panel-oidc-session")


def _redirect_uri(request_host: str) -> str:
    domain = _oidc_env("CLOUDRON_APP_DOMAIN")
    host = domain or request_host
    scheme = "https"
    return f"{scheme}://{host}/auth/openid/callback"


def build_login_url(request_host: str) -> tuple[str, str, str]:
    """Return (login_url, state, nonce). Caller stores state+nonce in cookies."""
    state = _secrets.token_urlsafe(24)
    nonce = _secrets.token_urlsafe(24)
    params = {
        "response_type": "code",
        "client_id": _oidc_env("CLOUDRON_OIDC_CLIENT_ID"),
        "redirect_uri": _redirect_uri(request_host),
        "scope": SCOPES,
        "state": state,
        "nonce": nonce,
    }
    url = f"{_oidc_env('CLOUDRON_OIDC_AUTH_ENDPOINT')}?{urllib.parse.urlencode(params)}"
    return url, state, nonce


async def exchange_code(code: str, request_host: str, nonce: str) -> str:
    """Exchange an authorization code for an ID token, returning the username.

    Validates the ID token signature (JWKS), issuer, audience, nonce, and
    expiry. Returns the `sub` claim (Cloudron = username) on success.
    Raises ValueError on any failure.
    """
    issuer = _oidc_env("CLOUDRON_OIDC_ISSUER")
    client_id = _oidc_env("CLOUDRON_OIDC_CLIENT_ID")
    client_secret = _oidc_env("CLOUDRON_OIDC_CLIENT_SECRET")
    token_endpoint = _oidc_env("CLOUDRON_OIDC_TOKEN_ENDPOINT")
    keys_endpoint = _oidc_env("CLOUDRON_OIDC_KEYS_ENDPOINT")

    token_resp = None
    id_token = None
    async with httpx.AsyncClient(timeout=20.0) as client:
        # Token exchange (client_secret_post).
        token_resp = await client.post(
            token_endpoint,
            data={
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": _redirect_uri(request_host),
                "client_id": client_id,
                "client_secret": client_secret,
            },
        )
        if token_resp.status_code != 200:
            logger.warning("OIDC token exchange failed: %s %s", token_resp.status_code, token_resp.text[:200])
            raise ValueError("token exchange failed")
        id_token = token_resp.json().get("id_token")
        if not id_token:
            logger.warning("OIDC token response missing id_token")
            raise ValueError("missing id_token")

        # Fetch JWKS.
        keys_resp = await client.get(keys_endpoint)
        if keys_resp.status_code != 200:
            logger.warning("OIDC JWKS fetch failed: %s", keys_resp.status_code)
            raise ValueError("jwks fetch failed")
        jwks = keys_resp.json()

    # Validate signature + standard claims (exp/nbf) via claims.validate(),
    # and check iss/aud/nonce manually (authlib's JWTClaims.validate() only
    # handles exp/nbf; issuer/audience/nonce are checked explicitly).
    try:
        jwk_set = JsonWebKey.import_key_set(jwks)
        jwt = JsonWebToken(["RS256", "EdDSA"])
        claims = jwt.decode(id_token, jwk_set)
        claims.validate()  # validates exp + nbf
        # Issuer + audience + nonce (replay protection).
        if claims.get("iss") != issuer:
            logger.warning("OIDC id_token issuer mismatch: %s", claims.get("iss"))
            raise ValueError("issuer mismatch")
        if claims.get("aud") != client_id:
            logger.warning("OIDC id_token audience mismatch: %s", claims.get("aud"))
            raise ValueError("audience mismatch")
        if claims.get("nonce") != nonce:
            logger.warning("OIDC id_token nonce mismatch")
            raise ValueError("nonce mismatch")
    except Exception as e:
        logger.warning("OIDC id_token validation failed: %s", e)
        raise ValueError("id_token validation failed") from e

    username = claims.get("sub")
    if not username:
        logger.warning("OIDC id_token missing sub claim")
        raise ValueError("missing sub claim")

    return str(username)


def write_session(identity: str) -> str:
    """Create a signed session token for the given identity (username)."""
    return _session_serializer().dumps({"identity": identity})


def read_session(token: str) -> str | None:
    """Return the identity from a signed session token, or None if invalid."""
    if not token:
        return None
    try:
        data = _session_serializer().loads(token, max_age=SESSION_MAX_AGE)
        return data.get("identity") if isinstance(data, dict) else None
    except (BadSignature, SignatureExpired):
        return None