tests/test_auth_service.py

"""``AuthService`` 的单元测试（任务 6）。

覆盖需求：
- 5.4：加盐哈希可校验；存储格式为 ``salt$hash``，非无盐 SHA-256；同一密码两次
  哈希不同（随机盐）；兼容旧无盐记录校验。
- 5.1：未设置 ``ADMIN_INIT_PASSWORD`` 时不创建默认账户，且日志无明文密码；
  已设置时以加盐哈希创建管理员，且日志 / 哈希均不泄露明文密码。
- 15.5：角色隔离 ``can_view_audit`` / ``visible_skills`` 按 admin / researcher /
  auditor 区分。

测试自包含：用注入的假存储（admin_exists / create_admin）完全替换数据库依赖，
不触网、不写盘，可在 CI 与离线环境稳定运行。导入路径由 tests/conftest.py 设置，
底座以顶层包 ``services`` 导入（与 test_llm_service.py 一致）。
"""

from __future__ import annotations

import hashlib
import logging

import pytest

from services.auth_service import (  # noqa: E402
    AuthService,
    hash_password,
    verify_password,
    is_legacy_hash,
    ROLE_ADMIN,
    ROLE_RESEARCHER,
    ROLE_AUDITOR,
)


# ---------------------------------------------------------------------------
# 加盐哈希：往返校验、格式、随机盐（需求 5.4）
# ---------------------------------------------------------------------------

def test_hash_then_verify_roundtrip():
    """正确密码校验通过，错误密码校验失败。"""
    stored = hash_password("S3cret-pw")
    assert verify_password("S3cret-pw", stored) is True
    assert verify_password("wrong-pw", stored) is False


def test_stored_format_is_salt_dollar_hash():
    """存储格式为 ``salt$hash``：含单个 ``$`` 分隔，盐与哈希均为十六进制。"""
    stored = hash_password("pw")
    assert stored.count("$") == 1
    salt_hex, hash_hex = stored.split("$")
    assert salt_hex and hash_hex
    # 均可被解析为十六进制
    int(salt_hex, 16)
    int(hash_hex, 16)


def test_stored_is_not_bare_unsalted_sha256():
    """存储值不得是无盐 SHA-256（64 位十六进制、无 ``$``）。"""
    stored = hash_password("pw")
    assert "$" in stored
    bare = hashlib.sha256("pw".encode("utf-8")).hexdigest()
    assert stored != bare
    assert not is_legacy_hash(stored)


def test_same_password_hashes_differ_due_to_random_salt():
    """同一密码两次哈希结果不同（随机盐），但都能各自校验通过。"""
    h1 = hash_password("same-pw")
    h2 = hash_password("same-pw")
    assert h1 != h2
    assert verify_password("same-pw", h1) is True
    assert verify_password("same-pw", h2) is True


def test_explicit_salt_is_deterministic():
    """显式传盐时哈希确定（便于测试），不同盐得到不同哈希。"""
    h1 = hash_password("pw", salt="00" * 16)
    h2 = hash_password("pw", salt="00" * 16)
    h3 = hash_password("pw", salt="11" * 16)
    assert h1 == h2
    assert h1 != h3
    assert verify_password("pw", h1) is True


# ---------------------------------------------------------------------------
# 向后兼容：旧无盐 SHA-256 记录仍可校验（需求 5.4 迁移）
# ---------------------------------------------------------------------------

def test_legacy_unsalted_sha256_still_verifies():
    """旧无盐 SHA-256 哈希在新校验逻辑下仍可通过 / 拒绝。"""
    legacy = hashlib.sha256("legacy-pw".encode("utf-8")).hexdigest()
    assert is_legacy_hash(legacy) is True
    assert verify_password("legacy-pw", legacy) is True
    assert verify_password("nope", legacy) is False


def test_verify_rejects_empty_and_malformed():
    """空值与畸形存储值安全拒绝，不抛异常。"""
    assert verify_password("pw", "") is False
    assert verify_password("pw", None) is False  # type: ignore[arg-type]
    assert verify_password("pw", "$abc") is False
    assert verify_password("pw", "nothex$nothex") is False


# ---------------------------------------------------------------------------
# 角色隔离（需求 15.5）
# ---------------------------------------------------------------------------

def test_can_view_audit_by_role():
    """审计区可见性：管理员与审计员可见，研发人员不可见。"""
    svc = AuthService()
    assert svc.can_view_audit(ROLE_ADMIN) is True
    assert svc.can_view_audit(ROLE_AUDITOR) is True
    assert svc.can_view_audit(ROLE_RESEARCHER) is False


