Mapping the Agent Before Anyone Attacks It

An agent that reads tickets, queries BigQuery, and sends email usually gets reviewed one tool at a time, and each tool has an owner and an access review. What often has no owner is the chain between them: text a customer writes in a ticket can steer what the agent queries and where it sends the result. Much of the agent-specific risk sits in that chain.

This series takes one working agent from threat model to release. Each part picks one way to attack it, shows the attack in a generic form, and then adds controls, along with what each one costs and where it stops helping. Our Missing Primitives series covered the building blocks of trustworthy agents in general. Here I apply them, plus the platform features that have shipped since, to one system you could run.

This first part describes the agent, maps where data and instructions get in, lines the threats up against current OWASP and MITRE ATLAS material, and adds a first control: a policy check on every tool call the agent makes through ADK.

The agent this series attacks

The reference agent is an internal operations agent, a composite of a setup we often see in client work: Microsoft Entra ID for workforce identity, Google Cloud for data and agents. An employee signs in through Entra and uses the agent for three kinds of task:

• Triage a ticket: read it, search internal documents for the relevant runbook, and add an internal comment.
• Answer a data question: query BigQuery and email the answer to colleagues.
• Reply to a ticket: read a customer's ticket, look up the facts in BigQuery, and send the customer a reply.

The agent is built with Google's Agent Development Kit (ADK) and runs on Agent Runtime (formerly Agent Engine), part of Gemini Enterprise Agent Platform (Google Cloud, n.d.c). Teams also run ADK agents on Cloud Run, and where that changes the answer, a short Cloud Run note says so. The agent reaches its tools over the Model Context Protocol (MCP):

• BigQuery, through Google's remote MCP server at bigquery.googleapis.com/mcp. As of October 2026 its documentation names six tools, and "[t]he only MCP tool that isn't read-only is execute_sql". The agent is meant to use execute_sql_readonly, which doesn't support DML, DDL, or Python UDFs (Google Cloud, n.d.o).
• Tickets, through an MCP server that the ticketing vendor operates.
• Documents, through an MCP server the company runs in front of its document store.
• Email, through an MCP server the company runs in front of its mail relay.

Keeping the agent off execute_sql is an IAM job, since the in-process controls later in this post only govern a misled model; the checks at the end of the post cover it.

Figure 1 numbers the flows. The tool results in flow 4 are where the design gets harder, because text written outside the company comes back to the agent inside them, while flow 5 sends mail to the same people who can write that text. The company backend matters as much as the runtime: it is the only caller the agent should accept, and it decides which employee and which task each session belongs to.

Two identity questions sit under every arrow in Figure 1: how an employee's Entra sign-in reaches the agent and binds to the session, which Part 2 covers, and whose authority the agent uses at each MCP server, which Part 3 covers. Google's Agent Identity gives each agent its own SPIFFE-based identity and can hold tokens that users delegate to it (Google Cloud, n.d.b). Until Part 3 shows how those tokens are handled, I treat code in the agent's process as able to use any credential the agent holds.

What is worth protecting

An agent's assets go beyond the data behind its tools. For this one:

• The data behind the tools: BigQuery tables, internal documents, and ticket history, including customer details.
• Credentials: tokens for each MCP server, refresh tokens held on the employee's behalf (by the agent or a credential broker), and wherever delegated tokens are stored for each user.
• The ability to act: sending mail from a company domain and writing to tickets both carry the company's name, even when the content is wrong.
• Session state and memory: what the agent carries from one turn or one session to the next.
• The record of what happened: tool-call logs and the decisions around them, which incident response depends on.
• Spend: model tokens and BigQuery bytes scanned, which a looping agent can burn through.

Where instructions get in

Web applications can keep code and data on separate paths, as parameterised queries do, and many injection defences rely on that separation. A language model gets its instructions and its data in the same context window, as text. Anything that reaches the context can be read by the model as a request, whatever channel it arrived through.

Table 1. Where text enters the reference agent's context.

Only the first two rows are written by people the agent works for. The third is written by whoever runs each server: Google, the ticketing vendor, and the company's own teams. The next three are written by anyone who can get text into a ticket, a document, or a table, and the last by earlier sessions, which may themselves have read any of those. The BigQuery row is easy to miss: a table that stores ticket text or customer notes carries untrusted content even though the tool reading it is internal.

The employee's own request isn't always benign either. The confirmations later in this post are approved by the employee who asked, so they help against a misled model and do little against an employee who wants data out. Where that matters, external sends of query results need a second approver or the company's existing data loss prevention controls.

Three properties in one agent

Simon Willison names the combination that turns injected text into data loss: access to private data, exposure to untrusted content, and the ability to communicate externally, which he calls the lethal trifecta (Willison, 2025). An agent with all three can be steered by text it reads into sending what it can read to wherever the text says.

