"""
Authentication: local password login and OAuth2 authorization-code (browser).

FastAPI security integration
----------------------------
- ``OAuth2PasswordBearer`` + ``APIKeyCookie``: extract PyGitWeb's opaque session id from
  ``Authorization: Bearer`` or the login cookie. This is *not* an IdP JWT; OpenAPI still
  documents the password ``/token`` flow that mints these session ids.
- ``OAuth2PasswordRequestForm``: local ``POST /token`` only (resource-owner password grant).
- External IdP login (Google/GitHub) uses standard authorization-code redirects implemented
  in ``auth_oauth``; ``OAuth2AuthorizationCodeBearer`` is intentionally not used because
  clients never receive or send the provider's access token—only our session id.
"""

from __future__ import annotations

import base64
import binascii
import hmac
import secrets
import threading
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Annotated, Literal

from fastapi import APIRouter, Depends, Form, HTTPException, Query, status
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.security import APIKeyCookie, OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field, field_validator

from pygitweb.auth_config import (
    HttpKey,
    LocalUser,
    auth_config_path,
    is_auth_configured,
    is_local_auth_available,
    is_oauth_auth_available,
    write_auth_config,
)
from pygitweb.auth_oauth import OAUTH_STATE_COOKIE_NAME, oauth_router
from pygitweb.config import auth_config, settings
from pygitweb.gravatar import gravatar_url
from pygitweb.password_hash import hash_password, verify_password
from pygitweb.permissions import (
    Permission,
    PermissionPrincipal,
    has_ref_write_permission,
)
from pygitweb.projects import git_get_project_owner
from pygitweb.sessions import AuthMethod, create_session, get_session, revoke_session
from pygitweb.templates_env import render_page

ACCESS_TOKEN_COOKIE_NAME = "pygitweb_access_token"
LOGIN_COOKIE_MAX_AGE = 60 * 60 * 24 * 7

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
_access_token_cookie = APIKeyCookie(name=ACCESS_TOKEN_COOKIE_NAME, auto_error=False)

TAG_AUTH = "auth"
TAG_AUTH_LOCAL = "auth - local"
TAG_AUTH_OAUTH = "auth - OAuth"

auth_router = APIRouter()
auth_router.include_router(oauth_router)


class User(BaseModel):
    username: str
    email: str | None = None


SmartHttpAuthError = Literal["unauthorized", "unavailable"]
_INVALID_KEY_ACTOR = "INVALID_KEY"


@dataclass(frozen=True, slots=True)
class SmartHttpCredentials:
    required: bool
    principal: PermissionPrincipal | None = None
    readonly: bool = False
    actor: str = ""
    error: SmartHttpAuthError | None = None


@dataclass(frozen=True, slots=True)
class PresentedHttpKey:
    principal: PermissionPrincipal
    readonly: bool
    key_id: str
    expired: bool


_SMART_HTTP_CHALLENGE = {"WWW-Authenticate": 'Basic realm="pygitweb", Bearer'}


def principal_from_session(token: str | None) -> PermissionPrincipal | None:
    if not token:
        return None
    rec = get_session(token)
    if rec is None:
        return None
    return PermissionPrincipal(username=rec.username, email=rec.email)


def _anonymous_principal_for_permissions() -> PermissionPrincipal | None:
    for entry in auth_config.local_users:
        if getattr(entry, "assume_anonymous", False):
            return PermissionPrincipal(username=entry.user, email=entry.email)
    return None


def _principal_identity_for_permissions(token: str | None) -> str | None:
    principal = principal_from_session(token)
    if principal is not None:
        return principal.identity
    anonymous = _anonymous_principal_for_permissions()
    if anonymous is not None:
        return anonymous.identity
    return None


def has_permission(
    permission: Permission,
    project: str | None = None,
    *,
    token: str | None = None,
) -> bool:
    if not settings.AUTH:
        return True
    identity = _principal_identity_for_permissions(token)
    if identity is None:
        return False
    return auth_config.oauth_permissions.has_permission(identity, permission, project)


