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) # 1Note: 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 dictVerificationResult
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.jsonlSee 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.