shaguabijia-app-server/app/admin/deps.py

"""Admin API 共享依赖:DB session、当前 admin、角色守卫、客户端 IP。

仿 app/api/deps.py 的 get_current_user,但验的是独立的 admin JWT、查的是 admin_user 表。
"""
from __future__ import annotations

from typing import Annotated

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

from app.admin.repositories import admin_user as admin_repo
from app.admin.security import AdminTokenError, decode_admin_token
from app.db.session import get_db
from app.models.admin import AdminUser

_bearer = HTTPBearer(auto_error=False, scheme_name="AdminBearer")


def get_current_admin(
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
    db: Annotated[Session, Depends(get_db)],
) -> AdminUser:
    """从 Authorization: Bearer <admin_token> 解出当前管理员。失败统一 401。"""
    if credentials is None or credentials.scheme.lower() != "bearer":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="missing bearer token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    try:
        payload = decode_admin_token(credentials.credentials)
    except AdminTokenError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=str(e),
            headers={"WWW-Authenticate": "Bearer"},
        ) from e

    admin = admin_repo.get_by_id(db, int(payload["sub"]))
    if admin is None or admin.status != "active":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="admin not found or disabled",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return admin


CurrentAdmin = Annotated[AdminUser, Depends(get_current_admin)]
AdminDb = Annotated[Session, Depends(get_db)]


def require_role(*roles: str):
    """角色守卫依赖工厂。super_admin 恒通过(全权)。

    用法:在路由签名加 `_: Annotated[AdminUser, Depends(require_role("finance"))]`,
    或 `dependencies=[Depends(require_role("finance"))]`(不需要拿 admin 时)。
    """
    allowed = set(roles)

    def _checker(admin: CurrentAdmin) -> AdminUser:
        if admin.role != "super_admin" and admin.role not in allowed:
            need = sorted(allowed | {"super_admin"})
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"role '{admin.role}' not allowed (need one of {need})",
            )
        return admin

    return _checker


def get_client_ip(request: Request) -> str:
    """取客户端 IP(审计日志用)。生产经 nginx 反代,优先 X-Forwarded-For 第一段;否则直连 IP。

    ⚠️ XFF 可被客户端伪造。nginx 必须用 `proxy_set_header X-Forwarded-For $remote_addr`
    覆盖客户端传入值(见 M4 部署),否则审计里的 IP 可被伪造。审计 IP 仅作记录、不参与鉴权。
    """
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else ""