core-api/src/auth/oidc.py

"""
OIDC Authentication Module

Provides OAuth2/OIDC token validation for FastAPI using Authentik as IdP.
Implements bearer token authentication with JWT verification.
"""
from fastapi import Depends, HTTPException, Security, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
import httpx
from functools import lru_cache
from typing import Dict, Optional
from src.logging_config import get_logger

logger = get_logger(__name__)
security = HTTPBearer(auto_error=False)


class OIDCConfig:
    """OIDC configuration from environment"""

    def __init__(self):
        # These will be set from environment variables in config.py
        self.enabled = False
        self.issuers: list[str] = []
        self.audiences: list[str] = []

    def configure(self, enabled: bool, issuers: list[str], audiences: list[str]):
        """Configure OIDC settings"""
        self.enabled = enabled
        self.issuers = [iss.rstrip('/') for iss in issuers]  # Normalize without trailing slash
        self.audiences = audiences
        logger.info(f"OIDC configured: enabled={enabled}, issuers={self.issuers}, audiences={audiences}")

    def get_jwks_uri(self, issuer: str) -> str:
        """Get JWKS URI for a specific issuer"""
        return f"{issuer.rstrip('/')}/jwks/"

    def is_valid_issuer(self, issuer: str) -> bool:
        """Check if issuer is in the allowed list"""
        normalized = issuer.rstrip('/')
        return normalized in self.issuers


# Global OIDC config instance
oidc_config = OIDCConfig()


# Per-issuer JWKS cache
_jwks_cache: Dict[str, Dict] = {}


def get_jwks_for_issuer(issuer: str) -> Dict:
    """
    Fetch JSON Web Key Set (JWKS) for a specific issuer.

    Cached per-issuer to avoid repeated requests. Cache is cleared on server restart.

    Args:
        issuer: The token issuer URL

    Returns:
        JWKS dictionary containing public keys for token verification

    Raises:
        HTTPException: If JWKS fetch fails
    """
    if not oidc_config.enabled:
        return {}

    normalized_issuer = issuer.rstrip('/')

    # Return cached JWKS if available
    if normalized_issuer in _jwks_cache:
        return _jwks_cache[normalized_issuer]

    jwks_uri = oidc_config.get_jwks_uri(normalized_issuer)

    try:
        logger.debug(f"Fetching JWKS from {jwks_uri}")
        response = httpx.get(jwks_uri, timeout=10.0)
        response.raise_for_status()
        jwks = response.json()
        logger.info(f"JWKS fetched successfully for {normalized_issuer} ({len(jwks.get('keys', []))} keys)")
        _jwks_cache[normalized_issuer] = jwks
        return jwks
    except Exception as e:
        logger.error(f"Failed to fetch JWKS from {jwks_uri}: {e}")
        raise HTTPException(
            status_code=503,
            detail="Authentication service unavailable"
        )


