Skip to content
GitHub

API Reference

Classes

Section titled “Classes”

Admin panel contributor for audit log management.

Registered via lexigram.admin.contributors entry point. Provides audit log search, filter, export, and verification status. Dependencies are resolved from the container in on_admin_boot.

__init__
def __init__() -> None
on_admin_boot
async def on_admin_boot(container: ContainerResolverProtocol) -> None

Resolve audit dependencies from the DI container.

Parameters
ParameterTypeDescription
`container`ContainerResolverProtocolThe DI container resolver.
get_menu_items
async def get_menu_items() -> list[dict[str, str]]

Return admin menu items for the audit module.

search
async def search(query: AuditQuery) -> list[AuditEntry]

Search audit entries by query filters.

Parameters
ParameterTypeDescription
`query`AuditQueryFilter criteria.
Returns
TypeDescription
list[AuditEntry]Matching audit entries, newest-first.
export
async def export(
    query: AuditQuery,
    format: str = 'json'
) -> bytes

Export filtered audit entries as JSON or CSV.

Parameters
ParameterTypeDescription
`query`AuditQueryFilter criteria.
`format`str``"json"`` or ``"csv"``.
Returns
TypeDescription
bytesEncoded bytes of the export.
verification_status
async def verification_status() -> dict[str, Any]

Return latest verification results.

Returns
TypeDescription
dict[str, Any]Dict with ``verified`` bool and ``mismatches`` count.

Composite provider that wires the full Lexigram audit stack.

Composes AuditCoreProvider, AuditRetentionProvider, AuditVerifierProvider, and optionally AuditAdminProvider.

Parameters
ParameterTypeDescription
`config`Audit configuration. Defaults to AuditConfig() if not provided.
`enable_admin`Whether to register the admin panel contributor.
__init__
def __init__(
    config: AuditConfig | None = None,
    enable_admin: bool = True
) -> None
register
async def register(container: ContainerRegistrarProtocol) -> None

Delegate registration to all sub-providers.

boot
async def boot(container: BootContainerProtocol) -> None

Delegate boot to all sub-providers.

shutdown
async def shutdown() -> None

Shutdown in reverse registration order.

Configuration for the audit subsystem.

Attributes: store_backend: Backend type — "sql" or "memory". table_name: SQL table name for the unified audit store. hmac_key: HMAC key for checksum computation (bytes). retention_policy: Retention rules; defaults to 365 days. verification_schedule: Cron expression for scheduled verification. verification_batch_size: Entries to verify per verification run. enable_admin: Whether to register the AuditAdminContributor.

Core audit logger implementing AuditLoggerProtocol.

Fire-tolerant: log() catches all exceptions and emits a warning. Audit failure must never block the operation that triggered it.

Parameters
ParameterTypeDescription
`store`Underlying storage backend (AuditStoreProtocol).
`retention`Optional retention policy for computing entry expiry.
__init__
def __init__(
    store: AuditStoreProtocol,
    retention: RetentionPolicyProtocol | None = None
) -> None
log
async def log(entry: AuditEntry) -> None

Record an audit entry. Never raises.

Parameters
ParameterTypeDescription
`entry`AuditEntryThe audit event to persist.
query
async def query(query: AuditQuery) -> list[AuditEntry]

Query entries matching filters. Returns empty list on error.

Parameters
ParameterTypeDescription
`query`AuditQueryFilter criteria encapsulated in an AuditQuery object.
Returns
TypeDescription
list[AuditEntry]List of matching entries, newest-first.

Lexigram audit module.

Registers the full audit stack including store, logger, retention, verification, and optional admin panel contributor.

Usage

app = Application()
app.use(AuditModule.configure(
    hmac_key=b"secret",
    retention_days=365,
))
app = Application()
app.use(AuditModule.configure(
    hmac_key=b"secret",
    retention_days=365,
))
configure
def configure(
    cls,
    *,
    hmac_key: bytes | None = None,
    store_backend: str = 'sql',
    table_name: str = 'audit_log',
    retention_days: int = 365,
    enable_admin: bool = True,
    **overrides: Any
) -> DynamicModule

Configure the audit module.

Parameters
ParameterTypeDescription
`hmac_key`bytes | NoneHMAC key for checksum computation.
`store_backend`str``"sql"`` or ``"memory"``.
`table_name`strSQL table name.
`retention_days`intDefault retention in days.
`enable_admin`boolRegister admin contributor. **overrides: Additional AuditConfig fields.
Returns
TypeDescription
DynamicModuleDynamicModule ready for ``app.use()``.

