Lab Specification — Module S04: CTF Harness Engineering

fixtures/
  web-1/      description.txt + url.txt          (Web)
  web-2/      description.txt + url.txt          (Web)
  pwn-1/      description.txt + vuln1 (ELF)      (Pwn)
  pwn-2/      description.txt + vuln2 (ELF)      (Pwn)
  forensic-1/ description.txt + capture.pcap     (Forensics)

from pydantic import BaseModel, Field
from typing import Literal
import subprocess, re

Domain = Literal["web", "pwn", "crypto", "forensic", "reversing", "misc", "uncertain"]

class RouterInput(BaseModel):
    description: str
    artifact_path: str | None = None
    artifact_kind: str | None = None  # "elf" | "pcap" | "image" | "url" | None

class RouterOutput(BaseModel):
    domain: Domain
    confidence: float
    signals: list[str]   # which signals drove the decision
    fallback_needed: bool

async def meta_agent_route(input: RouterInput) -> RouterOutput:
    """
    Lightweight classifier. Uses BOTH text and file-type signals.
    Returns 'uncertain' rather than guessing when signals conflict.
    """
    signals = []
    text_signals = classify_text(input.description)
    signals.extend(text_signals)

    if input.artifact_path:
        file_signal = probe_artifact(input.artifact_path)
        signals.append(file_signal)

    # Combine signals; if text and file disagree, mark uncertain
    domain, confidence = reconcile_signals(signals)
    return RouterOutput(
        domain=domain,
        confidence=confidence,
        signals=[s.description for s in signals],
        fallback_needed=(domain == "uncertain"),
    )

def probe_artifact(path: str) -> Signal:
    """Run cheap probes: file(1), checksec for ELF, header bytes for pcap."""
    raw = subprocess.run(["file", path], capture_output=True, text=True).stdout
    if "ELF" in raw:
        # For ELF, checksec tells us if it's likely a Pwn target
        sec = subprocess.run(["checksec", path], capture_output=True, text=True).stdout
        return Signal(kind="elf", value=raw + sec,
                      description=f"file: ELF, checksec: {sec.strip()}")
    if "pcap" in raw or "capture file" in raw:
        return Signal(kind="pcap", value=raw, description="file: pcap capture")
    # ... image, archive, etc.
    return Signal(kind="unknown", value=raw, description=f"file: {raw.strip()}")

async def triage_and_reroute(input: RouterInput) -> RouterOutput:
    """
    Runs when meta-agent returns 'uncertain'. Gathers disambiguating signal
    via cheap probes, then re-routes.
    """
    extra_signals = []
    if input.artifact_path:
        # For ELF: run strings frequency, look for Pwn-style symbols
        # For pcap: run tshark protocol summary
        extra_signals.append(deep_probe(input.artifact_path))
    # Re-classify with extra signal
    return await meta_agent_route(RouterInput(
        description=input.description + "\n" + " ".join(s.description for s in extra_signals),
        artifact_path=input.artifact_path,
    ))

import time
from enum import Enum

class HitlMode(Enum):
    ENABLED = "enabled"     # every tool call waits for human approval
    DISABLED = "disabled"   # every tool call executes immediately

async def run_solve(tool_calls: list, mode: HitlMode, hitl_callback=None) -> SolveReport:
    started = time.monotonic()
    timings = []
    for call in tool_calls:
        t0 = time.monotonic()
        if mode == HitlMode.ENABLED and hitl_callback:
            await hitl_callback(call)  # simulates a human approval round-trip
        result = await call.execute()
        t1 = time.monotonic()
        timings.append({"call": call.name, "duration_s": t1 - t0})
    return SolveReport(total_s=time.monotonic() - started, per_call=timings, mode=mode)

git clone https://github.com/CTFd/CTFd.git
cd CTFd
docker compose up -d
# Navigate to http://localhost:8000, complete setup, create a team,
# create one challenge with flag flag{extracted_and_submitted},
# note the challenge ID and your API token.

import re
from typing import Callable

DEFAULT_PATTERNS = [
    re.compile(r"flag\{[a-zA-Z0-9_\-]{8,128}\}", re.IGNORECASE),
    re.compile(r"CTF\{[a-zA-Z0-9_\-]{8,128}\}"),
    # Load competition-specific prefix from rules.json if present
]

class FlagExtractor:
    """Middleware: scans EVERY tool output for flag patterns."""
    def __init__(self, patterns=DEFAULT_PATTERNS, verifier: Callable | None = None):
        self.patterns = patterns
        self.verifier = verifier
        self.captured: list[str] = []

    def scan(self, tool_output: str, challenge_id: str) -> list[FlagCandidate]:
        found = []
        for p in self.patterns:
            for match in p.findall(tool_output):
                found.append(FlagCandidate(flag=match, source_challenge=challenge_id))
        # Dedup
        seen, unique = set(), []
        for c in found:
            if c.flag not in seen:
                seen.add(c.flag)
                unique.append(c)
        return unique

