harness_zbt/backend/app/domains/identity/service.py

"""H5 用户和后台管理员的认证业务逻辑。

路由层负责收发 HTTP 数据，本文件负责验证码、登录、令牌刷新和退出等业务规则。
"""

import re
from dataclasses import replace
from datetime import UTC, datetime, timedelta
from typing import Any

from app.core.config import Settings
from app.core.errors import AppError
from app.core.identifiers import new_ulid
from app.core.passwords import PasswordService
from app.core.security import JwtService
from app.domains.identity.models import AdminRole, AdminUser, AuthSession, H5User
from app.domains.identity.repository import IdentityRepository

MOBILE_PATTERN = re.compile(r"^1\d{10}$")
ADMIN_USERNAME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$")
ROLE_PERMISSION_CATALOG: tuple[dict[str, str], ...] = (
    {"code": "dashboard:read", "name": "查看运营总览", "group": "运营分析"},
    {"code": "order:read", "name": "查看全部订单", "group": "保单业务"},
    {"code": "policy:read", "name": "查看全部保单", "group": "保单业务"},
    {"code": "user:read", "name": "查看用户信息", "group": "客户管理"},
    {"code": "user:write", "name": "维护用户状态", "group": "客户管理"},
    {"code": "product:read", "name": "查看产品配置", "group": "产品管理"},
    {"code": "product:write", "name": "维护产品与版本", "group": "产品管理"},
    {"code": "attribution:read", "name": "查看推广归因", "group": "渠道推广"},
    {"code": "role:read", "name": "查看角色权限", "group": "系统治理"},
    {"code": "role:write", "name": "编辑角色权限", "group": "系统治理"},
)
ROLE_NAMES = {
    "SUPER_ADMIN": "超级管理员",
    "OPERATOR": "运营人员",
    "REVIEWER": "审核人员",
    "SALESPERSON": "业务员",
}
ALLOWED_DATA_SCOPES = {"ALL", "MASKED_ALL", "READ_ONLY", "SELF"}


