# Lab Specification — Capstone C2: Build a Smart Contract or Cloud Security Harness on tau-security **Course**: 2A — Building AI Harnesses for Cybersecurity **Module**: C2 — Build a Smart Contract or Cloud Security Harness **Duration**: 90 minutes (the lab IS the capstone — three phases matching C2.1/C2.2/C2.3) **Environment**: Python 3.11+, `pip install tau-ai`, the `tau-security` package, an LLM API key (OpenAI or Anthropic), and domain-specific tooling. Smart contract track: Foundry, Slither, a forked-mainnet RPC or Damn Vulnerable DeFi. Cloud track: AWS CLI, Prowler, Terraform, an isolated AWS account or LocalStack. > This lab is the capstone. It covers all three sub-sections. By the end, you have a domain-specific `SecurityHarness` that runs against a real lab target, produces a client-ready report, and scores against a recognized benchmark. The point of C2 is that the SAME harness brain works for any domain: tau-security gives you the `AgentHarness` loop, scope enforcement, evidence chain, triage, and reporting. YOU supply the domain tools and the domain scope. The deliverable is the published one-page summary. --- ## Learning objectives 1. Choose a domain and produce an authorization/scope document expressed through `Scope` and `ScopeRule`. 2. Write domain-specific security tools (smart contract analyzers or cloud posture scanners) as `AgentTool` instances following the tau-security pattern. 3. Configure a `SecurityHarness` with domain tools registered via `extra_tools` and a domain scope. 4. Run the harness against a real lab target and capture structured findings with full evidence records. 5. Score against the domain benchmark and produce a one-page results summary for publication. --- ## Prerequisites — install (5 min, before Phase 1) ```bash pip install tau-ai cd course/02a-security-harnesses/tau-security pip install -e ".[dev]" ``` Confirm the harness brain loads and that the five default tools are present. tau-ai ships `OpenAICompatibleProvider` (OpenAI, OpenRouter, local servers) and `AnthropicProvider` (Claude). The examples use OpenAI; swap to `AnthropicProvider(AnthropicConfig(api_key="sk-ant-..."))` with `model="claude-3-5-sonnet"` if you prefer Claude. ```python from tau_security import SecurityHarness, SecurityHarnessConfig, Scope from tau_ai import OpenAICompatibleProvider from tau_ai.env import OpenAICompatibleConfig config = SecurityHarnessConfig( provider=OpenAICompatibleProvider(OpenAICompatibleConfig(api_key="sk-...")), model="gpt-4o", scope=Scope.for_appsec("/tmp"), ) h = SecurityHarness(config) print("default tools:", [t.name for t in h._build_tools()]) ``` In C2 you will largely IGNORE the defaults and supply your own domain tools via `extra_tools`. The defaults (port scan, HTTP probe, code scan) still run, which is fine — for smart contracts, `code_scan` is useful; for cloud, you may never call them. The architecture stays the same. --- ## Phase 1 — Design and Scoping (C2.1, 15 min) ### 1.1 Choose your domain and lab target Pick a track. You need a real lab target you can run against in 60 minutes. - **Smart contract track**: Install Foundry. Clone Damn Vulnerable DeFi (`git clone https://github.com/OpenZeppelin/damn-vulnerable-defi`) or prepare a forked-mainnet script. Select 2-3 challenges for the benchmark subset (e.g., unstoppable, side-entrance, backdoor). - **Cloud track**: Provision an isolated AWS account (or start LocalStack with `localstack start`). Use Terraform to create deliberately misconfigured resources (public S3 bucket, over-permissive IAM role, exposed security group). Record your choice in `design-doc.md`. ### 1.2 Define the authorization document and the Scope object tau-security's `Scope` is the legal boundary — `assert_in_scope` fires inside every tool executor. For C2 you build a domain-specific scope. **Smart contract** — scope to local files/contracts (no network mainnet mutation): ```python from tau_security import Scope, ScopeRule scope = Scope(rules=[ ScopeRule(pattern="/abs/path/to/damn-vulnerable-defi", type="path", rule="allow"), ScopeRule(pattern="/abs/path/to/damn-vulnerable-defi/solutions", type="path", rule="deny"), ]) ``` **Cloud** — scope to the isolated account/region endpoints. LocalStack runs on `localhost:4566`: ```python scope = Scope(rules=[ ScopeRule(pattern="localhost:4566", type="host", rule="allow"), # LocalStack endpoint ScopeRule(pattern="127.0.0.1:4566", type="host", rule="allow"), ScopeRule(pattern="0.0.0.0", type="host", rule="deny"), # never real AWS ]) ``` For a real (isolated) AWS account, use the account ID and regions in the authorization doc; scope enforcement then gates which endpoints your cloud tools may hit. Authorization document (save to `design-doc.md`): ```markdown ## Authorization and scope document Smart contract: Contracts in scope: [list challenge contract addresses] Forked block: [block number, if using forked mainnet] Authorized actions: read source, mutate in fork, deploy PoC contracts Engagement boundary: DVD challenge contracts only — no mainnet state mutation Teardown: destroy forked mainnet instance after run Cloud: Account ID: [isolated account] · Regions: [us-east-1] Authorized actions: read config, apply remediation in isolated account, red-team within blast radius Engagement boundary: no cross-account actions, no production data access Teardown: terraform destroy ``` ### 1.3 Define the domain tool suite Each domain tool follows the tau-security pattern: JSON-schema `input_schema`, async `execute(arguments, signal)`, `assert_in_scope` gate, evidence capture via the shared context, structured `AgentToolResult`. **Smart contract tools** (you implement in Phase 2): Slither wrapper, Foundry test runner, forked-mainnet PoC builder, evidence recorder (the built-in `record_finding` covers this). **Cloud tools**: Prowler/config scanner, IAM analyzer, attack-surface mapper, remediation applier, evidence recorder. Record your chosen tool set and the three modes (Detect / Patch or Remediate / Exploit) in the design doc. **Checkpoint**: `design-doc.md` has domain choice, the exact `Scope(...)` call, the authorization document, the tool suite, and the three modes. --- ## Phase 2 — Build and Run (C2.2, 60 min) ### 2.1 Write your domain tools as AgentTool instances This is the core of C2: you write domain tools that plug into the same harness brain. The pattern is identical to C1's custom tool and to tau-security's `tools.py`. Build a shared `ToolContext`, call `assert_in_scope`, capture evidence, return `_success`/`_blocked`. **Smart contract track — Slither wrapper** (`sc_tools.py`): ```python import asyncio, uuid, json from collections.abc import Mapping from tau_agent.tools import AgentTool, AgentToolResult from tau_security.tools import ToolContext, _success, _blocked, _capture_evidence from tau_security.scope import assert_in_scope from tau_security.evidence import Finding def create_slither_tool(ctx: ToolContext) -> AgentTool: """Run Slither static analysis on an in-scope contract directory.""" async def execute(arguments: Mapping[str, object], signal=None) -> AgentToolResult: del signal contract_dir = str(arguments.get("path", "")) tool_call_id = str(uuid.uuid4())[:8] # Scope gate — only in-scope paths may be analyzed try: assert_in_scope(ctx.scope, contract_dir, "slither") except Exception as e: return _blocked(tool_call_id, "slither", str(e)) proc = await asyncio.create_subprocess_exec( "slither", contract_dir, "--json", "-", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await proc.communicate() output = stdout.decode(errors="replace") if stdout else stderr.decode(errors="replace") evidence = _capture_evidence( ctx, tool_call_id, "command_output", f"slither {contract_dir}", output, ) # Parse Slither JSON and promote detectors to findings try: data = json.loads(output) for det in data.get("results", {}).get("detectors", []): finding = Finding.create( title=det.get("check", "Slither finding"), severity=_slither_impact_to_severity(det.get("impact", "low")), location=str(det.get("elements", [{}])[0].get("source_mapping", {})), description=det.get("description", ""), confidence=_slither_conf(det.get("confidence", "low")), cwe_id=det.get("cwe", ""), ) ctx.chain.add_finding(finding) ctx.chain.link_evidence(finding.id, evidence.id) if ctx.on_finding: ctx.on_finding(finding) except json.JSONDecodeError: pass # Slither returned text, not JSON — evidence still captured return _success(tool_call_id, "slither", f"Slither complete on {contract_dir}. See findings.") return AgentTool( name="slither", description=( "Run Slither static analysis on an in-scope Solidity contract or directory. " "Returns detector findings (reentrancy, unchecked calls, etc.) and records " "them with evidence. Only in-scope paths can be analyzed." ), input_schema={ "type": "object", "properties": {"path": {"type": "string", "description": "Contract file or directory"}}, "required": ["path"], }, executor=execute, ) def _slither_impact_to_severity(impact: str) -> str: return {"high": "critical", "medium": "high", "low": "medium", "informational": "info"}.get(impact, "low") def _slither_conf(conf: str) -> float: return {"high": 0.85, "medium": 0.7, "low": 0.5}.get(conf, 0.5) ``` Build the rest of your suite the same way: a `create_foundry_test_tool` (runs `forge test`, captures output as evidence, promotes failures to findings), a `create_poc_builder_tool` (deploys a PoC against the forked mainnet and records whether it succeeded). **Cloud track — Prowler wrapper** (`cloud_tools.py`): ```python def create_prowler_tool(ctx: ToolContext) -> AgentTool: """Run Prowler against an in-scope AWS/LocalStack endpoint.""" async def execute(arguments: Mapping[str, object], signal=None) -> AgentToolResult: del signal endpoint = str(arguments.get("endpoint", "localhost:4566")) tool_call_id = str(uuid.uuid4())[:8] try: assert_in_scope(ctx.scope, endpoint, "prowler") except Exception as e: return _blocked(tool_call_id, "prowler", str(e)) proc = await asyncio.create_subprocess_exec( "prowler", "aws", "--endpoint-url", f"http://{endpoint}", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await proc.communicate() output = stdout.decode(errors="replace") evidence = _capture_evidence(ctx, tool_call_id, "command_output", f"prowler aws --endpoint {endpoint}", output) # Parse Prowler JSON lines and promote failed checks to findings for line in output.splitlines(): try: chk = json.loads(line) if chk.get("Status") == "FAIL": finding = Finding.create( title=chk.get("CheckID", "Prowler finding"), severity=_prowler_severity(chk.get("Severity", "low")), location=f"{chk.get('ResourceType')}/{chk.get('ResourceARN')}", description=chk.get("Message", ""), confidence=0.7, ) ctx.chain.add_finding(finding) ctx.chain.link_evidence(finding.id, evidence.id) if ctx.on_finding: ctx.on_finding(finding) except json.JSONDecodeError: continue return _success(tool_call_id, "prowler", f"Prowler complete against {endpoint}.") return AgentTool( name="prowler", description=( "Run Prowler config scanning against an in-scope AWS endpoint (LocalStack or " "isolated account). Returns posture findings (public buckets, permissive IAM, " "exposed SGs) and records them with evidence." ), input_schema={ "type": "object", "properties": {"endpoint": {"type": "string", "description": "host:port of the AWS endpoint"}}, "required": ["endpoint"], }, executor=execute, ) def _prowler_severity(sev: str) -> str: return {"critical": "critical", "high": "high", "medium": "medium", "low": "low"}.get(sev, "low") ``` Build the rest of the cloud suite the same way: `create_iam_analyzer_tool`, `create_remediation_tool` (applies a fix and re-scans), `create_redteam_tool` (proves an exposure is exploitable). ### 2.2 Configure the SecurityHarness with domain tools The harness brain is domain-independent. You supply the domain via a custom `Scope`, a domain `system_prompt`, and your domain tools wired in by overriding `_build_tools` on a small subclass. The override is how your tools get the shared `ToolContext` (scope + evidence chain) — passing tools via `extra_tools` won't share the chain, because that field is read before the harness's chain exists. ```python import asyncio from tau_security import SecurityHarness, SecurityHarnessConfig, Scope, ScopeRule from tau_ai import OpenAICompatibleProvider from tau_ai.env import OpenAICompatibleConfig from sc_tools import create_slither_tool, create_foundry_test_tool, create_poc_builder_tool from tau_security.tools import ( create_record_finding_tool, create_generate_report_tool, ) # Domain scope (smart contract — local files only) scope = Scope(rules=[ ScopeRule(pattern="/abs/path/to/dvd", type="path", rule="allow"), ]) config = SecurityHarnessConfig( provider=OpenAICompatibleProvider(OpenAICompatibleConfig(api_key="sk-...")), model="gpt-4o", scope=scope, autonomy="gated", system_prompt=DOMAIN_SYSTEM_PROMPT, # see below — you customize the state machine ) class SmartContractHarness(SecurityHarness): """SecurityHarness + smart-contract domain tools, sharing one scope + chain.""" def _build_tools(self): # record_finding + generate_report are still useful for evidence and the report. # The web-focused defaults (port_scan, http_probe, code_scan) are dropped. return [ create_record_finding_tool(self._ctx), create_generate_report_tool(self._ctx), create_slither_tool(self._ctx), create_foundry_test_tool(self._ctx), create_poc_builder_tool(self._ctx), ] harness = SmartContractHarness(config) ``` `self._ctx` is the `ToolContext` the harness builds from your `scope` and its own `EvidenceChain`. Passing it into your domain tool factories means scope checks and evidence land in the same chain as the built-ins — that is what makes the final report and `verify_integrity()` correct. The default `SECURITY_SYSTEM_PROMPT` encodes the web/bug-bounty state machine (Recon → Exploit → Evidence → Report). For C2, override it with a domain state machine so the model drives the right tools. Example for smart contracts: ```python DOMAIN_SYSTEM_PROMPT = """\ You are a smart contract security harness operating under a strict engagement scope. ## Operating model — three modes 1. DETECT — Use slither to find vulnerabilities in the in-scope contract directory. Record every detector finding with evidence. 2. PATCH — For each confirmed finding, propose a fix. Verify the fix with the foundry_test tool (tests pass) and re-run slither (finding gone). 3. EXPLOIT — For each confirmed finding, use poc_builder to deploy a proof-of-concept against the forked mainnet. Record whether the exploit succeeded. ## Critical rules - SCOPE IS ABSOLUTE. Only the in-scope contract directory may be analyzed. - EVIDENCE IS MANDATORY. Every finding must have linked evidence. - Use record_finding for confirmed issues. Use generate_report for the deliverable. """ ``` For cloud, the modes are Detect (Prowler) / Remediate (apply + rescan) / Exploit (red-team the exposure). The structure of the system prompt is identical — only the tool names and verbs change. ### 2.3 Run against a real lab target Wire the event listener and run the harness against your target: ```python from tau_agent import ToolExecutionEndEvent async def run_challenge(challenge_name: str): print(f"\n=== {challenge_name} ===") prompt = ( f"Analyze the {challenge_name} challenge at /abs/path/to/dvd/{challenge_name}. " f"Run slither, record findings, propose patches, and build a PoC exploit." ) async for event in harness.prompt(prompt): if isinstance(event, ToolExecutionEndEvent): status = "OK" if event.result.ok else "BLOCKED" print(f" [{status}] {event.result.name}") confirmed = harness.triage() print(f"\nConfirmed findings: {len(confirmed)}") return confirmed asyncio.run(run_challenge("unstoppable")) ``` ```bash # Or run the whole harness as a script python harness.py --challenge unstoppable --fork-block 15000000 ``` **Smart contract**: For each challenge, the harness should Detect the vulnerability (Slither finding), generate a Patch verified by both gates (re-detect clean + Foundry tests pass), and build an Exploit PoC that succeeds on the forked mainnet. **Cloud**: For each misconfigured resource, the harness should Detect the posture violation (Prowler), apply a Remediation verified by both gates (re-scan clean + app still runs), and attempt an Exploit proving the exposure is real. ### 2.4 Capture structured findings with evidence Every confirmed finding carries the full evidence chain. Verify it is complete and tamper-evident: ```python assert harness.verify_integrity(), "Evidence chain broken — tamper detected" for finding in harness.confirmed_findings: assert finding.evidence_ids, f"Finding {finding.id} has no linked evidence" print(f"Chain head hash: {harness.chain._previous_hash}") print(f"Total evidence: {len(harness.chain.evidence)}") ``` ### 2.5 Generate the client-ready report ```python open("report.json", "w").write(harness.report("json")) open("report.html", "w").write(harness.report("html")) harness.export_evidence("evidence.json") ``` The tau-security report generator produces the same structure for any domain: summary stats, severity breakdown, confirmed findings table, methodology, evidence chain hash, and integrity verification. For C2, the `location` and `cwe_id` fields carry domain meaning (contract path / line, or resource ARN / control ID). Dual output: JSON + HTML. **Checkpoint**: You have a harness that ran three modes against a real lab target, produced confirmed findings with linked evidence, verified chain integrity, and generated a client-ready report. --- ## Phase 3 — Benchmark and Publish (C2.3, 15 min) ### 3.1 Score against the domain benchmark Score all three modes. The harness gives you the raw data via `harness.confirmed_findings` and `harness.chain.evidence`; the scorer just compares against the known answer key. **Smart contract (EVMbench subset):** ```python def score_smart_contract(harness, challenges): known = {c: load_known_vuln(c) for c in challenges} detect = sum(1 for c in challenges if any(known[c] in f.title.lower() for f in harness.confirmed_findings)) / len(challenges) patch = sum(1 for c in challenges if patch_verified(c)) / len(challenges) exploit = sum(1 for c in challenges if exploit_succeeded(c)) / len(challenges) return {"detect": detect, "patch": patch, "exploit": exploit} ``` **Cloud (posture benchmark):** ```python def score_cloud(harness, misconfigs): detect = sum(1 for m in misconfigs if harness_found(harness, m)) / len(misconfigs) remediate = sum(1 for m in misconfigs if remediation_verified(m)) / len(misconfigs) exploit = sum(1 for m in misconfigs if exploit_succeeded(m)) / len(misconfigs) return {"detect": detect, "remediate": remediate, "exploit": exploit} ``` Report all three scores together. ### 3.2 Produce the one-page results summary ```markdown # [Harness Name] — Smart Contract Security Harness Benchmark Target: Damn Vulnerable DeFi (unstoppable, side-entrance, backdoor) Date: 2026-07-09 Built on: tau-security (SecurityHarness + custom domain tools) | Mode | Score | |----------|-------| | Detect | 67% | | Patch | 33% | | Exploit | 67% | | Property | Result | |----------|--------| | Scope enforcement | 100% OOS blocked | | Evidence chain integrity | PASS | | Confirmed findings with evidence | 9 of 9 | Client report: [report.html](report.html) · [report.json](report.json) Evidence export: [evidence.json](evidence.json) Harness: tau-security SecurityHarness · scope-enforced · 3-mode pipeline · tamper-evident chain ``` ### 3.3 Publish - **GitHub**: Create a repository with the harness config, your domain tools, the report, and the evidence. Use the one-page summary as the README. - **LinkedIn**: Write a post with the benchmark table and a link to the report. Position it as a falsifiable claim: "Here are the scores — run the same challenges and compare." - **Deepthreat.ai**: Submit as a demonstration asset showing the harness running against a real target with scored results. ### 3.4 Teardown **Smart contract**: destroy the forked mainnet instance. **Cloud**: run `terraform destroy` to remove all provisioned resources from the isolated AWS account. Verify teardown is complete — no resources left running. --- ## Deliverables checklist - [ ] `design-doc.md` — domain choice, the exact `Scope(...)` call, authorization document, tool suite, three modes - [ ] `sc_tools.py` or `cloud_tools.py` — domain tools as `AgentTool` instances following the tau-security pattern - [ ] `harness.py` — `SecurityHarness` configured with domain `extra_tools`, domain `Scope`, and a domain `system_prompt` - [ ] Real lab target run results — confirmed findings with evidence for known vulnerabilities/misconfigs - [ ] `report.html` and `report.json` — client-ready report - [ ] `evidence.json` — full evidence export with verified integrity - [ ] Benchmark scores — all three modes, reported together - [ ] One-page summary (`BENCHMARK.md`) — published to GitHub/LinkedIn/Deepthreat.ai - [ ] Teardown complete (if cloud track) ## Extension If you finish early: port your domain tools to the OTHER track (smart contract → cloud or vice versa) to confirm the harness architecture is domain-independent — the `SecurityHarness`, `Scope`, `EvidenceChain`, `TriagePipeline`, and report generator are unchanged; only `extra_tools` and `system_prompt` change. Or add a fourth mode (Formal Verification for smart contracts, or Continuous Monitoring for cloud). Or run the C1 benchmark gauntlet (OOS calls, injection payload, InjecAgent) against this domain-specific harness to verify the safety properties still hold.