async def verify_flag(candidate: FlagCandidate, challenge_description: str) -> bool:
    """
    A flag-shaped string is not necessarily the answer to THIS challenge.
    Examples, decoys, and challenge-description text all match the pattern.
    The verifier rejects obvious decoys; a small LLM call confirms plausibility.
    """
    # Cheap heuristic first: reject flags found in the challenge description itself
    if candidate.flag in challenge_description:
        return False
    # Reject known decoy prefixes
    if candidate.flag.lower().startswith("flag{example"):
        return False
    # Optionally: small LLM call to confirm plausibility
    return await llm_confirm_plausibility(candidate.flag, challenge_description)

import requests, time
from collections import defaultdict

class CTFdClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.token = token
        self.last_submit_at = 0.0
        self.wrong_attempts = defaultdict(int)
        self.locked = set()

    def submit(self, challenge_id: int, flag: str) -> SubmissionResult:
        # Rate-limit: 2-3s between submissions
        elapsed = time.monotonic() - self.last_submit_at
        if elapsed < 2.5:
            time.sleep(2.5 - elapsed)
        # Per-challenge lockout after 5 wrong attempts
        if challenge_id in self.locked:
            return SubmissionResult(status="locked", raw="per-challenge lockout")
        r = requests.post(
            f"{self.base_url}/api/v1/challenges/attempt",
            headers={"Authorization": f"Token {self.token}",
                     "Content-Type": "application/json"},
            json={"challenge_id": challenge_id, "submission": flag},
            timeout=10,
        )
        self.last_submit_at = time.monotonic()
        data = r.json()
        status = data.get("data", {}).get("status", "unknown")
        if status == "incorrect":
            self.wrong_attempts[challenge_id] += 1
            if self.wrong_attempts[challenge_id] >= 5:
                self.locked.add(challenge_id)
        return SubmissionResult(status=status, raw=data)

# Lab Specification — Module S04: CTF Harness Engineering

**Course**: 2A — Building AI Harnesses for Cybersecurity
**Module**: S04 — CTF Harness Engineering
**Duration**: 90–120 minutes (three labs, one per sub-section)
**Environment**: Python 3.11+, Docker, a local CTFd instance (Docker), pwntools, binwalk, file, strings, checksec (pwntools provides `checksec`), requests. No live external targets — all challenges are local Docker containers or local files.

---

## Learning objectives

1. Build a domain router that reads a challenge description plus file artifacts and dispatches to the correct specialized agent, with a TriageAgent fallback for uncertain cases.
2. Measure the wall-clock cost of HITL approval gates versus autonomous execution on the same challenge, and document the safety decision per the three-criterion test.
3. Build an automated flag extractor (regex sweep middleware) and a CTFd submission client, run end-to-end against a local CTFd instance.

---

## Phase 1 — The Domain Router (35 min)

### 1.1 Set up the challenge fixture

Create five challenge fixtures spanning three domains. Each fixture is a directory with `description.txt` and one or more artifacts:

```
fixtures/
  web-1/      description.txt + url.txt          (Web)
  web-2/      description.txt + url.txt          (Web)
  pwn-1/      description.txt + vuln1 (ELF)      (Pwn)
  pwn-2/      description.txt + vuln2 (ELF)      (Pwn)
  forensic-1/ description.txt + capture.pcap     (Forensics)
```

Descriptions should be deliberately ambiguous in at least one case (e.g., a Pwn challenge described as "analyze this binary") to force the router to use file-type signals.

### 1.2 Implement the meta-agent classifier

```python
from pydantic import BaseModel, Field
from typing import Literal
import subprocess, re

Domain = Literal["web", "pwn", "crypto", "forensic", "reversing", "misc", "uncertain"]

class RouterInput(BaseModel):
    description: str
    artifact_path: str | None = None
    artifact_kind: str | None = None  # "elf" | "pcap" | "image" | "url" | None

class RouterOutput(BaseModel):
    domain: Domain
    confidence: float
    signals: list[str]   # which signals drove the decision
    fallback_needed: bool

async def meta_agent_route(input: RouterInput) -> RouterOutput:
    """
    Lightweight classifier. Uses BOTH text and file-type signals.
    Returns 'uncertain' rather than guessing when signals conflict.
    """
    signals = []
    text_signals = classify_text(input.description)
    signals.extend(text_signals)

    if input.artifact_path:
        file_signal = probe_artifact(input.artifact_path)
        signals.append(file_signal)

    # Combine signals; if text and file disagree, mark uncertain
    domain, confidence = reconcile_signals(signals)
    return RouterOutput(
        domain=domain,
        confidence=confidence,
        signals=[s.description for s in signals],
        fallback_needed=(domain == "uncertain"),
    )

def probe_artifact(path: str) -> Signal:
    """Run cheap probes: file(1), checksec for ELF, header bytes for pcap."""
    raw = subprocess.run(["file", path], capture_output=True, text=True).stdout
    if "ELF" in raw:
        # For ELF, checksec tells us if it's likely a Pwn target
        sec = subprocess.run(["checksec", path], capture_output=True, text=True).stdout
        return Signal(kind="elf", value=raw + sec,
                      description=f"file: ELF, checksec: {sec.strip()}")
    if "pcap" in raw or "capture file" in raw:
        return Signal(kind="pcap", value=raw, description="file: pcap capture")
    # ... image, archive, etc.
    return Signal(kind="unknown", value=raw, description=f"file: {raw.strip()}")
```

