202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
# -*- 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")
|