def _owner_matches_principal(project: str, principal: PermissionPrincipal) -> bool:
    if not settings.AUTH:
        return False
    owner = git_get_project_owner(project)
    if not owner or not owner.strip():
        return False
    owner_norm = owner.strip()
    identity = principal.identity
    if hmac.compare_digest(owner_norm, identity):
        return True
    return owner_norm.lower() == identity.lower()


def _owner_matches(project: str, token: str | None) -> bool:
    if not token:
        return False
    principal = principal_from_session(token)
    if principal is None:
        return False
    return _owner_matches_principal(project, principal)


def principal_can_read_project(project: str, principal: PermissionPrincipal) -> bool:
    if not settings.AUTH:
        return True
    if _owner_matches_principal(project, principal):
        return True
    return auth_config.oauth_permissions.has_permission(principal.identity, Permission.READ, project)


def principal_can_write_project(project: str, principal: PermissionPrincipal) -> bool:
    if not settings.AUTH:
        return True
    if _owner_matches_principal(project, principal):
        return True
    return auth_config.oauth_permissions.has_permission(principal.identity, Permission.WRITE, project)


def can_read_project(project: str, token: str | None) -> bool:
    if not settings.AUTH:
        return True
    if _owner_matches(project, token):
        return True
    return has_permission(Permission.READ, project, token=token)


def can_write_project(project: str, token: str | None) -> bool:
    if not settings.AUTH:
        return True
    if _owner_matches(project, token):
        return True
    return has_permission(Permission.WRITE, project, token=token)


def can_push_ref(project: str, principal: PermissionPrincipal, ref_name: str) -> bool:
    """Per-ref push check.

    Project owners may push any ref. Otherwise, if any ``pgw.write_ref`` rule
    applies to the project, only matching ``write_ref`` grants allow the push
    (whitelist). With no ``write_ref`` rules for the project, project write is
    enough. Names outside ``refs/`` are rejected.
    """
    if not settings.AUTH:
        return True
    if _owner_matches_principal(project, principal):
        return True
    return has_ref_write_permission(
        auth_config.oauth_permissions,
        principal.identity,
        project,
        ref_name,
    )


def ensure_read_project(project: str, token: str | None) -> None:
    if can_read_project(project, token):
        return
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project")


def ensure_write_project(project: str, token: str | None) -> None:
    if not settings.AUTH:
        return
    if not is_auth_configured(auth_config):
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Authentication enabled but auth.json is not configured",
        )
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if decode_access_token(token) is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if can_write_project(project, token):
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Insufficient permissions",
    )


def ensure_permission(
    permission: Permission,
    project: str | None = None,
    *,
    token: str | None,
) -> None:
    ensure_active_user_if_auth_enabled(token)
    if project is not None:
        ensure_read_project(project, token)
    if has_permission(permission, project, token=token):
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Insufficient permissions",
    )


def require_permission(
    permission: Permission,
    *,
    project_from_query: bool = False,
    project_from_form: bool = False,
) -> Callable[..., None]:
    """Return a FastAPI dependency that enforces ``permission`` when auth is enabled."""

    if project_from_query:

        def _dep_query_project(
            token: Annotated[str | None, Depends(access_token_from_request)],
            project: Annotated[str, Query()],
        ) -> None:
            ensure_permission(permission, project, token=token)

        return _dep_query_project

    if project_from_form:

        def _dep_form_project(
            token: Annotated[str | None, Depends(access_token_from_request)],
            project: Annotated[str, Form()],
        ) -> None:
            ensure_permission(permission, project, token=token)

        return _dep_form_project

    def _dep(
        token: Annotated[str | None, Depends(access_token_from_request)],
    ) -> None:
        ensure_permission(permission, None, token=token)

    return _dep


