bh-audit-logger

Cloud-agnostic Python utilities for emitting privacy-preserving audit events for behavioral healthcare systems.

Events conform to bh-audit-schema v2.0: https://github.com/bh-healthcare/bh-audit-schema

Why

Audit logging in healthcare is often inconsistent across services and jobs. This library provides a small, boring, correct baseline for emitting structured audit events from any Python code — Lambdas, workers, CLIs, ETL jobs, cron scripts — without logging raw PHI.

It is not tied to FastAPI (see bh-fastapi-audit for middleware-based logging).

Install

pip install bh-audit-logger               # core (zero dependencies)
pip install bh-audit-logger[dynamodb]      # + DynamoDB sink (boto3)
pip install bh-audit-logger[cli]           # + bh-audit verify CLI (typer)
pip install bh-audit-logger[jsonschema]    # + runtime schema validation
pip install bh-audit-logger[all]           # everything

Quickstart

pip install bh-audit-logger

from bh_audit_logger import AuditLogger, AuditLoggerConfig

logger = AuditLogger(
    config=AuditLoggerConfig(
        service_name="sample-datalake",
        service_environment="prod",
    )
)

logger.audit(
    "READ",
    actor={"subject_id": "service_lambda", "subject_type": "service"},
    resource={"type": "Patient", "id": "patient_123"},
    outcome={"status": "SUCCESS"},
    correlation={"request_id": "req_abc"},
)

By default, events are emitted as one compact JSON line via Python logging (stdout-friendly).

Example output

{"schema_version":"2.0","event_id":"6d3f0f6b-0c1a-4b9f-9d6f-9f6f7f5b2b0a","timestamp":"2026-03-28T12:00:00.000Z","service":{"name":"sample-datalake","environment":"prod"},"actor":{"subject_id":"service_lambda","subject_type":"service"},"action":{"type":"READ","data_classification":"UNKNOWN"},"resource":{"type":"Patient","id":"patient_123"},"outcome":{"status":"SUCCESS"},"correlation":{"request_id":"req_abc"}}

Production usage: container logging

from bh_audit_logger import AuditLogger, AuditLoggerConfig, LoggingSink

logger = AuditLogger(
    config=AuditLoggerConfig(
        service_name="my-service",
        service_environment="prod",
    ),
    sink=LoggingSink(logger_name="bh.audit", level="INFO"),
)

Works anywhere stdout is collected: CloudWatch, GCP Cloud Logging, Azure Monitor, Kubernetes logging pipelines.

Production hardening

Frozen config

AuditLoggerConfig is frozen after creation (@dataclass(frozen=True)) to prevent runtime mutation of security settings:

config = AuditLoggerConfig(
    service_name="my-service",
    metadata_allowlist=frozenset({"batch_id", "region"}),
)
config.sanitize_errors = False  # raises AttributeError

Sink failure isolation

By default, sink failures are logged but never propagate to your application logic:

config = AuditLoggerConfig(
    service_name="my-service",
    emit_failure_mode="log",       # "silent", "log" (default), or "raise"
    failure_logger_name="bh.audit.internal",
)

Metadata restrictions

Metadata values are enforced to be scalar JSON types (str, int, float, bool, None). Dict, list, and tuple values are silently dropped. Long strings are truncated:

config = AuditLoggerConfig(
    service_name="my-service",
    metadata_allowlist=frozenset({"batch_id", "region"}),
    max_metadata_value_length=200,
)

Internal counters

Track emission health via lightweight counters:

logger = AuditLogger(config=config)
# ... emit events ...
print(logger.stats.snapshot())
# {"events_emitted_total": 42, "emit_failures_total": 0, "events_dropped_total": 0,
#  "validation_failures_total": 0, "validation_time_ms_total": 0.0}

Non-blocking async emission (optional)

v0.3 adds EmitQueue for async emission from async contexts:

from bh_audit_logger import EmitQueue

queue = EmitQueue(sink, stats, maxsize=5000)
queue.start()
queue.enqueue(event)
# ... later ...
await queue.shutdown()

Runtime schema validation

v0.4 adds optional runtime validation of emitted events against the vendored JSON schema. This catches schema-invalid events before they reach your sink.

