| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import logging |
| 5 | |
| 6 | from app.audit import StructuredMcpAuditLogger |
| 7 | from app.contracts import McpRiskClass, McpToolKind |
| 8 | |
| 9 | |
| 10 | def test_structured_audit_logs_metadata_trace_and_redacts_sensitive_details( |
| 11 | auth, caplog, monkeypatch |
| 12 | ) -> None: |
| 13 | class SpanContext: |
| 14 | is_valid = True |
| 15 | trace_id = 0x1234 |
| 16 | |
| 17 | class Span: |
| 18 | @staticmethod |
| 19 | def get_span_context(): |
| 20 | return SpanContext() |
| 21 | |
| 22 | monkeypatch.setattr("app.audit.trace.get_current_span", lambda: Span()) |
| 23 | logger = StructuredMcpAuditLogger(logging.getLogger("mcp-audit-test"), environment="TEST") |
| 24 | with caplog.at_level(logging.INFO, logger="mcp-audit-test"): |
| 25 | logger.emit( |
| 26 | "MCP_TOOL_INVOKED", |
| 27 | tool="get_company_analysis", |
| 28 | request_id="audit-5a", |
| 29 | auth=auth, |
| 30 | risk_class=McpRiskClass.SAFE_READ, |
| 31 | kind=McpToolKind.INTERNAL, |
| 32 | details={ |
| 33 | "quantity": 100, |
| 34 | "averageCost": 5, |
| 35 | "token": "Bearer abc.def.ghi", |
| 36 | "safeCode": "COMPANY_READ", |
| 37 | }, |
| 38 | ) |
| 39 | record = json.loads(caplog.records[-1].message) |
| 40 | assert record["event"] == "MCP_TOOL_INVOKED" |
| 41 | assert record["requestId"] == "audit-5a" |
| 42 | assert record["traceId"] == "00000000000000000000000000001234" |
| 43 | assert record["details"] == {"safeCode": "COMPANY_READ"} |
| 44 | assert "quantity" not in caplog.records[-1].message |
| 45 | assert "averageCost" not in caplog.records[-1].message |