PayrollManagementSystem/backend/app/api/deps.py

"""FastAPI 公共依赖注入。"""

from typing import Annotated

from fastapi import Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session

from app.core.constants import ROLE_ADMIN
from app.core.db import get_db
from app.core.exceptions import AuthError, ForbiddenError
from app.core.security import decode_token
from app.models import User
from app.repositories import user_repository
from app.services import token_blacklist
from app.services.audit_service import RequestMeta

# auto_error=False：令牌缺失时由我们自己抛 AuthError，
# 才能返回统一响应信封，而不是 FastAPI 默认的 {"detail": "Not authenticated"}
_bearer_scheme = HTTPBearer(auto_error=False, description="Bearer 访问令牌")

BearerCredentials = Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)]
DbSession = Annotated[Session, Depends(get_db)]


def client_ip(request: Request) -> str | None:
    """取客户端 IP。

    当前取直连地址。生产环境经 Nginx 反代后直连地址会变成代理地址，
    届时需改为读取 `X-Forwarded-For` 的**第一个**值 —— 且必须确保该头
    由网关覆写，否则客户端可以伪造。此改动留到部署阶段一并处理。
    """
    return request.client.host if request.client else None


def get_request_meta(request: Request) -> RequestMeta:
    """提取 HTTP 请求元信息，供操作日志审计使用。"""
    return RequestMeta(
        ip=client_ip(request),
        user_agent=request.headers.get("user-agent"),
        method=request.method,
        path=request.url.path,
    )


RequestMetaDep = Annotated[RequestMeta, Depends(get_request_meta)]


def get_raw_token(credentials: BearerCredentials) -> str:
    """取原始令牌字符串。登出时需要它来解析 jti。"""
    if credentials is None or not credentials.credentials:
        raise AuthError()
    return credentials.credentials


def get_current_user(credentials: BearerCredentials, session: DbSession) -> User:
    """从 Bearer 令牌解析当前登录用户。

    **三重校验，缺一不可**：

    1. 签名与有效期（`decode_token`）
    2. 是否已被登出拉黑（Redis `ys:token:blacklist:{jti}`）
    3. **令牌版本与库中是否一致** —— 禁用 / 改密 / 强制下线会把
       `users.token_version` 加一，使此前签发的所有令牌立即失效。
       这一步**不依赖 Redis**，因此 Redis 故障或数据被清空时，
       离职员工依然无法继续访问

    Redis 不可用时第 2 步会抛异常而非放行 —— **有意的 fail-closed**：
    对能查看全员薪酬的系统，宁可拒绝请求也不能放行状态不明的令牌。

    :raises AuthError: 任一项校验未通过（401）
    """
    if credentials is None or not credentials.credentials:
        raise AuthError()

    payload = decode_token(credentials.credentials)

    jti = payload.get("jti")
    if jti and token_blacklist.is_revoked(jti):
        raise AuthError("登录状态已失效，请重新登录")

    subject = payload.get("sub")
    user = user_repository.get_by_id(session, int(subject)) if _is_digits(subject) else None
    if user is None or user.status != 1:
        raise AuthError()

    if payload.get("ver") != user.token_version:
        raise AuthError("登录状态已失效，请重新登录")

    return user


def _is_digits(value: object) -> bool:
    """判断令牌中的 sub 是否为合法的主键字符串。

    令牌内容不可信，直接 int() 会因非法输入抛 ValueError 而不是 401。
    """
    return isinstance(value, str) and value.isdigit()


def require_admin(current_user: Annotated[User, Depends(get_current_user)]) -> User:
    """要求当前用户具备管理员角色。

    需求 2.5.1 只定义了「系统管理员（全部权限）」与「普通用户（业务操作）」两个
    角色，故权限判定用**角色粗粒度**即可，不必引入权限码表 —— `permissions`
    与 `role_permissions` 表已建好，等业务方给出权限清单后再启用。

    :raises ForbiddenError: 已登录但不具备管理员角色（403）
    """
    if ROLE_ADMIN not in user_repository.get_role_codes(current_user):
        raise ForbiddenError("需要管理员权限")
    return current_user


AdminUser = Annotated[User, Depends(require_admin)]