pip install bh-audit-logger[jsonschema]

from bh_audit_logger import AuditLogger, AuditLoggerConfig

logger = AuditLogger(
    config=AuditLoggerConfig(
        service_name="my-service",
        validate_events=True,                    # enable runtime validation
        validation_failure_mode="drop",          # "drop" (default), "log_and_emit", or "raise"
        target_schema_version="2.0",             # "1.0", "1.1", or "2.0" (default)
    )
)

Validation timing

Validation adds measurable latency. Track it via stats:

stats = logger.stats.snapshot()
print(stats["validation_time_ms_total"])  # cumulative ms spent in schema validation

DENIED outcomes

v0.4 adds audit_access_denied() for authorization denials (distinct from operational failures):

logger.audit_access_denied(
    "READ",
    error_type="RoleDenied",
    error_message="Role 'viewer' lacks access to ClinicalNote",
    actor={"subject_id": "user-42", "subject_type": "human"},
    resource={"type": "ClinicalNote", "id": "note-555"},
)

Cross-org access detection

Use owner_org_id in the actor block to flag cross-organization access attempts:

logger.audit_access_denied(
    "EXPORT",
    error_type="CrossOrgAccessDenied",
    error_message="Actor org-200 cannot export resources owned by org-300",
    actor={
        "subject_id": "user-77",
        "subject_type": "human",
        "org_id": "org-200",
        "owner_org_id": "org-300",
    },
    resource={"type": "PatientRecord"},
)

Schema version negotiation

Target a specific schema version for backward compatibility:

config = AuditLoggerConfig(
    service_name="my-service",
    target_schema_version="1.1",  # emit v1.1-compatible events (default is "2.0")
)

When targeting v1.0, DENIED outcomes are automatically downgraded to FAILURE (since v1.0 does not support DENIED). When targeting v1.x with agent/delegation events, set agent_event_downgrade_mode="strip" to collapse to the authorizing human actor, or leave the default "raise" to surface misconfigurations loudly.

Sinks

Pass any sink to AuditLogger(config=..., sink=...). Omit sink to get LoggingSink by default.

For DynamoDBSink production deployment (table creation, IAM, environment configuration), see docs/deploying-dynamodb.md.

Configuration

AuditLoggerConfig fields (frozen after creation):

Typed event blocks

v0.3+ exports TypedDict definitions for all event sub-blocks:

from bh_audit_logger import (
    AuditEvent, ServiceBlock, ActorBlock, ActionBlock,
    ResourceBlock, OutcomeBlock, CorrelationBlock,
    ActionType, ActorType, OutcomeStatus, DataClassification,
)

PHI-safe by default

• No request/response bodies — the library never tries to capture payloads
• Metadata is opt-in and strictly allowlisted — only keys in metadata_allowlist pass through; values must be scalar JSON types
• Error messages are sanitized — SSN, email, phone patterns are redacted and messages are length-capped
• PHI safety is enforced by tests that assert synthetic PHI tokens never appear in emitted events

Agent attribution (v2.0)

v2.0 adds first-class support for AI agent audit trails. Use audit_agent_action() to emit a delegated-action event with a full RFC 0001 delegation block:

logger.audit_agent_action(
    "READ",
    agent_subject_id="agent_cred_009",
    agent={"interface": "mcp", "vendor": "acme", "model_family": "model_x", "model_version": "x-3.1"},
    authorizing_subject_id="user_123",
    agent_session_id="agsess_01J9X0001",
    delegation_type="supervised",
    resource={"type": "Patient", "id": "pat_456", "patient_id": "pat_456"},
    phi_touched=True,
    data_classification="PHI",
    chain_depth=1,
)

When the authenticating credential and the acting agent instance are distinct identities (RFC 0001 three-role model), pass acting_subject_id= to record the instance in delegation.acting.subject_id while actor.subject_id keeps the credential. The agent descriptor is strictly validated: only the six schema keys are accepted (interface, vendor, model_family, model_version, harness, harness_version), string values only, 128 chars max — free-text prompt content can never ride along.

To record when a human cancels an agent session — action_name carries the override category (RFC 0001 §7.3); metadata keys must be allowlisted in AuditLoggerConfig(metadata_allowlist={"override_category"}) or they are dropped:

