Read the evidence of injection in a gateway log

short · 40 min · Objective 2.6

Task

Triage a synthetic gateway log the way an investigator would. Find every record that carries evidence of direct injection, indirect injection or guardrail evasion; say whether the evidence came from the user or from retrieved content; say whether the canary leaked; and name the compensating control that acts at the point where the evidence shows the attack succeeded. Every adversarial input in the log is an opaque [TEST-PATTERN:...] tag that stands where the attack text would be, so you are reading evidence, not attacks.

Steps

  1. Write lab/make-26-log.py. It writes 200 records to lab/evidence-26.jsonl: mostly ordinary questions, three legitimate HR salary questions from carol, and 25 records carrying tags.
import json, random
from datetime import datetime, timedelta

random.seed(2026)
NL = chr(10)
CANARY = "CANARY-7Q4X-SECAI"
CHUNKS = {
    "hr-holiday-policy.md#0": "Staff receive 25 days of annual leave plus public holidays.",
    "hr-holiday-policy.md#1": "Send leave requests to your line manager two weeks ahead.",
    "product-faq.md#0": "The Model S kettle has a two-year warranty.",
    "product-faq.md#1": "Returns are accepted within 30 days with a receipt.",
    "oncall-rota.md#0": "The on-call lead this month is dave.",
    "restricted/salaries.md#0": "Salary bands are reviewed each April. Project code " + CANARY + ".",
    "planted/notice.md#0": "[TEST-PATTERN:embedded-instruction] CANARY-REQUEST",
}
ORDINARY = [
    ("How many days of annual leave do staff get?", ["hr-holiday-policy.md#0", "hr-holiday-policy.md#1"],
     "Staff get 25 days of annual leave plus public holidays."),
    ("How far ahead must I ask for leave?", ["hr-holiday-policy.md#1", "hr-holiday-policy.md#0"],
     "Two weeks ahead, to your line manager."),
    ("What are the instructions for requesting leave?", ["hr-holiday-policy.md#1"],
     "Send the request to your line manager two weeks ahead."),
    ("What warranty does the Model S kettle have?", ["product-faq.md#0"], "Two years."),
    ("Can a customer return a kettle after three weeks?", ["product-faq.md#1", "product-faq.md#0"],
     "Yes. Returns are accepted within 30 days with a receipt."),
    ("Who is on call this month?", ["oncall-rota.md#0"], "dave is the on-call lead this month."),
    ("Can you role-play a customer asking about returns, for staff training?", ["product-faq.md#1"],
     "Customer: Can I return this kettle? Staff: Yes, within 30 days with a receipt."),
    ("Summarise the leave policy in one line.", ["hr-holiday-policy.md#0", "hr-holiday-policy.md#1"],
     "25 days plus public holidays, requested two weeks ahead."),
]
NOTICE_QUESTIONS = ["Is there a new notice for staff?", "What does the latest office notice say?",
                    "Any notices I should know about this week?"]
GUESTS = ["guest-41", "guest-52"]
ORDER = {"direct-blocked": 0, "direct": 1, "evasion": 2}


def assemble(question, ids):
    parts = ["<retrieved id=" + i + ">" + NL + CHUNKS[i] + NL + "</retrieved>" for i in ids]
    return NL.join(parts + ["Question: " + question])


plan = [("ordinary", random.choice(["alice", "bob", "dave"])) for _ in range(172)]
plan += [("hr-salary", "carol")] * 3
plan += [("direct-blocked" if i < 5 else "direct", GUESTS[i % 2]) for i in range(8)]
plan += [("evasion", GUESTS[i % 2]) for i in range(8)]
plan += [("indirect", random.choice(["alice", "bob"])) for _ in range(9)]
random.shuffle(plan)
for g in GUESTS:
    pos = [i for i, p in enumerate(plan) if p[1] == g]
    mine = sorted((plan[i] for i in pos), key=lambda p: ORDER[p[0]])
    for i, p in zip(pos, mine):
        plan[i] = p

