border-red-post-nalati/backend/app/services/auth.py

"""Database-backed sessions; no bearer secrets are stored in the database."""

import hashlib
import hmac
import re
import secrets
from dataclasses import dataclass
from datetime import timedelta
from functools import lru_cache

from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError
from fastapi import Depends, HTTPException, Request, Response
from sqlalchemy import case, delete, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session

from app.core.config import settings
from app.db.session import get_db
from app.models.auth import AuthRateLimit, AuthSession, User
from app.models.catalog import utc_now
from app.schemas.auth import SessionRead, UserRead

COOKIE_NAME = "nalati_session"
UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
PASSWORD_HASHER = PasswordHasher(time_cost=2, memory_cost=19456, parallelism=1)


@lru_cache(maxsize=1)
def dummy_hash():
    return PASSWORD_HASHER.hash(secrets.token_urlsafe(32))


def check_password(encoded: str, password: str) -> bool:
    try:
        return PASSWORD_HASHER.verify(encoded, password)
    except (VerificationError, InvalidHashError):
        return False


def token_hash(raw: str) -> str:
    return hashlib.sha256(raw.encode()).hexdigest()


def csrf_token(raw: str) -> str:
    # Domain-separated derivative; knowing the CSRF token does not reveal the cookie.
    return hmac.new(raw.encode(), b"nalati-csrf-v1", hashlib.sha256).hexdigest()


def require_origin(request: Request):
    if request.headers.get("origin") not in settings.auth_origin_list:
        raise HTTPException(403, "请求来源不受信任")


def limit_auth_requests(db: Session, request: Request, username: str | None = None):
    """Atomic, shared across workers and restarts; never trust forwarded client headers here."""
    now = utc_now()
    window = timedelta(minutes=5)
    buckets = [("ip:" + (request.client.host if request.client else "unknown"), 30)]
    if username is not None:
        buckets.append(("username:" + username, 10))
    blocked = False
    for identity, maximum in buckets:
        statement = insert(AuthRateLimit).values(
            key=token_hash(identity), window_start=now, attempts=1
        )
        expired = AuthRateLimit.window_start <= now - window
        statement = statement.on_conflict_do_update(
            index_elements=[AuthRateLimit.key],
            set_={
                "window_start": case((expired, now), else_=AuthRateLimit.window_start),
                "attempts": case((expired, 1), else_=AuthRateLimit.attempts + 1),
            },
        ).returning(AuthRateLimit.attempts)
        blocked |= db.scalar(statement) > maximum
    db.execute(delete(AuthRateLimit).where(AuthRateLimit.window_start < now - timedelta(days=1)))
    # Persist even failed authentication; no password or raw address is logged.
    db.commit()
    if blocked:
        raise HTTPException(429, "认证请求过于频繁，请稍后重试", headers={"Retry-After": "300"})


@dataclass
class CurrentSession:
    user: User
    session: AuthSession
    raw_token: str

    def response(self):
        return SessionRead(
            **UserRead.model_validate(self.user).model_dump(),
            csrf_token=csrf_token(self.raw_token),
        )


def require_session(request: Request, db: Session = Depends(get_db)) -> CurrentSession:
    raw = request.cookies.get(COOKIE_NAME, "")
    if not re.fullmatch(r"[A-Za-z0-9_-]{43}", raw):
        raise HTTPException(401, "请先登录")
    session = db.get(AuthSession, token_hash(raw))
    if session is None:
        raise HTTPException(401, "登录已失效，请重新登录")
    # Lock the user before checking the current session again, serializing password changes/login.
    user = db.scalar(select(User).where(User.id == session.user_id).with_for_update())
    session = db.scalar(
        select(AuthSession)
        .where(AuthSession.token_hash == token_hash(raw))
        .execution_options(populate_existing=True)
    )
    if (
        user is None
        or session is None
        or session.expires_at <= utc_now()
        or session.credential_version != user.credential_version
    ):
        raise HTTPException(401, "登录已失效，请重新登录")
    if request.method in UNSAFE_METHODS:
        require_origin(request)
        submitted = request.headers.get("x-csrf-token", "")
        if not hmac.compare_digest(submitted.encode(), csrf_token(raw).encode()):
            raise HTTPException(403, "CSRF 校验失败，请刷新登录状态后重试")
    return CurrentSession(user, session, raw)


def require_admin(current: CurrentSession = Depends(require_session)):
    if current.user.role != "admin":
        raise HTTPException(403, "需要管理员权限")
    return current


def issue_session(db: Session, user: User, request: Request, response: Response):
    now = utc_now()
    old = request.cookies.get(COOKIE_NAME, "")
    db.execute(delete(AuthSession).where(AuthSession.expires_at <= now))
    if old:
        db.execute(delete(AuthSession).where(AuthSession.token_hash == token_hash(old)))
    raw = secrets.token_urlsafe(32)
    session = AuthSession(
        token_hash=token_hash(raw),
        user_id=user.id,
        credential_version=user.credential_version,
        expires_at=now + timedelta(seconds=settings.auth_session_seconds),
    )
    db.add(session)
    db.commit()
    response.set_cookie(
        COOKIE_NAME,
        raw,
        max_age=settings.auth_session_seconds,
        httponly=True,
        secure=settings.auth_cookie_secure,
        samesite="lax",
        path="/",
    )
    return CurrentSession(user, session, raw).response()


def clear_cookie(response: Response):
    response.delete_cookie(
        COOKIE_NAME, path="/", httponly=True, secure=settings.auth_cookie_secure, samesite="lax"
    )