group_fqcd_jr/app/service/auth_service.py

"""账号密码登录：校验密码、签发访问令牌、留痕审计。

## 为什么放在平台侧

认证是**平台级能力**：所有业务域共用同一套 RBAC（`sys_user_role` → `sys_role_permission`
→ `sys_permission`），`docs/05` §11 把"JWT 签发、刷新、注销"划给统一身份认证模块。
本模块只做**登录这一步**（账号密码换令牌）；刷新与注销留给后续迭代 ——
`app/core/security.py` 已经留好 `RevocationStore` 协议，接上 Redis 即可。

放在 Service 层而不是业务 Agent 里，是因为它不属于任何一个业务域：让业务分支自己加登录
路由，等于又开一条绕过公共鉴权的路径（`AGENTS.md` 规则 7）。

## 令牌里为什么只放 `sub`

`JwtAuthenticator.authenticate` 只从令牌取 `sub`（用户 id），角色/权限/数据范围由
`IdentityService.resolve` **每次请求查库**解析（`identity_repository.load_context`：
`Fresh RBAC reads make revocation immediate`）。这是有意设计——权限变更立即生效、不受
令牌有效期拖累。所以登录只要签一个含 `sub` 的 JWT，**现有鉴权链路一行都不用改**。

三个角色的区分（客户 / 员工 / 管理员）因此已经完备：`bootstrap.py` 里各 Agent 的
`allowed_roles` 早就分开了（面向客户的 Agent 只要 `customer`、`RiskAgent` 要
`risk_operator`/`admin`、`PlatformProbeAgent` 只要 `admin`），此前唯独缺"怎么证明你是谁"。

## 安全约定（金融场景，逐条对应下面的实现）

1. **不区分失败原因**。用户不存在、密码错、账号停用、密码未初始化 —— 对外**同一条** 401
   消息。`docs/05` §3.6 只给了一个 `AUTHENTICATION_REQUIRED`，客户端本来也不该据 message
   区分。否则这个接口就成了账号枚举器。
2. **防时序枚举**。用户不存在时**照样跑一次 bcrypt 比对**（`_DUMMY_HASH`）。否则
   "查无此人"会明显快于"密码错"，同样能枚举出哪些账号存在。
3. **成功与失败都审计**。金融场景必须能回答"谁、什么时候、从哪、试图登录哪个账号、成没成"。
   `interaction_audit.actor_id` 可空，正是为失败场景准备的。
4. **绝不记录密码**。`detail` 里只有用户名与失败原因，没有任何形式的 password 字段。
5. **密码哈希用 bcrypt**。`cryptography` 是给 JWT（RS256）用的，它不提供密码哈希。
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from functools import lru_cache
from pathlib import Path
from typing import Any
from uuid import uuid4

import bcrypt
import jwt
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import get_settings
from app.core.contracts import RequestContext
from app.core.errors import UnauthorizedAgentError
from app.model.audit import InteractionAudit
from app.service.identity_service import IdentityService

logger = logging.getLogger(__name__)

#: 访问令牌有效期。与 `publish_*` 脚本长期使用的 30 分钟一致；权限不放在令牌里，
#: 所以这个值只影响"要不要重新登录"，不影响权限变更的生效速度。
ACCESS_TOKEN_TTL_SECONDS = 1800

#: 对外统一的失败消息。刻意不区分原因，见模块文档第 1 条。
INVALID_CREDENTIALS_MESSAGE = "用户名或密码不正确"

#: 用户不存在时用来比对的固定哈希，见模块文档第 2 条。
#: 用 `bcrypt.hashpw` 现算一次即可，不需要是"某个真实用户的密码"。
_DUMMY_HASH = bcrypt.hashpw(b"not-a-real-password", bcrypt.gensalt())


@lru_cache(maxsize=1)
def _private_key() -> str:
    """签发私钥。只在本进程内缓存，不落任何地方、不进日志。"""
    settings = get_settings()
    path = Path(settings.jwt_private_key_path)
    if not path.is_absolute():
        path = Path.cwd() / path
    return path.read_text(encoding="utf-8")


def issue_access_token(user_id: int) -> tuple[str, int]:
    """签一个只含 `sub` 的访问令牌，返回 (token, expires_in 秒)。

    `security.py` 的 `options={"require": [...]}` 要求
    `sub/iss/aud/exp/nbf/jti` 齐全，缺任何一个都会被判非法令牌。
    """
    settings = get_settings()
    now = datetime.now(UTC)
    expires_in = ACCESS_TOKEN_TTL_SECONDS
    token = jwt.encode(
        {
            "sub": str(user_id),
            "iss": settings.jwt_issuer,
            "aud": settings.jwt_audience,
            "iat": now,
            "nbf": now - timedelta(seconds=5),
            "exp": now + timedelta(seconds=expires_in),
            "jti": str(uuid4()),
        },
        _private_key(),
        algorithm=settings.jwt_algorithm,
    )
    return token, expires_in


def hash_password(password: str) -> str:
    """生成 bcrypt 哈希（成本因子用库默认值）。

    与 `verify_password` 放在一起，是为了让"写密码"和"校验密码"永远用同一套算法 ——
    两个工具脚本（`set_user_password.py` / `create_test_user.py`）都从这里取，
    避免第三次复制粘贴出不一致的实现。
    """
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, stored_hash: str | None) -> bool:
    """常数时间的密码校验；任何异常都当校验失败。

    `stored_hash` 在本项目的现状是**占位符**（种子写 `'x'`、worker 身份写
    `!worker-only-no-password-login!`），它们都不是合法 bcrypt 格式，
    `bcrypt.checkpw` 会抛 `ValueError` —— 必须吞掉并返回 False，
    否则"没设过密码的账号"会变成 500 而不是 401。
    """
    if not stored_hash:
        return False
    try:
        return bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8"))
    except (ValueError, TypeError):
        return False


class AuthService:
    """登录入口。只依赖一个数据库会话，不持有请求上下文（登录时还没有身份）。"""

    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def login(self, username: str, password: str, *, trace_id: str) -> dict[str, Any]:
        """校验账号密码并签发令牌。

        失败一律抛 `UnauthorizedAgentError`（401 `AUTHENTICATION_REQUIRED`），
        由 `app/main.py` 的 `AgentError` 处理器输出 `docs/05` §3.4 的统一错误信封。
        """
        row = (
            await self.session.execute(
                text(
                    "SELECT id, username, password_hash, status "
                    "FROM sys_user WHERE username = :username LIMIT 1"
                ),
                {"username": username},
            )
        ).mappings().first()

        # 不存在时也跑一次 bcrypt，让"查无此人"与"密码错"的耗时一致（见模块文档第 2 条）。
        # 注意不能指望 `verify_password(password, None)` 代劳 —— 它对空哈希直接返回 False，
        # 那就等于"查无此人"立刻返回，时序差异照样能用来枚举账号。
        stored_hash = str(row["password_hash"]) if row is not None else None
        matched = verify_password(password, stored_hash)

        if row is None:
            verify_password(password, _DUMMY_HASH.decode("utf-8"))
            await self._audit(
                actor_id=None,
                action_type="auth.login_failed",
                detail={"username": username, "reason": "user_not_found", "trace_id": trace_id},
            )
            raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE)

        user_id = int(row["id"])
        if not matched:
            await self._audit(
                actor_id=user_id,
                action_type="auth.login_failed",
                detail={"username": username, "reason": "bad_password", "trace_id": trace_id},
            )
            raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE)

        # 账号停用、角色读取失败等一律归到同一条 401：身份解析走的就是请求期那条链路，
        # 保证"能登录"与"登录后能用"用的是同一套判断。
        try:
            resolved = await self._resolve(user_id, trace_id)
        except Exception as exc:
            await self._audit(
                actor_id=user_id,
                action_type="auth.login_failed",
                detail={
                    "username": username,
                    "reason": f"identity_unavailable:{type(exc).__name__}",
                    "trace_id": trace_id,
                },
            )
            raise UnauthorizedAgentError(INVALID_CREDENTIALS_MESSAGE) from exc

        token, expires_in = issue_access_token(user_id)
        await self._audit(
            actor_id=user_id,
            action_type="auth.login_succeeded",
            detail={
                "username": username,
                "roles": list(resolved.roles),
                "trace_id": trace_id,
            },
        )
        return {
            "access_token": token,
            "token_type": "Bearer",
            "expires_in": expires_in,
            "user_id": str(user_id),
            # 前端据此决定进哪个界面；**鉴权仍以库里实时数据为准**，不看这两个字段。
            "roles": list(resolved.roles),
            "data_scope": resolved.data_scope,
        }

    async def _resolve(self, user_id: int, trace_id: str) -> RequestContext:
        """复用请求期的身份解析，保证登录与后续调用看到的是同一套 RBAC。"""
        identity = RequestContext(user_id=str(user_id), trace_id=trace_id)
        return await IdentityService().resolve(identity)

    async def _audit(
        self, *, actor_id: int | None, action_type: str, detail: dict[str, Any]
    ) -> None:
        """登录审计。

        与业务写入分开提交：登录失败时**也要**留下记录，不能因为随后抛异常而被回滚掉。
        审计写失败不阻断登录流程（只告警）——否则审计表的问题会变成"谁都登不进来"。
        """
        self.session.add(
            InteractionAudit(
                actor_type="user",
                actor_id=actor_id,
                portal=None,
                session_id=None,
                action_type=action_type,
                detail=detail,
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
        )
        try:
            await self.session.commit()
        except Exception:
            logger.warning("login audit write failed", exc_info=True)
            await self.session.rollback()