jeevesagent.security

Security harness: permissions, hooks, sandbox, audit.

Submodules

Attributes

Classes

AllowAll

Trivial permission policy: every call is allowed.

AuditLog

The append-only signed log surface.

DictSecrets

In-process Secrets backed by an explicit dict.

EnvSecrets

Reads secrets from os.environ.

FileAuditLog

JSONL append-only audit log with HMAC signatures.

FilesystemSandbox

Restrict a tool host's path-typed arguments to declared roots.

HookRegistry

Implements HookHost.

InMemoryAuditLog

List-backed signed audit log.

Mode

Enum where members are also (and must be) strings

NoSandbox

Pass-through wrapper around a ToolHost.

PerUserPermissions

Map user_id to a different permission policy.

StandardPermissions

Mode + allow/deny-list permission policy.

SubprocessSandbox

Run each tool call in a fresh child Python process.

Functions

verify_signature(→ bool)

Recompute the HMAC and compare against the stored signature.

Package Contents

class jeevesagent.security.AllowAll[source]

Trivial permission policy: every call is allowed.

The default for Agent when no permissions are configured.

async check(call: jeevesagent.core.types.ToolCall, *, context: collections.abc.Mapping[str, Any], user_id: str | None = None) jeevesagent.core.types.PermissionDecision[source]
class jeevesagent.security.AuditLog[source]

Bases: Protocol

The append-only signed log surface.

user_id (M9) is a top-level field on every entry, populated from the live RunContext. Backends MUST accept the kwarg on append and the query filter so multi-tenant audit queries work without payload-digging.

async append(*, session_id: str, actor: str, action: str, payload: dict[str, Any], user_id: str | None = None) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None, user_id: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
class jeevesagent.security.DictSecrets(initial: dict[str, str] | None = None)[source]

In-process Secrets backed by an explicit dict.

Useful in tests and for callers that fetch secrets once at startup (from a config file, a one-shot Vault read, etc.) and want to make them available to the framework without leaking them into os.environ.

Mutable: store() updates the in-process map. Not durable across process restarts.

lookup_sync(ref: str) str | None[source]
redact(text: str) str[source]
async resolve(ref: str) str[source]
async store(ref: str, value: str) None[source]
class jeevesagent.security.EnvSecrets[source]

Reads secrets from os.environ.

The default Secrets impl wired by Agent when the caller doesn’t pass an explicit one. Behaviour matches the pre-M10 framework: API keys are looked up as the corresponding environment variable name (OPENAI_API_KEY, ANTHROPIC_API_KEY, etc.).

lookup_sync(ref: str) str | None[source]
redact(text: str) str[source]
async resolve(ref: str) str[source]
abstractmethod store(ref: str, value: str) None[source]
Async:

class jeevesagent.security.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], user_id: str | None = None) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None, user_id: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
property path: pathlib.Path
class jeevesagent.security.FilesystemSandbox(inner: jeevesagent.core.protocols.ToolHost, *, roots: collections.abc.Iterable[str | pathlib.Path], path_args: collections.abc.Iterable[str] | None = None, auto_detect: bool = True)[source]

Restrict a tool host’s path-typed arguments to declared roots.

async call(tool: str, args: collections.abc.Mapping[str, Any], *, call_id: str = '') jeevesagent.core.types.ToolResult[source]
async list_tools(*, query: str | None = None) list[jeevesagent.core.types.ToolDef][source]
async watch() collections.abc.AsyncIterator[jeevesagent.core.types.ToolEvent][source]
property inner: jeevesagent.core.protocols.ToolHost
property roots: tuple[pathlib.Path, Ellipsis]
class jeevesagent.security.HookRegistry[source]

Implements HookHost.

async on_event(event: jeevesagent.core.types.Event) None[source]
async post_tool(call: jeevesagent.core.types.ToolCall, result: jeevesagent.core.types.ToolResult, *, user_id: str | None = None) None[source]

Best-effort post-tool callbacks. Failures and timeouts are absorbed so they cannot affect the result the loop returns. user_id follows the same forwarded-but-not-required pattern as pre_tool().

async pre_tool(call: jeevesagent.core.types.ToolCall, *, user_id: str | None = None) jeevesagent.core.types.PermissionDecision[source]

Run all pre-tool hooks. First deny wins; otherwise allow.

