Claude's Corner: Hex Security, AI Agents That Hack Before Attackers Do

---
description: Build a Hex Security clone, autonomous AI penetration testing agents that find and chain vulnerabilities continuously
---

# Build Hex Security: Autonomous AI Penetration Testing

## What You're Building
An agentic penetration testing platform that deploys AI-powered security agents to continuously scan web applications and APIs for vulnerabilities. Instead of a once-a-year manual pentest, the system runs 24/7, discovers attack chains across your entire surface, and delivers proof-of-concept exploits with remediation steps. Target users: security-conscious startups and mid-market companies who ship fast and need continuous coverage.

## Tech Stack
- **Frontend:** Next.js 14 with App Router, Tailwind CSS, shadcn/ui for dashboard and findings viewer
- **Backend:** Python (FastAPI) for the API layer, Celery + Redis for async job queue, PostgreSQL (Supabase) for findings storage
- **AI/ML:** Anthropic Claude or OpenAI GPT-4o for reasoning/orchestration, smaller models for classification
- **Security Tooling:** Nuclei (template scanner), httpx, subfinder, custom Python HTTP client for API fuzzing
- **Infrastructure:** Docker for isolated scan environments, Railway or Render for deployment, Redis Cloud for queuing

## Step 1: Project Setup

```bash
mkdir hex-security-clone && cd hex-security-clone
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn celery redis supabase anthropic httpx playwright pydantic python-jose
npx create-next-app@latest frontend --typescript --tailwind --app
cd frontend && npx shadcn@latest init
npx shadcn@latest add card badge button table dialog progress
```

## Step 2: Core Data Models

```sql
CREATE TABLE scan_targets (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id UUID REFERENCES auth.users,
  name TEXT NOT NULL,
  url TEXT NOT NULL,
  scope_config JSONB DEFAULT '{}'::jsonb,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE scan_jobs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  target_id UUID REFERENCES scan_targets,
  status TEXT DEFAULT 'queued',
  started_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  findings_count INT DEFAULT 0,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE findings (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  scan_job_id UUID REFERENCES scan_jobs,
  target_id UUID REFERENCES scan_targets,
  title TEXT NOT NULL,
  severity TEXT NOT NULL,
  cvss_score DECIMAL(3,1),
  vulnerability_class TEXT,
  endpoint TEXT,
  http_request TEXT,
  http_response TEXT,
  poc_steps TEXT,
  remediation TEXT,
  verified BOOLEAN DEFAULT FALSE,
  created_at TIMESTAMPTZ DEFAULT NOW()
);
```

## Step 3: LLM Orchestration Engine

```python
# backend/agents/orchestrator.py
import anthropic, json

ORCHESTRATOR_PROMPT = """You are an expert penetration tester AI. Analyze this attack surface and generate the top 10 vulnerability hypotheses.

Return JSON: {"hypotheses": [{"title": "...", "vulnerability_class": "sqli|idor|auth_bypass|xss|ssrf", "severity": "critical|high|medium|low", "test_type": "http_probe", "test_config": {"url": "...", "payloads": [...], "detection": "..."}}]}

Attack surface: {attack_surface}"""

class PentestOrchestrator:
    def __init__(self):
        self.client = anthropic.Anthropic()

    def generate_hypotheses(self, attack_surface):
        response = self.client.messages.create(
            model="claude-opus-4-5", max_tokens=4096,
            messages=[{"role": "user", "content": ORCHESTRATOR_PROMPT.format(attack_surface=json.dumps(attack_surface))}]
        )
        return json.loads(response.content[0].text)["hypotheses"]

    def analyze_finding(self, hypothesis, evidence):
        prompt = f"""Write a pentest finding report as JSON:
{{"confirmed": true/false, "title": "...", "severity": "...", "cvss_score": 0.0-10.0, "poc_steps": "...", "remediation": "..."}}
Hypothesis: {json.dumps(hypothesis)}
Evidence: {json.dumps(evidence)}"""
        response = self.client.messages.create(
            model="claude-opus-4-5", max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        return json.loads(response.content[0].text)
```

## Step 4: HTTP Fuzzing Agent

```python
# backend/agents/http_fuzzer.py
import httpx, asyncio

SQLI_PAYLOADS = ["'", "1 OR 1=1", "1; SELECT sleep(2)--", "' UNION SELECT NULL--"]
IDOR_TRANSFORMS = [lambda x: str(int(x)+1), lambda x: str(int(x)-1), lambda x: "0"]

class HTTPFuzzer:
    def __init__(self, base_url, cookies=None):
        self.base_url = base_url.rstrip("/")
        self.client = httpx.AsyncClient(cookies=cookies or {}, timeout=10.0, verify=False)

    async def test_sqli(self, endpoint, params):
        findings = []
        for param, value in params.items():
            for payload in SQLI_PAYLOADS:
                r = await self.client.get(f"{self.base_url}{endpoint}", params={**params, param: payload})
                if any(e in r.text.lower() for e in ["sql", "syntax error", "mysql", "postgresql"]):
                    findings.append({"type": "sqli", "param": param, "payload": payload, "evidence": r.text[:500]})
        return {"endpoint": endpoint, "findings": findings}

    async def test_idor(self, endpoint, id_param, current_id):
        findings = []
        for fn in IDOR_TRANSFORMS:
            test_id = fn(current_id)
            r = await self.client.get(f"{self.base_url}{endpoint}", params={id_param: test_id})
            if r.status_code == 200 and len(r.content) > 100:
                findings.append({"type": "idor", "original": current_id, "accessed": test_id})
        return {"endpoint": endpoint, "findings": findings}
```

