Lab Specification — Module S12: Advanced Smart Contract Harnesses

# Load the EVMbench subset metadata
import json

evmbench_subset = json.load(open("evmbench-subset.json"))
# Each entry: {repo, contract, vulnerability_class, expected_finding, exploit_expected, patch_expected}
print(f"Subset: {len(evmbench_subset)} vulnerabilities across {len(set(v['repo'] for v in evmbench_subset))} repos")

class EVMbenchScorer:
    def __init__(self, harness):
        self.harness = harness  # the S11 audit harness

    def score_detect(self, vuln) -> bool:
        """Did the harness find the known vulnerability?"""
        findings = self.harness.detect(vuln["contract_source"])
        return any(self.matches_expected(f, vuln["expected_finding"]) for f in findings)

    def score_patch(self, vuln) -> dict:
        """Did the harness generate a correct patch?"""
        patch = self.harness.generate_patch(vuln)
        if not patch:
            return {"generated": False, "correct": False, "preserves_behavior": False}
        # Gate 1: original finding gone, no new findings
        finding_gone = self.harness.verify_gate1(vuln, patch)
        # Gate 2: test suite passes (behavior preserved)
        behavior_ok = self.harness.verify_gate2(vuln, patch)
        return {
            "generated": True,
            "correct": finding_gone,
            "preserves_behavior": behavior_ok,
            "quality": "pass" if (finding_gone and behavior_ok) else "fail"
        }

    def score_exploit(self, vuln) -> dict:
        """Did the harness build a working PoC?"""
        poc = self.harness.build_exploit_poc(vuln)
        if not poc:
            return {"built": False, "runs": False, "succeeds": False}
        result = self.harness.run_poc_forked(poc, vuln)
        return {"built": True, "runs": result.executed, "succeeds": result.exploit_succeeded}

    def score_all(self, subset) -> dict:
        detect_hits = sum(1 for v in subset if self.score_detect(v))
        patch_results = [self.score_patch(v) for v in subset]
        exploit_results = [self.score_exploit(v) for v in subset]
        return {
            "detect_recall": detect_hits / len(subset),
            "patch_quality": sum(1 for p in patch_results if p["quality"] == "pass") / len(subset),
            "exploit_success_rate": sum(1 for e in exploit_results if e["succeeds"]) / len(subset),
            "total": len(subset),
            "per_vuln": [
                {"vuln_id": v["id"], "detect": self.score_detect(v),
                 "patch": self.score_patch(v), "exploit": self.score_exploit(v)}
                for v in subset
            ]
        }

def results_table(scores):
    print(f"{'Vuln ID':<12} {'Class':<20} {'Detect':<10} {'Patch':<15} {'Exploit':<10}")
    print("-" * 67)
    for v in scores["per_vuln"]:
        d = "PASS" if v["detect"] else "FAIL"
        p = v["patch"]["quality"]
        e = "PASS" if v["exploit"]["succeeds"] else "FAIL"
        print(f"{v['vuln_id']:<12} {v.get('class','?'):<20} {d:<10} {p:<15} {e:<10}")
    print("-" * 67)
    print(f"{'TOTAL':<12} {'':<20} {scores['detect_recall']:.0%}{'':<5} {scores['patch_quality']:.0%}{'':<8} {scores['exploit_success_rate']:.0%}")

# Install Solana CLI
sh -c "$(curl -sSfL https://release.solana.com/stable/install)"
# Install Anchor
cargo install --git https://github.com/coral-xyz/anchor anchor-cli --tag v0.30.0
# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

// Example vulnerable Anchor program — missing has_one constraint
#[derive(Accounts)]
pub struct Withdraw<'info> {
    #[account(mut)]
    pub user: Signer<'info>,
    #[account(mut)]
    pub vault: Account<'info, Vault>,  // BUG: no has_one = user constraint
    // An attacker can pass any vault account, not just one owned by `user`
}

