jeevesagent.security.audit

Append-only audit log.

Every meaningful event in the loop — run start/finish, tool dispatch, tool result, permission decision — gets a signed entry on the audit log. Two backends ship:

Both compute an HMAC-SHA256 signature over a canonicalised representation of the entry’s content fields, keyed by a per-log secret. The signature lets compliance tooling detect tampering; verify_signature() recomputes it and compares.

The log is conceptually monotonic: seq is per-log and never re-used. FileAuditLog recovers the highest seq from the file on startup so multiple processes can append in turn.

Classes

AuditLog

The append-only signed log surface.

FileAuditLog

JSONL append-only audit log with HMAC signatures.

InMemoryAuditLog

List-backed signed audit log.

Functions

stream_entries(...)

Yield every entry currently in log in seq order.

verify_signature(→ bool)

Recompute the HMAC and compare against the stored signature.

Module Contents

class jeevesagent.security.audit.AuditLog[source]

Bases: Protocol

The append-only signed log surface.

async append(*, session_id: str, actor: str, action: str, payload: dict[str, Any]) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
class jeevesagent.security.audit.FileAuditLog(path: str | pathlib.Path, *, secret: str = '')[source]

JSONL append-only audit log with HMAC signatures.

On construction we read any pre-existing entries to recover the highest seq, so a process restart picks up where the last one left off.

async append(*, session_id: str, actor: str, action: str, payload: dict[str, Any]) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
property path: pathlib.Path
class jeevesagent.security.audit.InMemoryAuditLog(*, secret: str = '')[source]

List-backed signed audit log.

async all_entries() list[jeevesagent.core.types.AuditEntry][source]
async append(*, session_id: str, actor: str, action: str, payload: dict[str, Any]) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
async jeevesagent.security.audit.stream_entries(log: AuditLog) collections.abc.AsyncIterator[jeevesagent.core.types.AuditEntry][source]

Yield every entry currently in log in seq order.

A polling helper for compliance tooling. Doesn’t tail — that comes later when we add an on-write notification stream.

jeevesagent.security.audit.verify_signature(entry: jeevesagent.core.types.AuditEntry, secret: str) bool[source]

Recompute the HMAC and compare against the stored signature.