track-LICA/backend/app/services/auth_service.py

"""认证服务 — 对接 MOM 系统 sys_user 表 + Track 自有 JWT（双 Token 架构）"""
from fastapi import HTTPException, status, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from werkzeug.security import check_password_hash

from app.core.config import settings
from app.core.security import (
    create_access_token,
    create_refresh_token,
    decode_token,
    ALGORITHM,
    TOKEN_TYPE_ACCESS,
    TOKEN_TYPE_REFRESH,
)
from app.core.mom_database import MomSessionLocal
from app.core.logging import user_var
from app.core.roles import SUPER_ADMIN
from app.schemas.user import LoginResponse, UserResponse

security = HTTPBearer()


def login(username: str, password: str) -> LoginResponse:
    """登录 — 签发双 Token（Access + Refresh）"""
    db = MomSessionLocal()
    try:
        # 1. LIKE '%/username' 定位 MOM sys_user 账号，再按部门收敛 ——
        #    IRIS 与 LICA 共用同一个 MOM 库，不加部门条件的话，另一个部门的
        #    普通账号也能登进来。
        #
        #    ⚠️ 唯一的例外是 SUPER_ADMIN：超管**跨部门放行**，供运维/管理员在
        #    两个实例之间切换。其余角色（INBOUND / SUPERVISOR / WAREHOUSE_MGR /
        #    SALES）必须严格属于本部门（settings.ORG_DEPARTMENT）。
        #
        #    不匹配时统一报「用户名或密码错误」，不区分「账号不存在」与「存在但
        #    不属于本部门」，避免给账号探测者提供线索。
        from sqlalchemy import text
        result = db.execute(
            text(
                "SELECT id, username, department, role, password_hash "
                "FROM sys_user "
                "WHERE username LIKE :pattern "
                "  AND (department = :dept OR role = :super_admin)"
            ),
            {
                "pattern": f"%/{username}",
                "dept": settings.ORG_DEPARTMENT,
                "super_admin": SUPER_ADMIN,
            },
        )
        row = result.fetchone()

        if not row:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="用户名或密码错误",
            )

        user_id, full_username, department, role, password_hash = row

        # 2. Werkzeug scrypt 密码验证
        if not check_password_hash(password_hash, password):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="用户名或密码错误",
            )

        # 3. 解析 display_name（"张三/zhangsan01" → "张三"）
        display_name = full_username.split("/")[0] if "/" in full_username else full_username

        token_data = {
            "sub": str(user_id),
            "role": role or "operator",
            "username": username,
            "display_name": display_name,
        }

        return LoginResponse(
            access_token=create_access_token(data=token_data),
            refresh_token=create_refresh_token(data=token_data),
            user=UserResponse(
                id=str(user_id),
                username=username,
                display_name=display_name,
                role=role or "operator",
            ),
        )

    finally:
        db.close()


def refresh_access_token(refresh_token: str) -> dict:
    """
    使用 Refresh Token 换取新的 Access Token。
    校验：
    1. Token 签名是否有效
    2. Token type 是否为 "refresh"
    3. Token 是否未过期
    """
    try:
        payload = decode_token(refresh_token)
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh Token 无效或已过期，请重新登录",
        )

    # 校验 token 类型
    if payload.get("type") != TOKEN_TYPE_REFRESH:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的 Token 类型，仅接受 Refresh Token",
        )

    # 提取用户信息，签发新的 Access Token
    access_token = create_access_token(
        data={
            "sub": payload.get("sub"),
            "role": payload.get("role", "operator"),
            "username": payload.get("username", ""),
            "display_name": payload.get("display_name", ""),
        }
    )

    return {"access_token": access_token, "token_type": "bearer"}


async def get_current_user(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    """从 Bearer Token 解析当前用户（仅接受 Access Token）"""
    token = credentials.credentials
    try:
        payload = decode_token(token)
        user_id = payload.get("sub")
        if not user_id:
            raise HTTPException(status_code=401, detail="无效的 Token")

        # 校验：仅接受 access token
        if payload.get("type") == TOKEN_TYPE_REFRESH:
            raise HTTPException(
                status_code=401,
                detail="请使用 Access Token 访问 API，Refresh Token 仅用于刷新",
            )

        # 操作人身份要写两处，用途不同，缺一不可：
        # 1) contextvar —— 供本请求任务内的业务/service 日志使用；
        # 2) request.state —— 中间件在独立 task 中执行（Starlette 的
        #    BaseHTTPMiddleware 用 anyio start_soon 起新 task，而 asyncio
        #    每个 Task 会复制 context），因此中间件读不到路由内改的
        #    contextvar，只能通过 ASGI scope 承载的 state 拿到。
        # username 即 assignee_id 口径，比数字 id 直观得多。
        user_label = payload.get("username") or user_id
        user_var.set(user_label)
        request.state.audit_user = user_label
        request.state.audit_display_name = payload.get("display_name") or ""
        request.state.audit_role = payload.get("role") or ""

        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="无效的 Token")