# -*- coding: utf-8 -*- """④-1 本地账号认证：用户模型 + 密码哈希 —— issue #150 / PRD 8.2。 登录认证是写操作的入口闸门（PRD 8.2「未登录不可访问写操作」）。本模块提供： - 密码哈希：`PBKDF2-HMAC-SHA256`（hashlib 标准库，盐 16B、迭代 200000， 与 Django/OWASP 2023 推荐量级一致），存储形如 `pbkdf2_sha256$$$`； - 用户模型：`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$$` 形式的密码哈希。""" 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")