def test_can_run_analysis_by_role():
    """分析执行权限：管理员与研发人员可执行，审计员只读不可执行。"""
    svc = AuthService()
    assert svc.can_run_analysis(ROLE_ADMIN) is True
    assert svc.can_run_analysis(ROLE_RESEARCHER) is True
    assert svc.can_run_analysis(ROLE_AUDITOR) is False


def test_visible_skills_differ_by_role():
    """可见 Skill 因角色而异：admin/researcher 见全部分析 Skill，auditor 不见。"""
    svc = AuthService(all_skill_ids=["stability", "compatibility", "general_qa"])
    admin_skills = svc.visible_skills(ROLE_ADMIN)
    researcher_skills = svc.visible_skills(ROLE_RESEARCHER)
    auditor_skills = svc.visible_skills(ROLE_AUDITOR)

    assert admin_skills == {"stability", "compatibility", "general_qa"}
    assert researcher_skills == {"stability", "compatibility", "general_qa"}
    assert auditor_skills == set()
    # 审计员可见集合与研发人员不同（隔离生效）
    assert auditor_skills != researcher_skills


def test_legacy_user_role_treated_as_researcher():
    """历史 ``user`` 角色与未知角色按最小权限归一化为研发人员。"""
    svc = AuthService()
    assert svc.normalize_role("user") == ROLE_RESEARCHER
    assert svc.normalize_role(None) == ROLE_RESEARCHER
    assert svc.normalize_role("unknown") == ROLE_RESEARCHER
    assert svc.can_view_audit("user") is False
    assert svc.is_admin("user") is False


# ---------------------------------------------------------------------------
# 管理员初始化与日志无密码（需求 5.1 / 5.2）
# ---------------------------------------------------------------------------

def test_init_admin_skips_when_password_unset(caplog):
    """未设置 ADMIN_INIT_PASSWORD 时不创建账户，日志提示「管理员未初始化」。"""
    svc = AuthService()
    created = []

    def admin_exists():
        return False

    def create_admin(email, password_hash):
        created.append((email, password_hash))

    with caplog.at_level(logging.DEBUG, logger="services.auth_service"):
        result = svc.init_admin(
            admin_exists=admin_exists,
            create_admin=create_admin,
            env={},  # 无 ADMIN_INIT_PASSWORD
        )

    assert result is False
    assert created == []  # 未创建任何账户
    assert any("管理员未初始化" in r.getMessage() for r in caplog.records)


def test_init_admin_creates_salted_hash_when_password_set(caplog):
    """设置 ADMIN_INIT_PASSWORD 时以加盐哈希创建管理员，日志不含明文密码。"""
    secret = "Env-Admin-Pw-12345"
    svc = AuthService(admin_email="admin")
    created = []

    def admin_exists():
        return False

    def create_admin(email, password_hash):
        created.append((email, password_hash))

    with caplog.at_level(logging.DEBUG, logger="services.auth_service"):
        result = svc.init_admin(
            admin_exists=admin_exists,
            create_admin=create_admin,
            env={"ADMIN_INIT_PASSWORD": secret},
        )

    assert result is True
    assert len(created) == 1
    email, stored = created[0]
    assert email == "admin"
    # 存储为加盐哈希，且可校验
    assert "$" in stored and not is_legacy_hash(stored)
    assert verify_password(secret, stored) is True
    # 日志与存储哈希均不泄露明文密码
    for record in caplog.records:
        assert secret not in record.getMessage()
    assert secret not in stored


def test_init_admin_skips_when_admin_exists(caplog):
    """管理员已存在时跳过创建（即便设置了环境变量）。"""
    svc = AuthService()
    created = []

    with caplog.at_level(logging.DEBUG, logger="services.auth_service"):
        result = svc.init_admin(
            admin_exists=lambda: True,
            create_admin=lambda e, h: created.append((e, h)),
            env={"ADMIN_INIT_PASSWORD": "whatever"},
        )

    assert result is False
    assert created == []


def test_no_password_leaks_in_any_init_log(caplog):
    """综合断言：初始化全流程日志中不出现任何明文密码（需求 5.2）。"""
    secret = "Sensitive-PLAINTEXT-PW"
    svc = AuthService()
    with caplog.at_level(logging.DEBUG, logger="services.auth_service"):
        svc.init_admin(
            admin_exists=lambda: False,
            create_admin=lambda e, h: None,
            env={"ADMIN_INIT_PASSWORD": secret},
        )
    for record in caplog.records:
        assert secret not in record.getMessage()


if __name__ == "__main__":  # pragma: no cover
    import sys
    sys.exit(pytest.main([__file__, "-v"]))