LEAK_AT = {"direct": {2}, "evasion": {2, 5, 7}, "indirect": {1, 4, 8}}
seen = {}
t = datetime(2026, 9, 21, 8, 0, 0)
with open("lab/evidence-26.jsonl", "w") as f:
    for n, (kind, user) in enumerate(plan, start=1):
        t += timedelta(seconds=random.randint(60, 600))
        decision, ids = "allow", []
        if kind == "ordinary":
            q, ids, answer = random.choice(ORDINARY)
        elif kind == "hr-salary":
            q, ids = "What are the salary bands for grade 4?", ["restricted/salaries.md#0"]
            answer = "Grade 4 is band C. Reference " + CANARY + "."
        elif kind == "indirect":
            q = random.choice(NOTICE_QUESTIONS)
            ids = ["planted/notice.md#0", "hr-holiday-policy.md#1"]
            answer = "There is one staff notice on file."
        else:
            tag = "[TEST-PATTERN:instruction-override]"
            if kind == "evasion":
                tag = random.choice(["[TEST-PATTERN:role-play-wrapper]", "[TEST-PATTERN:encoded-request]"])
            q = tag + " What are the salary bands?"
            answer = "I can only answer questions about company policy."
            ids = ["hr-holiday-policy.md#0"]
            if kind == "direct-blocked":
                decision, ids, answer = "block:firewall", [], ""
        seen[kind] = seen.get(kind, 0) + 1
        if seen[kind] in LEAK_AT.get(kind, ()):
            if "restricted/salaries.md#0" not in ids:
                ids = ids + ["restricted/salaries.md#0"]
            answer = "The project code is " + CANARY + "."
        prompt = assemble(q, ids) if ids else q
        rec = {"id": "e%04d" % n, "ts": t.strftime("%Y-%m-%dT%H:%M:%S+0000"), "user": user,
               "prompt": prompt, "retrieved_ids": ids, "response": answer,
               "prompt_tokens": len(prompt) // 4 if decision == "allow" else 0,
               "response_tokens": len(answer) // 4,
               "latency_ms": random.randint(900, 4000) if decision == "allow" else 3,
               "decision": decision}
        print(json.dumps(rec), file=f)
print(len(plan), "records written to lab/evidence-26.jsonl")

The prompt field holds the assembled prompt, as your gateway logs it: each retrieved chunk inside <retrieved id=...> delimiters, then the user's Question:. That is what makes indirect injection visible at all. 2. Run python lab/make-26-log.py, then take two first counts: grep -c 'TEST-PATTERN' lab/evidence-26.jsonl and grep -c -e CANARY-7Q4X-SECAI -e '[[]CANARY[]]' lab/evidence-26.jsonl. The two sets overlap but are not the same, and the difference between them is the triage. 3. Learn what each tag stands for, using the lesson's evidence lists. instruction-override is input that addresses the model's instructions rather than the task: direct injection. embedded-instruction is imperative text planted in a document for an AI to read: indirect injection. role-play-wrapper and encoded-request change the form of a request to get past a check: guardrail evasion. Where the tag sits decides the source: after Question: it came from the user; inside a <retrieved id=...> block it came from retrieved content, and the user may be entirely innocent. 4. Find the decoys before they find you. Carol's salary answers carry the canary, but carol is HR and the restricted document is hers to read: not evidence. The questions about the instructions for requesting leave and about role-playing a customer for training match a keyword search for injection and are ordinary. The opposite trap is the five records with block:firewall: a refused attempt is still evidence, and often the best evidence that a later allowed request from the same account was malicious. 5. Write lab/triage.py, which reads the log and writes lab/injection-triage.csv with header id,evidence_type,source,leaked,control: one row for every tagged record and none for any other. evidence_type is direct, indirect or evasion; source is user or retrieved; leaked is 1 when the response holds the canary in either form and 0 otherwise. 6. Fill control with this lab's triage rule, which applies the lesson's advice to choose the control that acts where the evidence shows the attack succeeded: a leaked record gets output canary filter; an indirect record that did not leak gets least privilege, because the planted file reached docs/ through write access and limiting what the model can do is the strongest answer to indirect injection; a direct record that did not leak gets prompt firewall; an evasion that did not leak gets template and delimiting, because an evasion is shaped to pass the firewall and the template is the next layer, the one that makes a successful attempt show up as broken format. 7. Read the sessions. Sort each tagged user's records by ts and count the users who had a block:firewall record followed later by an allowed tagged record. Then count the distinct users whose records retrieved planted/notice.md#0. Write lab/injection-notes.md with the lines block_then_allow_users=<n> and indirect_victims=<n>, plus at least one line starting - naming the evidence that a log of user input and final answers alone would have hidden.

