vps-manager/tests/test_credential_service.py

"""凭据服务（密码库 + 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