audit

audit

Append-only JSONL audit log for credential-bearing broker requests.

Records one line per phantom-token-validated proxy call so the operator can answer "what did this subject's credentials get used for?" after the fact. The schema is deliberately small — caller-supplied labels (scope, subject, credential_set, provider) plus request shape (method, path, status, outcome, duration_ms). No request bodies, no response bodies; the broker is a transparent proxy and audit's job is forensic context, not deep packet capture.

One JSONL file under the vault state dir, shared across every subject the broker has ever served. The sandbox makes no per-subject layout decisions because it doesn't model "subject" as anything other than an opaque label — read-side filtering is the orchestrator's job.

Soft-fail semantics throughout: a missing parent dir, a full disk, an EACCES on the file all degrade gracefully to a single WARNING log line. The proxy's primary job is forwarding, not auditing; an audit write must never block or kill a credential-bearing request.

AuditWriter(path)

Append-only JSONL writer with one process-wide async lock per file.

Holds a single line-buffered append handle for the lifetime of the broker — opening per request would dominate the sub-millisecond proxy hot path. Concurrent aiohttp handlers serialise through an asyncio.Lock so their JSONL bytes can't interleave.

Parameters:

Name	Type	Description	Default
path	Path	Where to write. Parent directories and the file itself are created lazily on first write so the broker can be started against a vault dir that doesn't exist yet (fresh installs, tests using tmp_path).	required

Bind the writer to path without touching the filesystem.

Source code in src/terok_sandbox/vault/daemon/audit.py

def __init__(self, path: Path) -> None:
    """Bind the writer to *path* without touching the filesystem."""
    self._path = path
    self._lock = asyncio.Lock()
    self._fh: TextIOBase | None = None
    self._open_failed = False

write(entry) async

Append one JSON-encoded line. Soft-fail on every error path.

Marshals entry with compact separators (no whitespace) and writes json + "\n" under the per-writer lock. Line-buffered text mode flushes after every newline, so a crash mid-broker loses at most the most recent partial line.

The actual file write goes through asyncio.to_thread so a slow disk / stuck mount can't stall the broker's event loop — the write happens on the default executor, the lock keeps ordering across concurrent callers, and the soft-fail OSError path stays the same shape as the sync version.

Serialization failures (a non-JSON-serializable field slipping into entry) are also soft-failed — a malformed audit entry must not kill a credential-bearing request.

Source code in src/terok_sandbox/vault/daemon/audit.py

async def write(self, entry: dict[str, Any]) -> None:
    """Append one JSON-encoded line.  Soft-fail on every error path.

    Marshals *entry* with compact separators (no whitespace) and
    writes ``json + "\\n"`` under the per-writer lock.  Line-buffered
    text mode flushes after every newline, so a crash mid-broker
    loses at most the most recent partial line.

    The actual file write goes through ``asyncio.to_thread`` so a
    slow disk / stuck mount can't stall the broker's event loop —
    the write happens on the default executor, the lock keeps
    ordering across concurrent callers, and the soft-fail
    ``OSError`` path stays the same shape as the sync version.

    Serialization failures (a non-JSON-serializable field slipping
    into *entry*) are also soft-failed — a malformed audit entry
    must not kill a credential-bearing request.
    """
    try:
        line = json.dumps(entry, separators=(",", ":")) + "\n"
    except (TypeError, ValueError) as exc:
        _logger.warning(
            "Credential audit entry not JSON-serializable (%s): %s", self._path, exc
        )
        return
    async with self._lock:
        if self._fh is None and not self._open_failed:
            self._fh = self._lazy_open()
        if self._fh is None:
            return
        try:
            await asyncio.to_thread(self._fh.write, line)
        except OSError as exc:
            _logger.warning("Credential audit write failed (%s): %s", self._path, exc)

close() async

Close the underlying handle if open. Idempotent.

Source code in src/terok_sandbox/vault/daemon/audit.py

async def close(self) -> None:
    """Close the underlying handle if open.  Idempotent."""
    async with self._lock:
        if self._fh is not None:
            try:
                self._fh.close()
            finally:
                self._fh = None

credential_audit_log_path(vault_root)

Return the canonical credential-audit JSONL path under vault_root.

Source code in src/terok_sandbox/vault/daemon/audit.py

def credential_audit_log_path(vault_root: Path) -> Path:
    """Return the canonical credential-audit JSONL path under *vault_root*."""
    return vault_root / _AUDIT_BASENAME