Skip to main content

Trailproof Class

The Trailproof class is the main entry point. It manages the hash chain, store, and optional signer.

Constructor

tp = Trailproof(
    store="memory",              # "memory" (default) or "jsonl"
    path=None,                   # file path (required for jsonl store)
    signing_key=None,            # HMAC-SHA256 key (optional)
    default_tenant_id=None,      # applied to every event if not specified
)

emit()

Record a new event in the audit trail.
event = tp.emit(
    event_type="myapp.user.login",     # required
    actor_id="user-42",                 # required
    tenant_id="acme-corp",              # required (or use default_tenant_id)
    payload={"ip": "1.2.3.4"},          # required
    trace_id="trace-abc",               # optional
    session_id="session-xyz",           # optional
)
Returns: TrailEvent — the complete event with auto-generated fields. Throws: ValidationError if required fields are missing or empty. Behavior:
  • Auto-generates event_id (UUID v4) and timestamp (ISO-8601 UTC)
  • Computes hash using the hash chain engine
  • Sets prev_hash to the previous event’s hash (or genesis hash for the first event)
  • If a signing key is configured, computes and sets signature
  • Appends the event to the store

query()

Search events with filters and pagination.
result = tp.query(
    event_type="myapp.user.login",       # optional
    actor_id="user-42",                   # optional
    tenant_id="acme-corp",                # optional
    trace_id="trace-abc",                 # optional
    session_id="session-xyz",             # optional
    from_time="2025-01-01T00:00:00Z",    # optional
    to_time="2025-12-31T23:59:59Z",      # optional
    limit=100,                            # default 100
    cursor=None,                          # for pagination
)

result.events       # list[TrailEvent]
result.next_cursor  # str | None
Returns: QueryResult { events, next_cursor }. All filters are optional. No filters returns all events up to limit. Filters use exact match except from_time and to_time which are range filters.

verify()

Walk the hash chain and check every event’s hash.
result = tp.verify()

result.intact   # bool -- True if no tampering
result.total    # int -- number of events checked
result.broken   # list[int] -- indices of broken events
Returns: VerifyResult { intact, total, broken }. Empty chain returns { intact: true, total: 0, broken: [] }.
Verification does not throw on broken chains — it returns the result. Check result.intact to determine if the chain is valid.

get_trace() / getTrace()

Get all events for a specific trace ID, ordered by timestamp.
events = tp.get_trace("trace-abc")
# returns: list[TrailEvent]
Returns: List of TrailEvent objects matching the trace ID, ordered by timestamp.

flush()

Ensure all buffered events are persisted to disk.
tp.flush()
No-op for the memory store. For the JSONL store, ensures all buffered writes are flushed to disk.