backend/core/subscription/auth_utils.py

"""Shared auth helpers: dev-token gate, Clerk issuer allowlist, JWKS resolution."""

from __future__ import annotations

import os
from typing import Optional, Set


DEV_TEST_TOKEN = "dev_test_token"
DEV_TEST_USER = "test_dev_user"


def is_dev_environment() -> bool:
    env = (os.getenv("ENV") or os.getenv("ENVIRONMENT") or "").strip().lower()
    return env in ("dev", "development", "local")


def is_production_environment() -> bool:
    """Treat unset ENV as production (HF Spaces / fail-closed)."""
    env = (os.getenv("ENV") or os.getenv("ENVIRONMENT") or "").strip().lower()
    if not env:
        return True
    return env in ("prod", "production", "hf", "huggingface")


def is_dev_token_allowed() -> bool:
    """Honor dev_test_token only when explicitly enabled — never via HF defaults."""
    flag = (os.getenv("ALLOW_DEV_TOKEN") or "").strip().lower()
    if flag in ("1", "true", "yes", "on"):
        return True
    return is_dev_environment()


def is_dev_test_token(token: Optional[str]) -> bool:
    return bool(token) and token == DEV_TEST_TOKEN and is_dev_token_allowed()


def get_allowed_clerk_issuers() -> Set[str]:
    """Allowlisted Clerk issuers from env (never from unverified JWT)."""
    issuers: Set[str] = set()
    for key in ("CLERK_ISSUER", "CLERK_FRONTEND_API"):
        raw = (os.getenv(key) or "").strip()
        if raw:
            issuers.add(raw.rstrip("/"))
    multi = (os.getenv("CLERK_ISSUERS") or "").strip()
    if multi:
        for part in multi.split(","):
            part = part.strip().rstrip("/")
            if part:
                issuers.add(part)
    return issuers


def get_configured_jwks_url() -> Optional[str]:
    url = (os.getenv("CLERK_JWKS_URL") or "").strip()
    return url or None


def get_clerk_audience() -> Optional[str]:
    aud = (os.getenv("CLERK_AUDIENCE") or os.getenv("CLERK_JWT_AUDIENCE") or "").strip()
    return aud or None


def resolve_jwks_url_for_issuer(issuer: str) -> str:
    """
    Resolve JWKS URL without trusting attacker-controlled issuers.

    Prefer CLERK_JWKS_URL. Otherwise build JWKS only for allowlisted issuers.
    """
    configured = get_configured_jwks_url()
    if configured:
        return configured

    allowed = get_allowed_clerk_issuers()
    normalized = (issuer or "").rstrip("/")
    if not allowed:
        raise ValueError(
            "CLERK_ISSUER or CLERK_JWKS_URL must be configured; refusing JWKS fetch"
        )
    if normalized not in allowed:
        raise ValueError("Untrusted token issuer")
    return f"{normalized}/.well-known/jwks.json"