# Lab Specification — Module S03: Pentest and Red Team Harnesses **Course**: 2A — Building AI Harnesses for Cybersecurity **Module**: S03 — Pentest and Red Team Harnesses **Duration**: 150–180 minutes (substantial — four labs, one per sub-section) **Environment**: Python 3.11+, Docker, Docker Compose. All targets are isolated Docker containers on a user-defined network. No live targets. Assumes S02 lab complete (engagement memory, evidence logger, wrapped tools are reused). --- ## Learning objectives 1. Implement an attack graph as the harness's primary state structure, with MITRE ATT&CK node labels and the five-state lifecycle. Run a 10-step engagement and visualize the graph after each step. 2. Simulate an exploit failure mid-chain and verify the RedTeamLLM plan correction loop selects and executes an alternative path. 3. Configure a Docker-based offensive harness with dual containment (outbound scope enforcement + inbound isolation) and verify both directions independently. 4. Build a report generator that turns structured findings into a client-ready HTML report with CVSS drafts (flagged), evidence citations, and CWE/OWASP-mapped remediation. --- ## Shared setup — target environment All four labs use the same isolated Docker target network. Create it once: ```bash # Create an isolated network for lab targets only docker network create --internal s03-target-net # Target 1: deliberately vulnerable web app (Metasploitable-style) cat > docker-compose.yml <<'EOF' version: "3.8" services: vuln-web: image: vulnerables/web-dvwa:latest networks: [target-net] ports: ["8081:80"] vuln-ssh: image: linuxserver/openssh-server:latest environment: - PASSWORD_ACCESS=true - USER_PASSWORD=labpassword - USER_NAME=labuser networks: [target-net] ports: ["2222:2222"] networks: target-net: external: true name: s03-target-net EOF docker compose up -d ``` The `--internal` flag means the target network has no default route to the host or the internet — targets can talk to each other and to containers you attach, but cannot reach out. This is the inbound isolation boundary you will build the agent container against. --- ## Phase 1 — Attack Graph as Primary State (40 min) Implement the attack graph from S03.1. This is the structure every subsequent lab builds on. ### 1.1 The AttackNode and AttackGraph schemas ```python from pydantic import BaseModel, Field from datetime import datetime from enum import Enum import json, uuid class NodeState(str, Enum): UNEXPLORED = "unexplored" ATTEMPTED = "attempted" SUCCEEDED = "succeeded" FAILED = "failed" OUT_OF_SCOPE = "out_of_scope" class AttackNode(BaseModel): id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8]) tactic: str # e.g. "TA0001 Initial Access" technique: str # e.g. "T1190 Exploit Public-Facing Application" target: str state: NodeState = NodeState.UNEXPLORED prerequisites: list[str] = Field(default_factory=list) # node IDs evidence_refs: list[str] = Field(default_factory=list) attempts: list[dict] = Field(default_factory=list) last_updated: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z") class AttackGraph(BaseModel): engagement_id: str nodes: dict[str, AttackNode] = Field(default_factory=dict) current_position: list[str] = Field(default_factory=list) scope_ref: str def add_node(self, **kwargs) -> AttackNode: node = AttackNode(**kwargs) self.nodes[node.id] = node return node def mark_failed(self, node_id: str): self.nodes[node_id].state = NodeState.FAILED self._persist() def mark_succeeded(self, node_id: str, evidence_id: str): n = self.nodes[node_id] n.state = NodeState.SUCCEEDED n.evidence_refs.append(evidence_id) if node_id not in self.current_position: self.current_position.append(node_id) # unlock downstream nodes whose prereqs are now met self._persist() def met_prerequisites_for(self, node: AttackNode) -> bool: return all( self.nodes[pid].state == NodeState.SUCCEEDED for pid in node.prerequisites if pid in self.nodes ) def available_techniques(self) -> list[AttackNode]: """Nodes the agent can attempt right now: unexplored, prereqs met, in scope.""" return [ n for n in self.nodes.values() if n.state == NodeState.UNEXPLORED and self.met_prerequisites_for(n) ] def _persist(self, path: str = "attack-graph.json"): with open(path, "w") as f: f.write(self.model_dump_json(indent=2)) ``` ### 1.2 Seed a 10-step engagement Seed the graph with a 10-objective kill chain spanning the ATT&CK tactics. This is a scripted engagement against the DVWA container — you are testing the graph mechanics, not the exploits (which are simulated as success/fail for this lab). ```python graph = AttackGraph(engagement_id="eng-001", scope_ref="scope.json") # TA0001 Initial Access n1 = graph.add_node(tactic="TA0001 Initial Access", technique="T1190 Exploit Public-Facing Application", target="vuln-web:80") # TA0002 Execution n2 = graph.add_node(tactic="TA0002 Execution", technique="T1059 Command and Scripting Interpreter", target="vuln-web:80", prerequisites=[n1.id]) # TA0003 Persistence n3 = graph.add_node(tactic="TA0003 Persistence", technique="T1098 Account Manipulation", target="vuln-web:80", prerequisites=[n2.id]) # TA0004 Privilege Escalation n4 = graph.add_node(tactic="TA0004 Privilege Escalation", technique="T1068 Exploitation for Privilege Escalation", target="vuln-web:80", prerequisites=[n2.id]) # TA0005 Defense Evasion n5 = graph.add_node(tactic="TA0005 Defense Evasion", technique="T1070 Indicator Removal", target="vuln-web:80", prerequisites=[n4.id]) # TA0006 Credential Access n6 = graph.add_node(tactic="TA0006 Credential Access", technique="T1552 Unsecured Credentials", target="vuln-web:80", prerequisites=[n4.id]) # TA0007 Discovery n7 = graph.add_node(tactic="TA0007 Discovery", technique="T1046 Network Service Discovery", target="vuln-web:80", prerequisites=[n3.id]) # TA0008 Lateral Movement n8 = graph.add_node(tactic="TA0008 Lateral Movement", technique="T1021 Remote Services", target="vuln-ssh:22", prerequisites=[n6.id, n7.id]) # TA0009 Collection n9 = graph.add_node(tactic="TA0009 Collection", technique="T1005 Data from Local System", target="vuln-ssh:22", prerequisites=[n8.id]) # TA0010 Exfiltration n10 = graph.add_node(tactic="TA0010 Exfiltration", technique="T1041 Exfiltration Over C2 Channel", target="vuln-ssh:22", prerequisites=[n9.id]) ``` ### 1.3 Run the engagement and visualize Implement a simple executor that steps through `available_techniques()`, simulates success, marks nodes, and visualizes the graph after each step. ```python def visualize(graph: AttackGraph, step: int): """ASCII visualization — for production use Graphviz.""" print(f"\n=== Step {step} === current_position: {graph.current_position}") for nid, node in graph.nodes.items(): prereqs_met = graph.met_prerequisites_for(node) marker = { NodeState.UNEXPLORED: ("[*]" if prereqs_met else "[ ]"), NodeState.ATTEMPTED: "[~]", NodeState.SUCCEEDED: "[+]", NodeState.FAILED: "[x]", NodeState.OUT_OF_SCOPE: "[-]", }[node.state] print(f" {marker} {node.technique} ({node.target})") # Simulate: step through, mark each available node succeeded for step in range(1, 11): available = graph.available_techniques() if not available: print("No available techniques — engagement complete or blocked.") break node = available[0] # (in a real harness, execute the wrapped tool here) graph.mark_succeeded(node.id, evidence_id=f"ev-{step}") visualize(graph, step) ``` ### 1.4 Verify cross-session continuity 1. After step 5, serialize the graph (automatic via `_persist`). 2. Restart your Python session. Load `attack-graph.json`. 3. Confirm `current_position` and node states are intact. 4. Issue the orientation prompt: print available techniques and resume from step 6. ### Deliverable - [ ] `AttackGraph` with five-state lifecycle and prerequisite checking - [ ] 10-node engagement seeded across all ATT&CK tactics - [ ] Visualization after each step showing state transitions - [ ] Cross-session resume verified (graph loaded from disk, engagement continued) --- ## Phase 2 — Plan Correction on Simulated Failure (35 min) Using the attack graph from Phase 1, simulate a hard failure at step 3 (Persistence) and verify the plan correction loop from S03.2 selects an alternative. ### 2.1 Build the knowledge base ```python class KnowledgeBase: """Maps tactics to candidate techniques, ranked by target context.""" def __init__(self): self.techniques = { "TA0003 Persistence": [ {"id": "T1098", "name": "Account Manipulation", "context_score": 0.8}, {"id": "T1543", "name": "Service Execution", "context_score": 0.6}, {"id": "T1136", "name": "Create Account", "context_score": 0.5}, ], "TA0001 Initial Access": [ {"id": "T1190", "name": "Exploit Public-Facing Application", "context_score": 0.9}, {"id": "T1078", "name": "Valid Accounts", "context_score": 0.7}, {"id": "T1110", "name": "Brute Force", "context_score": 0.4}, ], } def find_techniques(self, tactic: str, prerequisites_met: bool, exclude_failed: list[str] = None) -> list[dict]: exclude_failed = exclude_failed or [] candidates = [ t for t in self.techniques.get(tactic, []) if t["id"] not in exclude_failed ] return sorted(candidates, key=lambda t: -t["context_score"]) ``` ### 2.2 The plan correction loop ```python async def plan_correction(failed_node: AttackNode, graph: AttackGraph, kb: KnowledgeBase) -> AttackNode | None: """RedTeamLLM four-step loop. Returns the new node to attempt, or None.""" # 1. DETECT — caller has confirmed this is a hard failure graph.mark_failed(failed_node.id) # collect all failed technique IDs for this tactic (exclude_failed=True) failed_techniques = [ n.technique.split()[0] for n in graph.nodes.values() if n.state == NodeState.FAILED and n.tactic == failed_node.tactic ] # 2. RE-HYPOTHESIZE alternatives = kb.find_techniques( tactic=failed_node.tactic, prerequisites_met=graph.met_prerequisites_for(failed_node), exclude_failed=failed_techniques, ) if not alternatives: print(f"No viable path for {failed_node.tactic} — escalate.") return None # 3. RE-PLAN — add a NEW unexplored node (never retry the old one) next_tech = alternatives[0] new_node = graph.add_node( tactic=failed_node.tactic, technique=f"{next_tech['id']} {next_tech['name']}", target=failed_node.target, prerequisites=failed_node.prerequisites, # same prereqs state=NodeState.UNEXPLORED, ) print(f"Plan correction: {failed_node.technique} FAILED → trying {new_node.technique}") return new_node # 4. RETRY — caller executes this new node ``` ### 2.3 Simulate the failure 1. Run steps 1 and 2 (Initial Access, Execution) — succeed. 2. At step 3 (T1098 Account Manipulation), inject a simulated hard failure: ```python # Step 3: simulate hard failure persistence_node = [n for n in graph.available_techniques() if n.tactic == "TA0003 Persistence"][0] print(f"Simulating HARD FAILURE on {persistence_node.technique}") kb = KnowledgeBase() new_node = await plan_correction(persistence_node, graph, kb) assert new_node is not None, "Plan correction should find an alternative" assert new_node.technique != persistence_node.technique, "Must be a DIFFERENT technique" assert new_node.state == NodeState.UNEXPLORED, "New node starts unexplored" assert persistence_node.state == NodeState.FAILED, "Old node marked failed" # Execute the new node, succeed, continue the engagement graph.mark_succeeded(new_node.id, evidence_id="ev-3b") ``` 3. Verify the engagement continues to step 10 using the alternative path. ### 2.4 Verify the substitution invariant Confirm that: - The failed technique (`T1098`) does NOT reappear as a candidate in subsequent corrections for the same tactic. - If all alternatives exhaust, `plan_correction` returns `None` and the engagement reports partial objectives. ### Deliverable - [ ] Knowledge base with ranked alternative techniques per tactic - [ ] Plan correction loop implementing all four steps (detect, re-hypothesize, re-plan, retry) - [ ] Simulated hard failure at step 3 correctly triggers alternative selection - [ ] `exclude_failed=True` verified — failed techniques never re-candidate - [ ] Exhaustion case handled (returns None, engagement reports partial) --- ## Phase 3 — Dual Containment in Docker (40 min) The load-bearing lab. Configure the agent container with dual containment and verify both boundaries independently. ### 3.1 The scope file ```json // scope.json { "engagement_id": "eng-001", "valid_until": "2026-07-15T23:59:59Z", "in_scope": { "targets": ["vuln-web", "vuln-ssh"], "networks": ["s03-target-net"], "ports": ["80/tcp", "22/tcp"] }, "out_of_scope": { "explicit_deny": ["192.168.0.0/16", "10.0.0.0/8", "169.254.169.254", "172.16.0.0/12"] }, "roe": { "max_requests_per_second": 5 } } ``` ### 3.2 The agent container with dual containment ```dockerfile # Dockerfile.agent FROM python:3.11-slim RUN apt-get update && apt-get install -y --no-install-recommends \ nmap curl iproute2 iptables squid && rm -rf /var/lib/apt/lists/* WORKDIR /agent COPY . /agent # No host credential stores mounted — enforced at run time # The container runs on an isolated network namespace CMD ["python", "-m", "agent"] ``` The egress proxy (squid) configuration, generated from `scope.json`: ``` # squid.conf — generated from scope.json acl in_scope dstdomain vuln-web vuln-ssh acl denied_net dst 192.168.0.0/16 10.0.0.0/8 169.254.169.254 172.16.0.0/12 http_access deny denied_net http_access allow in_scope http_access deny all http_port 3128 ``` ### 3.3 The compose with isolated networks ```yaml # docker-compose.agent.yml version: "3.8" services: agent: build: . networks: - agent-egress # only path out is via the proxy volumes: - ./attack-graph.json:/agent/attack-graph.json # NOTE: no ~/.ssh, no ~/.aws, no host credential dirs mounted egress-proxy: image: ubuntu/squid:latest volumes: - ./squid.conf:/etc/squid/squid.conf:ro networks: - agent-egress - target-net # proxy can reach targets; agent cannot (except via proxy) cap_add: [NET_ADMIN] networks: agent-egress: internal: false # proxy's only path to targets target-net: external: true name: s03-target-net ``` The key: the agent container is on `agent-egress` only. It is NOT on `target-net`. The proxy bridges the two networks — the agent's only path to targets is through the proxy, which enforces the ACL. ### 3.4 Verify outbound scope enforcement From inside the agent container: ```bash # SHOULD SUCCEED — in-scope target via proxy curl -x http://egress-proxy:3128 http://vuln-web/ # expect DVWA page # SHOULD BE REFUSED — out-of-scope canary # Start a canary on the host: python3 -m http.server 9999 curl -x http://egress-proxy:3128 http://host.docker.internal:9999/ # expect TCP_RESET / 403 # SHOULD BE REFUSED — metadata endpoint (simulate) curl -x http://egress-proxy:3128 http://169.254.169.254/latest/meta-data/ # expect denied ``` Record the results. The in-scope connection succeeds; the out-of-scope canary and metadata endpoint are refused. ### 3.5 Verify inbound isolation From the target network, attempt to reach the agent or host: ```bash # Exec into the vuln-web container (which is on target-net) docker exec vuln-web bash -c "apt-get update && apt-get install -y iproute2 && ip route" # Confirm there is no route to the host LAN or the agent-egress network # Attempt to reach the agent container — should fail docker exec vuln-web bash -c "curl -m 3 http://agent:8000/ || echo 'INBOUND BLOCKED (correct)'" # Attempt to reach the host's metadata or LAN — should fail docker exec vuln-web bash -c "curl -m 3 http://192.168.1.1/ || echo 'LATERAL MOVEMENT BLOCKED (correct)'" ``` ### Deliverable - [ ] Agent container on isolated network; only path out via egress proxy - [ ] Squid ACL generated from scope.json; host credential stores not mounted - [ ] Outbound verified: in-scope succeeds, out-of-scope canary + metadata refused - [ ] Inbound verified: target cannot reach agent or host LAN - [ ] Both directions verified INDEPENDENTLY (active tests, not config review) --- ## Phase 4 — Report Generator (35 min) Build a report generator that takes structured findings and produces a client-ready HTML report with CVSS drafts, evidence citations, and remediation. ### 4.1 The finding and remediation schemas ```python from pydantic import BaseModel from typing import Literal class RemediationRecord(BaseModel): short_term: str long_term: str owasp_control: str # mapped via lookup, NOT model-generated effort: Literal["low", "medium", "high"] class Finding(BaseModel): id: str title: str severity: Literal["critical", "high", "medium", "low", "informational"] cvss_draft: str cvss_auto_scored: bool = True # ALWAYS true from the harness description: str affected_assets: list[str] evidence_refs: list[str] recreation_steps: list[str] remediation: RemediationRecord cwe: str # e.g. "CWE-79" discovered_via_node: str ``` ### 4.2 Deterministic CWE-to-OWASP lookup ```python # The control mapping is a LOOKUP TABLE — it cannot hallucinate. CWE_TO_OWASP = { "CWE-79": {"owasp": "A03:2021-Injection", "control": "ASVS 5.3.4 (output encoding)"}, "CWE-89": {"owasp": "A03:2021-Injection", "control": "ASVS 5.3.1 (parameterized queries)"}, "CWE-287": {"owasp": "A07:2021-Identification & Auth Failures", "control": "ASVS 2.1.1 (MFA)"}, "CWE-269": {"owasp": "A05:2021-Security Misconfiguration", "control": "ASVS 4.1.1 (least privilege)"}, "CWE-522": {"owasp": "A02:2021-Cryptographic Failures", "control": "ASVS 6.2.2 (password hashing)"}, } def map_remediation(cwe: str, short_term: str, long_term: str, effort: str) -> RemediationRecord: mapping = CWE_TO_OWASP.get(cwe, {"owasp": "UNCATEGORIZED", "control": "manual review required"}) return RemediationRecord( short_term=short_term, long_term=long_term, owasp_control=f"{mapping['owasp']} — {mapping['control']}", effort=effort, ) ``` ### 4.3 The report renderer (HTML with CVSS flagged as draft) ```python from jinja2 import Template REPORT_TEMPLATE = """\ Engagement {{ engagement_id }} — Report