logger.audit_human_override(
    "agsess_01J9X0001",
    actor={"subject_id": "user_123", "subject_type": "human"},
    action_name="cancel_agent_session",
    metadata={"override_category": "user_cancelled"},
)

For session lifecycle bookends:

agent_desc = {"interface": "mcp", "vendor": "acme", "model_family": "model_x"}

logger.audit_agent_session_start(
    agent_subject_id="agent_cred_009",
    agent=agent_desc,
    authorizing_subject_id="user_123",
    agent_session_id="agsess_01J9X0001",
    delegation_type="supervised",
)

# ... agent does work ...

logger.audit_agent_session_end(
    agent_subject_id="agent_cred_009",
    agent=agent_desc,
    authorizing_subject_id="user_123",
    agent_session_id="agsess_01J9X0001",
    delegation_type="supervised",
)

audit() also gained delegation= and action_name= kwargs for callers who build the delegation block manually or need a named action on any event.

See docs/migrating-1.1-to-2.0.md for the full delegation model and DynamoDB query guidance.

Schema conformance

All events conform to bh-audit-schema v2.0 (currently vendoring tag v2.0.0, commit 4b505efcd93e1d5f5001b606f7c4dd3252de7ad2). The v2.0 schema adds:

• "agent" actor type and "OVERRIDE" action type
• delegation block for agent attribution (RFC 0001)
• Three conditional validation rules (agent⇒delegation; chain_depth≥2⟺parent_agent_session_id; OVERRIDE⇒human+AgentSession)
• New types: DelegationBlock, ActingBlock, AuthorizingBlock, AgentDescriptorBlock, DelegationType, AgentInterface

Schema 1.0 and 1.1 remain supported via target_schema_version. See the migration guide for downgrade options.

Optional schema validation

pip install bh-audit-logger[jsonschema]

from bh_audit_logger import validate_event

event = {...}
validate_event(event)  # raises ValidationError on failure

Validates against the vendored bh-audit-schema v2.0 JSON schema included in the package.

Chain hashing (integrity)

v1.0 adds tamper-evident audit trails via SHA-256 chain hashing. Each event gets an integrity block with event_hash, prev_event_hash, and hash_alg:

config = AuditLoggerConfig(
    service_name="my-service",
    enable_integrity=True,       # SHA-256 chain hashing
    hash_algorithm="sha256",     # or "sha384", "sha512"
)
logger = AuditLogger(config=config)

For DynamoDB-backed multi-process chain state:

from bh_audit_logger import DynamoDBChainState

chain_state = DynamoDBChainState(table_name="bh_chain_state", service_name="my-service")
logger = AuditLogger(config=config, chain_state=chain_state)

Verifier CLI

v1.0 adds bh-audit verify for chain integrity verification:

pip install bh-audit-logger[cli]

# Verify a JSONL ledger file
bh-audit verify --source file --path /var/log/audit/events.jsonl

# Verify from DynamoDB
bh-audit verify --source dynamodb --table bh_audit_events --service intake-api

# JSON output for CI pipelines
bh-audit verify --source file --path events.jsonl --format json

Exit codes: 0 = PASS, 1 = FAIL, 2 = ERROR.

Programmatic verification:

from bh_audit_logger import verify_chain

result = verify_chain(events)
assert result.result == "PASS"

Telemetry

v1.0 adds opt-in, privacy-first telemetry. Off by default. No PII, no PHI, no event content -- only aggregate counters.

config = AuditLoggerConfig(
    service_name="my-service",
    telemetry_enabled=True,  # explicit opt-in required
)

See docs/telemetry.md for the full privacy commitment and payload format.

Emission tier (RFC §11)

This library provides instrumented/cooperative-tier emission (SDK instrumentation), as defined in RFC §11. Events are emitted explicitly by instrumented application code, not captured transparently at a network or OS layer.

Related projects

• bh-audit-schema: github.com/bh-healthcare/bh-audit-schema — the schema standard
• bh-fastapi-audit: github.com/bh-healthcare/bh-fastapi-audit — FastAPI middleware for automatic audit logging

License

Apache 2.0