chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
+216
@@ -0,0 +1,216 @@
|
||||
reports/
|
||||
exports/
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[codz]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py.cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
cover/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
.pybuilder/
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
# For a library or package, you might want to ignore these files since the code is
|
||||
# intended to run in multiple environments; otherwise, check them in:
|
||||
# .python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# UV
|
||||
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
#uv.lock
|
||||
|
||||
# poetry
|
||||
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
|
||||
#poetry.lock
|
||||
#poetry.toml
|
||||
|
||||
# pdm
|
||||
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
|
||||
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
|
||||
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
|
||||
#pdm.lock
|
||||
#pdm.toml
|
||||
.pdm-python
|
||||
.pdm-build/
|
||||
|
||||
# pixi
|
||||
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
|
||||
#pixi.lock
|
||||
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
|
||||
# in the .venv directory. It is recommended not to include this directory in version control.
|
||||
.pixi
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.envrc
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
|
||||
# pytype static type analyzer
|
||||
.pytype/
|
||||
|
||||
# Cython debug symbols
|
||||
cython_debug/
|
||||
|
||||
# PyCharm
|
||||
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
|
||||
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
|
||||
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||
#.idea/
|
||||
|
||||
# Abstra
|
||||
# Abstra is an AI-powered process automation framework.
|
||||
# Ignore directories containing user credentials, local state, and settings.
|
||||
# Learn more at https://abstra.io/docs
|
||||
.abstra/
|
||||
|
||||
# Visual Studio Code
|
||||
# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
|
||||
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
|
||||
# and can be added to the global gitignore or merged into this file. However, if you prefer,
|
||||
# you could uncomment the following to ignore the entire vscode folder
|
||||
# .vscode/
|
||||
|
||||
# Ruff stuff:
|
||||
.ruff_cache/
|
||||
|
||||
# PyPI configuration file
|
||||
.pypirc
|
||||
|
||||
# Cursor
|
||||
# Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to
|
||||
# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
|
||||
# refer to https://docs.cursor.com/context/ignore-files
|
||||
.cursorignore
|
||||
.cursorindexingignore
|
||||
|
||||
# Marimo
|
||||
marimo/_static/
|
||||
marimo/_lsp/
|
||||
__marimo__/
|
||||
reports/
|
||||
exports/
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2026 sooryathejas
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,5 @@
|
||||
FROM huihui_ai/qwen3.5-abliterated:9b
|
||||
PARAMETER num_ctx 16384
|
||||
PARAMETER temperature 0.7
|
||||
PARAMETER top_k 10
|
||||
PARAMETER top_p 0.9
|
||||
@@ -0,0 +1,355 @@
|
||||
# METATRON
|
||||
AI-powered penetration testing assistant using local LLM on linux (Parrot OS)
|
||||
# 🔱 METATRON
|
||||
### AI-Powered Penetration Testing Assistant
|
||||
|
||||
<p align="center">
|
||||
<img src="screenshots/banner.png" alt="Metatron Banner" width="800"/>
|
||||
</p>
|
||||
|
||||
<p align="center">
|
||||
<img src="https://img.shields.io/badge/Python-3.x-blue?style=for-the-badge&logo=python"/>
|
||||
<img src="https://img.shields.io/badge/OS-Parrot%20Linux-green?style=for-the-badge&logo=linux"/>
|
||||
<img src="https://img.shields.io/badge/AI-metatron--qwen-red?style=for-the-badge"/>
|
||||
<img src="https://img.shields.io/badge/DB-MariaDB-orange?style=for-the-badge&logo=mariadb"/>
|
||||
<img src="https://img.shields.io/badge/License-MIT-yellow?style=for-the-badge"/>
|
||||
</p>
|
||||
|
||||
---
|
||||
|
||||
## 📌 What is Metatron?
|
||||
|
||||
**Metatron** is a CLI-based AI penetration testing assistant that runs entirely on your local machine — no cloud, no API keys, no subscriptions.
|
||||
|
||||
You give it a target IP or domain. It runs real recon tools (nmap, whois, whatweb, curl, dig, nikto), feeds all results to a locally running AI model, and the AI analyzes the target, identifies vulnerabilities, suggests exploits, and recommends fixes. Everything gets saved to a MariaDB database with full scan history.
|
||||
|
||||
---
|
||||
|
||||
## ✨ Features
|
||||
|
||||
- 🤖 **Local AI Analysis** — powered by `metatron-qwen` via Ollama, runs 100% offline
|
||||
- 🔍 **Automated Recon** — nmap, whois, whatweb, curl headers, dig DNS, nikto
|
||||
- 🌐 **Web Search** — DuckDuckGo search + CVE lookup (no API key needed)
|
||||
- 🗄️ **MariaDB Backend** — full scan history with 5 linked tables
|
||||
- ✏️ **Edit / Delete** — modify any saved result directly from the CLI
|
||||
- 🔁 **Agentic Loop** — AI can request more tool runs mid-analysis
|
||||
- 🚫 **No API Keys** — everything is free and local
|
||||
-📤 Export Reports
|
||||
|
||||
Metatron allows you to export scan results into clean, shareable report formats by selecting '2.view history'->select slno and export
|
||||
|
||||
📄 PDF — professional vulnerability reports
|
||||
🌐 HTML — browser-viewable reports
|
||||
---
|
||||
|
||||
## 🖥️ Screenshots
|
||||
|
||||
<p align="center">
|
||||
<img src="screenshots/main_menu.png" alt="Main Menu" width="700"/>
|
||||
<br><i>Main Menu</i>
|
||||
</p>
|
||||
|
||||
<p align="center">
|
||||
<img src="screenshots/scan_running.png" alt="Scan Running" width="700"/>
|
||||
<br><i>Recon tools running on target</i>
|
||||
</p>
|
||||
|
||||
<p align="center">
|
||||
<img src="screenshots/ai_analysis.png" alt="AI Analysis" width="700"/>
|
||||
<br><i>metatron-qwen analyzing scan results</i>
|
||||
</p>
|
||||
|
||||
<p align="center">
|
||||
<img src="screenshots/results.png" alt="Results" width="700"/>
|
||||
<br><i>Vulnerabilities saved to database</i>
|
||||
</p>
|
||||
<p align="center"> <img src="screenshots/export_menu.png" alt="Export Menu" width="700"/> <br><i>Export scan results as PDF and or HTML</i> </p>
|
||||
---
|
||||
|
||||
## 🧱 Tech Stack
|
||||
|
||||
| Component | Technology |
|
||||
|------------|-------------------------------------|
|
||||
| Language | Python 3 |
|
||||
| AI Model | metatron-qwen (fine-tuned Qwen 3.5) |
|
||||
| Base Model | huihui_ai/qwen3.5-abliterated:9b |
|
||||
| LLM Runner | Ollama |
|
||||
| Database | MariaDB |
|
||||
| OS | Parrot OS (Debian-based) |
|
||||
| Search | DuckDuckGo (free, no key) |
|
||||
|
||||
---
|
||||
|
||||
## ⚙️ Installation
|
||||
|
||||
### 1. Clone the repository
|
||||
|
||||
```bash
|
||||
git clone https://github.com/sooryathejas/METATRON.git
|
||||
cd METATRON
|
||||
```
|
||||
|
||||
### 2. Create and activate virtual environment
|
||||
|
||||
```bash
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate
|
||||
```
|
||||
|
||||
### 3. Install Python dependencies
|
||||
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
### 4. Install system tools
|
||||
|
||||
```bash
|
||||
sudo apt install nmap whois whatweb curl dnsutils nikto
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🤖 AI Model Setup
|
||||
|
||||
### Step 1 — Install Ollama
|
||||
|
||||
```bash
|
||||
curl -fsSL https://ollama.com/install.sh | sh
|
||||
```
|
||||
|
||||
### Step 2 — Download the base model
|
||||
|
||||
```bash
|
||||
ollama pull huihui_ai/qwen3.5-abliterated:9b
|
||||
```
|
||||
|
||||
> ⚠️ This model requires at least 8.4 GB of RAM. If your system has less, use the 4b variant:
|
||||
> ```bash
|
||||
> ollama pull huihui_ai/qwen3.5-abliterated:4b
|
||||
> ```
|
||||
> Then edit `Modelfile` and change the FROM line to the 4b model.
|
||||
|
||||
### Step 3 — Build the custom metatron-qwen model
|
||||
|
||||
The repo includes a `Modelfile` that fine-tunes the base model with pentest-specific parameters:
|
||||
|
||||
```bash
|
||||
ollama create metatron-qwen -f Modelfile
|
||||
```
|
||||
|
||||
This creates your local `metatron-qwen` model with:
|
||||
- 16,384 token context window
|
||||
- Temperature: 0.7
|
||||
- Top-k: 10
|
||||
- Top-p: 0.9
|
||||
|
||||
### Step 4 — Verify the model exists
|
||||
|
||||
```bash
|
||||
ollama list
|
||||
```
|
||||
|
||||
You should see `metatron-qwen` in the list.
|
||||
|
||||
---
|
||||
|
||||
## 🗄️ Database Setup
|
||||
|
||||
### Step 1 — Make sure MariaDB is running
|
||||
|
||||
```bash
|
||||
sudo systemctl start mariadb
|
||||
sudo systemctl enable mariadb
|
||||
```
|
||||
|
||||
### Step 2 — Create the database and user
|
||||
|
||||
```bash
|
||||
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;
|
||||
```
|
||||
|
||||
### Step 3 — Create the tables
|
||||
|
||||
```bash
|
||||
mysql -u metatron -p123 metatron
|
||||
```
|
||||
|
||||
```sql
|
||||
CREATE TABLE history (
|
||||
sl_no INT AUTO_INCREMENT PRIMARY KEY,
|
||||
target VARCHAR(255) NOT NULL,
|
||||
scan_date DATETIME NOT NULL,
|
||||
status VARCHAR(50) DEFAULT 'active'
|
||||
);
|
||||
|
||||
CREATE TABLE vulnerabilities (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
sl_no INT,
|
||||
vuln_name TEXT,
|
||||
severity VARCHAR(50),
|
||||
port VARCHAR(20),
|
||||
service VARCHAR(100),
|
||||
description TEXT,
|
||||
FOREIGN KEY (sl_no) REFERENCES history(sl_no)
|
||||
);
|
||||
|
||||
CREATE TABLE fixes (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
sl_no INT,
|
||||
vuln_id INT,
|
||||
fix_text TEXT,
|
||||
source VARCHAR(50),
|
||||
FOREIGN KEY (sl_no) REFERENCES history(sl_no),
|
||||
FOREIGN KEY (vuln_id) REFERENCES vulnerabilities(id)
|
||||
);
|
||||
|
||||
CREATE TABLE exploits_attempted (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
sl_no INT,
|
||||
exploit_name TEXT,
|
||||
tool_used TEXT,
|
||||
payload LONGTEXT,
|
||||
result TEXT,
|
||||
notes TEXT,
|
||||
FOREIGN KEY (sl_no) REFERENCES history(sl_no)
|
||||
);
|
||||
|
||||
CREATE TABLE summary (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
sl_no INT,
|
||||
raw_scan LONGTEXT,
|
||||
ai_analysis LONGTEXT,
|
||||
risk_level VARCHAR(50),
|
||||
generated_at DATETIME,
|
||||
FOREIGN KEY (sl_no) REFERENCES history(sl_no)
|
||||
);
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🚀 Usage
|
||||
|
||||
Metatron needs **two terminal tabs** to run.
|
||||
|
||||
### Terminal 1 — Load the AI model
|
||||
|
||||
```bash
|
||||
ollama run metatron-qwen
|
||||
```
|
||||
|
||||
Wait until you see the `>>>` prompt. This means the model is loaded into memory and ready. You can leave this terminal running in the background.
|
||||
|
||||
### Terminal 2 — Launch Metatron
|
||||
|
||||
```bash
|
||||
cd ~/METATRON
|
||||
source venv/bin/activate
|
||||
python metatron.py
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Walkthrough
|
||||
|
||||
**1. Main menu appears:**
|
||||
```
|
||||
[1] New Scan
|
||||
[2] View History
|
||||
[3] Exit
|
||||
```
|
||||
|
||||
**2. Select [1] New Scan → enter your target:**
|
||||
```
|
||||
[?] Enter target IP or domain: 192.168.1.1
|
||||
```
|
||||
or
|
||||
```
|
||||
[?] Enter target IP or domain: example.com
|
||||
```
|
||||
|
||||
**3. Select recon tools to run:**
|
||||
```
|
||||
[1] nmap
|
||||
[2] whois
|
||||
[3] whatweb
|
||||
[4] curl headers
|
||||
[5] dig DNS
|
||||
[6] nikto
|
||||
[a] Run all (except nikto)
|
||||
[n] Run all + nikto (slow)
|
||||
```
|
||||
|
||||
**4. Metatron runs the tools, feeds results to the AI, and prints the analysis.**
|
||||
|
||||
**5. Everything is saved to MariaDB automatically.**
|
||||
|
||||
**6. After the scan you can edit or delete any result.**
|
||||
|
||||
---
|
||||
|
||||
## 📁 Project Structure
|
||||
|
||||
```
|
||||
METATRON/
|
||||
├── metatron.py ← main CLI entry point
|
||||
├── db.py ← MariaDB connection and all CRUD operations
|
||||
├── tools.py ← recon tool runners (nmap, whois, etc.)
|
||||
├── llm.py ← Ollama interface and AI tool dispatch loop
|
||||
├── search.py ← DuckDuckGo web search and CVE lookup
|
||||
├── Modelfile ← custom model config for metatron-qwen
|
||||
├── requirements.txt ← Python dependencies
|
||||
├── .gitignore ← excludes venv, pycache, db files
|
||||
├── LICENSE ← MIT License
|
||||
├── README.md ← this file
|
||||
└── screenshots/ ← terminal screenshots for documentation
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🗃️ Database Schema
|
||||
|
||||
All 5 tables are linked by `sl_no` (session number) from the `history` table:
|
||||
|
||||
```
|
||||
history ← one row per scan session (sl_no is the spine)
|
||||
│
|
||||
├── vulnerabilities ← vulns found, linked by sl_no
|
||||
│ │
|
||||
│ └── fixes ← fixes per vuln, linked by vuln_id + sl_no
|
||||
│
|
||||
├── exploits_attempted ← exploits tried, linked by sl_no
|
||||
│
|
||||
└── summary ← full AI analysis dump, linked by sl_no
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## ⚠️ Disclaimer
|
||||
|
||||
This tool is intended for **educational purposes and authorized penetration testing only**.
|
||||
|
||||
- Only use Metatron on systems you own or have **explicit written permission** to test.
|
||||
- Unauthorized scanning or exploitation of systems is **illegal**.
|
||||
- The author is not responsible for any misuse of this tool.
|
||||
|
||||
---
|
||||
|
||||
## 👤 Author
|
||||
|
||||
**Soorya Thejas**
|
||||
- GitHub: [@sooryathejas](https://github.com/sooryathejas)
|
||||
|
||||
---
|
||||
|
||||
## 📄 License
|
||||
|
||||
This project is licensed under the MIT License — see the [LICENSE](LICENSE) file for details.
|
||||
@@ -0,0 +1,7 @@
|
||||
# WeHub 来源说明
|
||||
|
||||
- 原始项目：`sooryathejas/METATRON`
|
||||
- 原始仓库：https://github.com/sooryathejas/METATRON
|
||||
- 导入方式：上游默认分支的最新快照
|
||||
- 原作者、版权和许可证信息以原始仓库及本仓库 LICENSE 为准
|
||||
- 本文件仅用于记录来源，不代表 WeHub 是原项目作者
|
||||
@@ -0,0 +1,351 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
METATRON - db.py
|
||||
MariaDB connection + all read/write/edit/delete operations
|
||||
Database: metatron
|
||||
"""
|
||||
|
||||
import mysql.connector
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# CONNECTION
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def get_connection():
|
||||
"""Returns a MariaDB connection. No password (local setup)."""
|
||||
return mysql.connector.connect(
|
||||
host="localhost",
|
||||
user="metatron",
|
||||
password="123",
|
||||
database="metatron"
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# WRITE FUNCTIONS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def create_session(target: str) -> int:
|
||||
"""Insert new row into history. Returns sl_no."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
c.execute(
|
||||
"INSERT INTO history (target, scan_date, status) VALUES (%s, %s, %s)",
|
||||
(target, now, "active")
|
||||
)
|
||||
conn.commit()
|
||||
sl_no = c.lastrowid
|
||||
conn.close()
|
||||
return sl_no
|
||||
|
||||
|
||||
def save_vulnerability(sl_no: int, vuln_name: str, severity: str,
|
||||
port: str, service: str, description: str) -> int:
|
||||
"""Insert a vulnerability. Returns its id."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.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 = c.lastrowid
|
||||
conn.close()
|
||||
return vuln_id
|
||||
|
||||
|
||||
def save_fix(sl_no: int, vuln_id: int, fix_text: str, source: str = "ai"):
|
||||
"""Insert a fix linked to a vulnerability."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.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()
|
||||
conn.close()
|
||||
|
||||
|
||||
def save_exploit(sl_no, exploit_name, tool_used, payload, result, notes):
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("""
|
||||
INSERT INTO exploits_attempted
|
||||
(sl_no, exploit_name, tool_used, payload, result, notes)
|
||||
VALUES (%s, %s, %s, %s, %s, %s)
|
||||
""", (
|
||||
sl_no,
|
||||
str(exploit_name or "")[:1000],
|
||||
str(tool_used or "")[:500],
|
||||
str(payload or ""),
|
||||
str(result or "")[:2000],
|
||||
str(notes or "")
|
||||
))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def save_summary(sl_no: int, raw_scan: str, ai_analysis: str, risk_level: str):
|
||||
"""Insert the full session summary."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
c.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, now))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# READ FUNCTIONS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def get_all_history():
|
||||
"""Return all rows from history ordered by newest first."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT sl_no, target, scan_date, status FROM history ORDER BY sl_no DESC")
|
||||
rows = c.fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
def get_session(sl_no: int) -> dict:
|
||||
"""Return everything linked to a sl_no across all tables."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
|
||||
c.execute("SELECT * FROM history WHERE sl_no = %s", (sl_no,))
|
||||
history = c.fetchone()
|
||||
|
||||
c.execute("SELECT * FROM vulnerabilities WHERE sl_no = %s", (sl_no,))
|
||||
vulns = c.fetchall()
|
||||
|
||||
c.execute("SELECT * FROM fixes WHERE sl_no = %s", (sl_no,))
|
||||
fixes = c.fetchall()
|
||||
|
||||
c.execute("SELECT * FROM exploits_attempted WHERE sl_no = %s", (sl_no,))
|
||||
exploits = c.fetchall()
|
||||
|
||||
c.execute("SELECT * FROM summary WHERE sl_no = %s", (sl_no,))
|
||||
summary = c.fetchone()
|
||||
|
||||
conn.close()
|
||||
|
||||
return {
|
||||
"history": history,
|
||||
"vulns": vulns,
|
||||
"fixes": fixes,
|
||||
"exploits": exploits,
|
||||
"summary": summary
|
||||
}
|
||||
|
||||
|
||||
def get_vulnerabilities(sl_no: int):
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT * FROM vulnerabilities WHERE sl_no = %s", (sl_no,))
|
||||
rows = c.fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
def get_fixes(sl_no: int):
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT * FROM fixes WHERE sl_no = %s", (sl_no,))
|
||||
rows = c.fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
def get_exploits(sl_no: int):
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT * FROM exploits_attempted WHERE sl_no = %s", (sl_no,))
|
||||
rows = c.fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# EDIT FUNCTIONS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def edit_vulnerability(vuln_id: int, field: str, value: str):
|
||||
"""Edit a single field in vulnerabilities by id."""
|
||||
allowed = {"vuln_name", "severity", "port", "service", "description"}
|
||||
if field not in allowed:
|
||||
print(f"[!] Invalid field: {field}. Allowed: {allowed}")
|
||||
return
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute(
|
||||
f"UPDATE vulnerabilities SET {field} = %s WHERE id = %s",
|
||||
(value, vuln_id)
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] vulnerabilities.{field} updated for id={vuln_id}")
|
||||
|
||||
|
||||
def edit_fix(fix_id: int, fix_text: str):
|
||||
"""Edit the fix_text of a fix by id."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("UPDATE fixes SET fix_text = %s WHERE id = %s", (fix_text, fix_id))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] fix id={fix_id} updated.")
|
||||
|
||||
|
||||
def edit_exploit(exploit_id: int, field: str, value: str):
|
||||
"""Edit a single field in exploits_attempted by id."""
|
||||
allowed = {"exploit_name", "tool_used", "payload", "result", "notes"}
|
||||
if field not in allowed:
|
||||
print(f"[!] Invalid field: {field}. Allowed: {allowed}")
|
||||
return
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute(
|
||||
f"UPDATE exploits_attempted SET {field} = %s WHERE id = %s",
|
||||
(value, exploit_id)
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] exploits_attempted.{field} updated for id={exploit_id}")
|
||||
|
||||
|
||||
def edit_summary_risk(sl_no: int, risk_level: str):
|
||||
"""Update the risk level on a summary."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("UPDATE summary SET risk_level = %s WHERE sl_no = %s", (risk_level, sl_no))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] Summary risk_level updated for SL#{sl_no}")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# DELETE FUNCTIONS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def delete_vulnerability(vuln_id: int):
|
||||
"""Delete a single vulnerability and its linked fixes."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("DELETE FROM fixes WHERE vuln_id = %s", (vuln_id,))
|
||||
c.execute("DELETE FROM vulnerabilities WHERE id = %s", (vuln_id,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] Vulnerability id={vuln_id} and its fixes deleted.")
|
||||
|
||||
|
||||
def delete_exploit(exploit_id: int):
|
||||
"""Delete a single exploit attempt."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("DELETE FROM exploits_attempted WHERE id = %s", (exploit_id,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] Exploit id={exploit_id} deleted.")
|
||||
|
||||
|
||||
def delete_fix(fix_id: int):
|
||||
"""Delete a single fix."""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("DELETE FROM fixes WHERE id = %s", (fix_id,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] Fix id={fix_id} deleted.")
|
||||
|
||||
|
||||
def delete_full_session(sl_no: int):
|
||||
"""
|
||||
Wipe everything linked to a sl_no across all 5 tables.
|
||||
Order matters — delete children before parent (FK constraints).
|
||||
"""
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("DELETE FROM fixes WHERE sl_no = %s", (sl_no,))
|
||||
c.execute("DELETE FROM exploits_attempted WHERE sl_no = %s", (sl_no,))
|
||||
c.execute("DELETE FROM vulnerabilities WHERE sl_no = %s", (sl_no,))
|
||||
c.execute("DELETE FROM summary WHERE sl_no = %s", (sl_no,))
|
||||
c.execute("DELETE FROM history WHERE sl_no = %s", (sl_no,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"[+] Full session SL#{sl_no} deleted from all tables.")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# DISPLAY HELPERS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def print_history(rows):
|
||||
print("\n" + "─"*65)
|
||||
print(f"{'SL#':<6} {'TARGET':<28} {'DATE':<22} {'STATUS'}")
|
||||
print("─"*65)
|
||||
for row in rows:
|
||||
print(f"{row[0]:<6} {row[1]:<28} {str(row[2]):<22} {row[3]}")
|
||||
print()
|
||||
|
||||
|
||||
def print_session(data: dict):
|
||||
h = data["history"]
|
||||
print(f"\n{'═'*60}")
|
||||
print(f" SL# {h[0]} | Target: {h[1]} | {h[2]} | {h[3]}")
|
||||
print(f"{'═'*60}")
|
||||
|
||||
print("\n[ VULNERABILITIES ]")
|
||||
if data["vulns"]:
|
||||
for v in data["vulns"]:
|
||||
print(f" id={v[0]} | {v[2]} | Severity: {v[3]} | Port: {v[4]} | Service: {v[5]}")
|
||||
print(f" {v[6]}")
|
||||
else:
|
||||
print(" None recorded.")
|
||||
|
||||
print("\n[ FIXES ]")
|
||||
if data["fixes"]:
|
||||
for f in data["fixes"]:
|
||||
print(f" id={f[0]} | vuln_id={f[2]} | [{f[4]}] {f[3]}")
|
||||
else:
|
||||
print(" None recorded.")
|
||||
|
||||
print("\n[ EXPLOITS ATTEMPTED ]")
|
||||
if data["exploits"]:
|
||||
for e in data["exploits"]:
|
||||
print(f" id={e[0]} | {e[2]} | Tool: {e[3]} | Result: {e[5]}")
|
||||
print(f" Payload: {e[4]}")
|
||||
print(f" Notes: {e[6]}")
|
||||
else:
|
||||
print(" None recorded.")
|
||||
|
||||
print("\n[ SUMMARY ]")
|
||||
if data["summary"]:
|
||||
s = data["summary"]
|
||||
print(f" Risk Level : {s[4]}")
|
||||
print(f" Generated : {s[5]}")
|
||||
print(f"\n AI Analysis:\n {s[3][:500]}{'...' if len(str(s[3])) > 500 else ''}")
|
||||
else:
|
||||
print(" None recorded.")
|
||||
print()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# QUICK CONNECTION TEST
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
conn = get_connection()
|
||||
print("[+] MariaDB connection successful.")
|
||||
print("[+] Database: metatron")
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
print(f"[!] Connection failed: {e}")
|
||||
@@ -0,0 +1,421 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import datetime
|
||||
import mysql.connector
|
||||
from reportlab.lib.pagesizes import A4
|
||||
from reportlab.lib import colors
|
||||
from reportlab.lib.units import mm
|
||||
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
|
||||
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, HRFlowable
|
||||
from reportlab.lib.enums import TA_CENTER
|
||||
|
||||
SEVERITY_COLORS = {
|
||||
"critical": "#c0392b",
|
||||
"high": "#e67e22",
|
||||
"medium": "#f1c40f",
|
||||
"low": "#27ae60",
|
||||
"unknown": "#7f8c8d",
|
||||
}
|
||||
|
||||
RISK_COLORS = {
|
||||
"CRITICAL": "#c0392b",
|
||||
"HIGH": "#e67e22",
|
||||
"MEDIUM": "#f1c40f",
|
||||
"LOW": "#27ae60",
|
||||
"UNKNOWN": "#7f8c8d",
|
||||
}
|
||||
|
||||
|
||||
def get_connection():
|
||||
return mysql.connector.connect(
|
||||
host="localhost",
|
||||
user="metatron",
|
||||
password="123",
|
||||
database="metatron"
|
||||
)
|
||||
|
||||
|
||||
def fetch_session(sl_no: int) -> dict:
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT * FROM history WHERE sl_no = %s", (sl_no,))
|
||||
history = c.fetchone()
|
||||
c.execute("SELECT * FROM vulnerabilities WHERE sl_no = %s", (sl_no,))
|
||||
vulns = c.fetchall()
|
||||
c.execute("SELECT * FROM fixes WHERE sl_no = %s", (sl_no,))
|
||||
fixes = c.fetchall()
|
||||
c.execute("SELECT * FROM exploits_attempted WHERE sl_no = %s", (sl_no,))
|
||||
exploits = c.fetchall()
|
||||
c.execute("SELECT * FROM summary WHERE sl_no = %s", (sl_no,))
|
||||
summary = c.fetchone()
|
||||
conn.close()
|
||||
return {"history": history, "vulns": vulns, "fixes": fixes,
|
||||
"exploits": exploits, "summary": summary}
|
||||
|
||||
|
||||
def fetch_all_history():
|
||||
conn = get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT sl_no, target, scan_date, status FROM history ORDER BY sl_no DESC")
|
||||
rows = c.fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
def export_pdf(data: dict, output_dir: str) -> str:
|
||||
h = data["history"]
|
||||
sl = h[0]
|
||||
tgt = h[1]
|
||||
date = str(h[2])
|
||||
risk = data["summary"][4] if data["summary"] else "UNKNOWN"
|
||||
ai = data["summary"][3] if data["summary"] else ""
|
||||
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
safe = tgt.replace("https://","").replace("http://","").replace("/","_").replace(".","_")
|
||||
filename = os.path.join(output_dir, f"metatron_SL{sl}_{safe}.pdf")
|
||||
doc = SimpleDocTemplate(filename, pagesize=A4,
|
||||
topMargin=15*mm, bottomMargin=15*mm,
|
||||
leftMargin=15*mm, rightMargin=15*mm)
|
||||
|
||||
title_style = ParagraphStyle("t", fontSize=22, fontName="Helvetica-Bold",
|
||||
textColor=colors.HexColor("#c0392b"), spaceAfter=4)
|
||||
sub_style = ParagraphStyle("s", fontSize=10, fontName="Helvetica",
|
||||
textColor=colors.HexColor("#555555"), spaceAfter=2)
|
||||
h1_style = ParagraphStyle("h1", fontSize=13, fontName="Helvetica-Bold",
|
||||
textColor=colors.HexColor("#2c3e50"),
|
||||
spaceBefore=10, spaceAfter=4)
|
||||
body_style = ParagraphStyle("b", fontSize=9, fontName="Helvetica",
|
||||
textColor=colors.black, leading=13)
|
||||
code_style = ParagraphStyle("c", fontSize=7.5, fontName="Courier",
|
||||
textColor=colors.HexColor("#2c3e50"),
|
||||
backColor=colors.HexColor("#f4f4f4"),
|
||||
leading=11, leftIndent=6, rightIndent=6,
|
||||
spaceBefore=2, spaceAfter=2)
|
||||
footer_style = ParagraphStyle("f", fontSize=7,
|
||||
textColor=colors.HexColor("#aaaaaa"),
|
||||
alignment=TA_CENTER)
|
||||
story = []
|
||||
|
||||
story.append(Paragraph("METATRON", title_style))
|
||||
story.append(Paragraph("AI Penetration Testing Report", sub_style))
|
||||
story.append(HRFlowable(width="100%", thickness=1.5,
|
||||
color=colors.HexColor("#c0392b"), spaceAfter=8))
|
||||
|
||||
risk_color = colors.HexColor(RISK_COLORS.get(risk.upper(), "#7f8c8d"))
|
||||
meta = [["Target", tgt], ["Scan Date", date],
|
||||
["Session", f"SL# {sl}"], ["Risk Level", risk],
|
||||
["Generated", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")]]
|
||||
mt = Table(meta, colWidths=[35*mm, 130*mm])
|
||||
mt.setStyle(TableStyle([
|
||||
("FONTNAME", (0,0), (-1,-1), "Helvetica"),
|
||||
("FONTSIZE", (0,0), (-1,-1), 9),
|
||||
("FONTNAME", (0,0), (0,-1), "Helvetica-Bold"),
|
||||
("TEXTCOLOR", (0,0), (0,-1), colors.HexColor("#2c3e50")),
|
||||
("TEXTCOLOR", (1,3), (1,3), risk_color),
|
||||
("FONTNAME", (1,3), (1,3), "Helvetica-Bold"),
|
||||
("ROWBACKGROUNDS", (0,0), (-1,-1), [colors.HexColor("#f9f9f9"), colors.white]),
|
||||
("GRID", (0,0), (-1,-1), 0.3, colors.HexColor("#dddddd")),
|
||||
("PADDING", (0,0), (-1,-1), 5),
|
||||
]))
|
||||
story.append(mt)
|
||||
story.append(Spacer(1, 10))
|
||||
|
||||
story.append(Paragraph("Vulnerabilities", h1_style))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=6))
|
||||
if data["vulns"]:
|
||||
vd = [["#", "Vulnerability", "Severity", "Port", "Service"]]
|
||||
for v in data["vulns"]:
|
||||
vd.append([str(v[0]), str(v[2] or "-"),
|
||||
str(v[3] or "-").upper(), str(v[4] or "-"), str(v[5] or "-")])
|
||||
vt = Table(vd, colWidths=[10*mm, 72*mm, 24*mm, 18*mm, 28*mm], repeatRows=1)
|
||||
vts = [
|
||||
("FONTNAME", (0,0), (-1,0), "Helvetica-Bold"),
|
||||
("FONTSIZE", (0,0), (-1,-1), 8),
|
||||
("BACKGROUND", (0,0), (-1,0), colors.HexColor("#2c3e50")),
|
||||
("TEXTCOLOR", (0,0), (-1,0), colors.white),
|
||||
("GRID", (0,0), (-1,-1), 0.3, colors.HexColor("#dddddd")),
|
||||
("PADDING", (0,0), (-1,-1), 5),
|
||||
("ROWBACKGROUNDS", (0,1), (-1,-1), [colors.HexColor("#f9f9f9"), colors.white]),
|
||||
]
|
||||
for i, v in enumerate(data["vulns"], 1):
|
||||
sc = colors.HexColor(SEVERITY_COLORS.get((v[3] or "unknown").lower(), "#7f8c8d"))
|
||||
vts.append(("TEXTCOLOR", (2,i), (2,i), sc))
|
||||
vts.append(("FONTNAME", (2,i), (2,i), "Helvetica-Bold"))
|
||||
vt.setStyle(TableStyle(vts))
|
||||
story.append(vt)
|
||||
story.append(Spacer(1, 6))
|
||||
|
||||
story.append(Paragraph("Vulnerability Details", h1_style))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=6))
|
||||
for v in data["vulns"]:
|
||||
sc = colors.HexColor(SEVERITY_COLORS.get((v[3] or "unknown").lower(), "#7f8c8d"))
|
||||
lbl = ParagraphStyle("vl", fontSize=9, fontName="Helvetica-Bold", textColor=sc)
|
||||
story.append(Paragraph(f"[{(v[3] or 'UNKNOWN').upper()}] {v[2]}", lbl))
|
||||
if v[6]:
|
||||
story.append(Paragraph(str(v[6]), body_style))
|
||||
story.append(Spacer(1, 4))
|
||||
else:
|
||||
story.append(Paragraph("No vulnerabilities recorded.", body_style))
|
||||
|
||||
story.append(Spacer(1, 6))
|
||||
story.append(Paragraph("Fixes & Mitigations", h1_style))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=6))
|
||||
if data["fixes"]:
|
||||
for f in data["fixes"]:
|
||||
story.append(Paragraph(f"Fix for vuln id={f[2]}:", body_style))
|
||||
story.append(Paragraph(str(f[3] or "-"), code_style))
|
||||
story.append(Spacer(1, 3))
|
||||
else:
|
||||
story.append(Paragraph("No fixes recorded.", body_style))
|
||||
|
||||
story.append(Spacer(1, 6))
|
||||
story.append(Paragraph("Exploits Attempted", h1_style))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=6))
|
||||
if data["exploits"]:
|
||||
ed = [["#", "Exploit", "Tool", "Result"]]
|
||||
for e in data["exploits"]:
|
||||
ed.append([str(e[0]), str(e[2] or "-")[:60],
|
||||
str(e[3] or "-")[:30], str(e[5] or "-")[:30]])
|
||||
et = Table(ed, colWidths=[10*mm, 80*mm, 40*mm, 28*mm])
|
||||
et.setStyle(TableStyle([
|
||||
("FONTNAME", (0,0), (-1,0), "Helvetica-Bold"),
|
||||
("FONTSIZE", (0,0), (-1,-1), 8),
|
||||
("BACKGROUND", (0,0), (-1,0), colors.HexColor("#2c3e50")),
|
||||
("TEXTCOLOR", (0,0), (-1,0), colors.white),
|
||||
("GRID", (0,0), (-1,-1), 0.3, colors.HexColor("#dddddd")),
|
||||
("PADDING", (0,0), (-1,-1), 5),
|
||||
("ROWBACKGROUNDS", (0,1), (-1,-1), [colors.HexColor("#f9f9f9"), colors.white]),
|
||||
]))
|
||||
story.append(et)
|
||||
else:
|
||||
story.append(Paragraph("No exploits recorded.", body_style))
|
||||
|
||||
story.append(Spacer(1, 6))
|
||||
story.append(Paragraph("AI Analysis Summary", h1_style))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=6))
|
||||
if ai:
|
||||
for line in str(ai).split("\n"):
|
||||
line = line.strip()
|
||||
if line:
|
||||
story.append(Paragraph(line, body_style))
|
||||
story.append(Spacer(1, 2))
|
||||
else:
|
||||
story.append(Paragraph("No AI analysis recorded.", body_style))
|
||||
|
||||
story.append(Spacer(1, 10))
|
||||
story.append(HRFlowable(width="100%", thickness=0.5,
|
||||
color=colors.HexColor("#dddddd"), spaceAfter=4))
|
||||
story.append(Paragraph(
|
||||
"Generated by METATRON — AI Penetration Testing Assistant | "
|
||||
"github.com/sooryathejas/METATRON | For authorized use only.",
|
||||
footer_style))
|
||||
|
||||
doc.build(story)
|
||||
return filename
|
||||
|
||||
|
||||
def export_html(data: dict, output_dir: str) -> str:
|
||||
h = data["history"]
|
||||
sl = h[0]
|
||||
tgt = h[1]
|
||||
date = str(h[2])
|
||||
risk = data["summary"][4] if data["summary"] else "UNKNOWN"
|
||||
ai = data["summary"][3] if data["summary"] else ""
|
||||
rc = RISK_COLORS.get(risk.upper(), "#7f8c8d")
|
||||
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
safe = tgt.replace("https://","").replace("http://","").replace("/","_").replace(".","_")
|
||||
filename = os.path.join(output_dir, f"metatron_SL{sl}_{safe}.html")
|
||||
vuln_rows = ""
|
||||
for v in data["vulns"]:
|
||||
sc = SEVERITY_COLORS.get((v[3] or "unknown").lower(), "#7f8c8d")
|
||||
vuln_rows += (f"<tr><td>{v[0]}</td>"
|
||||
f"<td><strong>{v[2]}</strong><br><small>{v[6] or ''}</small></td>"
|
||||
f"<td><span style='color:{sc};font-weight:bold'>"
|
||||
f"{(v[3] or 'unknown').upper()}</span></td>"
|
||||
f"<td>{v[4] or '-'}</td><td>{v[5] or '-'}</td></tr>")
|
||||
|
||||
fix_rows = ""
|
||||
for f in data["fixes"]:
|
||||
fix_rows += (f"<tr><td>{f[0]}</td><td>vuln #{f[2]}</td>"
|
||||
f"<td><code>{f[3] or '-'}</code></td>"
|
||||
f"<td>{f[4] or 'ai'}</td></tr>")
|
||||
|
||||
exp_rows = ""
|
||||
for e in data["exploits"]:
|
||||
exp_rows += (f"<tr><td>{e[0]}</td><td>{e[2] or '-'}</td>"
|
||||
f"<td>{e[3] or '-'}</td>"
|
||||
f"<td><code>{str(e[4] or '-')[:80]}</code></td>"
|
||||
f"<td>{e[5] or '-'}</td></tr>")
|
||||
|
||||
ai_html = "".join(f"<p>{line}</p>"
|
||||
for line in str(ai).split("\n") if line.strip())
|
||||
|
||||
html = f"""<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<title>Metatron Report — {tgt}</title>
|
||||
<style>
|
||||
*{{box-sizing:border-box;margin:0;padding:0}}
|
||||
body{{font-family:'Segoe UI',sans-serif;background:#0d0d0d;color:#e0e0e0;padding:30px}}
|
||||
.container{{max-width:960px;margin:auto}}
|
||||
.header{{border-left:5px solid #c0392b;padding-left:16px;margin-bottom:30px}}
|
||||
.header h1{{font-size:2.2em;color:#c0392b}}
|
||||
.header p{{color:#888;font-size:.95em}}
|
||||
.meta-grid{{display:grid;grid-template-columns:1fr 1fr;gap:12px;margin-bottom:30px}}
|
||||
.meta-card{{background:#1a1a1a;border:1px solid #333;border-radius:6px;padding:14px}}
|
||||
.meta-card .label{{font-size:.75em;color:#888;text-transform:uppercase;margin-bottom:4px}}
|
||||
.meta-card .value{{font-size:1.1em;font-weight:bold}}
|
||||
.risk{{color:{rc}}}
|
||||
section{{margin-bottom:30px}}
|
||||
section h2{{font-size:1.2em;color:#c0392b;border-bottom:1px solid #333;
|
||||
padding-bottom:8px;margin-bottom:14px}}
|
||||
table{{width:100%;border-collapse:collapse;font-size:.88em}}
|
||||
th{{background:#1e1e1e;color:#aaa;text-align:left;padding:10px;
|
||||
font-size:.8em;text-transform:uppercase;border-bottom:2px solid #333}}
|
||||
td{{padding:10px;border-bottom:1px solid #222;vertical-align:top}}
|
||||
tr:hover td{{background:#1a1a1a}}
|
||||
code{{background:#1e1e1e;padding:2px 6px;border-radius:3px;
|
||||
font-family:monospace;font-size:.85em;color:#e74c3c}}
|
||||
.ai-box{{background:#111;border:1px solid #333;border-radius:6px;
|
||||
padding:16px;font-size:.9em;line-height:1.7;color:#ccc}}
|
||||
.ai-box p{{margin-bottom:8px}}
|
||||
.footer{{text-align:center;color:#444;font-size:.78em;
|
||||
margin-top:40px;border-top:1px solid #222;padding-top:16px}}
|
||||
a{{color:#555}}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="container">
|
||||
|
||||
<div class="header">
|
||||
<h1>🔱 METATRON</h1>
|
||||
<p>AI Penetration Testing Report</p>
|
||||
</div>
|
||||
|
||||
<div class="meta-grid">
|
||||
<div class="meta-card">
|
||||
<div class="label">Target</div>
|
||||
<div class="value">{tgt}</div>
|
||||
</div>
|
||||
<div class="meta-card">
|
||||
<div class="label">Session</div>
|
||||
<div class="value">SL# {sl}</div>
|
||||
</div>
|
||||
<div class="meta-card">
|
||||
<div class="label">Scan Date</div>
|
||||
<div class="value">{date}</div>
|
||||
</div>
|
||||
<div class="meta-card">
|
||||
<div class="label">Risk Level</div>
|
||||
<div class="value risk">{risk}</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<section>
|
||||
<h2>Vulnerabilities</h2>
|
||||
{'<table><thead><tr><th>#</th><th>Vulnerability</th><th>Severity</th><th>Port</th><th>Service</th></tr></thead><tbody>' + vuln_rows + '</tbody></table>' if data["vulns"] else '<p style="color:#888">None recorded.</p>'}
|
||||
</section>
|
||||
|
||||
<section>
|
||||
<h2>Fixes & Mitigations</h2>
|
||||
{'<table><thead><tr><th>#</th><th>Vuln</th><th>Fix</th><th>Source</th></tr></thead><tbody>' + fix_rows + '</tbody></table>' if data["fixes"] else '<p style="color:#888">None recorded.</p>'}
|
||||
</section>
|
||||
|
||||
<section>
|
||||
<h2>Exploits Attempted</h2>
|
||||
{'<table><thead><tr><th>#</th><th>Exploit</th><th>Tool</th><th>Payload</th><th>Result</th></tr></thead><tbody>' + exp_rows + '</tbody></table>' if data["exploits"] else '<p style="color:#888">None recorded.</p>'}
|
||||
</section>
|
||||
|
||||
<section>
|
||||
<h2>AI Analysis Summary</h2>
|
||||
<div class="ai-box">
|
||||
{ai_html if ai_html else '<p style="color:#888">None recorded.</p>'}
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<div class="footer">
|
||||
Generated by METATRON —
|
||||
<a href="https://github.com/sooryathejas/METATRON">github.com/sooryathejas/METATRON</a>
|
||||
— For authorized use only.
|
||||
</div>
|
||||
|
||||
</div>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
with open(filename, "w") as f:
|
||||
f.write(html)
|
||||
return filename
|
||||
|
||||
|
||||
def export_menu(data: dict):
|
||||
if not data["history"]:
|
||||
print("[!] No session data to export.")
|
||||
return
|
||||
|
||||
h = data["history"]
|
||||
sl = h[0]
|
||||
tgt = h[1]
|
||||
|
||||
print(f"\n\033[33m{'─'*20} EXPORT SL#{sl} — {tgt} {'─'*20}\033[0m")
|
||||
print(" [1] PDF report")
|
||||
print(" [2] HTML report")
|
||||
print(" [3] Both")
|
||||
print(" [4] Back")
|
||||
print(f"\033[90m{'─'*60}\033[0m")
|
||||
|
||||
choice = input("\033[36mExport format: \033[0m").strip()
|
||||
output_dir = os.path.expanduser("~/METATRON/reports")
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
if choice == "1":
|
||||
p = export_pdf(data, output_dir)
|
||||
print(f"\033[92m[+] PDF saved: {p}\033[0m")
|
||||
elif choice == "2":
|
||||
p = export_html(data, output_dir)
|
||||
print(f"\033[92m[+] HTML saved: {p}\033[0m")
|
||||
elif choice == "3":
|
||||
p1 = export_pdf(data, output_dir)
|
||||
p2 = export_html(data, output_dir)
|
||||
print(f"\033[92m[+] PDF : {p1}\033[0m")
|
||||
print(f"\033[92m[+] HTML : {p2}\033[0m")
|
||||
elif choice == "4":
|
||||
return
|
||||
else:
|
||||
print("\033[93m[!] Invalid choice.\033[0m")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("\n\033[91m METATRON — Standalone Report Exporter\033[0m")
|
||||
print("\033[90m ─────────────────────────────────────\033[0m\n")
|
||||
|
||||
rows = fetch_all_history()
|
||||
if not rows:
|
||||
print("[!] No sessions found in database.")
|
||||
exit()
|
||||
|
||||
print(f"{'SL#':<6} {'TARGET':<28} {'DATE':<22} {'STATUS'}")
|
||||
print("─" * 65)
|
||||
for row in rows:
|
||||
print(f"{row[0]:<6} {row[1]:<28} {str(row[2]):<22} {row[3]}")
|
||||
print()
|
||||
|
||||
sl_input = input("\033[36mEnter SL# to export: \033[0m").strip()
|
||||
if not sl_input.isdigit():
|
||||
print("[!] Invalid SL#.")
|
||||
exit()
|
||||
|
||||
data = fetch_session(int(sl_input))
|
||||
if not data["history"]:
|
||||
print(f"[!] SL# {sl_input} not found.")
|
||||
exit()
|
||||
|
||||
export_menu(data)
|
||||
@@ -0,0 +1,387 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
METATRON - llm.py
|
||||
Ollama interface for metatron-qwen model.
|
||||
Builds prompts, handles AI responses, runs tool dispatch loop.
|
||||
Model: metatron-qwen (fine-tuned from huihui_ai/qwen3.5-abliterated:9b)
|
||||
"""
|
||||
|
||||
import re
|
||||
import requests
|
||||
import json
|
||||
from tools import run_tool_by_command, run_nmap, run_curl_headers
|
||||
from search import handle_search_dispatch
|
||||
|
||||
OLLAMA_URL = "http://localhost:11434/api/chat"
|
||||
MODEL_NAME = "metatron-qwen"
|
||||
MAX_TOKENS = 8192
|
||||
MAX_TOOL_LOOPS = 9 # max times AI can call tools per session
|
||||
OLLAMA_TIMEOUT = 600
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# SYSTEM PROMPT
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
SYSTEM_PROMPT = """You are METATRON, an elite AI penetration testing assistant running on Parrot OS.
|
||||
You are precise, technical, and direct. No fluff.
|
||||
|
||||
You have access to real tools. To use them, write tags in your response:
|
||||
|
||||
[TOOL: nmap -sV 192.168.1.1] → runs nmap or any CLI tool
|
||||
[SEARCH: CVE-2021-44228 exploit] → searches the web via DuckDuckGo
|
||||
|
||||
Rules:
|
||||
- Always analyze scan data thoroughly before suggesting exploits
|
||||
- List vulnerabilities with: name, severity (critical/high/medium/low), port, service
|
||||
- For each vulnerability, suggest a concrete fix
|
||||
- If you need more information, use [SEARCH:] or [TOOL:]
|
||||
- Format vulnerabilities clearly so they can be saved to a database
|
||||
- Be specific about CVE IDs when you know them
|
||||
- Always give a final risk rating: CRITICAL / HIGH / MEDIUM / LOW
|
||||
|
||||
Output format for vulnerabilities (use this exactly):
|
||||
VULN: <name> | SEVERITY: <level> | PORT: <port> | SERVICE: <service>
|
||||
DESC: <description>
|
||||
FIX: <fix recommendation>
|
||||
|
||||
Output format for exploits:
|
||||
EXPLOIT: <name> | TOOL: <tool> | PAYLOAD: <payload or description>
|
||||
RESULT: <expected result>
|
||||
NOTES: <any notes>
|
||||
|
||||
End your analysis with:
|
||||
RISK_LEVEL: <CRITICAL|HIGH|MEDIUM|LOW>
|
||||
SUMMARY: <2-3 sentence overall summary>
|
||||
IMPORTANT: Never use markdown bold (**text**) or
|
||||
headers (## text). Plain text only. No exceptions.
|
||||
IMPORTANT RULES FOR ACCURACY:
|
||||
- nmap filtered or no-response means INCONCLUSIVE not vulnerable
|
||||
- Never assert a server version without seeing it in scan output
|
||||
- Never infer CVEs from guessed versions
|
||||
- curl timeouts and HTTP_CODE=000 mean the host is unreachable not exploitable
|
||||
- ab and stress tools are not Slowloris unless confirmed
|
||||
- Only assign CRITICAL if there is direct evidence of exploitability
|
||||
- If evidence is weak mark severity as LOW with note: unconfirmed"""
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# OLLAMA API CALL
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def ask_ollama(messages: list) -> str:
|
||||
try:
|
||||
payload = {
|
||||
"model": MODEL_NAME,
|
||||
"messages": messages,
|
||||
"stream": False,
|
||||
"options": {
|
||||
"num_predict": MAX_TOKENS,
|
||||
"temperature": 0.7,
|
||||
"top_p": 0.9,
|
||||
}
|
||||
}
|
||||
print(f"\n[*] Sending to {MODEL_NAME}...")
|
||||
resp = requests.post(OLLAMA_URL, json=payload, timeout=OLLAMA_TIMEOUT)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
response = data.get("message", {}).get("content", "").strip()
|
||||
if not response:
|
||||
return "[!] Model returned empty response."
|
||||
return response
|
||||
except requests.exceptions.ConnectionError:
|
||||
return "[!] Cannot connect to Ollama. Is it running? Try: ollama serve"
|
||||
except requests.exceptions.Timeout:
|
||||
return "[!] Ollama timed out. Model may be loading, try again."
|
||||
except requests.exceptions.HTTPError as e:
|
||||
return f"[!] Ollama HTTP error: {e}"
|
||||
except Exception as e:
|
||||
return f"[!] Unexpected error: {e}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# TOOL DISPATCH
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def extract_tool_calls(response: str) -> list:
|
||||
"""
|
||||
Extract all [TOOL: ...] and [SEARCH: ...] tags from AI response.
|
||||
Returns list of tuples: [("TOOL", "nmap -sV x.x.x.x"), ("SEARCH", "CVE...")]
|
||||
"""
|
||||
calls = []
|
||||
|
||||
tool_matches = re.findall(r'\[TOOL:\s*(.+?)\]', response)
|
||||
search_matches = re.findall(r'\[SEARCH:\s*(.+?)\]', response)
|
||||
|
||||
for m in tool_matches:
|
||||
calls.append(("TOOL", m.strip()))
|
||||
for m in search_matches:
|
||||
calls.append(("SEARCH", m.strip()))
|
||||
|
||||
return calls
|
||||
|
||||
def summarize_tool_output(raw_output: str) -> str:
|
||||
"""
|
||||
Compress raw tool output into security-relevant bullet points
|
||||
before injecting into the LLM context.
|
||||
Keeps context size manageable across rounds.
|
||||
"""
|
||||
if len(raw_output) < 500:
|
||||
return raw_output
|
||||
|
||||
try:
|
||||
payload = {
|
||||
"model": MODEL_NAME,
|
||||
"messages": [
|
||||
{"role": "system", "content": "You are a security data compressor. Extract only security-relevant facts. Return maximum 15 bullet points. Plain text only. No markdown."},
|
||||
{"role": "user", "content": f"Compress this tool output:\n{raw_output[:6000]}"} ],
|
||||
"stream": False,
|
||||
"options": {
|
||||
"num_predict": 512,
|
||||
"temperature": 0.2,
|
||||
"top_p": 0.9,
|
||||
}
|
||||
}
|
||||
resp = requests.post(OLLAMA_URL, json=payload, timeout=120)
|
||||
resp.raise_for_status()
|
||||
summary = resp.json().get("message", {}).get("content", "").strip()
|
||||
return summary if summary else raw_output
|
||||
except Exception:
|
||||
return raw_output
|
||||
def run_tool_calls(calls: list) -> str:
|
||||
"""
|
||||
Execute all tool/search calls and return combined results string.
|
||||
"""
|
||||
if not calls:
|
||||
return ""
|
||||
|
||||
results = ""
|
||||
for call_type, call_content in calls:
|
||||
print(f"\n [DISPATCH] {call_type}: {call_content}")
|
||||
|
||||
if call_type == "TOOL":
|
||||
output = run_tool_by_command(call_content)
|
||||
elif call_type == "SEARCH":
|
||||
output = handle_search_dispatch(call_content)
|
||||
else:
|
||||
output = f"[!] Unknown call type: {call_type}"
|
||||
|
||||
compressed = summarize_tool_output(output.strip())
|
||||
results += f"\n[{call_type} RESULT: {call_content}]\n"
|
||||
results += "─" * 40 + "\n"
|
||||
results += compressed + "\n"
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# PARSER — extract structured data from AI output
|
||||
# ─────────────────────────────────────────────
|
||||
def _clean(line: str) -> str:
|
||||
return re.sub(r'\*+', '', line).strip()
|
||||
def parse_vulnerabilities(response: str) -> list:
|
||||
"""
|
||||
Parse VULN: lines from AI response into dicts.
|
||||
Returns list of vulnerability dicts ready for db.save_vulnerability()
|
||||
"""
|
||||
vulns = []
|
||||
lines = response.splitlines()
|
||||
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = _clean(lines[i])
|
||||
if line.startswith("VULN:"):
|
||||
vuln = {
|
||||
"vuln_name": "",
|
||||
"severity": "medium",
|
||||
"port": "",
|
||||
"service": "",
|
||||
"description": "",
|
||||
"fix": ""
|
||||
}
|
||||
|
||||
# parse header line: VULN: name | SEVERITY: x | PORT: x | SERVICE: x
|
||||
parts = line.split("|")
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if part.startswith("VULN:"):
|
||||
vuln["vuln_name"] = part.replace("VULN:", "").strip()
|
||||
elif part.startswith("SEVERITY:"):
|
||||
vuln["severity"] = part.replace("SEVERITY:", "").strip().lower()
|
||||
elif part.startswith("PORT:"):
|
||||
vuln["port"] = part.replace("PORT:", "").strip()
|
||||
elif part.startswith("SERVICE:"):
|
||||
vuln["service"] = part.replace("SERVICE:", "").strip()
|
||||
|
||||
# look ahead for DESC: and FIX: lines
|
||||
j = i + 1
|
||||
while j < len(lines) and j <= i + 5:
|
||||
next_line = _clean(lines[j])
|
||||
if next_line.startswith(("VULN:", "EXPLOIT:", "RISK_LEVEL:", "SUMMARY:")):
|
||||
break
|
||||
if next_line.startswith("DESC:"):
|
||||
vuln["description"] = next_line.replace("DESC:", "").strip()
|
||||
elif next_line.startswith("FIX:"):
|
||||
vuln["fix"] = next_line.replace("FIX:", "").strip()
|
||||
j += 1
|
||||
|
||||
if vuln["vuln_name"]:
|
||||
vulns.append(vuln)
|
||||
|
||||
i += 1
|
||||
|
||||
return vulns
|
||||
|
||||
|
||||
def parse_exploits(response: str) -> list:
|
||||
"""
|
||||
Parse EXPLOIT: lines from AI response into dicts.
|
||||
Returns list of exploit dicts ready for db.save_exploit()
|
||||
"""
|
||||
exploits = []
|
||||
lines = response.splitlines()
|
||||
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = _clean(lines[i])
|
||||
if line.startswith("EXPLOIT:"):
|
||||
exploit = {
|
||||
"exploit_name": "",
|
||||
"tool_used": "",
|
||||
"payload": "",
|
||||
"result": "unknown",
|
||||
"notes": ""
|
||||
}
|
||||
|
||||
parts = line.split("|")
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if part.startswith("EXPLOIT:"):
|
||||
exploit["exploit_name"] = part.replace("EXPLOIT:", "").strip()
|
||||
elif part.startswith("TOOL:"):
|
||||
exploit["tool_used"] = part.replace("TOOL:", "").strip()
|
||||
elif part.startswith("PAYLOAD:"):
|
||||
exploit["payload"] = part.replace("PAYLOAD:", "").strip()
|
||||
|
||||
j = i + 1
|
||||
while j < len(lines) and j <= i + 4:
|
||||
next_line = _clean(lines[j])
|
||||
if next_line.startswith(("VULN:", "EXPLOIT:", "RISK_LEVEL:", "SUMMARY:")):
|
||||
break
|
||||
if next_line.startswith("RESULT:"):
|
||||
exploit["result"] = next_line.replace("RESULT:", "").strip()
|
||||
elif next_line.startswith("NOTES:"):
|
||||
exploit["notes"] = next_line.replace("NOTES:", "").strip()
|
||||
j += 1
|
||||
|
||||
if exploit["exploit_name"]:
|
||||
exploits.append(exploit)
|
||||
|
||||
i += 1
|
||||
|
||||
return exploits
|
||||
|
||||
|
||||
def parse_risk_level(response: str) -> str:
|
||||
"""Extract RISK_LEVEL from AI response."""
|
||||
match = re.search(r'RISK_LEVEL:\s*(CRITICAL|HIGH|MEDIUM|LOW)', response, re.IGNORECASE)
|
||||
return match.group(1).upper() if match else "UNKNOWN"
|
||||
|
||||
|
||||
def parse_summary(response: str) -> str:
|
||||
match = re.search(r'SUMMARY:\s*(.+)', response, re.IGNORECASE)
|
||||
return match.group(1).strip() if match else ""
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# MAIN ANALYSIS FUNCTION
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def analyse_target(target: str, raw_scan: str) -> dict:
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": SYSTEM_PROMPT
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": f"""TARGET: {target}
|
||||
|
||||
RECON DATA:
|
||||
{raw_scan}
|
||||
|
||||
Analyze this target completely. Use [TOOL:] or [SEARCH:] if you need more information.
|
||||
List all vulnerabilities, fixes, and suggest exploits where applicable."""
|
||||
}
|
||||
]
|
||||
|
||||
final_response = ""
|
||||
|
||||
for loop in range(MAX_TOOL_LOOPS):
|
||||
response = ask_ollama(messages)
|
||||
|
||||
print(f"\n{'─'*60}")
|
||||
print(f"[METATRON - Round {loop + 1}]")
|
||||
print(f"{'─'*60}")
|
||||
print(response)
|
||||
|
||||
final_response = response
|
||||
|
||||
tool_calls = extract_tool_calls(response)
|
||||
if not tool_calls:
|
||||
print("\n[*] No tool calls. Analysis complete.")
|
||||
break
|
||||
|
||||
tool_results = run_tool_calls(tool_calls)
|
||||
|
||||
# add assistant response and tool results as new messages
|
||||
messages.append({
|
||||
"role": "assistant",
|
||||
"content": response
|
||||
})
|
||||
messages.append({
|
||||
"role": "user",
|
||||
"content": f"""[TOOL RESULTS]
|
||||
{tool_results}
|
||||
|
||||
Continue your analysis with this new information.
|
||||
If analysis is complete, give the final RISK_LEVEL and SUMMARY."""
|
||||
})
|
||||
|
||||
vulnerabilities = parse_vulnerabilities(final_response)
|
||||
exploits = parse_exploits(final_response)
|
||||
risk_level = parse_risk_level(final_response)
|
||||
summary = parse_summary(final_response)
|
||||
|
||||
print(f"\n[+] Parsed: {len(vulnerabilities)} vulns, {len(exploits)} exploits | Risk: {risk_level}")
|
||||
|
||||
return {
|
||||
"full_response": final_response,
|
||||
"vulnerabilities": vulnerabilities,
|
||||
"exploits": exploits,
|
||||
"risk_level": risk_level,
|
||||
"summary": summary,
|
||||
"raw_scan": raw_scan
|
||||
}
|
||||
# ─────────────────────────────────────────────
|
||||
# QUICK TEST
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("[ llm.py test — direct AI query ]\n")
|
||||
|
||||
# test if ollama is reachable
|
||||
try:
|
||||
r = requests.get("http://localhost:11434", timeout=5)
|
||||
print("[+] Ollama is running.")
|
||||
except Exception:
|
||||
print("[!] Ollama not reachable. Run: ollama serve")
|
||||
exit(1)
|
||||
|
||||
target = input("Test target: ").strip()
|
||||
test_scan = f"Test recon for {target} — nmap and whois data would appear here."
|
||||
result = analyse_target(target, test_scan)
|
||||
|
||||
print(f"\nRisk Level : {result['risk_level']}")
|
||||
print(f"Summary : {result['summary']}")
|
||||
print(f"Vulns found: {len(result['vulnerabilities'])}")
|
||||
print(f"Exploits : {len(result['exploits'])}")
|
||||
+430
@@ -0,0 +1,430 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
METATRON - metatron.py
|
||||
Main CLI entry point. Wires db.py + tools.py + search.py + llm.py together.
|
||||
Run with: python metatron.py
|
||||
"""
|
||||
from export import export_menu
|
||||
import os
|
||||
import sys
|
||||
from db import (
|
||||
get_connection,
|
||||
create_session,
|
||||
save_vulnerability,
|
||||
save_fix,
|
||||
save_exploit,
|
||||
save_summary,
|
||||
get_all_history,
|
||||
get_session,
|
||||
get_vulnerabilities,
|
||||
get_fixes,
|
||||
get_exploits,
|
||||
edit_vulnerability,
|
||||
edit_fix,
|
||||
edit_exploit,
|
||||
edit_summary_risk,
|
||||
delete_vulnerability,
|
||||
delete_exploit,
|
||||
delete_fix,
|
||||
delete_full_session,
|
||||
print_history,
|
||||
print_session
|
||||
)
|
||||
from tools import interactive_tool_run, format_recon_for_llm, run_default_recon
|
||||
from llm import analyse_target
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# BANNER
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def banner():
|
||||
os.system("clear")
|
||||
print("""
|
||||
\033[91m
|
||||
███╗ ███╗███████╗████████╗ █████╗ ████████╗██████╗ ██████╗ ███╗ ██╗
|
||||
████╗ ████║██╔════╝╚══██╔══╝██╔══██╗╚══██╔══╝██╔══██╗██╔═══██╗████╗ ██║
|
||||
██╔████╔██║█████╗ ██║ ███████║ ██║ ██████╔╝██║ ██║██╔██╗ ██║
|
||||
██║╚██╔╝██║██╔══╝ ██║ ██╔══██║ ██║ ██╔══██╗██║ ██║██║╚██╗██║
|
||||
██║ ╚═╝ ██║███████╗ ██║ ██║ ██║ ██║ ██║ ██║╚██████╔╝██║ ╚████║
|
||||
╚═╝ ╚═╝╚══════╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝
|
||||
\033[0m
|
||||
\033[90mAI Penetration Testing Assistant | Model: metatron-qwen | Parrot OS\033[0m
|
||||
\033[90m─────────────────────────────────────────────────────────────────────\033[0m
|
||||
""")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# HELPERS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def divider(label=""):
|
||||
if label:
|
||||
print(f"\n\033[33m{'─'*20} {label} {'─'*20}\033[0m")
|
||||
else:
|
||||
print(f"\033[90m{'─'*60}\033[0m")
|
||||
|
||||
|
||||
def prompt(text):
|
||||
return input(f"\033[36m{text}\033[0m").strip()
|
||||
|
||||
|
||||
def success(text):
|
||||
print(f"\033[92m[+] {text}\033[0m")
|
||||
|
||||
|
||||
def warn(text):
|
||||
print(f"\033[93m[!] {text}\033[0m")
|
||||
|
||||
|
||||
def error(text):
|
||||
print(f"\033[91m[✗] {text}\033[0m")
|
||||
|
||||
|
||||
def info(text):
|
||||
print(f"\033[94m[*] {text}\033[0m")
|
||||
|
||||
|
||||
def confirm(question: str) -> bool:
|
||||
ans = prompt(f"{question} [y/N]: ").lower()
|
||||
return ans == "y"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# NEW SCAN
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def new_scan():
|
||||
divider("NEW SCAN")
|
||||
target = prompt("[?] Enter target IP or domain: ")
|
||||
if not target:
|
||||
warn("No target entered.")
|
||||
return
|
||||
|
||||
# check if target was scanned before
|
||||
history = get_all_history()
|
||||
past = [row for row in history if row[1] == target]
|
||||
if past:
|
||||
warn(f"Target '{target}' has been scanned before ({len(past)} time(s)).")
|
||||
if not confirm("Continue with a new scan?"):
|
||||
return
|
||||
|
||||
# create session in history table first
|
||||
sl_no = create_session(target)
|
||||
success(f"Session created — SL# {sl_no}")
|
||||
|
||||
# run recon tools
|
||||
divider("RECON")
|
||||
info("Choose recon tools to run:")
|
||||
raw_scan = interactive_tool_run(target)
|
||||
|
||||
if not raw_scan.strip():
|
||||
warn("No scan data collected. Aborting.")
|
||||
delete_full_session(sl_no)
|
||||
return
|
||||
|
||||
# send to AI
|
||||
divider("AI ANALYSIS")
|
||||
result = analyse_target(target, raw_scan)
|
||||
|
||||
# ── save everything to DB ──────────────────
|
||||
divider("SAVING TO DATABASE")
|
||||
|
||||
# save vulnerabilities and their fixes
|
||||
for vuln in result["vulnerabilities"]:
|
||||
vuln_id = save_vulnerability(
|
||||
sl_no,
|
||||
vuln["vuln_name"],
|
||||
vuln["severity"],
|
||||
vuln["port"],
|
||||
vuln["service"],
|
||||
vuln["description"]
|
||||
)
|
||||
if vuln.get("fix"):
|
||||
save_fix(sl_no, vuln_id, vuln["fix"], source="ai")
|
||||
success(f"Saved vuln: {vuln['vuln_name']} [{vuln['severity']}]")
|
||||
|
||||
# save exploits
|
||||
for exp in result["exploits"]:
|
||||
save_exploit(
|
||||
sl_no,
|
||||
exp["exploit_name"],
|
||||
exp["tool_used"],
|
||||
exp["payload"],
|
||||
exp["result"],
|
||||
exp["notes"]
|
||||
)
|
||||
success(f"Saved exploit: {exp['exploit_name']}")
|
||||
|
||||
# save summary
|
||||
save_summary(
|
||||
sl_no,
|
||||
result["raw_scan"],
|
||||
result["full_response"],
|
||||
result["risk_level"]
|
||||
)
|
||||
|
||||
success(f"All data saved. SL# {sl_no} | Risk: {result['risk_level']}")
|
||||
divider()
|
||||
|
||||
# show results and offer edit/delete
|
||||
data = get_session(sl_no)
|
||||
print_session(data)
|
||||
|
||||
if confirm("Edit or delete anything in this session?"):
|
||||
edit_delete_menu(sl_no)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# VIEW HISTORY
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def view_history():
|
||||
divider("SCAN HISTORY")
|
||||
rows = get_all_history()
|
||||
|
||||
if not rows:
|
||||
warn("No scans in database yet.")
|
||||
return
|
||||
|
||||
print_history(rows)
|
||||
|
||||
sl_no_str = prompt("Enter SL# to view details (or press Enter to go back): ")
|
||||
if not sl_no_str:
|
||||
return
|
||||
|
||||
try:
|
||||
sl_no = int(sl_no_str)
|
||||
except ValueError:
|
||||
error("Invalid SL#.")
|
||||
return
|
||||
|
||||
data = get_session(sl_no)
|
||||
if not data["history"]:
|
||||
error(f"SL# {sl_no} not found.")
|
||||
return
|
||||
|
||||
print_session(data)
|
||||
|
||||
if confirm("Export this session?"):
|
||||
export_menu(data)
|
||||
|
||||
if confirm("Edit or delete anything in this session?"):
|
||||
edit_delete_menu(sl_no)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# EDIT / DELETE MENU
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def edit_delete_menu(sl_no: int):
|
||||
while True:
|
||||
divider(f"EDIT / DELETE — SL# {sl_no}")
|
||||
print(" [1] Edit a vulnerability")
|
||||
print(" [2] Edit a fix")
|
||||
print(" [3] Edit an exploit")
|
||||
print(" [4] Edit risk level")
|
||||
print(" [5] Delete a vulnerability")
|
||||
print(" [6] Delete a fix")
|
||||
print(" [7] Delete an exploit")
|
||||
print(" [8] Delete FULL session (all tables)")
|
||||
print(" [9] Back")
|
||||
divider()
|
||||
|
||||
choice = prompt("Choice: ")
|
||||
|
||||
# ── EDIT VULNERABILITY ─────────────────
|
||||
if choice == "1":
|
||||
vulns = get_vulnerabilities(sl_no)
|
||||
if not vulns:
|
||||
warn("No vulnerabilities recorded for this session.")
|
||||
continue
|
||||
|
||||
print("\n[ VULNERABILITIES ]")
|
||||
for v in vulns:
|
||||
print(f" id={v[0]} | {v[2]} | {v[3]} | port {v[4]} | {v[5]}")
|
||||
|
||||
vid = prompt("Enter vulnerability id to edit: ")
|
||||
if not vid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
print(" Fields: vuln_name / severity / port / service / description")
|
||||
field = prompt("Field to edit: ").strip()
|
||||
value = prompt(f"New value for '{field}': ")
|
||||
edit_vulnerability(int(vid), field, value)
|
||||
|
||||
# ── EDIT FIX ──────────────────────────
|
||||
elif choice == "2":
|
||||
fixes = get_fixes(sl_no)
|
||||
if not fixes:
|
||||
warn("No fixes recorded for this session.")
|
||||
continue
|
||||
|
||||
print("\n[ FIXES ]")
|
||||
for f in fixes:
|
||||
print(f" id={f[0]} | vuln_id={f[2]} | {f[3][:80]}")
|
||||
|
||||
fid = prompt("Enter fix id to edit: ")
|
||||
if not fid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
new_text = prompt("New fix text: ")
|
||||
edit_fix(int(fid), new_text)
|
||||
|
||||
# ── EDIT EXPLOIT ──────────────────────
|
||||
elif choice == "3":
|
||||
exploits = get_exploits(sl_no)
|
||||
if not exploits:
|
||||
warn("No exploits recorded for this session.")
|
||||
continue
|
||||
|
||||
print("\n[ EXPLOITS ]")
|
||||
for e in exploits:
|
||||
print(f" id={e[0]} | {e[2]} | tool: {e[3]} | result: {e[5]}")
|
||||
|
||||
eid = prompt("Enter exploit id to edit: ")
|
||||
if not eid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
print(" Fields: exploit_name / tool_used / payload / result / notes")
|
||||
field = prompt("Field to edit: ").strip()
|
||||
value = prompt(f"New value for '{field}': ")
|
||||
edit_exploit(int(eid), field, value)
|
||||
|
||||
# ── EDIT RISK LEVEL ───────────────────
|
||||
elif choice == "4":
|
||||
print(" Options: CRITICAL / HIGH / MEDIUM / LOW")
|
||||
risk = prompt("New risk level: ").upper()
|
||||
if risk not in ("CRITICAL", "HIGH", "MEDIUM", "LOW"):
|
||||
error("Invalid risk level.")
|
||||
continue
|
||||
edit_summary_risk(sl_no, risk)
|
||||
|
||||
# ── DELETE VULNERABILITY ──────────────
|
||||
elif choice == "5":
|
||||
vulns = get_vulnerabilities(sl_no)
|
||||
if not vulns:
|
||||
warn("No vulnerabilities to delete.")
|
||||
continue
|
||||
|
||||
print("\n[ VULNERABILITIES ]")
|
||||
for v in vulns:
|
||||
print(f" id={v[0]} | {v[2]} | {v[3]}")
|
||||
|
||||
vid = prompt("Enter vulnerability id to delete: ")
|
||||
if not vid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
if confirm(f"Delete vulnerability id={vid} and its linked fixes?"):
|
||||
delete_vulnerability(int(vid))
|
||||
|
||||
# ── DELETE FIX ────────────────────────
|
||||
elif choice == "6":
|
||||
fixes = get_fixes(sl_no)
|
||||
if not fixes:
|
||||
warn("No fixes to delete.")
|
||||
continue
|
||||
|
||||
print("\n[ FIXES ]")
|
||||
for f in fixes:
|
||||
print(f" id={f[0]} | vuln_id={f[2]} | {f[3][:80]}")
|
||||
|
||||
fid = prompt("Enter fix id to delete: ")
|
||||
if not fid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
if confirm(f"Delete fix id={fid}?"):
|
||||
delete_fix(int(fid))
|
||||
|
||||
# ── DELETE EXPLOIT ────────────────────
|
||||
elif choice == "7":
|
||||
exploits = get_exploits(sl_no)
|
||||
if not exploits:
|
||||
warn("No exploits to delete.")
|
||||
continue
|
||||
|
||||
print("\n[ EXPLOITS ]")
|
||||
for e in exploits:
|
||||
print(f" id={e[0]} | {e[2]} | result: {e[5]}")
|
||||
|
||||
eid = prompt("Enter exploit id to delete: ")
|
||||
if not eid.isdigit():
|
||||
error("Invalid id.")
|
||||
continue
|
||||
|
||||
if confirm(f"Delete exploit id={eid}?"):
|
||||
delete_exploit(int(eid))
|
||||
|
||||
# ── DELETE FULL SESSION ───────────────
|
||||
elif choice == "8":
|
||||
if confirm(f"\n\033[91mPermanently delete ENTIRE session SL# {sl_no} from all tables?\033[0m"):
|
||||
delete_full_session(sl_no)
|
||||
success(f"Session SL# {sl_no} wiped.")
|
||||
return # go back to main menu
|
||||
|
||||
# ── BACK ──────────────────────────────
|
||||
elif choice == "9":
|
||||
break
|
||||
|
||||
else:
|
||||
warn("Invalid choice.")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# DB CONNECTION CHECK
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def check_db():
|
||||
try:
|
||||
conn = get_connection()
|
||||
conn.close()
|
||||
return True
|
||||
except Exception as e:
|
||||
error(f"MariaDB connection failed: {e}")
|
||||
error("Make sure MariaDB is running: sudo systemctl start mariadb")
|
||||
return False
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# MAIN MENU
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def main_menu():
|
||||
while True:
|
||||
banner()
|
||||
print(" \033[92m[1]\033[0m New Scan")
|
||||
print(" \033[92m[2]\033[0m View History")
|
||||
print(" \033[92m[3]\033[0m Exit")
|
||||
divider()
|
||||
|
||||
choice = prompt("metatron> ")
|
||||
|
||||
if choice == "1":
|
||||
new_scan()
|
||||
input("\n\033[90mPress Enter to continue...\033[0m")
|
||||
|
||||
elif choice == "2":
|
||||
view_history()
|
||||
input("\n\033[90mPress Enter to continue...\033[0m")
|
||||
|
||||
elif choice == "3":
|
||||
print("\n\033[91m[*] Shutting down Metatron. Stay legal.\033[0m\n")
|
||||
sys.exit(0)
|
||||
|
||||
else:
|
||||
warn("Invalid choice.")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# ENTRY POINT
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
if not check_db():
|
||||
sys.exit(1)
|
||||
main_menu()
|
||||
@@ -0,0 +1,24 @@
|
||||
anyio==4.13.0
|
||||
beautifulsoup4==4.14.3
|
||||
certifi==2026.2.25
|
||||
charset-normalizer==3.4.6
|
||||
click==8.3.1
|
||||
ddgs==9.12.1
|
||||
duckduckgo_search==8.1.1
|
||||
h11==0.16.0
|
||||
httpcore==1.0.9
|
||||
httpx==0.28.1
|
||||
idna==3.11
|
||||
lxml==6.0.2
|
||||
markdown-it-py==4.0.0
|
||||
mdurl==0.1.2
|
||||
mysql-connector-python==9.6.0
|
||||
pillow==12.2.0
|
||||
primp==1.2.1
|
||||
Pygments==2.20.0
|
||||
reportlab==4.4.10
|
||||
requests==2.33.1
|
||||
rich==14.3.3
|
||||
soupsieve==2.8.3
|
||||
typing_extensions==4.15.0
|
||||
urllib3==2.6.3
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 428 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 799 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 184 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 204 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 343 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 250 KiB |
@@ -0,0 +1,175 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
METATRON - search.py
|
||||
Free web search via DuckDuckGo — no API key needed.
|
||||
Also fetches and extracts plain text from URLs.
|
||||
Used by LLM tool dispatch when AI writes [SEARCH: query]
|
||||
"""
|
||||
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from ddgs import DDGS # pip install duckduckgo-search
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# DDG SEARCH
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def web_search(query: str, max_results: int = 5) -> str:
|
||||
"""
|
||||
Search DuckDuckGo and return formatted results.
|
||||
No API key. No rate limit issues for reasonable usage.
|
||||
Returns a string ready to paste into LLM prompt.
|
||||
"""
|
||||
print(f" [*] Searching: {query}")
|
||||
try:
|
||||
with DDGS() as ddgs:
|
||||
results = list(ddgs.text(query, max_results=max_results))
|
||||
|
||||
if not results:
|
||||
return "[!] No search results found."
|
||||
|
||||
output = f"[WEB SEARCH RESULTS FOR: {query}]\n"
|
||||
output += "─" * 50 + "\n"
|
||||
for i, r in enumerate(results, 1):
|
||||
output += f"\n[{i}] {r['title']}\n"
|
||||
output += f" URL : {r['href']}\n"
|
||||
output += f" Snippet : {r['body']}\n"
|
||||
|
||||
return output
|
||||
|
||||
except Exception as e:
|
||||
return f"[!] Search failed: {e}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# CVE SPECIFIC SEARCH
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def search_cve(cve_id: str) -> str:
|
||||
"""
|
||||
Search for a specific CVE.
|
||||
Queries DDG then also hits cve.mitre.org directly.
|
||||
"""
|
||||
print(f" [*] Looking up {cve_id}...")
|
||||
|
||||
# DDG search first
|
||||
ddg_results = web_search(f"{cve_id} vulnerability exploit details", max_results=3)
|
||||
|
||||
# Direct MITRE fetch
|
||||
mitre_url = f"https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve_id}"
|
||||
mitre_data = fetch_page(mitre_url, max_chars=2000)
|
||||
|
||||
return f"{ddg_results}\n\n[MITRE CVE LOOKUP: {cve_id}]\n{mitre_data}"
|
||||
|
||||
|
||||
def search_exploit(service: str, version: str) -> str:
|
||||
"""
|
||||
Search for known exploits for a service + version combo.
|
||||
e.g. search_exploit("apache", "2.4.49")
|
||||
"""
|
||||
query = f"{service} {version} exploit CVE vulnerability 2023 2024"
|
||||
return web_search(query, max_results=5)
|
||||
|
||||
|
||||
def search_fix(vuln_name: str) -> str:
|
||||
"""
|
||||
Search for mitigation/fix for a vulnerability.
|
||||
"""
|
||||
query = f"how to fix {vuln_name} security mitigation patch"
|
||||
return web_search(query, max_results=3)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# PAGE FETCHER
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def fetch_page(url: str, max_chars: int = 3000) -> str:
|
||||
"""
|
||||
Fetch a URL and return extracted plain text.
|
||||
Strips all HTML tags. Truncated to max_chars for LLM context.
|
||||
"""
|
||||
try:
|
||||
headers = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/120.0"}
|
||||
resp = requests.get(url, headers=headers, timeout=15)
|
||||
resp.raise_for_status()
|
||||
|
||||
soup = BeautifulSoup(resp.text, "html.parser")
|
||||
|
||||
# remove nav/footer/script noise
|
||||
for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
|
||||
tag.decompose()
|
||||
|
||||
text = soup.get_text(separator="\n", strip=True)
|
||||
|
||||
# collapse blank lines
|
||||
lines = [l for l in text.splitlines() if l.strip()]
|
||||
clean = "\n".join(lines)
|
||||
|
||||
if len(clean) > max_chars:
|
||||
clean = clean[:max_chars] + f"\n... [truncated at {max_chars} chars]"
|
||||
|
||||
return clean
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
return "[!] Could not connect to URL — check network."
|
||||
except requests.exceptions.Timeout:
|
||||
return "[!] Page fetch timed out."
|
||||
except requests.exceptions.HTTPError as e:
|
||||
return f"[!] HTTP error: {e}"
|
||||
except Exception as e:
|
||||
return f"[!] Fetch failed: {e}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# TOOL DISPATCH HANDLER
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def handle_search_dispatch(query: str) -> str:
|
||||
"""
|
||||
Called by llm.py when AI writes [SEARCH: something].
|
||||
Smartly routes to CVE lookup, exploit search, or general search.
|
||||
"""
|
||||
query = query.strip()
|
||||
|
||||
# CVE pattern — CVE-YYYY-NNNNN
|
||||
import re
|
||||
cve_pattern = re.compile(r'CVE-\d{4}-\d{4,7}', re.IGNORECASE)
|
||||
cve_match = cve_pattern.search(query)
|
||||
if cve_match:
|
||||
return search_cve(cve_match.group())
|
||||
|
||||
# exploit keywords
|
||||
if any(word in query.lower() for word in ["exploit", "poc", "payload", "rce", "lfi", "sqli"]):
|
||||
return web_search(query + " exploit poc github", max_results=5)
|
||||
|
||||
# fix/patch keywords
|
||||
if any(word in query.lower() for word in ["fix", "patch", "mitigate", "harden", "secure"]):
|
||||
return search_fix(query)
|
||||
|
||||
# default general search
|
||||
return web_search(query, max_results=5)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# QUICK TEST
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("[ search.py test ]\n")
|
||||
print("[1] General search")
|
||||
print("[2] CVE lookup")
|
||||
print("[3] Fetch a URL")
|
||||
choice = input("Choice: ").strip()
|
||||
|
||||
if choice == "1":
|
||||
q = input("Query: ").strip()
|
||||
print(web_search(q))
|
||||
|
||||
elif choice == "2":
|
||||
cve = input("CVE ID (e.g. CVE-2021-44228): ").strip()
|
||||
print(search_cve(cve))
|
||||
|
||||
elif choice == "3":
|
||||
url = input("URL: ").strip()
|
||||
print(fetch_page(url))
|
||||
@@ -0,0 +1,251 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
METATRON - tools.py
|
||||
Recon tool runners — all output returned as strings to feed into the LLM.
|
||||
Tools used: nmap, whois, whatweb, curl, dig, nikto
|
||||
OS: Parrot OS (all these tools are pre-installed or easily available)
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# BASE RUNNER
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def run_tool(command: list, timeout: int = 120) -> str:
|
||||
"""
|
||||
Execute a shell command, return combined stdout + stderr as string.
|
||||
Never crashes the program — always returns something.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout
|
||||
)
|
||||
output = result.stdout.strip()
|
||||
errors = result.stderr.strip()
|
||||
|
||||
if output and errors:
|
||||
return output + "\n[STDERR]\n" + errors
|
||||
elif output:
|
||||
return output
|
||||
elif errors:
|
||||
return errors
|
||||
else:
|
||||
return "[!] Tool returned no output."
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return f"[!] Timed out after {timeout}s: {' '.join(command)}"
|
||||
except FileNotFoundError:
|
||||
return f"[!] Tool not found: {command[0]} — install it with: sudo apt install {command[0]}"
|
||||
except Exception as e:
|
||||
return f"[!] Unexpected error running {command[0]}: {e}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# INDIVIDUAL TOOLS
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def run_nmap(target: str) -> str:
|
||||
"""
|
||||
nmap -sV -sC -T4 --open
|
||||
-sV : detect service versions
|
||||
-sC : run default scripts (basic vuln checks)
|
||||
-T4 : aggressive timing (faster)
|
||||
--open : only show open ports
|
||||
"""
|
||||
print(f" [*] nmap -sV -sC -T4 --open {target}")
|
||||
return run_tool(["nmap", "-sV", "-sC", "-T4", "--open", target], timeout=180)
|
||||
|
||||
|
||||
def run_whois(target: str) -> str:
|
||||
"""
|
||||
whois — domain registration, registrar, IP ownership info
|
||||
"""
|
||||
print(f" [*] whois {target}")
|
||||
return run_tool(["whois", target], timeout=30)
|
||||
|
||||
|
||||
def run_whatweb(target: str) -> str:
|
||||
"""
|
||||
whatweb -a 3 — fingerprint web technologies, CMS, frameworks, headers
|
||||
-a 3 : aggression level 3 (active but not destructive)
|
||||
"""
|
||||
print(f" [*] whatweb -a 3 {target}")
|
||||
return run_tool(["whatweb", "-a", "3", target], timeout=60)
|
||||
|
||||
|
||||
def run_curl_headers(target: str) -> str:
|
||||
"""
|
||||
curl -sI — fetch HTTP headers only
|
||||
Reveals: server software, X-Powered-By, cookies, security headers (or lack of them)
|
||||
"""
|
||||
print(f" [*] curl -sI http://{target}")
|
||||
output = run_tool([
|
||||
"curl", "-sI",
|
||||
"--max-time", "10",
|
||||
"--location", # follow redirects
|
||||
f"http://{target}"
|
||||
], timeout=20)
|
||||
|
||||
# also try https
|
||||
https_output = run_tool([
|
||||
"curl", "-sI",
|
||||
"--max-time", "10",
|
||||
"--location",
|
||||
"-k", # ignore cert errors
|
||||
f"https://{target}"
|
||||
], timeout=20)
|
||||
|
||||
return f"[HTTP Headers]\n{output}\n\n[HTTPS Headers]\n{https_output}"
|
||||
|
||||
|
||||
def run_dig(target: str) -> str:
|
||||
"""
|
||||
dig — DNS records: A, MX, NS, TXT
|
||||
Useful for subdomains, mail servers, SPF/DKIM info
|
||||
"""
|
||||
print(f" [*] dig {target} ANY")
|
||||
a_record = run_tool(["dig", "+short", "A", target], timeout=15)
|
||||
mx_record = run_tool(["dig", "+short", "MX", target], timeout=15)
|
||||
ns_record = run_tool(["dig", "+short", "NS", target], timeout=15)
|
||||
txt_record= run_tool(["dig", "+short", "TXT", target], timeout=15)
|
||||
|
||||
return (
|
||||
f"[A Records]\n{a_record}\n\n"
|
||||
f"[MX Records]\n{mx_record}\n\n"
|
||||
f"[NS Records]\n{ns_record}\n\n"
|
||||
f"[TXT Records]\n{txt_record}"
|
||||
)
|
||||
|
||||
|
||||
def run_nikto(target: str) -> str:
|
||||
"""
|
||||
nikto -h — web server vulnerability scanner
|
||||
Checks for outdated software, dangerous files, misconfigurations
|
||||
WARNING: noisy tool, only run with permission
|
||||
"""
|
||||
print(f" [*] nikto -h {target} (this may take a while...)")
|
||||
return run_tool(["nikto", "-h", target, "-nointeractive"], timeout=300)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# MAIN RECON PIPELINE
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
TOOLS_MENU = {
|
||||
"1": ("nmap", run_nmap),
|
||||
"2": ("whois", run_whois),
|
||||
"3": ("whatweb", run_whatweb),
|
||||
"4": ("curl headers", run_curl_headers),
|
||||
"5": ("dig DNS", run_dig),
|
||||
"6": ("nikto", run_nikto),
|
||||
}
|
||||
|
||||
|
||||
def run_default_recon(target: str) -> dict:
|
||||
"""
|
||||
Run the standard recon pipeline (everything except nikto).
|
||||
Returns a dict of {tool_name: output_string}.
|
||||
Nikto is excluded by default — too slow/noisy for auto-run.
|
||||
"""
|
||||
print(f"\n[*] Starting recon on: {target}")
|
||||
print("─" * 50)
|
||||
|
||||
results = {}
|
||||
results["nmap"] = run_nmap(target)
|
||||
results["whois"] = run_whois(target)
|
||||
results["whatweb"] = run_whatweb(target)
|
||||
results["curl_headers"] = run_curl_headers(target)
|
||||
results["dig"] = run_dig(target)
|
||||
|
||||
print("─" * 50)
|
||||
print("[+] Recon complete.\n")
|
||||
return results
|
||||
|
||||
|
||||
def run_single_tool(tool_key: str, target: str) -> str:
|
||||
"""Run one tool by its menu key. Used by AI tool dispatch."""
|
||||
if tool_key in TOOLS_MENU:
|
||||
name, func = TOOLS_MENU[tool_key]
|
||||
return func(target)
|
||||
return f"[!] Unknown tool key: {tool_key}"
|
||||
|
||||
|
||||
def format_recon_for_llm(results: dict) -> str:
|
||||
"""
|
||||
Flatten the recon results dict into one clean string
|
||||
to paste into the LLM prompt.
|
||||
"""
|
||||
output = ""
|
||||
for tool, data in results.items():
|
||||
output += f"\n{'='*50}\n"
|
||||
output += f"[ {tool.upper()} OUTPUT ]\n"
|
||||
output += f"{'='*50}\n"
|
||||
output += data.strip() + "\n"
|
||||
return output
|
||||
|
||||
|
||||
ALLOWED_TOOLS = {"nmap", "whois", "whatweb", "curl", "dig", "nikto"}
|
||||
|
||||
def run_tool_by_command(command_str: str) -> str:
|
||||
parts = command_str.strip().split()
|
||||
if not parts:
|
||||
return "[!] Empty command."
|
||||
|
||||
# allowlist only — reject anything not in the list
|
||||
tool = parts[0].lower().split("/")[-1] # handles /bin/nmap etc
|
||||
if tool not in ALLOWED_TOOLS:
|
||||
return f"[!] Tool '{parts[0]}' is not permitted. Allowed: {ALLOWED_TOOLS}"
|
||||
|
||||
return run_tool(parts)
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# INTERACTIVE TOOL SELECTOR (called from CLI)
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def interactive_tool_run(target: str) -> str:
|
||||
"""
|
||||
Let user manually pick which tools to run.
|
||||
Returns combined output string.
|
||||
"""
|
||||
print("\n[ SELECT TOOLS TO RUN ]")
|
||||
for key, (name, _) in TOOLS_MENU.items():
|
||||
print(f" [{key}] {name}")
|
||||
print(" [a] Run all (except nikto)")
|
||||
print(" [n] Run all + nikto (slow)")
|
||||
|
||||
choice = input("\nChoice(s) e.g. 1 2 4 or a: ").strip().lower()
|
||||
|
||||
if choice == "a":
|
||||
results = run_default_recon(target)
|
||||
return format_recon_for_llm(results)
|
||||
|
||||
if choice == "n":
|
||||
results = run_default_recon(target)
|
||||
results["nikto"] = run_nikto(target)
|
||||
return format_recon_for_llm(results)
|
||||
|
||||
combined = {}
|
||||
for key in choice.split():
|
||||
if key in TOOLS_MENU:
|
||||
name, func = TOOLS_MENU[key]
|
||||
print(f"\n[*] Running {name}...")
|
||||
combined[name] = func(target)
|
||||
else:
|
||||
print(f"[!] Unknown option: {key}")
|
||||
|
||||
return format_recon_for_llm(combined)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# QUICK TEST
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
target = input("Enter test target (IP or domain): ").strip()
|
||||
results = run_default_recon(target)
|
||||
print(format_recon_for_llm(results))
|
||||
Reference in New Issue
Block a user