class H5AuthService:
    """处理 H5 用户认证流程。"""

    def __init__(
        self,
        settings: Settings,
        repository: IdentityRepository,
        jwt_service: JwtService,
    ) -> None:
        self._settings = settings
        self._repository = repository
        self._jwt = jwt_service

    def request_code(self, mobile: str) -> dict[str, int]:
        """校验手机号并请求验证码；当前开发环境使用固定验证码。"""
        self._validate_mobile(mobile)
        if self._settings.app_env == "production" or not self._settings.dev_tools_enabled:
            raise AppError("DEPENDENCY_UNAVAILABLE", "当前环境未配置验证码服务", 503)
        return {"expires_in": 300}

    def login(self, mobile: str, code: str) -> dict[str, Any]:
        """验证手机号和验证码，必要时创建用户，并签发一对 JWT。"""
        self._validate_mobile(mobile)
        if code != self._settings.dev_fixed_otp:
            raise AppError("INVALID_OTP", "验证码错误或已过期", 400)
        user = self._repository.get_h5_user_by_mobile(mobile)
        if user is None:
            # 首次登录即注册：数据库中没有该手机号时创建一条用户记录。
            user = H5User(
                id=new_ulid(),
                mobile=mobile,
                mobile_masked=self._mask_mobile(mobile),
                display_name=f"用户{mobile[-4:]}",
                status="ACTIVE",
                created_at=datetime.now(UTC),
            )
            self._repository.save_h5_user(user)
        if user.status != "ACTIVE":
            raise AppError("PERMISSION_DENIED", "用户已被禁用", 403)
        # session_id 把 access token、refresh token 与服务端会话记录关联起来。
        session_id = new_ulid()
        tokens = self._jwt.issue_pair(
            subject_id=user.id,
            subject_type="H5_USER",
            audience="zbt-h5",
            session_id=session_id,
        )
        self._repository.save_session(
            AuthSession(
                id=session_id,
                subject_type="H5_USER",
                subject_id=user.id,
                refresh_jti_hash=self._jwt.hash_token_identifier(tokens.refresh_jti),
                expires_at=datetime.now(UTC) + timedelta(days=self._settings.refresh_token_days),
            )
        )
        return {
            "user": self._user_data(user),
            "tokens": {
                "access_token": tokens.access_token,
                "refresh_token": tokens.refresh_token,
                "token_type": "bearer",
                "access_expires_in": tokens.access_expires_in,
                "refresh_expires_in": tokens.refresh_expires_in,
            },
        }

    def authenticate_access(self, token: str) -> H5User:
        """验证 access token 和服务端会话，返回当前登录用户。"""
        payload = self._jwt.decode_access(token, audience="zbt-h5")
        session = self._repository.get_session(str(payload["sid"]))
        if session is None or not session.active:
            raise AppError("AUTH_REQUIRED", "登录会话已失效", 401)
        user = self._repository.get_h5_user(str(payload["sub"]))
        if user is None or user.status != "ACTIVE":
            raise AppError("AUTH_REQUIRED", "用户不存在或已禁用", 401)
        return user

    def refresh(self, refresh_token: str) -> dict[str, Any]:
        """用 refresh token 换取新令牌，并使旧 refresh token 失效。"""
        payload = self._jwt.decode_refresh(refresh_token, audience="zbt-h5")
        session = self._repository.get_session(str(payload["sid"]))
        # 数据库只保存 jti 的哈希，数据库泄漏时不会直接暴露令牌标识。
        presented_jti_hash = self._jwt.hash_token_identifier(str(payload["jti"]))
        if session is None or not session.active or session.refresh_jti_hash != presented_jti_hash:
            raise AppError("AUTH_REQUIRED", "刷新凭证已失效", 401)
        user = self._repository.get_h5_user(str(payload["sub"]))
        if user is None or user.status != "ACTIVE":
            raise AppError("AUTH_REQUIRED", "用户不存在或已禁用", 401)
        tokens = self._jwt.issue_pair(
            subject_id=user.id,
            subject_type="H5_USER",
            audience="zbt-h5",
            session_id=session.id,
        )
        session.refresh_jti_hash = self._jwt.hash_token_identifier(tokens.refresh_jti)
        session.expires_at = datetime.now(UTC) + timedelta(days=self._settings.refresh_token_days)
        self._repository.save_session(session)
        return {
            "user": self._user_data(user),
            "tokens": {
                "access_token": tokens.access_token,
                "refresh_token": tokens.refresh_token,
                "token_type": "bearer",
                "access_expires_in": tokens.access_expires_in,
                "refresh_expires_in": tokens.refresh_expires_in,
            },
        }

    def logout(self, access_token: str) -> None:
        """撤销服务端会话；即使 JWT 尚未过期，之后也无法继续使用。"""
        payload = self._jwt.decode_access(access_token, audience="zbt-h5")
        session = self._repository.get_session(str(payload["sid"]))
        if session is not None and session.revoked_at is None:
            session.revoked_at = datetime.now(UTC)
            self._repository.save_session(session)

    @staticmethod
    def _validate_mobile(mobile: str) -> None:
        if not MOBILE_PATTERN.fullmatch(mobile):
            raise AppError("VALIDATION_ERROR", "手机号格式无效", 400)

    @staticmethod
    def _mask_mobile(mobile: str) -> str:
        return f"{mobile[:3]}****{mobile[-4:]}"

    @staticmethod
    def _user_data(user: H5User) -> dict[str, str]:
        return {
            "id": user.id,
            "mobile": user.mobile,
            "mobile_masked": user.mobile_masked,
            "display_name": user.display_name,
            "status": user.status,
        }


