activity-core/src/activity_core/audit_projection.py
tegwick 26934e25b9
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Container Image / build-and-push (push) Successful in 21s
Enforce bounded operation guardrails
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f
2026-08-23 12:31:13 +02:00

68 lines
1.8 KiB
Python

"""Bounded, non-secret projections for durable audit surfaces."""
from __future__ import annotations
from typing import Any
_SENSITIVE_KEYS = frozenset(
{
"api_key",
"api_token",
"access_token",
"refresh_token",
"authorization",
"client_secret",
"cookie",
"credential",
"credentials",
"messages",
"model_messages",
"password",
"private_key",
"provider_payload",
"provider_response",
"raw_output",
"raw_output_preview",
"raw_prompt",
"rendered_prompt",
"secret",
"token",
"tool_error",
"tool_output",
}
)
def bounded_audit_projection(
value: Any,
*,
max_depth: int = 8,
max_items: int = 100,
max_string: int = 4000,
) -> Any:
"""Return a JSON-like audit projection without raw or credential fields.
The projection is intentionally lossy. Workflow evaluation uses the full
in-memory context; only the durable audit copy is bounded here.
"""
def project(item: Any, depth: int) -> Any:
if depth > max_depth:
return "<depth-limit>"
if item is None or isinstance(item, (bool, int, float)):
return item
if isinstance(item, str):
return item[:max_string]
if isinstance(item, dict):
result: dict[str, Any] = {}
for raw_key, child in list(item.items())[:max_items]:
key = str(raw_key)
if key.strip().lower() in _SENSITIVE_KEYS:
continue
result[key] = project(child, depth + 1)
return result
if isinstance(item, (list, tuple)):
return [project(child, depth + 1) for child in item[:max_items]]
return str(item)[:max_string]
return project(value, 0)