async def get_current_user(
    credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
) -> Optional[Dict]:
    """
    Validate OIDC token from Authorization: Bearer header

    Extracts and validates JWT token from request header. Verifies:
    - Token signature using JWKS
    - Token expiration
    - Issuer matches Authentik
    - Audience matches core-api

    Args:
        credentials: HTTP Bearer token from Authorization header

    Returns:
        User claims dictionary containing email, name, groups, etc.
        Returns None if OIDC is disabled (allows unauthenticated access)

    Raises:
        HTTPException 401: If token is invalid, expired, or missing when OIDC enabled
    """
    # If OIDC is disabled, allow all requests (no authentication)
    if not oidc_config.enabled:
        logger.debug("OIDC disabled - allowing unauthenticated access")
        return None

    # OIDC enabled - token required
    if not credentials:
        logger.warning("Authentication required but no token provided")
        raise HTTPException(
            status_code=401,
            detail="Authentication required",
            headers={"WWW-Authenticate": "Bearer"},
        )

    token = credentials.credentials

    try:
        # First, extract issuer and audience from unverified claims
        unverified_claims = jwt.get_unverified_claims(token)
        token_issuer = unverified_claims.get("iss", "")
        token_audience = unverified_claims.get("aud", "")

        # Validate issuer is in our allowed list
        if not oidc_config.is_valid_issuer(token_issuer):
            logger.warning(f"Invalid token issuer: {token_issuer}")
            raise HTTPException(status_code=401, detail="Invalid token issuer")

        # Validate audience is in our allowed list
        if token_audience not in oidc_config.audiences:
            logger.warning(f"Invalid token audience: {token_audience}")
            raise HTTPException(status_code=401, detail="Invalid token audience")

        # Decode token header to get key ID
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        if not kid:
            raise HTTPException(status_code=401, detail="Invalid token format")

        # Find matching key in JWKS for this specific issuer
        jwks = get_jwks_for_issuer(token_issuer)
        rsa_key = None

        for key in jwks.get("keys", []):
            if key.get("kid") == kid:
                rsa_key = key
                break

        if not rsa_key:
            logger.warning(f"No matching key found for kid: {kid}")
            raise HTTPException(status_code=401, detail="Invalid token key")

        # Verify and decode token using the token's actual issuer and audience
        payload = jwt.decode(
            token,
            rsa_key,
            algorithms=["RS256"],
            audience=token_audience,  # Use the token's audience (already validated)
            issuer=token_issuer,  # Use the token's issuer (already validated)
        )

        user_email = payload.get("email", "unknown")
        logger.info(f"Authenticated user: {user_email} (issuer: {token_issuer})")

        return payload

    except jwt.ExpiredSignatureError:
        logger.warning("Token expired")
        raise HTTPException(
            status_code=401,
            detail="Token expired",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except jwt.JWTClaimsError as e:
        logger.warning(f"Invalid token claims: {e}")
        raise HTTPException(
            status_code=401,
            detail="Invalid token claims",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except JWTError as e:
        logger.error(f"JWT validation error: {e}")
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except Exception as e:
        logger.error(f"Unexpected authentication error: {e}")
        raise HTTPException(
            status_code=500,
            detail="Authentication error",
        )


async def get_admin_user(
    user: Optional[Dict] = Depends(get_current_user)
) -> Dict:
    """
    Require admin group membership

    Use this dependency for endpoints that require admin access.
    Checks if user is member of 'admin' group in Authentik.

    Args:
        user: User claims from get_current_user

    Returns:
        User claims dictionary if user is admin

    Raises:
        HTTPException 403: If user is not in admin group
        HTTPException 401: If OIDC enabled but user not authenticated
    """
    # If OIDC disabled, allow all (backward compatibility)
    if not oidc_config.enabled or user is None:
        logger.debug("OIDC disabled - allowing admin access")
        return {"email": "unauthenticated", "groups": ["admin"]}

    # Check admin group membership
    groups = user.get("groups", [])

    if "admin" not in groups and "authentik Admins" not in groups:
        user_email = user.get("email", "unknown")
        logger.warning(f"User {user_email} attempted admin access (groups: {groups})")
        raise HTTPException(
            status_code=403,
            detail="Admin access required"
        )

    return user


async def get_optional_user(
    credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
) -> Optional[Dict]:
    """
    Optional authentication - allows both authenticated and unauthenticated access

    Use for endpoints that should be accessible to everyone but can provide
    enhanced functionality for authenticated users.

    Args:
        credentials: HTTP Bearer token from Authorization header

    Returns:
        User claims if valid token provided, None otherwise
    """
    if not credentials or not oidc_config.enabled:
        return None

    try:
        return await get_current_user(credentials)
    except HTTPException:
        # Invalid token - return None instead of raising
        return None


async def get_forward_auth_user(
    request: Request
) -> Optional[Dict]:
    """
    Authentik Forward Auth authentication for external access via NPM

    This dependency allows:
    - External access through api.schweitz.net (with Authentik forward auth headers) - REQUIRES authentication
    - Internal direct access (no forward auth headers) - ALLOWED without authentication

    When accessing through NPM with Authentik forward auth enabled, NPM adds headers like:
    - X-authentik-username
    - X-authentik-email
    - X-authentik-groups
    - X-authentik-name
    - X-authentik-uid

    Args:
        request: FastAPI request object containing headers

    Returns:
        User info dict if authenticated via forward auth headers
        None if accessed internally (no forward auth headers)

    Raises:
        HTTPException 401: If forward auth headers present but invalid/incomplete
    """
    # Check for Authentik forward auth headers
    username = request.headers.get("x-authentik-username")
    email = request.headers.get("x-authentik-email")
    groups = request.headers.get("x-authentik-groups")
    name = request.headers.get("x-authentik-name")
    uid = request.headers.get("x-authentik-uid")

    # If NO forward auth headers present, this is internal access - allow it
    if not username and not email:
        logger.debug("No forward auth headers - allowing internal access")
        return None

    # Forward auth headers present (external access via api.schweitz.net)
    # Validate authentication
    if not username or not email:
        logger.warning("Incomplete forward auth headers detected")
        raise HTTPException(
            status_code=401,
            detail="Authentication required - incomplete forward auth headers"
        )

    # Parse groups (comma-separated string to list)
    groups_list = [g.strip() for g in groups.split(",")] if groups else []

    user_info = {
        "username": username,
        "email": email,
        "name": name or username,
        "groups": groups_list,
        "uid": uid,
        "auth_method": "forward_auth"
    }

    logger.info(f"Authenticated via forward auth: {email} (groups: {groups_list})")
    return user_info


async def get_forward_auth_admin(
    user: Optional[Dict] = Depends(get_forward_auth_user)
) -> Dict:
    """
    Require admin access for external requests, allow all internal requests

    Use this dependency for endpoints that require admin access when accessed
    externally through api.schweitz.net, but allow unrestricted internal access.

    Args:
        user: User info from get_forward_auth_user

    Returns:
        User info dict if user is admin or if accessed internally

    Raises:
        HTTPException 403: If external user is not in admin/authentik Admins group
    """
    # Internal access (no forward auth headers) - allow all
    if user is None:
        logger.debug("Internal access - allowing without admin check")
        return {"email": "internal", "groups": ["admin"], "auth_method": "internal"}

    # External access - check admin group membership
    groups = user.get("groups", [])

    if "admin" not in groups and "authentik Admins" not in groups:
        user_email = user.get("email", "unknown")
        logger.warning(f"User {user_email} attempted admin access (groups: {groups})")
        raise HTTPException(
            status_code=403,
            detail="Admin access required"
        )

    return user