The user_id kwarg is forwarded for protocol parity (M9); the bundled HookRegistry doesn’t itself dispatch per-user, but custom HookHost implementations can route on it. Individual hook callables continue to receive only (call,) to keep the existing decorator API stable; hooks that need the user_id can call get_run_context() themselves.

register_event(hook: EventHook) EventHook[source]
register_post_tool(hook: PostToolHook) PostToolHook[source]
register_pre_tool(hook: PreToolHook) PreToolHook[source]
event_hooks: list[EventHook] = []
hook_timeout_s: float = 5.0
post_tool_hooks: list[PostToolHook] = []
pre_tool_hooks: list[PreToolHook] = []
class jeevesagent.security.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], user_id: str | None = None) jeevesagent.core.types.AuditEntry[source]
async query(*, session_id: str | None = None, action: str | None = None, user_id: str | None = None) list[jeevesagent.core.types.AuditEntry][source]
class jeevesagent.security.Mode[source]

Bases: enum.StrEnum

Enum where members are also (and must be) strings

Initialize self. See help(type(self)) for accurate signature.

ACCEPT_EDITS = 'acceptEdits'
BYPASS = 'bypassPermissions'
DEFAULT = 'default'
class jeevesagent.security.NoSandbox(inner: jeevesagent.core.protocols.ToolHost)[source]

Pass-through wrapper around a ToolHost.

async call(tool: str, args: collections.abc.Mapping[str, Any], *, call_id: str = '') jeevesagent.core.types.ToolResult[source]
async list_tools(*, query: str | None = None) list[jeevesagent.core.types.ToolDef][source]
async watch() collections.abc.AsyncIterator[jeevesagent.core.types.ToolEvent][source]
property inner: jeevesagent.core.protocols.ToolHost
class jeevesagent.security.PerUserPermissions(*, policies: collections.abc.Mapping[str | None, Any], default: Any)[source]

Map user_id to a different permission policy.

The common multi-tenant shape: admins get one policy, paid users get another, free users get a third. Construct with a mapping of user_id -> Permissions plus a default fallback for unmapped users (including None):

from jeevesagent import (
    PerUserPermissions, StandardPermissions, Mode,
)

permissions = PerUserPermissions(
    policies={
        "admin_alice": StandardPermissions(mode=Mode.BYPASS),
        "paid_user_42": StandardPermissions(
            mode=Mode.ACCEPT_EDITS,
        ),
    },
    default=StandardPermissions(
        mode=Mode.DEFAULT,
        denied_tools=["bash", "delete_user"],
    ),
)
agent = Agent("...", permissions=permissions)

Each check call routes to the policy keyed by user_id (the live RunContext’s value, threaded through by the agent loop). When no policy matches, the default decides — most apps want a strict default and add permissive policies for trusted users.

async check(call: jeevesagent.core.types.ToolCall, *, context: collections.abc.Mapping[str, Any], user_id: str | None = None) jeevesagent.core.types.PermissionDecision[source]
class jeevesagent.security.StandardPermissions(*, mode: Mode = Mode.DEFAULT, allowed_tools: list[str] | None = None, denied_tools: list[str] | None = None)[source]

Mode + allow/deny-list permission policy.

async check(call: jeevesagent.core.types.ToolCall, *, context: collections.abc.Mapping[str, Any], user_id: str | None = None) jeevesagent.core.types.PermissionDecision[source]
classmethod strict() StandardPermissions[source]

Convenience: default-mode permissions with no overrides.

class jeevesagent.security.SubprocessSandbox(inner: jeevesagent.core.protocols.ToolHost, *, timeout_seconds: float = 30.0)[source]

Run each tool call in a fresh child Python process.

async call(tool: str, args: collections.abc.Mapping[str, Any], *, call_id: str = '') jeevesagent.core.types.ToolResult[source]
async list_tools(*, query: str | None = None) list[jeevesagent.core.types.ToolDef][source]
async watch() collections.abc.AsyncIterator[jeevesagent.core.types.ToolEvent][source]
property inner: jeevesagent.core.protocols.ToolHost
property timeout_seconds: float
jeevesagent.security.verify_signature(entry: jeevesagent.core.types.AuditEntry, secret: str) bool[source]

Recompute the HMAC and compare against the stored signature.

jeevesagent.security.PostToolHook
jeevesagent.security.PreToolHook