"""
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_branch_read_permission
from pygitweb.projects import git_get_project_owner
from pygitweb.sessions import 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
	tail: 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_read_branch(project: str, ref_name: str, token: str | None) -> bool:
	if not settings.AUTH:
		return True
	if _owner_matches(project, token):
		return True
	identity = _principal_identity_for_permissions(token)
	if identity is None:
		return False
	return has_branch_read_permission(auth_config.oauth_permissions, 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.tail} (EXPIRED)"
	return f"{owner} {matched.tail}"


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
	tail = presented_secret[-4:]
	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,
						tail=tail,
						expired=True,
					)
				continue
			return PresentedHttpKey(principal=principal, readonly=key.readonly, tail=tail, 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_for_principal(username: str, email: str | None) -> LocalUser | None:
	for entry in auth_config.local_users:
		if _const_eq(entry.user, username):
			return entry
	if not email:
		return None
	folded = email.casefold()
	for entry in auth_config.local_users:
		if entry.email and _const_eq(entry.email.casefold(), folded):
			return entry
	return None


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) -> IssuedApiKey:
	with _API_KEY_LOCK:
		user = _local_user_for_principal(username, email)
		if user is None:
			raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
		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),
		)
		previous = list(user.keys)
		user.keys = [*previous, stored]
		try:
			_persist_auth_config()
		except Exception:
			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) -> None:
	with _API_KEY_LOCK:
		user = _local_user_for_principal(username, email)
		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 _session_identity(token: str | None) -> tuple[str, str | None]:
	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


@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",
		)
	u = decode_access_token(token or "") if token else None
	if u is None:
		return RedirectResponse(url="/login?next=/user", status_code=303)
	rec = get_session(token or "")
	email = rec.email if rec is not None else None
	local_user = _local_user_for_principal(u.username, email)
	return render_page(
		"user_profile.html",
		title=f"{settings.SITE_NAME} - Account",
		username=u.username,
		api_keys=_public_key_rows(local_user) if local_user is not None else None,
		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 local user. "
		"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 = _session_identity(token)
	return issue_api_key(username, email, key_id=body.id, readonly=body.readonly)


@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 local 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 = _session_identity(token)
	revoke_api_key(username, email, id.strip())
	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