Auditor

The Auditor provides tamper-proof, append-only audit trails for all enforcement actions using SHA-256 Merkle chains. Every policy evaluation is recorded with cryptographic integrity guarantees, enabling compliance verification and forensic analysis.

Quick Start

from enforcecore import Auditor, verify_trail

# Create an auditor
auditor = Auditor(output_path="./audit_logs/trail.jsonl")

# Record an event
entry = auditor.record(
    tool_name="web_search",
    policy_name="production",
    decision="allowed",
)
print(entry.entry_hash)  # "sha256:a3f2c..."

# Verify trail integrity
result = verify_trail("./audit_logs/trail.jsonl")
print(result.is_valid)       # True
print(result.total_entries)  # 1
Info

Note: When using the @enforce decorator, audit recording happens automatically. You only need the Auditor class directly for custom pipelines or manual audit operations.

Auditor Class

Constructor

Auditor(
    output_path: str | Path = "./audit_logs/trail.jsonl",
    witness: WitnessBackend | None = None,
    immutable: bool = False,
    backend: AuditBackend | None = None,
)
Parameter Type Default Description
output_path str | Path ./audit_logs/trail.jsonl Path to the JSONL audit file. Created automatically.
witness WitnessBackend | None None External witness for tamper detection.
immutable bool False Set OS-level append-only flag (chattr +a on Linux, chflags uappend on macOS).
backend AuditBackend | None None Custom storage backend. Default: JsonlBackend.

Methods

record(**kwargs) -> AuditEntry

Records an enforcement event to the audit trail. Each entry is hash-chained to the previous one.

entry = auditor.record(
    tool_name="execute_shell",
    policy_name="production",
    policy_version="1.0",
    decision="blocked",
    violation_type="tool_denied",
    violation_reason="Action 'execute_shell' is in the deny list",
    overhead_ms=0.056,
    call_duration_ms=0.0,
    input_redactions=2,
    output_redactions=0,
)
Parameter Type Default Description
tool_name str required The tool/function that was evaluated.
policy_name str required Name of the policy applied.
policy_version str "" Version of the policy.
decision str "allowed" Decision: "allowed" or "blocked".
call_id str "" Unique call identifier for correlation.
violation_type str | None None Type of violation (if blocked).
violation_reason str | None None Human-readable violation reason.
overhead_ms float 0.0 Enforcement overhead in milliseconds.
call_duration_ms float 0.0 Total call duration in milliseconds.
input_redactions int 0 Number of PII entities redacted from input.
output_redactions int 0 Number of PII entities redacted from output.

Returns: AuditEntry with computed entry_hash and previous_hash.

close()

Closes the auditor and flushes the backend.

auditor.close()

Standalone Functions

verify_trail(path) -> VerificationResult

Verifies the integrity of an audit trail by recomputing the Merkle chain.

from enforcecore import verify_trail

result = verify_trail("./audit_logs/trail.jsonl")
if result.is_valid:
    print(f"Verified: {result.total_entries} entries, chain intact")
else:
    for error in result.errors:
        print(f"ERROR: {error}")

load_trail(path) -> list[AuditEntry]

Loads all entries from an audit file for inspection.

from enforcecore import load_trail

entries = load_trail("./audit_logs/trail.jsonl")
for entry in entries:
    print(f"{entry.timestamp} | {entry.tool_name} | {entry.decision}")

Data Structures

AuditEntry

A single entry in the audit trail with Merkle chain info.

Attribute Type Description
entry_id str Unique UUID.
call_id str Correlation ID for the call.
timestamp str ISO 8601 timestamp (UTC).
tool_name str The tool evaluated.
policy_name str Policy that was applied.
policy_version str Policy version.
decision str "allowed" or "blocked".
violation_type str | None Violation category.
violation_reason str | None Violation description.
overhead_ms float Enforcement overhead (ms).
call_duration_ms float Total call time (ms).
input_redactions int PII entities redacted from input.
output_redactions int PII entities redacted from output.
previous_hash str SHA-256 hash of the previous entry (chain link).
entry_hash str SHA-256 hash of this entry.

Key methods:

entry.compute_hash()    # Compute SHA-256 hash
entry.seal()            # Compute and set entry_hash
entry.to_dict()         # Serialize to dict
entry.to_json()         # Serialize to JSON string
AuditEntry.from_dict(d) # Deserialize from dict

VerificationResult

Returned by verify_trail().

Attribute Type Description
is_valid bool True if the chain is intact.
total_entries int Number of entries verified.
chain_intact bool Whether the hash chain is unbroken.
errors list[str] List of integrity error messages.
witness_verified bool Whether external witness validated.

Backends

EnforceCore supports pluggable audit backends:

Backend Description
JsonlBackend Default. Append-only JSONL files. Thread-safe.
NullBackend Discards entries. For testing/benchmarking.
CallbackBackend Calls a custom function for each entry.
MultiBackend Multiplexes entries to multiple backends simultaneously.
from enforcecore.auditor.backends import MultiBackend, JsonlBackend, CallbackBackend

# Write to file AND send to external system
backend = MultiBackend([
    JsonlBackend("./audit_logs/trail.jsonl"),
    CallbackBackend(lambda entry: send_to_siem(entry)),
])

auditor = Auditor(backend=backend)

Witness System

Witnesses provide external tamper detection by independently recording hashes:

from enforcecore.auditor.witness import FileWitness

witness = FileWitness("/var/log/ec-witness.jsonl")
auditor = Auditor(
    output_path="./audit_logs/trail.jsonl",
    witness=witness,
)

Audit Rotation

Configure automatic log rotation via environment variables:

Variable Default Description
ENFORCECORE_AUDIT_ROTATE_MB 100 Rotate when file exceeds this size.
ENFORCECORE_AUDIT_RETAIN_DAYS 90 Delete rotated files after this many days.
ENFORCECORE_AUDIT_COMPRESS true Compress rotated files with gzip.

Log Format

Entries are stored as newline-delimited JSON (.jsonl):

{"entry_id":"550e8400-...","previous_hash":"sha256:0000...","entry_hash":"sha256:a3f2c...","timestamp":"2026-02-25T10:30:00Z","tool_name":"web_search","policy_name":"production","decision":"allowed","overhead_ms":0.056}

CLI Commands

# Verify audit trail integrity
enforcecore audit verify ./audit_logs/trail.jsonl

# Verify with external witness
enforcecore audit verify-witness ./audit_logs/trail.jsonl ./witness.jsonl

See Also

  • Enforcer API — The enforcement engine that generates audit events.
  • Security Model — How the audit trail fits into the security architecture.
  • GDPR Compliance — How audit trails support Article 30 compliance.