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

Environment hooks в Gemini Managed Agents: блокировка и аудит tool calls
Temporary fallback cover; replace in editorial pass.
Level: advanced · Reading time: 45 minutes · Updated:

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:

We want four guarantees, and we want to be able to check each one:

  1. A write to protected/settings.yaml is refused before the tool runs, including tricks like out/../protected/settings.yaml.
  2. 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.
  3. Every allowed write_file is checked after it runs. The file must exist, its content must match what was requested, and .json files must parse.
  4. 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 pointSeesCan doCannot do
Pre-executionTool name, arguments, call IDAllow or deny; record the decisionKnow what the tool would actually have done
Post-executionThe same, plus the tool resultCheck side effects, flag problems, record the outcomeUndo 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.

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.

EventPre-hookTool runs?Post-hookFinal exit
01-blockeddeny: write outside writable dirs: protected/settings.yamlNoNot called2
02-traversaldeny: same reason (path resolved first)NoNot called2
03-allowedallow: inside out/Yesok, no problems0
04-bad-jsonallow: inside out/Yesflag: invalid JSON1
05-malformeddeny: hook error: JSONDecodeErrorNoNot called2
06-unknown-tooldeny: tool not in allowlist: shellNoNot called2

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:

  1. Where are pre- and post-execution hooks declared (environment config, agent definition, API field)?
  2. What is the exact payload schema, and how does it name the tool, the arguments and the call ID?
  3. How does a hook signal deny: exit code, response field, or both? What does the agent see when a call is denied?
  4. What happens on hook timeout or crash? Is the default fail-open or fail-closed, and can you change it?
  5. Do hooks run inside the same environment as the tools (where the agent could change them) or outside it?
  6. 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

Limitations

Checklist before you rely on it

For related work on controlling and testing agents, see the guides index. For the terms used here, see the glossary.

We publish what works for us—and implement the same solutions for your business. We design AI automation, Telegram bots, chats, and AI agents for real-world processes. Discuss your project →