def get_local_user(username: str) -> User | None:
    if not settings.AUTH or not is_local_auth_available(auth_config):
        return None
    for entry in auth_config.local_users:
        if not entry.pass_hash and not entry.password:
            continue
        if hmac.compare_digest(entry.user, username):
            return User(username=entry.user, email=entry.email)
    return None


def authenticate_local_user(username: str, password: str) -> User | None:
    if auth_config.auth_mode not in ("local", "both"):
        return None
    for entry in auth_config.local_users:
        if not hmac.compare_digest(entry.user, username):
            continue
        if entry.pass_hash:
            if not verify_password(password, entry.pass_hash, config_salt=auth_config.salt):
                return None
            return User(username=entry.user, email=entry.email)
        if entry.password is not None and hmac.compare_digest(entry.password, password):
            return User(username=entry.user, email=entry.email)
        return None
    return None


def _const_eq(left: str, right: str) -> bool:
    if len(left) != len(right):
        return False
    return hmac.compare_digest(left, right)


def _http_key_username_matches(user: str, email: str | None, username: str) -> bool:
    if _const_eq(user, username):
        return True
    if not email:
        return False
    return _const_eq(email.casefold(), username.casefold())


def _key_owner_name(principal: PermissionPrincipal) -> str:
    email = principal.email.strip() if principal.email else ""
    return email or principal.username


def _http_key_actor(matched: PresentedHttpKey) -> str:
    owner = _key_owner_name(matched.principal)
    if matched.expired:
        return f"{owner} {matched.key_id} (EXPIRED)"
    return f"{owner} {matched.key_id}"


def match_presented_http_key(username: str, password: str) -> PresentedHttpKey | None:
    presented_user = username.strip()
    presented_secret = password.strip()
    if not presented_user or not presented_secret:
        return None
    now = datetime.now(UTC)
    expired_match: PresentedHttpKey | None = None
    for entry in auth_config.local_users:
        if not _http_key_username_matches(entry.user, entry.email, presented_user):
            continue
        principal = PermissionPrincipal(username=entry.user, email=entry.email)
        for key in entry.keys:
            if not verify_password(presented_secret, key.secret, config_salt=auth_config.salt):
                continue
            if key.is_expired(now):
                if expired_match is None:
                    expired_match = PresentedHttpKey(
                        principal=principal,
                        readonly=key.readonly,
                        key_id=key.id,
                        expired=True,
                    )
                continue
            return PresentedHttpKey(
                principal=principal,
                readonly=key.readonly,
                key_id=key.id,
                expired=False,
            )
    return expired_match


def authenticate_http_key(username: str, password: str) -> tuple[PermissionPrincipal, bool] | None:
    matched = match_presented_http_key(username, password)
    if matched is None or matched.expired:
        return None
    return matched.principal, matched.readonly


def _parse_basic_authorization(value: str) -> tuple[str, str] | None:
    scheme, _, param = value.strip().partition(" ")
    if scheme.lower() != "basic" or not param.strip():
        return None
    try:
        decoded = base64.b64decode(param.strip(), validate=True)
        text = decoded.decode("utf-8")
    except (binascii.Error, UnicodeDecodeError, ValueError):
        return None
    username, sep, password = text.partition(":")
    if not sep:
        return None
    return username, password


def unauthorized_smart_http() -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Not authenticated",
        headers=_SMART_HTTP_CHALLENGE,
    )


def raise_for_smart_http_credentials(creds: SmartHttpCredentials) -> None:
    if creds.error == "unavailable":
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Authentication enabled but auth.json is not configured",
        )
    if creds.error == "unauthorized":
        raise unauthorized_smart_http()


def _assumable_public_username(project: str, *, write: bool) -> str:
    if write:
        return ""
    anonymous = _anonymous_principal_for_permissions()
    if anonymous is None or not can_read_project(project, None):
        return ""
    return anonymous.username


