Skip to content
Packages Examples Agents Blog Get started

API Reference

Classes

Section titled “Classes”

Admin panel contributor for audit log management.

Registered via oridecon.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) -> None

Resolve audit dependencies from the DI container.

Parameters
ParameterTypeDescription
`container`ContainerResolverProtocol | NoneThe DI container resolver.
get_navigation_items
def get_navigation_items() -> Sequence[NavigationContribution]

Return the navigation items for this contributor.

get_management_pages
def get_management_pages() -> Sequence[ManagementPageDefinition]

Return the management page definitions for this contributor.

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 Oridecon audit stack.

Composes AuditCoreProvider, AuditRetentionProvider, AuditVerifierProvider, AuditSchedulingProvider, and optionally AuditAdminProvider.

Parameters
ParameterTypeDescription
`config`Audit configuration. When ``None``, the orchestrator injects the typed ``audit`` yaml section after construction and sub-providers are composed in register.
`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.

Late config binding: the orchestrator injects the typed audit section (via config_key) after construction and before this call. If configure() ran with no explicit config, compose now so the automatic path behaves identically to the explicit one.

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.

Oridecon audit module.

Global like EventsModule/QueueModule so any consumer module (e.g. an app’s infrastructure provider) can resolve audit protocols without declaring explicit imports.

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,
))

Or with an explicit AuditConfig section

app.use(AuditModule.configure(config=AuditConfig(store_backend="memory")))
app.use(AuditModule.configure(config=AuditConfig(store_backend="memory")))
configure
def configure(
    cls,
    config: AuditConfig | None = None,
    *,
    hmac_key: bytes | None = None,
    store_backend: str | None = None,
    table_name: str | None = None,
    retention_days: int | None = None,
    enable_admin: bool = True,
    **overrides: Any
) -> DynamicModule

Configure the audit module.

Parameters
ParameterTypeDescription
`config`AuditConfig | NoneExplicit AuditConfig section. When provided it wins over the keyword shortcuts below.
`hmac_key`bytes | NoneHMAC key for checksum computation.
`store_backend`str | None``"sql"`` or ``"memory"``.
`table_name`str | NoneSQL table name.
`retention_days`int | NoneDefault retention in days.
`enable_admin`boolRegister admin contributor. **overrides: Additional AuditConfig fields.
Returns
TypeDescription
DynamicModuleDynamicModule ready for ``app.use()``.

Note

Called with no arguments, the module passes None through so the orchestrator injects the typed audit yaml section before registration (framework defaults apply when no section exists).

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(dry_run: bool = False) -> int

Evaluate all entries and purge those past expiry.

Parameters
ParameterTypeDescription
`dry_run`boolWhen True, only count entries that would be purged without deleting anything. Defaults to False.
Returns
TypeDescription
intNumber of entries purged (or that would be purged in dry-run).

Tamper detection via HMAC-SHA256 checksum verification.

Implements AuditVerifierProtocol. Entries are verified by recomputing the HMAC over the canonical persisted row (entry_to_row) and comparing against the checksum the store read back. Entries written before checksums existed carry no stored checksum and are reported honestly as unverifiable (no_checksum_present) — never silently clean, never falsely tampered.

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.

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: AuditEntry) -> AuditMismatch | None

Verify checksum for a single entry.

Recomputed the HMAC over the canonical persisted row and compares it against the stored checksum. Older entries may carry v1 checksums while new entries carry v2; both are accepted. Entries with no stored checksum are reported as unverifiable.

Parameters
ParameterTypeDescription
`entry`AuditEntryThe audit entry to verify.
Returns
TypeDescription
AuditMismatch | NoneNone when the entry verifies clean; an AuditMismatch whose reason is ``checksum_mismatch`` when tampered or ``no_checksum_present`` when the entry carries no stored checksum and cannot be verified.

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.

delete_expired
async def delete_expired(cutoff: datetime) -> int

Delete entries whose stored expiry precedes or equals cutoff.

Mirrors the SQL store: only entries stamped with the __expires_at metadata key are candidates for deletion.

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”

audited
def audited(
    action: str,
    *,
    resource_type: str = '',
    severity: str = 'medium'
) -> Callable[[F], F]
Mark an async function for automatic audit logging.

Attaches audit metadata that an audit middleware or interceptor can read to automatically log an AuditEntry on successful execution.

Parameters
ParameterTypeDescription
`action`strDot-notation action identifier (e.g. ``"user.update"``).
`resource_type`strKind of affected resource (e.g. ``"User"``).
`severity`strDefault severity level for the audit entry.
Returns
TypeDescription
Callable[[F], F]Decorator that attaches audit metadata to the function.

Example

@audited("user.update", resource_type="User", severity="medium")
async def update_user(self, user_id: str, data: dict) -> User:
    ...
@audited("user.update", resource_type="User", severity="medium")
async def update_user(self, user_id: str, data: dict) -> User:
    ...

compute_audit_checksum
def compute_audit_checksum(
    entry_data: dict[str, Any],
    key: bytes,
    schema_version: int = 2
) -> str
Compute HMAC-SHA256 hex digest for audit entry data.

Includes entry_schema_version in the canonical form so the verifier knows which fields were expected at write time.

Parameters
ParameterTypeDescription
`entry_data`dict[str, Any]Dictionary of audit entry fields.
`key`bytesHMAC secret key bytes.
`schema_version`intSchema version to embed (default 2). Existing entries use 1; new entries use 2.
Returns
TypeDescription
strHex-encoded HMAC-SHA256 digest.

verify_audit_checksum
def verify_audit_checksum(
    entry_data: dict[str, Any],
    key: bytes,
    expected: str,
    schema_version: int | None = None
) -> bool
Verify expected checksum matches computed checksum using constant-time comparison.

When schema_version is None, the version is extracted from entry_data (defaults to 1 if absent) so old entries verify against v1 checksums and new entries verify against v2.

Parameters
ParameterTypeDescription
`entry_data`dict[str, Any]Dictionary of audit entry fields.
`key`bytesHMAC secret key bytes.
`expected`strPreviously stored checksum hex string.
`schema_version`int | NoneExplicit schema version override. When ``None``, extracted from entry_data or defaults to 1.
Returns
TypeDescription
boolTrue if checksum matches, False if tampered.