feat(#150): 本地账号认证 users.py（PBKDF2 密码哈希 + UserStore，对齐 rbac.py 三级角色）

2026-08-05 02:05:21 +00:00
parent b2fa36bd35
commit b5b29d28d7
1 changed files with 201 additions and 0 deletions
@@ -0,0 +1,201 @@
+# -*- coding: utf-8 -*-
+"""④-1 本地账号认证：用户模型 + 密码哈希 —— issue #150 / PRD 8.2。
+
+登录认证是写操作的入口闸门（PRD 8.2「未登录不可访问写操作」）。本模块提供：
+
+- 密码哈希：`PBKDF2-HMAC-SHA256`（hashlib 标准库，盐 16B、迭代 200000，
+  与 Django/OWASP 2023 推荐量级一致），存储形如 `pbkdf2_sha256$<iter>$<salt>$<hash>`；
+- 用户模型：`User`（id/username/password_hash/role/active），角色对齐
+  `core/template-console/rbac.py` 的三级（readonly/engineer/admin）；
+- `UserStore`：内存/可持久化用户仓，提供 create / authenticate / get / list。
+
+数据可落 PostgreSQL users 表（#30 schema，见 `postgres_users_schema.py`）；
+本模块零运行时依赖（仅标准库），`UserStore` 默认内存，接入 PG 时实现
+`PgUserBackend` 即可，上层 API 不变。
+
+安全要点：
+- 永不存储明文密码；`authenticate` 用恒定时间比较防时序侧信道（`hmac.compare_digest`）；
+- 内置初始管理员账号仅在首次初始化时创建，生产部署必须改密。
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import os
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+# ---------------------------------------------------------------------------
+# 常量（与 #30 users 表 schema 对齐）
+# ---------------------------------------------------------------------------
+
+PBKDF2_ALGO = "sha256"
+PBKDF2_ITER = 200_000          # OWASP 2023 推荐（SHA256）
+SALT_BYTES = 16
+HASH_BYTES = 32
+HASH_SCHEME = "pbkdf2_sha256"  # 存储前缀，便于将来升级算法
+
+# 角色与 rbac.py 一致（readonly/engineer/admin）
+VALID_ROLES = ("readonly", "engineer", "admin")
+
+
+# ---------------------------------------------------------------------------
+# 密码哈希
+# ---------------------------------------------------------------------------
+
+def hash_password(password: str, *, iterations: int = PBKDF2_ITER,
+                  salt: Optional[bytes] = None) -> str:
+    """返回 `scheme$iter$<salt-b64>$<hash-b64>` 形式的密码哈希。"""
+    if not isinstance(password, str) or not password:
+        raise ValueError("password must be a non-empty string")
+    if iterations < 100_000:
+        raise ValueError("iterations must be >= 100000")
+    raw_salt = salt if salt is not None else os.urandom(SALT_BYTES)
+    if len(raw_salt) != SALT_BYTES:
+        raise ValueError("salt must be %d bytes" % SALT_BYTES)
+    dk = hashlib.pbkdf2_hmac(PBKDF2_ALGO, password.encode("utf-8"),
+                             raw_salt, iterations, HASH_BYTES)
+    return "%s$%d$%s$%s" % (HASH_SCHEME, iterations,
+                            base64.b64encode(raw_salt).decode("ascii"),
+                            base64.b64encode(dk).decode("ascii"))
+
+
+def verify_password(password: str, stored: str) -> bool:
+    """恒定时间校验密码。`stored` 为 `hash_password` 的输出。"""
+    try:
+        scheme, iter_s, salt_b64, hash_b64 = stored.split("$")
+        if scheme != HASH_SCHEME:
+            return False
+        iterations = int(iter_s)
+        raw_salt = base64.b64decode(salt_b64)
+        expected = base64.b64decode(hash_b64)
+    except (ValueError, AttributeError):
+        return False
+    dk = hashlib.pbkdf2_hmac(PBKDF2_ALGO, password.encode("utf-8"),
+                             raw_salt, iterations, len(expected))
+    return hmac.compare_digest(dk, expected)  # 恒定时间比较，防时序侧信道
+
+
+# ---------------------------------------------------------------------------
+# 用户模型
+# ---------------------------------------------------------------------------
+
+@dataclass
+class User:
+    id: int
+    username: str
+    password_hash: str
+    role: str = "readonly"          # readonly/engineer/admin
+    active: bool = True
+    created_at: float = field(default_factory=time.time)
+    last_login_at: Optional[float] = None
+
+    def __post_init__(self) -> None:
+        if self.role not in VALID_ROLES:
+            raise ValueError("role must be one of %r" % (VALID_ROLES,))
+
+    @property
+    def is_admin(self) -> bool:
+        return self.role == "admin"
+
+    def to_public(self) -> Dict:
+        """对外暴露（不含 password_hash）。"""
+        return {
+            "id": self.id,
+            "username": self.username,
+            "role": self.role,
+            "active": self.active,
+            "created_at": self.created_at,
+            "last_login_at": self.last_login_at,
+        }
+
+
+# ---------------------------------------------------------------------------
+# 用户仓
+# ---------------------------------------------------------------------------
+
+class UserStore:
+    """内存用户仓（默认）。接 PostgreSQL 时换 PgUserBackend，API 不变。"""
+
+    def __init__(self) -> None:
+        self._users: Dict[int, User] = {}
+        self._by_name: Dict[str, int] = {}
+        self._next_id = 1
+        self._lock = threading.Lock()
+
+    def create(self, username: str, password: str, *, role: str = "readonly",
+               active: bool = True) -> User:
+        username = (username or "").strip()
+        if not username or len(username) > 64:
+            raise ValueError("username must be 1..64 chars")
+        if len(password) < 8:
+            raise ValueError("password must be >= 8 chars")
+        with self._lock:
+            if username in self._by_name:
+                raise ValueError("username already exists: %s" % username)
+            uid = self._next_id
+            user = User(id=uid, username=username,
+                        password_hash=hash_password(password), role=role, active=active)
+            self._users[uid] = user
+            self._by_name[username] = uid
+            self._next_id += 1
+            return user
+
+    def set_password(self, user_id: int, new_password: str) -> None:
+        if len(new_password) < 8:
+            raise ValueError("password must be >= 8 chars")
+        with self._lock:
+            user = self._users.get(user_id)
+            if user is None:
+                raise KeyError("user not found: %r" % (user_id,))
+            user.password_hash = hash_password(new_password)
+
+    def set_role(self, user_id: int, role: str) -> None:
+        if role not in VALID_ROLES:
+            raise ValueError("role must be one of %r" % (VALID_ROLES,))
+        with self._lock:
+            user = self._users.get(user_id)
+            if user is None:
+                raise KeyError("user not found: %r" % (user_id,))
+            user.role = role
+
+    def set_active(self, user_id: int, active: bool) -> None:
+        with self._lock:
+            user = self._users.get(user_id)
+            if user is None:
+                raise KeyError("user not found: %r" % (user_id,))
+            user.active = active
+
+    def get(self, user_id: int) -> Optional[User]:
+        return self._users.get(user_id)
+
+    def get_by_name(self, username: str) -> Optional[User]:
+        uid = self._by_name.get((username or "").strip())
+        return self._users.get(uid) if uid is not None else None
+
+    def list(self) -> List[User]:
+        return [self._users[i] for i in sorted(self._users)]
+
+    def authenticate(self, username: str, password: str) -> Optional[User]:
+        """成功返回 User（并刷新 last_login_at），失败返回 None。
+
+        失败原因不区分"用户不存在"与"密码错"，避免用户名枚举。
+        """
+        user = self.get_by_name(username)
+        if user is None or not user.active:
+            return None
+        if not verify_password(password, user.password_hash):
+            return None
+        with self._lock:
+            user.last_login_at = time.time()
+        return user
+
+    def ensure_bootstrap_admin(self, username: str, password: str) -> User:
+        """首次初始化管理员账号（已存在则返回，不改密）。"""
+        existing = self.get_by_name(username)
+        if existing is not None:
+            return existing
+        return self.create(username, password, role="admin")