def load_smart_http_credentials(
    authorization: str | None,
    cookie_token: str | None,
    *,
    project: str,
    write: bool,
) -> SmartHttpCredentials:
    if not settings.AUTH:
        return SmartHttpCredentials(required=False)
    if not is_auth_configured(auth_config) and not any(user.keys for user in auth_config.local_users):
        return SmartHttpCredentials(required=True, error="unavailable")
    if authorization:
        scheme, _, rest = authorization.strip().partition(" ")
        scheme_name = scheme.lower()
        if scheme_name == "basic":
            parsed = _parse_basic_authorization(authorization)
            if parsed is None:
                return SmartHttpCredentials(required=True, actor=_INVALID_KEY_ACTOR, error="unauthorized")
            matched = match_presented_http_key(parsed[0], parsed[1])
            if matched is None:
                return SmartHttpCredentials(required=True, actor=_INVALID_KEY_ACTOR, error="unauthorized")
            actor = _http_key_actor(matched)
            if matched.expired:
                return SmartHttpCredentials(required=True, actor=actor, error="unauthorized")
            return SmartHttpCredentials(
                required=True,
                principal=matched.principal,
                readonly=matched.readonly,
                actor=actor,
            )
        if scheme_name == "bearer":
            bearer_principal = principal_from_session(rest.strip())
            if bearer_principal is None:
                return SmartHttpCredentials(required=True, error="unauthorized")
            return SmartHttpCredentials(
                required=True,
                principal=bearer_principal,
                readonly=False,
                actor=_key_owner_name(bearer_principal),
            )
        return SmartHttpCredentials(required=True, error="unauthorized")
    if cookie_token and cookie_token.strip():
        cookie_principal = principal_from_session(cookie_token)
        if cookie_principal is None:
            return SmartHttpCredentials(required=True, error="unauthorized")
        return SmartHttpCredentials(
            required=True,
            principal=cookie_principal,
            readonly=False,
            actor=_key_owner_name(cookie_principal),
        )
    return SmartHttpCredentials(required=True, actor=_assumable_public_username(project, write=write))


def decode_access_token(token: str) -> User | None:
    if not settings.AUTH:
        return None
    rec = get_session(token)
    if rec is None:
        return None
    return User(username=rec.username)


def safe_next_url(next_raw: str | None) -> str:
    n = (next_raw or "/").strip()
    if not n.startswith("/") or n.startswith("//"):
        return "/"
    return n


def access_token_from_request(
    bearer: Annotated[str | None, Depends(oauth2_scheme)],
    cookie_token: Annotated[str | None, Depends(_access_token_cookie)],
) -> str | None:
    b = bearer.strip() if bearer else None
    c = cookie_token.strip() if cookie_token else None
    return b or c


def ensure_active_user_if_auth_enabled(token: str | None) -> None:
    if not settings.AUTH:
        return
    if not is_auth_configured(auth_config):
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Authentication enabled but auth.json is not configured",
        )
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if decode_access_token(token) is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )


def require_active_user_if_auth_enabled(
    token: Annotated[str | None, Depends(access_token_from_request)],
) -> None:
    ensure_active_user_if_auth_enabled(token)


@auth_router.get(
    "/login",
    response_class=HTMLResponse,
    tags=[TAG_AUTH],
    summary="Sign-in page",
    description="HTML sign-in page. Shows local and/or OAuth options depending on auth.json.",
)
async def login_page(
    next: Annotated[str, Query()] = "/",
) -> HTMLResponse:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    next_safe = safe_next_url(next)
    return render_page(
        "login.html",
        title=f"{settings.SITE_NAME} - Sign in",
        next=next_safe,
        error=None,
        local_available=is_local_auth_available(auth_config),
        oauth_available=is_oauth_auth_available(auth_config),
        oauth_provider=str(auth_config.oauth_provider).strip() or "OAuth",
    )


