Zen AI Pentest Framework

Zen AI Pentest Framework

Skill by ara.so — Security Skills collection.

Overview

Zen-AI-Pentest is a production-ready, AI-powered autonomous penetration testing framework that orchestrates 72+ real security tools through an intelligent multi-agent system. It executes actual tools (Nmap, Nuclei, SQLMap, FFuF, etc.) with safety controls, not mocks or simulations.

Key capabilities:

• Autonomous AI agents using ReAct pattern (Reason → Act → Observe → Reflect)
• Real tool execution in Docker sandbox with 4-level safety controls
• Risk engine with false positive reduction and CVSS/EPSS scoring
• FastAPI backend with WebSocket real-time updates
• PostgreSQL persistence and Redis caching
• JWT authentication with RBAC
• Professional PDF/HTML reporting with compliance mapping
• CI/CD integration (GitHub Actions, GitLab CI, Jenkins)

Installation

Prerequisites

• Python 3.10+
• Docker and Docker Compose
• PostgreSQL 13+ (or use Docker Compose)
• Redis (optional, for caching)

Quick Start with Docker Compose

# Clone repository
git clone https://github.com/SHAdd0WTAka/Zen-Ai-Pentest.git
cd Zen-Ai-Pentest

# Set environment variables
cp .env.example .env
# Edit .env with your API keys and configuration

# Start full stack
docker-compose up -d

# Access web UI
# http://localhost:3000 (default credentials: admin/admin)

# Access API docs
# http://localhost:8000/docs

Manual Installation

# Install Python dependencies
pip install -r requirements.txt

# Install security tools (Ubuntu/Debian)
sudo apt-get update
sudo apt-get install -y nmap nuclei sqlmap ffuf whatweb wafw00f subfinder httpx nikto

# Initialize database
python -m alembic upgrade head

# Run API server
uvicorn api.main:app --host 0.0.0.0 --port 8000 --reload

# In another terminal, run frontend
cd frontend
npm install
npm run dev

Configuration

Environment Variables

Create .env file in project root:

# API Keys (REQUIRED)
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

# Database
DATABASE_URL=postgresql://zen:zen_password@localhost:5432/zen_pentest
REDIS_URL=redis://localhost:6379/0

# Security
JWT_SECRET_KEY=your-random-secret-key-min-32-chars
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

# API Configuration
API_HOST=0.0.0.0
API_PORT=8000
CORS_ORIGINS=http://localhost:3000,http://localhost:5173

# Safety Controls (0-3)
# 0: Read-only info gathering
# 1: Active scanning (port scans)
# 2: Vulnerability probing
# 3: Exploit validation (requires VPN)
DEFAULT_SAFETY_LEVEL=1
REQUIRE_VPN_FOR_LEVEL_3=true

# Notifications (optional)
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-app-password

# Agent Configuration
MAX_ITERATIONS=10
AGENT_TIMEOUT=300
ENABLE_MEMORY=true
MEMORY_WINDOW=5

# Tool Timeouts (seconds)
NMAP_TIMEOUT=300
NUCLEI_TIMEOUT=600
SQLMAP_TIMEOUT=900

Configuration File

config/config.yaml:

agents:
  default_model: "gpt-4"
  fallback_model: "claude-3-opus"
  temperature: 0.7
  max_tokens: 4096
  
tools:
  enabled:
    - nmap
    - nuclei
    - sqlmap
    - ffuf
    - subfinder
  
  nmap:
    default_args: ["-sV", "-sC", "-T4"]
    max_ports: 10000
    
  nuclei:
    templates_dir: "/nuclei-templates"
    severity: ["critical", "high", "medium"]
    
reporting:
  formats: ["pdf", "html", "json"]
  templates_dir: "templates/reports"
  
compliance:
  frameworks: ["OWASP", "PCI-DSS", "GDPR"]

Core API Usage

Authentication

import requests

API_BASE = "http://localhost:8000"

# Login
response = requests.post(
    f"{API_BASE}/api/v1/auth/login",
    json={"username": "admin", "password": "admin"}
)
token = response.json()["access_token"]

# Use token in headers
headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

Create and Execute Scan

