aps-agent/tests/golden/test_audit_events_ingest.py

# ============================================================
# POST /api/gov/audit/events 审计事件上报端点（矩阵 103 收口，round-45 HH）
#   H1 web 登录态 → 事件写入审计链（actor=登录用户）
#   H2 web 未登录 → 401（端点内显式要求登录身份，与其他 gov 端点语义一致）
#   H3 desktop nonce 门禁：缺失/错误 nonce → 403；正确 nonce（无 cookie）→ 200 落链
#   H4 idempotencyKey 幂等：重复上报不重复落链
#   H5 非法载荷 → 4xx 明确（422 校验失败 / 403 门禁）
# ============================================================
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from server.sidecar import SidecarIdentityApp
from tests.auth_provider import install_test_auth

NONCE = "a" * 64


class _Store:
    """最小世界状态替身（与 test_audit_anchor_api.py 同构）。"""

    def __init__(self) -> None:
        self.tenant_uuid = "platform"
        self.world_key = "default"
        self.data: dict = {"auditEvents": []}

    def next_id(self, _kind: str) -> int:
        return len(self.data.get("auditEvents", [])) + 1

    def save(self) -> None:
        pass


@pytest.fixture()
def web_client(monkeypatch):
    """web 模式：无 APS_SIDECAR_NONCE（无 nonce 门禁）+ 测试登录身份。"""
    import server.gateway.app as gateway_module
    install_test_auth(monkeypatch, "tenant-audit-events-0000000000000000001")
    store = _Store()
    monkeypatch.setattr(gateway_module, "get_store", lambda: store)
    monkeypatch.delenv("APS_SIDECAR_NONCE", raising=False)
    c = TestClient(gateway_module.create_app())
    login = c.post("/api/auth/login", json={"username": "planner", "password": "test"})
    assert login.status_code == 200, login.text
    return c, store


@pytest.fixture()
def desktop_app(monkeypatch):
    """desktop 模式：SidecarIdentityApp nonce 门禁 + APS_SIDECAR_NONCE 环境变量。"""
    import server.gateway.app as gateway_module
    install_test_auth(monkeypatch, "tenant-audit-events-0000000000000000002")
    store = _Store()
    monkeypatch.setattr(gateway_module, "get_store", lambda: store)
    monkeypatch.setenv("APS_SIDECAR_NONCE", NONCE)
    inner = gateway_module.create_app()
    return TestClient(SidecarIdentityApp(inner, NONCE)), store


def _event(**overrides):
    payload = {
        "category": "UPGRADE",
        "action": "upgrade.succeeded",
        "power": "P0",
        "actor": "desktop-updater",
        "target": {"type": "UPGRADE"},
        "rationale": {"fromVersion": "0.1.0", "toVersion": "0.2.0"},
        "timestamp": "2026-08-02T10:00:00+08:00",
        "result": "SUCCESS",
        "idempotencyKey": "evt-1",
    }
    payload.update(overrides)
    return payload


# ---------------- H1：web 登录态写入审计 ----------------
def test_web_logged_in_writes_audit_event(web_client):
    c, store = web_client
    r = c.post("/api/gov/audit/events", json=_event())
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["ok"] is True and body["duplicate"] is False
    assert body["event"]["hash"]                          # 链式哈希已生成
    # 事件真实落链（gov/audit 可读），web 模式 actor 以登录身份为准
    chain = c.get("/api/gov/audit").json()
    assert [e["action"] for e in chain["events"]] == ["upgrade.succeeded"]
    assert chain["events"][0]["actor"] == "planner"
    assert len(store.data["auditEvents"]) == 1


# ---------------- H2：web 未登录 401 ----------------
def test_web_anonymous_gets_401(web_client):
    c, _ = web_client
    c.cookies.clear()
    r = c.post("/api/gov/audit/events", json=_event())
    assert r.status_code == 401, r.text


# ---------------- H5：非法载荷 4xx 明确 ----------------
def test_web_invalid_payload_gets_4xx(web_client):
    c, _ = web_client
    r = c.post("/api/gov/audit/events", json=_event(power="P9"))   # power 越界
    assert r.status_code == 400, r.text
    assert r.json()["detail"]["code"] == "INVALID_POWER"
    r2 = c.post("/api/gov/audit/events", json={"action": ""})      # action 为空
    assert r2.status_code == 400, r2.text
    assert r2.json()["detail"]["code"] == "INVALID_ACTION"
    r3 = c.post("/api/gov/audit/events", data="not-json")          # 非 JSON 请求体
    assert r3.status_code == 400, r3.text
    assert r3.json()["detail"]["code"] == "INVALID_JSON"


# ---------------- H3：desktop nonce 门禁 ----------------
def test_desktop_nonce_gate_missing_nonce_403(desktop_app):
    c, _ = desktop_app
    r = c.post("/api/gov/audit/events", json=_event())
    assert r.status_code == 403, r.text                  # 无 nonce → 门禁 403


def test_desktop_nonce_gate_wrong_nonce_403(desktop_app):
    c, _ = desktop_app
    r = c.post("/api/gov/audit/events", json=_event(),
               headers={"x-aps-sidecar-nonce": "b" * 64})
    assert r.status_code == 403, r.text                  # 错误 nonce → 403


def test_desktop_updater_with_nonce_writes_event(desktop_app):
    """updater 场景：无 cookie、仅带 nonce → 200 且事件落链（nonce 门禁即身份）。"""
    c, store = desktop_app
    r = c.post("/api/gov/audit/events", json=_event(),
               headers={"x-aps-sidecar-nonce": NONCE})
    assert r.status_code == 200, r.text
    assert store.data["auditEvents"][0]["actor"] == "desktop-updater"
    assert store.data["auditEvents"][0]["rationale"]["idempotencyKey"] == "evt-1"
    # 链完整性：新事件可被 verify_audit_chain 校验（哈希链无断点）
    from server.agent_core.registry import verify_audit_chain
    assert verify_audit_chain(store.data["auditEvents"])["ok"] is True


# ---------------- H4：idempotencyKey 幂等 ----------------
def test_idempotency_key_dedupes_repeated_report(web_client):
    c, store = web_client
    first = c.post("/api/gov/audit/events", json=_event())
    assert first.status_code == 200 and first.json()["duplicate"] is False
    event_id = first.json()["eventId"]
    second = c.post("/api/gov/audit/events", json=_event())          # 同 idempotencyKey
    assert second.status_code == 200, second.text
    assert second.json()["duplicate"] is True
    assert second.json()["eventId"] == event_id
    assert len(store.data["auditEvents"]) == 1                       # 未重复落链
    third = c.post("/api/gov/audit/events", json=_event(idempotencyKey="evt-2"))
    assert third.status_code == 200 and third.json()["duplicate"] is False
    assert len(store.data["auditEvents"]) == 2