@auth_router.post(
    "/login",
    response_model=None,
    tags=[TAG_AUTH_LOCAL],
    summary="Local sign-in (form)",
    description="Browser form login. Sets the pygitweb_access_token session cookie on success.",
    responses={303: {"description": "Redirect to next URL with session cookie set"}},
)
async def login_form_submit(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
    next: Annotated[str, Form()] = "/",
) -> HTMLResponse | RedirectResponse:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    if not is_local_auth_available(auth_config):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Local sign-in is not enabled")
    next_safe = safe_next_url(next)
    user = authenticate_local_user(username.strip(), password)
    if user is None:
        return render_page(
            "login.html",
            title=f"{settings.SITE_NAME} - Sign in",
            next=next_safe,
            error="Incorrect username or password.",
            local_available=True,
            oauth_available=is_oauth_auth_available(auth_config),
            oauth_provider=str(auth_config.oauth_provider).strip() or "OAuth",
        )
    sid = create_session(user.username, email=user.email, auth_method="local")
    resp = RedirectResponse(url=next_safe, status_code=303)
    resp.set_cookie(
        key=ACCESS_TOKEN_COOKIE_NAME,
        value=sid,
        httponly=True,
        samesite="lax",
        max_age=LOGIN_COOKIE_MAX_AGE,
        path="/",
    )
    return resp


@auth_router.get(
    "/logout",
    tags=[TAG_AUTH],
    summary="Sign out",
    description="Revokes the current session (cookie or Bearer) and clears auth cookies.",
    responses={303: {"description": "Redirect to next URL"}},
)
async def logout_page(
    token: Annotated[str | None, Depends(access_token_from_request)],
    next: Annotated[str, Query()] = "/",
) -> RedirectResponse:
    revoke_session(token)
    resp = RedirectResponse(url=safe_next_url(next), status_code=303)
    resp.delete_cookie(key=ACCESS_TOKEN_COOKIE_NAME, path="/")
    resp.delete_cookie(key=OAUTH_STATE_COOKIE_NAME, path="/")
    return resp


@auth_router.post(
    "/token",
    tags=[TAG_AUTH_LOCAL],
    summary="Obtain session token (local)",
    description=(
        "OAuth2 password flow for API clients. Returns an opaque session id as access_token; "
        "use Authorization: Bearer on protected routes. Not an identity-provider JWT."
    ),
)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]) -> dict[str, str]:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    if not is_local_auth_available(auth_config):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Local sign-in is not enabled")
    user = authenticate_local_user(form_data.username, form_data.password)
    if user is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    sid = create_session(user.username, email=user.email, auth_method="local")
    return {"access_token": sid, "token_type": "bearer"}


@auth_router.get(
    "/auth/status",
    tags=[TAG_AUTH],
    summary="Auth status (navbar)",
    description="Lightweight JSON for the navbar script. Does not return 401 when anonymous.",
)
async def auth_status(
    token: Annotated[str | None, Depends(access_token_from_request)],
) -> dict[str, bool | str | None]:
    if not settings.AUTH:
        return {"auth_enabled": False, "username": None, "gravatar_url": None}
    if not token:
        return {"auth_enabled": True, "username": None, "gravatar_url": None}
    rec = get_session(token)
    if rec is None:
        return {"auth_enabled": True, "username": None, "gravatar_url": None}
    avatar = gravatar_url(rec.email) if rec.email else None
    return {"auth_enabled": True, "username": rec.username, "gravatar_url": avatar}


_API_KEY_LOCK = threading.Lock()


class IssueApiKeyRequest(BaseModel):
    id: str = Field(min_length=1)
    readonly: bool = False

    @field_validator("id", mode="before")
    @classmethod
    def _strip_id(cls, value: object) -> object:
        if not isinstance(value, str):
            return value
        stripped = value.strip()
        if not stripped:
            raise ValueError("must be a non-empty string")
        return stripped


