Archived
snowy-yang
96af56cc27
拆掉两类垫片的最后一处：auth_system 原为同步 sqlite3 栈，web 层与 services 的
worker 只能靠 asyncio.to_thread 逐个桥接才不阻塞事件循环——每次 DB 访问都白付一次
线程池调度，且「同步调异步」的边界散落在 100+ 个调用点。B1 已把 worker 变成协程，
本批把访问层本身也迁到原生异步，两条线在此合拢。
改动（按子系统）：
- auth_system：db.py 照 core/db.py 现有范式重建为原生异步访问层（aiosqlite + WAL）
——读路径 _q 每次独立连接、无锁、可并发；写路径 _x / write_transaction 用
asyncio.Lock 串行化，写者不阻塞读者。users / cards / credits / ledger / logs /
settings / mail / geo / backup 九模块随之全部 async/await；security.py 是纯计算
（pbkdf2/normalize）不动。__init__ 去掉 _DB_LOCK 与 get_conn 导出。
- web：app / deps / billing / tasks_registry 与 routers/{admin_site,auth,music,
register,user} 直接 await，拆掉包裹的 to_thread。deps.secret_key 改 async，
连带 _parse_cookie 也 async（会话解析要读密钥）。
- services：_taskcore / tasks / downloads / podcast 的账本调用与查询直接 await。
4 处 start_task 调用点（download.py）确认已 await。
设计取舍（为什么这么做，而不是直接包一层）：
- 锁按**事件循环**隔离（_LOOP_LOCKS 以 id(loop) 为键）：asyncio.Lock 绑定创建它的
loop，应用主 loop 与测试各自的 loop 并存，跨 loop 复用会抛 "bound to a different
event loop"。
- WAL 切换用**独立锁** _wal_lock 而非 _write_lock：execute() 已持写锁，再在锁内
connect→_configure 会自锁死（asyncio.Lock 非重入）。切换 + 标记在专用锁内串行并
二次复查，否则并发首连会同时执行 PRAGMA journal_mode=WAL 争排他锁报
database is locked（core/db.py 同款教训，实测复现）。
- force_wal 保留：journal_mode 持久化在库文件头，库文件被删重建时新文件不继承，
必须重设。exists() 判定须在 aiosqlite.connect **之前**（connect 会创建空文件）。
- connect() 用 DB_PATH.parent 兜底建目录，优于 DATA_DIR：生产两者相同，但测试
monkeypatch 只换 DB_PATH 时兜底仍能命中正确目录。
留在 to_thread 上的仅剩**正当的阻塞卸载**，不是同步门面：PBKDF2 的 10 万次迭代
（登录链路每次都要跑）、smtplib 发信、备份文件的 copy2 / os.replace、secret.key
的文件 IO。原有 aria2 / downloader / zip 侧的 to_thread 未动。
不变量逐条保持（迁移只改执行模型，不改语义）：
- ledger_settle 幂等：账本非 open 直接 return 0，重启后重放不重复退款；
- 终态先到先得：_mark_task_terminal 对已 done/error/canceled 的任务直接 return，
堵住 canceled→error 的非法迁移边；
- settle 必须在任务锁**外**调用（ledger_settle 要抢 SQLite 写锁，busy 最长 15s，
持任务锁会让全站进度/提交/取消随库停摆）；
- reserve_and_open 的预扣与开账同事务，消除原两步之间的崩溃丢账窗口。
路径真源红线：DATA_DIR / DB_PATH / BACKUP_DIR / SECRET_KEY_PATH 仍是 auth_system.db
的**模块全局**（用与 core.db 相同的表达式，但**不是**它的别名）。测试只 patch
auth_system.db.*，若改成 core.db 的别名会静默把测试写到生产库。已核 data/site.db
读写前后 sha256 完全一致。
顺带清掉 get_conn：迁移后它只剩「转调 connect()」一层空壳、全仓零消费者（测试
直接开裸连接、备份用 db.connect()），保留等于留一条「拿裸连接自己写」的口子。
测试：19 个文件同步迁移——10 个文件的 web.deps.secret_key 替身改 async；
test_db_conn.py 的多线程首连用例按异步语义拆成 2 条（单 loop gather(16) 覆盖批量
并发首连、WAL 就绪后 8 线程各自 loop 建连覆盖跨 loop）；test_billing_scope.py 去掉
多余的 @pytest.mark.anyio；被 await 且**正常返回**的 monkeypatch 替身一律改 async
（仅 raise 的同步替身仍可用）。
门禁：ruff 全过 / pyright 0-0-0（standard 模式含 tests 与顶层模块）/ pytest
384 项全绿（0 失败 0 error 0 skipped；383→384 因 test_db_conn 拆用例）。
data/site.db 读写前后 sha256 一致（a19f5db…6487），无 -wal/-shm 边车残留。
下一步 B3：删 core/sync_http.py、core/async_bridge.py、core/http_errors.py 及
tests/test_async_bridge.py——三者在生产侧已零消费者（http_errors 仅 sync_http
自用，requests 已非声明依赖）。
405 lines
17 KiB
Python
405 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""路径常量、连接与建表（原生异步 SQLite 访问层：aiosqlite + WAL）。
|
||
|
||
（自 auth_system.py 拆出；本文件由同步 sqlite3 栈迁移为原生异步，设计照
|
||
`core/db.py`——同一套「读写同时进行」口径：）
|
||
|
||
- WAL 模式：读事务与写事务互不阻塞（读快照隔离）；
|
||
- 读路径（`_q`）：每次操作独立连接、无锁、可并发；
|
||
- 写路径（`_x` / `write_transaction`）：SQLite 全库同一时刻只允许一个写者，
|
||
用 `asyncio.Lock` 串行化写事务，但写者不阻塞读者；
|
||
- `busy_timeout`：多进程/多连接下遇短暂写锁自旋等待而非立刻 SQLITE_BUSY。
|
||
|
||
路径常量仍是本模块的**唯一真源**：DB 访问函数在**调用时**读取本模块的
|
||
`DB_PATH` 等全局（测试 monkeypatch `auth_system.db.DB_PATH` 依旧生效）。
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import secrets
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from collections.abc import AsyncGenerator, Sequence
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
|
||
import aiosqlite
|
||
|
||
from .security import (
|
||
_hash_new,
|
||
_normalize_email,
|
||
)
|
||
|
||
# ============ 路径与常量 ============
|
||
# 必须与 core.db.DATA_DIR 用**同一条**表达式：两处指向同一个 data/site.db。
|
||
# 此前这里写死 Path('data') 而 core 侧认 SITE_DATA_DIR——一旦运维设置该变量，
|
||
# 站点库与设置/公告库会落到两个不同文件（脑裂），且不会报错。
|
||
DATA_DIR = Path(os.environ.get('SITE_DATA_DIR') or 'data')
|
||
|
||
DB_PATH = DATA_DIR / 'site.db'
|
||
|
||
BACKUP_DIR = DATA_DIR / 'backups'
|
||
|
||
SECRET_KEY_PATH = DATA_DIR / 'secret.key'
|
||
|
||
|
||
DEFAULT_ADMIN_USER = 'admin'
|
||
|
||
SESSION_KEY = 'site_uid'
|
||
|
||
DEFAULT_REGISTER_CREDITS = 3 # 注册直接赠送 3 次下载次数（无需领取；TRIAL_CREDITS 死常量已删）
|
||
|
||
|
||
# 锁按事件循环隔离：asyncio.Lock 绑定创建它的 loop，跨 loop 复用会抛
|
||
# "bound to a different event loop"（应用主 loop + 测试各自的 loop 并存，必须分开持有）。
|
||
_LOOP_LOCKS: dict[tuple[int, str], asyncio.Lock] = {}
|
||
|
||
|
||
def _loop_lock(kind: str) -> asyncio.Lock:
|
||
key = (id(asyncio.get_running_loop()), kind)
|
||
lock = _LOOP_LOCKS.get(key)
|
||
if lock is None:
|
||
lock = asyncio.Lock()
|
||
_LOOP_LOCKS[key] = lock
|
||
return lock
|
||
|
||
|
||
def _write_lock() -> asyncio.Lock:
|
||
"""写事务串行锁（每事件循环一个）。"""
|
||
return _loop_lock('write')
|
||
|
||
|
||
def _wal_lock() -> asyncio.Lock:
|
||
"""WAL 切换专用锁（不能用 _write_lock：execute() 已持它再调
|
||
connect→_configure 会自锁死）。"""
|
||
return _loop_lock('wal')
|
||
|
||
|
||
# journal_mode=WAL 持久化在库文件头里，按路径设一次即可。每连接重复执行
|
||
# 实测多付 ~1ms（新连接首条语句要建 WAL 索引读锁），且并发首连会争排他锁
|
||
# 报 database is locked（core/db.py 同款教训）。专用锁 + 二次复查防护。
|
||
_WAL_READY_FOR: str | None = None
|
||
|
||
|
||
# ============ 数据库 ============
|
||
|
||
|
||
async def _configure(conn: aiosqlite.Connection, force_wal: bool = False) -> None:
|
||
global _WAL_READY_FOR
|
||
conn.row_factory = aiosqlite.Row
|
||
# busy_timeout 先设：让后续语句遇锁自旋等待，而不是立刻报 database is locked
|
||
await conn.execute('PRAGMA busy_timeout=15000')
|
||
key = str(DB_PATH) # 调用时读取模块全局——测试 monkeypatch 后仍生效
|
||
if force_wal or _WAL_READY_FOR != key:
|
||
# 切换日志模式需要排他锁，且 `_WAL_READY_FOR` 的检查不是原子的——
|
||
# 并发连接须在专用锁内串行完成「切换 + 标记」，并二次复查，否则批量并发
|
||
# 连接会同时执行 PRAGMA journal_mode=WAL 争锁（实测 database is locked）
|
||
async with _wal_lock():
|
||
if force_wal or _WAL_READY_FOR != key:
|
||
await conn.execute('PRAGMA journal_mode=WAL')
|
||
_WAL_READY_FOR = key
|
||
await conn.execute('PRAGMA synchronous=NORMAL') # WAL 下兼顾安全与吞吐
|
||
await conn.execute('PRAGMA foreign_keys=ON')
|
||
|
||
|
||
async def connect() -> aiosqlite.Connection:
|
||
"""新建一个已配置好的连接（调用方负责 close）。"""
|
||
# exists() 必须在 aiosqlite.connect 之前判断：connect 会创建空文件，之后
|
||
# exists 恒为 True，库文件被删重建的场景就检测不到了（journal_mode 不持久
|
||
# 于新文件，必须重设）。
|
||
force_wal = not DB_PATH.exists()
|
||
# mkdir 移出 happy path（实测 190us/次）：connect 直接失败（目录缺失）才补建
|
||
# 重试。库文件被删的场景同样由此兜底重建。用 DB_PATH.parent 而非 DATA_DIR：
|
||
# 生产两者相同，测试 monkeypatch 只换 DB_PATH 时兜底仍能命中正确目录
|
||
# （core/db.py 用 DATA_DIR 的前提是两处都换）。
|
||
try:
|
||
conn = await aiosqlite.connect(str(DB_PATH), timeout=15)
|
||
except sqlite3.OperationalError:
|
||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
conn = await aiosqlite.connect(str(DB_PATH), timeout=15)
|
||
await _configure(conn, force_wal)
|
||
return conn
|
||
|
||
|
||
# 注：同步时代的公开连接门面 get_conn() 已删除——它在迁移后只剩「转调 connect()」
|
||
# 一层空壳且全仓零消费者（测试直接开裸连接，备份用 db.connect()）。保留它等于留一条
|
||
# 「拿裸连接自己写」的口子，与本次「拆掉同步门面、写路径统一走 _x/write_transaction」
|
||
# 的目标相悖。
|
||
|
||
|
||
# ============ 读路径（并发，无锁） ============
|
||
|
||
|
||
async def _q(sql: str, params: Sequence = ()) -> list[aiosqlite.Row]:
|
||
"""单语句读：**不持锁**。
|
||
|
||
WAL 模式下每个连接读到一致性快照，读与写互不阻塞；旧实现对读也加锁，
|
||
等于把全部读串行化并被写挡住。此处仅放开单语句读——写事务（_x 与各
|
||
`write_transaction` 块）仍全程串行，故不引入写-写竞争。
|
||
"""
|
||
conn = await connect()
|
||
try:
|
||
cur = await conn.execute(sql, params)
|
||
return list(await cur.fetchall())
|
||
finally:
|
||
await conn.close()
|
||
|
||
|
||
# ============ 写路径（串行，不阻塞读） ============
|
||
|
||
|
||
async def _x(sql: str, params: Sequence = ()) -> int:
|
||
"""执行单条写语句，返回 lastrowid（INSERT）或 rowcount（UPDATE/DELETE）"""
|
||
async with _write_lock():
|
||
conn = await connect()
|
||
try:
|
||
cur = await conn.execute(sql, params)
|
||
await conn.commit()
|
||
return cur.lastrowid if cur.lastrowid else cur.rowcount
|
||
finally:
|
||
await conn.close()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def write_transaction() -> AsyncGenerator[aiosqlite.Connection, None]:
|
||
"""写事务上下文：多条写语句原子提交（串行锁内）。
|
||
|
||
用法：
|
||
async with write_transaction() as conn:
|
||
await conn.execute(...)
|
||
await conn.execute(...)
|
||
# 退出即 commit；异常则 rollback
|
||
"""
|
||
async with _write_lock():
|
||
conn = await connect()
|
||
try:
|
||
yield conn
|
||
await conn.commit()
|
||
except Exception:
|
||
await conn.rollback()
|
||
raise
|
||
finally:
|
||
await conn.close()
|
||
|
||
|
||
_SCHEMA_SQL = '''
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
username TEXT UNIQUE NOT NULL,
|
||
password_hash TEXT NOT NULL,
|
||
salt TEXT NOT NULL,
|
||
email TEXT DEFAULT '',
|
||
role TEXT DEFAULT 'user',
|
||
duration_seconds INTEGER DEFAULT 0,
|
||
trial_device TEXT DEFAULT '',
|
||
register_ip TEXT DEFAULT '',
|
||
register_ua TEXT DEFAULT '',
|
||
netease_cookie TEXT DEFAULT '',
|
||
ann_seen TEXT DEFAULT '',
|
||
last_active INTEGER DEFAULT 0,
|
||
created_at INTEGER DEFAULT 0,
|
||
last_login INTEGER DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS email_codes (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
email TEXT NOT NULL,
|
||
code TEXT NOT NULL,
|
||
expires_at INTEGER NOT NULL,
|
||
created_at INTEGER DEFAULT 0,
|
||
purpose TEXT DEFAULT 'register'
|
||
);
|
||
CREATE TABLE IF NOT EXISTS cards (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
code TEXT UNIQUE NOT NULL,
|
||
minutes INTEGER NOT NULL,
|
||
created_by TEXT DEFAULT '',
|
||
created_at INTEGER DEFAULT 0,
|
||
used_by TEXT DEFAULT '',
|
||
used_at INTEGER DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
type TEXT DEFAULT 'success',
|
||
username TEXT DEFAULT '',
|
||
ip TEXT DEFAULT '',
|
||
action TEXT DEFAULT '',
|
||
detail TEXT DEFAULT '',
|
||
time TEXT DEFAULT ''
|
||
);
|
||
'''
|
||
|
||
|
||
async def _create_schema(conn: aiosqlite.Connection) -> None:
|
||
"""建表 + 单行配置种子行（幂等：全部 IF NOT EXISTS / INSERT OR IGNORE）。"""
|
||
await conn.executescript(_SCHEMA_SQL)
|
||
# 单行配置表
|
||
await conn.execute('''CREATE TABLE IF NOT EXISTS smtp_config (
|
||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||
host TEXT DEFAULT '', port INTEGER DEFAULT 587,
|
||
sender TEXT DEFAULT '', password TEXT DEFAULT '',
|
||
updated_at INTEGER DEFAULT 0
|
||
)''')
|
||
await conn.execute('INSERT OR IGNORE INTO smtp_config (id, port) VALUES (1, 587)')
|
||
# 键值设置表
|
||
await conn.execute('''CREATE TABLE IF NOT EXISTS settings (
|
||
key TEXT PRIMARY KEY, value TEXT DEFAULT ''
|
||
)''')
|
||
await conn.execute('''CREATE TABLE IF NOT EXISTS announcements (
|
||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||
title TEXT DEFAULT '', content TEXT DEFAULT '',
|
||
popup_mode TEXT DEFAULT 'off',
|
||
updated_at INTEGER DEFAULT 0
|
||
)''')
|
||
await conn.execute('INSERT OR IGNORE INTO announcements (id) VALUES (1)')
|
||
# 计费账本（任务实扣/交付/退款/终态的持久化唯一记录，结算幂等）
|
||
await conn.execute('''CREATE TABLE IF NOT EXISTS billing_ledger (
|
||
task_id TEXT PRIMARY KEY,
|
||
user_id INTEGER NOT NULL,
|
||
reserved INTEGER DEFAULT 0,
|
||
delivered INTEGER DEFAULT 0,
|
||
refunded INTEGER DEFAULT 0,
|
||
status TEXT DEFAULT 'open',
|
||
created_at INTEGER DEFAULT 0,
|
||
settled_at INTEGER DEFAULT 0
|
||
)''')
|
||
|
||
|
||
async def _apply_column_migrations(conn: aiosqlite.Connection) -> None:
|
||
"""轻量迁移：新增列。仅"列已存在"可忽略（其余迁移失败必须暴露，不能吞掉）"""
|
||
_already: set[str] = set()
|
||
trows = await (await conn.execute("SELECT name FROM sqlite_master WHERE type='table'")).fetchall()
|
||
for trow in trows:
|
||
crows = await (await conn.execute(f"PRAGMA table_info({trow['name']})")).fetchall()
|
||
for crow in crows:
|
||
_already.add(f"{trow['name']}.{crow['name']}")
|
||
for alter_sql, tbl_col in (
|
||
('ALTER TABLE users ADD COLUMN status INTEGER DEFAULT 1', 'users.status'),
|
||
('ALTER TABLE users ADD COLUMN login_seq INTEGER DEFAULT 0', 'users.login_seq'),
|
||
('ALTER TABLE users ADD COLUMN recent_provinces TEXT DEFAULT \'[]\'', 'users.recent_provinces'),
|
||
('ALTER TABLE users ADD COLUMN ban_reason TEXT DEFAULT \'\'', 'users.ban_reason'),
|
||
('ALTER TABLE cards ADD COLUMN note TEXT DEFAULT \'\'', 'cards.note'),
|
||
('ALTER TABLE cards ADD COLUMN expire_at INTEGER DEFAULT 0', 'cards.expire_at'),
|
||
('ALTER TABLE cards ADD COLUMN disabled INTEGER DEFAULT 0', 'cards.disabled'),
|
||
('ALTER TABLE email_codes ADD COLUMN purpose TEXT DEFAULT \'register\'', 'email_codes.purpose'),
|
||
):
|
||
if tbl_col in _already:
|
||
continue # 幂等：列已存在
|
||
try:
|
||
await conn.execute(alter_sql)
|
||
except sqlite3.OperationalError as e:
|
||
if 'duplicate column' not in str(e).lower():
|
||
# 磁盘/锁/结构等真实故障：抛出，禁止静默带病初始化
|
||
raise
|
||
|
||
|
||
async def _ensure_email_unique_index(conn: aiosqlite.Connection) -> None:
|
||
"""非空邮箱唯一（部分索引，不约束历史空串）。
|
||
|
||
存量库若已有重复非空邮箱，索引创建失败仅告警不阻断启动
|
||
（查重逻辑在注册/建号路由另有兜底）。
|
||
"""
|
||
try:
|
||
await conn.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email_nonempty '
|
||
'ON users(email) WHERE email != \'\'')
|
||
except sqlite3.IntegrityError:
|
||
print('[警告] users.email 存在重复的非空邮箱，唯一索引未创建（请人工清理后重启）')
|
||
|
||
|
||
async def _migrate_user_emails(conn: aiosqlite.Connection) -> None:
|
||
"""users.email 口径统一为归一化（lower + 剥 Gmail +tag）——与
|
||
get_user_by_email/找回/邮箱登录的查询口径对齐。逐行迁移，归一化后与他
|
||
人撞唯一索引的行保留原值跳过（与"存量重复仅告警"同口径）。幂等：已
|
||
归一化的行零写入。
|
||
"""
|
||
urows = await (await conn.execute("SELECT id, email FROM users WHERE email != ''")).fetchall()
|
||
for urow in urows:
|
||
fixed = _normalize_email(urow['email'])
|
||
if not fixed or fixed == urow['email']:
|
||
continue
|
||
try:
|
||
await conn.execute('UPDATE users SET email = ? WHERE id = ?', (fixed, urow['id']))
|
||
except sqlite3.IntegrityError:
|
||
print(f"[警告] 用户 {urow['id']} 的邮箱归一化后与他人重复，保留原值")
|
||
|
||
|
||
async def _ensure_initial_admin(conn: aiosqlite.Connection) -> None:
|
||
"""初始管理员（新装不再使用公开默认口令，随机生成并写入日志一次）。
|
||
|
||
引导条件改为"不存在任何管理员角色"——按用户名查会在
|
||
管理员改名（如 admin→qfbsh）后误判缺失，重启即重建出一个新的 admin 账号。
|
||
"""
|
||
row = await (await conn.execute("SELECT id FROM users WHERE role = 'admin' LIMIT 1")).fetchone()
|
||
if row:
|
||
return
|
||
initial_pass = secrets.token_urlsafe(12)
|
||
await conn.execute(
|
||
'INSERT INTO users (username, password_hash, salt, role, duration_seconds, created_at) '
|
||
'VALUES (?, ?, ?, ?, -1, ?)',
|
||
(DEFAULT_ADMIN_USER, _hash_new(initial_pass), '',
|
||
'admin', int(time.time())))
|
||
# ——非 ASCII 输出须防 GBK 重定向下的 UnicodeEncodeError
|
||
try:
|
||
print(f'[初始化] 已创建管理员账号 admin，初始密码：{initial_pass}（请立即登录并修改，本提示不再显示）')
|
||
except UnicodeEncodeError:
|
||
print('[init] admin account created, initial password printed above failed to encode')
|
||
|
||
|
||
async def _init_db_locked(conn: aiosqlite.Connection) -> None:
|
||
"""在**已持有写锁**的连接上执行全部结构演进（不自行取锁）。
|
||
|
||
restore_backup 需要在自己的写锁临界区里对新库跑一遍初始化，不能调用会
|
||
再次取写锁的 init_db（asyncio.Lock 非重入，会自锁死），故拆出本函数。
|
||
"""
|
||
await _create_schema(conn)
|
||
await _apply_column_migrations(conn)
|
||
await _ensure_email_unique_index(conn)
|
||
await _migrate_user_emails(conn)
|
||
await _ensure_initial_admin(conn)
|
||
|
||
|
||
async def init_db() -> None:
|
||
"""初始化表结构与默认数据"""
|
||
DATA_DIR.mkdir(parents=True, exist_ok=True) # noqa: ASYNC240 - 启动期一次的目录创建，无需线程池
|
||
async with write_transaction() as conn:
|
||
await _init_db_locked(conn)
|
||
|
||
|
||
def _get_secret_key_sync() -> str:
|
||
"""同步实现（见 get_secret_key 的异步包装）：原子写入 + 失败显式告警"""
|
||
key = ''
|
||
try:
|
||
if SECRET_KEY_PATH.exists():
|
||
key = SECRET_KEY_PATH.read_text(encoding='utf-8').strip()
|
||
if key:
|
||
return key
|
||
except OSError:
|
||
pass
|
||
new_key = secrets.token_hex(32)
|
||
try:
|
||
SECRET_KEY_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
# 先写临时文件再原子替换，避免部分写入留下损坏密钥
|
||
tmp = SECRET_KEY_PATH.with_suffix('.key.tmp')
|
||
tmp.write_text(new_key, encoding='utf-8')
|
||
os.replace(str(tmp), str(SECRET_KEY_PATH))
|
||
# 并发初始化者以"替换后文件里的值"为准（最后写入者胜出但全员一致）
|
||
final = SECRET_KEY_PATH.read_text(encoding='utf-8').strip()
|
||
return final or new_key
|
||
except OSError as e:
|
||
# 写入失败不能装作成功：显著告警，返回临时值（本进程会话可用，重启后失效）
|
||
try:
|
||
print(f'[警告] 会话密钥持久化失败（{e}），本进程将使用临时密钥，重启后所有会话失效！', file=sys.stderr)
|
||
except UnicodeEncodeError:
|
||
print('[warn] secret key persistence failed; using temporary key for this process', file=sys.stderr)
|
||
return new_key
|
||
|
||
|
||
async def get_secret_key() -> str:
|
||
"""获取（或生成）持久化的会话签名密钥（原子写入 + 失败显式告警）。
|
||
|
||
内部是**同步阻塞文件 IO**（exists/read/write/os.replace），整段经
|
||
`asyncio.to_thread` 卸载到线程——这是正当的阻塞 IO 卸载（登录链路每次
|
||
签会话都要读一次密钥），不是把异步包装成同步的门面。
|
||
"""
|
||
return await asyncio.to_thread(_get_secret_key_sync)
|