Purges expired audit entries and emits a meta-audit log on each run.
Parameters
ParameterTypeDescription
`store`The underlying store to purge entries from.
`retention`Retention policy to evaluate entries.
`audit_logger`Optional audit logger for meta-audit events.
__init__
def __init__(
    store: AuditStoreProtocol,
    retention: RetentionPolicyProtocol,
    audit_logger: AuditLoggerProtocol | None = None
) -> None
purge_expired
async def purge_expired() -> int

Evaluate all entries and purge those past expiry.

Returns
TypeDescription
intNumber of entries purged.

Tamper detection via HMAC-SHA256 checksum verification.

Implements AuditVerifierProtocol. Checksum verification is only meaningful when entries are stored with checksums (i.e. the SQL backend with an HMAC key configured). The in-memory store does not persist checksums, so verify_recent and verify_entry are no-ops in that case.

Parameters
ParameterTypeDescription
`store`Audit store backend (any AuditStoreProtocol implementation).
`config`Audit configuration containing the HMAC key.
__init__
def __init__(
    store: AuditStoreProtocol,
    config: AuditConfig
) -> None
verify_recent
async def verify_recent(
    *,
    limit: int = 100
) -> list[AuditMismatch]

Verify checksums for the most recent entries.

Because AuditEntry does not carry a stored checksum field, full tamper-detection is only possible when the SQL backend is in use and checksums are persisted in the DB column. Without stored checksums this method logs the entries and returns an empty list (no mismatches detected / not applicable).

Parameters
ParameterTypeDescription
`limit`intNumber of recent entries to verify.
Returns
TypeDescription
list[AuditMismatch]List of AuditMismatch objects (empty = all verified or not applicable).
verify_entry
async def verify_entry(entry_id: str) -> bool

Verify checksum for a single entry.

This is a no-op at the protocol level because AuditEntry does not carry a stored checksum field. When using the SQL backend with HMAC enabled, verification should be performed directly against the DB row rather than through the store protocol.

Parameters
ParameterTypeDescription
`entry_id`strID of the entry to verify.
Returns
TypeDescription
boolTrue (checksum verification not applicable at protocol level).

In-memory audit store implementing AuditStoreProtocol.

Uses a bounded deque. Thread-safe for single-event-loop async usage.

Parameters
ParameterTypeDescription
`max_entries`Maximum capacity. Oldest entries dropped when full.
__init__
def __init__(max_entries: int = 10000) -> None
append
async def append(entry: AuditEntry) -> None

Persist a single audit entry.

query
async def query(query: AuditQuery) -> list[AuditEntry]

Retrieve entries matching filters, newest-first.

count
async def count(query: AuditQuery) -> int

Return count of entries matching filters.

clear
def clear() -> None

Clear all entries (test helper).

Evaluates retention policies to determine entry expiry.

Implements RetentionPolicyProtocol.

Parameters
ParameterTypeDescription
`policy`Retention policy configuration.
__init__
def __init__(policy: RetentionPolicy) -> None
evaluate
async def evaluate(entry: AuditEntry) -> RetentionDecision

Determine retention decision based on severity and source.

Parameters
ParameterTypeDescription
`entry`AuditEntryThe audit entry to evaluate.
Returns
TypeDescription
RetentionDecisionRetentionDecision.RETAIN if indefinite, RETAIN_UNTIL otherwise.
get_expiry
async def get_expiry(entry: AuditEntry) -> datetime | None

Return expiry datetime for an entry, or None for indefinite retention.

Parameters
ParameterTypeDescription
`entry`AuditEntryThe audit entry to evaluate.
Returns
TypeDescription
datetime | NoneUTC expiry datetime, or None.

Functions

Section titled “Functions”

compute_audit_checksum
def compute_audit_checksum(
    entry_data: dict[str, Any],
    key: bytes
) -> str
Compute HMAC-SHA256 hex digest for audit entry data.
Parameters
ParameterTypeDescription
`entry_data`dict[str, Any]Dictionary of audit entry fields.
`key`bytesHMAC secret key bytes.
Returns
TypeDescription
strHex-encoded HMAC-SHA256 digest.

verify_audit_checksum
def verify_audit_checksum(
    entry_data: dict[str, Any],
    key: bytes,
    expected: str
) -> bool
Verify expected checksum matches computed checksum using constant-time comparison.
Parameters
ParameterTypeDescription
`entry_data`dict[str, Any]Dictionary of audit entry fields.
`key`bytesHMAC secret key bytes.
`expected`strPreviously stored checksum hex string.
Returns
TypeDescription
boolTrue if checksum matches, False if tampered.