feat(vault): 凭据库（密码+2FA 动态码）与 MASTER_KEY 密钥托管
credentials 表为登录凭据唯一事实源（站点×登录方式，含 oauth/2FA）；账号密码读写重定向凭据层并幂等迁移历史数据；TOTP 按 RFC6238 零依赖自实现，绑定需当前动态码校验；Key Escrow 防 MASTER_KEY 遗失；前端新增凭据库视图与账号 2FA 联动。64 pytest + 16 浏览器端到端验证通过。
This commit is contained in:
gouki
@@ -0,0 +1,162 @@
|
||||
"""凭据中心化迁移测试（accounts.login_password → credentials）
|
||||
|
||||
模拟旧库（accounts 无 credential_id 列）→ 跑迁移 → 断言：
|
||||
密文原样搬入、site 取平台显示名、账号关联回填、原字段置空、重跑幂等、
|
||||
无密码账号不产生空条目。
|
||||
|
||||
运行：.venv/bin/pytest tests/test_credential_migration.py -v
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
import sqlalchemy as sa
|
||||
from cryptography.fernet import Fernet
|
||||
from sqlmodel import Session, SQLModel, create_engine, select
|
||||
|
||||
import app.database as db
|
||||
from app.core import crypto
|
||||
from app.core.config import settings
|
||||
from app.models.asset import AIAccount, Account, Asset, CloudflareDetail, DomainDetail, VPSDetail
|
||||
from app.models.credential import Credential, LoginType
|
||||
from app.models.provider import Provider
|
||||
from app.models.ssl import SiteCert, Subdomain
|
||||
|
||||
# 旧结构 accounts 表：无 credential_id 列（迁移应自动补列）
|
||||
OLD_ACCOUNTS_DDL = """
|
||||
CREATE TABLE accounts (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name VARCHAR NOT NULL,
|
||||
platform VARCHAR,
|
||||
remark VARCHAR,
|
||||
login_user VARCHAR,
|
||||
login_password_encrypted VARCHAR,
|
||||
api_config_encrypted VARCHAR,
|
||||
last_synced_at DATETIME,
|
||||
created_at DATETIME NOT NULL
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def legacy_engine(tmp_path, monkeypatch):
|
||||
"""旧结构测试库：新表齐备，但 accounts 缺 credential_id 列"""
|
||||
monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
|
||||
crypto._get_fernet.cache_clear() # 避免复用其它测试缓存的 Fernet 实例
|
||||
engine = create_engine(
|
||||
f"sqlite:///{tmp_path / 'test.db'}", connect_args={"check_same_thread": False}
|
||||
)
|
||||
# 除 Account 外按当前模型建表（含新 credentials 表；accounts 用旧 DDL）
|
||||
SQLModel.metadata.create_all(
|
||||
engine,
|
||||
tables=[
|
||||
Provider.__table__, Asset.__table__, VPSDetail.__table__, DomainDetail.__table__,
|
||||
AIAccount.__table__, CloudflareDetail.__table__, Subdomain.__table__,
|
||||
SiteCert.__table__, Credential.__table__,
|
||||
],
|
||||
)
|
||||
with engine.begin() as conn:
|
||||
conn.execute(sa.text(OLD_ACCOUNTS_DDL))
|
||||
# 迁移函数引用模块级 assets_engine → 指向测试库
|
||||
monkeypatch.setattr(db, "assets_engine", engine)
|
||||
yield engine
|
||||
crypto._get_fernet.cache_clear()
|
||||
|
||||
|
||||
def _seed(engine, pwd_aliyun: str, pwd_forum: str) -> None:
|
||||
"""造迁移前现场：平台账号（有密码）/ 无平台账号（有密码）/ 无密码账号"""
|
||||
with Session(engine) as s:
|
||||
s.add(Provider(slug="aliyun", name="阿里云", category="vps"))
|
||||
s.commit()
|
||||
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO accounts (name, platform, login_user, login_password_encrypted, created_at) "
|
||||
"VALUES ('me@gmail.com', 'aliyun', 'me@gmail.com', :pwd, :now)"
|
||||
),
|
||||
{"pwd": pwd_aliyun, "now": now},
|
||||
)
|
||||
# platform 为空串：site 应回退 '未分类'；无 login_user：username 回退 name
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO accounts (name, platform, login_user, login_password_encrypted, created_at) "
|
||||
"VALUES ('forum-user', '', NULL, :pwd, :now)"
|
||||
),
|
||||
{"pwd": pwd_forum, "now": now},
|
||||
)
|
||||
# 无密码账号：不应产生凭据条目
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO accounts (name, platform, created_at) VALUES ('no-pwd', 'vultr', :now)"
|
||||
),
|
||||
{"now": now},
|
||||
)
|
||||
|
||||
|
||||
def test_migration_moves_passwords_to_vault(legacy_engine):
|
||||
engine = legacy_engine
|
||||
pwd_aliyun = crypto.encrypt("p@ss-阿里云")
|
||||
pwd_forum = crypto.encrypt("forum-pass-中文")
|
||||
_seed(engine, pwd_aliyun, pwd_forum)
|
||||
|
||||
db._migrate_assets_db()
|
||||
|
||||
with Session(engine) as s:
|
||||
creds = s.exec(select(Credential).order_by(Credential.id)).all()
|
||||
# 仅两条有密码的账号被迁移，无密码账号不产生空条目
|
||||
assert len(creds) == 2
|
||||
|
||||
c1, c2 = creds
|
||||
# 密文原样搬入（不重新加密，避免中间态明文暴露），仍可解密还原
|
||||
assert c1.password_encrypted == pwd_aliyun
|
||||
assert crypto.decrypt(c1.password_encrypted) == "p@ss-阿里云"
|
||||
# site 取平台显示名（join providers），username 取 login_user
|
||||
assert c1.site == "阿里云"
|
||||
assert c1.username == "me@gmail.com"
|
||||
assert c1.login_type == LoginType.PASSWORD
|
||||
# 无 provider 匹配 + platform 空 → site 回退 '未分类'；username 回退 name
|
||||
assert c2.password_encrypted == pwd_forum
|
||||
assert c2.site == "未分类"
|
||||
assert c2.username == "forum-user"
|
||||
|
||||
# 账号侧：credential_id 回填、原密码字段置空（唯一事实源，防双份漂移）
|
||||
accs = {a.name: a for a in s.exec(select(Account)).all()}
|
||||
assert accs["me@gmail.com"].credential_id == c1.id
|
||||
assert accs["me@gmail.com"].login_password_encrypted is None
|
||||
assert accs["forum-user"].credential_id == c2.id
|
||||
assert accs["forum-user"].login_password_encrypted is None
|
||||
assert accs["no-pwd"].credential_id is None
|
||||
|
||||
|
||||
def test_migration_is_idempotent(legacy_engine):
|
||||
"""重跑迁移不得重复建条目（幂等条件：credential_id 已非空则跳过）"""
|
||||
engine = legacy_engine
|
||||
_seed(engine, crypto.encrypt("pw1"), crypto.encrypt("pw2"))
|
||||
|
||||
db._migrate_assets_db()
|
||||
db._migrate_assets_db()
|
||||
db._migrate_assets_db()
|
||||
|
||||
with Session(engine) as s:
|
||||
assert len(s.exec(select(Credential)).all()) == 2
|
||||
|
||||
|
||||
def test_migration_adds_credential_id_column(legacy_engine):
|
||||
"""旧库缺 credential_id 列时迁移应自动补列"""
|
||||
engine = legacy_engine
|
||||
with engine.connect() as conn:
|
||||
cols = {c["name"] for c in sa.inspect(conn).get_columns("accounts")}
|
||||
assert "credential_id" not in cols # 前置：确实是旧结构
|
||||
|
||||
db._migrate_assets_db()
|
||||
|
||||
with engine.connect() as conn:
|
||||
cols = {c["name"] for c in sa.inspect(conn).get_columns("accounts")}
|
||||
indexes = {ix["name"] for ix in sa.inspect(conn).get_indexes("accounts")}
|
||||
assert "credential_id" in cols
|
||||
|
||||
db._migrate_indexes()
|
||||
with engine.connect() as conn:
|
||||
indexes = {ix["name"] for ix in sa.inspect(conn).get_indexes("accounts")}
|
||||
assert "ix_accounts_credential_id" in indexes
|
||||
@@ -0,0 +1,351 @@
|
||||
"""凭据服务（密码库 + 2FA）与账号联动测试
|
||||
|
||||
覆盖：创建/更新/删除语义、密码加解密、oauth 条目、2FA 绑定强制校验
|
||||
（错码拒绝）、动态码生成、搜索过滤、重复标记，以及账号侧密码重定向
|
||||
（唯一事实源）与生命周期解耦（删账号不删凭据）。
|
||||
|
||||
运行：.venv/bin/pytest tests/test_credential_service.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from cryptography.fernet import Fernet
|
||||
from fastapi import HTTPException
|
||||
from sqlmodel import Session, SQLModel, create_engine, select
|
||||
|
||||
from app.core import crypto, totp
|
||||
from app.core.config import settings
|
||||
from app.models.asset import Account
|
||||
from app.models.credential import Credential, LoginType
|
||||
from app.models.provider import Provider
|
||||
from app.schemas.account import AccountCreate, AccountUpdate
|
||||
from app.schemas.credential import CredentialCreate, CredentialUpdate, OtpBindRequest
|
||||
from app.services import account_service, credential_service
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def session(tmp_path, monkeypatch):
|
||||
"""独立测试库（新结构全表）+ 预置一个平台供 site 显示名映射"""
|
||||
monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
|
||||
crypto._get_fernet.cache_clear()
|
||||
engine = create_engine(
|
||||
f"sqlite:///{tmp_path / 'svc.db'}", connect_args={"check_same_thread": False}
|
||||
)
|
||||
SQLModel.metadata.create_all(engine)
|
||||
with Session(engine) as s:
|
||||
s.add(Provider(slug="aliyun", name="阿里云", category="vps"))
|
||||
s.commit()
|
||||
yield s
|
||||
crypto._get_fernet.cache_clear()
|
||||
|
||||
|
||||
def _otp_pair():
|
||||
"""生成一对可用的 (secret, 当前动态码)"""
|
||||
secret = totp.random_secret()
|
||||
return secret, totp.totp_at(secret)[0]
|
||||
|
||||
|
||||
# ----------------------------- 创建 / 读取 ----------------------------- #
|
||||
def test_create_with_password(session):
|
||||
read = credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", username="me@gmail.com", password="s3cret-中文")
|
||||
)
|
||||
assert read.has_password is True
|
||||
assert read.has_otp is False
|
||||
# Read 永不返回明文/密文
|
||||
dumped = read.model_dump()
|
||||
assert "password" not in dumped and "password_encrypted" not in dumped
|
||||
|
||||
revealed = credential_service.reveal_password(session, read.id)
|
||||
assert revealed["password"] == "s3cret-中文"
|
||||
assert revealed["username"] == "me@gmail.com"
|
||||
# 库里确实是密文
|
||||
row = session.get(Credential, read.id)
|
||||
assert row.password_encrypted != "s3cret-中文"
|
||||
assert crypto.decrypt(row.password_encrypted) == "s3cret-中文"
|
||||
|
||||
|
||||
def test_create_oauth_entry(session):
|
||||
"""授权登录条目：无密码也是正常记录，可检索可回忆"""
|
||||
read = credential_service.create_credential(
|
||||
session,
|
||||
CredentialCreate(
|
||||
site="某论坛", username="me@gmail.com",
|
||||
login_type=LoginType.OAUTH, oauth_provider="google",
|
||||
),
|
||||
)
|
||||
assert read.login_type == LoginType.OAUTH
|
||||
assert read.oauth_provider == "google"
|
||||
assert read.has_password is False
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.reveal_password(session, read.id)
|
||||
assert e.value.status_code == 404
|
||||
|
||||
|
||||
def test_create_requires_site(session):
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.create_credential(session, CredentialCreate(site=" "))
|
||||
assert e.value.status_code == 400
|
||||
|
||||
|
||||
# ----------------------------- 2FA 绑定校验 ----------------------------- #
|
||||
def test_create_with_otp_requires_code(session):
|
||||
"""绑定 2FA 必须同时提供当前动态码（防 secret 手误成废条目）"""
|
||||
secret, _ = _otp_pair()
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", otp_secret=secret)
|
||||
)
|
||||
assert e.value.status_code == 400 and "动态码" in e.value.detail
|
||||
|
||||
|
||||
def test_create_with_otp_wrong_code_rejected(session):
|
||||
secret, _ = _otp_pair()
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", otp_secret=secret, otp_code="000000")
|
||||
)
|
||||
assert e.value.status_code == 400 and "不匹配" in e.value.detail
|
||||
# 校验失败不得落库
|
||||
assert session.exec(select(Credential)).first() is None
|
||||
|
||||
|
||||
def test_create_with_otp_valid_code(session):
|
||||
secret, code = _otp_pair()
|
||||
read = credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", username="me@gmail.com", otp_secret=secret, otp_code=code)
|
||||
)
|
||||
assert read.has_otp is True
|
||||
otp = credential_service.current_otp(session, read.id)
|
||||
assert len(otp["code"]) == 6 and otp["code"].isdigit()
|
||||
assert 1 <= otp["expires_in"] <= 30
|
||||
# secret 加密存储，且与录入值等价（能算出同样的码）
|
||||
row = session.get(Credential, read.id)
|
||||
assert row.otp_secret_encrypted != secret
|
||||
assert totp.verify(crypto.decrypt(row.otp_secret_encrypted), otp["code"])
|
||||
|
||||
|
||||
def test_create_with_otpauth_uri(session):
|
||||
"""支持直接粘贴 otpauth:// 链接录入（自动提取 secret）"""
|
||||
secret, code = _otp_pair()
|
||||
uri = f"otpauth://totp/GitHub:me%40gmail.com?secret={secret}&issuer=GitHub"
|
||||
read = credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", otp_secret=uri, otp_code=code)
|
||||
)
|
||||
assert read.has_otp is True
|
||||
row = session.get(Credential, read.id)
|
||||
assert crypto.decrypt(row.otp_secret_encrypted) == secret # 只存 secret，不存整条 URI
|
||||
|
||||
|
||||
def test_create_with_invalid_secret_rejected(session):
|
||||
secret, code = _otp_pair()
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="x", otp_secret="otpauth://hotp/a?secret=AA", otp_code=code)
|
||||
)
|
||||
assert e.value.status_code == 400 and "secret 无效" in e.value.detail
|
||||
|
||||
|
||||
def test_current_otp_unbound_404(session):
|
||||
read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
|
||||
with pytest.raises(HTTPException) as e:
|
||||
credential_service.current_otp(session, read.id)
|
||||
assert e.value.status_code == 404
|
||||
|
||||
|
||||
def test_bind_and_unbind_otp(session):
|
||||
read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
|
||||
secret, code = _otp_pair()
|
||||
updated = credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret, code=code))
|
||||
assert updated.has_otp is True
|
||||
# 错码解绑不了也绑不上
|
||||
with pytest.raises(HTTPException):
|
||||
credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret, code="111111"))
|
||||
unbound = credential_service.unbind_otp(session, read.id)
|
||||
assert unbound.has_otp is False
|
||||
|
||||
|
||||
# ----------------------------- 更新 / 删除语义 ----------------------------- #
|
||||
def test_update_password_semantics(session):
|
||||
"""None=不改，''=清除，非空=重加密（与账号凭证惯例一致）"""
|
||||
read = credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", password="old")
|
||||
)
|
||||
# None：不修改
|
||||
r = credential_service.update_credential(session, read.id, CredentialUpdate(note="记一笔"))
|
||||
assert r.note == "记一笔"
|
||||
assert credential_service.reveal_password(session, read.id)["password"] == "old"
|
||||
# 非空：重加密
|
||||
credential_service.update_credential(session, read.id, CredentialUpdate(password="new-中文"))
|
||||
assert credential_service.reveal_password(session, read.id)["password"] == "new-中文"
|
||||
# 空串：清除
|
||||
r = credential_service.update_credential(session, read.id, CredentialUpdate(password=""))
|
||||
assert r.has_password is False
|
||||
with pytest.raises(HTTPException):
|
||||
credential_service.reveal_password(session, read.id)
|
||||
|
||||
|
||||
def test_update_fields(session):
|
||||
read = credential_service.create_credential(session, CredentialCreate(site="GitHub", username="a@b.c"))
|
||||
r = credential_service.update_credential(
|
||||
session, read.id,
|
||||
CredentialUpdate(site="GitLab", username="", login_type=LoginType.OTHER, url="https://x", note=None),
|
||||
)
|
||||
assert r.site == "GitLab"
|
||||
assert r.username is None # 空串归一化为 None
|
||||
assert r.login_type == LoginType.OTHER
|
||||
assert r.url == "https://x"
|
||||
|
||||
|
||||
def test_delete_unlinks_account(session):
|
||||
"""删除凭据：账号保留，仅解除关联（生命周期解耦）"""
|
||||
acc = account_service.create_account(
|
||||
session, AccountCreate(name="me@gmail.com", platform="aliyun", login_password="pw")
|
||||
)
|
||||
cred_id = acc.credential_id
|
||||
assert cred_id is not None
|
||||
|
||||
credential_service.delete_credential(session, cred_id)
|
||||
|
||||
refreshed = session.get(Account, acc.id)
|
||||
assert refreshed is not None # 账号还在
|
||||
assert refreshed.credential_id is None # 关联已解除
|
||||
assert session.get(Credential, cred_id) is None
|
||||
|
||||
|
||||
# ----------------------------- 列表 / 搜索 / 重复标记 ----------------------------- #
|
||||
def test_list_search_and_filters(session):
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", username="me@gmail.com", password="p1")
|
||||
)
|
||||
secret, code = _otp_pair()
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="阿里云", username="me@gmail.com", password="p2",
|
||||
otp_secret=secret, otp_code=code)
|
||||
)
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="某论坛", login_type=LoginType.OAUTH, oauth_provider="google")
|
||||
)
|
||||
|
||||
all_creds = credential_service.list_credentials(session)
|
||||
assert len(all_creds) == 3
|
||||
|
||||
# 搜索命中 site / username / note
|
||||
assert len(credential_service.list_credentials(session, q="git")) == 1
|
||||
assert len(credential_service.list_credentials(session, q="me@gmail.com")) == 2
|
||||
# 登录方式过滤
|
||||
assert len(credential_service.list_credentials(session, login_type="oauth")) == 1
|
||||
# 2FA 过滤
|
||||
assert len(credential_service.list_credentials(session, has_otp=True)) == 1
|
||||
assert len(credential_service.list_credentials(session, has_otp=False)) == 2
|
||||
|
||||
|
||||
def test_duplicate_flag(session):
|
||||
"""同 (site, username) 多条时全部标记 duplicate（录入提示用）"""
|
||||
credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", username="me@gmail.com")
|
||||
)
|
||||
second = credential_service.create_credential(
|
||||
session, CredentialCreate(site="github", username="ME@gmail.com") # 大小写/空白归一化
|
||||
)
|
||||
assert second.duplicate is True
|
||||
rows = credential_service.list_credentials(session)
|
||||
assert all(r.duplicate for r in rows)
|
||||
# 同站不同用户名不算重复
|
||||
third = credential_service.create_credential(
|
||||
session, CredentialCreate(site="GitHub", username="other@gmail.com")
|
||||
)
|
||||
assert third.duplicate is False
|
||||
|
||||
|
||||
# ----------------------------- 账号联动（唯一事实源） ----------------------------- #
|
||||
def test_account_create_redirects_password(session):
|
||||
"""账号录入密码 → 自动建凭据并关联；账号表不再存密码密文"""
|
||||
acc = account_service.create_account(
|
||||
session,
|
||||
AccountCreate(name="me@gmail.com", platform="aliyun",
|
||||
login_user="me@gmail.com", login_password="pw-中文"),
|
||||
)
|
||||
assert acc.credential_id is not None
|
||||
assert acc.has_login_password is True
|
||||
assert acc.has_otp is False
|
||||
|
||||
row = session.get(Account, acc.id)
|
||||
assert row.login_password_encrypted is None # 唯一事实源在凭据表
|
||||
cred = session.get(Credential, acc.credential_id)
|
||||
assert cred.site == "阿里云" # 平台 slug → 显示名
|
||||
assert cred.username == "me@gmail.com"
|
||||
assert crypto.decrypt(cred.password_encrypted) == "pw-中文"
|
||||
|
||||
# 账号查看密码接口仍可用（重定向到凭据）
|
||||
revealed = account_service.reveal_password(session, acc.id)
|
||||
assert revealed["password"] == "pw-中文"
|
||||
assert revealed["login_user"] == "me@gmail.com"
|
||||
|
||||
|
||||
def test_account_create_without_password_no_credential(session):
|
||||
"""无密码账号不产生空凭据条目"""
|
||||
acc = account_service.create_account(session, AccountCreate(name="x@y.z", platform="aliyun"))
|
||||
assert acc.credential_id is None
|
||||
assert session.exec(select(Credential)).first() is None
|
||||
|
||||
|
||||
def test_account_update_syncs_credential(session):
|
||||
"""改名/换平台/改密码都同步到关联凭据"""
|
||||
acc = account_service.create_account(
|
||||
session, AccountCreate(name="me@gmail.com", platform="aliyun", login_password="pw1")
|
||||
)
|
||||
# 改密码
|
||||
r = account_service.update_account(session, acc.id, AccountUpdate(login_password="pw2"))
|
||||
assert r.has_login_password is True
|
||||
assert account_service.reveal_password(session, acc.id)["password"] == "pw2"
|
||||
# 换平台 → site 同步（无匹配 provider 时回退 slug 原文）
|
||||
r = account_service.update_account(session, acc.id, AccountUpdate(platform="vultr"))
|
||||
cred = session.get(Credential, r.credential_id)
|
||||
assert cred.site == "vultr"
|
||||
# 清密码（空串）→ 凭据保留但无密码
|
||||
r = account_service.update_account(session, acc.id, AccountUpdate(login_password=""))
|
||||
assert r.has_login_password is False
|
||||
assert session.get(Credential, r.credential_id) is not None
|
||||
|
||||
|
||||
def test_account_otp_visible_in_read(session):
|
||||
"""账号 Read 暴露 has_otp，供前端在账号侧展示「获取验证码」"""
|
||||
acc = account_service.create_account(
|
||||
session, AccountCreate(name="me@gmail.com", platform="aliyun", login_password="pw")
|
||||
)
|
||||
secret, code = _otp_pair()
|
||||
credential_service.bind_otp(session, acc.credential_id, OtpBindRequest(secret=secret, code=code))
|
||||
listed = {a.id: a for a in account_service.list_accounts(session)}
|
||||
assert listed[acc.id].has_otp is True
|
||||
assert listed[acc.id].has_login_password is True
|
||||
# 账号侧取动态码（走凭据）
|
||||
otp = credential_service.current_otp(session, acc.credential_id)
|
||||
assert totp.verify(secret, otp["code"])
|
||||
|
||||
|
||||
def test_account_delete_keeps_credential(session):
|
||||
"""删除账号不删凭据（密码库独立留存，仍可查/可复制）"""
|
||||
acc = account_service.create_account(
|
||||
session, AccountCreate(name="me@gmail.com", platform="aliyun", login_password="pw")
|
||||
)
|
||||
cred_id = acc.credential_id
|
||||
account_service.delete_account(session, acc.id)
|
||||
|
||||
cred = session.get(Credential, cred_id)
|
||||
assert cred is not None
|
||||
assert crypto.decrypt(cred.password_encrypted) == "pw"
|
||||
# 凭据列表不再标记「来自平台账号」
|
||||
rows = credential_service.list_credentials(session)
|
||||
assert rows[0].account_id is None and rows[0].account_name is None
|
||||
|
||||
|
||||
def test_credential_list_marks_source_account(session):
|
||||
"""凭据列表反向标注来源账号（密码库中区分「平台账号」与游离条目）"""
|
||||
acc = account_service.create_account(
|
||||
session, AccountCreate(name="me@gmail.com", platform="aliyun", login_password="pw")
|
||||
)
|
||||
credential_service.create_credential(session, CredentialCreate(site="某论坛", username="a@b.c"))
|
||||
rows = {r.site: r for r in credential_service.list_credentials(session)}
|
||||
assert rows["阿里云"].account_id == acc.id
|
||||
assert rows["阿里云"].account_name == "me@gmail.com"
|
||||
assert rows["某论坛"].account_id is None
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Key Escrow（MASTER_KEY 密钥托管）测试
|
||||
|
||||
覆盖两层：
|
||||
- crypto 层：build_escrow/recover_master_key 往返、错误钥匙拒绝、格式校验
|
||||
- CLI 层：setup → 模拟 .env 丢失 key → recover --write 找回回写（§3.3 丢失演练）
|
||||
|
||||
运行：.venv/bin/pytest tests/test_escrow.py -v
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from app.core import crypto
|
||||
from app.core.config import settings
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||||
SETUP_SCRIPT = BASE_DIR / "scripts" / "setup_key_escrow.py"
|
||||
RECOVER_SCRIPT = BASE_DIR / "scripts" / "recover_master_key.py"
|
||||
|
||||
|
||||
# ----------------------------- crypto 层 ----------------------------- #
|
||||
def test_build_recover_roundtrip(monkeypatch):
|
||||
"""托管往返：build_escrow 后用同一把钥匙能还原 MASTER_KEY"""
|
||||
monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
|
||||
restore_key = Fernet.generate_key().decode()
|
||||
escrow = crypto.build_escrow(restore_key)
|
||||
assert escrow.startswith(crypto.ESCROW_PREFIX)
|
||||
assert crypto.recover_master_key(restore_key, escrow) == settings.MASTER_KEY
|
||||
|
||||
|
||||
def test_recover_wrong_key_rejected(monkeypatch):
|
||||
"""错误恢复钥匙必须抛 ValueError（不能静默返回错误 key）"""
|
||||
monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
|
||||
escrow = crypto.build_escrow(Fernet.generate_key().decode())
|
||||
with pytest.raises(ValueError):
|
||||
crypto.recover_master_key(Fernet.generate_key().decode(), escrow)
|
||||
|
||||
|
||||
def test_recover_invalid_prefix():
|
||||
"""缺 v1: 前缀视为格式无效"""
|
||||
with pytest.raises(ValueError):
|
||||
crypto.recover_master_key(Fernet.generate_key().decode(), "garbage-no-prefix")
|
||||
|
||||
|
||||
def test_build_escrow_requires_master_key(monkeypatch):
|
||||
"""MASTER_KEY 未配置时拒绝建立托管（避免生成解不开真实数据的废档）"""
|
||||
monkeypatch.setattr(settings, "MASTER_KEY", "")
|
||||
with pytest.raises(RuntimeError):
|
||||
crypto.build_escrow(Fernet.generate_key().decode())
|
||||
|
||||
|
||||
# ----------------------------- CLI 端到端 ----------------------------- #
|
||||
def _run_script(script: Path, env_extra: dict, args: list) -> subprocess.CompletedProcess:
|
||||
"""以子进程运行脚本（环境变量覆盖 .env：python-dotenv 不覆盖已存在变量）"""
|
||||
env = {**os.environ, **env_extra}
|
||||
return subprocess.run(
|
||||
[sys.executable, str(script), *args], capture_output=True, text=True, env=env
|
||||
)
|
||||
|
||||
|
||||
# Fernet key = base64url(32 字节) = 43 个 [A-Za-z0-9_-] 字符 + '='
|
||||
_FERNET_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{43}=$")
|
||||
|
||||
|
||||
def _extract_restore_key(stdout: str) -> str:
|
||||
"""从 setup 输出提取 RESTORE_KEY（44 字符 Fernet key，独立一行）"""
|
||||
for line in stdout.splitlines():
|
||||
s = line.strip()
|
||||
if _FERNET_KEY_RE.match(s):
|
||||
return s
|
||||
raise AssertionError("setup 输出未包含 RESTORE_KEY:\n" + stdout)
|
||||
|
||||
|
||||
def test_cli_loss_drill(tmp_path):
|
||||
"""丢失演练：托管建立 → .env 中 key 变为错误值 → recover --write 找回真实 key"""
|
||||
master_key = Fernet.generate_key().decode()
|
||||
escrow_path = tmp_path / "master_key.escrow"
|
||||
env_path = tmp_path / ".env"
|
||||
# 模拟丢失现场：.env 里是错误的 key，其它配置需原样保留
|
||||
env_path.write_text("API_KEY=abc\nMASTER_KEY=lost-wrong-key\n", encoding="utf-8")
|
||||
|
||||
# 1. 建立托管（MASTER_KEY 环境变量 = 正确 key）
|
||||
r = _run_script(SETUP_SCRIPT, {"MASTER_KEY": master_key}, ["--escrow", str(escrow_path)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert escrow_path.exists()
|
||||
assert (escrow_path.stat().st_mode & 0o777) == 0o600
|
||||
restore_key = _extract_restore_key(r.stdout)
|
||||
|
||||
# 2. 回验模式：--key 校验托管可解且与当前 MASTER_KEY 一致
|
||||
r = _run_script(
|
||||
SETUP_SCRIPT, {"MASTER_KEY": master_key},
|
||||
["--key", restore_key, "--escrow", str(escrow_path)],
|
||||
)
|
||||
assert r.returncode == 0 and "通过" in r.stdout
|
||||
|
||||
# 3. 恢复：即使运行时环境里的 MASTER_KEY 是错的，也能凭 RESTORE_KEY 找回并回写 .env
|
||||
r = _run_script(
|
||||
RECOVER_SCRIPT, {"MASTER_KEY": "irrelevant", "RESTORE_KEY": restore_key},
|
||||
["--escrow", str(escrow_path), "--env", str(env_path), "--write"],
|
||||
)
|
||||
assert r.returncode == 0, r.stderr
|
||||
text = env_path.read_text(encoding="utf-8")
|
||||
assert f"MASTER_KEY={master_key}" in text
|
||||
assert "API_KEY=abc" in text # 其它配置行不动
|
||||
assert "lost-wrong-key" not in text
|
||||
assert (tmp_path / ".env.bak-pre-recover").exists() # 回写前已备份
|
||||
|
||||
# 4. 错误钥匙恢复必须失败（退出码 1，.env 不被污染）
|
||||
r = _run_script(
|
||||
RECOVER_SCRIPT, {"RESTORE_KEY": Fernet.generate_key().decode()},
|
||||
["--escrow", str(escrow_path), "--env", str(env_path), "--write"],
|
||||
)
|
||||
assert r.returncode == 1 and "失败" in (r.stdout + r.stderr)
|
||||
assert f"MASTER_KEY={master_key}" in env_path.read_text(encoding="utf-8")
|
||||
|
||||
# 5. 幂等：托管已存在时默认不重建
|
||||
r = _run_script(SETUP_SCRIPT, {"MASTER_KEY": master_key}, ["--escrow", str(escrow_path)])
|
||||
assert r.returncode == 2 and "已存在" in r.stderr
|
||||
|
||||
# 6. --force 重建：新钥匙生效，旧钥匙作废
|
||||
r = _run_script(
|
||||
SETUP_SCRIPT, {"MASTER_KEY": master_key}, ["--escrow", str(escrow_path), "--force"]
|
||||
)
|
||||
assert r.returncode == 0
|
||||
new_key = _extract_restore_key(r.stdout)
|
||||
assert new_key != restore_key
|
||||
r = _run_script(
|
||||
RECOVER_SCRIPT, {"RESTORE_KEY": new_key}, ["--escrow", str(escrow_path)]
|
||||
)
|
||||
assert r.returncode == 0 and f"MASTER_KEY={master_key}" in r.stdout
|
||||
r = _run_script(
|
||||
RECOVER_SCRIPT, {"RESTORE_KEY": restore_key}, ["--escrow", str(escrow_path)]
|
||||
)
|
||||
assert r.returncode == 1 # 旧钥匙已作废
|
||||
@@ -0,0 +1,112 @@
|
||||
"""TOTP（RFC 6238）自实现测试
|
||||
|
||||
官方向量锚定正确性（RFC 6238 附录 B：SHA1、20 字节 ASCII secret
|
||||
"12345678901234567890"），覆盖：码生成、剩余秒数、录入校验窗口、
|
||||
otpauth URI 解析/重建、随机 secret 可用性。
|
||||
|
||||
运行：.venv/bin/pytest tests/test_totp.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core import totp
|
||||
|
||||
# RFC 6238 附录 B 官方向量 secret 的 base32 形式
|
||||
RFC_SECRET_B32 = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
||||
# (时刻 T, 官方 8 位码)；本实现为 6 位 = 8 位码取模 10^6 补零
|
||||
RFC_VECTORS = [
|
||||
(59, "94287082"),
|
||||
(1111111109, "07081804"),
|
||||
(1111111111, "14050471"),
|
||||
(1234567890, "89005924"),
|
||||
(2000000000, "69279037"),
|
||||
(20000000000, "65353130"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ts,code8", RFC_VECTORS)
|
||||
def test_rfc6238_official_vectors(ts, code8):
|
||||
"""逐条对齐 RFC 6238 附录 B 官方向量（6 位截断）"""
|
||||
code6, _ = totp.totp_at(RFC_SECRET_B32, ts=ts)
|
||||
assert code6 == str(int(code8) % 10**6).zfill(6)
|
||||
|
||||
|
||||
def test_expires_in():
|
||||
"""剩余秒数：T=59 窗口已过 29s 剩 1s；T=60 新窗口剩 30s"""
|
||||
_, left = totp.totp_at(RFC_SECRET_B32, ts=59)
|
||||
assert left == 1
|
||||
_, left = totp.totp_at(RFC_SECRET_B32, ts=60)
|
||||
assert left == 30
|
||||
|
||||
|
||||
def test_verify_window_tolerance():
|
||||
"""录入校验容忍 ±1 步进时钟偏差，隔两个窗口必须失败"""
|
||||
code, _ = totp.totp_at(RFC_SECRET_B32, ts=59) # counter=1
|
||||
assert totp.verify(RFC_SECRET_B32, code, ts=59)
|
||||
assert totp.verify(RFC_SECRET_B32, code, ts=61) # counter=2，窗口含 1
|
||||
assert totp.verify(RFC_SECRET_B32, code, ts=80) # counter=2
|
||||
assert not totp.verify(RFC_SECRET_B32, code, ts=150) # counter=5，超出窗口
|
||||
|
||||
|
||||
def test_verify_rejects_bad_input():
|
||||
assert not totp.verify(RFC_SECRET_B32, "12345") # 位数不对
|
||||
assert not totp.verify(RFC_SECRET_B32, "abcdef") # 非数字
|
||||
assert not totp.verify(RFC_SECRET_B32, "000000", ts=59) # 错误码
|
||||
assert not totp.verify(RFC_SECRET_B32, "", ts=59)
|
||||
|
||||
|
||||
def test_b32decode_tolerant():
|
||||
"""小写/空白/缺填充都应正确解码"""
|
||||
raw = totp.b32decode(RFC_SECRET_B32)
|
||||
assert totp.b32decode(RFC_SECRET_B32.lower()) == raw
|
||||
assert totp.b32decode("gezdg nbvgy 3tqoj qgezd gnbvg y3tqo jq") == raw
|
||||
assert totp.b32decode(RFC_SECRET_B32.rstrip("=")) == raw
|
||||
|
||||
|
||||
def test_parse_otpauth_uri_full():
|
||||
uri = (
|
||||
"otpauth://totp/GitHub:me@example.com"
|
||||
"?secret=JBSWY3DPEHPK3PXP&issuer=GitHub&algorithm=SHA1&period=30&digits=6"
|
||||
)
|
||||
p = totp.parse_otpauth_uri(uri)
|
||||
assert p["secret"] == "JBSWY3DPEHPK3PXP"
|
||||
assert p["issuer"] == "GitHub"
|
||||
assert p["account"] == "me@example.com"
|
||||
|
||||
|
||||
def test_parse_otpauth_uri_label_only():
|
||||
"""无 issuer 参数时从 Label 前缀提取，account 需 URL 解码"""
|
||||
p = totp.parse_otpauth_uri("otpauth://totp/Google:me%40gmail.com?secret=JBSWY3DPEHPK3PXP")
|
||||
assert p["issuer"] == "Google"
|
||||
assert p["account"] == "me@gmail.com"
|
||||
|
||||
|
||||
def test_parse_otpauth_uri_rejects():
|
||||
with pytest.raises(ValueError):
|
||||
totp.parse_otpauth_uri("https://example.com") # 非 otpauth 协议
|
||||
with pytest.raises(ValueError):
|
||||
totp.parse_otpauth_uri("otpauth://hotp/x?secret=JBSWY3DPEHPK3PXP") # 非 TOTP
|
||||
with pytest.raises(ValueError):
|
||||
totp.parse_otpauth_uri("otpauth://totp/x?issuer=y") # 缺 secret
|
||||
with pytest.raises(ValueError):
|
||||
totp.parse_otpauth_uri("otpauth://totp/x?secret=JBSWY3DPEHPK3PXP&digits=8") # 非 6 位
|
||||
with pytest.raises(ValueError):
|
||||
totp.parse_otpauth_uri("otpauth://totp/x?secret=JBSWY3DPEHPK3PXP&period=60") # 非 30s
|
||||
|
||||
|
||||
def test_build_otpauth_uri_roundtrip():
|
||||
"""重建的 URI 可被解析回同样的 secret/issuer/account（换机导出场景）"""
|
||||
uri = totp.build_otpauth_uri("JBSWY3DPEHPK3PXP", "GitHub", "me@example.com")
|
||||
p = totp.parse_otpauth_uri(uri)
|
||||
assert p["secret"] == "JBSWY3DPEHPK3PXP"
|
||||
assert p["issuer"] == "GitHub"
|
||||
assert p["account"] == "me@example.com"
|
||||
|
||||
|
||||
def test_random_secret_usable():
|
||||
"""随机 secret 生成的码应能通过自身校验（录入闭环）"""
|
||||
s = totp.random_secret()
|
||||
code, left = totp.totp_at(s)
|
||||
assert len(code) == 6 and code.isdigit()
|
||||
assert 1 <= left <= 30
|
||||
assert totp.verify(s, code)
|
||||
Reference in New Issue
Block a user