The reference agent has all three, and the ticket reply task uses all three in one session. It reads a ticket (untrusted content), queries BigQuery (private data), and sends email to a customer (external communication). It is often the task teams most want to automate, which is why this series attacks it.

One path, end to end

Figure 2 shows the path in its generic form. It needs no software vulnerability, and each step is a call the agent is allowed to make.

The ticket text asks, in ordinary language, for some extra context to be included in the reply. The model has been told to be helpful with tickets, the request reads like part of the job, and the query and the email are both tools it was given. Whether a given model follows the embedded text can't be assumed either way, and that is part of the problem. A few test runs without a leak don't show the path is closed.

Each component in Figure 2 behaves as designed. The ticketing system stores what customers write, the BigQuery MCP server runs a read-only query the credentials allow, and the email server sends mail. The leak comes from the order of the calls, so this post starts with a control that can see that order. It also comes to a simple recommendation for the ticket reply task: let the agent draft the reply and have the employee send it.

If the rows in Figure 2 hold customers' personal data, the outcome is a data breach, whatever the AI Act says about the agent. In the EU, GDPR Article 33(1) requires the controller to notify the supervisory authority "without undue delay and, where feasible, not later than 72 hours after having become aware of it", unless the breach is unlikely to result in a risk to people's rights and freedoms (European Union, 2016). In the US, every state has a breach notification law (National Conference of State Legislatures, 2022), and a financial firm regulated in New York has to notify the Department of Financial Services "no later than 72 hours after determining that a cybersecurity incident has occurred" (New York State Department of Financial Services, n.d.). An EU financial entity also reports major ICT-related incidents to its competent authority under DORA (European Union, 2022). Which of these applies depends on the data and the firm, so check with counsel, but I'd put the agent on the same breach runbook as any other system that holds customer data before it ships.

A first control: a policy check on each tool call

The first control sits where the sequence is visible: between the model proposing a tool call and the tool running. ADK provides the hook. A before_tool_callback is "[c]alled just before a specific tool's run_async method is invoked, after the LLM has generated a function call for it"; returning None lets the tool run, and returning a dictionary skips it (Google, n.d.b). ATLAS describes the same idea as mitigation AML.M0030, Restrict AI Agent Tool Invocation on Untrusted Data, with human approval as AML.M0029 (MITRE, 2026).

The policy check is deny by default and works from three inputs:

• The task, fixed by the application when it creates the session, from the entry point the employee used. The model doesn't choose it, so text in a ticket shouldn't be able to promote a triage session into a reply session, provided nothing else can write that state (see Session state and credentials below).
• The tool's effect class: reads untrusted content, reads sensitive data, writes internally, or sends.
• What the session has already been exposed to: whether it has read untrusted content and whether it has read sensitive data.

From those it decides allow, confirm, or deny, checking the rows of Table 2 in order.

Table 2. How the policy check decides.

An address with anything unusual in it, including a plus tag or non-ASCII characters, counts as external. Recipient fields are denied rather than parsed when they aren't plain lists, an unexpected argument could carry an address or data, and control or bidirectional formatting characters in a subject could add mail headers or disguise what the approver sees. The policy check doesn't inspect what the body says.

Listing 1 keeps send_email in the ticket reply task so the deny is visible. For most teams a drafting tool is the better fit, as the section on costs explains.

Python

