Skip to content
Lexigram
GitHubDiscord

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.

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

Resolve audit dependencies from the DI container.

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

Return admin menu items for the audit module.

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.
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.
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.
def __init__(
    config: AuditConfig | None = None,
    enable_admin: bool = True
) -> None
async def register(container: ContainerRegistrarProtocol) -> None

Delegate registration to all sub-providers.

async def boot(container: ContainerResolverProtocol) -> None

Delegate boot to all sub-providers.

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.
def __init__(
    store: AuditStoreProtocol,
    retention: RetentionPolicyProtocol | None = None
) -> None
async def log(entry: AuditEntry) -> None

Record an audit entry. Never raises.

Parameters
ParameterTypeDescription
`entry`AuditEntryThe audit event to persist.
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,
))
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.
def __init__(
    store: AuditStoreProtocol,
    retention: RetentionPolicyProtocol,
    audit_logger: AuditLoggerProtocol | None = None
) -> None
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.
def __init__(
    store: AuditStoreProtocol,
    config: AuditConfig
) -> None
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).
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.
def __init__(max_entries: int = 10000) -> None
async def append(entry: AuditEntry) -> None

Persist a single audit entry.

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

Retrieve entries matching filters, newest-first.

async def count(query: AuditQuery) -> int

Return count of entries matching filters.

def clear() -> None

Clear all entries (test helper).

Evaluates retention policies to determine entry expiry.

Implements RetentionPolicyProtocol.

Parameters
ParameterTypeDescription
`policy`Retention policy configuration.
def __init__(policy: RetentionPolicy) -> None
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.
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”

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.

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.