## Step 5: Celery Scan Worker

```python
# backend/workers/scan_worker.py
from celery import Celery
from agents.orchestrator import PentestOrchestrator
from agents.http_fuzzer import HTTPFuzzer
import asyncio
from db import supabase

app = Celery("hex_clone", broker="redis://localhost:6379/0")

@app.task
def run_pentest_scan(scan_job_id, target_id):
    supabase.table("scan_jobs").update({"status": "running"}).eq("id", scan_job_id).execute()
    try:
        target = supabase.table("scan_targets").select("url").eq("id", target_id).single().execute().data
        # 1. Enumerate attack surface (simplified, add subfinder/crawling here)
        attack_surface = {"base_url": target["url"], "endpoints": ["/api/users", "/api/posts"]}
        # 2. Generate hypotheses via LLM
        orchestrator = PentestOrchestrator()
        hypotheses = orchestrator.generate_hypotheses(attack_surface)
        # 3. Execute and verify each
        findings = []
        fuzzer = HTTPFuzzer(target["url"])
        for h in hypotheses:
            evidence = asyncio.run(fuzzer.test_sqli(h["test_config"]["url"], {"id": "1"}))
            analysis = orchestrator.analyze_finding(h, evidence)
            if analysis.get("confirmed"):
                supabase.table("findings").insert({
                    "scan_job_id": scan_job_id, "target_id": target_id,
                    "title": analysis["title"], "severity": analysis["severity"],
                    "cvss_score": analysis["cvss_score"], "poc_steps": analysis["poc_steps"],
                    "remediation": analysis["remediation"], "verified": True
                }).execute()
                findings.append(analysis)
        supabase.table("scan_jobs").update({"status": "completed", "findings_count": len(findings)}).eq("id", scan_job_id).execute()
    except Exception as e:
        supabase.table("scan_jobs").update({"status": "failed"}).eq("id", scan_job_id).execute()
        raise
```

## Step 6: Frontend Dashboard

```tsx
// app/dashboard/findings/page.tsx
import { createClient } from "@/lib/supabase/server";
import { Badge } from "@/components/ui/badge";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table";

const SEVERITY_COLORS = {
  critical: "bg-red-500/10 text-red-600", high: "bg-orange-500/10 text-orange-600",
  medium: "bg-yellow-500/10 text-yellow-600", low: "bg-blue-500/10 text-blue-600",
};

export default async function FindingsPage() {
  const supabase = await createClient();
  const { data: findings } = await supabase.from("findings")
    .select("id, title, severity, cvss_score, endpoint, created_at")
    .order("cvss_score", { ascending: false }).limit(100);
  return (
    <Card>
      <CardHeader><CardTitle>Findings</CardTitle></CardHeader>
      <CardContent>
        <Table>
          <TableHeader><TableRow>
            <TableHead>Finding</TableHead><TableHead>Severity</TableHead>
            <TableHead>CVSS</TableHead><TableHead>Endpoint</TableHead>
          </TableRow></TableHeader>
          <TableBody>
            {findings?.map(f => (
              <TableRow key={f.id}>
                <TableCell className="font-medium">{f.title}</TableCell>
                <TableCell><Badge className={SEVERITY_COLORS[f.severity as keyof typeof SEVERITY_COLORS]}>{f.severity}</Badge></TableCell>
                <TableCell>{f.cvss_score}</TableCell>
                <TableCell className="font-mono text-xs">{f.endpoint}</TableCell>
              </TableRow>
            ))}
          </TableBody>
        </Table>
      </CardContent>
    </Card>
  );
}
```

## Step 7: Deploy

```bash
# Railway (backend + Redis)
railway login && railway new hex-security-backend
railway add redis
railway variables set ANTHROPIC_API_KEY=... SUPABASE_URL=... SUPABASE_SERVICE_ROLE_KEY=...
railway up

# Vercel (frontend)
cd frontend && vercel deploy --prod

# Start worker
celery -A workers.scan_worker worker --loglevel=info --concurrency=4
```

## Key Insights
- The LLM does planning and analysis; actual probes are deterministic HTTP calls. This distinction matters for reliability.
- Multi-step exploit chaining is the hard moat, no traditional scanner can chain endpoint A's info leak with endpoint B's missing auth check.
- False positive rate is the product. Every finding must be machine-verified before delivery or security teams will stop trusting you.
- Start with OWASP Top 10, SQL injection, IDOR, broken auth, XSS cover 80% of real-world findings.

## Gotchas
- **Scope creep**: Hard-code scope limits before anything. An agent that scans third-party services will get you sued.
- **Rate limiting your own agents**: Add jitter and rate limits, your scan traffic will look like an attack. Always get written permission first.
- **Nuclei template noise**: The open-source templates have many false positives. Curate heavily, only use high-confidence templates for automated reporting.
- **LLM hallucination in reports**: Add a validation step that re-runs the exact PoC request before marking a finding as confirmed. Claude will occasionally hallucinate confirmed vulnerabilities from ambiguous evidence.