"""Deny-by-default policy check for the reference agent's tool calls.""" import re import unicodedata from dataclasses import dataclass from enum import Enum class Effect(Enum): READ_UNTRUSTED = "read_untrusted" # content anyone can write: tickets, documents  READ_SENSITIVE = "read_sensitive" # internal data: BigQuery  WRITE_INTERNAL = "write_internal" # changes inside the company: ticket comments  SEND = "send" # leaves the agent: email  TOOLS = { "get_ticket": Effect.READ_UNTRUSTED, "search_documents": Effect.READ_UNTRUSTED, "execute_sql_readonly": Effect.READ_SENSITIVE, "add_ticket_comment": Effect.WRITE_INTERNAL, "send_email": Effect.SEND, } # The application sets the task from the entry point the employee used. # The model never chooses it. TASKS = { "triage_ticket": {"get_ticket", "search_documents", "add_ticket_comment"}, "data_question": {"execute_sql_readonly", "send_email"}, "ticket_reply": {"get_ticket", "execute_sql_readonly", "send_email"}, } # A strict local part: anything unusual counts as external, including a # plus tag, which can route mail into a ticket or a list. INTERNAL = re.compile(r"[a-z0-9._-]+@example\.com") # The email tool's arguments. Any other key could carry an address or data. SEND_FIELDS = {"to", "cc", "bcc", "subject", "body"} @dataclass class Session: untrusted_seen: bool = False sensitive_read: bool = False @dataclass(frozen=True) class Decision: outcome: str # "allow", "confirm", or "deny"  reason: str def _recipients(args: dict): """Every address field, or None if any field isn't a list of strings.""" found = [] for field in ("to", "cc", "bcc"): value = args.get(field, []) if not isinstance(value, list) or not all(isinstance(r, str) for r in value): return None found += value return found def _unsafe(c: str) -> bool: """Control characters, line separators, and bidirectional formatting.""" return (unicodedata.category(c) == "Cc" or c in "\u2028\u2029\u200e\u200f" or "\u202a" <= c <= "\u202e" or "\u2066" <= c <= "\u2069") def _external(recipients) -> bool: return any(not (r.isascii() and INTERNAL.fullmatch(r.lower())) for r in recipients) def decide(task: str, tool: str, args: dict, s: Session) -> Decision: effect = TOOLS.get(tool) if effect is None: return Decision("deny", f"unknown tool {tool}") if tool not in TASKS.get(task, set()): return Decision("deny", f"{tool} is not allowed for task {task}") if effect is Effect.SEND: if set(args) - SEND_FIELDS: return Decision("deny", "unexpected argument") subject, body = args.get("subject", ""), args.get("body", "") if not (isinstance(subject, str) and isinstance(body, str)): return Decision("deny", "subject and body must be text") if any(_unsafe(c) for c in subject): return Decision("deny", "control or formatting character in subject") recipients = _recipients(args) if not recipients: return Decision("deny", "recipients missing or malformed") if _external(recipients): if s.untrusted_seen and s.sensitive_read: return Decision("deny", "external send after untrusted content and sensitive data") return Decision("confirm", "external recipient") if effect in (Effect.SEND, Effect.WRITE_INTERNAL) and s.untrusted_seen: return Decision("confirm", "side effect after untrusted content") # Record what the session has now been exposed to.  if effect is Effect.READ_UNTRUSTED: s.untrusted_seen = True if effect is Effect.READ_SENSITIVE: s.sensitive_read = True return Decision("allow", "within task policy") 

Rust

//! Deny-by-default policy check for the reference agent's tool calls. use serde_json::Value; #[derive(Clone, Copy, PartialEq, Debug)] pub enum Effect { ReadUntrusted, // content anyone can write: tickets, documents ReadSensitive, // internal data: BigQuery WriteInternal, // changes inside the company: ticket comments Send, // leaves the agent: email } pub fn effect_of(tool: &str) -> Option<Effect> { match tool { "get_ticket" | "search_documents" => Some(Effect::ReadUntrusted), "execute_sql_readonly" => Some(Effect::ReadSensitive), "add_ticket_comment" => Some(Effect::WriteInternal), "send_email" => Some(Effect::Send), _ => None, } } // The application sets the task from the entry point the employee used. // The model never chooses it. pub fn task_allows(task: &str, tool: &str) -> bool { let allowed: &[&str] = match task { "triage_ticket" => &["get_ticket", "search_documents", "add_ticket_comment"], "data_question" => &["execute_sql_readonly", "send_email"], "ticket_reply" => &["get_ticket", "execute_sql_readonly", "send_email"], _ => &[], }; allowed.contains(&tool) } #[derive(Default, Debug)] pub struct Session { pub untrusted_seen: bool, pub sensitive_read: bool, } #[derive(PartialEq, Debug)] pub enum Decision { Allow, Confirm(&'static str), Deny(String), } /// Every address field, or None if any field isn't a list of strings. fn recipients(args: &Value) -> Option<Vec<String>> { let mut found = Vec::new(); for field in ["to", "cc", "bcc"] { match args.get(field) { None => {} Some(Value::Array(items)) => { for item in items { found.push(item.as_str()?.to_string()); } } Some(_) => return None, } } Some(found) } /// The email tool's arguments. Any other key could carry an address or data. const SEND_FIELDS: [&str; 5] = ["to", "cc", "bcc", "subject", "body"]; /// Control characters, line separators, and bidirectional formatting. fn unsafe_char(c: char) -> bool { c.is_control() || matches!(c, '\u{2028}' | '\u{2029}' | '\u{200e}' | '\u{200f}') || ('\u{202a}'..='\u{202e}').contains(&c) || ('\u{2066}'..='\u{2069}').contains(&c) } /// A strict local part: anything unusual counts as external, including a /// plus tag, which can route mail into a ticket or a list. fn is_internal(address: &str) -> bool { if !address.is_ascii() { return false; } let a = address.to_ascii_lowercase(); match a.strip_suffix("@example.com") { Some(local) => { !local.is_empty() && local.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || "._-".contains(c)) } None => false, } } pub fn decide(task: &str, tool: &str, args: &Value, s: &mut Session) -> Decision { let Some(effect) = effect_of(tool) else { return Decision::Deny(format!("unknown tool {tool}")); }; if !task_allows(task, tool) { return Decision::Deny(format!("{tool} is not allowed for task {task}")); } if effect == Effect::Send { let known = args .as_object() .is_some_and(|o| o.keys().all(|k| SEND_FIELDS.contains(&k<...