261 lines
8.9 KiB
Python
261 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""The Analyzer — Web GUI Server
|
|
Run: python3 server.py [port]
|
|
Opens a dark hacker-themed web interface at http://localhost:5000
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import queue
|
|
import threading
|
|
import subprocess
|
|
from pathlib import Path
|
|
from flask import Flask, render_template, request, jsonify, Response, send_from_directory, stream_with_context
|
|
|
|
# Paths
|
|
HERE = Path(__file__).parent
|
|
REPO = HERE.parent
|
|
ANALYZER = REPO / "analyzer"
|
|
REPORTS = REPO / "reports"
|
|
VECTORS = REPO / "vectors"
|
|
|
|
app = Flask(__name__, template_folder=HERE / "templates", static_folder=HERE / "static")
|
|
|
|
# Active scan tracking
|
|
active_scans = {}
|
|
scan_history = []
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────
|
|
|
|
def get_vectors():
|
|
"""Return list of available vectors with metadata."""
|
|
vectors = []
|
|
for f in sorted(VECTORS.glob("*.sh")):
|
|
num = f.stem.split("-")[0]
|
|
name = "-".join(f.stem.split("-")[1:])
|
|
desc = ""
|
|
sev = ""
|
|
with open(f) as fh:
|
|
for line in fh:
|
|
if line.startswith("# Desc:"):
|
|
desc = line.replace("# Desc:", "").strip()
|
|
if line.startswith("# Severity:"):
|
|
sev = line.replace("# Severity:", "").strip()
|
|
vectors.append({
|
|
"num": int(num),
|
|
"name": name,
|
|
"file": f.name,
|
|
"desc": desc,
|
|
"severity": sev
|
|
})
|
|
return vectors
|
|
|
|
def get_reports():
|
|
"""Return list of completed reports."""
|
|
reports = []
|
|
for f in sorted(REPORTS.glob("*.md"), reverse=True):
|
|
if f.name.startswith("."):
|
|
continue
|
|
size = f.stat().st_size
|
|
target = "Unknown"
|
|
vulns = 0
|
|
date = ""
|
|
with open(f) as fh:
|
|
for line in fh:
|
|
if line.startswith("**Target:**"):
|
|
target = line.replace("**Target:**", "").strip()
|
|
if line.startswith("**Date:**"):
|
|
date = line.replace("**Date:**", "").strip()
|
|
if line.startswith("### "):
|
|
vulns += 1
|
|
html_path = f.with_suffix(".html")
|
|
has_html = html_path.exists()
|
|
reports.append({
|
|
"file": f.name,
|
|
"target": target,
|
|
"date": date,
|
|
"vulns": vulns,
|
|
"size": size,
|
|
"html": has_html
|
|
})
|
|
return reports
|
|
|
|
# ─── SSE Scan Runner ──────────────────────────────────────
|
|
|
|
class ScanRunner:
|
|
"""Runs the analyzer and streams output via SSE."""
|
|
|
|
def __init__(self, target, mode, vectors=None):
|
|
self.target = target
|
|
self.mode = mode
|
|
self.vectors = vectors
|
|
self.q = queue.Queue()
|
|
self.process = None
|
|
self.done = False
|
|
self.scan_id = str(int(time.time()))
|
|
self.start_time = time.time()
|
|
|
|
def run(self):
|
|
"""Run the scan in a thread, pushing output to queue."""
|
|
try:
|
|
env = os.environ.copy()
|
|
env["OLLAMA_HOST"] = env.get("OLLAMA_HOST", "http://10.30.20.110:11434")
|
|
env["OLLAMA_MODEL"] = env.get("OLLAMA_MODEL", "granite4.1:8b")
|
|
env["TERM"] = "xterm"
|
|
|
|
cmd = ["bash", str(ANALYZER), self.target, self.mode]
|
|
|
|
self.process = subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
env=env,
|
|
text=True,
|
|
bufsize=1,
|
|
cwd=str(REPO)
|
|
)
|
|
|
|
for line in iter(self.process.stdout.readline, ""):
|
|
if line:
|
|
self.q.put({"type": "output", "data": line.rstrip()})
|
|
|
|
self.process.wait()
|
|
|
|
# Find latest report
|
|
latest_report = None
|
|
reports = sorted(REPORTS.glob("*.md"), key=lambda x: x.stat().st_mtime, reverse=True)
|
|
for r in reports:
|
|
if not r.name.startswith("."):
|
|
latest_report = r.name
|
|
break
|
|
|
|
elapsed = time.time() - self.start_time
|
|
self.q.put({
|
|
"type": "complete",
|
|
"data": {
|
|
"exit_code": self.process.returncode,
|
|
"elapsed": f"{elapsed:.1f}s",
|
|
"report": latest_report
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
self.q.put({"type": "error", "data": str(e)})
|
|
finally:
|
|
self.done = True
|
|
|
|
def stream(self):
|
|
"""Generator for SSE streaming."""
|
|
while not self.done or not self.q.empty():
|
|
try:
|
|
msg = self.q.get(timeout=0.5)
|
|
yield f"data: {json.dumps(msg)}\n\n"
|
|
except queue.Empty:
|
|
yield ": keepalive\n\n"
|
|
|
|
# ─── Routes ───────────────────────────────────────────────
|
|
|
|
@app.route("/")
|
|
def index():
|
|
vectors = get_vectors()
|
|
reports = get_reports()
|
|
return render_template("index.html", vectors=vectors, reports=reports)
|
|
|
|
@app.route("/api/vectors")
|
|
def api_vectors():
|
|
return jsonify(get_vectors())
|
|
|
|
@app.route("/api/reports")
|
|
def api_reports():
|
|
return jsonify(get_reports())
|
|
|
|
@app.route("/api/scan", methods=["POST"])
|
|
def start_scan():
|
|
data = request.get_json()
|
|
target = data.get("target", "").strip()
|
|
mode = data.get("mode", "quick")
|
|
vector_list = data.get("vectors", "")
|
|
|
|
if not target:
|
|
return jsonify({"error": "Target URL required"}), 400
|
|
|
|
# Add https:// if missing
|
|
if not target.startswith("http"):
|
|
target = "https://" + target
|
|
|
|
runner = ScanRunner(target, mode, vector_list)
|
|
scan_id = runner.scan_id
|
|
active_scans[scan_id] = runner
|
|
|
|
thread = threading.Thread(target=runner.run, daemon=True)
|
|
thread.start()
|
|
|
|
return jsonify({"scan_id": scan_id, "status": "started"})
|
|
|
|
@app.route("/api/scan/<scan_id>/stream")
|
|
def scan_stream(scan_id):
|
|
runner = active_scans.get(scan_id)
|
|
if not runner:
|
|
return jsonify({"error": "Scan not found"}), 404
|
|
|
|
return Response(
|
|
stream_with_context(runner.stream()),
|
|
mimetype="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no"
|
|
}
|
|
)
|
|
|
|
@app.route("/api/scan/<scan_id>/stop", methods=["POST"])
|
|
def stop_scan(scan_id):
|
|
runner = active_scans.get(scan_id)
|
|
if runner and runner.process:
|
|
runner.process.terminate()
|
|
return jsonify({"status": "stopped"})
|
|
return jsonify({"error": "Scan not found"}), 404
|
|
|
|
@app.route("/report/<filename>")
|
|
def view_report(filename):
|
|
md_path = REPORTS / filename
|
|
html_path = md_path.with_suffix(".html")
|
|
|
|
if html_path.exists():
|
|
return send_from_directory(str(REPORTS), html_path.name)
|
|
|
|
# Convert markdown to HTML on the fly
|
|
if md_path.exists():
|
|
content = md_path.read_text()
|
|
import markdown
|
|
html = markdown.markdown(content, extensions=["fenced_code", "tables"])
|
|
return f"""<!DOCTYPE html><html><head>
|
|
<meta charset="utf-8"><title>Analyzer Report</title>
|
|
<style>body{{font-family:-apple-system,sans-serif;background:#0f172a;color:#e2e8f0;max-width:800px;margin:40px auto;padding:20px;line-height:1.6}}
|
|
h1{{color:#f97316}} h2{{color:#f1f5f9;border-bottom:1px solid #334155;padding-bottom:8px}}
|
|
a{{color:#22d3ee}} code{{background:#1e293b;padding:2px 6px;border-radius:4px;font-size:14px}}
|
|
pre{{background:#1e293b;padding:16px;border-radius:8px;overflow-x:auto}}
|
|
</style></head><body>{html}</body></html>"""
|
|
|
|
return "Report not found", 404
|
|
|
|
@app.route("/report/<filename>/raw")
|
|
def raw_report(filename):
|
|
md_path = REPORTS / filename
|
|
if md_path.exists():
|
|
return Response(md_path.read_text(), mimetype="text/plain")
|
|
return "Not found", 404
|
|
|
|
# ─── Main ─────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 5000
|
|
print(f"\n ╔══════════════════════════════════════╗")
|
|
print(f" ║ THE ANALYZER — Web Interface ║")
|
|
print(f" ║ ║")
|
|
print(f" ║ http://localhost:{port} ║")
|
|
print(f" ╚══════════════════════════════════════╝\n")
|
|
app.run(host="0.0.0.0", port=port, debug=False, threaded=True)
|