class IssuedApiKey(BaseModel):
    id: str
    secret: str
    readonly: bool
    time_issued: datetime
    time_expiry: datetime | None = None


def _local_user_by_email(email: str | None) -> LocalUser | None:
    if not email:
        return None
    folded = email.strip().casefold()
    if not folded:
        return None
    for entry in auth_config.local_users:
        if entry.email and _const_eq(entry.email.casefold(), folded):
            return entry
    return None


def _local_user_for_principal(username: str, email: str | None) -> LocalUser | None:
    for entry in auth_config.local_users:
        if _const_eq(entry.user, username):
            return entry
    return _local_user_by_email(email)


def _key_owner(username: str, email: str | None, auth_method: AuthMethod) -> LocalUser | None:
    if auth_method == "oauth":
        return _local_user_by_email(email)
    return _local_user_for_principal(username, email)


def _new_passwordless_user(email: str | None, auth_method: AuthMethod) -> LocalUser:
    if auth_method != "oauth":
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
    stored = email.strip() if email else ""
    if len(stored) < 2:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="An email is required to store API keys")
    for entry in auth_config.local_users:
        if _const_eq(entry.user.casefold(), stored.casefold()):
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="A local account already uses this name",
            )
    return LocalUser(user=stored, email=stored)


def _drop_local_user(user: LocalUser) -> None:
    for index, entry in enumerate(auth_config.local_users):
        if entry is user:
            del auth_config.local_users[index]
            return


def _key_expiry(issued: datetime) -> datetime | None:
    days = settings.API_KEY_EXPIRY_DAYS
    if days <= 0:
        return None
    return issued + timedelta(days=days)


def _persist_auth_config() -> None:
    write_auth_config(auth_config_path(settings), auth_config)


def _public_key_rows(user: LocalUser) -> list[dict[str, str | bool | None]]:
    rows: list[dict[str, str | bool | None]] = []
    for key in user.keys:
        rows.append({
            "id": key.id,
            "readonly": key.readonly,
            "time_issued": key.time_issued.isoformat(),
            "time_expiry": key.time_expiry.isoformat() if key.time_expiry else None,
        })
    return rows


def issue_api_key(
    username: str,
    email: str | None,
    *,
    key_id: str,
    readonly: bool,
    auth_method: AuthMethod = "local",
) -> IssuedApiKey:
    with _API_KEY_LOCK:
        user = _key_owner(username, email, auth_method)
        created = False
        if user is None:
            user = _new_passwordless_user(email, auth_method)
            created = True
        if len(user.keys) >= settings.API_KEY_MAX_COUNT:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=f"API key limit reached ({settings.API_KEY_MAX_COUNT})",
            )
        if any(_const_eq(existing.id, key_id) for existing in user.keys):
            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="An API key with this id already exists")
        issued = datetime.now(UTC)
        plaintext = secrets.token_urlsafe(32)
        stored = HttpKey(
            id=key_id,
            secret=hash_password(plaintext, config_salt=auth_config.salt),
            readonly=readonly,
            time_issued=issued,
            time_expiry=_key_expiry(issued),
        )
        if created:
            auth_config.local_users.append(user)
        previous = list(user.keys)
        user.keys = [*previous, stored]
        try:
            _persist_auth_config()
        except Exception:
            if created:
                _drop_local_user(user)
            else:
                user.keys = previous
            raise
        return IssuedApiKey(
            id=stored.id,
            secret=plaintext,
            readonly=stored.readonly,
            time_issued=stored.time_issued,
            time_expiry=stored.time_expiry,
        )


def revoke_api_key(
    username: str,
    email: str | None,
    key_id: str,
    *,
    auth_method: AuthMethod = "local",
) -> None:
    with _API_KEY_LOCK:
        user = _key_owner(username, email, auth_method)
        if user is None:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
        remaining = [key for key in user.keys if not _const_eq(key.id, key_id)]
        if len(remaining) == len(user.keys):
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such API key")
        previous = list(user.keys)
        user.keys = remaining
        try:
            _persist_auth_config()
        except Exception:
            user.keys = previous
            raise


