from __future__ import annotations class FakeGovernanceAuditService: def __init__(self): self.calls = [] def coverage(self, **filters): self.calls.append(("coverage", filters)) return { "period_start": filters["period_start"].isoformat(), "period_end": filters["period_end"].isoformat(), "categories": [ { "category": "authentication", "count": 1, "available": True, "latest_at": "2026-07-30T00:00:00Z", } ], } def list_events(self, **filters): self.calls.append(("events", filters)) return { "records": [ { "event_uid": "authentication:1", "category": "authentication", "action": "login", "status": "success", "safe_detail": {"username": "admin"}, } ], "page": int(filters["page"]), "page_size": int(filters["page_size"]), "total": 1, } def create_seal(self, **payload): self.calls.append(("create_seal", payload)) return { "uid": "seal-1", "event_count": 1, "root_hash": "a" * 64, "signature": "b" * 64, } def list_seals(self, *, limit): self.calls.append(("list_seals", {"limit": limit})) return [{"uid": "seal-1", "event_count": 1}] def verify_seal(self, seal_uid): self.calls.append(("verify_seal", {"seal_uid": seal_uid})) return {"uid": seal_uid, "integrity_status": "intact"} def _client(monkeypatch, *, role="admin"): from app import create_app from app.api.system import governance_audit service = FakeGovernanceAuditService() identity = { "id": "00000000-0000-7000-8000-000000000111", "roles": [role], } monkeypatch.setattr( "app.core.system.permissions.authenticate_request", lambda: identity, ) monkeypatch.setattr( governance_audit, "get_governance_audit_service", lambda: service, ) monkeypatch.setattr( governance_audit, "get_security_checks", lambda: { "seal_ready": True, "assurance": "签名封存用于检测篡改，不等于阻止数据库管理员修改", "checks": [{"code": "safe_error_boundary", "status": "passed"}], }, ) app = create_app() app.config.update(TESTING=True) return app.test_client(), service def test_admin_reads_security_coverage_events_and_seals(monkeypatch): http, service = _client(monkeypatch) window = ( "period_start=2026-07-01T00:00:00Z&" "period_end=2026-07-30T00:00:00Z" ) security = http.get( "/api/system/governance-audit/security-checks" ) coverage = http.get( f"/api/system/governance-audit/coverage?{window}" ) events = http.get( f"/api/system/governance-audit/events?{window}" "&category=authentication&page=1&page_size=20" ) seals = http.get("/api/system/governance-audit/seals?limit=25") assert security.status_code == 200 assert security.get_json()["data"]["seal_ready"] is True assert coverage.status_code == 200 assert events.status_code == 200 assert events.get_json()["data"]["records"][0]["safe_detail"] == { "username": "admin" } assert seals.status_code == 200 assert [call[0] for call in service.calls] == [ "coverage", "events", "list_seals", ] def test_seal_creation_uses_server_identity_and_verification_is_explicit( monkeypatch, ): http, service = _client(monkeypatch) created = http.post( "/api/system/governance-audit/seals", json={ "period_start": "2026-07-01T00:00:00Z", "period_end": "2026-07-30T00:00:00Z", "categories": ["authentication", "ingestion"], "sealed_by": "attacker-controlled", }, ) verified = http.post( "/api/system/governance-audit/seals/seal-1/verify" ) assert created.status_code == 200 assert service.calls[0][1]["actor_uid"] == ( "00000000-0000-7000-8000-000000000111" ) assert "sealed_by" not in service.calls[0][1] assert verified.status_code == 200 assert verified.get_json()["data"]["integrity_status"] == "intact" def test_viewer_and_editor_cannot_read_or_seal(monkeypatch): for role in ("viewer", "editor"): http, _service = _client(monkeypatch, role=role) assert http.get( "/api/system/governance-audit/events" ).status_code == 403 assert http.post( "/api/system/governance-audit/seals", json={}, ).status_code == 403 def test_invalid_query_and_repository_failure_return_safe_contract(monkeypatch): from app.core.system.governance_audit import GovernanceAuditInvalid http, service = _client(monkeypatch) def invalid(**_kwargs): raise GovernanceAuditInvalid("page_size must be between 1 and 100") service.list_events = invalid response = http.get( "/api/system/governance-audit/events" "?period_start=2026-07-01T00:00:00Z" "&period_end=2026-07-30T00:00:00Z&page_size=101" ) assert response.status_code == 400 assert response.get_json()["error"]["code"] == ( "GOVERNANCE_AUDIT_INVALID" ) def unavailable(**_kwargs): raise RuntimeError( "postgresql://admin:database-secret@db/dataops " "authorization=Bearer-secret" ) service.list_events = unavailable failed = http.get( "/api/system/governance-audit/events" "?period_start=2026-07-01T00:00:00Z" "&period_end=2026-07-30T00:00:00Z" ) payload = failed.get_json() assert failed.status_code == 503 assert payload["message"] == "审计与运行证据暂不可用" assert "database-secret" not in repr(payload) assert "Bearer-secret" not in repr(payload) def test_production_without_dedicated_key_blocks_seal_and_verify(monkeypatch): from flask import Flask from app.api.system import governance_audit production = Flask(__name__) production.config.update( FLASK_ENV="production", TESTING=False, ) development = Flask(__name__) development.config.update( FLASK_ENV="development", TESTING=False, ) with production.app_context(): assert governance_audit._sealing_allowed(False) is False assert governance_audit._sealing_allowed(True) is True with development.app_context(): assert governance_audit._sealing_allowed(False) is True http, service = _client(monkeypatch) monkeypatch.setattr( governance_audit, "_sealing_allowed", lambda _dedicated_ready: False, ) created = http.post( "/api/system/governance-audit/seals", json={ "period_start": "2026-07-01T00:00:00Z", "period_end": "2026-07-30T00:00:00Z", }, ) verified = http.post( "/api/system/governance-audit/seals/seal-1/verify" ) assert created.status_code == 503 assert verified.status_code == 503 assert service.calls == []