Detect privilege escalation in agent behavior

Introduction

When you grant an AI agent permission to read configs, modify access controls, and restart services, you've handed it the building blocks of an escalation chain — even though no individual tool call looks dangerous on its own. Teams that ship agents without trajectory-level monitoring usually find out the hard way: a prompt-injected agent strings three approved actions together, dumps credentials, and exfiltrates data through API calls that the per-call policy gate happily authorised. By the end of this lesson you'll be able to detect the four canonical escalation patterns — sequence chains, parameter injection, repeated denials, and scope creep — and wire an automated response that logs evidence, blocks the session, and alerts the security team the moment a pattern fires.

Key Terminology

• Privilege escalation — an agent acquiring capabilities beyond what its operator authorised; emerges from sequences of individually-approved actions, not from a single forbidden call.
• Tool call trajectory — the ordered, time-stamped record of every tool an agent invoked in a session (parameters, results, allow/deny outcome); the unit of analysis for escalation detection.
• Prompt injection — adversary-supplied text embedded in data the agent reads (documents, API responses, database rows) that redirects the agent toward unauthorised actions while leaving the user's original instruction intact.
• Escalation chain — a sequence of tool calls where each step is benign in isolation but the combined effect achieves unauthorised access; canonical example read_config → modify_config → restart_service.
• Scope creep — an agent operating on resources outside its declared namespace, database, path-prefix, or API-domain allowlist, signalling that authority is leaking beyond the assignment boundary.

Concepts

Why agents escalate privileges

Agents do not "decide" to escalate the way a human attacker does. Two mechanisms drive it. Prompt injection plants instructions in data the agent processes — a wiki page, a database row, an HTTP response — that tell the model to take unauthorised actions while keeping the user's request superficially intact. Goal-directed reasoning is more insidious: an agent told to "fix the production database" may reason that the fix requires admin credentials, read config files to find them, modify access controls, and restart services — an escalation chain that no human authorised and no single tool call would have denied.

Sequence-based patterns

Most escalations are recognisable as a fixed sequence of tool calls within a bounded time window. The canonical chain read_config → modify_config → restart_service is low/medium/high risk per step but unambiguous in combination. A sequence detector keeps a list of named patterns — each with an ordered tool list, severity, and window — then scans every session's trajectory for matches. Other high-yield patterns: read_secrets → http_request (credential harvest), get_permissions → set_permissions (self-escalation), query_database → write_file → http_request (data staging for exfiltration). The detector returns one record per match so the responder (see Code Walkthrough) can act per detection, not per session.

Parameter injection