def _account_keys(
    username: str,
    email: str | None,
    auth_method: AuthMethod,
) -> tuple[list[dict[str, str | bool | None]] | None, str | None]:
    if auth_method == "oauth":
        user = _local_user_by_email(email)
        if user is not None:
            return _public_key_rows(user), user.user
        stored = email.strip() if email else ""
        if len(stored) < 2:
            return None, None
        return [], stored
    user = _local_user_for_principal(username, email)
    if user is None:
        return None, None
    return _public_key_rows(user), user.user


def _session_identity(token: str | None) -> tuple[str, str | None, AuthMethod]:
    ensure_active_user_if_auth_enabled(token)
    rec = get_session(token or "")
    if rec is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return rec.username, rec.email, rec.auth_method


@auth_router.get(
    "/user",
    response_class=HTMLResponse,
    response_model=None,
    tags=[TAG_AUTH],
    summary="Account page",
    description="HTML account page for the signed-in user.",
)
async def user_account_page(
    token: Annotated[str | None, Depends(access_token_from_request)],
) -> HTMLResponse | RedirectResponse:
    if not settings.AUTH:
        return render_page(
            "message.html",
            title=f"{settings.SITE_NAME} - Account",
            paragraphs=["Authentication is not enabled on this server."],
            muted=True,
        )
    if not is_auth_configured(auth_config):
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Authentication is misconfigured",
        )
    rec = get_session(token or "")
    if rec is None:
        return RedirectResponse(url="/login?next=/user", status_code=303)
    api_keys, api_key_user = _account_keys(rec.username, rec.email, rec.auth_method)
    if api_key_user is not None and _const_eq(api_key_user, rec.username):
        api_key_user = None
    return render_page(
        "user_profile.html",
        title=f"{settings.SITE_NAME} - Account",
        username=rec.username,
        api_keys=api_keys,
        api_key_user=api_key_user,
        api_key_max=settings.API_KEY_MAX_COUNT,
        api_key_expiry_days=settings.API_KEY_EXPIRY_DAYS,
    )


@auth_router.post(
    "/user/keys",
    response_model=IssuedApiKey,
    tags=[TAG_AUTH],
    summary="Issue an API key",
    description=(
        "Creates a Smart HTTP API key for the signed-in user. "
        "An OAuth session is stored as a passwordless local account, matched by email, "
        "the first time it issues a key. The plaintext secret is returned once. "
        "Only its scrypt hash is stored."
    ),
)
async def issue_user_api_key(
    body: IssueApiKeyRequest,
    token: Annotated[str | None, Depends(access_token_from_request)],
) -> IssuedApiKey:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    username, email, auth_method = _session_identity(token)
    return issue_api_key(username, email, key_id=body.id, readonly=body.readonly, auth_method=auth_method)


@auth_router.delete(
    "/user/keys",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=[TAG_AUTH],
    summary="Revoke an API key",
    description="Removes one Smart HTTP API key from the signed-in user.",
)
async def revoke_user_api_key(
    token: Annotated[str | None, Depends(access_token_from_request)],
    id: Annotated[str, Query(min_length=1)],
) -> Response:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    username, email, auth_method = _session_identity(token)
    revoke_api_key(username, email, id.strip(), auth_method=auth_method)
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@auth_router.get(
    "/users/me",
    response_model=User,
    tags=[TAG_AUTH],
    summary="Current user",
    description="Returns the username for the current session (cookie or Bearer).",
)
async def read_users_me(
    token: Annotated[str | None, Depends(access_token_from_request)],
) -> User:
    if not settings.AUTH:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
    ensure_active_user_if_auth_enabled(token)
    u = decode_access_token(token or "")
    if u is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return u