"""Verified Huggingface OAuth identities and same-origin mutation checks.

The identity comes only from Gradio/Authlib's signed OAuth session.  A public
username, request JSON field, or arbitrary bearer/header is never an identity.
"""
from __future__ import annotations

from dataclasses import dataclass
import hashlib
import hmac
import math
import os
import secrets
import time
from urllib.parse import urlsplit

from fastapi import HTTPException, Request


_CSRF_SECRET = (os.environ.get('OAUTH_CLIENT_SECRET') or secrets.token_hex(32)).encode()


@dataclass(frozen=True)
class Identity:
    owner: str
    username: str
    session_key: str


def identity_from_oauth(oauth) -> Identity | None:
    """Validate a verified server-side OAuth result, independently of cookies."""
    if not isinstance(oauth, dict):
        return None
    profile = oauth.get('userinfo')
    if not isinstance(profile, dict):
        return None
    owner, token = profile.get('sub'), oauth.get('access_token')
    if not isinstance(owner, str) or not owner or len(owner) > 256:
        return None
    if any(ord(c) < 32 for c in owner) or not isinstance(token, str) or not token:
        return None
    try:
        expires = float(oauth['expires_at'])
    except (KeyError, ValueError, TypeError, OverflowError):
        return None
    if not math.isfinite(expires) or expires <= time.time():
        return None
    username = profile.get('preferred_username')
    if not isinstance(username, str) or not username:
        username = profile.get('name') if isinstance(profile.get('name'), str) else 'Account'
    return Identity(owner, username[:256], token + '\0' + str(expires))


def optional_identity(request: Request) -> Identity | None:
    """Prefer the verified partitioned application session when present."""
    if 'supra_history_app_identity' in request.scope:
        # An invalid/expired application cookie must not silently fall back to
        # a previous account in the older, unpartitioned Gradio session.
        return request.scope['supra_history_app_identity']
    session = request.scope.get('session')
    return identity_from_oauth(session.get('oauth_info')) if isinstance(session, dict) else None


def require_identity(request: Request) -> Identity:
    identity = optional_identity(request)
    if identity is None:
        raise HTTPException(401, 'Sign in with Huggingface to view your history.')
    return identity


def csrf_token(identity: Identity, secret: bytes | None = None) -> str:
    """Stable in an OAuth session, including multiple tabs; never reveals tokens."""
    value = (identity.owner + '\0' + identity.session_key).encode('utf-8')
    return hmac.new(secret or _CSRF_SECRET, value, hashlib.sha256).hexdigest()


def account_key(identity: Identity, secret: bytes | None = None) -> str:
    """Opaque account identity stable across logins, distinct from CSRF values."""
    value = ('supra-history-account\0' + identity.owner).encode('utf-8')
    return hmac.new(secret or _CSRF_SECRET, value, hashlib.sha256).hexdigest()


def _origin(value: str) -> str | None:
    try:
        parts = urlsplit(value)
        if parts.scheme not in ('http', 'https') or not parts.netloc:
            return None
        if parts.username is not None or parts.password is not None:
            return None
        if parts.path not in ('', '/') or parts.query or parts.fragment:
            return None
        host = parts.hostname
        port = parts.port
        if not host:
            return None
        host = '[' + host.lower() + ']' if ':' in host else host.lower()
        default = (parts.scheme == 'https' and port == 443) or (parts.scheme == 'http' and port == 80)
        return parts.scheme.lower() + '://' + host + (':' + str(port) if port and not default else '')
    except (ValueError, TypeError):
        return None


def trusted_origin_set(origins=()) -> frozenset[str]:
    values = list(origins)
    for host in os.environ.get('SPACE_HOST', '').split(','):
        if host.strip():
            values.append('https://' + host.strip().rstrip('/'))
    return frozenset(value for value in (_origin(str(v)) for v in values) if value)


def require_mutation(request: Request, origins=(), secret: bytes | None = None) -> Identity:
    identity = require_identity(request)
    allowed = trusted_origin_set(origins)
    # Local tests/previews can use their own origin. Production derives the fixed
    # origin from SPACE_HOST instead of trusting arbitrary Host headers.
    if not allowed and not os.environ.get('SPACE_ID'):
        current = _origin(str(request.base_url))
        if current:
            allowed = frozenset((current,))
    origin = _origin(request.headers.get('origin', ''))
    fetch_site = request.headers.get('sec-fetch-site', '').lower()
    if origin not in allowed or fetch_site == 'cross-site':
        raise HTTPException(403, 'This action must be submitted from this website.')
    supplied = request.headers.get('x-csrf-token', '')
    if len(supplied) != 64 or not hmac.compare_digest(supplied, csrf_token(identity, secret)):
        raise HTTPException(403, 'Your session changed. Refresh the page and try again.')
    return identity