diff --git a/core/auth/users.py b/core/auth/users.py new file mode 100644 index 0000000..238ad1b --- /dev/null +++ b/core/auth/users.py @@ -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$$$`; +- 用户模型:`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")