Even a single tool call can be an escalation vector if its parameters carry SQL injection (UNION SELECT, ' OR 1=1), command injection (; rm -rf, $(cat /etc/passwd)), path traversal (../../etc/shadow), LDAP filter manipulation, or server-side template syntax. A parameter-injection detector compiles a regex catalogue per attack family and scans every string parameter passed to every tool. It runs alongside the sequence detector — escalation can arrive in one rich call or in many small ones.

Repeated permission-denied attempts

Persistent unauthorised intent shows up as denial clusters: after read_secrets is denied, the agent tries get_config (hoping secrets leak), then list_files against /etc/secrets, then execute_command "cat /etc/secrets". Each denial is correctly enforced, but three related denials inside a five-minute window means the session should be blocked, not just rate-limited.

Scope creep

Every agent should ship with a ScopeDefinition — allowed namespaces, databases, path prefixes, API domains. Scope creep is any tool call whose parameters reference a resource outside that allowlist. An agent assigned production-web touching production-database is creep; an agent calling https://attacker.example/upload when only api.internal is allowlisted is creep with critical severity.

LLM trajectory analysis and automated response

Rule-based detectors miss subtle cases — slow scope expansion across sessions, unauthorised outcomes assembled from individually-authorised steps. A second-pass LLM analyser reads the formatted trajectory and returns a risk score, named patterns with evidence, and an allow|monitor|block recommendation. Whichever detector fires, the response protocol is identical: log the trajectory with detection evidence, block the session, page the security team with severity and pattern name.

Code Walkthrough

Now that you have the four detector concepts and the shared response protocol in mind, the pipeline below stitches them into one entry point. It demonstrates sequence-based detection, scope-creep checking, denial-cluster counting, and the three-step response (log, block, alert). The parameter-injection scanner and the LLM second pass plug into the same evaluate_call interface.

Code snippetpython
1from datetime import datetime
2from typing import Any, Dict, List, Optional
3from urllib.parse import urlparse
4from pydantic import BaseModel, Field
5
6class ToolCallRecord(BaseModel):
7    timestamp: datetime
8    tool_name: str
9    parameters: Dict[str, Any]
10    session_id: str
11    was_allowed: bool
12    result: Optional[str] = None
13
14class EscalationPattern(BaseModel):
15    pattern_id: str
16    name: str
17    tool_sequence: List[str]
18    severity: str                       # critical | high | medium
19    window_seconds: int = 300
20
21class ScopeDefinition(BaseModel):
22    agent_id: str
23    allowed_namespaces: List[str] = Field(default_factory=list)
24    allowed_api_domains: List[str] = Field(default_factory=list)
25
26DEFAULT_PATTERNS = [
27    EscalationPattern(
28        pattern_id="config_modify_restart",
29        name="Config Modification and Restart",
30        tool_sequence=["read_config", "modify_config", "restart_service"],
31        severity="critical",
32    ),
33    EscalationPattern(
34        pattern_id="credential_harvest",
35        name="Credential Harvesting",
36        tool_sequence=["read_secrets", "http_request"],
37        severity="critical",
38        window_seconds=120,
39    ),
40    EscalationPattern(
41        pattern_id="permission_self_escalation",
42        name="Permission Self-Escalation",
43        tool_sequence=["get_permissions", "set_permissions"],
44        severity="critical",
45        window_seconds=180,
46    ),
47]
48
49class EscalationPipeline:
50    """Sequence + scope + denial detectors sharing one response path."""
51
52    def __init__(self, scope: ScopeDefinition, responder,
53                 patterns: List[EscalationPattern] = DEFAULT_PATTERNS,
54                 denial_threshold: int = 3, denial_window: int = 300):
55        self.scope = scope
56        self.responder = responder
57        self.patterns = patterns
58        self.denial_threshold = denial_threshold
59        self.denial_window = denial_window
60
61    async def evaluate_call(
62        self, trajectory: List[ToolCallRecord], call: ToolCallRecord,
63    ) -> Optional[Dict[str, Any]]:
64        if creep := self._check_scope(call):
65            return await self._respond(call.session_id, creep, trajectory)
66
67        full = trajectory + [call]
68        if cluster := self._denial_cluster(full):
69            return await self._respond(call.session_id, cluster, full)
70
71        for pattern in self.patterns:
72            if match := self._match_sequence(full, pattern):
73                detection = {
74                    "pattern_id": pattern.pattern_id,
75                    "pattern_name": pattern.name,
76                    "severity": pattern.severity,
77                    "matched_calls": [c.tool_name for c in match],
78                }
79                return await self._respond(call.session_id, detection, full)
80        return None
81
82    def _check_scope(self, call: ToolCallRecord) -> Optional[Dict[str, Any]]:
83        ns = call.parameters.get("namespace")
84        if ns and ns not in self.scope.allowed_namespaces:
85            return {"pattern_name": "scope_creep", "severity": "high",
86                    "resource": "namespace", "requested": ns}
87        url = call.parameters.get("url")
88        if url:
89            host = urlparse(url).hostname
90            if host and host not in self.scope.allowed_api_domains:
91                return {"pattern_name": "scope_creep", "severity": "critical",
92                        "resource": "api_domain", "requested": host}
93        return None
94
95    def _denial_cluster(
96        self, trajectory: List[ToolCallRecord],
97    ) -> Optional[Dict[str, Any]]:
98        denials = [c for c in trajectory if not c.was_allowed]
99        if len(denials) < self.denial_threshold:
100            return None
101        window = denials[-self.denial_threshold:]
102        span = (window[-1].timestamp - window[0].timestamp).total_seconds()
103        if span <= self.denial_window:
104            return {"pattern_name": "repeated_denial", "severity": "high",
105                    "denied_tools": [c.tool_name for c in window]}
106        return None
107
108    def _match_sequence(
109        self, trajectory: List[ToolCallRecord], pattern: EscalationPattern,
110    ) -> Optional[List[ToolCallRecord]]:
111        seq, n = pattern.tool_sequence, len(pattern.tool_sequence)
112        for i, start in enumerate(trajectory):
113            if start.tool_name != seq[0]:
114                continue
115            match, idx = [start], 1
116            for nxt in trajectory[i + 1:]:
117                if idx >= n:
118                    break
119                elapsed = (nxt.timestamp - start.timestamp).total_seconds()
120                if elapsed > pattern.window_seconds:
121                    break
122                if nxt.tool_name == seq[idx]:
123                    match.append(nxt)
124                    idx += 1
125            if idx == n:
126                return match
127        return None
128
129    async def _respond(self, session_id, detection, trajectory):
130        await self.responder.log(session_id, detection, trajectory)
131        await self.responder.block(session_id, detection["pattern_name"])
132        await self.responder.alert(detection["severity"], detection)
133        return detection

You'll know it works when an injected read_config → modify_config → restart_service sequence fires config_modify_restart within the 300-second window, the session is blocked, and the security channel receives exactly one alert per detection — not one per tool call.

Do's and Don'ts

Building on the detectors, response protocol, and discipline application above, here are the operating rules that keep the pipeline trustworthy in production.

Do's

1. ✓Do analyse the full trajectory — escalation is a sequence property, not a per-call property; per-call policy gates cannot see chains.
2. ✓Do block the session, not just the call — once a pattern fires, the agent has unauthorised intent and further calls compound risk.
3. ✓Do ship every agent with an explicit ScopeDefinition — namespaces, databases, path prefixes, API domains; a missing allowlist is an open allowlist.

Don'ts

1. ✗Don't rely solely on regex parameter scanning — chained calls with clean parameters slip through; pair it with sequence detection.
2. ✗Don't treat repeated denials as harmless — three denials in five minutes is the agent searching for a way around the gate, not a stuck retry loop.
3. ✗Don't put detection inside the agent's own process — host the pipeline out-of-band so a prompt-injected agent cannot disable its own monitor.