Build an Agent Data-Exfiltration Detection Workflow Against Memory Heist & GitLost Patterns

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

July 2026 delivered the two clearest cases of agent data exfiltration yet documented. In the claude.ai memory heist, user memory was pulled out by making the agent follow a crafted web_fetch link — the agent did what the tool description said, and the data walked out through a channel everyone trusted. Days later, the GitLost pattern exfiltrated private repo READMEs from GitHub Agentic Workflows by embedding repo content in crafted public issues that workflows then routed outward as normal output. Neither attack needed a firewall bypass. Both needed the agent to ship data where it should not go. This dispatch builds exfil-guard, a LangGraph workflow that classifies every outbound data flow — fetch targets, payload sizes, memory-access requests — applies egress allowlists, redacts PII, and blocks or routes suspicious patterns to a human approval gate. Follow both incidents in the latest AI news hub, then build the gate.

Why fetch tools and memory access are the new exfiltration channel

The memory heist and GitLost attacks share a structural root: agentic tools blur the boundary between reading and sending. A web_fetch that reads a URL is also a web_fetch that can POST to one. A workflow that summarizes repo content into an issue is also a workflow that can embed that content where it will be indexed publicly. Exfiltration no longer requires a malicious tool — it requires a trusted tool used against an unexpected target. The only scalable defense is to classify every outbound flow before release: what tool, to what target, how much data, and does it touch memory context.

Architecture

flowchart TD
    A[Outbound data flow] --> B[Classify tool + target]
    B --> C[Egress allowlist check]
    C --> D[PII + payload-size scan]
    D --> E[Memory-access request check]
    E --> F{Risk score}
    F -- allow --> G[Release flow]
    F -- redact --> H[Redact PII] --> G
    F -- human --> I[Human approval gate]
    I -- approved --> G
    I -- denied --> J[Block + quarantine]
    F -- block --> J
    G --> K[Append-only audit log]
    J --> K

Project setup

mkdir exfil-guard && cd exfil-guard
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai pydantic httpx

# .env
OPENAI_API_KEY=sk-...
EGRESS_ALLOWLIST_URL=https://raw.githubusercontent.com/acme/egress-allowlist/main/allowlist.json
FETCH_PROXY_URL=http://localhost:8080
MEMORY_API_URL=http://localhost:9000
REDACT_PII=true
REVIEW_THRESHOLD=0.4
BLOCK_THRESHOLD=0.7
AUDIT_LOG_PATH=./audit/exfil-guard.log
APPROVAL_CHANNEL=slack

schemas.py

from pydantic import BaseModel, Field
from typing import Optional, Literal

class OutboundFlow(BaseModel):
    flow_id: str
    source: str = Field(..., description="agent | tool | workflow")
    tool: str = Field(..., description="web_fetch | memory_read | http_post | git_push")
    target: str = Field(..., description="destination URL, host, or endpoint")
    payload_size: int = Field(..., description="bytes of outbound data")
    content_sample: str = Field(default="", description="prefix of outbound content")
    asks_memory: bool = Field(False, description="requested memory-context access")

class FlowAssessment(BaseModel):
    flow_id: str
    allowlisted: bool
    pii_detected: bool
    risk_score: float = Field(..., ge=0.0, le=1.0)
    reasons: list[str] = Field(default_factory=list)
    verdict: Literal["allow", "redact", "human", "block"] = "block"

class AuditRecord(BaseModel):
    flow_id: str
    tool: str
    target: str
    payload_size: int
    risk_score: float
    verdict: str
    redaction_applied: bool = False

tools.py

import os, re, hashlib
import httpx

PII_PATTERNS = [
    r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b",
    r"\b\d{3}-\d{2}-\d{4}\b",
    r"\b(?:\d[ -]*?){13,16}\b",
    r"\b(?:api[_-]?key|token|secret)\s*[:=]\s*\S+",
]

def detect_pii(text: str) -> bool:
    return any(re.search(p, text, re.I) for p in PII_PATTERNS)

def redact(text: str) -> str:
    red = text
    for p in PII_PATTERNS:
        red = re.sub(p, "[REDACTED]", red, flags=re.I)
    return red

