Environment hooks в Gemini Managed Agents: блокировка и аудит tool calls

A system prompt that says "never touch the config directory" is a request, not a rule. A remote agent running in a managed environment needs rules that apply at the moment it calls a tool: one check just before execution that can refuse the call, and one just after that records what happened and confirms the result. This guide builds both. By the end you will have a pre-execution hook that blocks a test file write, a post-execution hook that checks every allowed write automatically, and an audit log that shows if anyone edited it.
The concrete case
Picture a remote AI agent that writes reports into a workspace. It can call three tools: read_file, write_file and, through a misconfiguration, a general shell tool. The workspace has two areas:
out/: where the agent is supposed to write results;protected/: settings the agent should never change.
We want four guarantees, and we want to be able to check each one:
- A write to
protected/settings.yamlis refused before the tool runs, including tricks likeout/../protected/settings.yaml. - Any tool the policy doesn't know (here,
shell) is refused. If the event is malformed or the hook crashes, the call is refused too. - Every allowed
write_fileis checked after it runs. The file must exist, its content must match what was requested, and.jsonfiles must parse. - Every decision, allowed or denied, goes into an append-only audit log whose entries are hash-chained, so a later edit shows up.
Why hooks and not just instructions
A hook is code the runtime calls at a fixed point in the life of a tool call. The model doesn't choose whether to run it. That is the whole difference. An instruction in the prompt can be ignored, misread or overridden by prompt injection hidden in a document the agent reads. A pre-execution hook sees the actual tool name and arguments the runtime is about to execute, and it has the last word.
The two hook points do different jobs:
| Hook point | Sees | Can do | Cannot do |
|---|---|---|---|
| Pre-execution | Tool name, arguments, call ID | Allow or deny; record the decision | Know what the tool would actually have done |
| Post-execution | The same, plus the tool result | Check side effects, flag problems, record the outcome | Undo a side effect that already happened |
Because of the last column, anything that must not happen at all belongs in the pre-hook. The post-hook is for checking and evidence, not for prevention.
Step 0. Prerequisites and layout
You need Python 3.9 or later and nothing else. Make this layout:
hooks-lab/
├── hooks/
│ ├── audit.py # shared hash-chained logger
│ ├── pre_tool_guard.py # pre-execution hook
│ └── post_tool_audit.py # post-execution hook
├── events/ # test tool-call events
├── simulate.py # local stand-in for the runtime
├── verify_audit.py # checks the log chain
├── sandbox/ # the agent's workspace
│ ├── out/
│ └── protected/
└── audit/ # created on first run
mkdir -p hooks-lab/{hooks,events,sandbox/out,sandbox/protected}
cd hooks-lab
printf 'debug: false\n' > sandbox/protected/settings.yaml
export WORKSPACE_ROOT=./sandbox
export AUDIT_LOG=./audit/tool-calls.jsonl
The local sandbox here is just a folder, not a security boundary. It lets you test the hook logic in a repeatable way before you attach the hooks to a remote environment.
Step 1. Define the hook contract
Before writing code, write down the contract. When the platform's real contract is different, you change one adapter instead of the policy.
- Input: one JSON object on stdin, shaped like
{"call_id": "...", "tool_name": "...", "arguments": {...}}. The post-hook also gets"result". - Pre-hook output: one JSON line,
{"decision": "allow"|"deny", "reason": "..."}. Exit code0means allow and2means deny. - Post-hook output:
{"status": "ok"|"flag"|"error", "problems": [...]}. Exit code0means ok, anything else means it needs attention. - Failure rule: if the pre-hook can't decide (bad input, exception, timeout), the answer is deny. This is fail-closed behaviour, and it matters more than any single rule in the policy.
Step 2. A shared, tamper-evident audit logger
Each record stores the hash of the record before it, and its own hash is computed over everything except that field. If you edit, delete or reorder a line, the chain breaks from that point onward. The log does not store file contents, only a SHA-256 of the arguments, so secrets the agent writes don't end up in the audit trail.
# hooks/audit.py
import hashlib
import json
import os
import time
LOG = os.environ.get("AUDIT_LOG", "./audit/tool-calls.jsonl")
GENESIS = "0" * 64
def _digest(record):
canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def _last_hash():
try:
with open(LOG, encoding="utf-8") as f:
lines = [line for line in f if line.strip()]
except FileNotFoundError:
return GENESIS
return json.loads(lines[-1])["entry_hash"] if lines else GENESIS
def append(phase, event, status, detail):
os.makedirs(os.path.dirname(LOG) or ".", exist_ok=True)
ev = event if isinstance(event, dict) else {}
args = json.dumps(ev.get("arguments"), sort_keys=True)
record = {
"ts": round(time.time(), 3),
"phase": phase,
"call_id": ev.get("call_id"),
"tool_name": ev.get("tool_name"),
"args_sha256": hashlib.sha256(args.encode("utf-8")).hexdigest(),
"status": status,
"detail": detail,
"prev_hash": _last_hash(),
}
record["entry_hash"] = _digest(record)
with open(LOG, "a", encoding="utf-8") as f:
f.write(json.dumps(record, sort_keys=True) + "\n")
Step 3. The pre-execution hook: block before it runs
The policy is an allowlist. Read tools pass, write tools pass only inside out/, and everything else is denied. Paths are resolved with realpath and compared with commonpath, which closes the simple path traversal cases (../, absolute paths, symlinks that already exist).
# hooks/pre_tool_guard.py
import json
import os
import sys
from audit import append
ROOT = os.path.realpath(os.environ.get("WORKSPACE_ROOT", "./sandbox"))
WRITABLE = [os.path.join(ROOT, "out")]
READ_TOOLS = {"read_file", "list_dir"}
WRITE_TOOLS = {"write_file"}
def decide(event):
tool = event.get("tool_name")
args = event.get("arguments") or {}
if tool in READ_TOOLS:
return "allow", "read-only tool"
if tool not in WRITE_TOOLS:
return "deny", f"tool not in allowlist: {tool}"
raw = args.get("path")
if not isinstance(raw, str) or not raw:
return "deny", "missing path"
target = os.path.realpath(os.path.join(ROOT, raw))
for allowed in WRITABLE:
if os.path.commonpath([target, allowed]) == allowed:
return "allow", f"inside {os.path.relpath(allowed, ROOT)}/"
return "deny", f"write outside writable dirs: {os.path.relpath(target, ROOT)}"
def main():
event = None
try:
event = json.loads(sys.stdin.read())
decision, reason = decide(event)
except Exception as exc:
decision, reason = "deny", f"hook error: {type(exc).__name__}"
try:
append("pre", event, decision, reason)
except Exception:
decision, reason = "deny", "audit write failed"
print(json.dumps({"decision": decision, "reason": reason}))
sys.exit(0 if decision == "allow" else 2)
if __name__ == "__main__":
main()
Two design choices are worth noting. First, if the audit write fails, the call is denied. An allowed call with no record is exactly what an audit is supposed to rule out. Second, a non-object event such as [] raises AttributeError inside decide, which ends up as a deny rather than a crash that the runtime might read as "no objection".
Step 4. The post-execution hook: check what actually happened
# hooks/post_tool_audit.py
import json
import os
import sys
from audit import append
ROOT = os.path.realpath(os.environ.get("WORKSPACE_ROOT", "./sandbox"))
def check(event):
if event.get("tool_name") != "write_file":
return "ok", []
args = event["arguments"]
path = os.path.realpath(os.path.join(ROOT, args["path"]))
if not os.path.isfile(path):
return "flag", ["file not found after write"]
with open(path, "rb") as f:
data = f.read()
problems = []
if data != args.get("content", "").encode("utf-8"):
problems.append("content differs from requested")
if path.endswith(".json"):
try:
json.loads(data)
except ValueError:
problems.append("invalid JSON")
return ("flag" if problems else "ok"), problems
def main():
event = None
try:
event = json.loads(sys.stdin.read())
status, problems = check(event)
except Exception as exc:
status, problems = "error", [f"hook error: {type(exc).__name__}"]
append("post", event, status, "; ".join(problems) or "checks passed")
print(json.dumps({"status": status, "problems": problems}))
sys.exit(0 if status == "ok" else 1)
if __name__ == "__main__":
main()
Replace the JSON check with whatever "correct" means for your outputs: a schema validator, a linter, a unit test, a size limit. Keep it fast and deterministic, because it runs after every allowed call.
Step 5. A local runtime stand-in and test events
simulate.py plays the role of the managed runtime. It sends the raw event to the pre-hook with a timeout, runs the tool only if the call is allowed, then runs the post-hook. The order is the same one you expect from the platform, so the policy can be tested without the platform.
# simulate.py
import json
import os
import subprocess
import sys
ROOT = os.path.realpath(os.environ.get("WORKSPACE_ROOT", "./sandbox"))
TIMEOUT = 5
def run_hook(script, payload):
proc = subprocess.run(
[sys.executable, script], input=payload,
capture_output=True, text=True, timeout=TIMEOUT,
)
return proc.returncode, proc.stdout.strip()
def execute(event):
if event["tool_name"] != "write_file":
return {"ok": True, "note": "simulated"}
args = event["arguments"]
path = os.path.realpath(os.path.join(ROOT, args["path"]))
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(args.get("content", ""))
return {"ok": True}
raw = open(sys.argv[1], encoding="utf-8").read()
try:
code, out = run_hook("hooks/pre_tool_guard.py", raw)
except subprocess.TimeoutExpired:
code, out = 2, '{"decision": "deny", "reason": "pre-hook timeout"}'
print("pre :", code, out)
if code != 0:
sys.exit(2)
event = json.loads(raw)
result = execute(event)
code, out = run_hook("hooks/post_tool_audit.py", json.dumps({**event, "result": result}))
print("post:", code, out)
sys.exit(code)
Now make six events. Each one targets one guarantee from the case above.
cat > events/01-blocked.json <<'EOF'
{"call_id": "t1", "tool_name": "write_file",
"arguments": {"path": "protected/settings.yaml", "content": "debug: true\n"}}
EOF
cat > events/02-traversal.json <<'EOF'
{"call_id": "t2", "tool_name": "write_file",
"arguments": {"path": "out/../protected/settings.yaml", "content": "debug: true\n"}}
EOF
cat > events/03-allowed.json <<'EOF'
{"call_id": "t3", "tool_name": "write_file",
"arguments": {"path": "out/report.json", "content": "{\"status\": \"ok\"}\n"}}
EOF
cat > events/04-bad-json.json <<'EOF'
{"call_id": "t4", "tool_name": "write_file",
"arguments": {"path": "out/broken.json", "content": "{status: ok"}}
EOF
printf 'this is not json' > events/05-malformed.json
cat > events/06-unknown-tool.json <<'EOF'
{"call_id": "t6", "tool_name": "shell", "arguments": {"cmd": "rm -rf out"}}
EOF
Run them all:
for e in events/*.json; do
echo "== $e"; python3 simulate.py "$e"; echo "exit=$?"
done
Expected results
These are the outcomes the code above is written to produce. Compare them with what you actually see, and treat any mismatch as a bug to fix before you go further.
| Event | Pre-hook | Tool runs? | Post-hook | Final exit |
|---|---|---|---|---|
| 01-blocked | deny: write outside writable dirs: protected/settings.yaml | No | Not called | 2 |
| 02-traversal | deny: same reason (path resolved first) | No | Not called | 2 |
| 03-allowed | allow: inside out/ | Yes | ok, no problems | 0 |
| 04-bad-json | allow: inside out/ | Yes | flag: invalid JSON | 1 |
| 05-malformed | deny: hook error: JSONDecodeError | No | Not called | 2 |
| 06-unknown-tool | deny: tool not in allowlist: shell | No | Not called | 2 |
Then confirm that the blocked write really didn't happen:
cat sandbox/protected/settings.yaml # must still print: debug: false
Event 04 shows what the post-hook is for and where it stops. The bad file was written, because the post-hook can only report it. If invalid output must never land in out/, move that check into the pre-hook (validate content before the write), or have the agent write to a staging directory that a separate step promotes.
Step 6. Verify the audit chain
# verify_audit.py
import json
import sys
sys.path.insert(0, "hooks")
from audit import GENESIS, LOG, _digest
prev, n = GENESIS, 0
with open(LOG, encoding="utf-8") as f:
for n, line in enumerate(f, 1):
rec = json.loads(line)
claimed = rec.pop("entry_hash")
if rec["prev_hash"] != prev or _digest(rec) != claimed:
print(f"BROKEN at line {n}")
sys.exit(1)
prev = claimed
print(f"OK: {n} entries")
python3 verify_audit.py
# expected after the six events: OK: 8 entries
# (6 pre-hook records + 2 post-hook records for events 03 and 04)
Now break it on purpose, to prove the check works:
cp audit/tool-calls.jsonl /tmp/backup.jsonl
sed -i '1s/"deny"/"allow"/' audit/tool-calls.jsonl
python3 verify_audit.py # expected: BROKEN at line 1
cp /tmp/backup.jsonl audit/tool-calls.jsonl
Be precise about what this proves. The chain shows edits to the log. It does not stop someone with write access from rewriting the whole file and recomputing every hash. To protect against that, regularly copy the latest entry_hash somewhere the agent environment can't write: a separate storage bucket, a ticket, or a commit in another repository.
Step 7. Attach the hooks to the managed environment
This is where you need the platform documentation for your Gemini Managed Agents version. Before you wire anything, find answers to these questions there:
- Where are pre- and post-execution hooks declared (environment config, agent definition, API field)?
- What is the exact payload schema, and how does it name the tool, the arguments and the call ID?
- How does a hook signal deny: exit code, response field, or both? What does the agent see when a call is denied?
- What happens on hook timeout or crash? Is the default fail-open or fail-closed, and can you change it?
- Do hooks run inside the same environment as the tools (where the agent could change them) or outside it?
- Which tools go through hooks? Are built-in or platform tools also covered?
Once you have the answers, write a thin adapter that converts the platform payload into the contract from Step 1, and leave the policy code as it is. The shape below is hypothetical. It shows where the pieces go, not real field names:
# HYPOTHETICAL SHAPE — map to the real schema from the official docs
environment:
hooks:
pre_tool_execution:
command: ["python3", "/opt/hooks/pre_tool_guard.py"]
timeout_seconds: 5
on_failure: deny # confirm this option exists; if not, see Failure cases
post_tool_execution:
command: ["python3", "/opt/hooks/post_tool_audit.py"]
timeout_seconds: 10
env:
WORKSPACE_ROOT: /workspace
AUDIT_LOG: /var/log/agent/tool-calls.jsonl
Repeat the checks from Step 5 against the real environment: ask the agent to do each of the six actions in plain language, then read the audit log. A guarantee only counts as verified on the platform once you have seen it work there.
Failure cases to test on purpose
- Hook timeout. Add
time.sleep(10)to the pre-hook. In the simulator this turns into a deny. On the platform, check what happens. If the default lets the call through, you have a fail-open gap, and you can't close it from inside the hook. - Hook not found or not executable. Rename the script. A runtime that treats "hook missing" as "no objection" quietly disables your policy.
- Symlink created after the check.
realpathresolves links that exist at check time. If the agent can create a link insideout/pointing atprotected/and then write through it, the hook alone won't catch it (this is a time-of-check to time-of-use race). Deny symlink creation too, or enforce write access at the filesystem level. - Tool aliases. If a second tool such as
apply_patchorshellcan also write files, it gets around a rule that only knowswrite_file. The allowlist in Step 3 denies unknown tools for exactly this reason. Keep it that way. - Parallel calls.
audit.pyreads the last hash and then appends, which isn't atomic. Two hooks running at once can fork the chain. Wrap the append in a file lock (fcntl.flock), or send records to a single writer process. - Audit log inside the agent's reach. If the log path is writable by the agent's tools, put it outside
WORKSPACE_ROOTand add a deny rule for it.
Limitations
- Hooks control tool calls. They don't control what the model says, and they don't cover side effects that happen outside the tools the hooks see.
- The post-hook checks after the fact. It reports damage but doesn't undo it.
- The path policy is only as strong as the runtime's guarantee that the path it checked is the path the tool uses. Filesystem permissions are still the backstop.
- A hash chain shows edits to the log. It doesn't prevent a full rewrite unless you anchor hashes outside the environment.
- Hook latency adds to every call. Measure it in your own environment rather than assuming it is negligible.
Checklist before you rely on it
- [ ] All six local events produce the expected results.
- [ ]
protected/settings.yamlis unchanged after the run. - [ ]
verify_audit.pypasses, and fails after a deliberate edit. - [ ] Platform behaviour on hook timeout, crash and missing hook is confirmed from the docs and by a test.
- [ ] Every tool that can write is either covered by the policy or denied.
- [ ] The audit log sits outside the agent's writable area, and its latest hash is copied somewhere else.
For related work on controlling and testing agents, see the guides index. For the terms used here, see the glossary.