fix: harden tool safety guard enforcement

Author: impself

docs/tool_safety_guard.md

@@ -51,10 +51,10 @@ permissions, and runtime resource limits.
 
 | Layer | Owns |
 |---|---|
-| **ToolScriptSafetyFilter** | Pre-execution static policy decision and redaction |
-| **SafetyCheckedExecutor** | Applies the same decision before delegated execution |
+| **SafetyWrappedCallable / SafetyCheckedExecutor** | Current enforcement path: scan, await audit, then delegate only when allowed |
+| **ToolScriptSafetyFilter** | Normalizes and records decisions for wrappers and a future SDK terminal hook |
 | **Wrapper / Sandbox / Runtime** | Runtime isolation: CPU, memory, PID, FS, network hard limits |
-| **Audit / Telemetry** | Evidence that a decision occurred; **not** enforcement |
+| **Audit / Telemetry** | Decision evidence; required audit persistence is part of the wrapper's fail-closed gate |
 
 ## Quick start
 
@@ -119,6 +119,13 @@ safe_run = SafetyWrappedCallable(
 safe_run("ls -la")  # raises BlockedExecutionError if policy denies
 ```
 
+Call ``await safe_run.call_async(...)`` instead when the delegate or caller
+already runs inside an event loop; this preserves the required-audit
+before-delegate guarantee.
+For a tool-style callable, set ``argv_kw``, ``cwd_kw``, ``env_kw``,
+``metadata_kw``, and ``output_bytes_kw`` to the corresponding argument
+names so the normalized request contains every available execution field.
+
 ### Wrapping a code executor
 
 ```python
@@ -148,6 +155,7 @@ change between releases so policy overrides remain stable.
 | `FILE004_DOTENV_READ` | file | deny | Reads of `.env` files |
 | `NET001_DOMAIN_NOT_ALLOWED` | network | deny | Requests/curl/wget to non-allowlisted hosts |
 | `NET002_DYNAMIC_TARGET` | network | review | Computed network destination |
+| `NET003_IP_LITERAL` | network | deny | IP literal when `deny_ip_literals` is enabled |
 | `PROC001_PROCESS_EXEC` | process | review | Subprocess or command not on allow list |
 | `PROC002_SHELL_INJECTION` | process | deny | `shell=True` with shell grammar |
 | `PROC003_SHELL_OPERATOR` | process | review | `;`, `&&`, `\|`, `&`, command substitution |
@@ -156,13 +164,13 @@ change between releases so policy overrides remain stable.
 | `RES001_UNBOUNDED_LOOP` | resource | deny | `while True` without break |
 | `RES002_FORK_BOMB` | resource | deny | Classic `:(){ :\|:& };:` pattern |
 | `RES003_LONG_SLEEP` | resource | deny | Sleeps exceeding policy limit |
-| `RES004_CONCURRENCY` | resource | deny | Fan-out exceeding `max_parallel_tasks` |
+| `RES004_CONCURRENCY` | resource | deny | Fan-out exceeding `max_parallel_tasks` or `max_processes` |
 | `RES005_LARGE_WRITE` | resource | deny | Writes exceeding `max_file_write_bytes` |
 | `SECRET001_LOG_SINK` | secret | deny | Tainted value into print/log |
 | `SECRET002_FILE_SINK` | secret | deny | Tainted value into file write |
 | `SECRET003_NETWORK_SINK` | secret | deny | Tainted value into network payload |
 | `PARSE001_UNCERTAIN` | analysis | review | Syntax error or unknown construct |
-| `OBF001_DYNAMIC_EXEC` | analysis | review | `eval`, `exec`, `importlib`, reflective calls |
+| `OBF001_DYNAMIC_EXEC` | analysis | review | `eval`, `exec`, indirect Bash execution, interpreter payloads |
 | `SAFE000` | safe | allow | No findings |
 | `GUARD001_INTERNAL_ERROR` | analysis | deny | Internal guard failure (fail closed) |
 
@@ -272,19 +280,20 @@ never emitted as span attributes or metric labels.
 | 0 | Final decision was ``allow`` |
 | 2 | Final decision was ``deny`` |
 | 3 | Final decision was ``needs_human_review`` |
-| 4 | Invalid input / policy error |
+| 4 | Invalid input, policy, or required-audit error |
 
 ## Integration with the SDK
 
-The standalone package is duck-typed to ``BaseFilter`` so it can be
-composed with the framework's filter runner once the terminal-phase
-ordering seam is exposed. Until then, wrap the executor or callable
-explicitly via ``tool.wrapper``.
+The current SDK does not expose a terminal filter phase after
+``ToolCallbackFilter``. A configured ``filters=`` instance can therefore
+scan arguments that a later callback changes; it is not a secure enforcement
+point. Use ``SafetyWrappedCallable`` or ``SafetyCheckedExecutor`` today:
+both scan, await the audit event, and only then invoke their delegate.
 
-The :class:`ToolScriptSafetyFilter` carries a ``terminal_before_handler``
-attribute (always ``True``). When the framework adopts the terminal
-phase marker, the filter will automatically run after
-``ToolCallbackFilter`` and prevent TOCTOU mutations.
+``ToolScriptSafetyFilter`` provides matching ``_before``/``_after`` hooks
+and a ``terminal_before_handler`` marker for a future SDK terminal phase.
+That marker is metadata only until the framework implements ordering after
+all argument-mutating callbacks. The wrapper remains mandatory until then.
 
 ## Custom rules
 
@@ -324,6 +333,10 @@ Rules must be pure: no file I/O, network access, or process creation.
 * **Runtime downloads.** A script that downloads and executes a payload
   in two stages defeats static analysis. Network egress policy and
   sandboxing are mandatory.
+* **Runtime limits.** The wrapper validates the declared timeout and caps
+  returned output, but it cannot impose CPU, memory, PID, file-size, or
+  network limits on an arbitrary executor. Configure those in the sandbox
+  or CodeExecutor runtime as well.
 * **Shell grammar holes.** The bash lexer-lite is conservative: any
   unbalanced quote or unsupported substitution becomes
   ``PARSE001_UNCERTAIN``.

scripts/tool_safety_check.py

@@ -37,6 +37,7 @@
     sys.path.insert(0, str(_REPO_ROOT))
 
 from tool.safety._audit import InMemoryAuditSink, JsonlAuditSink  # noqa: E402
+from tool.safety._exceptions import SafetyAuditError  # noqa: E402
 from tool.safety._guard import ToolSafetyGuard  # noqa: E402
 from tool.safety._models import (  # noqa: E402
     SafetyDecision,
@@ -59,11 +60,15 @@ def main(argv: Sequence[str] | None = None) -> int:
         return 4
     guard = ToolSafetyGuard(policy)
     audit_sink = _resolve_audit_sink(args, policy)
-    if args.manifest:
-        return _run_manifest(guard, audit_sink, args)
-    if args.request_json:
-        return _run_request_json(guard, audit_sink, args)
-    return _run_single(guard, audit_sink, args)
+    try:
+        if args.manifest:
+            return _run_manifest(guard, audit_sink, args)
+        if args.request_json:
+            return _run_request_json(guard, audit_sink, args)
+        return _run_single(guard, audit_sink, args)
+    except SafetyAuditError as exc:
+        print(f"audit error: {exc}", file=sys.stderr)
+        return 4
 
 
 def _build_parser() -> argparse.ArgumentParser:
@@ -165,7 +170,13 @@ def _run_manifest(guard: ToolSafetyGuard,
             "matches_expected": _decision_matches(report, expected),
         })
         blocked = report.decision != SafetyDecision.ALLOW
-        asyncio.run(_emit_audit(audit_sink, report, request, blocked=blocked))
+        asyncio.run(_emit_audit(
+            audit_sink,
+            report,
+            request,
+            blocked=blocked,
+            required=guard.policy.audit.required,
+        ))
         exit_code = max(exit_code, _exit_for_decision(report.decision))
     if args.manifest_output:
         with open(args.manifest_output, "w", encoding="utf-8") as handle:
@@ -182,8 +193,13 @@ def _emit(guard: ToolSafetyGuard,
           args: argparse.Namespace,
           request: SafetyScanRequest) -> int:
     report = guard.scan(request)
-    asyncio.run(_emit_audit(audit_sink, report, request,
-                            blocked=report.decision != SafetyDecision.ALLOW))
+    asyncio.run(_emit_audit(
+        audit_sink,
+        report,
+        request,
+        blocked=report.decision != SafetyDecision.ALLOW,
+        required=guard.policy.audit.required,
+    ))
     payload = report.model_dump_json(indent=2)
     if args.output:
         with open(args.output, "w", encoding="utf-8") as handle:
@@ -193,7 +209,8 @@ def _emit(guard: ToolSafetyGuard,
 
 
 async def _emit_audit(audit_sink: Any, report: SafetyReport,
-                      request: SafetyScanRequest, *, blocked: bool) -> None:
+                      request: SafetyScanRequest, *, blocked: bool,
+                      required: bool) -> None:
     import datetime as _dt
     event = build_audit_event(
         report=report,
@@ -205,8 +222,11 @@ async def _emit_audit(audit_sink: Any, report: SafetyReport,
     try:
         await audit_sink.emit(event)
     except Exception as exc:
-        # Audit failures should not crash the CLI; surface them on stderr.
         print(f"audit emit warning: {exc}", file=sys.stderr)
+        if required:
+            if isinstance(exc, SafetyAuditError):
+                raise
+            raise SafetyAuditError("unexpected audit emit failure") from exc
 
 
 def _build_request(args: argparse.Namespace) -> SafetyScanRequest:

tests/tool_safety/test_bash_scanner.py

@@ -60,6 +60,12 @@ def test_pip_install_denies(guard):
     assert report.decision == SafetyDecision.DENY
 
 
+def test_python_module_pip_install_denies(guard):
+    report = scan(guard, "python -m pip install numpy\n")
+    assert "DEP001_ENV_MUTATION" in report.rule_ids
+    assert report.decision == SafetyDecision.DENY
+
+
 def test_apt_install_denies(guard):
     report = scan(guard, "apt-get install -y curl\n")
     assert "DEP001_ENV_MUTATION" in report.rule_ids
@@ -101,6 +107,29 @@ def test_dynamic_eval_review(guard):
     assert "OBF001_DYNAMIC_EXEC" in report.rule_ids
 
 
+@pytest.mark.parametrize("script", [
+    "source ./payload.sh\n",
+    "xargs -a commands.txt sh\n",
+    r"find . -exec sh {} \;\n",
+])
+def test_indirect_execution_requires_review(guard, script):
+    report = scan(guard, script)
+    assert "OBF001_DYNAMIC_EXEC" in report.rule_ids
+    assert report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW
+
+
+@pytest.mark.parametrize("script", [
+    "python payload.py\n",
+    "node payload.js\n",
+    "bash -x payload.sh\n",
+    "python -m untrusted_module\n",
+])
+def test_interpreter_payload_requires_review(guard, script):
+    report = scan(guard, script)
+    assert "OBF001_DYNAMIC_EXEC" in report.rule_ids
+    assert report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW
+
+
 def test_shebang_and_comment_ignored(guard):
     report = scan(guard, "#!/usr/bin/env bash\n# comment\necho hi\n")
     assert report.decision == SafetyDecision.ALLOW

tests/tool_safety/test_cli.py

@@ -59,6 +59,16 @@ def test_invalid_policy_exit_4(tmp_path):
     assert proc.returncode == 4
 
 
+def test_required_audit_failure_exits_4(tmp_path):
+    rc, out, err = _run_cli(
+        "--language", "python",
+        "--script", "print('hello')",
+        "--audit-file", str(tmp_path),
+    )
+    assert rc == 4
+    assert "audit error:" in err
+
+
 def test_manifest_writes_output(tmp_path):
     manifest = REPO_ROOT / "examples" / "tool_safety" / "samples" / "manifest.yaml"
     output = tmp_path / "out.json"

tests/tool_safety/test_cross_field_scanner.py

@@ -61,16 +61,16 @@ def test_denied_executable_in_argv_blocks(guard):
     assert report.decision == SafetyDecision.DENY
 
 
-def test_sensitive_env_triggers_review(guard):
+def test_unused_sensitive_env_does_not_trigger_review(guard):
     request = SafetyScanRequest(
         tool_name="t",
         language=ScriptLanguage.PYTHON,
         script="print('hi')",
         env={"API_TOKEN": "abc"},
     )
     report = guard.scan(request)
-    assert "SECRET001_LOG_SINK" in report.rule_ids
-    assert report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW
+    assert "SECRET001_LOG_SINK" not in report.rule_ids
+    assert report.decision == SafetyDecision.ALLOW
 
 
 def test_output_budget_enforced(guard):

tests/tool_safety/test_filter.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
+from types import SimpleNamespace
+
 import pytest
 
 from tool.safety._audit import InMemoryAuditSink
@@ -56,6 +59,30 @@ def test_audit_event_one_per_call(flt):
     assert len(flt.audit_sink.events) == 3  # type: ignore[attr-defined]
 
 
+def test_check_async_persists_audit_before_returning(flt):
+    async def run():
+        decision, _ = await flt.check_async(
+            "workspace_exec", {"command": "echo hi"})
+        assert decision == SafetyDecision.ALLOW
+        assert len(flt.audit_sink.events) == 1  # type: ignore[attr-defined]
+
+    asyncio.run(run())
+
+
+def test_filter_audits_request_build_failure(flt):
+    rsp = {}
+
+    async def run():
+        await flt._before(
+            SimpleNamespace(tool_name="workspace_exec"), {}, rsp)
+
+    asyncio.run(run())
+
+    assert rsp["is_continue"] is False
+    assert len(flt.audit_sink.events) == 1  # type: ignore[attr-defined]
+    assert flt.audit_sink.events[0].decision == SafetyDecision.DENY  # type: ignore[attr-defined]
+
+
 def test_filter_redacts_env_in_trace(flt):
     flt.check("workspace_exec", {
         "command": "echo hi",

tests/tool_safety/test_python_scanner.py

@@ -63,6 +63,26 @@ def test_allowlist_network_allows(guard):
     assert report.decision == SafetyDecision.ALLOW
 
 
+def test_ip_literal_policy_can_block_an_allowlisted_ip(strict_policy_dict):
+    policy_dict = strict_policy_dict.copy()
+    policy_dict["network"] = {
+        "allow_domains": ["127.0.0.1"],
+        "deny_ip_literals": True,
+    }
+    blocked_guard = ToolSafetyGuard(load_safety_policy_dict(policy_dict))
+    script = "import requests\nrequests.get('http://127.0.0.1:8080')\n"
+    blocked = scan(blocked_guard, script)
+    assert "NET003_IP_LITERAL" in blocked.rule_ids
+    assert blocked.decision == SafetyDecision.DENY
+
+    policy_dict["network"] = {
+        "allow_domains": ["127.0.0.1"],
+        "deny_ip_literals": False,
+    }
+    allowed_guard = ToolSafetyGuard(load_safety_policy_dict(policy_dict))
+    assert scan(allowed_guard, script).decision == SafetyDecision.ALLOW
+
+
 def test_fstring_allowlist_network_allows(guard):
     script = "import requests\nrequests.get(f'https://api.github.com/users/{name}')\n"
     report = scan(guard, script)
@@ -108,13 +128,36 @@ def test_long_sleep_denies(guard):
     assert report.decision == SafetyDecision.DENY
 
 
+def test_process_limit_uses_max_processes(strict_policy_dict):
+    policy_dict = strict_policy_dict.copy()
+    policy_dict["limits"] = dict(policy_dict["limits"])
+    policy_dict["limits"]["max_parallel_tasks"] = 10
+    policy_dict["limits"]["max_processes"] = 2
+    g = ToolSafetyGuard(load_safety_policy_dict(policy_dict))
+    report = scan(
+        g,
+        "import multiprocessing\nmultiprocessing.Pool(processes=3)\n",
+    )
+    assert "RES004_CONCURRENCY" in report.rule_ids
+    assert report.decision == SafetyDecision.DENY
+
+
 def test_secret_to_print_denies(guard):
     script = "import os\nv=os.environ['API_TOKEN']\nprint(v)\n"
     report = scan(guard, script)
     assert "SECRET001_LOG_SINK" in report.rule_ids
     assert report.decision == SafetyDecision.DENY
 
 
+def test_literal_api_key_to_print_denies_and_redacts(guard):
+    secret = "sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ123456"
+    report = scan(guard, f"print('{secret}')\n")
+    assert "SECRET001_LOG_SINK" in report.rule_ids
+    assert report.decision == SafetyDecision.DENY
+    assert report.redacted is True
+    assert secret not in report.model_dump_json()
+
+
 def test_secret_to_log_denies(guard):
     script = (
         "import os, logging\n"