### 1.3 Implement the TriageAgent fallback

```python
async def triage_and_reroute(input: RouterInput) -> RouterOutput:
    """
    Runs when meta-agent returns 'uncertain'. Gathers disambiguating signal
    via cheap probes, then re-routes.
    """
    extra_signals = []
    if input.artifact_path:
        # For ELF: run strings frequency, look for Pwn-style symbols
        # For pcap: run tshark protocol summary
        extra_signals.append(deep_probe(input.artifact_path))
    # Re-classify with extra signal
    return await meta_agent_route(RouterInput(
        description=input.description + "\n" + " ".join(s.description for s in extra_signals),
        artifact_path=input.artifact_path,
    ))
```

### 1.4 Test the router

For each of the 5 fixtures:

1. Run the meta-agent. Record the predicted domain and confidence.
2. If uncertain, run the TriageAgent fallback. Record the re-routed domain.
3. Compare to ground truth (the directory name's prefix).
4. Compute router accuracy: correct on first dispatch / total.

### Deliverable
- [ ] Router implementation with text + file-type signals
- [ ] TriageAgent fallback for uncertain cases
- [ ] Accuracy report across 5 fixtures (target: 4/5 first-dispatch correct, 5/5 after triage)
- [ ] Documented case where file-type signal overrode a misleading text description

---

## Phase 2 — HITL vs No-HITL Timing (25 min)

### 2.1 Set up the challenge

Use the `pwn-1` fixture (a simple stack-overflow binary you own, in a Docker container). The challenge solves in roughly 4 tool calls: `checksec`, `file`, `ROPgadget` (or offset find), and `send exploit payload`.

### 2.2 Implement two execution modes

```python
import time
from enum import Enum

class HitlMode(Enum):
    ENABLED = "enabled"     # every tool call waits for human approval
    DISABLED = "disabled"   # every tool call executes immediately

async def run_solve(tool_calls: list, mode: HitlMode, hitl_callback=None) -> SolveReport:
    started = time.monotonic()
    timings = []
    for call in tool_calls:
        t0 = time.monotonic()
        if mode == HitlMode.ENABLED and hitl_callback:
            await hitl_callback(call)  # simulates a human approval round-trip
        result = await call.execute()
        t1 = time.monotonic()
        timings.append({"call": call.name, "duration_s": t1 - t0})
    return SolveReport(total_s=time.monotonic() - started, per_call=timings, mode=mode)
```

The `hitl_callback` simulates a human approval delay (sample a realistic distribution: mean 27s, stddev 5s — measured from real HITL sessions).

### 2.3 Run the comparison

Run the same challenge three times in each mode and record wall-clock time.

| Run | HITL enabled (s) | HITL disabled (s) |
| --- | --- | --- |
| 1 | ... | ... |
| 2 | ... | ... |
| 3 | ... | ... |
| **mean** | ... | ... |

### 2.4 Apply the safety decision

Document, for this specific challenge:

- (a) Is the target owned or sandboxed? (Yes — Docker container you built.)
- (b) Is it isolated from production at the network layer? (Verify: no host network, no shared volumes with sensitive data.)
- (c) What is the worst-case action? (Container restart, lost time.)

State the decision: HITL on or off, with which gate retained (the credential-reuse gate is the one to discuss).

### Deliverable
- [ ] Timing comparison table (3 runs each mode)
- [ ] Mean wall-clock difference and the percentage of time spent waiting for approval
- [ ] Three-criterion safety analysis with explicit go/no-go decision per criterion
- [ ] Statement of which gate is retained even in autonomous mode and why

---

## Phase 3 — Flag Extractor and CTFd Submission (40 min)

### 3.1 Stand up a local CTFd

```bash
git clone https://github.com/CTFd/CTFd.git
cd CTFd
docker compose up -d
# Navigate to http://localhost:8000, complete setup, create a team,
# create one challenge with flag flag{extracted_and_submitted},
# note the challenge ID and your API token.
```

### 3.2 Implement the flag extractor middleware

```python
import re
from typing import Callable

DEFAULT_PATTERNS = [
    re.compile(r"flag\{[a-zA-Z0-9_\-]{8,128}\}", re.IGNORECASE),
    re.compile(r"CTF\{[a-zA-Z0-9_\-]{8,128}\}"),
    # Load competition-specific prefix from rules.json if present
]

class FlagExtractor:
    """Middleware: scans EVERY tool output for flag patterns."""
    def __init__(self, patterns=DEFAULT_PATTERNS, verifier: Callable | None = None):
        self.patterns = patterns
        self.verifier = verifier
        self.captured: list[str] = []

    def scan(self, tool_output: str, challenge_id: str) -> list[FlagCandidate]:
        found = []
        for p in self.patterns:
            for match in p.findall(tool_output):
                found.append(FlagCandidate(flag=match, source_challenge=challenge_id))
        # Dedup
        seen, unique = set(), []
        for c in found:
            if c.flag not in seen:
                seen.add(c.flag)
                unique.append(c)
        return unique
```

### 3.3 Implement the verifier

```python
async def verify_flag(candidate: FlagCandidate, challenge_description: str) -> bool:
    """
    A flag-shaped string is not necessarily the answer to THIS challenge.
    Examples, decoys, and challenge-description text all match the pattern.
    The verifier rejects obvious decoys; a small LLM call confirms plausibility.
    """
    # Cheap heuristic first: reject flags found in the challenge description itself
    if candidate.flag in challenge_description:
        return False
    # Reject known decoy prefixes
    if candidate.flag.lower().startswith("flag{example"):
        return False
    # Optionally: small LLM call to confirm plausibility
    return await llm_confirm_plausibility(candidate.flag, challenge_description)
```

### 3.4 Implement the CTFd submission client

```python
import requests, time
from collections import defaultdict

class CTFdClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.token = token
        self.last_submit_at = 0.0
        self.wrong_attempts = defaultdict(int)
        self.locked = set()

    def submit(self, challenge_id: int, flag: str) -> SubmissionResult:
        # Rate-limit: 2-3s between submissions
        elapsed = time.monotonic() - self.last_submit_at
        if elapsed < 2.5:
            time.sleep(2.5 - elapsed)
        # Per-challenge lockout after 5 wrong attempts
        if challenge_id in self.locked:
            return SubmissionResult(status="locked", raw="per-challenge lockout")
        r = requests.post(
            f"{self.base_url}/api/v1/challenges/attempt",
            headers={"Authorization": f"Token {self.token}",
                     "Content-Type": "application/json"},
            json={"challenge_id": challenge_id, "submission": flag},
            timeout=10,
        )
        self.last_submit_at = time.monotonic()
        data = r.json()
        status = data.get("data", {}).get("status", "unknown")
        if status == "incorrect":
            self.wrong_attempts[challenge_id] += 1
            if self.wrong_attempts[challenge_id] >= 5:
                self.locked.add(challenge_id)
        return SubmissionResult(status=status, raw=data)
```

### 3.5 Wire it together and run end-to-end

1. Simulate four tool outputs — three that contain no flag, one that contains `flag{extracted_and_submitted}`.
2. Run each through the FlagExtractor middleware.
3. Run any candidates through the verifier.
4. Submit verified flags via the CTFdClient.
5. Inject a decoy (`flag{example_decoy}`) into one of the outputs. Verify the verifier drops it.
6. Inject the actual flag string into the challenge description. Verify the verifier drops it (it appears in description text).
7. Confirm CTFd returns `correct` for the legitimate flag and `incorrect` for an invented wrong flag.

### Deliverable
- [ ] FlagExtractor middleware with configurable patterns
- [ ] Verifier that drops decoys (description text, example prefixes)
- [ ] CTFdClient with rate-limiting (2.5s) and per-challenge lockout (5 wrong)
- [ ] End-to-end run: simulated tool outputs → extraction → verification → CTFd `correct`
- [ ] Demonstration that the decoy flag is dropped and never submitted
- [ ] Demonstration of all three CTFd response states (correct / incorrect / already_solved)

---

## Stretch goals

1. **Parallel solve with intra-challenge hypotheses.** Take the `pwn-1` fixture and spawn two solve sessions in parallel — one pursuing a stack-overflow hypothesis, one pursuing a format-string hypothesis. First session to produce a verified flag wins; kill the other. Measure wall-clock versus single-session.
2. **Full scorecard.** Implement the `CTFScorecard` interface from the teaching document. Run the harness against 3 local CTFd challenges concurrently and produce a real-time scorecard showing per-challenge state, sessions, attempts, and first-blood timestamps.
3. **Router accuracy benchmark.** Expand the fixture set to 15 challenges across all 6 domains. Measure first-dispatch router accuracy and post-TriageAgent accuracy. Identify which domains the router misroutes most often and add a domain-specific signal to fix the worst case.