# Create a new scan
scan_data = {
    "target": "example.com",
    "scan_type": "full",  # Options: quick, full, custom
    "tools": ["nmap", "nuclei", "whatweb"],
    "safety_level": 1,  # 0-3
    "options": {
        "ports": "1-1000",
        "aggressive": False,
        "parallel": True
    }
}

response = requests.post(
    f"{API_BASE}/api/v1/scans",
    json=scan_data,
    headers=headers
)
scan_id = response.json()["scan_id"]
print(f"Scan created: {scan_id}")

# Get scan status
status = requests.get(
    f"{API_BASE}/api/v1/scans/{scan_id}",
    headers=headers
).json()
print(f"Status: {status['state']}")  # IDLE, PLANNING, EXECUTING, etc.

# Get results
results = requests.get(
    f"{API_BASE}/api/v1/scans/{scan_id}/results",
    headers=headers
).json()

print(f"Findings: {len(results['findings'])}")
for finding in results['findings']:
    print(f"  - {finding['severity']}: {finding['title']}")

WebSocket Real-Time Updates

import asyncio
import websockets
import json

async def monitor_scan(scan_id, token):
    uri = f"ws://localhost:8000/api/v1/ws/scans/{scan_id}?token={token}"
    
    async with websockets.connect(uri) as websocket:
        async for message in websocket:
            data = json.loads(message)
            print(f"[{data['type']}] {data['message']}")
            
            if data['type'] == 'completed':
                break

# Run
asyncio.run(monitor_scan(scan_id, token))

Python SDK Usage

Basic Scan Workflow

from zen_pentest import ZenClient
from zen_pentest.agents import AutonomousAgent
from zen_pentest.tools import ToolRegistry

# Initialize client
client = ZenClient(
    api_url="http://localhost:8000",
    api_key=os.getenv("ZEN_API_KEY")
)

# Quick scan
scan = client.create_scan(
    target="example.com",
    scan_type="quick"
)
scan.wait_for_completion(timeout=600)
report = scan.get_report(format="pdf")
report.save("report.pdf")

# Advanced scan with custom agent
agent = AutonomousAgent(
    llm="gpt-4",
    tools=["nmap", "nuclei", "sqlmap"],
    safety_level=2,
    memory_enabled=True
)

result = agent.run_scan(
    target="https://example.com",
    objectives=[
        "Map attack surface",
        "Identify web vulnerabilities",
        "Test for SQL injection"
    ]
)

print(f"State transitions: {result.state_history}")
print(f"Findings: {len(result.findings)}")
print(f"Risk score: {result.risk_score}")

Custom Tool Integration

from zen_pentest.tools import BaseTool
from zen_pentest.schemas import ToolResult

class CustomScannerTool(BaseTool):
    name = "custom_scanner"
    description = "Custom vulnerability scanner"
    risk_level = 1
    
    def validate_input(self, target: str) -> bool:
        # Validate target (no private IPs, etc.)
        return self.is_public_target(target)
    
    def execute(self, target: str, **kwargs) -> ToolResult:
        # Your custom logic
        results = self._run_custom_scan(target)
        
        return ToolResult(
            success=True,
            data=results,
            metadata={"tool": self.name, "target": target}
        )
    
    def parse_output(self, raw_output: str) -> dict:
        # Parse tool output
        return {"vulnerabilities": [...]}

# Register custom tool
registry = ToolRegistry()
registry.register(CustomScannerTool())

# Use in scan
agent = AutonomousAgent(tools=["custom_scanner"])
result = agent.run_scan(target="example.com")

Multi-Agent Coordination

from zen_pentest.agents import ResearcherAgent, AnalystAgent, CoordinatorAgent

# Create specialized agents
researcher = ResearcherAgent(
    tools=["subfinder", "httpx", "whatweb"]
)

analyst = AnalystAgent(
    tools=["nuclei", "ffuf", "nikto"]
)

# Coordinate workflow
coordinator = CoordinatorAgent(
    agents=[researcher, analyst],
    strategy="sequential"  # or "parallel"
)

result = coordinator.execute_mission(
    target="example.com",
    depth=3
)

# Researcher findings feed into analyst
for finding in result.findings: