#!/usr/bin/env python3 """Local mock of a VULNERABLE RSFiles! install for testing cve_2026_57827.py. This is a safe, self-contained simulation running on 127.0.0.1 only — it does not touch any real system. """ import http.server import socketserver import re import sys from urllib.parse import urlparse, parse_qs PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 VERSION = sys.argv[2] if len(sys.argv) > 2 else "1.17.11" # 1.17.11 = vulnerable stored_shell = {} class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): u = urlparse(self.path) q = parse_qs(u.query) # a tiny "home page" so opening the root shows something useful if u.path in ("/", "/index.php"): body = ( f"" f"

Mock RSFiles! server

" f"

Version: {VERSION} " f"({ 'VULNERABLE' if _is_vulnerable(VERSION) else 'PATCHED' })

" f"

This is a simulation, not a real Joomla site. Available endpoints:

" f"" f"

Uploaded shells live under /downloads/ and /briefcase/.

" f"" ) self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write(body.encode()) return if u.path == "/debug": body = repr(stored_shell).encode() self.send_response(200) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(body) return # manifest -> report version if u.path.endswith("rsfiles.xml"): body = f'{VERSION}' self.send_response(200) self.send_header("Content-Type", "text/xml") self.end_headers() self.wfile.write(body.encode()) return # component presence marker if "com_rsfiles" in u.path and not u.path.startswith("/downloads") \ and not u.path.startswith("/briefcase"): self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write(b"RSFiles component here") return # shell serving from /downloads/ and /briefcase/ — simulate PHP execution for prefix in ("/downloads/", "/briefcase/"): if u.path.startswith(prefix): name = u.path[len(prefix):] if name in stored_shell: body = stored_shell[name] # fake PHP execution: only run if the ?t= token matches the # token embedded in the shell source ($t="...") m = re.search(rb'\$t="([0-9a-f]+)"', body) tok_ok = m and q.get("t", [""])[0] == m.group(1).decode() if tok_ok: if "del" in q: stored_shell.pop(name, None) body = b"DEL" elif "c" in q: body = b"C|uid=0(root) gid=0(root) groups=0(root) hostname|E" else: # rendered shell page: HTML, no raw PHP source body = (b"

think

Dir: /downloads

") else: body = b"404 not found" # token mismatch -> PHP dies with 404 code = 200 if tok_ok else 404 self.send_response(code) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write(body) return self.send_response(404) self.end_headers() self.wfile.write(b"not found") return self.send_response(404) self.end_headers() self.wfile.write(b"not found") def do_POST(self): if "rsfiles.upload" in self.path: # read multipart and extract the uploaded file's content length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length) ct = self.headers.get("Content-Type", "") bm = re.search(r"boundary=([^;]+)", ct) boundary = bm.group(1).strip('"').encode() if bm else b"----x" name, body = None, None for part in raw.split(b"--" + boundary): if b'name="file"' in part and b'filename="' in part: m = re.search(rb'filename="([^"]+)"', part) name = m.group(1).decode() if m else "shell.php" # content is after the blank line separating headers from body body = part.split(b"\r\n\r\n", 1)[-1].rstrip(b"\r\n") if name is None: name = "shell.php" if body is None: body = raw # vulnerable: accepts any file type, stores in /downloads stored_shell[name] = body self.send_response(200) self.end_headers() self.wfile.write(b'{"ok":1,"path":"/downloads/' + name.encode() + b'"}') return self.send_response(403) self.end_headers() self.wfile.write(b"no") def serve_forever(self): socketserver.TCPServer.allow_reuse_address = True with socketserver.TCPServer(("127.0.0.1", PORT), Handler) as httpd: print(" Mock RSFiles! server running:", flush=True) print(f" URL : http://127.0.0.1:{PORT}", flush=True) print(f" Version : {VERSION} ({'VULNERABLE' if _is_vulnerable(VERSION) else 'PATCHED'})", flush=True) print(" Status : waiting for connections on 127.0.0.1 (Ctrl+C to stop)", flush=True) print(flush=True) try: httpd.serve_forever() except KeyboardInterrupt: print("\n Stopping mock server. Bye!") def _is_vulnerable(version: str) -> bool: """Rough check: version below 1.17.12 is the vulnerable one for this CVE.""" try: parts = [int(x) for x in version.split(".")] return (parts[0], parts[1], parts[2]) < (1, 17, 12) except (ValueError, IndexError): return True if __name__ == "__main__": Handler.serve_forever(Handler)