core.security¶
Security pipeline modules: tool execution gate, manifest validation, autonomy policy, approval workflow, and process isolation.
Tool Host¶
core.security.tool_host
¶
Capability-gated tool execution boundary.
Sits between the agent pipeline and raw tool functions. Every tool invocation passes through this host which enforces:
- Manifest validation — the tool must have a registered manifest.
- Permission check — AutonomyPolicy tier + side-effect matrix.
- Approval gating — HITL workflow for tools that require it.
- Timeout enforcement — each tool has a max execution budget.
- Output size limit — truncated responses that exceed the manifest
max_output_bytes. - Audit logging — structured events for every decision point.
- Event publishing — typed events to the EventBus for observers.
The orchestrator should never call arbitrary Python tool functions directly. This module is the security boundary.
Typical usage
tool_host = RealToolHost(
manifests=create_builtin_manifests(),
autonomy_policy=AutonomyPolicy(tier=AutonomyTier.OPERATOR),
approval_workflow=ApprovalWorkflow(),
event_bus=event_bus,
audit_log=audit_log,
)
await tool_host.register_handler("web_search", web_search_fn)
result = await tool_host.execute(
tool_name="web_search",
arguments={"query": "latest news"},
caller="agent",
)
RealToolHost
¶
Capability-gated tool execution host.
Validates manifests, checks permissions, applies autonomy policy, runs approval prompts when needed, and invokes tools through a narrow execution context.
Parameters¶
manifests :
Tool manifest registry.
autonomy_policy :
Tiered autonomy policy engine.
approval_workflow :
HITL approval workflow.
event_bus :
Async event bus for publishing events.
audit_log :
Structured audit trail.
default_timeout :
Fallback timeout (seconds) for tools that don't specify one
in their manifest. 0 means no timeout.
isolation_manager :
Optional process isolation layer for WRITE/DESTRUCTIVE tools.
When provided, tools with sufficient side-effect levels are
executed in sandboxed subprocesses with path/domain validation,
timeout enforcement, and Windows job object containment.
manifest_names: list[str]
property
¶
List of tool names that have manifests.
registered_tools: list[str]
property
¶
List of tool names that have handlers registered.
turn_id: str
property
¶
Current turn identifier.
approve_tool_call(*, tool_name: str, arguments: dict[str, Any], grant: bool, persistent: bool = False) -> None
async
¶
Record a manual approval or denial decision.
This looks up any pending approval request for the tool and resolves it, or creates a persistent grant/denial.
execute(*, tool_name: str, arguments: dict[str, Any], correlation_id: str = '', caller: str = '') -> dict[str, Any]
async
¶
Validate manifest → check policy → HITL if needed → execute.
Returns a dict with result, success, and optional error.
Steps¶
- Resolve manifest (fail fast if unknown tool).
- Check autonomy policy permission.
- Check persistent denial (fail immediately).
- Check persistent grant (skip approval if granted).
- Run approval workflow if needed.
- Execute tool handler with timeout enforcement.
- Enforce output size limit.
- Publish events and audit log.
get_tool_manifest(tool_name: str) -> dict[str, Any] | None
async
¶
Return the manifest for a single tool, or None.
has_handler(tool_name: str) -> bool
¶
Return True if a handler is registered for tool_name.
list_tools() -> list[dict[str, Any]]
async
¶
Return active tool manifests.
register_handler(tool_name: str, handler: ToolHandler, *, module_name: str | None = None, function_name: str | None = None) -> None
¶
Register the execution function for a tool.
The handler is looked up by tool_name at execution time.
The handler may be a sync function or an async coroutine.
Parameters¶
module_name : Python module containing the handler (for subprocess isolation). Required when PROCESS-level isolation is configured and the tool has WRITE side effects. Item #20. function_name : Function name within the module (for subprocess isolation).
set_turn(turn_id: str) -> None
¶
Start a new generation turn, resetting call counts.
Tool Host Adapter¶
core.security.tool_host_adapter
¶
Tool execution adapter — bridges Agent methods to RealToolHost.
Converts Agent method calls into
RealToolHost-compatible handler functions so that every tool
invocation flows through the 8-step security pipeline (manifest → policy
→ denial → grant → approval → execute → output limit → audit).
Usage¶
from core.tool_host_adapter import ToolHostAdapter
adapter = ToolHostAgent(agents)
await adapter.register_handlers(tool_host)
See Also¶
core/tool_host.py-- RealToolHost with 8-step pipeline.core/agent_classes.py-- Agent methods (web_search, fetch_url).
ToolHostAdapter
¶
Wires existing Agent methods into RealToolHost.
Parameters¶
agents :
List of Agent instances to register handlers for.
build_tool_host(event_bus=None, audit_log=None, approval_workflow=None, autonomy_policy=None, isolation_manager=None, default_timeout: float = 30.0) -> RealToolHost
staticmethod
¶
Instantiate a RealToolHost with sensible defaults.
Parameters¶
event_bus :
Defaults to EventBus() when omitted.
audit_log :
Defaults to InMemoryAuditLog() (test-only). In production,
pass SQLiteStoreAuditAdapter(store) instead.
approval_workflow :
Defaults to a fresh ApprovalWorkflow() when omitted.
autonomy_policy :
Defaults to AutonomyTier.OPERATOR when omitted.
isolation_manager :
When omitted, no process isolation is applied.
default_timeout :
Seconds before tool execution times out (default 30).
Note¶
Production code should inject SQLiteStoreAuditAdapter (via the
adapter exported from core.audit_log) rather than relying on
the in-memory default.
register_handlers(tool_host: RealToolHost) -> None
async
¶
Register each agent's tool methods with tool_host.
All tools are agent-aware — the correct agent is resolved at
execution time from the _caller string so that toasts (and any
per-agent state such as system_prompt_override) use the calling
agent's identity. Falls back to the first agent when the caller
cannot be parsed.
Tool Manifest¶
core.security.tool_manifest
¶
Versioned tool manifest schema.
Defines the contract for every built-in and plugin tool: identity, versioning, input/output schemas, side-effect classification, permission boundaries, confirmation policy, and audit metadata.
This module is independent of the Ollama-specific TOOL_SCHEMAS in
tools/schemas.py — manifests describe what a tool is, while schemas
describe how to call it through a particular model backend.
ToolManifest
dataclass
¶
Immutable, versioned manifest for a single tool.
Parameters¶
name :
Unique tool identifier (snake_case, e.g. "web_search").
version :
Semver string ("1.0.0").
description :
Human-readable description shown to users and approval dialogs.
summary :
One-line summary for compact displays (defaults to description's
first sentence).
side_effect_level :
Classification of the tool's blast radius.
confirmation_policy :
When human approval is required.
input_schema :
JSON Schema object describing the arguments dict. Keys are
property names, values are JSON Schema subschemas.
output_schema :
Optional JSON Schema describing the tool's return value.
allowed_paths :
Filesystem path prefixes the tool may access (empty = no filesystem
access). Glob patterns ("*") supported.
allowed_domains :
Network domains the tool may contact (empty = no network access).
Wildcard domains ("*.example.com") supported.
timeout_seconds :
Maximum execution time. 0 means no timeout.
max_output_bytes :
Maximum response size in bytes. 0 means unlimited.
max_calls_per_turn :
Maximum invocations per generation turn. 0 means unlimited.
audit_labels :
Free-form tags for audit filtering (e.g. ["network", "facts"]).
depends_on :
Other tool names this tool may internally invoke.
metadata :
Arbitrary extra key-values (not enforced by the runtime).
Raises¶
ValueError If name is invalid or version doesn't match semver.
from_dict(data: dict[str, Any]) -> ToolManifest
classmethod
¶
Deserialize from a dict.
Accepts both string and enum values for side_effect_level and
confirmation_policy.
needs_approval_for(autonomy_tier: int = 0) -> bool
¶
Determine if this tool needs approval at the given autonomy tier.
This is a heuristic combining confirmation policy and side-effect level. The real decision should go through the ToolHost + AutonomyPolicy.
save(path: str | Path) -> None
¶
Serialize this manifest to a JSON file.
to_dict() -> dict[str, Any]
¶
Serialize to a plain dict (enums become strings).
to_ollama_schema() -> dict[str, Any]
¶
Produce an Ollama-compatible tool dict from this manifest.
Returns a dict matching the shape expected by Ollama's tools
parameter::
{"function": {"name": ..., "description": ..., "parameters": {...}}}
SideEffectLevel
¶
Bases: StrEnum
Classifies the blast radius of a tool execution.
Levels¶
none : Pure computation or no-ops. No external state changes. read : Reads external state (files, network, environment) but never mutates. write : Creates or modifies external state (files, settings, scheduled tasks). destructive : Irreversible or high-risk operations (deletion, subprocess execution, credential access, plugin installation).
severity: int
property
¶
Numeric severity for comparisons (0 = safest).
ConfirmationPolicy
¶
Bases: StrEnum
When does a tool call require human approval?
Policies¶
never : Always auto-execute (safe read-only or computation tools). first_time : Require approval on first execution; remember the grant thereafter. always : Require approval for every invocation. autonomy_gated : Defer to the AutonomyPolicy engine — tier and side-effect level determine whether approval is needed.
create_builtin_manifests() -> ToolManifests
¶
Return the default manifests for the 4 built-in tools.
This serves as the canonical declaration of tool capabilities, side-effects, and permission requirements.
Autonomy Policy¶
core.security.autonomy_policy
¶
Autonomy mode policy engine.
Implements four tiers of autonomy — Companion, Assistant, Operator, Automation — each with different default tool permissions, side-effect limits, and confirmation rules. This is the code-enforced policy layer; prompt text may describe the mode but cannot replace these checks.
The decision matrix per tier:
+----------------+--------+-----------+-----------+-------------+ | Tier | None | Read | Write | Destructive | +----------------+--------+-----------+-----------+-------------+ | Companion (0) | deny | deny | deny | deny | | Assistant (1) | allow | allow | deny | deny | | Operator (2) | allow | allow | allow | deny | | Automation (3) | allow | allow | allow | allow |
requires_approval() determines whether a tool call must pause for human
confirmation even if tool_allowed() returns True. The matrix:
+----------------+--------+-----------+-----------+-------------+ | Tier | None | Read | Write | Destructive | +----------------+--------+-----------+-----------+-------------+ | Companion (0) | approve| approve | approve | approve | | Assistant (1) | skip | skip | N/A | N/A | | Operator (2) | skip | skip | approve | N/A | | Automation (3) | skip | skip | skip | approve |
Max tool calls per turn also varies by tier (see _DEFAULT_MAX_TOOL_CALLS).
AutonomyPolicy
¶
Code-enforced autonomy policy with four tiers.
Parameters¶
tier:
Starting autonomy tier. Defaults to COMPANION (safest).
allowed_tools:
Explicitly allowed tool names. An empty set means "allow all
tools that pass the tier matrix". A non-empty set acts as a
whitelist: only named tools are considered.
denied_tools:
Tool names that are never allowed, regardless of tier.
Takes precedence over allowed_tools.
max_tool_calls:
Override the default per-tier limit. 0 means unlimited.
auto_approve_none:
If True, tools with side-effect level "none" bypass approval
checks even at Companion tier.
Examples¶
policy = AutonomyPolicy(tier=AutonomyTier.ASSISTANT) policy.tool_allowed("web_search", SideEffectLevel.READ.value) True policy.tool_allowed("write_file", SideEffectLevel.WRITE.value) False
add_denied_tool(tool_name: str) -> None
¶
Permanently deny a tool, regardless of tier.
get_tier_defaults() -> TierDefaults
¶
Return the TierDefaults for the current tier.
list_tiers() -> list[TierDefaults]
¶
Return all tier descriptors in order.
max_tool_calls_per_turn() -> int
¶
Return the max recursive tool calls allowed per turn.
Returns a tier-specific default unless overridden via
set_max_tool_calls().
remove_denied_tool(tool_name: str) -> None
¶
Remove a tool from the denied list.
requires_approval(tool_name: str) -> bool
¶
requires_approval_for_level(tool_name: str, side_effect_level: str) -> bool
¶
Check approval need for a specific side-effect level.
Use this method when you know the tool's side-effect level (e.g., from the ToolManifest) rather than guessing.
Parameters¶
tool_name:
The tool identifier.
side_effect_level:
One of "none", "read", "write", "destructive".
Returns¶
bool
True if human confirmation is required.
set_allowed_tools(tools: set[str]) -> None
¶
Set an explicit whitelist of allowed tools.
Pass an empty set to clear the whitelist (allow all tools that pass the tier matrix).
set_max_tool_calls(value: int) -> None
¶
Override the per-turn tool call limit.
Parameters¶
value:
Maximum calls. Pass 0 to revert to tier default.
set_tier(tier: int) -> None
¶
AutonomyTier
¶
Bases: IntEnum
Four autonomy tiers with increasing tool access.
ASSISTANT = 1
class-attribute
instance-attribute
¶
Read-only tools allowed. No mutations.
AUTONOMY = 3
class-attribute
instance-attribute
¶
Full tool access. Destructive actions still require approval.
COMPANION = 0
class-attribute
instance-attribute
¶
All tool calls blocked. Pure conversation mode.
OPERATOR = 2
class-attribute
instance-attribute
¶
Read + write tools allowed. Destructive actions blocked.
Approval Workflow¶
core.security.approval_workflow
¶
Human-in-the-loop approval workflow.
Provides a typed approval flow for risky actions: file writes, deletions, subprocess execution, network calls, plugin installation, scheduled automation, and multi-step autonomous tasks.
The workflow engine integrates with: - AutonomyPolicy — tier-based permission and approval checks - ToolManifest — per-tool confirmation policies - EventBus — approval request/grant/deny/timeout events - AuditLog — structured audit trail for each decision
Key concepts¶
ApprovalRequest— immutable descriptor for a pending approval. Includes tool name, arguments, side-effect level, reason, optional diff/preview, timeout, and grant mode.ApprovalDecision— enum of outcomes: granted, denied, timed_out.ApprovalResult— frozen record linking a request to its decision, timestamp, and optional rationale.ApprovalWorkflow— manages the lifecycle: creating requests, granting/denying, expiring timed-out entries, tracking persistent grants, and maintaining decision history.
Grant modes¶
- one_time — the decision applies only to this specific request.
- persistent — the decision is remembered for future requests with
the same
tool_name+ matching arguments (exact match on argument keys). Persistent grants can be revoked.
Timeout semantics¶
Each request has a timeout (default 60 seconds). Expired requests are
automatically resolved as timed_out (treated as denial for execution
purposes). Callers should poll get_pending() or subscribe to
APPROVAL_TIMEOUT events from the EventBus.
Examples¶
import asyncio from core.approval_workflow import ApprovalWorkflow from core.tool_manifest import SideEffectLevel
async def main(): ... wf = ApprovalWorkflow(default_timeout=30.0) ... req_id = await wf.create( ... tool_name="write_file", ... arguments={"path": "/tmp/hello.txt", "content": "hi"}, ... side_effect_level=SideEffectLevel.WRITE, ... reason="Agent wants to write a configuration file.", ... ) ... pending = await wf.get_pending(req_id) ... print(pending) # ApprovalRequest(...) ... result = await wf.grant(req_id, rationale="User approved") ... print(result.decision) # ApprovalDecision.GRANTED
asyncio.run(main())
ApprovalWorkflow
¶
Manages the lifecycle of human-in-the-loop approvals.
Parameters¶
default_timeout:
Default timeout in seconds for new approval requests (default 60).
max_pending:
Maximum number of concurrent pending requests. If exceeded,
create() raises RuntimeError. Set to 0 for unlimited.
grant_tool_fn:
Optional async callback invoked when a persistent grant/deny is
resolved. Signature: async(tool_name: str, decision: str) -> None.
Typically writes to SQLiteStore.grant_tool().
list_grants_fn:
Optional async callback used by load_persistent_grants() to
retrieve persisted grants. Signature: async() -> list[dict].
Each dict must have tool_name, decision ("GRANTED" or
"DENIED"), and persistent (bool) keys. Typically calls
SQLiteStore.list_tool_grants() and serializes the results.
Examples¶
wf = ApprovalWorkflow(default_timeout=30.0)
create a request, grant it, check history¶
req_id = wf.create_sync(tool_name="web_search", arguments={"query": "config"}, ... side_effect_level="read", reason="retrieve config") result = wf.grant_sync(req_id) result.is_approved True
history_count: int
property
¶
Total decisions recorded.
pending_count: int
property
¶
Number of currently pending (non-expired) requests.
check_persistent(tool_name: str) -> ApprovalDecision | None
async
¶
Check if there is a persistent grant/deny for a tool.
Returns GRANTED if the tool has a persistent grant,
DENIED if persistently denied, or None if no
persistent decision exists.
clear_history() -> int
async
¶
Clear all decision history. Returns count cleared.
create(*, tool_name: str, arguments: dict[str, Any], side_effect_level: str, reason: str, preview: str | None = None, timeout_seconds: float | None = None, grant_mode: ApprovalGrantMode = ApprovalGrantMode.ONE_TIME, correlation_id: str = '', caller: str = '') -> str
async
¶
Create a new approval request.
Parameters¶
tool_name:
The tool requesting approval.
arguments:
Arguments that would be passed to the tool.
side_effect_level:
One of "none", "read", "write", "destructive".
reason:
Human-readable explanation of the proposed action.
preview:
Optional diff/preview of the change.
timeout_seconds:
Override the default timeout.
grant_mode:
Whether to remember the decision persistently.
correlation_id:
Links to a broader operation.
caller:
The requesting agent/component.
Returns¶
str
The request_id to pass to grant() / deny().
Raises¶
RuntimeError
If max_pending would be exceeded.
ValueError
If an invalid side_effect_level is provided.
create_sync(*, tool_name: str, arguments: dict[str, Any], side_effect_level: str, reason: str, preview: str | None = None, timeout_seconds: float | None = None, grant_mode: ApprovalGrantMode = ApprovalGrantMode.ONE_TIME, correlation_id: str = '', caller: str = '') -> str
¶
Synchronous wrapper around create().
deny(request_id: str, *, rationale: str = '') -> ApprovalResult
async
¶
deny_sync(request_id: str, *, rationale: str = '') -> ApprovalResult
¶
Synchronous wrapper around deny().
get_history(*, tool_name: str | None = None, limit: int = 50) -> list[ApprovalResult]
async
¶
Return decision history, most recent first.
Parameters¶
tool_name: Optional filter by tool name. limit: Maximum entries to return.
get_pending(request_id: str) -> ApprovalRequest | None
async
¶
Return the pending request, or None if resolved/expired.
get_result(request_id: str) -> ApprovalResult | None
async
¶
Look up the result for a resolved request.
grant(request_id: str, *, rationale: str = '') -> ApprovalResult
async
¶
grant_sync(request_id: str, *, rationale: str = '') -> ApprovalResult
¶
Synchronous wrapper around grant().
list_pending() -> list[ApprovalRequest]
async
¶
Return all non-expired pending requests.
list_persistent() -> dict[str, str]
async
¶
Return a snapshot of all persistent grants and denials.
Returns¶
dict[str, str]
Mapping of tool_name -> "GRANTED" or "DENIED".
load_persistent_grants() -> int
async
¶
Load persistent grants/denials from the persistence backend.
Uses list_grants_fn callback if configured. Populates the
in-memory _persistent_grants and _persistent_denials sets
so that check_persistent() works after a restart.
Returns¶
int Number of persistent grants loaded.
Notes¶
This should be called once at startup (e.g. in main.py
after the SQLiteStore is initialized). If no list_grants_fn
is configured, returns 0 immediately.
revoke_persistent(tool_name: str) -> bool
async
¶
Remove any persistent grant or denial for a tool.
If a persistence callback is configured, the revocation is also
propagated to the backend (e.g. SQLiteStore.revoke_tool_grant()).
wait_for_resolution(request_id: str, timeout: float = 60.0) -> ApprovalResult | None
async
¶
Block until a pending request is resolved or the timeout elapses.
Parameters¶
request_id:
The ID returned by create().
timeout:
Maximum seconds to wait. After expiry the request is
auto-resolved as TIMED_OUT.
Returns¶
ApprovalResult | None
The decision if resolved within the window, or a TIMED_OUT
result if the waiter expired. Returns None if the request
does not exist.
ApprovalResult
dataclass
¶
Immutable record of an approval decision.
Parameters¶
request_id: The request this result refers to. decision: The outcome (granted / denied / timed_out). rationale: Optional human explanation for the decision. decided_at: UTC timestamp of the decision. persistent: Whether the decision was saved as a persistent grant.
Audit Log¶
core.security.audit_log
¶
Structured audit log — append-only trail for runtime events.
Provides:
- AuditLevel — severity enumeration (debug → critical)
- AuditEvent — frozen, immutable event record
- InMemoryAuditLog — dict-backed implementation (tests, CI)
- SqliteAuditLog — deprecated; use SQLiteStoreAuditAdapter instead
Both implementations satisfy the AuditLog Protocol in
core/service_protocols.py.
AuditEvent
dataclass
¶
Immutable audit event record.
Parameters¶
id :
Unique identifier (auto-generated via AuditEvent.create()).
timestamp :
UTC timestamp.
event_type :
String identifying the event category.
source :
Component that generated the event (e.g. "tool_host").
correlation_id :
Groups related events (one user turn, one tool execution chain).
data :
Arbitrary structured payload.
level :
Severity level.
create(*, event_type: str, source: str, correlation_id: str = '', data: dict[str, Any] | None = None, level: AuditLevel | str = AuditLevel.INFO) -> AuditEvent
classmethod
¶
Create a new audit event with auto-generated ID and timestamp.
Parameters¶
level :
Can be either an AuditLevel enum or a string.
to_dict() -> dict[str, Any]
¶
Serialize to a plain dict.
AuditEventType
¶
Canonical event type strings for consistency across the codebase.
These are conventional constants — the audit log accepts arbitrary strings, but using these ensures uniformity in queries and dashboards.
AuditLevel
¶
Bases: StrEnum
Severity classification for audit events.
Levels¶
DEBUG : Internal state changes, trace-level detail. INFO : Normal operational events (tool calls, model switches). WARNING : Policy denials, rate-limit hits, retries, fallbacks. ERROR : Failures that affect functionality (tool errors, timeouts). CRITICAL : Security violations, data corruption, unrecoverable errors.
CorrelationScope
¶
Context-manager-style correlation ID generator.
Usage::
corr = CorrelationScope()
with corr as cid:
log(event_type="x", correlation_id=cid)
InMemoryAuditLog
¶
Thread-safe in-memory audit log.
Satisfies the AuditLog Protocol. Suitable for tests, CI, and
short-lived processes.
events() -> list[AuditEvent]
¶
Return all events (for test assertions).
SQLiteStoreAuditAdapter
¶
Adapts SQLiteStore audit methods to the AuditLog Protocol.
Wraps SQLiteStore.log_audit() and SQLiteStore.query_audit() so
the store's audit_events table satisfies the same interface as
InMemoryAuditLog (and the deprecated SqliteAuditLog).
Parameters¶
store :
A SQLiteStore instance (must have log_audit and
query_audit async methods).
SqliteAuditLog
¶
Persistent SQLite-backed audit log.
.. deprecated:: 2026-06-14
Use SQLiteStoreAuditAdapter (wrapping SQLiteStore) instead.
This class is kept for backwards compatibility with existing tests
and standalone scripts but will be removed in a future release.
Satisfies the AuditLog Protocol. Uses WAL mode for concurrent
reads/writes and bounded retention.
Parameters¶
db_path :
Path to the SQLite database file.
max_age_days :
Automatically purge entries older than this many days on each log()
call. 0 disables rotation (default).
Process Isolation¶
core.security.isolation
¶
Process isolation for risky tool execution.
Provides path/domain validation, workspace confinement, subprocess sandbox execution, and Windows job object containment. Plugs into the ToolHost pipeline as an optional pre-execution validation + isolation layer.
Design goals¶
- Path validation — prevent directory traversal, symlink escapes, and access outside allowed workspace roots.
- Domain validation — enforce network allowlists for tools that contact external services.
- Subprocess sandbox — run WRITE/DESTRUCTIVE tools in child processes with timeout, filtered environment, and constrained cwd.
- Windows job objects — contain child process trees so orphaned processes don't escape the sandbox.
- Graceful degradation — falls back to in-process execution when subprocess isolation is unavailable (e.g. editable installs without module-level functions).
ArgumentScanner
¶
Scans tool arguments for paths and domains that violate sandbox rules.
Checks both explicit path/domain parameters and text content that might contain embedded paths or URLs.
scan(arguments: dict[str, Any]) -> list[str]
¶
Scan arguments for path/domain violations.
Returns a list of warning/error strings (empty = clean).
DomainDeniedError
¶
Bases: IsolationError
Domain is not in the allowed list.
DomainValidator
¶
Validates network domains against an allowlist.
extract_domains(text: str) -> list[str]
¶
Extract obvious hostnames/URLs from a string.
is_allowed(domain: str) -> bool
¶
Return True if domain matches any allowed domain pattern.
validate(domain: str) -> str
¶
Validate domain is in the allowed list.
Raises DomainDeniedError if the domain is blocked.
InProcessExecutor
¶
Executes tool handlers in the main process (async-safe).
execute(handler: Callable, arguments: dict[str, Any], timeout: float | None = None) -> dict[str, Any]
async
¶
Execute handler with arguments, optionally with timeout.
IsolationError
¶
Bases: Exception
Base class for isolation violations.
IsolationLevel
¶
Bases: Enum
How much isolation a tool execution receives.
FULL = 'full'
class-attribute
instance-attribute
¶
Process isolation + OS-level containment (job objects / sandbox).
NONE = 'none'
class-attribute
instance-attribute
¶
No isolation — runs in the main process (default for read-only).
PROCESS = 'process'
class-attribute
instance-attribute
¶
Runs in a child Python process with timeout + env filtering.
SANDBOXED = 'sandboxed'
class-attribute
instance-attribute
¶
Runs in-process with argument validation (path/domain checks).
IsolationManager
¶
Coordinates path/domain validation and subprocess isolation.
This is the main entry point for ToolHost integration. It decides the isolation level for a tool call, validates arguments, and either delegates to the in-process or subprocess executor.
Parameters¶
config : SandboxConfig Sandbox settings (workspace root, allowed paths/domains, etc).
execute(*, handler: Callable, arguments: dict[str, Any], side_effect: str = 'none', module_name: str | None = None, function_name: str | None = None, timeout: float | None = None) -> SandboxResult
async
¶
Execute a tool handler with appropriate isolation.
For NONE/SANDBOXED: runs in-process with optional arg validation. For PROCESS/FULL: runs in a child process with sandboxing.
Parameters¶
handler : The tool handler callable (used for in-process execution). arguments : JSON-serializable tool arguments. side_effect : Side-effect level string ("none" / "read" / "write" / "destructive"). module_name : Python module containing the handler (for subprocess mode). function_name : Function name within the module (for subprocess mode). timeout : Execution timeout in seconds (overrides config default).
get_isolation_level(side_effect: str) -> IsolationLevel
¶
Return the isolation level for a given side-effect classification.
NONE side-effect level is never upgraded (no side effects = no isolation needed, regardless of config floor). Other levels are upgraded to match the config's minimum isolation_level.
should_isolate(side_effect: str) -> bool
¶
Return True if tools with this side-effect level need isolation.
validate_arguments(arguments: dict[str, Any]) -> list[str]
¶
Scan arguments for path/domain violations.
Returns a list of issues (empty = clean).
PathTraversalError
¶
Bases: IsolationError
Path escapes allowed workspace roots.
PathValidator
¶
Validates filesystem paths against workspace roots.
extract_paths(text: str) -> list[str]
¶
Extract obvious file paths from a string (for content scanning).
Looks for strings that look like absolute or relative paths. This is a best-effort heuristic, not a parser.
is_within_roots(path_str: str) -> bool
¶
Return True if path_str resolves inside any allowed root.
validate(path_str: str, must_exist: bool = False) -> Path
¶
Validate path_str is within allowed roots.
Returns the resolved Path. Raises PathTraversalError if the path escapes all allowed roots.
SandboxConfig
dataclass
¶
Configuration for a tool sandbox.
effective_roots: list[Path]
property
¶
Return resolved workspace roots for path validation.
SandboxExecutionError
¶
Bases: IsolationError
Subprocess sandbox failed to execute the tool.
SandboxResult
dataclass
¶
Result of a sandboxed tool execution.
exception_type: str = ''
class-attribute
instance-attribute
¶
Optional exception class name for error classification (e.g. 'SandboxTimeoutError', 'PathTraversalError', 'DomainDeniedError').
SandboxTimeoutError
¶
Bases: IsolationError
Subprocess sandbox exceeded its execution timeout.
SubprocessExecutor
¶
Executes tool handlers in an isolated child Python process.
The child process has: - Constrained working directory (workspace root) - Filtered environment variables - Execution timeout (process is killed on timeout) - On Windows: job object containment for child process trees
execute(*, module_name: str, function_name: str, arguments: dict[str, Any], timeout: float = 30.0) -> SandboxResult
async
¶
Execute module_name:function_name in an isolated subprocess.
Returns a SandboxResult with the output, timing, and status.
Content Guards¶
core.security.content_guards
¶
Content guards — untrusted-content wrappers and injection detection.
Production modules for separating untrusted content (web search results, file contents, tool output) from system prompts, with heuristic injection detection and policy reminders at the tool boundary.
Components¶
UntrustedContent— frozen wrapper marking a string as untrusted, providing sanitized views and provenance trackingContentCategory— enum classifying the source of untrusted contentInjectionFinding— frozen record of a detected injection attemptContentGuard— heuristic scanner for prompt injection, command injection, path traversal, and cross-agent smuggling in untrusted textPolicyReminder— generates system-level reminders to prepend when injecting untrusted content into the agent's context
ContentCategory
¶
Bases: Enum
Source of untrusted content, used for provenance tracking.
ContentGuard
¶
Heuristic scanner for untrusted content.
Scans text for prompt injection, command injection, path traversal, and cross-agent smuggling patterns. Returns structured findings rather than a boolean — the caller decides how to act on them.
Design note: heuristic detection is not foolproof. This module supplements (not replaces) system prompts, ToolHost policy checks, and user approval workflows.
is_suspicious(text: str) -> bool
classmethod
¶
Return True if any injection pattern is detected.
scan(text: str) -> list[InjectionFinding]
classmethod
¶
Scan text for all injection categories. Returns a list of findings.
wrap(text: str, *, category: ContentCategory, source_identifier: str = '') -> UntrustedContent
classmethod
¶
Scan and wrap text as UntrustedContent in one call.
InjectionFinding
dataclass
¶
A single detected injection attempt.
Parameters¶
category :
Type of injection detected (prompt_injection, command_injection,
path_traversal, cross_agent_smuggle).
severity :
"high" / "medium" / "low" — how dangerous the pattern is.
pattern :
The regex pattern that matched (for audit trails).
match_text :
The substring that triggered the finding (truncated to 128 chars).
position :
Character offset of the match in the original text (-1 if unknown).
PolicyReminder
¶
Generates system-level reminders for tool boundaries.
When untrusted content is injected into the agent's context (e.g., web search results or file contents), this module generates a reminder to prepend that reinforces safety policies. The reminder is proportional to the threat level detected by ContentGuard.
for_tool_output(tool_name: str, output: str) -> tuple[str, UntrustedContent]
classmethod
¶
Scan tool output and generate a reminder + wrapper.
Convenience method for ToolHost integration. Scans the raw tool output, wraps it as UntrustedContent, and returns both the policy reminder text (if any) and the wrapped content.
Returns¶
reminder : Policy reminder string (may be empty if no findings). wrapped : The UntrustedContent wrapper with scan results.
generate(content: UntrustedContent) -> str
classmethod
¶
Generate an appropriate policy reminder for content.
Returns a string to prepend before the untrusted content in the agent's context. Uses the escalated reminder if high-severity findings are present; otherwise uses the default reminder. If no findings at all, returns an empty string (no reminder needed).
UntrustedContent
dataclass
¶
Wrapper marking a string as untrusted external content.
Provides sanitized views, provenance metadata, and safe formatting for injection into the agent's context window.
Parameters¶
raw : The original untrusted text (stored but not used directly in prompts). category : Source classification (web_search, file_read, etc.). source_identifier : Human-readable identifier for the source (e.g., URL, file path). findings : Pre-computed injection findings from ContentGuard.scan().
has_high_severity: bool
property
¶
Return True if any finding is marked "high" severity.
is_suspicious: bool
property
¶
Return True if any injection findings were detected.
for_prompt() -> str
¶
Format content safely for injection into the agent's context.
Wraps the content with a header indicating its untrusted origin and truncates very long content to prevent context overflow.
sanitized() -> str
¶
Return content with HTML/script tags stripped for safe display.
warning_header() -> str
¶
Generate a warning prefix if injection patterns were detected.
Loop Guards¶
core.security.loop_guards
¶
Loop guards — autonomous loop containment and rate limits.
Provides enforcement primitives for auto-chat and autonomous modes so
they cannot run indefinitely. The caller (main.py orchestrator) checks
LoopGuard.should_continue() before each turn, and can use
RateLimiter to throttle per-mode message rates.
Components¶
BudgetExceededError— raised when a loop guard is violatedLoopBudget— frozen dataclass declaring max turns, duration, tool countLoopState— tracks current consumption against a LoopBudgetLoopGuard— stateful guard: record turns/tool errors, check budget, resetRateLimitExceeded— frozen result from RateLimiterRateLimiter— sliding-window rate limiter for per-mode throttling
BudgetExceededError
¶
Bases: Exception
Raised when a loop guard budget is exceeded.
Attributes¶
guard_name : Name of the budget that was exceeded (e.g., ""turns"", ""duration"") current : The value that exceeded the limit limit : The configured maximum
LoopBudget
dataclass
¶
Configurable budget for an autonomous loop.
Parameters¶
max_turns : Maximum number of conversational turns (0 = unlimited). max_duration_seconds : Maximum wall-clock duration in seconds (0 = unlimited). max_tool_calls_per_turn : Maximum tool calls allowed in a single turn (0 = unlimited). max_total_tool_calls : Maximum cumulative tool calls across all turns (0 = unlimited). idle_timeout_seconds : Seconds of inactivity before the loop should stop (0 = no timeout).
is_unlimited: bool
property
¶
Return True if all budgets are disabled (unlimited).
LoopGuard
¶
Stateful guard for autonomous loops.
The caller should check should_continue() before each turn and
after each tool call. If the budget is exceeded, a
BudgetExceededError is raised indicating which limit was hit.
Parameters¶
budget :
The LoopBudget to enforce.
raise_on_exceed :
If True, raise BudgetExceededError on violation.
If False, silently return False from checks.
budget: LoopBudget
property
¶
Configured budget (immutable).
state: LoopState
property
¶
Current consumption state (read-only access).
check_before_tool_call() -> None
¶
Call before executing a tool. Raises if tool budget exceeded.
check_before_turn() -> None
¶
Call before starting a new turn. Raises if turn budget exceeded.
check_duration() -> None
¶
Check elapsed duration against the budget.
check_idle_timeout() -> None
¶
Check idle time against the budget.
record_activity() -> None
¶
Record generic activity (e.g., token generation, TTS synthesis).
record_tool_call() -> None
¶
Record a tool call in the current turn.
record_turn_start() -> None
¶
Record the start of a new conversational turn.
reset() -> None
¶
Reset all counters and timestamps. Use to restart a fresh loop.
should_continue() -> bool
¶
Check all budget limits. Returns False if any are exceeded.
start() -> None
¶
Mark the loop as started, recording the start time.
LoopState
dataclass
¶
Tracks current resource consumption against a LoopBudget.
Updated by the LoopGuard as the loop progresses. The caller reads this to display progress hints in the UI (e.g., ""Turn 3/20"").
Fields¶
turns_completed : Number of completed conversational turns. tool_calls_this_turn : Tool calls in the current turn (resets each turn). total_tool_calls : Cumulative tool calls across all turns. start_time : Monotonic clock time when the loop started (-1 = not started). last_activity_time : Monotonic clock time of the last activity (-1 = none).
RateLimitExceeded
dataclass
¶
Result when a rate limit is exceeded.
Parameters¶
mode : The mode that was rate-limited. max_per_window : The configured maximum events per window. retry_after_seconds : Seconds until the next event can be processed.
RateLimiter
¶
Sliding-window rate limiter for per-mode throttling.
Tracks timestamps of events in a rolling window. When the number
of events in the window exceeds the configured limit, subsequent
calls return RateLimitExceeded instead of accepting.
Parameters¶
max_events : Maximum events allowed within the time window. window_seconds : Sliding window duration in seconds.
allow(mode: str = 'default') -> RateLimitExceeded | None
¶
Check if a new event is allowed for mode.
Returns None if the event is allowed (and records it).
Returns RateLimitExceeded if the limit has been hit.
remaining(mode: str = 'default') -> int
¶
Return the number of events remaining in the current window.
reset(mode: str | None = None) -> None
¶
Clear rate limit history. If mode is None, clear all modes.
Permission Boundaries¶
core.security.permission_boundaries
¶
Network and filesystem permission boundaries.
Enforces default-deny for tool execution: no network, no writes, no traversal unless explicitly allowed by autonomy mode and user grants. Provides path validation, sensitive location blocking, and workspace containment.
Components
PermissionLevel— enum of permission granularitiesPermissionScope— enum of access scope (read, write, network, execute)PathBoundary— workspace root with allowed/denied pathsSensitiveLocation— paths that are blocked by defaultPermissionCheckResult— result of a permission checkPermissionManager— enforces permission boundaries
PathBoundary
dataclass
¶
Workspace root with allowed/denied paths.
Parameters¶
workspace_root : Root directory for all filesystem operations. allowed_subdirs : Additional allowed subdirectories within workspace. denied_subdirs : Subdirectories that are denied even within workspace. allow_symlinks : Whether symbolic links are allowed. max_depth : Maximum directory depth for operations (0 = unlimited).
PermissionCheckResult
dataclass
¶
Result of a permission check.
Parameters¶
allowed : Whether the operation is allowed. scope : The scope that was checked. reason : Human-readable explanation. path : Path that was checked (if applicable). level : Permission level that was checked against.
PermissionLevel
¶
Bases: Enum
Permission granularity for tool execution.
PermissionManager
¶
Enforces permission boundaries for tool execution.
Default-deny: no network, no writes, no traversal unless explicitly allowed by autonomy mode and user grants.
Parameters¶
workspace_root : Root directory for all filesystem operations. permission_level : Default permission level. allowed_domains : Set of allowed network domains. user_grants : Set of user-granted permissions.
check_network(domain: str) -> PermissionCheckResult
¶
check_path(path: str | Path, scope: PermissionScope = PermissionScope.READ) -> PermissionCheckResult
¶
get_violations() -> list[PermissionCheckResult]
¶
Return all violations found so far.
grant(permission: str) -> None
¶
Grant a user permission.
is_granted(permission: str) -> bool
¶
Check if a permission has been granted.
reset() -> None
¶
Clear all violations.
revoke(permission: str) -> None
¶
Revoke a user permission.
PermissionScope
¶
Bases: Enum
Scope of a permission check.
SensitiveLocation
¶
Paths and patterns that are blocked by default.
is_sensitive(path: str | Path) -> bool
classmethod
¶
Check if a path is in a sensitive location.
create_default_permission_manager(*, workspace_root: str | Path | None = None) -> PermissionManager
¶
Create a permission manager with sensible defaults.
Default: READ_ONLY, no network, workspace-contained.
is_local_host(domain: str) -> bool
¶
Check if a domain is localhost.
is_path_traversal(path: str) -> bool
¶
Check if a path contains traversal attempts.
normalize_path(path: str | Path) -> Path
¶
Normalize a path (resolve symlinks, remove traversal).
Provenance Policy¶
core.security.provenance
¶
Model and dependency provenance policy.
Defines how models, plugins, and native dependencies are selected, downloaded, verified, cached, updated, and removed. Includes checksum validation, trusted source manifests, and user consent tracking.
Components
ProvenanceKind— enum of artifact kinds (MODEL, PLUGIN, NATIVE_LIB)ProvenanceRecord— immutable record of a single artifactProvenanceStore— persistent store for provenance recordsSourceManifest— trusted source registryChecksumError/ProvenanceError— exception hierarchyverify_checksum()— sha256 verification utilityartifact_path()— canonical cache directory layout
ArtifactStatus
¶
Bases: Enum
Lifecycle status of a tracked artifact.
ChecksumError
¶
Bases: ProvenanceError
Checksum verification failed for a downloaded artifact.
ConsentRequiredError
¶
Bases: ProvenanceError
User consent is required before proceeding.
ProvenanceError
¶
Bases: Exception
Base exception for provenance policy violations.
ProvenanceKind
¶
Bases: Enum
Kind of artifact being tracked.
ProvenanceRecord
dataclass
¶
Immutable record of a single artifact's provenance.
Parameters¶
name :
Human-readable name (e.g. "qwen3-embedding:0.6b").
kind :
Kind of artifact (MODEL, PLUGIN, NATIVE_LIB).
source :
URL or source identifier (e.g. "https://..." or "ollama").
version :
Semantic version or tag string.
checksum :
Expected SHA-256 hex digest (empty string if not yet known).
path :
Local file system path where the artifact is stored.
status :
Current lifecycle status.
installed_at :
ISO-8601 timestamp of installation (empty if not yet installed).
consent_given :
Whether the user explicitly consented to this artifact.
consent_at :
ISO-8601 timestamp of consent (empty if not yet given).
metadata :
Arbitrary additional context (architecture, license, size_bytes).
ProvenanceStore
¶
Persistent store for provenance records.
Manages the lifecycle of artifact provenance: adding records, looking them up, updating status, recording consent, and removal.
Parameters¶
storage_path :
File path for JSON persistence (None = in-memory only).
manifest :
Optional trusted source manifest for validation.
manifest: SourceManifest
property
¶
Return the trusted source manifest.
add(name: str, kind: ProvenanceKind, source: str, version: str, *, checksum: str = '', path: str = '', metadata: dict[str, Any] | None = None) -> ProvenanceRecord
¶
Add a new artifact record.
Raises¶
UntrustedSourceError If the source is not in the trusted manifest.
check_consent(name: str) -> bool
¶
Check whether user consent has been given for an artifact.
get(name: str) -> ProvenanceRecord | None
¶
Look up an artifact record by name.
list_records(kind: ProvenanceKind | None = None, status: ArtifactStatus | None = None) -> list[ProvenanceRecord]
¶
List artifact records, optionally filtered by kind and/or status.
record_consent(name: str) -> ProvenanceRecord | None
¶
Record explicit user consent for an artifact.
remove(name: str) -> bool
¶
Remove an artifact record. Returns True if it existed.
save_manifest(path: str | Path | None = None) -> None
¶
Persist the trusted source manifest to a JSON file.
update_status(name: str, status: ArtifactStatus) -> ProvenanceRecord | None
¶
Update an artifact's status. Returns the updated record.
validate(name: str) -> list[str]
¶
Validate an artifact record. Returns a list of warnings.
SourceManifest
dataclass
¶
Registry of trusted artifact sources.
Tracks which URLs/domains are considered trusted for downloading models, plugins, and native libraries.
Parameters¶
name :
Human-readable manifest name (e.g. "default").
trusted_sources :
Set of trusted source URLs or domain patterns.
allowed_checksums :
Mapping of artifact name → expected SHA-256 hex digest.
metadata :
Arbitrary additional context (created_at, updated_at, author).
add_source(source: str) -> None
¶
Add a trusted source (idempotent).
get_checksum(name: str) -> str | None
¶
Get the expected checksum for an artifact, or None.
is_trusted(source: str) -> bool
¶
Check if a source URL is in the trusted list.
remove_source(source: str) -> bool
¶
Remove a trusted source. Returns True if it was present.
set_checksum(name: str, checksum: str) -> None
¶
Set the expected checksum for an artifact.
UntrustedSourceError
¶
Bases: ProvenanceError
Attempted to fetch from an untrusted source.
artifact_path(cache_dir: str | Path, kind: ProvenanceKind, name: str) -> Path
¶
Compute the canonical cache path for an artifact.
Layout::
<cache_dir>/<kind_value>/<name>/
Parameters¶
cache_dir : Root directory for cached artifacts. kind : Kind of the artifact. name : Name of the artifact (used as a subdirectory).
Returns¶
Path The directory where the artifact should be stored.
compute_checksum(file_path: str | Path) -> str
¶
Compute the SHA-256 checksum of a file.
default_manifest() -> SourceManifest
¶
Create a default source manifest with common trusted sources.
Includes Ollama library, PyPI, and conda-forge as trusted sources.
verify_checksum(file_path: str | Path, expected: str) -> bool
¶
Verify a file's SHA-256 checksum against an expected value.
Parameters¶
file_path : Path to the file to verify. expected : Expected SHA-256 hex digest (64 characters).
Returns¶
bool
True if the checksum matches.
Raises¶
ChecksumError If the checksum does not match. FileNotFoundError If the file does not exist.