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,
    OtpVerifyRequest,
)
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_without_code_marks_unverified(session):
    """拿不到当前动态码也允许先录入 secret

    真实场景下 secret 多为历史留存，录入那一刻手上并没有能出码的验证器；
    强约束只会把人挡在门外。改为存下来并标记未验证，事后补验。
    """
    secret, _ = _otp_pair()
    read = credential_service.create_credential(
        session, CredentialCreate(site="GitHub", otp_secret=secret)
    )
    assert read.has_otp is True and read.otp_verified is False
    # 未验证不影响出码（系统本身就是验证器），只是不保证站点认
    otp = credential_service.current_otp(session, read.id)
    assert len(otp["code"]) == 6


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
    assert read.otp_verified 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 and read.otp_verified 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 and updated.otp_verified is True
    # 提供了动态码就必须对：错码拒绝，且不覆盖已绑定的 secret
    with pytest.raises(HTTPException):
        credential_service.bind_otp(session, read.id, OtpBindRequest(secret="JBSWY3DPEHPK3PXP", code="111111"))
    row = session.get(Credential, read.id)
    assert crypto.decrypt(row.otp_secret_encrypted) == secret
    unbound = credential_service.unbind_otp(session, read.id)
    assert unbound.has_otp is False and unbound.otp_verified is False


def test_bind_otp_without_code_marks_unverified(session):
    """编辑路径同样允许留空：先存 secret，标记未验证"""
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    secret, _ = _otp_pair()
    updated = credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret))
    assert updated.has_otp is True and updated.otp_verified is False


def test_verify_otp_promotes_to_verified(session):
    """后置补验：从能出码的设备取当前码，通过后解除「未验证」"""
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    secret, code = _otp_pair()
    credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret))
    updated = credential_service.verify_otp(session, read.id, OtpVerifyRequest(code=code))
    assert updated.otp_verified is True
    assert session.get(Credential, read.id).otp_verified_at is not None


def test_verify_otp_wrong_code_keeps_unverified(session):
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    secret, _ = _otp_pair()
    credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret))
    with pytest.raises(HTTPException) as e:
        credential_service.verify_otp(session, read.id, OtpVerifyRequest(code="000000"))
    assert e.value.status_code == 400 and "不匹配" in e.value.detail
    assert session.get(Credential, read.id).otp_verified_at is None


@pytest.mark.parametrize("bad", ["12345", "abcdef", "", "1234567"])
def test_verify_otp_rejects_malformed_code(session, bad):
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    secret, _ = _otp_pair()
    credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret))
    with pytest.raises(HTTPException) as e:
        credential_service.verify_otp(session, read.id, OtpVerifyRequest(code=bad))
    assert e.value.status_code == 400


def test_verify_otp_requires_bound_secret(session):
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    with pytest.raises(HTTPException) as e:
        credential_service.verify_otp(session, read.id, OtpVerifyRequest(code="123456"))
    assert e.value.status_code == 400 and "未绑定" in e.value.detail


def test_rebinding_secret_resets_verified(session):
    """换 secret 必须重新验证：旧验证状态不能顺延给新 secret"""
    read = credential_service.create_credential(session, CredentialCreate(site="GitHub"))
    secret, code = _otp_pair()
    credential_service.bind_otp(session, read.id, OtpBindRequest(secret=secret, code=code))
    assert session.get(Credential, read.id).otp_verified_at is not None
    new_secret, _ = _otp_pair()
    updated = credential_service.bind_otp(session, read.id, OtpBindRequest(secret=new_secret))
    assert updated.otp_verified 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