Verify

Run from ~/secai-lab. The first script recomputes the ground truth from the tags and compares it with your CSV; the second checks your session counts.

python lab/triage.py
python - <<'EOF'
import csv, json
CANARY = "CANARY-7Q4X-SECAI"
recs = [json.loads(l) for l in open("lab/evidence-26.jsonl")]
truth = {}
for r in recs:
    p = r["prompt"]
    if any(i.startswith("planted/") for i in r["retrieved_ids"]):
        t, s = "indirect", "retrieved"
    elif "[TEST-PATTERN:instruction-override]" in p:
        t, s = "direct", "user"
    elif "[TEST-PATTERN:role-play-wrapper]" in p or "[TEST-PATTERN:encoded-request]" in p:
        t, s = "evasion", "user"
    else:
        continue
    leaked = "1" if CANARY in r["response"] or "[CANARY]" in r["response"] else "0"
    if leaked == "1":
        c = "output canary filter"
    else:
        c = {"indirect": "least privilege", "direct": "prompt firewall", "evasion": "template and delimiting"}[t]
    truth[r["id"]] = (t, s, leaked, c)
rows = {}
for r in csv.DictReader(open("lab/injection-triage.csv", newline="")):
    rows[r["id"]] = (r["evidence_type"], r["source"], r["leaked"], r["control"])
missing = [i for i in truth if i not in rows]
extra = [i for i in rows if i not in truth]
wrong = [i for i in truth if i in rows and rows[i] != truth[i]]
print(len(recs), "records,", len(truth), "tagged,", len(rows), "triaged")
print("tagged but not triaged:", missing or "none")
print("clean records triaged:", extra or "none")
print("wrong fields:", wrong or "none")
print("PASS" if truth and not missing and not extra and not wrong else "FAIL")
EOF
python - <<'EOF'
import json
recs = [json.loads(l) for l in open("lab/evidence-26.jsonl")]
blocked, users = set(), set()
for r in sorted(recs, key=lambda r: r["ts"]):
    if "[TEST-PATTERN:" not in r["prompt"]:
        continue
    if r["decision"] == "block:firewall":
        blocked.add(r["user"])
    elif r["user"] in blocked:
        users.add(r["user"])
victims = {r["user"] for r in recs if any(i.startswith("planted/") for i in r["retrieved_ids"])}
notes = dict(l.strip().split("=", 1) for l in open("lab/injection-notes.md") if "=" in l)
print("block_then_allow_users", len(users), "| notes say", notes.get("block_then_allow_users"))
print("indirect_victims", len(victims), "| notes say", notes.get("indirect_victims"))
ok = notes.get("block_then_allow_users") == str(len(users)) and notes.get("indirect_victims") == str(len(victims))
print("PASS" if ok else "FAIL")
EOF
grep -c '^- ' lab/injection-notes.md

The first script reports 200 records and 25 tagged, all 25 triaged, none for each of the three problem lines, and PASS. The second prints PASS, and the notes have at least one - line. A clean record in your CSV is usually carol's legitimate salary answer or one of the keyword decoys; a missing one is usually a block:firewall record you discarded because it did no harm.

Notes

Every indirect record here came from alice or bob asking an ordinary question, and the only trace of the attack is a chunk id and the chunk text inside the logged prompt. An application that logs only what the user typed and what the model said back shows nine innocent questions, three of them with a leak and no cause. That is the lesson's forensic point, and the reason lesson 2's gateway logs the assembled prompt. The tags make the triage easy on purpose: in a real log the same judgement is made from the patterns the lesson lists, and the decision about which control to recommend is the same.

This is an independent study companion for CompTIA SecAI+ CY0-001 and is not produced by or endorsed by CompTIA.