metatron-pentest-assistant

# METATRON Penetration Testing Assistant

> Skill by [ara.so](https://ara.so) — Daily 2026 Skills collection.

METATRON is a CLI-based AI penetration testing assistant that runs entirely locally — no cloud, no API keys. It orchestrates recon tools (nmap, whois, whatweb, curl, dig, nikto), feeds results to a locally running fine-tuned LLM (`metatron-qwen` via Ollama), and stores all findings in MariaDB with full scan history, vulnerability tracking, and PDF/HTML export.

---

## Architecture Overview

```
metatron.py     ← CLI entry point, main menu, scan orchestration
db.py           ← MariaDB CRUD (history, vulns, fixes, exploits, summary)
tools.py        ← Recon tool runners (nmap, whois, whatweb, curl, dig, nikto)
llm.py          ← Ollama interface, agentic loop, AI tool dispatch
search.py       ← DuckDuckGo search + CVE lookup (no API key)
Modelfile       ← Custom metatron-qwen model config
```

**Database spine:** every scan creates a `sl_no` in `history`; all other tables link via `sl_no`.

---

## Installation

### 1. Clone and set up Python environment

```bash
git clone https://github.com/sooryathejas/METATRON.git
cd METATRON
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```

### 2. Install system recon tools

```bash
sudo apt install nmap whois whatweb curl dnsutils nikto
```

### 3. Install Ollama and pull base model

```bash
curl -fsSL https://ollama.com/install.sh | sh

# 8GB+ RAM:
ollama pull huihui_ai/qwen3.5-abliterated:9b

# <8GB RAM — use 4b and edit Modelfile FROM line accordingly:
ollama pull huihui_ai/qwen3.5-abliterated:4b
```

### 4. Build the custom metatron-qwen model

```bash
ollama create metatron-qwen -f Modelfile
ollama list   # verify metatron-qwen appears
```

**Modelfile** (the repo ships this — key parameters):
```
FROM huihui_ai/qwen3.5-abliterated:9b
PARAMETER num_ctx 16384
PARAMETER temperature 0.7
PARAMETER top_k 10
PARAMETER top_p 0.9
```

To use 4b instead, edit `Modelfile`:
```
FROM huihui_ai/qwen3.5-abliterated:4b
```
Then rebuild: `ollama create metatron-qwen -f Modelfile`

### 5. Set up MariaDB

```bash
sudo systemctl start mariadb
sudo systemctl enable mariadb
mysql -u root
```

```sql
CREATE DATABASE metatron;
CREATE USER 'metatron'@'localhost' IDENTIFIED BY '123';
GRANT ALL PRIVILEGES ON metatron.* TO 'metatron'@'localhost';
FLUSH PRIVILEGES;
EXIT;
```

Create all tables:
```bash
mysql -u metatron -p123 metatron < schema.sql
```

Or manually (paste from README schema block). The 5 tables:
- `history` — one row per scan session (spine)
- `vulnerabilities` — findings per session
- `fixes` — remediation per vulnerability
- `exploits_attempted` — exploit attempts per session
- `summary` — raw scan + full AI analysis dump

---

## Running METATRON

METATRON requires **two terminals**:

**Terminal 1 — Load model into memory:**
```bash
ollama run metatron-qwen
# Wait for >>> prompt before proceeding
```

**Terminal 2 — Launch the assistant:**
```bash
cd ~/METATRON
source venv/bin/activate
python metatron.py
```

### Main Menu Flow

```
[1] New Scan      → enter target IP/domain → select tools → AI analyzes → saved to DB
[2] View History  → browse past scans → view/edit/delete/export
[3] Exit
```

### New Scan — Tool Selection

```
[1] nmap
[2] whois
[3] whatweb
[4] curl headers
[5] dig DNS
[6] nikto
[a] Run all (except nikto)
[n] Run all + nikto (slow, thorough)
```

### Exporting Reports

From **View History → select scan → export**:
- `PDF` — professional vulnerability report
- `HTML` — browser-viewable report

---

## Code Examples

### Programmatically run a scan and save to DB (`db.py` patterns)

```python
import mysql.connector

def get_db_connection():
    return mysql.connector.connect(
        host="localhost",
        user="metatron",
        password="123",
        database="metatron"
    )

def create_scan_session(target: str) -> int:
    """Create a new history entry, return sl_no."""
    from datetime import datetime
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO history (target, scan_date, status) VALUES (%s, %s, %s)",
        (target, datetime.now(), "active")
    )
    conn.commit()
    sl_no = cursor.lastrowid
    cursor.close()
    conn.close()
    return sl_no

def save_vulnerability(sl_no: int, vuln_name: str, severity: str,
                       port: str, service: str, description: str) -> int:
    """Save a vulnerability finding, return vuln id."""
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO vulnerabilities
           (sl_no, vuln_name, severity, port, service, description)
           VALUES (%s, %s, %s, %s, %s, %s)""",
        (sl_no, vuln_name, severity, port, service, description)
    )
    conn.commit()
    vuln_id = cursor.lastrowid
    cursor.close()
    conn.close()
    return vuln_id

def save_fix(sl_no: int, vuln_id: int, fix_text: str, source: str = "AI"):
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO fixes (sl_no, vuln_id, fix_text, source) VALUES (%s, %s, %s, %s)",
        (sl_no, vuln_id, fix_text, source)
    )
    conn.commit()
    cursor.close()
    conn.close()

def save_summary(sl_no: int, raw_scan: str, ai_analysis: str, risk_level: str):
    from datetime import datetime
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO summary (sl_no, raw_scan, ai_analysis, risk_level, generated_at)
           VALUES (%s, %s, %s, %s, %s)""",
        (sl_no, raw_scan, ai_analysis, risk_level, datetime.now())
    )
    conn.commit()
    cursor.close()
    conn.close()

def get_scan_history():
    """Retrieve all scan sessions."""
    conn = get_db_connection()
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM history ORDER BY scan_date DESC")
    rows = cursor.fetchall()
    cursor.close()
    conn.close()
    return rows

def get_vulnerabilities_for_scan(sl_no: int):
    conn = get_db_connection()
    cursor = conn.cursor(dictionary=True)
    cursor.execute(
        "SELECT * FROM vulnerabilities WHERE sl_no = %s", (sl_no,)
    )
    rows = cursor.fetchall()
    cursor.close()
    conn.close()
    return rows
```

### Running recon tools (`tools.py` patterns)

```python
import subprocess

def run_nmap(target: str) -> str:
    """Run nmap service/version scan."""
    result = subprocess.run(
        ["nmap", "-sV", "-sC", "-T4", target],
        capture_output=True, text=True, timeout=120
    )
    return result.stdout + result.stderr

def run_whois(target: str) -> str:
    result = subprocess.run(
        ["whois", target],
        capture_output=True, text=True, timeout=30
    )
    return result.stdout

def run_whatweb(target: str) -> str:
    result = subprocess.run(
        ["whatweb", "-a", "3", target],
        capture_output=True, text=True, timeout=60
    )
    return result.stdout

def run_curl_headers(target: str) -> str:
    result = subprocess.run(
        ["curl", "-I", "-L", "--max-time", "15", target],
        capture_output=True, text=True, timeout=20
    )
    return result.stdout

def run_dig(target: str) -> str:
    result = subprocess.run(
        ["dig", target, "ANY"],
        capture_output=True, text=True, timeout=15
    )
    return result.stdout

def run_nikto(target: str) -> str:
    """Slow but thorough web scanner."""
    result = subprocess.run(
        ["nikto", "-h", target],
        capture_output=True, text=True, timeout=300
    )
    return result.stdout

def run_selected_tools(target: str, selections: list) -> dict:
    """
    selections: list of tool names, e.g. ['nmap', 'whois', 'dig']
    Returns dict of {tool_name: output}
    """
    tool_map = {
        'nmap': run_nmap,
        'whois': run_whois,
        'whatweb': run_whatweb,
        'curl': run_curl_headers,
        'dig': run_dig,
        'nikto': run_nikto,
    }
    results = {}
    for tool in selections:
        if tool in tool_map:
            print(f"[*] Running {tool} on {target}...")
            try:
                results[tool] = tool_map[tool](target)
            except subprocess.TimeoutExpired:
                results[tool] = f"[TIMEOUT] {tool} timed out"
            except Exception as e:
                results[tool] = f"[ERROR] {tool}: {e}"
    return results
```

### Querying Ollama LLM (`llm.py` patterns)

```python
import requests
import json

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL_NAME = "metatron-qwen"

def query_llm(prompt: str, stream: bool = True) -> str:
    """Send prompt to metatron-qwen, return full response."""
    payload = {
        "model": MODEL_NAME,
        "prompt": prompt,
        "stream": stream
    }
    response = requests.post(OLLAMA_URL, json=payload, stream=stream)
    
    if not stream:
        return response.json().get("response", "")
    
    full_response = ""
    for line in response.iter_lines():
        if line:
            chunk = json.loads(line)
            token = chunk.get("response", "")
            print(token, end="", flush=True)
            full_response += token
            if chunk.get("done"):
                break
    print()
    return full_response

def build_pentest_prompt(target: str, scan_results: dict) -> str:
    """Build the analysis prompt from scan results."""
    combined = "\n\n".join(
        f"=== {tool.upper()} ===\n{output}"
        for tool, output in scan_results.items()
    )
    return f"""You are an expert penetration tester analyzing scan results for: {target}

SCAN RESULTS:
{combined}

Provide a structured analysis covering:
1. VULNERABILITIES FOUND — name, severity (Critical/High/Medium/Low), port, service, description
2. EXPLOIT SUGGESTIONS — specific tools or techniques for each vulnerability
3. RECOMMENDED FIXES — actionable remediation steps
4. OVERALL RISK LEVEL — Critical / High / Medium / Low

Format vulnerabilities as:
VULN: <name> | SEVERITY: <level> | PORT: <port> | SERVICE: <service>
DESC: <description>
FIX: <remediation>
"""

def analyze_target(target: str, scan_results: dict) -> str:
    prompt = build_pentest_prompt(target, scan_results)
    print("\n[🤖] metatron-qwen analyzing...\n")
    return query_llm(prompt)
```

### DuckDuckGo search and CVE lookup (`search.py` patterns)

```python
from duckduckgo_search import DDGS

def search_exploits(query: str, max_results: int = 5) -> list:
    """Search DuckDuckGo for exploit info — no API key needed."""
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=max_results))
    return results

def lookup_cve(cve_id: str) -> list:
    """Look up a CVE identifier."""
    query = f"{cve_id} vulnerability exploit details"
    return search_exploits(query)

def search_service_vulns(service: str, version: str) -> list:
    query = f"{service} {version} known vulnerabilities CVE exploit"
    return search_exploits(query)

# Usage example:
# results = lookup_cve("CVE-2021-44228")
# results = search_service_vulns("Apache", "2.4.49")
```

### Full scan pipeline (end-to-end)

```python
from tools import run_selected_tools
from llm import analyze_target
from db import (create_scan_session, save_vulnerability,
                save_fix, save_summary)

def run_full_scan(target: str, tools: list = None):
    if tools is None:
        tools = ['nmap', 'whois', 'whatweb', 'curl', 'dig']
    
    # 1. Create DB session
    sl_no = create_scan_session(target)
    print(f"[+] Scan session #{sl_no} created for {target}")
    
    # 2. Run recon
    scan_results = run_selected_tools(target, tools)
    raw_scan = "\n\n".join(f"{k}:\n{v}" for k, v in scan_results.items())
    
    # 3. AI analysis
    ai_output = analyze_target(target, scan_results)
    
    # 4. Parse and save (simplified — real parser in llm.py)
    # Save summary
    save_summary(sl_no, raw_scan, ai_output, risk_level="High")
    
    print(f"\n[✓] Results saved to database (sl_no={sl_no})")
    return sl_no, ai_output

# Run it:
# sl_no, analysis = run_full_scan("192.168.1.1", ['nmap', 'whois'])
```

---

## Common Patterns

### Check if Ollama model is running before scan

```python
import requests

def check_ollama_ready(model: str = "metatron-qwen") -> bool:
    try:
        resp = requests.get("http://localhost:11434/api/tags", timeout=5)
        models = [m["name"] for m in resp.json().get("models", [])]
        return any(model in m for m in models)
    except Exception:
        return False

if not check_ollama_ready():
    print("[!] metatron-qwen not found. Run: ollama run metatron-qwen")
    exit(1)
```

### Query scan history

```python
from db import get_db_connection

def get_full_scan_report(sl_no: int) -> dict:
    conn = get_db_connection()
    cursor = conn.cursor(dictionary=True)
    
    cursor.execute("SELECT * FROM history WHERE sl_no = %s", (sl_no,))
    history = cursor.fetchone()
    
    cursor.execute("SELECT * FROM vulnerabilities WHERE sl_no = %s", (sl_no,))
    vulns = cursor.fetchall()
    
    cursor.execute("SELECT * FROM summary WHERE sl_no = %s", (sl_no,))
    summary = cursor.fetchone()
    
    cursor.close()
    conn.close()
    
    return {"history": history, "vulnerabilities": vulns, "summary": summary}
```

### Add a custom recon tool

```python
# In tools.py — add your tool function:
def run_gobuster(target: str, wordlist: str = "/usr/share/wordlists/dirb/common.txt") -> str:
    result = subprocess.run(
        ["gobuster", "dir", "-u", f"http://{target}", "-w", wordlist],
        capture_output=True, text=True, timeout=180
    )
    return result.stdout

# Register it in the tool_map in run_selected_tools():
tool_map['gobuster'] = run_gobuster
```

---

## Troubleshooting

### `metatron-qwen` not found / connection refused
```bash
# Terminal 1: ensure model is loaded
ollama run metatron-qwen
# Should show >>> prompt

# Verify Ollama API is reachable
curl http://localhost:11434/api/tags
```

### Out of memory when running 9b model
```bash
# Switch to 4b: edit Modelfile first line:
# FROM huihui_ai/qwen3.5-abliterated:4b
ollama create metatron-qwen -f Modelfile
```

### MariaDB connection error
```bash
sudo systemctl status mariadb
sudo systemctl start mariadb

# Verify credentials work:
mysql -u metatron -p123 metatron -e "SHOW TABLES;"
```

### `mysql.connector` not found
```bash
source venv/bin/activate
pip install mysql-connector-python
```

### nmap requires root for SYN scan
```bash
sudo nmap -sV -sC -T4 <target>
# Or use TCP connect scan (no root needed):
nmap -sT -sV <target>
```

### Nikto timeout
Nikto is slow by design. Either use `[a]` (all without nikto) or increase the subprocess timeout in `tools.py`:
```python
result = subprocess.run(["nikto", "-h", target],
                        capture_output=True, text=True,
                        timeout=600)  # 10 minutes
```

### Slow AI responses
The 9b model needs time to load. If response is slow after the first query, it's still loading. Subsequent queries will be faster. Ensure no other GPU/memory-heavy processes are running.

---

## Configuration Reference

| Setting | Location | Default | Notes |
|---|---|---|---|
| DB host | `db.py` | `localhost` | Change for remote DB |
| DB user | `db.py` | `metatron` | Match SQL user created |
| DB password | `db.py` | `123` | Change in production |
| DB name | `db.py` | `metatron` | |
| Ollama URL | `llm.py` | `http://localhost:11434` | |
| Model name | `llm.py` | `metatron-qwen` | Must match `ollama list` |
| Context window | `Modelfile` | `16384` | Increase for large scans |
| Temperature | `Modelfile` | `0.7` | Lower = more deterministic |

> **Security note:** For production use, replace the hardcoded DB password with an environment variable: `os.environ.get("METATRON_DB_PASSWORD", "123")`

---

## Legal Disclaimer

METATRON is for **educational purposes and authorized penetration testing only**. Only scan systems you own or have explicit written permission to test. Unauthorized scanning is illegal.