01b3f27b0f
Some checks failed
CI / lint (pull_request) Failing after 44s
CI / test (3.10) (pull_request) Failing after 30s
CI / test (3.11) (pull_request) Failing after 33s
CI / test (3.12) (pull_request) Successful in 1m26s
CI / migrations (pull_request) Successful in 24s
CI / helm (pull_request) Successful in 20s
CI / docker (pull_request) Has been skipped
Frontend wiring: - Wire useMarkdownWorker into Markdown component (worker-first, sync fallback) - Wire useIndexedDB as primary storage in useChatHistory (500 msg cap, localStorage fallback) Backend depth: - Persistent audit store (SQLite, thread-safe, WAL mode) with record/query/filter - Wire audit store into session routes (session.create, prompt.submit events) - Wire audit store into audit export routes (persistent-first, telemetry fallback) - CSRF double-submit cookie pattern (token generation, cookie set, header validation) Production: - Helm chart CI: helm lint + helm template validation - Database migration CI: verify step in pipeline - Prometheus alerting rules (error rate, latency, pod restarts, memory, CPU, queue, health) - Rate limiting per API key (3x IP limit, sliding window, advisory) - WebSocket SSE fallback (auto-downgrade after MAX_RETRIES WS failures) Tests: 605 Python + 56 frontend = 661 total, 0 ruff errors Co-Authored-By: Nakamoto, S <defi@defi-oracle.io>
119 lines
3.2 KiB
Python
119 lines
3.2 KiB
Python
"""Audit log export endpoint.
|
|
|
|
Exports governance audit trail as CSV or JSON for compliance and review.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Query
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from fusionagi._logger import logger
|
|
from fusionagi.api.audit_store import get_audit_events
|
|
from fusionagi.api.dependencies import get_telemetry_tracer
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _get_audit_records(
|
|
task_id: str | None = None,
|
|
limit: int = 1000,
|
|
since: float | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Collect audit records from persistent store, falling back to telemetry tracer."""
|
|
# Try persistent audit store first
|
|
try:
|
|
records = get_audit_events(limit=limit, since=since)
|
|
if records:
|
|
return records
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback to telemetry tracer
|
|
tracer = get_telemetry_tracer()
|
|
if not tracer:
|
|
return []
|
|
|
|
traces = tracer.get_traces(task_id=task_id, limit=limit)
|
|
if since:
|
|
traces = [t for t in traces if t.get("timestamp", 0) >= since]
|
|
return traces
|
|
|
|
|
|
@router.get("/audit/export/json")
|
|
def export_audit_json(
|
|
task_id: str | None = None,
|
|
limit: int = Query(default=1000, le=10000),
|
|
since: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Export audit log as JSON.
|
|
|
|
Args:
|
|
task_id: Filter by task ID.
|
|
limit: Maximum records (default 1000, max 10000).
|
|
since: Unix timestamp filter (records after this time).
|
|
|
|
Returns:
|
|
Dict with records array and metadata.
|
|
"""
|
|
records = _get_audit_records(task_id=task_id, limit=limit, since=since)
|
|
logger.info("Audit log exported (JSON)", extra={"count": len(records)})
|
|
return {
|
|
"format": "json",
|
|
"count": len(records),
|
|
"exported_at": time.time(),
|
|
"records": records,
|
|
}
|
|
|
|
|
|
@router.get("/audit/export/csv")
|
|
def export_audit_csv(
|
|
task_id: str | None = None,
|
|
limit: int = Query(default=1000, le=10000),
|
|
since: float | None = None,
|
|
) -> StreamingResponse:
|
|
"""Export audit log as CSV download.
|
|
|
|
Args:
|
|
task_id: Filter by task ID.
|
|
limit: Maximum records (default 1000, max 10000).
|
|
since: Unix timestamp filter (records after this time).
|
|
|
|
Returns:
|
|
CSV file as streaming download.
|
|
"""
|
|
records = _get_audit_records(task_id=task_id, limit=limit, since=since)
|
|
|
|
# Collect all unique keys across records
|
|
all_keys: set[str] = set()
|
|
for r in records:
|
|
all_keys.update(r.keys())
|
|
fieldnames = sorted(all_keys)
|
|
|
|
output = io.StringIO()
|
|
writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore")
|
|
writer.writeheader()
|
|
for r in records:
|
|
# Flatten nested dicts to JSON strings
|
|
flat = {}
|
|
for k, v in r.items():
|
|
flat[k] = json.dumps(v) if isinstance(v, (dict, list)) else v
|
|
writer.writerow(flat)
|
|
|
|
output.seek(0)
|
|
logger.info("Audit log exported (CSV)", extra={"count": len(records)})
|
|
|
|
return StreamingResponse(
|
|
iter([output.getvalue()]),
|
|
media_type="text/csv",
|
|
headers={
|
|
"Content-Disposition": f"attachment; filename=fusionagi_audit_{int(time.time())}.csv",
|
|
},
|
|
)
|