class AdminAuthService:
    """处理后台管理员认证，与 H5 用户使用不同的 audience。"""

    def __init__(
        self,
        settings: Settings,
        repository: IdentityRepository,
        jwt_service: JwtService,
        password_service: PasswordService,
    ) -> None:
        self._settings = settings
        self._repository = repository
        self._jwt = jwt_service
        self._passwords = password_service

    def login(self, username: str, password: str) -> dict[str, Any]:
        user = self._repository.get_admin_user_by_username(username)
        if (
            user is None
            or user.status != "ACTIVE"
            or not self._passwords.verify(password, user.password_hash)
        ):
            raise AppError("AUTH_REQUIRED", "用户名或密码错误", 401)
        session_id = new_ulid()
        tokens = self._jwt.issue_pair(
            subject_id=user.id,
            subject_type="ADMIN_USER",
            audience="zbt-admin",
            session_id=session_id,
        )
        self._repository.save_session(
            AuthSession(
                id=session_id,
                subject_type="ADMIN_USER",
                subject_id=user.id,
                refresh_jti_hash=self._jwt.hash_token_identifier(tokens.refresh_jti),
                expires_at=datetime.now(UTC) + timedelta(days=self._settings.refresh_token_days),
            )
        )
        return {
            "user": self._user_data(user),
            "tokens": {
                "access_token": tokens.access_token,
                "refresh_token": tokens.refresh_token,
                "token_type": "bearer",
                "access_expires_in": tokens.access_expires_in,
                "refresh_expires_in": tokens.refresh_expires_in,
            },
        }

    def authenticate_access(self, token: str) -> AdminUser:
        payload = self._jwt.decode_access(token, audience="zbt-admin")
        session = self._repository.get_session(str(payload["sid"]))
        if session is None or not session.active:
            raise AppError("AUTH_REQUIRED", "登录会话已失效", 401)
        user = self._repository.get_admin_user(str(payload["sub"]))
        if user is None or user.status != "ACTIVE":
            raise AppError("AUTH_REQUIRED", "后台账号不存在或已禁用", 401)
        return user

    def refresh(self, refresh_token: str) -> dict[str, Any]:
        """轮换后台刷新令牌，并维持同一个服务端会话。"""
        payload = self._jwt.decode_refresh(refresh_token, audience="zbt-admin")
        session = self._repository.get_session(str(payload["sid"]))
        presented_jti_hash = self._jwt.hash_token_identifier(str(payload["jti"]))
        if session is None or not session.active or session.refresh_jti_hash != presented_jti_hash:
            raise AppError("AUTH_REQUIRED", "刷新凭证已失效", 401)
        user = self._repository.get_admin_user(str(payload["sub"]))
        if user is None or user.status != "ACTIVE":
            raise AppError("AUTH_REQUIRED", "后台账号不存在或已禁用", 401)
        tokens = self._jwt.issue_pair(
            subject_id=user.id,
            subject_type="ADMIN_USER",
            audience="zbt-admin",
            session_id=session.id,
        )
        session.refresh_jti_hash = self._jwt.hash_token_identifier(tokens.refresh_jti)
        session.expires_at = datetime.now(UTC) + timedelta(days=self._settings.refresh_token_days)
        self._repository.save_session(session)
        return {
            "user": self._user_data(user),
            "tokens": {
                "access_token": tokens.access_token,
                "refresh_token": tokens.refresh_token,
                "token_type": "bearer",
                "access_expires_in": tokens.access_expires_in,
                "refresh_expires_in": tokens.refresh_expires_in,
            },
        }

    def logout(self, access_token: str) -> None:
        """撤销后台服务端会话。"""
        payload = self._jwt.decode_access(access_token, audience="zbt-admin")
        session = self._repository.get_session(str(payload["sid"]))
        if session is not None and session.revoked_at is None:
            session.revoked_at = datetime.now(UTC)
            self._repository.save_session(session)

    def list_roles(self) -> dict[str, Any]:
        roles = self._repository.list_roles()
        return {
            "items": [self._role_data(role) for role in roles],
            "permission_catalog": list(ROLE_PERMISSION_CATALOG),
            "data_scopes": [
                {"code": "ALL", "name": "全部数据"},
                {"code": "MASKED_ALL", "name": "全部脱敏数据"},
                {"code": "READ_ONLY", "name": "只读数据"},
                {"code": "SELF", "name": "仅本人数据"},
            ],
        }

    def update_role(
        self,
        role_code: str,
        *,
        permissions: list[str],
        data_scope: str,
    ) -> dict[str, Any]:
        if role_code == "SUPER_ADMIN":
            raise AppError(
                "PROTECTED_ROLE",
                "超级管理员为系统保护角色，不允许修改",
                409,
            )
        if data_scope not in ALLOWED_DATA_SCOPES:
            raise AppError("VALIDATION_ERROR", "数据范围无效", 400)
        allowed_permissions = {item["code"] for item in ROLE_PERMISSION_CATALOG}
        requested_permissions = set(permissions)
        if not requested_permissions.issubset(allowed_permissions):
            raise AppError("VALIDATION_ERROR", "包含未定义的权限项", 400)
        role = next(
            (item for item in self._repository.list_roles() if item.code == role_code),
            None,
        )
        if role is None:
            raise AppError("ROLE_NOT_FOUND", "角色不存在", 404)
        updated = replace(
            role,
            name=ROLE_NAMES.get(role.code, role.name),
            data_scope=data_scope,
            permissions=tuple(sorted(requested_permissions)),
        )
        self._repository.save_role(updated)
        return self._role_data(updated)

    def list_managed_users(self) -> dict[str, Any]:
        return {
            "admin_users": [self._user_data(user) for user in self._repository.list_admin_users()],
            "customers": [self._customer_data(user) for user in self._repository.list_h5_users()],
        }

    def create_admin_user(
        self,
        *,
        username: str,
        password: str,
        display_name: str,
        role_code: str,
    ) -> dict[str, Any]:
        normalized_username = username.strip()
        normalized_name = display_name.strip()
        self._validate_admin_account_input(
            normalized_username,
            password,
            normalized_name,
        )
        if self._repository.get_admin_user_by_username(normalized_username) is not None:
            raise AppError("USERNAME_EXISTS", "后台账号已存在", 409)
        role = self._find_role(role_code)
        user = AdminUser(
            id=new_ulid(),
            username=normalized_username,
            password_hash=self._passwords.hash(password),
            display_name=normalized_name,
            status="ACTIVE",
            roles=(role.code,),
            permissions=role.permissions,
            data_scope=role.data_scope,
        )
        self._repository.save_admin_user(user)
        return self._user_data(user)

    def update_admin_user(
        self,
        actor: AdminUser,
        user_id: str,
        *,
        display_name: str,
        status: str,
        role_code: str,
    ) -> dict[str, Any]:
        target = self._repository.get_admin_user(user_id)
        if target is None:
            raise AppError("ADMIN_USER_NOT_FOUND", "后台账号不存在", 404)
        normalized_name = display_name.strip()
        if not normalized_name:
            raise AppError("VALIDATION_ERROR", "账号姓名不能为空", 400)
        if status not in {"ACTIVE", "DISABLED"}:
            raise AppError("VALIDATION_ERROR", "账号状态无效", 400)
        if target.id == actor.id and status != "ACTIVE":
            raise AppError("SELF_DISABLE_FORBIDDEN", "不能停用当前登录账号", 409)
        role = self._find_role(role_code)
        if "SUPER_ADMIN" in target.roles and (status != "ACTIVE" or role.code != "SUPER_ADMIN"):
            active_super_admins = [
                item
                for item in self._repository.list_admin_users()
                if item.status == "ACTIVE" and "SUPER_ADMIN" in item.roles
            ]
            if len(active_super_admins) <= 1:
                raise AppError(
                    "LAST_SUPER_ADMIN",
                    "系统必须保留至少一个启用的超级管理员",
                    409,
                )
        updated = replace(
            target,
            display_name=normalized_name,
            status=status,
            roles=(role.code,),
            permissions=role.permissions,
            data_scope=role.data_scope,
        )
        self._repository.save_admin_user(updated)
        return self._user_data(updated)

    def reset_admin_password(
        self,
        user_id: str,
        *,
        new_password: str,
    ) -> dict[str, Any]:
        target = self._repository.get_admin_user(user_id)
        if target is None:
            raise AppError("ADMIN_USER_NOT_FOUND", "后台账号不存在", 404)
        if len(new_password) < 8:
            raise AppError("VALIDATION_ERROR", "新密码至少需要8位", 400)
        updated = replace(
            target,
            password_hash=self._passwords.hash(new_password),
        )
        self._repository.save_admin_user(updated)
        return {"user_id": updated.id, "password_reset": True}

    def update_customer_status(
        self,
        user_id: str,
        *,
        status: str,
    ) -> dict[str, Any]:
        user = self._repository.get_h5_user(user_id)
        if user is None:
            raise AppError("H5_USER_NOT_FOUND", "投保用户不存在", 404)
        if status not in {"ACTIVE", "DISABLED"}:
            raise AppError("VALIDATION_ERROR", "用户状态无效", 400)
        updated = replace(user, status=status)
        self._repository.save_h5_user(updated)
        return self._customer_data(updated)

    def _find_role(self, role_code: str) -> AdminRole:
        role = next(
            (item for item in self._repository.list_roles() if item.code == role_code),
            None,
        )
        if role is None:
            raise AppError("ROLE_NOT_FOUND", "角色不存在", 404)
        return role

    @staticmethod
    def _validate_admin_account_input(
        username: str,
        password: str,
        display_name: str,
    ) -> None:
        if not ADMIN_USERNAME_PATTERN.fullmatch(username):
            raise AppError(
                "VALIDATION_ERROR",
                "账号需以字母开头，可包含字母、数字、点、下划线或短横线",
                400,
            )
        if len(password) < 8:
            raise AppError("VALIDATION_ERROR", "密码至少需要8位", 400)
        if not display_name:
            raise AppError("VALIDATION_ERROR", "账号姓名不能为空", 400)

    @staticmethod
    def _user_data(user: AdminUser) -> dict[str, Any]:
        return {
            "id": user.id,
            "username": user.username,
            "display_name": user.display_name,
            "status": user.status,
            "roles": sorted(user.roles),
            "permissions": sorted(user.permissions),
            "data_scope": user.data_scope,
        }

    @staticmethod
    def _role_data(role: AdminRole) -> dict[str, Any]:
        return {
            "code": role.code,
            "name": ROLE_NAMES.get(role.code, role.name),
            "data_scope": role.data_scope,
            "status": role.status,
            "permissions": sorted(role.permissions),
            "protected": role.code == "SUPER_ADMIN",
        }

    @staticmethod
    def _customer_data(user: H5User) -> dict[str, Any]:
        return {
            "id": user.id,
            "mobile": user.mobile,
            "mobile_masked": user.mobile_masked,
            "display_name": user.display_name,
            "status": user.status,
            "created_at": user.created_at,
        }