gouki
d711ad5827
credentials 表为登录凭据唯一事实源（站点×登录方式，含 oauth/2FA）；账号密码读写重定向凭据层并幂等迁移历史数据；TOTP 按 RFC6238 零依赖自实现，绑定需当前动态码校验；Key Escrow 防 MASTER_KEY 遗失；前端新增凭据库视图与账号 2FA 联动。64 pytest + 16 浏览器端到端验证通过。
117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
"""TOTP 动态验证码（RFC 6238，零依赖自实现）
|
||
|
||
用于凭据库 2FA：存储 base32 secret，按 30s 步长生成 6 位动态码。
|
||
不引第三方库（pyotp）的原因：零 SSH 部署链路不重装依赖，避免 update.sh
|
||
缺包导致服务自毁；正确性用 RFC 6238 附录 B 官方向量在 pytest 锚定
|
||
（tests/test_totp.py）。
|
||
"""
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import secrets
|
||
import struct
|
||
import time
|
||
from urllib.parse import parse_qs, quote, unquote, urlparse
|
||
|
||
PERIOD = 30 # 步长（秒），标准值
|
||
DIGITS = 6 # 码位数，标准值
|
||
|
||
|
||
def b32decode(secret: str) -> bytes:
|
||
"""Base32 解码（容错：去空白、大写化、自动补 = 填充）"""
|
||
s = "".join(secret.split()).upper()
|
||
return base64.b32decode(s + "=" * ((-len(s)) % 8))
|
||
|
||
|
||
def random_secret() -> str:
|
||
"""生成 20 字节随机 base32 secret（供「生成随机密钥」入口）"""
|
||
return base64.b32encode(secrets.token_bytes(20)).decode().rstrip("=")
|
||
|
||
|
||
def _hotp(key: bytes, counter: int, digits: int = DIGITS) -> str:
|
||
"""RFC 4226 HOTP：HMAC-SHA1(key, counter) 动态截断 → digits 位十进制码"""
|
||
digest = hmac.new(key, struct.pack(">Q", counter), hashlib.sha1).digest()
|
||
offset = digest[-1] & 0x0F
|
||
code = struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7FFFFFFF
|
||
return str(code % (10 ** digits)).zfill(digits)
|
||
|
||
|
||
def totp_at(
|
||
secret_b32: str, ts: float | None = None, period: int = PERIOD, digits: int = DIGITS
|
||
) -> tuple[str, int]:
|
||
"""计算指定时刻的 TOTP 码
|
||
|
||
返回 (code, expires_in)：expires_in 为当前码剩余有效秒数（前端倒计时用）。
|
||
ts 为 None 时取当前时间。
|
||
"""
|
||
now = time.time() if ts is None else ts
|
||
counter = int(now // period)
|
||
left = period - int(now % period)
|
||
return _hotp(b32decode(secret_b32), counter, digits), left
|
||
|
||
|
||
def verify(
|
||
secret_b32: str,
|
||
code: str,
|
||
window: int = 1,
|
||
ts: float | None = None,
|
||
period: int = PERIOD,
|
||
digits: int = DIGITS,
|
||
) -> bool:
|
||
"""校验用户输入的动态码（±window 个步进，容忍时钟偏差）
|
||
|
||
常量时间比较，防时序旁路。录入绑定时用 window=1 即可。
|
||
"""
|
||
code = "".join(code.split())
|
||
if not code.isdigit() or len(code) != digits:
|
||
return False
|
||
now = time.time() if ts is None else ts
|
||
base_counter = int(now // period)
|
||
key = b32decode(secret_b32)
|
||
return any(
|
||
hmac.compare_digest(_hotp(key, base_counter + off, digits), code)
|
||
for off in range(-window, window + 1)
|
||
)
|
||
|
||
|
||
def parse_otpauth_uri(uri: str) -> dict:
|
||
"""解析 otpauth://totp/Label?secret=...&issuer=...&period=30&digits=6
|
||
|
||
返回 {secret, issuer, account}；Label 形如 "Issuer:account" 或 "account"，
|
||
无 issuer 参数时从 Label 前缀提取。非 TOTP、缺 secret 或非标准参数抛 ValueError
|
||
（v1 仅支持 6 位/30s 标准参数，避免录入成功但生成码对不上）。
|
||
"""
|
||
parsed = urlparse(uri.strip())
|
||
if parsed.scheme != "otpauth":
|
||
raise ValueError("不是 otpauth:// 链接")
|
||
if parsed.netloc.lower() != "totp":
|
||
raise ValueError(f"仅支持 TOTP（当前类型：{parsed.netloc}）")
|
||
params = parse_qs(parsed.query)
|
||
secret = (params.get("secret") or [""])[0]
|
||
if not secret:
|
||
raise ValueError("链接缺少 secret 参数")
|
||
period = int((params.get("period") or [PERIOD])[0])
|
||
digits = int((params.get("digits") or [DIGITS])[0])
|
||
if period != PERIOD or digits != DIGITS:
|
||
raise ValueError(f"暂仅支持 {DIGITS} 位/{PERIOD}s 标准参数（当前 {digits} 位/{period}s）")
|
||
|
||
label = unquote(parsed.path.lstrip("/"))
|
||
issuer = (params.get("issuer") or [""])[0]
|
||
account = label
|
||
if ":" in label:
|
||
prefix, _, rest = label.partition(":")
|
||
account = rest or prefix
|
||
if not issuer:
|
||
issuer = prefix
|
||
return {"secret": secret, "issuer": issuer or None, "account": account or None}
|
||
|
||
|
||
def build_otpauth_uri(secret_b32: str, issuer: str | None, account: str | None) -> str:
|
||
"""重建 otpauth URI（导出/换机重新绑定用）"""
|
||
label = f"{issuer}:{account}" if issuer and account else (account or issuer or "")
|
||
q = f"secret={secret_b32}"
|
||
if issuer:
|
||
q += f"&issuer={quote(issuer)}"
|
||
return f"otpauth://totp/{quote(label)}?{q}"
|