#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 用户系统数据库模块 — BIZ-129 方案 C 表： - users 用户账号（含多用户预留字段 / 登录失败锁定字段） - sessions 服务端 session（存 token_hash，不存明文 cookie 值） - password_reset_tokens 找回密码一次性 token - operation_logs 操作日志（90 天保留） 约束对齐 — BIZ-127 架构侧清单： #2 sessions.token_hash = sha256(cookie_value).hexdigest() #5 users.tenant_id / users.status（数据层预留，UI 不写） #6 users.failed_attempts / users.locked_until（登录失败锁定） 升级路径：旧装已存在的 `users` / `sessions` 表会被 `migrate_legacy_schema()` 增量升级， 不会丢数据。新装直接走 CREATE TABLE。 """ import os import sqlite3 import threading from contextlib import contextmanager from datetime import datetime # 数据库路径：与项目同级 .data/lottery_auth.db BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE_DIR, '.data') DB_PATH = os.path.join(DATA_DIR, 'lottery_auth.db') _db_lock = threading.Lock() def _ensure_data_dir(): os.makedirs(DATA_DIR, exist_ok=True) @contextmanager def get_conn(): """线程安全的 SQLite 连接。""" _ensure_data_dir() conn = sqlite3.connect(DB_PATH, timeout=10, isolation_level=None) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode = WAL") try: yield conn finally: conn.close() # ============================================================ # 表结构（BIZ-127 8 条约束对齐版） # ============================================================ SCHEMA = [ """ CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'user', -- 'admin' / 'user' is_active INTEGER NOT NULL DEFAULT 1, tenant_id TEXT, -- #5 多用户预留 status TEXT NOT NULL DEFAULT 'active',-- #5 active/disabled failed_attempts INTEGER NOT NULL DEFAULT 0, -- #6 失败计数 locked_until TEXT, -- #6 锁定到期时间 created_at TEXT NOT NULL, last_login_at TEXT ) """, """ CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, token_hash TEXT NOT NULL UNIQUE, -- #2 sha256(cookie_value) user_id INTEGER NOT NULL, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, -- 默认 30 天 ip TEXT, user_agent TEXT, revoked INTEGER NOT NULL DEFAULT 0, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS password_reset_tokens ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, token TEXT NOT NULL UNIQUE, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, -- 2 小时有效 used_at TEXT, -- 使用时间（一次性） ip TEXT, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """, """ CREATE TABLE IF NOT EXISTS operation_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, -- 可为空（如登录失败） username TEXT, -- 冗余存储便于查询 action TEXT NOT NULL, -- login_success / login_fail / ... ip TEXT, user_agent TEXT, detail TEXT, -- JSON 字符串 success INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL ) """, "CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id)", "CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at)", "CREATE INDEX IF NOT EXISTS idx_sessions_token_hash ON sessions(token_hash)", "CREATE INDEX IF NOT EXISTS idx_reset_tokens_token ON password_reset_tokens(token)", "CREATE INDEX IF NOT EXISTS idx_op_logs_user ON operation_logs(user_id)", "CREATE INDEX IF NOT EXISTS idx_op_logs_action ON operation_logs(action)", "CREATE INDEX IF NOT EXISTS idx_op_logs_created ON operation_logs(created_at)", ] # ============================================================ # 数据迁移 SQL — 旧装兼容 # ============================================================ # 旧版 users 缺 tenant_id / status / failed_attempts / locked_until # 旧版 sessions 用 session_id 明文，需 rename → token_hash 并回填 sha256 # # 注意：SQLite ALTER TABLE 不能改 UNIQUE 约束列名，所以采用 # 1. add column token_hash_temp（无 UNIQUE） # 2. UPDATE token_hash_temp = sha256(session_id) # 3. drop 原 session_id 列（不支持 DROP COLUMN 的旧版 SQLite 走 fallback） # 4. rename token_hash_temp → token_hash # 的方式。新版 SQLite（≥3.35）支持 DROP COLUMN，本脚本两种路径都跑。 def _table_has_column(conn, table: str, column: str) -> bool: rows = conn.execute(f"PRAGMA table_info({table})").fetchall() return any(r['name'] == column for r in rows) def _sqlite_supports_drop_column(conn) -> bool: row = conn.execute("SELECT sqlite_version() AS v").fetchone() try: # SQLite 3.35.0 起支持 DROP COLUMN major, minor, _ = (int(x) for x in row['v'].split('.')) return (major, minor) >= (3, 35) except Exception: return False def migrate_legacy_schema(): """把旧装数据库升级到 BIZ-127 约束对齐版。幂等。 返回 dict 报告做了哪些迁移（便于测试断言）。 """ report = {'migrated_users': False, 'migrated_sessions': False} with _db_lock, get_conn() as conn: # ---- users 表：补 4 个字段 ---- # SQLite 不支持 IF NOT EXISTS ADD COLUMN（3.35 之前），用 PRAGMA 查 if not _table_has_column(conn, 'users', 'tenant_id'): conn.execute("ALTER TABLE users ADD COLUMN tenant_id TEXT") report['migrated_users'] = True if not _table_has_column(conn, 'users', 'status'): conn.execute("ALTER TABLE users ADD COLUMN status TEXT NOT NULL DEFAULT 'active'") report['migrated_users'] = True if not _table_has_column(conn, 'users', 'failed_attempts'): conn.execute("ALTER TABLE users ADD COLUMN failed_attempts INTEGER NOT NULL DEFAULT 0") report['migrated_users'] = True if not _table_has_column(conn, 'users', 'locked_until'): conn.execute("ALTER TABLE users ADD COLUMN locked_until TEXT") report['migrated_users'] = True # ---- sessions 表：session_id → token_hash（sha256 化）---- if _table_has_column(conn, 'sessions', 'session_id') \ and not _table_has_column(conn, 'sessions', 'token_hash'): # 用 temp 列 + 改名方式，不依赖 DROP COLUMN conn.execute("ALTER TABLE sessions ADD COLUMN token_hash_temp TEXT") # 回填 sha256(session_id) rows = conn.execute( "SELECT id, session_id FROM sessions WHERE session_id IS NOT NULL" ).fetchall() for r in rows: # Python 端算 sha256（避免 SQLite 函数兼容性问题） import hashlib digest = hashlib.sha256(r['session_id'].encode('utf-8')).hexdigest() conn.execute( "UPDATE sessions SET token_hash_temp = ? WHERE id = ?", (digest, r['id']), ) if _sqlite_supports_drop_column(conn): # 新版 SQLite：直接 drop 旧列 + rename temp conn.execute("ALTER TABLE sessions DROP COLUMN session_id") conn.execute("ALTER TABLE sessions RENAME COLUMN token_hash_temp TO token_hash") else: # 旧版 SQLite：建新表，复制数据，drop 旧表，rename conn.execute(""" CREATE TABLE sessions_new ( id INTEGER PRIMARY KEY AUTOINCREMENT, token_hash TEXT NOT NULL UNIQUE, user_id INTEGER NOT NULL, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, ip TEXT, user_agent TEXT, revoked INTEGER NOT NULL DEFAULT 0, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """) conn.execute(""" INSERT INTO sessions_new (id, token_hash, user_id, created_at, expires_at, ip, user_agent, revoked) SELECT id, token_hash_temp, user_id, created_at, expires_at, ip, user_agent, revoked FROM sessions """) conn.execute("DROP TABLE sessions") conn.execute("ALTER TABLE sessions_new RENAME TO sessions") report['migrated_sessions'] = True return report # ============================================================ # 初始化 # ============================================================ def init_db(): """初始化所有表（幂等），并对旧装做 schema 迁移。 新装：CREATE TABLE 走完即可。 旧装：CREATE TABLE IF NOT EXISTS 跳过已存在表，迁移 SQL 补齐字段。 """ with _db_lock, get_conn() as conn: cur = conn.cursor() for stmt in SCHEMA: cur.execute(stmt) # 迁移在 init 完成后单独跑（需要先有 users/sessions 表存在） migrate_legacy_schema() def now_iso(): return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S') if __name__ == '__main__': init_db() print(f"✅ 初始化完成：{DB_PATH}")