#!/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//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//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/") 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""" Analyzer Report {html}""" return "Report not found", 404 @app.route("/report//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)