class SolanaSecurityHarness:
    def detect(self, program_source):
        """Detect vulnerabilities in an Anchor program."""
        candidates = []
        # 1. Anchor constraint checking — analyze #[derive(Accounts)] for missing constraints
        candidates.extend(self.check_anchor_constraints(program_source))
        # 2. Custom Semgrep rules for Rust/Anchor patterns
        candidates.extend(self.run_semgrep_solana(program_source))
        # 3. LLM semantic reasoning on candidates
        return [self.llm_triage(c, program_source) for c in candidates]

    def check_anchor_constraints(self, source):
        """Check #[derive(Accounts)] structs for missing has_one, constraint, address."""
        findings = []
        for account_struct in extract_derive_accounts(source):
            for field in account_struct.fields:
                # Flag accounts without has_one or constraint that handle value transfer
                if field.is_mut and not field.has_constraint and field.involves_value:
                    findings.append({
                        "class": "account_confusion",
                        "location": account_struct.name,
                        "field": field.name,
                        "detail": f"Account '{field.name}' is mutable and handles value but has no has_one/constraint"
                    })
        return findings

    def build_exploit_poc(self, vuln):
        """Build a Solana exploit PoC using bankrun."""
        # Generate a TypeScript test using bankrun that:
        # 1. Sets up the vulnerable program
        # 2. Passes a malicious account (wrong type/owner)
        # 3. Asserts value was transferred incorrectly
        return self.generate_bankrun_test(vuln)

# Run the Solana harness
python3 solana_harness.py --program ./vulnerable-anchor/ --output solana-findings.json

findings = [
    {
        "id": "S11-F001",
        "title": "Reentrancy in Vault.withdraw allows drainage",
        "severity": "critical",  # harness draft severity
        "status": "open",
        "location": "Vault.sol:142, withdraw() function",
        "description": "The withdraw function makes an external call to msg.sender before updating the user's balance...",
        "impact": "An attacker can recursively call withdraw to drain the vault's entire ETH balance.",
        "poc": "test/Exploit.t.sol (forge test --match-test test_reentrancy_exploit)",
        "recommendation": "Apply checks-effects-interactions pattern: update balance before external call, or use ReentrancyGuard.",
        "economic_impact": "10,000 ETH at fork-block prices (~$17M)"
    },
    # ... additional findings from S11 labs
]

class AuditReportGenerator:
    SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "informational": 4}

    def generate(self, findings, scope):
        # Sort findings by severity
        sorted_findings = sorted(findings, key=lambda f: self.SEVERITY_ORDER[f["severity"]])
        return {
            "scope": scope,
            "methodology": self.methodology(),
            "findings_table": self.findings_table(sorted_findings),
            "severity_breakdown": self.severity_breakdown(sorted_findings),
            "detailed_findings": [self.detailed(f) for f in sorted_findings],
            "remediation_roadmap": self.roadmap(sorted_findings),
        }

    def methodology(self):
        return (
            "This audit was conducted using an AI-assisted harness combining static analysis "
            "(Slither, Mythril, Semgrep), LLM semantic reasoning, invariant extraction, "
            "Foundry-based exploit PoCs on forked mainnet, and cascaded verification for patches. "
            "All findings were human-confirmed before publication."
        )

    def findings_table(self, findings):
        return [{"id": f["id"], "title": f["title"], "severity": f["severity"],
                 "status": f["status"], "location": f["location"]} for f in findings]

    def severity_breakdown(self, findings):
        from collections import Counter
        counts = Counter(f["severity"] for f in findings)
        return {s: counts.get(s, 0) for s in ["critical", "high", "medium", "low", "informational"]}

    def detailed(self, finding):
        return {
            "id": finding["id"],
            "title": finding["title"],
            "severity": finding["severity"],
            "description": finding["description"],
            "impact": finding["impact"],
            "proof_of_concept": finding["poc"],
            "recommendation": finding["recommendation"],
            "economic_impact": finding.get("economic_impact"),
            "status": finding["status"],
        }

    def roadmap(self, findings):
        critical = [f for f in findings if f["severity"] == "critical"]
        high = [f for f in findings if f["severity"] == "high"]
        return {
            "immediate": [f["id"] for f in critical],  # fix before deployment
            "short_term": [f["id"] for f in high],      # fix within 1 sprint
            "note": "All Critical and High findings should be remediated before mainnet deployment. "
                    "Each fix must be verified: PoC re-run must fail, test suite must pass, Slither must be clean."
        }

# Generate Markdown report
report = generator.generate(findings, scope)
markdown = render_markdown(report)  # template-based rendering
with open("audit-report.md", "w") as f:
    f.write(markdown)

# Lab Specification — Module S12: Advanced Smart Contract Harnesses