async def fetch_allowlist() -> list[str]:
    async with httpx.AsyncClient(timeout=10) as c:
        r = await c.get(os.getenv("EGRESS_ALLOWLIST_URL"))
        r.raise_for_status()
        return r.json().get("hosts", [])

def target_allowlisted(target: str, allowlist: list[str]) -> bool:
    return any(h in target for h in allowlist)

graph.py

import os, hashlib
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from schemas import OutboundFlow, FlowAssessment, AuditRecord
from tools import detect_pii, redact, fetch_allowlist, target_allowlisted

class GuardState(TypedDict):
    flow: OutboundFlow
    assessment: FlowAssessment
    verdict: str
    redacted: str

def classify_node(state: GuardState) -> GuardState:
    f = state["flow"]
    allowlist = fetch_allowlist()
    allow = target_allowlisted(f.target, allowlist)
    pii = detect_pii(f.content_sample)
    reasons, score = [], 0.0
    if f.asks_memory:
        score += 0.3
        reasons.append("memory-context request")
    if f.tool == "web_fetch":
        score += 0.1
        reasons.append("fetch tool to external target")
    if not allow:
        score += 0.3
        reasons.append("target not allowlisted")
    if pii:
        score += 0.2
        reasons.append("PII detected in payload")
    if f.payload_size > 10_000_000:
        score += 0.2
        reasons.append("large payload")
    flow_id = hashlib.sha256(f.target.encode()).hexdigest()[:8]
    assessment = FlowAssessment(flow_id=flow_id, allowlisted=allow,
        pii_detected=pii, risk_score=round(min(score, 1.0), 2), reasons=reasons)
    return {**state, "assessment": assessment}

def route(state: GuardState) -> Literal["allow", "redact", "human", "block"]:
    a = state["assessment"]
    if a.pii_detected:
        return "redact"
    if a.risk_score >= float(os.getenv("BLOCK_THRESHOLD", "0.7")):
        return "block"
    if a.risk_score >= float(os.getenv("REVIEW_THRESHOLD", "0.4")):
        return "human"
    return "allow"

def allow_node(state: GuardState) -> GuardState:
    return {**state, "verdict": "allow"}

def redact_node(state: GuardState) -> GuardState:
    return {**state, "verdict": "redact", "redacted": redact(state["flow"].content_sample)}

def human_node(state: GuardState) -> GuardState:
    return {**state, "verdict": "human"}

def block_node(state: GuardState) -> GuardState:
    return {**state, "verdict": "block"}

def audit_node(state: GuardState) -> GuardState:
    a = state["assessment"]
    rec = AuditRecord(flow_id=a.flow_id, tool=state["flow"].tool,
        target=state["flow"].target, payload_size=state["flow"].payload_size,
        risk_score=a.risk_score, verdict=state["verdict"],
        redaction_applied=state["verdict"] == "redact")
    append_audit(rec)  # append-only writer, separate process
    return state

def build_graph():
    g = StateGraph(GuardState)
    g.add_node("classify", classify_node)
    for n in ("allow", "redact", "human", "block"):
        g.add_node(n, locals()[n + "_node"])
    g.add_node("audit", audit_node)
    g.set_entry_point("classify")
    g.add_conditional_edges("classify", route, {
        "allow": "allow", "redact": "redact", "human": "human", "block": "block"})
    for n in ("allow", "redact", "human", "block"):
        g.add_edge(n, "audit")
    g.add_edge("audit", END)
    return g.compile()

main.py

import os, asyncio, json
from schemas import OutboundFlow
from graph import build_graph

