QMDSearch/app/core/users.py

"""用户核心：用户记录存储与密码哈希

存储后端为 Redis（key: user:{username}，JSON 序列化，无 TTL 持久化）；
构造时传入 None 降级为进程级内存 dict（重启丢失，仅用于测试与 Redis 故障兜底）。
所有 Redis 读写异常统一包装为 UserStoreError 抛出（不静默），由上层转 2001。
密码哈希使用标准库 hashlib.pbkdf2_hmac（SHA256，100_000 迭代 + 16 字节随机 salt）。
"""

import hashlib
import hmac
import json
import re
import secrets
from dataclasses import asdict, dataclass
from datetime import UTC, datetime

from redis import asyncio as redis_async
from structlog.typing import FilteringBoundLogger

# 用户名规则：字母/数字/下划线/连字符，2~32 位
USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{2,32}$")

# 密码最小长度
MIN_PASSWORD_LENGTH = 8

# PBKDF2 迭代次数
_HASH_ITERATIONS = 100_000

# salt 字节数（hex 序列化后 32 字符）
_SALT_BYTES = 16

# Redis 键前缀
_USER_KEY_PREFIX = "user:"


class UserExistsError(Exception):
    """用户名已存在"""


class UserStoreError(Exception):
    """用户存储后端读写异常"""


@dataclass
class UserRecord:
    """用户记录（不含明文密码）"""

    username: str
    role: str  # "admin" | "user"
    password_hash: str  # hex
    salt: str  # hex, 16 字节
    must_change_password: bool
    created_at: str  # UTC ISO8601


class UserStore:
    """用户存储：Redis 持久化，构造传 None 降级为内存 dict"""

    def __init__(self, redis_client: redis_async.Redis | None) -> None:
        # redis.asyncio 客户端（需 decode_responses=True）；None → 内存降级
        self._redis = redis_client
        # 内存降级：username -> JSON 字符串（与 Redis 路径同构）
        self._memory: dict[str, str] = {}

    @staticmethod
    def _key(username: str) -> str:
        return f"{_USER_KEY_PREFIX}{username}"

    def hash_password(self, password: str, salt: bytes) -> str:
        """PBKDF2-HMAC-SHA256（100_000 迭代），返回 hex 摘要"""
        return hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, _HASH_ITERATIONS).hex()

    async def create(
        self,
        username: str,
        password: str,
        role: str = "user",
        must_change_password: bool = False,
    ) -> UserRecord:
        """创建用户

        重名抛 UserExistsError；用户名不符合 USERNAME_PATTERN 或密码过短抛 ValueError。
        """
        if not USERNAME_PATTERN.match(username):
            raise ValueError(f"用户名非法: {username!r}（须为 2~32 位字母/数字/_/-）")
        if len(password) < MIN_PASSWORD_LENGTH:
            raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH} 位")
        if await self.get(username) is not None:
            raise UserExistsError(f"用户已存在: {username}")
        salt = secrets.token_bytes(_SALT_BYTES)
        record = UserRecord(
            username=username,
            role=role,
            password_hash=self.hash_password(password, salt),
            salt=salt.hex(),
            must_change_password=must_change_password,
            created_at=datetime.now(UTC).isoformat(),
        )
        await self._write(record)
        return record

    async def get(self, username: str) -> UserRecord | None:
        """按用户名查询，不存在返回 None"""
        raw = await self._read_raw(username)
        if raw is None:
            return None
        return UserRecord(**json.loads(raw))

    async def list(self) -> list[UserRecord]:
        """列出全部用户（扫描 user:* 键）"""
        if self._redis is None:
            raws: list[str | None] = list(self._memory.values())
        else:
            try:
                raws = [await self._redis.get(key) async for key in self._redis.scan_iter(match=f"{_USER_KEY_PREFIX}*")]
            except Exception as e:
                raise UserStoreError("列出用户失败") from e
        return [UserRecord(**json.loads(raw)) for raw in raws if raw is not None]

    async def delete(self, username: str) -> bool:
        """删除用户，返回是否删除成功（幂等：不存在返回 False）"""
        if self._redis is None:
            return self._memory.pop(username, None) is not None
        try:
            return bool(await self._redis.delete(self._key(username)))
        except Exception as e:
            raise UserStoreError(f"删除用户失败: {username}") from e

    async def verify_password(self, username: str, password: str) -> UserRecord | None:
        """校验密码：成功返回用户记录，用户不存在或密码错误返回 None"""
        record = await self.get(username)
        if record is None:
            return None
        candidate = self.hash_password(password, bytes.fromhex(record.salt))
        # 常量时间比较，防时序侧信道
        if not hmac.compare_digest(candidate, record.password_hash):
            return None
        return record

    async def set_password(self, username: str, new_password: str) -> bool:
        """重置密码：重新生成 salt 并哈希，同时清除 must_change_password 标记

        用户不存在返回 False；新密码过短抛 ValueError。
        """
        if len(new_password) < MIN_PASSWORD_LENGTH:
            raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH} 位")
        record = await self.get(username)
        if record is None:
            return False
        salt = secrets.token_bytes(_SALT_BYTES)
        record.salt = salt.hex()
        record.password_hash = self.hash_password(new_password, salt)
        record.must_change_password = False
        await self._write(record)
        return True

    async def count_admins(self) -> int:
        """统计 admin 角色用户数"""
        return sum(1 for record in await self.list() if record.role == "admin")

    async def _read_raw(self, username: str) -> str | None:
        if self._redis is None:
            return self._memory.get(username)
        try:
            return await self._redis.get(self._key(username))
        except Exception as e:
            raise UserStoreError(f"读取用户失败: {username}") from e

    async def _write(self, record: UserRecord) -> None:
        raw = json.dumps(asdict(record), ensure_ascii=False)
        if self._redis is None:
            self._memory[record.username] = raw
            return
        try:
            await self._redis.set(self._key(record.username), raw)
        except Exception as e:
            raise UserStoreError(f"写入用户失败: {record.username}") from e


async def bootstrap_admin(store: UserStore, logger: FilteringBoundLogger) -> str | None:
    """空库引导：无任何用户时创建随机密码的 admin 账号

    明文密码通过 logger.warning 打印一次并返回（仅此一次机会）；
    已有用户时不做任何事，返回 None。
    """
    if await store.list():
        return None
    password = secrets.token_urlsafe(12)
    await store.create("admin", password, role="admin", must_change_password=True)
    logger.warning("已创建默认管理员，请立即登录并修改密码", username="admin", password=password)
    return password