**Course**: 2A — Building AI Harnesses for Cybersecurity
**Module**: S12 — Advanced Smart Contract Harnesses
**Duration**: 120–150 minutes (three labs, one per sub-section)
**Environment**: Python 3.11+, the S11 audit harness (complete), Foundry, Slither. For Lab 2: Solana CLI, Anchor, Rust. An LLM API (Claude or GPT-class).

> These labs build on the S11 harness. Lab 1 makes it measurable; Lab 2 ports it; Lab 3 turns its output into a deliverable.

---

## Learning objectives

1. Configure the audit harness to run against an EVMbench subset and produce a scored results table across Detect, Patch, and Exploit.
2. Run a Solana security harness against a known-vulnerable Anchor program and identify at least one vulnerability class.
3. Generate a client-ready audit report from Pillar 4 findings, matching the structure top firms publish.

---

## Phase 1 — EVMbench Subset Scoring (45 min)

### 1.1 Obtain an EVMbench subset

EVMbench's full dataset contains 117 vulnerabilities across 40 repositories. For this lab, select a subset of 10–15 vulnerabilities (e.g., the reentrancy and flash loan classes — 10 challenges total).

```python
# Load the EVMbench subset metadata
import json

evmbench_subset = json.load(open("evmbench-subset.json"))
# Each entry: {repo, contract, vulnerability_class, expected_finding, exploit_expected, patch_expected}
print(f"Subset: {len(evmbench_subset)} vulnerabilities across {len(set(v['repo'] for v in evmbench_subset))} repos")
```

### 1.2 Run the S11 harness against each vulnerability

For each vulnerability in the subset, run the three modes and score them independently.

```python
class EVMbenchScorer:
    def __init__(self, harness):
        self.harness = harness  # the S11 audit harness

    def score_detect(self, vuln) -> bool:
        """Did the harness find the known vulnerability?"""
        findings = self.harness.detect(vuln["contract_source"])
        return any(self.matches_expected(f, vuln["expected_finding"]) for f in findings)

    def score_patch(self, vuln) -> dict:
        """Did the harness generate a correct patch?"""
        patch = self.harness.generate_patch(vuln)
        if not patch:
            return {"generated": False, "correct": False, "preserves_behavior": False}
        # Gate 1: original finding gone, no new findings
        finding_gone = self.harness.verify_gate1(vuln, patch)
        # Gate 2: test suite passes (behavior preserved)
        behavior_ok = self.harness.verify_gate2(vuln, patch)
        return {
            "generated": True,
            "correct": finding_gone,
            "preserves_behavior": behavior_ok,
            "quality": "pass" if (finding_gone and behavior_ok) else "fail"
        }

    def score_exploit(self, vuln) -> dict:
        """Did the harness build a working PoC?"""
        poc = self.harness.build_exploit_poc(vuln)
        if not poc:
            return {"built": False, "runs": False, "succeeds": False}
        result = self.harness.run_poc_forked(poc, vuln)
        return {"built": True, "runs": result.executed, "succeeds": result.exploit_succeeded}

    def score_all(self, subset) -> dict:
        detect_hits = sum(1 for v in subset if self.score_detect(v))
        patch_results = [self.score_patch(v) for v in subset]
        exploit_results = [self.score_exploit(v) for v in subset]
        return {
            "detect_recall": detect_hits / len(subset),
            "patch_quality": sum(1 for p in patch_results if p["quality"] == "pass") / len(subset),
            "exploit_success_rate": sum(1 for e in exploit_results if e["succeeds"]) / len(subset),
            "total": len(subset),
            "per_vuln": [
                {"vuln_id": v["id"], "detect": self.score_detect(v),
                 "patch": self.score_patch(v), "exploit": self.score_exploit(v)}
                for v in subset
            ]
        }
```

### 1.3 Produce the scored results table

```python
def results_table(scores):
    print(f"{'Vuln ID':<12} {'Class':<20} {'Detect':<10} {'Patch':<15} {'Exploit':<10}")
    print("-" * 67)
    for v in scores["per_vuln"]:
        d = "PASS" if v["detect"] else "FAIL"
        p = v["patch"]["quality"]
        e = "PASS" if v["exploit"]["succeeds"] else "FAIL"
        print(f"{v['vuln_id']:<12} {v.get('class','?'):<20} {d:<10} {p:<15} {e:<10}")
    print("-" * 67)
    print(f"{'TOTAL':<12} {'':<20} {scores['detect_recall']:.0%}{'':<5} {scores['patch_quality']:.0%}{'':<8} {scores['exploit_success_rate']:.0%}")
```