async def main():
    graph = build_graph()
    flow = OutboundFlow(
        flow_id="flow-001",
        source="agent",
        tool="web_fetch",
        target="https://docs.acme.com/research",
        payload_size=4_500,
        content_sample="researcher alice@acme.com api_key=sk-... context...",
        asks_memory=True,
    )
    result = await graph.ainvoke({"flow": flow, "redacted": ""})
    print(json.dumps({
        "verdict": result["verdict"],
        "risk_score": result["assessment"].risk_score,
        "reasons": result["assessment"].reasons,
        "redacted_sample": result.get("redacted", ""),
    }, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

Scoring the Memory Heist and GitLost patterns

The classifier turns both July incidents into concrete checks. The memory-heist path is the asks_memory flag combined with a fetch to a non-allowlisted target — the exact shape of a memory exfiltration that rides on link-following. The GitLost path is a git_push or issue-write flow whose content sample contains repo-internal markers or path fragments and whose target is not in the allowlist. Neither is a perfect detector; that is why the classifier also scores payload size and PII, and why borderline flows go to a human rather than guessing. The score is a fuse, not a verdict — allowlist membership is the primary control, and the score decides how much human time each flow deserves.

Retry rules

• Allowlist fetches retry twice with exponential backoff (1s, 2s); a failed third fetch treats every target as not allowlisted — deny by default.
• Flow classification is deterministic and never retried; it consumes only already-captured flow metadata.
• PII redaction is deterministic; if redaction itself fails, the flow is blocked rather than released raw.
• Human approval notifications retry every 60s for up to 15 minutes; if no human responds, the flow expires blocked.
• The audit log write retries three times; if it fails, the flow is held and never released — no unlogged data leaves the boundary.

Redaction, release, and the human gate

When PII is detected, exfil-guard does not block the whole flow — it redacts and releases, so legitimate work continues while secrets stay home. The redacted sample is stored in the audit record so reviewers can confirm what was scrubbed. When the risk score lands between the review and block thresholds, the flow suspends and a human reviews the target, the payload sample, and the reasons list. Approve and it releases; deny and it is blocked and quarantined. Batch approvals hourly for high-frequency low-risk flows, and keep the gate bounded by an SLA so an unattended agent request expires blocked instead of lingering.

Testing with the July incident corpus

Reproduce both attacks in a test environment before connecting production agents. Feed exfil-guard a web_fetch flow to a non-allowlisted URL with memory context requested, and confirm it routes to the block path when the score crosses the threshold, or to the human gate when it lands in review territory. Feed it a git_push whose payload contains an internal repo README path, and confirm the same. Feed it a clean fetch to an allowlisted documentation host with no PII, and confirm it releases without human involvement. These three cases are the acceptance tests for the workflow — the memory-heist pattern, the GitLost pattern, and the legitimate baseline. The egress-governance patterns in the AI workflows library complement the same discipline.

The audit trail

Every classified flow is appended to audit/exfil-guard.log before release or block: flow id, tool, target, payload size, risk score, reasons, verdict, and redaction flag. The log is append-only and written by a separate process, so a compromised agent cannot delete the record of its own exfiltration attempt. For the GitLost class, the audit record also lets you reconstruct exactly which workflow routed which content where — the forensic trail that turns an incident response from guesswork into a timeline.

The bottom line

The memory heist and GitLost attacks proved that agent exfiltration now rides trusted tools and normal workflows. The defense is a gate on the way out: classify every flow, check the target against an egress allowlist, redact PII, score memory-access and size signals, and route borderline flows to a human. exfil-guard is the LangGraph implementation of that gate. Deploy it in front of fetch tools, memory access, and workflow output channels, and track the evolving attack patterns on latest AI news.

Frequently Asked Questions

What is exfil-guard?

A LangGraph workflow that classifies every outbound data flow — fetch targets, payload sizes, memory-access requests — applies egress allowlists, redacts PII, and blocks or routes suspicious exfiltration patterns to a human gate.

What was the claude.ai memory heist?

A July 2026 incident where user memory was exfiltrated by making the agent follow a crafted web_fetch link, which pulled data out through a channel the agent already trusted as normal tool use.

What is the GitLost pattern?

A July 2026 GitHub attack where private repo READMEs were exfiltrated by embedding repo content in crafted public issues, which agentic workflows then routed outward as normal output.

How does exfil-guard detect exfiltration?

It classifies each flow by tool and target, checks destinations against an egress allowlist, scans payloads for PII and size anomalies, and scores memory-access requests — then routes allow, redact, human, or block.

How does the human gate work?

When the risk score crosses the review threshold, the flow is suspended and a human reviews the target, payload sample, and reasons before approving, redacting, or blocking the flow.

Closing thoughts

Agent exfiltration no longer needs a malicious tool — it needs a trusted tool pointed at the wrong target. exfil-guard turns that realization into an outbound gate: classify, allowlist, redact, score, approve. Run it on every fetch, every push, every memory read, and the memory-heist and GitLost classes lose their channel. The defense-in-depth patterns in the AI workflows library complete the picture.