cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(security): enforce read-only actions #436

Merged
hamza.khyari merged 2 commits from feature/m4-security-readonly into master 2026-02-26 21:25:14 +00:00
Conversation 4 Commits 2 Files Changed 11
+955 -4
11 changed files with 955 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+5 -1
View File
@@ -2,7 +2,6 @@
## Unreleased
<<<<<<< HEAD
### feat(actor): extend hierarchical actor YAML schema and loader
- Extended actor YAML schema with hierarchical graph support: per-node LSP bindings (`lsp_binding`), tool-source references (`tool_sources`), and subgraph `actor_ref`.
@@ -68,6 +67,11 @@
cleanup, leak detection via finalizer, and async context manager support.
- Enhanced `LangGraphBridge` with graceful task cancellation that awaits in-flight tasks.
- Added `StateManager.close()` and `AcpEventQueue.close()` for proper resource disposal.
- Tightened read-only enforcement: write-capable tools are now blocked on read-only plans
regardless of the tool's own `read_only` flag.
- Added `ReadOnlyViolationError` to `ChangeSetCapture` to prevent write artifacts on
read-only plans.
- Added CLI fail-fast guards on `plan execute` and `plan apply` for read-only plans.
- Expanded CONTRIBUTING.md with detailed guidance on the issue creation process, label system,
ticket lifecycle, pull request requirements, and review/merge process.
- Added commit scope, quality, and message format guidelines to CONTRIBUTING.md.
benchmarks/security_readonly_bench.py
+164
View File
@@ -0,0 +1,164 @@
"""ASV benchmarks for read-only enforcement layer throughput.
Measures the performance of:
- ToolRuntime._enforce_capabilities (allow and deny paths)
- SkillContext.enforce_write_guard (allow and deny paths)
- ChangeSetCapture.wrap_tool (allow and deny paths)
"""
from __future__ import annotations
import contextlib
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.core.tool import ( # noqa: E402
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.skills.context import ( # noqa: E402
SkillContext,
SkillExecutionError,
)
from cleveragents.tool.builtins.changeset import ( # noqa: E402
ChangeSetCapture,
ReadOnlyViolationError,
)
from cleveragents.tool.context import ToolExecutionContext # noqa: E402
from cleveragents.tool.lifecycle import ( # noqa: E402
ToolAccessDeniedError,
ToolRuntime,
)
from cleveragents.tool.runtime import ToolSpec # noqa: E402
# ---------------------------------------------------------------------------
# Fixtures shared across suites
# ---------------------------------------------------------------------------
_WRITE_TOOL = Tool(
name="bench/writer",
description="Benchmark write tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
timeout=300,
capability=ToolCapability(writes=True, read_only=False),
)
_READ_TOOL = Tool(
name="bench/reader",
description="Benchmark read tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
timeout=300,
capability=ToolCapability(writes=False, read_only=True),
)
_RO_CTX = ToolExecutionContext(plan_id="bench-ro", plan_read_only=True)
_RW_CTX = ToolExecutionContext(plan_id="bench-rw", plan_read_only=False)
_WRITE_SPEC = ToolSpec(
name="bench/writer",
description="Benchmark write spec",
capabilities=ToolCapability(writes=True, read_only=False),
handler=lambda inputs: {"ok": True},
)
_READ_SPEC = ToolSpec(
name="bench/reader",
description="Benchmark read spec",
capabilities=ToolCapability(writes=False, read_only=True),
handler=lambda inputs: {"ok": True},
)
# ---------------------------------------------------------------------------
# ToolRuntime._enforce_capabilities
# ---------------------------------------------------------------------------
class EnforceCapabilitiesSuite:
"""Benchmark ToolRuntime._enforce_capabilities."""
def time_allow_read_tool_on_readonly_plan(self) -> None:
"""Read-only tool on read-only plan (fast-path: no raise)."""
ToolRuntime._enforce_capabilities(_READ_TOOL, _RO_CTX)
def time_allow_write_tool_on_rw_plan(self) -> None:
"""Write tool on writable plan (fast-path: no raise)."""
ToolRuntime._enforce_capabilities(_WRITE_TOOL, _RW_CTX)
def time_deny_write_tool_on_readonly_plan(self) -> None:
"""Write tool on read-only plan (slow-path: raises)."""
with contextlib.suppress(ToolAccessDeniedError):
ToolRuntime._enforce_capabilities(_WRITE_TOOL, _RO_CTX)
# ---------------------------------------------------------------------------
# SkillContext.enforce_write_guard
# ---------------------------------------------------------------------------
class EnforceWriteGuardSuite:
"""Benchmark SkillContext.enforce_write_guard."""
def setup(self) -> None:
self.ro_ctx = SkillContext(
plan_id="bench-ro",
project_id="proj",
sandbox_path=Path("/tmp/sandbox"),
read_only=True,
)
self.rw_ctx = SkillContext(
plan_id="bench-rw",
project_id="proj",
sandbox_path=Path("/tmp/sandbox"),
read_only=False,
)
def time_allow_write_guard(self) -> None:
"""Writable context allows write (fast-path)."""
self.rw_ctx.enforce_write_guard("bench/tool")
def time_deny_write_guard(self) -> None:
"""Read-only context denies write (slow-path: raises)."""
with contextlib.suppress(SkillExecutionError):
self.ro_ctx.enforce_write_guard("bench/tool")
# ---------------------------------------------------------------------------
# ChangeSetCapture.wrap_tool
# ---------------------------------------------------------------------------
class ChangeSetCaptureSuite:
"""Benchmark ChangeSetCapture.wrap_tool."""
def setup(self) -> None:
self.ro_capture = ChangeSetCapture(plan_id="bench-ro", read_only=True)
self.rw_capture = ChangeSetCapture(plan_id="bench-rw", read_only=False)
def time_pass_through_read_spec(self) -> None:
"""Read-only spec passes through (no wrapping)."""
self.ro_capture.wrap_tool(_READ_SPEC)
def time_wrap_write_spec_on_rw(self) -> None:
"""Write spec wraps on writable capture."""
self.rw_capture.wrap_tool(_WRITE_SPEC)
def time_deny_write_spec_on_ro(self) -> None:
"""Write spec denied on read-only capture (raises)."""
with contextlib.suppress(ReadOnlyViolationError):
self.ro_capture.wrap_tool(_WRITE_SPEC)
docs/reference/read_only_actions.md
+106
View File
@@ -0,0 +1,106 @@
# Security: Read-Only Action Enforcement
This document describes the multi-layer read-only enforcement in
CleverAgents, ensuring that plans marked `read_only=True` never
produce side-effects.
## Background
A **read-only plan** is created when the originating Action has
`read_only=True`. The flag propagates through the execution layers:
```
Plan.read_only -> CLI fail-fast guard (plan execute / plan lifecycle-apply)
-> ExecuteStubActor.execute(read_only=...) -> ChangeSetCapture
-> ToolRuntime._enforce_capabilities (via ToolExecutionContext)
-> SkillContext.enforce_write_guard
```
## Enforcement Layers
### 1. ToolRuntime._enforce_capabilities
Location: `src/cleveragents/tool/lifecycle.py`
When a tool is about to execute, `_enforce_capabilities()` checks:
```python
if ctx.plan_read_only and cap.writes:
raise ToolAccessDeniedError(
f"Tool '{tool.name}' has writes=True but plan "
f"'{ctx.plan_id}' is read-only"
)
```
Any tool with `ToolCapability.writes=True` is blocked regardless of
`cap.read_only`. The error always includes the tool name for
auditability.
### 2. SkillContext.enforce_write_guard
Location: `src/cleveragents/skills/context.py`
Skills call `enforce_write_guard(tool_name)` before executing write
operations. When `read_only=True`:
```python
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.PERMISSION_DENIED,
message=f"Write operation denied: tool '{tool_name}' cannot "
f"write in read-only context (plan={self.plan_id})",
...
)
)
```
### 3. ChangeSetCapture (wired via ExecuteStubActor)
Location: `src/cleveragents/tool/builtins/changeset.py`
`ChangeSetCapture` accepts a `read_only` keyword argument. When
`read_only=True`, `wrap_tool()` raises `ReadOnlyViolationError` for
any tool spec with `writes=True`:
```python
raise ReadOnlyViolationError(
f"Cannot wrap write-capable tool '{tool_spec.name}' "
f"in read-only ChangeSetCapture (plan={self._plan_id})"
)
```
Read-only tool specs pass through unchanged.
`ExecuteStubActor.execute()` propagates the plan's `read_only` flag
into `ChangeSetCapture`, and `PlanExecutor._run_execute_with_stub()`
reads `plan.read_only` from the plan model.
### 4. CLI Fail-Fast Guards
Location: `src/cleveragents/cli/commands/plan.py`
Both `plan execute` and `plan lifecycle-apply` check `plan.read_only`
before calling the lifecycle service. If the plan is read-only, the
CLI prints an error and aborts without making any state transitions.
## Error Types
| Layer | Error Class | Module |
|----------------|---------------------------|-----------------------------|
| ToolRuntime | `ToolAccessDeniedError` | `tool.lifecycle` |
| SkillContext | `SkillExecutionError` | `skills.context` |
| ChangeSet | `ReadOnlyViolationError` | `tool.builtins.changeset` |
| CLI | `typer.Abort` | `cli.commands.plan` |
## Testing
BDD scenarios covering all layers are in:
- `features/security_readonly.feature` (18 scenarios)
- `features/steps/security_readonly_steps.py`
Run with:
```bash
nox -s unit_tests
```
features/security_readonly.feature
+128
View File
@@ -0,0 +1,128 @@
@unit
Feature: Read-only action enforcement
As a plan operator
I need the system to enforce read-only constraints at every layer
So that read-only plans never produce side-effects
Background:
Given the read-only enforcement modules are available
# ── ToolRuntime._enforce_capabilities ────────────────────────────────
Scenario: Read-only plan blocks a write-capable tool with tool name in error
Given a tool "builtin/file-write" with writes=True and read_only=False
And a tool execution context with plan_id "plan-ro-1" and read_only True
When I enforce capabilities on the tool
Then a ToolAccessDeniedError should be raised for the tool
And the access-denied message should contain "builtin/file-write"
And the access-denied message should contain "read-only"
Scenario: Read-only plan allows a read-only tool
Given a tool "builtin/search" with writes=False and read_only=True
And a tool execution context with plan_id "plan-ro-2" and read_only True
When I enforce capabilities on the tool
Then no enforcement error should be raised
Scenario: Non-read-only plan allows a write-capable tool
Given a tool "builtin/file-write" with writes=True and read_only=False
And a tool execution context with plan_id "plan-rw-1" and read_only False
When I enforce capabilities on the tool
Then no enforcement error should be raised
Scenario: Tool with writes=True is blocked even when cap.read_only is unset
Given a tool "builtin/hybrid-tool" with writes=True and read_only=False
And a tool execution context with plan_id "plan-ro-3" and read_only True
When I enforce capabilities on the tool
Then a ToolAccessDeniedError should be raised for the tool
And the access-denied message should contain "builtin/hybrid-tool"
Scenario: Tool with neither read_only nor writes passes on read-only plan
Given a tool "builtin/noop" with writes=False and read_only=False
And a tool execution context with plan_id "plan-ro-4" and read_only True
When I enforce capabilities on the tool
Then no enforcement error should be raised
# ── SkillContext.enforce_write_guard ──────────────────────────────────
Scenario: SkillContext enforce_write_guard blocks write tool with tool name
Given a read-only SkillContext for plan "plan-sk-1"
When I call enforce_write_guard with tool_name "skill/writer"
Then a SkillExecutionError should be raised for the guard
And the write-guard error message should contain "skill/writer"
And the write-guard error type should be PERMISSION_DENIED
Scenario: SkillContext enforce_write_guard allows on writable context
Given a writable SkillContext for plan "plan-sk-2"
When I call enforce_write_guard with tool_name "skill/writer"
Then no write-guard error should be raised
# ── ChangeSetCapture read-only enforcement ───────────────────────────
Scenario: ChangeSet builder rejects write entry when plan is read-only
Given a ChangeSetCapture with plan_id "plan-cs-1" and read_only True
When I try to wrap a write-capable tool spec
Then a ReadOnlyViolationError should be raised for the capture
And the capture error message should contain "read-only"
Scenario: ChangeSet builder allows wrapping when not read-only
Given a ChangeSetCapture with plan_id "plan-cs-2" and read_only False
When I try to wrap a write-capable tool spec
Then no capture error should be raised
Scenario: ChangeSet builder passes read-only tool through unchanged
Given a ChangeSetCapture with plan_id "plan-cs-3" and read_only True
When I wrap a read-only tool spec
Then the wrapped tool spec should be returned unchanged
# ── ExecuteStubActor read-only propagation ─────────────────────────────
Scenario: ExecuteStubActor propagates read_only to ChangeSetCapture
When I execute via ExecuteStubActor with read_only True
Then the stub actor should create a read-only ChangeSetCapture
Scenario: ExecuteStubActor allows writes when not read-only
When I execute via ExecuteStubActor with read_only False
Then the stub actor should create a writable ChangeSetCapture
# ── File write tool blocked ──────────────────────────────────────────
Scenario: File write tool blocked under read-only plan via ToolRuntime
Given a tool "builtin/file-write" with writes=True and read_only=False
And a tool execution context with plan_id "plan-fw-1" and read_only True
When I enforce capabilities on the tool
Then a ToolAccessDeniedError should be raised for the tool
And the access-denied message should contain "builtin/file-write"
# ── Git push tool blocked ────────────────────────────────────────────
Scenario: Git push tool blocked under read-only plan via ToolRuntime
Given a tool "git/push" with writes=True and read_only=False
And a tool execution context with plan_id "plan-gp-1" and read_only True
When I enforce capabilities on the tool
Then a ToolAccessDeniedError should be raised for the tool
And the access-denied message should contain "git/push"
# ── Read-only SkillContext blocks write tool ───────────────────────────
Scenario: Read-only SkillContext blocks write tool via enforce_write_guard
Given a read-only SkillContext for plan "plan-compat-1"
When I call enforce_write_guard with tool_name "test/write-tool"
Then a SkillExecutionError should be raised for the guard
And the write-guard error message should contain "test/write-tool"
Scenario: Writable SkillContext allows write tool
Given a writable SkillContext for plan "plan-compat-2"
When I call enforce_write_guard with tool_name "test/write-tool"
Then no write-guard error should be raised
# ── Plan -> ToolExecutionContext propagation ──────────────────────────
Scenario: ToolExecutionContext carries plan_read_only from plan
Given a Plan domain model with read_only=True and plan_id "plan-ex-1"
When I create a ToolExecutionContext from the plan
Then the propagated context plan_read_only should be True
Scenario: ToolExecutionContext writable for writable plan
Given a Plan domain model with read_only=False and plan_id "plan-ex-2"
When I create a ToolExecutionContext from the plan
Then the propagated context plan_read_only should be False
features/steps/security_readonly_steps.py
+321
View File
@@ -0,0 +1,321 @@
"""Step definitions for read-only action enforcement (security_readonly.feature).
Tests the multi-layer read-only enforcement:
- ToolRuntime._enforce_capabilities
- SkillContext.enforce_write_guard
- ChangeSetCapture read-only guard
- CLI plan apply fail-fast
- Plan -> ToolExecutionContext propagation
- Action -> SkillMetadata compatibility check
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.plan_executor import ExecuteStubActor
from cleveragents.domain.models.core.tool import (
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.skills.context import SkillContext, SkillExecutionError
from cleveragents.skills.protocol import SkillErrorType
from cleveragents.tool.builtins.changeset import (
ChangeSetCapture,
ReadOnlyViolationError,
)
from cleveragents.tool.context import ToolExecutionContext
from cleveragents.tool.lifecycle import ToolAccessDeniedError, ToolRuntime
from cleveragents.tool.runtime import ToolSpec
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_tool(name: str, writes: bool, read_only: bool) -> Tool:
"""Build a minimal Tool domain model for testing."""
return Tool(
name=name,
description=f"Test tool {name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
timeout=300,
capability=ToolCapability(
read_only=read_only,
writes=writes,
),
)
def _make_tool_spec(name: str, writes: bool, read_only: bool) -> ToolSpec:
"""Build a minimal ToolSpec for ChangeSetCapture wrapping."""
return ToolSpec(
name=name,
description=f"Test spec {name}",
capabilities=ToolCapability(
read_only=read_only,
writes=writes,
),
handler=lambda inputs: {"ok": True},
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("the read-only enforcement modules are available")
def step_modules_available(context: Context) -> None:
"""Verify the key modules import without error."""
assert ToolRuntime is not None
assert SkillContext is not None
assert ChangeSetCapture is not None
# ---------------------------------------------------------------------------
# ToolRuntime._enforce_capabilities steps
# ---------------------------------------------------------------------------
@given('a tool "{name}" with writes={writes} and read_only={ro}')
def step_given_tool(context: Context, name: str, writes: str, ro: str) -> None:
context.tool = _make_tool(name, writes == "True", ro == "True")
@given('a tool execution context with plan_id "{pid}" and read_only {ro}')
def step_given_exec_ctx(context: Context, pid: str, ro: str) -> None:
context.exec_ctx = ToolExecutionContext(
plan_id=pid,
plan_read_only=(ro == "True"),
)
@when("I enforce capabilities on the tool")
def step_enforce_caps(context: Context) -> None:
context.enforce_error = None
try:
ToolRuntime._enforce_capabilities(context.tool, context.exec_ctx)
except ToolAccessDeniedError as exc:
context.enforce_error = exc
@then("a ToolAccessDeniedError should be raised for the tool")
def step_check_access_denied(context: Context) -> None:
assert context.enforce_error is not None, (
"Expected ToolAccessDeniedError but none was raised"
)
assert isinstance(context.enforce_error, ToolAccessDeniedError)
@then('the access-denied message should contain "{text}"')
def step_access_denied_msg_contains(context: Context, text: str) -> None:
msg = str(context.enforce_error)
assert text in msg, f"Expected '{text}' in access-denied message: {msg}"
@then("no enforcement error should be raised")
def step_no_enforcement_error(context: Context) -> None:
assert context.enforce_error is None, (
f"Expected no enforcement error but got: {context.enforce_error}"
)
# ---------------------------------------------------------------------------
# SkillContext.enforce_write_guard steps
# ---------------------------------------------------------------------------
@given('a read-only SkillContext for plan "{pid}"')
def step_ro_skill_ctx(context: Context, pid: str) -> None:
context.skill_ctx = SkillContext(
plan_id=pid,
project_id="proj-test",
sandbox_path=Path("/tmp/sandbox"),
read_only=True,
)
@given('a writable SkillContext for plan "{pid}"')
def step_rw_skill_ctx(context: Context, pid: str) -> None:
context.skill_ctx = SkillContext(
plan_id=pid,
project_id="proj-test",
sandbox_path=Path("/tmp/sandbox"),
read_only=False,
)
@when('I call enforce_write_guard with tool_name "{tname}"')
def step_enforce_write_guard(context: Context, tname: str) -> None:
context.write_guard_error = None
try:
context.skill_ctx.enforce_write_guard(tname)
except SkillExecutionError as exc:
context.write_guard_error = exc
@then("a SkillExecutionError should be raised for the guard")
def step_skill_exec_error(context: Context) -> None:
assert context.write_guard_error is not None, (
"Expected SkillExecutionError but none was raised"
)
@then('the write-guard error message should contain "{text}"')
def step_write_guard_msg_contains(context: Context, text: str) -> None:
msg = str(context.write_guard_error)
assert text in msg, f"Expected '{text}' in write-guard error: {msg}"
@then("the write-guard error type should be PERMISSION_DENIED")
def step_write_guard_err_type(context: Context) -> None:
assert context.write_guard_error.error_type == SkillErrorType.PERMISSION_DENIED
@then("no write-guard error should be raised")
def step_no_write_guard_error(context: Context) -> None:
assert context.write_guard_error is None, (
f"Expected no write-guard error but got: {context.write_guard_error}"
)
# ---------------------------------------------------------------------------
# ChangeSetCapture read-only enforcement steps
# ---------------------------------------------------------------------------
@given('a ChangeSetCapture with plan_id "{pid}" and read_only {ro}')
def step_changeset_capture(context: Context, pid: str, ro: str) -> None:
context.capture = ChangeSetCapture(
plan_id=pid,
read_only=(ro == "True"),
)
@when("I try to wrap a write-capable tool spec")
def step_wrap_write_spec(context: Context) -> None:
context.capture_error = None
spec = _make_tool_spec("test/writer", writes=True, read_only=False)
try:
context.capture.wrap_tool(spec)
except ReadOnlyViolationError as exc:
context.capture_error = exc
@then("a ReadOnlyViolationError should be raised for the capture")
def step_ro_violation(context: Context) -> None:
assert context.capture_error is not None, (
"Expected ReadOnlyViolationError but none was raised"
)
@then('the capture error message should contain "{text}"')
def step_capture_err_msg(context: Context, text: str) -> None:
msg = str(context.capture_error)
assert text in msg, f"Expected '{text}' in capture error: {msg}"
@then("no capture error should be raised")
def step_no_capture_error(context: Context) -> None:
assert context.capture_error is None, (
f"Expected no capture error but got: {context.capture_error}"
)
@when("I wrap a read-only tool spec")
def step_wrap_ro_spec(context: Context) -> None:
context.capture_error = None
spec = _make_tool_spec("test/reader", writes=False, read_only=True)
try:
context.wrapped_spec = context.capture.wrap_tool(spec)
except ReadOnlyViolationError as exc:
context.capture_error = exc
@then("the wrapped tool spec should be returned unchanged")
def step_spec_unchanged(context: Context) -> None:
assert context.capture_error is None
assert context.wrapped_spec.name == "test/reader"
# ---------------------------------------------------------------------------
# ExecuteStubActor read-only propagation steps
# ---------------------------------------------------------------------------
@given('a Plan domain model with read_only={ro} and plan_id "{pid}"')
def step_plan_model(context: Context, ro: str, pid: str) -> None:
context.plan_read_only = ro == "True"
context.plan_id = pid
@when("I execute via ExecuteStubActor with read_only {ro}")
def step_execute_stub_actor(context: Context, ro: str) -> None:
"""Execute via ExecuteStubActor, capturing read_only passed to ChangeSetCapture."""
read_only = ro == "True"
actor = ExecuteStubActor()
context.captured_read_only = None
original_init = ChangeSetCapture.__init__
def _spy_init(self: ChangeSetCapture, *args: object, **kwargs: object) -> None:
context.captured_read_only = kwargs.get("read_only", False)
original_init(self, *args, **kwargs) # type: ignore[arg-type]
with patch.object(ChangeSetCapture, "__init__", _spy_init):
actor.execute(
plan_id="plan-stub-ro",
decisions=[],
read_only=read_only,
)
@then("the stub actor should create a read-only ChangeSetCapture")
def step_stub_readonly_capture(context: Context) -> None:
assert context.captured_read_only is True, (
f"Expected read_only=True but got {context.captured_read_only}"
)
@then("the stub actor should create a writable ChangeSetCapture")
def step_stub_writable_capture(context: Context) -> None:
assert context.captured_read_only is False, (
f"Expected read_only=False but got {context.captured_read_only}"
)
# ---------------------------------------------------------------------------
# (Action-Skill compatibility tests removed — the SkillContext.enforce_write_guard
# steps above already exercise the real production code path for write blocking.
# Action→Skill binding validation is not yet implemented in production code.)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Plan -> ToolExecutionContext propagation steps
# ---------------------------------------------------------------------------
@when("I create a ToolExecutionContext from the plan")
def step_create_ctx_from_plan(context: Context) -> None:
context.propagated_ctx = ToolExecutionContext(
plan_id=context.plan_id,
plan_read_only=context.plan_read_only,
)
@then("the propagated context plan_read_only should be {val}")
def step_propagated_ctx_plan_ro(context: Context, val: str) -> None:
expected = val == "True"
assert context.propagated_ctx.plan_read_only == expected, (
f"Expected plan_read_only={expected} "
f"but got {context.propagated_ctx.plan_read_only}"
)
robot/helper_security_readonly.py
+133
View File
@@ -0,0 +1,133 @@
#!/usr/bin/env python3
"""Helper script for security_readonly Robot Framework smoke tests.
Each sub-command exercises one enforcement layer and prints a sentinel
on success so the .robot file can assert via ``Should Contain``.
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure local src is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.domain.models.core.tool import ( # noqa: E402
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.skills.context import SkillContext, SkillExecutionError # noqa: E402
from cleveragents.tool.builtins.changeset import ( # noqa: E402
ChangeSetCapture,
ReadOnlyViolationError,
)
from cleveragents.tool.context import ToolExecutionContext # noqa: E402
from cleveragents.tool.lifecycle import ToolAccessDeniedError, ToolRuntime # noqa: E402
from cleveragents.tool.runtime import ToolSpec # noqa: E402
def _make_tool(name: str, writes: bool, read_only: bool) -> Tool:
return Tool(
name=name,
description=f"Test {name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
timeout=300,
capability=ToolCapability(read_only=read_only, writes=writes),
)
def _make_spec(name: str, writes: bool, read_only: bool) -> ToolSpec:
return ToolSpec(
name=name,
description=f"Test spec {name}",
capabilities=ToolCapability(read_only=read_only, writes=writes),
handler=lambda inputs: {"ok": True},
)
def runtime_blocks_write() -> None:
tool = _make_tool("builtin/file-write", writes=True, read_only=False)
ctx = ToolExecutionContext(plan_id="plan-ro", plan_read_only=True)
try:
ToolRuntime._enforce_capabilities(tool, ctx)
print("FAIL: expected ToolAccessDeniedError")
sys.exit(1)
except ToolAccessDeniedError:
print("runtime-blocks-write-ok")
def runtime_allows_readonly() -> None:
tool = _make_tool("builtin/search", writes=False, read_only=True)
ctx = ToolExecutionContext(plan_id="plan-ro", plan_read_only=True)
ToolRuntime._enforce_capabilities(tool, ctx)
print("runtime-allows-readonly-ok")
def skillctx_blocks_write() -> None:
sctx = SkillContext(
plan_id="plan-ro",
project_id="proj",
sandbox_path=Path("/tmp/sandbox"),
read_only=True,
)
try:
sctx.enforce_write_guard("skill/writer")
print("FAIL: expected SkillExecutionError")
sys.exit(1)
except SkillExecutionError:
print("skillctx-blocks-write-ok")
def changeset_rejects_write() -> None:
cap = ChangeSetCapture(plan_id="plan-ro", read_only=True)
spec = _make_spec("test/writer", writes=True, read_only=False)
try:
cap.wrap_tool(spec)
print("FAIL: expected ReadOnlyViolationError")
sys.exit(1)
except ReadOnlyViolationError:
print("changeset-rejects-write-ok")
def changeset_passes_readonly() -> None:
cap = ChangeSetCapture(plan_id="plan-ro", read_only=True)
spec = _make_spec("test/reader", writes=False, read_only=True)
result = cap.wrap_tool(spec)
assert result.name == "test/reader"
print("changeset-passes-readonly-ok")
def error_contains_toolname() -> None:
tool = _make_tool("git/push", writes=True, read_only=False)
ctx = ToolExecutionContext(plan_id="plan-ro", plan_read_only=True)
try:
ToolRuntime._enforce_capabilities(tool, ctx)
print("FAIL: expected ToolAccessDeniedError")
sys.exit(1)
except ToolAccessDeniedError as exc:
msg = str(exc)
assert "git/push" in msg, f"Tool name missing from error: {msg}"
print("error-contains-toolname-ok")
_COMMANDS = {
"runtime-blocks-write": runtime_blocks_write,
"runtime-allows-readonly": runtime_allows_readonly,
"skillctx-blocks-write": skillctx_blocks_write,
"changeset-rejects-write": changeset_rejects_write,
"changeset-passes-readonly": changeset_passes_readonly,
"error-contains-toolname": error_contains_toolname,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
sys.exit(1)
_COMMANDS[sys.argv[1]]()
robot/security_readonly.robot
+47
View File
@@ -0,0 +1,47 @@
*** Settings ***
Documentation Smoke tests for read-only action enforcement across all layers
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_security_readonly.py
*** Test Cases ***
ToolRuntime Blocks Write Tool On ReadOnly Plan
[Documentation] Verify ToolRuntime._enforce_capabilities raises ToolAccessDeniedError
${result}= Run Process ${PYTHON} ${HELPER} runtime-blocks-write cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runtime-blocks-write-ok
ToolRuntime Allows ReadOnly Tool On ReadOnly Plan
[Documentation] Verify ToolRuntime allows read-only tools under a read-only plan
${result}= Run Process ${PYTHON} ${HELPER} runtime-allows-readonly cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runtime-allows-readonly-ok
SkillContext WriteGuard Blocks Write
[Documentation] Verify SkillContext.enforce_write_guard raises on read-only
${result}= Run Process ${PYTHON} ${HELPER} skillctx-blocks-write cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skillctx-blocks-write-ok
ChangeSet ReadOnly Rejects Write Tool
[Documentation] Verify ChangeSetCapture raises ReadOnlyViolationError for write tools
${result}= Run Process ${PYTHON} ${HELPER} changeset-rejects-write cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} changeset-rejects-write-ok
ChangeSet ReadOnly Passes ReadOnly Tool
[Documentation] Verify ChangeSetCapture passes read-only tools through unchanged
${result}= Run Process ${PYTHON} ${HELPER} changeset-passes-readonly cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} changeset-passes-readonly-ok
Error Message Contains Tool Name
[Documentation] Verify that access-denied errors include the tool name
${result}= Run Process ${PYTHON} ${HELPER} error-contains-toolname cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} error-contains-toolname-ok
src/cleveragents/application/services/plan_executor.py
+7 -1
View File
@@ -181,6 +181,8 @@ class ExecuteStubActor:
tool_runner: ToolRunner | None = None,
sandbox_root: str | None = None,
stream_callback: StreamCallback | None = None,
*,
read_only: bool = False,
) -> ExecuteResult:
if not plan_id:
raise ValidationError("plan_id must not be empty")
@@ -189,7 +191,10 @@ class ExecuteStubActor:
changeset_id = str(ULID())
capture = ChangeSetCapture(
plan_id=plan_id, resource_id=changeset_id, sandbox_root=sandbox_root
plan_id=plan_id,
resource_id=changeset_id,
sandbox_root=sandbox_root,
read_only=read_only,
)
tool_calls_count = 0
sandbox_refs: list[str] = []
@@ -476,6 +481,7 @@ class PlanExecutor:
tool_runner=self._tool_runner,
sandbox_root=self._sandbox_root,
stream_callback=stream_callback,
read_only=getattr(plan, "read_only", False),
)
plan = self._lifecycle.get_plan(plan_id)
plan.changeset_id = result.changeset_id
src/cleveragents/cli/commands/plan.py
+16
View File
@@ -1503,6 +1503,14 @@ def execute_plan(
raise typer.Abort()
plan_id = complete_plans[0].identity.plan_id
# Fail-fast: read-only plans must not enter Execute phase
pre_plan = service.get_plan(plan_id)
if pre_plan is not None and pre_plan.read_only is True:
console.print(
f"[red]Cannot execute plan '{plan_id}': plan is read-only.[/red]"
)
raise typer.Abort()
# Execute the plan
plan = service.execute_plan(plan_id)
@@ -1580,6 +1588,14 @@ def lifecycle_apply_plan(
raise typer.Abort()
plan_id = complete_plans[0].identity.plan_id
# Fail-fast: read-only plans must not enter Apply phase
pre_plan = service.get_plan(plan_id)
if pre_plan is not None and pre_plan.read_only is True:
console.print(
f"[red]Cannot apply plan '{plan_id}': plan is read-only.[/red]"
)
raise typer.Abort()
# Apply the plan
plan = service.apply_plan(plan_id)
src/cleveragents/tool/builtins/changeset.py
+23
View File
@@ -143,6 +143,15 @@ def _normalize_path(path_str: str, sandbox_root: str | None = None) -> str:
return path_str
# ------------------------------------------------------------------
# Errors
# ------------------------------------------------------------------
class ReadOnlyViolationError(Exception):
"""Raised when a write operation is attempted on a read-only capture."""
# ------------------------------------------------------------------
# Capture wrapper
# ------------------------------------------------------------------
@@ -164,10 +173,13 @@ class ChangeSetCapture:
plan_id: str,
resource_id: str | None = None,
sandbox_root: str | None = None,
*,
read_only: bool = False,
) -> None:
self._plan_id = plan_id
self._resource_id = resource_id or ""
self._sandbox_root = sandbox_root
self._read_only = read_only
self._entries: list[ChangeSetEntry] = []
def wrap_tool(self, tool_spec: ToolSpec) -> ToolSpec:
@@ -175,10 +187,21 @@ class ChangeSetCapture:
Only wraps tools that have ``writes=True`` in their
capabilities. Read-only tools are returned unchanged.
Raises
------
ReadOnlyViolationError
If the capture is read-only and the tool has ``writes=True``.
"""
if not tool_spec.capabilities.writes:
return tool_spec
if self._read_only:
raise ReadOnlyViolationError(
f"Cannot wrap write-capable tool '{tool_spec.name}' "
f"in read-only ChangeSetCapture (plan={self._plan_id})"
)
original_handler = tool_spec.handler
capture = self
src/cleveragents/tool/lifecycle.py
+5 -2
View File
@@ -673,8 +673,11 @@ class ToolRuntime:
"""
cap = tool.capability
# Read-only plan cannot use tools that write
if ctx.plan_read_only and not cap.read_only and cap.writes:
# Read-only plan cannot use tools that write.
# Any tool with writes=True is blocked regardless of its read_only flag
# (the ToolCapability model validator already prevents read_only=True
# combined with writes=True, but we enforce defensively).
if ctx.plan_read_only and cap.writes:
raise ToolAccessDeniedError(
f"Tool '{tool.name}' has writes=True but plan "
f"'{ctx.plan_id}' is read-only"