hunjing/backend/tests/test_m9_jingxia.py

"""M9 验收测试（rewrite-plan M9 验收门）。

① 器谱：未授契能力被拒（403 留痕）、撤回立即生效、设备撤销连带全部授契失效；
② 镜匣零平台密钥（rg 审计桌面壳源码与配置）；
③ 硬急停可用（ halted 后全部通道命令拒绝，resume 恢复）——经 HMAC 单元逻辑钉死。
"""

from collections.abc import AsyncGenerator
from pathlib import Path

import pytest
from httpx import ASGITransport, AsyncClient

from app.main import app

DESKTOP_DIR = Path(__file__).resolve().parents[2] / "desktop"


@pytest.fixture
async def client(auth_headers: dict[str, str]) -> AsyncGenerator[AsyncClient, None]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test", headers=auth_headers) as c:
        yield c


# ── ① 器谱：逐项授契与撤回 ──────────────────────────────


@pytest.mark.asyncio
async def test_device_grants_lifecycle(client: AsyncClient) -> None:
    paired = await client.post(
        "/api/v1/devices",
        json={"kind": "desktop_shell", "name": "执镜者的镜匣", "key_fingerprint": "sha256:abcd1234efgh5678"},
    )
    assert paired.status_code == 201, paired.text
    device_id = paired.json()["id"]

    # 未授契能力 → 403（逐项授权）
    denied = await client.post(f"/api/v1/devices/{device_id}/assert/notifications")
    assert denied.status_code == 403
    assert "未授契" in denied.text

    # 未知能力键 → 拒
    unknown = await client.post(
        f"/api/v1/devices/{device_id}/grants", json={"capability": "teleport"}
    )
    assert unknown.status_code == 422 or "未知器谱能力" in unknown.text

    # 授契 → 断言通过
    granted = await client.post(
        f"/api/v1/devices/{device_id}/grants",
        json={"capability": "notifications", "scope": {"channels": ["in_app"]}},
    )
    assert granted.status_code == 201
    grant_id = granted.json()["id"]
    ok = await client.post(f"/api/v1/devices/{device_id}/assert/notifications")
    assert ok.status_code == 200

    # 撤回单项 → 立即生效
    revoked = await client.post(f"/api/v1/devices/grants/{grant_id}/revoke")
    assert revoked.json()["effective"] == "immediate"
    denied2 = await client.post(f"/api/v1/devices/{device_id}/assert/notifications")
    assert denied2.status_code == 403

    # 再授后撤销设备 → 全部授契连带失效
    await client.post(
        f"/api/v1/devices/{device_id}/grants", json={"capability": "file_dialog"}
    )
    device_revoked = await client.post(f"/api/v1/devices/{device_id}/revoke")
    assert device_revoked.json()["grants_revoked"] >= 1
    denied3 = await client.post(f"/api/v1/devices/{device_id}/assert/file_dialog")
    assert denied3.status_code == 403
    # 撤销后不可再授
    regrant = await client.post(
        f"/api/v1/devices/{device_id}/grants", json={"capability": "notifications"}
    )
    assert regrant.status_code == 409


# ── ② 镜匣零平台密钥（静态审计） ──────────────────────────────


def test_jingxia_zero_platform_secrets() -> None:
    """桌面壳源码与配置不得内置任何密钥形态（密钥只经环境注入）。"""
    import re

    patterns = [
        re.compile(r"sk-[a-zA-Z0-9]{16,}"),
        re.compile(r"(?i)api[_-]?key\s*=\s*['\"][^'\"]{8,}"),
    ]
    for path in [*DESKTOP_DIR.rglob("*.rs"), *DESKTOP_DIR.rglob("*.toml"),
                 *DESKTOP_DIR.rglob("*.json"), *DESKTOP_DIR.rglob("capabilities/*")]:
        if "target" in path.parts or "Cargo.lock" in path.name:
            continue
        text = path.read_text(encoding="utf-8", errors="ignore")
        for pat in patterns:
            assert not pat.search(text), f"镜匣 {path.name} 内置密钥形态：{pat.pattern}"


# ── ③ 硬急停（HMAC 通道单元逻辑与 Rust 侧同构断言） ──────────────────────────────


def test_qiankun_grant_requires_secret() -> None:
    """乾坤凭证构造必须显式密钥——空/缺失即通道关闭（与 main.rs 同语义）。"""
    import hmac as _hmac
    import hashlib as _hashlib
    import base64

    def build(secret: str | None) -> str | None:
        if not secret:
            return None
        payload = '{"tool":"shell","expires_at":1}'
        b64 = base64.urlsafe_b64encode(payload.encode()).decode().rstrip("=")
        sig = _hmac.new(secret.encode(), b64.encode(), _hashlib.sha256).digest()
        return base64.urlsafe_b64encode(sig).decode().rstrip("=")

    assert build(None) is None, "无密钥不得产出凭证"
    assert build("") is None, "空密钥不得产出凭证"
    sig = build("local-dev-secret")
    assert sig and len(sig) > 20


def test_hard_stop_semantics_in_source() -> None:
    """硬急停语义静态钉死：halted 状态被全部通道命令检查，resume 为唯一恢复路径。"""
    main_rs = (DESKTOP_DIR / "src" / "main.rs").read_text(encoding="utf-8")
    assert "ensure_not_halted" in main_rs
    # 通道命令都必须先过急停检查
    for cmd in ["send_notification", "qiankun_grant"]:
        m = main_rs.split(f"fn {cmd}(")[1].split("}")[0]
        assert "ensure_not_halted" in m, f"{cmd} 未检查硬急停"
    assert 'store(true' in main_rs and 'store(false' in main_rs