### Deliverable
- [ ] Scored results table for the EVMbench subset (10+ vulnerabilities)
- [ ] All three scores reported: Detect recall, Patch quality, Exploit success rate
- [ ] Per-vulnerability breakdown showing which passed/failed each mode
- [ ] Analysis: which vulnerability classes does the harness handle best? Which need improvement?

---

## Phase 2 — Solana Anchor Security Harness (45 min)

### 2.1 Set up the Solana environment

```bash
# Install Solana CLI
sh -c "$(curl -sSfL https://release.solana.com/stable/install)"
# Install Anchor
cargo install --git https://github.com/coral-xyz/anchor anchor-cli --tag v0.30.0
# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
```

### 2.2 Obtain a known-vulnerable Anchor program

Use a deliberately vulnerable Solana program (e.g., from the Solana SEED or a CTF-style vulnerable Anchor program). Alternatively, construct one with a known account-confusion vulnerability.

```rust
// Example vulnerable Anchor program — missing has_one constraint
#[derive(Accounts)]
pub struct Withdraw<'info> {
    #[account(mut)]
    pub user: Signer<'info>,
    #[account(mut)]
    pub vault: Account<'info, Vault>,  // BUG: no has_one = user constraint
    // An attacker can pass any vault account, not just one owned by `user`
}
```

### 2.3 Build the Solana security harness

Adapt the S11 harness architecture to Solana. The three-mode structure ports; the tools change.

```python
class SolanaSecurityHarness:
    def detect(self, program_source):
        """Detect vulnerabilities in an Anchor program."""
        candidates = []
        # 1. Anchor constraint checking — analyze #[derive(Accounts)] for missing constraints
        candidates.extend(self.check_anchor_constraints(program_source))
        # 2. Custom Semgrep rules for Rust/Anchor patterns
        candidates.extend(self.run_semgrep_solana(program_source))
        # 3. LLM semantic reasoning on candidates
        return [self.llm_triage(c, program_source) for c in candidates]

    def check_anchor_constraints(self, source):
        """Check #[derive(Accounts)] structs for missing has_one, constraint, address."""
        findings = []
        for account_struct in extract_derive_accounts(source):
            for field in account_struct.fields:
                # Flag accounts without has_one or constraint that handle value transfer
                if field.is_mut and not field.has_constraint and field.involves_value:
                    findings.append({
                        "class": "account_confusion",
                        "location": account_struct.name,
                        "field": field.name,
                        "detail": f"Account '{field.name}' is mutable and handles value but has no has_one/constraint"
                    })
        return findings

    def build_exploit_poc(self, vuln):
        """Build a Solana exploit PoC using bankrun."""
        # Generate a TypeScript test using bankrun that:
        # 1. Sets up the vulnerable program
        # 2. Passes a malicious account (wrong type/owner)
        # 3. Asserts value was transferred incorrectly
        return self.generate_bankrun_test(vuln)
```

### 2.4 Run and identify the vulnerability

Run the harness against the vulnerable Anchor program. Identify at least one vulnerability class (account confusion, missing signer, PDA misuse, or integer arithmetic).

```bash
# Run the Solana harness
python3 solana_harness.py --program ./vulnerable-anchor/ --output solana-findings.json
```

### Deliverable
- [ ] Solana harness adapting the S11 architecture (Detect mode at minimum)
- [ ] Anchor constraint checker identifying missing has_one/constraint
- [ ] At least one vulnerability class identified in the test program
- [ ] Documentation of what ports (3-mode architecture) and what does not (Slither/Mythril)

---

## Phase 3 — Client-Ready Audit Report Generation (30 min)

### 3.1 Collect findings from Pillar 4

Aggregate the verified findings from the S11 labs and Phase 1 of this module into a structured finding set.

```python
findings = [
    {
        "id": "S11-F001",
        "title": "Reentrancy in Vault.withdraw allows drainage",
        "severity": "critical",  # harness draft severity
        "status": "open",
        "location": "Vault.sol:142, withdraw() function",
        "description": "The withdraw function makes an external call to msg.sender before updating the user's balance...",
        "impact": "An attacker can recursively call withdraw to drain the vault's entire ETH balance.",
        "poc": "test/Exploit.t.sol (forge test --match-test test_reentrancy_exploit)",
        "recommendation": "Apply checks-effects-interactions pattern: update balance before external call, or use ReentrancyGuard.",
        "economic_impact": "10,000 ETH at fork-block prices (~$17M)"
    },
    # ... additional findings from S11 labs
]
```

