Lab Specification — Capstone C2: Build a Smart Contract or Cloud Security Harness on tau-security

pip install tau-ai
cd course/02a-security-harnesses/tau-security
pip install -e ".[dev]"

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()])

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"),
])

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
])

## 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

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)

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")

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)

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.
"""

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"))

# Or run the whole harness as a script
python harness.py --challenge unstoppable --fork-block 15000000

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)}")

open("report.json", "w").write(harness.report("json"))
open("report.html", "w").write(harness.report("html"))
harness.export_evidence("evidence.json")

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}

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}

# [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

# 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.