vulncheck/app/auth/strategies/oidc_strategy.py

"""
OAuth 2.0 / OpenID Connect strategy.

Uses Authlib (authlib.integrations.starlette_client) for the heavy lifting:
- Discovery (.well-known/openid-configuration)
- PKCE (S256) on the authorization request
- ID-Token validation (signature via JWKS, iss, aud, exp, nbf, nonce)
- Userinfo fetch for claims not present in ID-Token

This strategy does NOT do credential auth — it's a callback-only strategy.
The router (app/routers/auth_oidc.py) drives the flow and hands the
verified `id_token`/`userinfo` payload to `parse_userinfo()` here.
"""
from __future__ import annotations

import logging
import os
from typing import List

from app.auth.strategies.base import AuthError, AuthStrategy, ExternalIdentity
from app.models.user import AuthProvider

logger = logging.getLogger(__name__)


def _split_groups(value) -> List[str]:
    """Groups claim can be list-of-strings (Entra), list-of-objs (some Keycloaks), or comma-string."""
    if value is None:
        return []
    if isinstance(value, list):
        out = []
        for v in value:
            if isinstance(v, str):
                out.append(v)
            elif isinstance(v, dict):
                name = v.get("name") or v.get("displayName") or v.get("id")
                if name:
                    out.append(str(name))
        return out
    if isinstance(value, str):
        return [g.strip() for g in value.split(",") if g.strip()]
    return []


class OidcStrategy(AuthStrategy):
    """
    OIDC. Configuration via env vars (OIDC_*). Provider is enabled only
    when discovery URL and client credentials are set.
    """

    name = AuthProvider.OIDC

    def __init__(self):
        self.discovery_url = os.getenv("OIDC_DISCOVERY_URL", "").strip()
        self.client_id = os.getenv("OIDC_CLIENT_ID", "").strip()
        self.client_secret = os.getenv("OIDC_CLIENT_SECRET", "").strip()
        self.redirect_uri = os.getenv("OIDC_REDIRECT_URI", "").strip()
        self.scopes = os.getenv("OIDC_SCOPES", "openid profile email").strip()
        self.provider_name = os.getenv("OIDC_PROVIDER_NAME", "OIDC").strip()

        # Claim names
        self.claim_username = os.getenv("OIDC_CLAIM_USERNAME", "preferred_username")
        self.claim_email = os.getenv("OIDC_CLAIM_EMAIL", "email")
        self.claim_groups = os.getenv("OIDC_CLAIM_GROUPS", "groups")
        self.claim_display = os.getenv("OIDC_CLAIM_DISPLAY", "name")
        self.claim_subject = os.getenv("OIDC_CLAIM_SUBJECT", "sub")

    @property
    def is_enabled(self) -> bool:
        return bool(self.discovery_url and self.client_id and self.client_secret and self.redirect_uri)

    def authenticate_credentials(self, username: str, password: str) -> ExternalIdentity:
        raise NotImplementedError("OIDC is redirect-based; use the /auth/oidc/* router.")

    # Called by the router after Authlib has validated the ID token.
    def parse_userinfo(self, userinfo: dict, id_token_claims: dict) -> ExternalIdentity:
        # Prefer ID-Token claims for security-relevant fields (signed, validated).
        sub = id_token_claims.get(self.claim_subject) or userinfo.get("sub")
        if not sub:
            raise AuthError(
                detail="OIDC response missing 'sub' claim",
                safe_message="Identity provider error",
            )

        email = id_token_claims.get(self.claim_email) or userinfo.get(self.claim_email)
        username = (
            id_token_claims.get(self.claim_username)
            or userinfo.get(self.claim_username)
            or (email.split("@")[0] if email else None)
            or sub
        )
        display = id_token_claims.get(self.claim_display) or userinfo.get(self.claim_display) or username

        # Groups can come from either ID-Token or userinfo. Merge.
        groups = list(
            set(
                _split_groups(id_token_claims.get(self.claim_groups))
                + _split_groups(userinfo.get(self.claim_groups))
            )
        )

        if not email:
            # Some IdPs require explicit 'email' scope to populate it. We can still
            # provision, but role-mapping by group remains the only path.
            logger.warning("OIDC userinfo has no email claim for sub=%s", sub[:8])
            email = f"{sub}@oidc.local"

        return ExternalIdentity(
            provider=AuthProvider.OIDC,
            external_id=str(sub),
            username=str(username),
            email=str(email),
            display_name=str(display) if display else None,
            groups=groups,
            extra={"iss": id_token_claims.get("iss")},
        )