### 3.2 Implement the report generator

```python
class AuditReportGenerator:
    SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "informational": 4}

    def generate(self, findings, scope):
        # Sort findings by severity
        sorted_findings = sorted(findings, key=lambda f: self.SEVERITY_ORDER[f["severity"]])
        return {
            "scope": scope,
            "methodology": self.methodology(),
            "findings_table": self.findings_table(sorted_findings),
            "severity_breakdown": self.severity_breakdown(sorted_findings),
            "detailed_findings": [self.detailed(f) for f in sorted_findings],
            "remediation_roadmap": self.roadmap(sorted_findings),
        }

    def methodology(self):
        return (
            "This audit was conducted using an AI-assisted harness combining static analysis "
            "(Slither, Mythril, Semgrep), LLM semantic reasoning, invariant extraction, "
            "Foundry-based exploit PoCs on forked mainnet, and cascaded verification for patches. "
            "All findings were human-confirmed before publication."
        )

    def findings_table(self, findings):
        return [{"id": f["id"], "title": f["title"], "severity": f["severity"],
                 "status": f["status"], "location": f["location"]} for f in findings]

    def severity_breakdown(self, findings):
        from collections import Counter
        counts = Counter(f["severity"] for f in findings)
        return {s: counts.get(s, 0) for s in ["critical", "high", "medium", "low", "informational"]}

    def detailed(self, finding):
        return {
            "id": finding["id"],
            "title": finding["title"],
            "severity": finding["severity"],
            "description": finding["description"],
            "impact": finding["impact"],
            "proof_of_concept": finding["poc"],
            "recommendation": finding["recommendation"],
            "economic_impact": finding.get("economic_impact"),
            "status": finding["status"],
        }

    def roadmap(self, findings):
        critical = [f for f in findings if f["severity"] == "critical"]
        high = [f for f in findings if f["severity"] == "high"]
        return {
            "immediate": [f["id"] for f in critical],  # fix before deployment
            "short_term": [f["id"] for f in high],      # fix within 1 sprint
            "note": "All Critical and High findings should be remediated before mainnet deployment. "
                    "Each fix must be verified: PoC re-run must fail, test suite must pass, Slither must be clean."
        }
```

### 3.3 Render the report

Generate the report as Markdown and/or HTML matching the format of published reports from Trail of Bits, Consensys Diligence, and OpenZeppelin.

```python
# Generate Markdown report
report = generator.generate(findings, scope)
markdown = render_markdown(report)  # template-based rendering
with open("audit-report.md", "w") as f:
    f.write(markdown)
```

### 3.4 Human review checkpoint

Before the report is "client-ready," a human auditor must:
1. Review every finding's severity (harness severity = draft; auditor = final).
2. Confirm the PoC references are reproducible.
3. Edit the methodology to accurately reflect what was done.
4. Approve the remediation roadmap priorities.

### Deliverable
- [ ] Structured finding set from Pillar 4 labs
- [ ] Report generator producing the six-section structure
- [ ] Severity breakdown table (Critical/High/Medium/Low/Informational counts)
- [ ] Markdown or HTML report matching top-firm format
- [ ] Documentation of the human review checkpoint (which severities were changed and why)

---

## Stretch goals

1. **Run Phase 1 against the full EVMbench dataset** (all 117 vulnerabilities). Compare your harness's three scores to Heimdallr's 92.45% detection and to the GPT-5.1 34% baseline. Document where your harness exceeds or falls short.
2. **Build a bridge security harness** for Phase 2. Model a simple lock-mint bridge as a cross-chain state machine, trace asset flows, and check the "minted <= locked" invariant. Identify how a signature-verification flaw would violate it.
3. **Add Exploit mode to the Solana harness** (Phase 2). Use `bankrun` to build and run a PoC that exploits the account-confusion vulnerability. Capture structured evidence (transaction signature, account state before/after).
4. **Automate the severity confirmation** — have a secondary LLM propose severity corrections for the human auditor to approve, pre-filtering findings where the draft severity is likely wrong.