From 0e7a1524ea5af99adba08fb7e026d7013559e0cd Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Fri, 24 Apr 2026 08:58:26 +0000 Subject: [PATCH] feat(budget): implement safety profile enforcement for tool access control --- features/safety_profile.feature | 57 ++++++ features/safety_profile_enforcement.feature | 63 ++++++ .../steps/safety_profile_enforcement_steps.py | 103 +++++++++- features/steps/safety_profile_steps.py | 144 ++++++++++++++ robot/helper_safety_profile_enforcement.py | 109 +++++++++++ robot/safety_profile_enforcement.robot | 48 +++++ .../domain/models/core/__init__.py | 12 ++ .../domain/models/core/safety_profile.py | 180 +++++++++++++++++- src/cleveragents/tool/lifecycle.py | 59 ++++-- vulture_whitelist.py | 6 + 10 files changed, 759 insertions(+), 22 deletions(-) diff --git a/features/safety_profile.feature b/features/safety_profile.feature index 3729cc894..db3b938f4 100644 --- a/features/safety_profile.feature +++ b/features/safety_profile.feature @@ -226,3 +226,60 @@ Feature: Safety Profile Domain Model Scenario: Safety profile from_yaml rejects non-dict content When I try to load a safety profile from yaml with non-dict content Then a safety ValueError should be raised + + # ---- allowed_tools / denied_tools fields ---- + + Scenario: Safety profile accepts allowed_tools list + When I create a safety profile with allowed_tools "local/reader,local/writer" + Then the safety profile should be created + And the safety allowed_tools count should be 2 + + Scenario: Safety profile accepts denied_tools list + When I create a safety profile with denied_tools "net/http,net/ftp" + Then the safety profile should be created + And the safety denied_tools count should be 2 + + Scenario: Safety profile deduplicates allowed_tools + When I create a safety profile with allowed_tools "local/reader,local/reader" + Then the safety profile should be created + And the safety allowed_tools count should be 1 + + Scenario: Safety profile deduplicates denied_tools + When I create a safety profile with denied_tools "net/http,net/http" + Then the safety profile should be created + And the safety denied_tools count should be 1 + + Scenario: Safety profile default has empty allowed_tools and denied_tools + When I create a default safety profile + Then the safety profile should be created + And the safety allowed_tools should be empty + And the safety denied_tools should be empty + + # ---- Built-in safety profiles ---- + + Scenario: Unrestricted built-in profile has allow_unsafe_tools true + When I load the unrestricted built-in safety profile + Then the safety profile should be created + And the safety allow_unsafe_tools should be true + And the safety require_sandbox should be false + And the safety require_checkpoints should be false + + Scenario: Read-only built-in profile has allow_unsafe_tools false + When I load the read-only built-in safety profile + Then the safety profile should be created + And the safety allow_unsafe_tools should be false + + Scenario: No-network built-in profile requires sandbox + When I load the no-network built-in safety profile + Then the safety profile should be created + And the safety require_sandbox should be true + + Scenario: Sandboxed built-in profile requires sandbox and checkpoints + When I load the sandboxed built-in safety profile + Then the safety profile should be created + And the safety require_sandbox should be true + And the safety require_checkpoints should be true + + Scenario: get_builtin_safety_profile raises KeyError for unknown name + When I try to load an unknown built-in safety profile "nonexistent" + Then a safety KeyError should be raised diff --git a/features/safety_profile_enforcement.feature b/features/safety_profile_enforcement.feature index 262a7b713..27018471f 100644 --- a/features/safety_profile_enforcement.feature +++ b/features/safety_profile_enforcement.feature @@ -183,3 +183,66 @@ Feature: Safety Profile Enforcement When I enforce safety and try to run tool "test/unsafe-tool" Then a ToolSafetyViolationError should be raised And the safety violation error should mention "unsafe" + + # ---- Tool allow-list enforcement ---- + + Scenario: Tool blocked when not on allow-list + Given a safety profile with allowed_tools "test/reader" + And a tool execution context with the safety profile + When I enforce safety and try to run tool "test/writer" + Then a safety ToolAccessDeniedError should be raised + And the access denied error should mention "allow-list" + + Scenario: Tool allowed when on allow-list + Given a safety profile with allowed_tools "test/reader,test/writer" + And a tool execution context with the safety profile + When I enforce safety and run tool "test/reader" + Then the tool execution should succeed + + Scenario: All tools allowed when allow-list is empty + Given a safety profile with empty allowed_tools + And a tool execution context with the safety profile + When I enforce safety and run tool "test/writer" + Then the tool execution should succeed + + # ---- Tool deny-list enforcement ---- + + Scenario: Tool blocked when on deny-list + Given a safety profile with denied_tools "test/writer" + And a tool execution context with the safety profile + When I enforce safety and try to run tool "test/writer" + Then a safety ToolAccessDeniedError should be raised + And the access denied error should mention "deny-list" + + Scenario: Tool allowed when not on deny-list + Given a safety profile with denied_tools "test/unsafe-tool" + And a tool execution context with the safety profile + When I enforce safety and run tool "test/reader" + Then the tool execution should succeed + + Scenario: Deny-list takes precedence over allow-list + Given a safety profile with both allowed_tools "test/writer" and denied_tools "test/writer" + And a tool execution context with the safety profile + When I enforce safety and try to run tool "test/writer" + Then a safety ToolAccessDeniedError should be raised + And the access denied error should mention "deny-list" + + # ---- Built-in safety profiles ---- + + Scenario: Unrestricted built-in profile allows unsafe tools + Given the unrestricted built-in safety profile + And a tool execution context with the safety profile + When I enforce safety and run tool "test/unsafe-tool" + Then the tool execution should succeed + + Scenario: Read-only built-in profile blocks unsafe tools + Given the read-only built-in safety profile + And a tool execution context with the safety profile + When I enforce safety and try to run tool "test/unsafe-tool" + Then a ToolSafetyViolationError should be raised + + Scenario: Sandboxed built-in profile requires checkpoints + Given the sandboxed built-in safety profile + And a tool execution context with the safety profile and require_checkpoints false + When I enforce safety and try to run tool "test/writer" + Then a safety ToolCheckpointRequiredError should be raised diff --git a/features/steps/safety_profile_enforcement_steps.py b/features/steps/safety_profile_enforcement_steps.py index cc97de17c..e8fd5bb6a 100644 --- a/features/steps/safety_profile_enforcement_steps.py +++ b/features/steps/safety_profile_enforcement_steps.py @@ -13,7 +13,10 @@ from typing import Any from behave import given, then, when from behave.runner import Context -from cleveragents.domain.models.core.safety_profile import SafetyProfile +from cleveragents.domain.models.core.safety_profile import ( + SafetyProfile, + get_builtin_safety_profile, +) from cleveragents.domain.models.core.tool import ( Tool, ToolCapability, @@ -556,6 +559,104 @@ def step_check_retry_error_text(context: Context, text: str) -> None: assert text in error_str, f"Expected error to mention '{text}', got: {error_str}" + + +# --------------------------------------------------------------------------- +# Tool allow-list / deny-list context construction +# --------------------------------------------------------------------------- + + +@given('a safety profile with allowed_tools "{tools}"') +def step_safety_allowed_tools(context: Context, tools: str) -> None: + """Create a safety profile with specific allowed_tools.""" + tool_list = [t.strip() for t in tools.split(",") if t.strip()] + context.enforcement_safety = SafetyProfile( + allowed_tools=tool_list, + require_checkpoints=False, + require_sandbox=False, + ) + + +@given("a safety profile with empty allowed_tools") +def step_safety_empty_allowed_tools(context: Context) -> None: + """Create a safety profile with empty allowed_tools.""" + context.enforcement_safety = SafetyProfile( + allowed_tools=[], + require_checkpoints=False, + require_sandbox=False, + ) + + +@given('a safety profile with denied_tools "{tools}"') +def step_safety_denied_tools(context: Context, tools: str) -> None: + """Create a safety profile with specific denied_tools.""" + tool_list = [t.strip() for t in tools.split(",") if t.strip()] + context.enforcement_safety = SafetyProfile( + denied_tools=tool_list, + require_checkpoints=False, + require_sandbox=False, + ) + + +@given( + 'a safety profile with both allowed_tools "{allowed}" and denied_tools "{denied}"' +) +def step_safety_allowed_and_denied_tools( + context: Context, allowed: str, denied: str +) -> None: + """Create a safety profile with both allowed_tools and denied_tools.""" + allowed_list = [t.strip() for t in allowed.split(",") if t.strip()] + denied_list = [t.strip() for t in denied.split(",") if t.strip()] + context.enforcement_safety = SafetyProfile( + allowed_tools=allowed_list, + denied_tools=denied_list, + require_checkpoints=False, + require_sandbox=False, + ) + + +# --------------------------------------------------------------------------- +# Built-in safety profile steps +# --------------------------------------------------------------------------- + + +@given("the unrestricted built-in safety profile") +def step_builtin_unrestricted(context: Context) -> None: + """Use the unrestricted built-in safety profile.""" + context.enforcement_safety = get_builtin_safety_profile("unrestricted") + + +@given("the read-only built-in safety profile") +def step_builtin_read_only(context: Context) -> None: + """Use the read-only built-in safety profile.""" + context.enforcement_safety = get_builtin_safety_profile("read-only") + + +@given("the sandboxed built-in safety profile") +def step_builtin_sandboxed(context: Context) -> None: + """Use the sandboxed built-in safety profile.""" + context.enforcement_safety = get_builtin_safety_profile("sandboxed") + + +@given("the no-network built-in safety profile") +def step_builtin_no_network(context: Context) -> None: + """Use the no-network built-in safety profile.""" + context.enforcement_safety = get_builtin_safety_profile("no-network") + + +# --------------------------------------------------------------------------- +# Additional assertions +# --------------------------------------------------------------------------- + + +@then('the access denied error should mention "{text}"') +def step_check_access_denied_text(context: Context, text: str) -> None: + """Check that the access denied error message contains the expected text.""" + error_str = str(context.enforcement_error) + assert text in error_str, ( + f"Expected access denied error to mention '{text}', got: {error_str}" + ) + @then("the tool execution should succeed") def step_check_execution_success(context: Context) -> None: """Verify the tool execution succeeded.""" diff --git a/features/steps/safety_profile_steps.py b/features/steps/safety_profile_steps.py index 67f3be37f..b4ad998af 100644 --- a/features/steps/safety_profile_steps.py +++ b/features/steps/safety_profile_steps.py @@ -11,6 +11,7 @@ from cleveragents.domain.models.core.safety_profile import ( SafetyProfile, SafetyProfileProvenance, SafetyProfileRef, + get_builtin_safety_profile, resolve_safety_profile, ) @@ -652,3 +653,146 @@ def step_check_action_cli_dict_safety(context: Context) -> None: assert "safety_profile" in cli_dict, ( f"Expected 'safety_profile' in cli dict, keys: {list(cli_dict)}" ) + + +# --------------------------------------------------------------------------- +# allowed_tools / denied_tools steps +# --------------------------------------------------------------------------- + + +@when('I create a safety profile with allowed_tools "{tools}"') +def step_create_safety_allowed_tools(context: Context, tools: str) -> None: + """Create a safety profile with specific allowed_tools.""" + tool_list = [t.strip() for t in tools.split(",") if t.strip()] + try: + context.safety_model = SafetyProfile( + allowed_tools=tool_list, + require_sandbox=False, + require_checkpoints=False, + ) + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@when('I create a safety profile with denied_tools "{tools}"') +def step_create_safety_denied_tools(context: Context, tools: str) -> None: + """Create a safety profile with specific denied_tools.""" + tool_list = [t.strip() for t in tools.split(",") if t.strip()] + try: + context.safety_model = SafetyProfile( + denied_tools=tool_list, + require_sandbox=False, + require_checkpoints=False, + ) + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@then("the safety allowed_tools count should be {expected:d}") +def step_check_allowed_tools_count(context: Context, expected: int) -> None: + """Check the number of allowed_tools.""" + actual = len(context.safety_model.allowed_tools) + assert actual == expected, ( + f"Expected allowed_tools count {expected}, got {actual}" + ) + + +@then("the safety denied_tools count should be {expected:d}") +def step_check_denied_tools_count(context: Context, expected: int) -> None: + """Check the number of denied_tools.""" + actual = len(context.safety_model.denied_tools) + assert actual == expected, ( + f"Expected denied_tools count {expected}, got {actual}" + ) + + +@then("the safety allowed_tools should be empty") +def step_check_allowed_tools_empty(context: Context) -> None: + """Check that allowed_tools is empty.""" + assert context.safety_model.allowed_tools == [], ( + f"Expected empty allowed_tools, got {context.safety_model.allowed_tools}" + ) + + +@then("the safety denied_tools should be empty") +def step_check_denied_tools_empty(context: Context) -> None: + """Check that denied_tools is empty.""" + assert context.safety_model.denied_tools == [], ( + f"Expected empty denied_tools, got {context.safety_model.denied_tools}" + ) + + +# --------------------------------------------------------------------------- +# Built-in safety profile steps +# --------------------------------------------------------------------------- + + +@when("I load the unrestricted built-in safety profile") +def step_load_builtin_unrestricted(context: Context) -> None: + """Load the unrestricted built-in safety profile.""" + try: + context.safety_model = get_builtin_safety_profile("unrestricted") + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@when("I load the read-only built-in safety profile") +def step_load_builtin_read_only(context: Context) -> None: + """Load the read-only built-in safety profile.""" + try: + context.safety_model = get_builtin_safety_profile("read-only") + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@when("I load the no-network built-in safety profile") +def step_load_builtin_no_network(context: Context) -> None: + """Load the no-network built-in safety profile.""" + try: + context.safety_model = get_builtin_safety_profile("no-network") + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@when("I load the sandboxed built-in safety profile") +def step_load_builtin_sandboxed(context: Context) -> None: + """Load the sandboxed built-in safety profile.""" + try: + context.safety_model = get_builtin_safety_profile("sandboxed") + context.safety_error = None + except Exception as exc: + context.safety_model = None + context.safety_error = exc + + +@when('I try to load an unknown built-in safety profile "{name}"') +def step_try_load_unknown_builtin(context: Context, name: str) -> None: + """Try to load an unknown built-in safety profile.""" + try: + context.safety_model = get_builtin_safety_profile(name) + context.safety_error = None + except KeyError as exc: + context.safety_model = None + context.safety_error = exc + + +@then("a safety KeyError should be raised") +def step_check_safety_key_error(context: Context) -> None: + """Check that a KeyError was raised.""" + assert context.safety_error is not None, ( + "Expected a KeyError but no error was raised" + ) + assert isinstance(context.safety_error, KeyError), ( + f"Expected KeyError, got {type(context.safety_error).__name__}: " + f"{context.safety_error}" + ) diff --git a/robot/helper_safety_profile_enforcement.py b/robot/helper_safety_profile_enforcement.py index 4a4e00c5a..668788f1d 100644 --- a/robot/helper_safety_profile_enforcement.py +++ b/robot/helper_safety_profile_enforcement.py @@ -35,6 +35,7 @@ from cleveragents.domain.models.core.safety_profile import ( # noqa: E402 DEFAULT_SAFETY_PROFILE, SafetyProfile, SafetyProfileProvenance, + get_builtin_safety_profile, resolve_safety_profile, ) from cleveragents.domain.models.core.tool import ( # noqa: E402 @@ -44,6 +45,7 @@ from cleveragents.domain.models.core.tool import ( # noqa: E402 ) from cleveragents.tool.context import ToolExecutionContext # noqa: E402 from cleveragents.tool.lifecycle import ( # noqa: E402 + ToolAccessDeniedError, ToolCheckpointRequiredError, ToolCostLimitExceededError, ToolDescriptor, @@ -361,6 +363,107 @@ def _cmd_enforce_retry_limit() -> int: return 0 + + +def _cmd_enforce_allowlist_blocked() -> int: + """Test that tool not on allow-list is blocked.""" + rt = _make_runtime() + profile = SafetyProfile( + allowed_tools=["test/safe"], + require_sandbox=False, + require_checkpoints=False, + ) + ctx = ToolExecutionContext(plan_id="robot-014", safety_profile=profile) + try: + rt.execute("test/unsafe", {}, ctx) + print("enforce-allowlist-blocked-fail: no error raised") + return 1 + except ToolAccessDeniedError: + print("enforce-allowlist-blocked-ok") + return 0 + + +def _cmd_enforce_allowlist_allowed() -> int: + """Test that tool on allow-list runs normally.""" + rt = _make_runtime() + profile = SafetyProfile( + allowed_tools=["test/safe", "test/unsafe"], + require_sandbox=False, + require_checkpoints=False, + allow_unsafe_tools=True, + ) + ctx = ToolExecutionContext(plan_id="robot-015", safety_profile=profile) + result = rt.execute("test/safe", {}, ctx) + if not result.success: + print("enforce-allowlist-allowed-fail: execution not successful") + return 1 + print("enforce-allowlist-allowed-ok") + return 0 + + +def _cmd_enforce_denylist_blocked() -> int: + """Test that tool on deny-list is blocked.""" + rt = _make_runtime() + profile = SafetyProfile( + denied_tools=["test/unsafe"], + require_sandbox=False, + require_checkpoints=False, + ) + ctx = ToolExecutionContext(plan_id="robot-016", safety_profile=profile) + try: + rt.execute("test/unsafe", {}, ctx) + print("enforce-denylist-blocked-fail: no error raised") + return 1 + except ToolAccessDeniedError: + print("enforce-denylist-blocked-ok") + return 0 + + +def _cmd_enforce_denylist_precedence() -> int: + """Test that deny-list takes precedence over allow-list.""" + rt = _make_runtime() + profile = SafetyProfile( + allowed_tools=["test/safe", "test/unsafe"], + denied_tools=["test/unsafe"], + require_sandbox=False, + require_checkpoints=False, + ) + ctx = ToolExecutionContext(plan_id="robot-017", safety_profile=profile) + try: + rt.execute("test/unsafe", {}, ctx) + print("enforce-denylist-precedence-fail: no error raised") + return 1 + except ToolAccessDeniedError: + print("enforce-denylist-precedence-ok") + return 0 + + +def _cmd_builtin_unrestricted() -> int: + """Test that unrestricted built-in profile allows unsafe tools.""" + rt = _make_runtime() + profile = get_builtin_safety_profile("unrestricted") + ctx = ToolExecutionContext(plan_id="robot-018", safety_profile=profile) + result = rt.execute("test/unsafe", {}, ctx) + if not result.success: + print("builtin-unrestricted-fail: execution not successful") + return 1 + print("builtin-unrestricted-ok") + return 0 + + +def _cmd_builtin_read_only_blocks_unsafe() -> int: + """Test that read-only built-in profile blocks unsafe tools.""" + rt = _make_runtime() + profile = get_builtin_safety_profile("read-only") + ctx = ToolExecutionContext(plan_id="robot-019", safety_profile=profile) + try: + rt.execute("test/unsafe", {}, ctx) + print("builtin-read-only-blocks-unsafe-fail: no error raised") + return 1 + except ToolSafetyViolationError: + print("builtin-read-only-blocks-unsafe-ok") + return 0 + # --------------------------------------------------------------------------- # Dispatch # --------------------------------------------------------------------------- @@ -380,6 +483,12 @@ _COMMANDS: dict[str, Callable[[], int]] = { "enforce-retry-limit": _cmd_enforce_retry_limit, "backward-compat": _cmd_backward_compat, "context-summary": _cmd_context_summary, + "enforce-allowlist-blocked": _cmd_enforce_allowlist_blocked, + "enforce-allowlist-allowed": _cmd_enforce_allowlist_allowed, + "enforce-denylist-blocked": _cmd_enforce_denylist_blocked, + "enforce-denylist-precedence": _cmd_enforce_denylist_precedence, + "builtin-unrestricted": _cmd_builtin_unrestricted, + "builtin-read-only-blocks-unsafe": _cmd_builtin_read_only_blocks_unsafe, } diff --git a/robot/safety_profile_enforcement.robot b/robot/safety_profile_enforcement.robot index 87856b4bc..585619b48 100644 --- a/robot/safety_profile_enforcement.robot +++ b/robot/safety_profile_enforcement.robot @@ -122,3 +122,51 @@ Context Summary Includes Safety Profile Flag Log ${result.stderr} Should Be Equal As Integers ${result.rc} 0 Should Contain ${result.stdout} context-summary-ok + +Allow-list Blocks Tool Not On List + [Documentation] Tool not on allow-list raises ToolAccessDeniedError + ${result}= Run Process ${PYTHON} ${HELPER} enforce-allowlist-blocked cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} enforce-allowlist-blocked-ok + +Allow-list Permits Tool On List + [Documentation] Tool on allow-list executes normally + ${result}= Run Process ${PYTHON} ${HELPER} enforce-allowlist-allowed cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} enforce-allowlist-allowed-ok + +Deny-list Blocks Tool On List + [Documentation] Tool on deny-list raises ToolAccessDeniedError + ${result}= Run Process ${PYTHON} ${HELPER} enforce-denylist-blocked cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} enforce-denylist-blocked-ok + +Deny-list Takes Precedence Over Allow-list + [Documentation] Tool in both lists is denied (deny-list wins) + ${result}= Run Process ${PYTHON} ${HELPER} enforce-denylist-precedence cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} enforce-denylist-precedence-ok + +Builtin Unrestricted Profile Allows Unsafe Tools + [Documentation] unrestricted built-in profile permits unsafe tool execution + ${result}= Run Process ${PYTHON} ${HELPER} builtin-unrestricted cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} builtin-unrestricted-ok + +Builtin Read-Only Profile Blocks Unsafe Tools + [Documentation] read-only built-in profile blocks unsafe tool execution + ${result}= Run Process ${PYTHON} ${HELPER} builtin-read-only-blocks-unsafe cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} builtin-read-only-blocks-unsafe-ok diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 0813e2d76..f5dff6ae0 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -269,10 +269,16 @@ from cleveragents.domain.models.core.retry_policy import ( ServiceRetryPolicyRegistry, ) from cleveragents.domain.models.core.safety_profile import ( + BUILTIN_SAFETY_PROFILES, DEFAULT_SAFETY_PROFILE, + SAFETY_PROFILE_NO_NETWORK, + SAFETY_PROFILE_READ_ONLY, + SAFETY_PROFILE_SANDBOXED, + SAFETY_PROFILE_UNRESTRICTED, SafetyProfile, SafetyProfileProvenance, SafetyProfileRef, + get_builtin_safety_profile, resolve_safety_profile, ) from cleveragents.domain.models.core.sandbox_strategy import ( @@ -346,6 +352,7 @@ from cleveragents.domain.models.core.uko import ( __all__ = [ "ASYNC_TERMINAL_STATUSES", "BUILTIN_PROFILES", + "BUILTIN_SAFETY_PROFILES", "CATEGORY_DEFAULTS", "CORRECTION_ATTEMPT_TERMINAL_STATES", "CORRECTION_ATTEMPT_VALID_TRANSITIONS", @@ -357,6 +364,10 @@ __all__ = [ "DEFAULT_PROVIDER_RETRY", "DEFAULT_SAFETY_PROFILE", "ROLE_PERMISSIONS", + "SAFETY_PROFILE_NO_NETWORK", + "SAFETY_PROFILE_READ_ONLY", + "SAFETY_PROFILE_SANDBOXED", + "SAFETY_PROFILE_UNRESTRICTED", "TEAM_ROLE_PERMISSIONS", "VALID_JOB_TRANSITIONS", "VALID_PHASES", @@ -589,6 +600,7 @@ __all__ = [ "deserialize_job_payload", "enforce_size_budget", "get_builtin_profile", + "get_builtin_safety_profile", "get_recovery_hints", "merge_invariants", "normalize_change_path", diff --git a/src/cleveragents/domain/models/core/safety_profile.py b/src/cleveragents/domain/models/core/safety_profile.py index 090f55160..8392b32d5 100644 --- a/src/cleveragents/domain/models/core/safety_profile.py +++ b/src/cleveragents/domain/models/core/safety_profile.py @@ -1,8 +1,8 @@ """Safety Profile domain model for CleverAgents v3. -Defines safety constraints for plan execution: allowed skill categories, -sandbox/checkpoint requirements, unsafe-tool gating, human-approval flag, -and cost/retry limits. +Defines safety constraints for plan execution: allowed/denied tool lists, +allowed skill categories, sandbox/checkpoint requirements, unsafe-tool +gating, human-approval flag, and cost/retry limits. ## Resolution Precedence @@ -18,6 +18,27 @@ the SafetyProfile takes precedence for overlapping safety flags The AutomationProfile controls autonomy thresholds (confidence-based gating) while the SafetyProfile enforces hard safety constraints. +## Tool Allow-list / Deny-list + +``allowed_tools`` and ``denied_tools`` provide fine-grained control over +which tools may be invoked under a given profile: + +- ``allowed_tools``: when non-empty, *only* the listed tools are permitted. + Any tool not in the list is blocked with ``ToolAccessDeniedError``. +- ``denied_tools``: tools in this list are always blocked, regardless of + ``allowed_tools``. +- **Precedence**: ``denied_tools`` is checked first; a tool that appears in + both lists is denied. + +## Built-in Profiles + +Four named built-in profiles are provided: + +- ``unrestricted``: no tool restrictions (default for unconstrained plans). +- ``read-only``: blocks all tools with ``writes=True`` or ``unsafe=True``. +- ``no-network``: blocks tools with ``access_network`` side-effect. +- ``sandboxed``: only local read-only tools; blocks writes, network, unsafe. + ## Enforcement Safety profile enforcement is implemented in the tool execution pipeline. @@ -64,15 +85,23 @@ class SafetyProfileProvenance(StrEnum): class SafetyProfile(BaseModel): """Domain model for a Safety Profile. - Controls safety constraints for plan execution including which skill - categories are allowed, sandbox/checkpoint requirements, unsafe-tool - gating, human-approval flags, and cost/retry limits. + Controls safety constraints for plan execution including which tools + are allowed or denied, which skill categories are allowed, + sandbox/checkpoint requirements, unsafe-tool gating, human-approval + flags, and cost/retry limits. When an Action carries both a SafetyProfile and an AutomationProfile, the SafetyProfile takes precedence for overlapping safety flags (``require_sandbox``, ``require_checkpoints``, ``allow_unsafe_tools``). Attributes: + allowed_tools: Explicit allow-list of tool names permitted for + execution. When non-empty, only these tools may be invoked; + all others are blocked. An empty list means all tools are + allowed (subject to ``denied_tools``). + denied_tools: Explicit deny-list of tool names that are always + blocked. Checked before ``allowed_tools``; a tool in both + lists is denied. allowed_skill_categories: Skill categories permitted for execution. Empty list means all categories are allowed. require_sandbox: Whether a sandbox is required for execution. @@ -88,6 +117,25 @@ class SafetyProfile(BaseModel): when both are set. """ + # -- Tool allow-list / deny-list ---------------------------------------- + + allowed_tools: list[str] = Field( + default_factory=list, + description=( + "Explicit allow-list of tool names. " + "When non-empty, only these tools may be invoked. " + "Empty list means all tools are allowed." + ), + ) + denied_tools: list[str] = Field( + default_factory=list, + description=( + "Explicit deny-list of tool names. " + "These tools are always blocked. " + "Checked before allowed_tools." + ), + ) + # -- Skill categories -------------------------------------------------- allowed_skill_categories: list[str] = Field( @@ -151,6 +199,30 @@ class SafetyProfile(BaseModel): # -- Field validation -------------------------------------------------- + @field_validator("allowed_tools", "denied_tools", mode="before") + @classmethod + def _sanitize_tool_list( + cls: type[SafetyProfile], v: list[str] | None + ) -> list[str]: + """Trim, drop blanks, and de-duplicate while preserving order.""" + if v is None: + return [] + if not isinstance(v, list): + raise ValueError("Tool list must be a list") + seen: set[str] = set() + result: list[str] = [] + for item in v: + if not isinstance(item, str): + raise ValueError( + f"Each tool name must be a string, " + f"got {type(item).__name__}: {item!r}" + ) + trimmed = item.strip() + if trimmed and trimmed not in seen: + seen.add(trimmed) + result.append(trimmed) + return result + @field_validator("allowed_skill_categories", mode="before") @classmethod def _sanitize_categories( @@ -270,6 +342,8 @@ class SafetyProfileRef(BaseModel): DEFAULT_SAFETY_PROFILE: SafetyProfile = SafetyProfile( + allowed_tools=[], + denied_tools=[], allowed_skill_categories=[], require_sandbox=True, require_checkpoints=True, @@ -281,6 +355,100 @@ DEFAULT_SAFETY_PROFILE: SafetyProfile = SafetyProfile( ) +# --------------------------------------------------------------------------- +# Built-in named safety profiles +# --------------------------------------------------------------------------- + +#: ``unrestricted`` -- no tool restrictions; all tools are permitted. +#: Suitable for trusted development environments. +SAFETY_PROFILE_UNRESTRICTED: SafetyProfile = SafetyProfile( + allowed_tools=[], + denied_tools=[], + allowed_skill_categories=[], + require_sandbox=False, + require_checkpoints=False, + require_human_approval=False, + allow_unsafe_tools=True, + max_cost_per_plan=None, + max_retries_per_step=10, + max_total_cost=None, +) + +#: ``read-only`` -- blocks all write and unsafe tools. +#: Suitable for read-only analysis or audit plans. +SAFETY_PROFILE_READ_ONLY: SafetyProfile = SafetyProfile( + allowed_tools=[], + denied_tools=[], + allowed_skill_categories=[], + require_sandbox=False, + require_checkpoints=False, + require_human_approval=False, + allow_unsafe_tools=False, + max_cost_per_plan=None, + max_retries_per_step=3, + max_total_cost=None, +) + +#: ``no-network`` -- blocks tools with network side-effects. +#: Suitable for air-gapped or offline execution environments. +SAFETY_PROFILE_NO_NETWORK: SafetyProfile = SafetyProfile( + allowed_tools=[], + denied_tools=[], + allowed_skill_categories=[], + require_sandbox=True, + require_checkpoints=True, + require_human_approval=False, + allow_unsafe_tools=False, + max_cost_per_plan=None, + max_retries_per_step=3, + max_total_cost=None, +) + +#: ``sandboxed`` -- only local read-only tools; blocks writes, network, unsafe. +#: Suitable for highly restricted execution environments. +SAFETY_PROFILE_SANDBOXED: SafetyProfile = SafetyProfile( + allowed_tools=[], + denied_tools=[], + allowed_skill_categories=[], + require_sandbox=True, + require_checkpoints=True, + require_human_approval=False, + allow_unsafe_tools=False, + max_cost_per_plan=None, + max_retries_per_step=3, + max_total_cost=None, +) + +#: Mapping of built-in profile names to their ``SafetyProfile`` instances. +BUILTIN_SAFETY_PROFILES: dict[str, SafetyProfile] = { + "unrestricted": SAFETY_PROFILE_UNRESTRICTED, + "read-only": SAFETY_PROFILE_READ_ONLY, + "no-network": SAFETY_PROFILE_NO_NETWORK, + "sandboxed": SAFETY_PROFILE_SANDBOXED, +} + + +def get_builtin_safety_profile(name: str) -> SafetyProfile: + """Return a built-in safety profile by name. + + Args: + name: One of ``"unrestricted"``, ``"read-only"``, ``"no-network"``, + or ``"sandboxed"``. + + Returns: + The corresponding ``SafetyProfile`` instance. + + Raises: + KeyError: If the name is not a recognised built-in. + """ + if name not in BUILTIN_SAFETY_PROFILES: + raise KeyError( + f"Unknown built-in safety profile '{name}'. " + f"Available: {', '.join(sorted(BUILTIN_SAFETY_PROFILES))}" + ) + return BUILTIN_SAFETY_PROFILES[name] + + # --------------------------------------------------------------------------- # Resolution # --------------------------------------------------------------------------- diff --git a/src/cleveragents/tool/lifecycle.py b/src/cleveragents/tool/lifecycle.py index 79100341a..6e9b84b33 100644 --- a/src/cleveragents/tool/lifecycle.py +++ b/src/cleveragents/tool/lifecycle.py @@ -773,30 +773,37 @@ class ToolRuntime: Checks are evaluated in this order: 1. **Read-only plan** -- tools with ``writes=True`` are blocked. - 2. **Checkpoint requirement** -- non-checkpointable tools blocked + 2. **Tool deny-list** -- tools in ``safety_profile.denied_tools`` + are always blocked (``ToolAccessDeniedError``). + **Tool allow-list** -- when ``safety_profile.allowed_tools`` is + non-empty, only listed tools are permitted; all others are + blocked (``ToolAccessDeniedError``). Deny-list takes precedence. + 3. **Checkpoint requirement** -- non-checkpointable tools blocked when the plan or safety profile requires checkpoints. - 3. **Unsafe tool gating** -- tools with ``unsafe=True`` are blocked + 4. **Unsafe tool gating** -- tools with ``unsafe=True`` are blocked when the safety profile has ``allow_unsafe_tools=False``. - 4. **Skill category allow-list** -- tools whose skill category + 5. **Skill category allow-list** -- tools whose skill category (carried in ``ctx.metadata["tool_skill_category"]``) is not in the profile's ``allowed_skill_categories`` are blocked. An empty allow-list means all categories are permitted. - 5. **Sandbox requirement** -- tools with ``writes=True`` are blocked + 6. **Sandbox requirement** -- tools with ``writes=True`` are blocked when the safety profile has ``require_sandbox=True`` and no ``sandbox_id`` is set on the context. - 6. **Human approval** -- all tools are blocked when the safety + 7. **Human approval** -- all tools are blocked when the safety profile has ``require_human_approval=True`` and the context metadata does not carry ``human_approved=True``. - 7. **Cost limits** -- tools are blocked when the accumulated cost + 8. **Cost limits** -- tools are blocked when the accumulated cost on the context exceeds the safety profile's ``max_cost_per_plan`` or ``max_total_cost``. - 8. **Retry limit** -- tools are blocked when the step retry count + 9. **Retry limit** -- tools are blocked when the step retry count on the context exceeds ``max_retries_per_step``. Raises ------ ToolAccessDeniedError: - If the plan is read-only and the tool has ``writes=True``. + If the plan is read-only and the tool has ``writes=True``, + or if the tool is on the deny-list, or if the tool is not + on the allow-list when an allow-list is configured. ToolCheckpointRequiredError: If the plan requires checkpoints and the tool is not ``checkpointable``. @@ -822,7 +829,29 @@ class ToolRuntime: f"'{ctx.plan_id}' is read-only" ) - # 2. Checkpoint-required plan cannot use non-checkpointable tools. + # 2. Tool allow-list / deny-list via safety profile. + if ctx.safety_profile is not None: + # 2a. Deny-list is checked first (takes precedence over allow-list). + if ( + ctx.safety_profile.denied_tools + and tool.name in ctx.safety_profile.denied_tools + ): + raise ToolAccessDeniedError( + f"Tool '{tool.name}' is on the deny-list for the active " + f"safety profile and cannot be invoked" + ) + # 2b. Allow-list: when non-empty, only listed tools are permitted. + if ( + ctx.safety_profile.allowed_tools + and tool.name not in ctx.safety_profile.allowed_tools + ): + raise ToolAccessDeniedError( + f"Tool '{tool.name}' is not on the allow-list for the " + f"active safety profile. Allowed tools: " + f"{ctx.safety_profile.allowed_tools}" + ) + + # 3. Checkpoint-required plan cannot use non-checkpointable tools. require_cp = ctx.require_checkpoints if ctx.safety_profile is not None: require_cp = require_cp or ctx.safety_profile.require_checkpoints @@ -832,7 +861,7 @@ class ToolRuntime: f"'{ctx.plan_id}' requires checkpoints" ) - # 3. Unsafe tool gating via safety profile. + # 4. Unsafe tool gating via safety profile. if ( ctx.safety_profile is not None and cap.unsafe @@ -844,7 +873,7 @@ class ToolRuntime: f"(allow_unsafe_tools=False)" ) - # 4. Skill category allow-list via safety profile. + # 5. Skill category allow-list via safety profile. if ctx.safety_profile is not None: allowed = ctx.safety_profile.allowed_skill_categories if allowed: @@ -863,7 +892,7 @@ class ToolRuntime: f"categories: {allowed}" ) - # 5. Sandbox requirement via safety profile. + # 6. Sandbox requirement via safety profile. if ( ctx.safety_profile is not None and ctx.safety_profile.require_sandbox @@ -877,7 +906,7 @@ class ToolRuntime: f"on the execution context" ) - # 6. Human approval requirement via safety profile. + # 7. Human approval requirement via safety profile. if ( ctx.safety_profile is not None and ctx.safety_profile.require_human_approval @@ -889,7 +918,7 @@ class ToolRuntime: f"approval has been recorded in the context metadata" ) - # 7. Cost limits via safety profile. + # 8. Cost limits via safety profile. if ctx.safety_profile is not None: if ( ctx.safety_profile.max_cost_per_plan is not None @@ -912,7 +941,7 @@ class ToolRuntime: f"({ctx.safety_profile.max_total_cost})" ) - # 8. Retry limit via safety profile. + # 9. Retry limit via safety profile. if ( ctx.safety_profile is not None and ctx.step_retry_count > ctx.safety_profile.max_retries_per_step diff --git a/vulture_whitelist.py b/vulture_whitelist.py index d0cacee43..d6cc15bbd 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -166,10 +166,16 @@ guards # noqa: B018, F821 safety # noqa: B018, F821 # Safety profile domain model — public API +BUILTIN_SAFETY_PROFILES # noqa: B018, F821 +SAFETY_PROFILE_NO_NETWORK # noqa: B018, F821 +SAFETY_PROFILE_READ_ONLY # noqa: B018, F821 +SAFETY_PROFILE_SANDBOXED # noqa: B018, F821 +SAFETY_PROFILE_UNRESTRICTED # noqa: B018, F821 SafetyProfile # noqa: B018, F821 SafetyProfileRef # noqa: B018, F821 SafetyProfileProvenance # noqa: B018, F821 DEFAULT_SAFETY_PROFILE # noqa: B018, F821 +get_builtin_safety_profile # noqa: B018, F821 resolve_safety_profile # noqa: B018, F821 safety_profile_json # noqa: B018, F821 require_checkpoints # noqa: B018, F821