Red Team Engagement Report

Engagement: {{ engagement_id }} · Generated: {{ generated_at }}

Executive Summary

{{ executive_summary }}

Findings ({{ findings|length }})

{% for f in findings %}

{{ f.title }} CVSS {{ f.cvss_draft }} — DRAFT, pending analyst review

Severity: {{ f.severity }} · CWE: {{ f.cwe }}

{{ f.description }}

Affected assets: {{ f.affected_assets|join(', ') }}

Reproduction:

    {% for step in f.recreation_steps %}
  1. {{ step }}
    2. {% endfor %}

Evidence: {% for ref in f.evidence_refs %} {{ ref }}{% if not loop.last %}, {% endif %} {% endfor %}

Remediation ({{ f.remediation.owasp_control }}):

{% endfor %} """ def render_report(engagement_id: str, findings: list[Finding], executive_summary: str) -> str: tpl = Template(REPORT_TEMPLATE) return tpl.render( engagement_id=engagement_id, findings=[f.model_dump() for f in findings], executive_summary=executive_summary, generated_at=datetime.utcnow().isoformat() + "Z", ) ``` ### 4.4 Generate a report from the attack graph 1. Extract findings from the Phase 1 attack graph: each succeeded node with an associated vulnerability becomes a finding. 2. Auto-score CVSS (clearly flagged as draft). 3. Map remediation via the CWE lookup. 4. Write the executive summary LAST, grounded in the specific finding set (name the actual top finding, cite the attack path). ```python findings = [ Finding( id="F-001", title="Remote Code Execution via Public-Facing Web Application", severity="critical", cvss_draft="9.8", # DRAFT — flagged in the template description="The DVWA instance permits command injection via the ping parameter...", affected_assets=["vuln-web:80"], evidence_refs=["ev-1", "ev-2"], recreation_steps=["Navigate to /vulnerabilities/exec/", "Submit ; id in the ip field"], remediation=map_remediation("CWE-79", "Disable the exec module", "Implement parameterized command execution", "medium"), cwe="CWE-79", discovered_via_node="n1", ), # ... one finding per succeeded node with a vulnerability ] executive_summary = ( "The engagement demonstrated full kill chain execution against the target estate, " "achieving Domain Admin equivalent access within the lab environment. The critical " "finding (CWE-79, CVSS 9.8 draft) allowed remote code execution on the public-facing " "web application, chaining through 10 ATT&CK techniques to exfiltration. Top priority: " "remediate the command injection (F-001) and enforce output encoding per OWASP ASVS 5.3.4." ) html = render_report("eng-001", findings, executive_summary) with open("report.html", "w") as f: f.write(html) print("Report written to report.html — open in a browser to verify.") ``` ### 4.5 Verify the report 1. Open `report.html` in a browser. 2. Confirm every CVSS score carries the visible "DRAFT — pending analyst review" marker. 3. Confirm every remediation cites a valid OWASP control from the lookup table (no hallucinated references). 4. Confirm evidence references are present and correspond to real evidence record IDs. 5. Confirm the executive summary names a specific finding — it must NOT be generic boilerplate. ### Deliverable - [ ] Finding and RemediationRecord schemas with CVSS auto-flagged - [ ] Deterministic CWE-to-OWASP lookup table (cannot hallucinate) - [ ] HTML report with all CVSS scores visibly flagged as draft - [ ] Evidence references present and clickable - [ ] Executive summary grounded in the specific finding set (not generic) --- ## Stretch goals 1. **Cross-format rendering**: render the same findings to PDF (via headless browser / Playwright) and JSON export from the same source data. Verify all three are consistent. 2. **Interactive graph visualization**: render the Phase 1 attack graph as an interactive Graphviz/SVG diagram embedded in the HTML report, with node states color-coded. 3. **Full engagement run**: combine all four phases — run the attack graph against the DVWA target through the dual-containment agent, trigger a real (lab-safe) exploit failure, observe plan correction, and generate the report from real findings (not seeded data). 4. **Quarantine vault**: implement the credential quarantine vault inside the agent container — an encrypted store with its own key, verified destroyed on container teardown. Confirm harvested target credentials never appear on the host filesystem.