Adding Negotiate/Kerberos authentication (#3404)

extra/kerberos/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""

extra/kerberos/aes.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Dependency-free AES (FIPS-197) block cipher with CBC mode, supporting 128- and 256-bit keys. It is
+# the primitive underneath Kerberos' AES-CTS-HMAC-SHA1-96 (RFC 3962) etypes, kept pure-Python so
+# '--auth-type=Negotiate' needs no third-party crypto library. Validated against the FIPS-197
+# known-answer vectors. Python 2.7 / 3.x.
+#
+# The state is a flat list of 16 ints in AES column-major order: byte i holds row (i % 4), column
+# (i // 4), i.e. column c occupies positions [4*c : 4*c + 4].
+
+def _gmul(a, b):
+    """Multiplication in GF(2**8) with the AES reduction polynomial 0x11b."""
+
+    p = 0
+    for _ in range(8):
+        if b & 1:
+            p ^= a
+        high = a & 0x80
+        a = (a << 1) & 0xff
+        if high:
+            a ^= 0x1b
+        b >>= 1
+    return p
+
+# GF(2**8) log/exp tables (generator 0x03) -> multiplicative inverse -> S-box (affine transform),
+# computed rather than transcribed so there is no 256-entry table to get wrong
+_EXP = [0] * 256
+_LOG = [0] * 256
+_x = 1
+for _i in range(255):
+    _EXP[_i] = _x
+    _LOG[_x] = _i
+    _x = _gmul(_x, 0x03)
+
+def _inv(b):
+    return 0 if b == 0 else _EXP[(255 - _LOG[b]) % 255]
+
+def _rotl8(b, n):
+    return ((b << n) | (b >> (8 - n))) & 0xff
+
+SBOX = []
+for _b in range(256):
+    _v = _inv(_b)
+    SBOX.append(_v ^ _rotl8(_v, 1) ^ _rotl8(_v, 2) ^ _rotl8(_v, 3) ^ _rotl8(_v, 4) ^ 0x63)
+
+INV_SBOX = [0] * 256
+for _b in range(256):
+    INV_SBOX[SBOX[_b]] = _b
+
+RCON = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1b, 0x36, 0x6c, 0xd8, 0xab, 0x4d]
+
+def _xor(a, b):
+    if len(a) != len(b):                                   # equal-length by construction; fail loud (not via assert, which -O strips)
+        raise ValueError("XOR operands differ in length")
+    return bytes(bytearray(x ^ y for x, y in zip(bytearray(a), bytearray(b))))
+
+class AES(object):
+    """AES-128/256 block cipher (16-byte block) with a minimal CBC mode."""
+
+    def __init__(self, key):
+        key = bytearray(key)
+        if len(key) not in (16, 32):
+            raise ValueError("AES key must be 16 or 32 bytes")
+        self.rounds = 10 if len(key) == 16 else 14
+        self._roundKeys = self._expand(key)
+
+    def _expand(self, key):
+        nk = len(key) // 4
+        words = [list(key[4 * i:4 * i + 4]) for i in range(nk)]
+        for i in range(nk, 4 * (self.rounds + 1)):
+            temp = list(words[i - 1])
+            if i % nk == 0:
+                temp = temp[1:] + temp[:1]                          # RotWord
+                temp = [SBOX[b] for b in temp]                      # SubWord
+                temp[0] ^= RCON[i // nk - 1]
+            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)])
+
+        roundKeys = []
+        for r in range(self.rounds + 1):
+            rk = []
+            for c in range(4):
+                rk.extend(words[4 * r + c])
+            roundKeys.append(rk)
+        return roundKeys
+
+    @staticmethod
+    def _addRoundKey(state, rk):
+        for i in range(16):
+            state[i] ^= rk[i]
+
+    @staticmethod
+    def _shiftRows(s):
+        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 _invShiftRows(s):
+        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 _mixColumns(s):
+        out = [0] * 16
+        for c in range(4):
+            col = s[4 * c:4 * c + 4]
+            out[4 * c + 0] = _gmul(col[0], 2) ^ _gmul(col[1], 3) ^ col[2] ^ col[3]
+            out[4 * c + 1] = col[0] ^ _gmul(col[1], 2) ^ _gmul(col[2], 3) ^ col[3]
+            out[4 * c + 2] = col[0] ^ col[1] ^ _gmul(col[2], 2) ^ _gmul(col[3], 3)
+            out[4 * c + 3] = _gmul(col[0], 3) ^ col[1] ^ col[2] ^ _gmul(col[3], 2)
+        return out
+
+    @staticmethod
+    def _invMixColumns(s):
+        out = [0] * 16
+        for c in range(4):
+            col = s[4 * c:4 * c + 4]
+            out[4 * c + 0] = _gmul(col[0], 14) ^ _gmul(col[1], 11) ^ _gmul(col[2], 13) ^ _gmul(col[3], 9)
+            out[4 * c + 1] = _gmul(col[0], 9) ^ _gmul(col[1], 14) ^ _gmul(col[2], 11) ^ _gmul(col[3], 13)
+            out[4 * c + 2] = _gmul(col[0], 13) ^ _gmul(col[1], 9) ^ _gmul(col[2], 14) ^ _gmul(col[3], 11)
+            out[4 * c + 3] = _gmul(col[0], 11) ^ _gmul(col[1], 13) ^ _gmul(col[2], 9) ^ _gmul(col[3], 14)
+        return out
+
+    def encryptBlock(self, block):
+        state = list(bytearray(block))
+        self._addRoundKey(state, self._roundKeys[0])
+        for r in range(1, self.rounds):
+            state = self._mixColumns(self._shiftRows([SBOX[b] for b in state]))
+            self._addRoundKey(state, self._roundKeys[r])
+        state = self._shiftRows([SBOX[b] for b in state])
+        self._addRoundKey(state, self._roundKeys[self.rounds])
+        return bytes(bytearray(state))
+
+    def decryptBlock(self, block):
+        state = list(bytearray(block))
+        self._addRoundKey(state, self._roundKeys[self.rounds])
+        for r in range(self.rounds - 1, 0, -1):
+            state = [INV_SBOX[b] for b in self._invShiftRows(state)]
+            self._addRoundKey(state, self._roundKeys[r])
+            state = self._invMixColumns(state)
+        state = [INV_SBOX[b] for b in self._invShiftRows(state)]
+        self._addRoundKey(state, self._roundKeys[0])
+        return bytes(bytearray(state))
+
+    def cbcEncrypt(self, iv, data):
+        if len(data) % 16 != 0:
+            raise ValueError("CBC input is not block-aligned")
+        prev, out = iv, []
+        for i in range(0, len(data), 16):
+            prev = self.encryptBlock(_xor(data[i:i + 16], prev))
+            out.append(prev)
+        return b"".join(out)
+
+    def cbcDecrypt(self, iv, data):
+        if len(data) % 16 != 0:
+            raise ValueError("CBC input is not block-aligned")
+        prev, out = iv, []
+        for i in range(0, len(data), 16):
+            block = data[i:i + 16]
+            out.append(_xor(self.decryptBlock(block), prev))
+            prev = block
+        return b"".join(out)

extra/kerberos/client.py

@@ -0,0 +1,332 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Dependency-free Kerberos 5 client (RFC 4120) built on the in-tree DER codec and RFC 3961/3962
+# crypto. Implements the AS exchange (password -> TGT) with PA-ENC-TIMESTAMP pre-authentication;
+# the TGS exchange and AP-REQ follow. Talks to the KDC over TCP (4-byte length framing).
+# Python 2.7 / 3.x.
+
+import os
+import socket
+import struct
+import time
+
+from extra.kerberos import der
+from extra.kerberos import spnego
+from extra.kerberos.crypto import ENCTYPES
+
+GSS_CHECKSUM_TYPE = 0x8003                                  # RFC 4121 section 4.1.1 authenticator checksum
+GSS_CHECKSUM_FLAGS = 0                                      # no GSS context flags requested (no mutual/deleg)
+TICKET_LIFETIME_SECONDS = 10 * 3600                        # requested 'till' offset (KDC clamps to its max)
+
+# message types
+AS_REQ, AS_REP, TGS_REQ, TGS_REP, AP_REQ, KRB_ERROR = 10, 11, 12, 13, 14, 30
+
+# principal name types
+NT_PRINCIPAL, NT_SRV_INST = 1, 2
+
+# PA-DATA types
+PA_TGS_REQ, PA_ENC_TIMESTAMP, PA_ETYPE_INFO2 = 1, 2, 19
+
+# KDC error code that carries the PA-ETYPE-INFO2 hint (etype/salt/iteration count) for pre-auth
+KDC_ERR_PREAUTH_REQUIRED = 25
+
+# key usages (RFC 4120 section 7.5.1)
+USAGE_AS_REQ_PA_ENC_TIMESTAMP = 1
+USAGE_AS_REP_ENCPART = 3
+USAGE_TGS_REQ_AUTH_CKSUM = 6
+USAGE_TGS_REQ_AUTH = 7
+USAGE_TGS_REP_ENCPART = 8
+USAGE_AP_REQ_AUTH = 11
+
+PVNO = 5
+DEFAULT_ETYPES = (18, 17, 23)                               # aes256-cts, aes128-cts, rc4-hmac (best first)
+KDC_TIMEOUT = 10                                            # seconds for the KDC TCP exchange
+MAX_KDC_RESPONSE = 8 * 1024 * 1024                          # cap on a KDC reply (guards a hostile length prefix)
+
+def _enctype(etype):
+    if etype not in ENCTYPES:
+        raise KerberosError(-1, "unsupported encryption type %d (only AES-CTS-HMAC-SHA1 is implemented)" % etype)
+    return ENCTYPES[etype]
+
+class KerberosError(Exception):
+    def __init__(self, code, text=None):
+        Exception.__init__(self, "KDC error %d%s" % (code, ": %s" % text if text else ""))
+        self.code = code
+
+# ---- EXPLICIT-tag unwrap helpers ------------------------------------------------------------------
+# Kerberos uses EXPLICIT tagging: an [n] field's content is a complete inner TLV, so it must be
+# peeled before the value can be read. _fields() maps a SEQUENCE's [n] children to that inner TLV.
+def _fields(sequenceContent):
+    out = {}
+    for tag, inner in der.children(sequenceContent):
+        if 0xA0 <= tag <= 0xBE:                             # context-specific, constructed [0]..[30]
+            out[tag - 0xA0] = inner
+    return out
+
+def _expInteger(field):
+    return der.decodeInteger(der.peel(field)[1])
+
+def _expString(field):
+    return der.decodeGeneralString(der.peel(field)[1])
+
+def _expOctet(field):
+    return bytes(der.peel(field)[1])
+
+def _expFields(field):
+    """For an [n] field whose inner TLV is a SEQUENCE, return that SEQUENCE's field map."""
+
+    return _fields(der.peel(field)[1])
+
+# ---- message building -----------------------------------------------------------------------------
+def _nonce():
+    return struct.unpack(">I", os.urandom(4))[0] & 0x7fffffff
+
+def _kerberosTime(offsetSeconds=0):
+    return time.strftime("%Y%m%d%H%M%SZ", time.gmtime(time.time() + offsetSeconds))
+
+def _principalName(nameType, components):
+    return der.sequence(
+        der.tagged(0, der.integer(nameType)),
+        der.tagged(1, der.sequenceOf([der.generalString(_) for _ in components])),
+    )
+
+def _encryptedData(etype, cipher, kvno=None):
+    parts = [der.tagged(0, der.integer(etype))]
+    if kvno is not None:
+        parts.append(der.tagged(1, der.integer(kvno)))
+    parts.append(der.tagged(2, der.octetString(cipher)))
+    return der.sequence(*parts)
+
+# ---- KDC transport (RFC 4120 section 7.2.2: 4-byte length-prefixed over TCP) ----------------------
+def _recvExactly(sock, count):
+    buf = b""
+    while len(buf) < count:
+        chunk = sock.recv(count - len(buf))
+        if not chunk:
+            raise KerberosError(-1, "connection closed by KDC")
+        buf += chunk
+    return buf
+
+def _sendReceive(host, port, request):
+    sock = socket.create_connection((host, port), timeout=KDC_TIMEOUT)
+    try:
+        sock.sendall(struct.pack(">I", len(request)) + request)
+        length = struct.unpack(">I", _recvExactly(sock, 4))[0]
+        if length > MAX_KDC_RESPONSE:
+            raise KerberosError(-1, "KDC reply length %d exceeds the sane maximum" % length)
+        return _recvExactly(sock, length)
+    finally:
+        sock.close()
+
+def _raiseIfError(message):
+    tag, content, _ = der.peel(message)
+    if tag == der.applicationTag(KRB_ERROR):
+        fields = _fields(der.peel(content)[1])
+        raise KerberosError(_expInteger(fields[6]) if 6 in fields else -1,
+                            _expString(fields[11]) if 11 in fields else None)
+    return tag, content
+
+def _parseEtypeInfo2(errorFields):
+    """Extract [(etype, salt, iterations), ...] from a KDC_ERR_PREAUTH_REQUIRED error's PA-ETYPE-INFO2,
+    telling us which etype/salt/s2kparams the KDC expects for the long-term key."""
+
+    out = []
+    if 12 not in errorFields:                              # no e-data
+        return out
+    try:
+        methodData = der.peel(errorFields[12])[1]          # e-data OCTET STRING -> METHOD-DATA (SEQ OF PA-DATA)
+        for _, paData in der.children(der.peel(methodData)[1]):
+            pa = _fields(paData)
+            if 1 in pa and 2 in pa and _expInteger(pa[1]) == PA_ETYPE_INFO2:
+                info = der.peel(pa[2])[1]                   # padata-value OCTET STRING -> ETYPE-INFO2 (SEQ OF entry)
+                for _, entry in der.children(der.peel(info)[1]):
+                    fields = _fields(entry)
+                    etype = _expInteger(fields[0])
+                    salt = _expOctet(fields[1]) if 1 in fields else None   # opaque octets for string2key (RFC 3961), not UTF-8
+                    iterations = None
+                    if 2 in fields:
+                        raw = bytes(der.peel(fields[2])[1]) # s2kparams: 4-byte BE iteration count for AES
+                        iterations = struct.unpack(">I", raw)[0] if len(raw) == 4 else None
+                    out.append((etype, salt, iterations))
+    except (KeyError, IndexError, ValueError, struct.error):
+        del out[:]                                          # malformed hint -> fall back to the default etype/salt
+    return out
+
+def _replyEtype(response):
+    """Return the etype of a KDC-REP's enc-part (which etype the KDC used for the client's key)."""
+
+    try:
+        rep = _fields(der.peel(der.peel(response)[1])[1])
+        return _expInteger(_expFields(rep[6])[0])
+    except (KeyError, IndexError, ValueError, struct.error):
+        raise KerberosError(-1, "malformed KDC reply")
+
+def _parseRep(response, key, usage, expectedNonce, expectedType):
+    """Parse an AS-REP / TGS-REP: decrypt its enc-part with 'key' under 'usage', returning the
+    opaque ticket and the freshly issued session key. The two replies are structurally identical.
+    The reply's application tag MUST match the expected message type, and the nonce carried in the
+    (integrity-protected) enc-part MUST equal the request nonce (RFC 4120)."""
+
+    try:                                                   # any structural defect in a hostile/truncated reply -> KerberosError
+        tag, repContent = _raiseIfError(response)
+        if tag != der.applicationTag(expectedType):
+            raise KerberosError(-1, "unexpected reply message type (tag 0x%02x)" % tag)
+        rep = _fields(der.peel(repContent)[1])
+        encData = _expFields(rep[6])                        # enc-part (EncryptedData)
+        repEtype = _expInteger(encData[0])
+        try:
+            encRepPart = _enctype(repEtype).decrypt(key, usage, _expOctet(encData[2]))
+        except ValueError:                                 # HMAC mismatch -> we hold the wrong long-term key
+            raise KerberosError(-1, "reply decryption failed (wrong password or salt)")
+
+        # Enc*RepPart = [APPLICATION 25/26] EncKDCRepPart ; key is field [0], nonce is field [2]
+        encKdcRep = _fields(der.peel(der.peel(encRepPart)[1])[1])
+        if _expInteger(encKdcRep[2]) != expectedNonce:
+            raise KerberosError(-1, "reply nonce does not match the request (possible replay)")
+        keyFields = _expFields(encKdcRep[0])
+
+        return {
+            "ticket": bytes(rep[5]),
+            "sessionKey": _expOctet(keyFields[1]),
+            "sessionKeyType": _expInteger(keyFields[0]),
+            "etype": repEtype,
+            "crealm": _expString(rep[3]),
+        }
+    except (KeyError, IndexError, ValueError, struct.error):
+        raise KerberosError(-1, "malformed KDC reply")
+
+def _reqBody(realm, snameType, snameComponents, etypes, nonce, cnameComponents=None):
+    parts = [der.tagged(0, der.bitString(b"\x00\x00\x00\x00"))]                      # kdc-options
+    if cnameComponents is not None:
+        parts.append(der.tagged(1, _principalName(NT_PRINCIPAL, cnameComponents)))  # cname (AS only)
+    parts.append(der.tagged(2, der.generalString(realm)))                           # realm
+    parts.append(der.tagged(3, _principalName(snameType, snameComponents)))         # sname
+    parts.append(der.tagged(5, der.generalizedTime(_kerberosTime(offsetSeconds=TICKET_LIFETIME_SECONDS))))  # till
+    parts.append(der.tagged(7, der.integer(nonce)))                                 # nonce
+    parts.append(der.tagged(8, der.sequenceOf([der.integer(_) for _ in etypes])))   # etype
+    return der.sequence(*parts)
+
+def _authenticator(crealm, cnameComponents, cksum=None):
+    parts = [
+        der.tagged(0, der.integer(PVNO)),
+        der.tagged(1, der.generalString(crealm)),
+        der.tagged(2, _principalName(NT_PRINCIPAL, cnameComponents)),
+    ]
+    if cksum is not None:
+        parts.append(der.tagged(3, der.sequence(der.tagged(0, der.integer(cksum[0])),
+                                                 der.tagged(1, der.octetString(cksum[1])))))
+    parts.append(der.tagged(4, der.integer(struct.unpack(">I", os.urandom(4))[0] % 1000000)))  # cusec
+    parts.append(der.tagged(5, der.generalizedTime(_kerberosTime())))               # ctime
+    return der.application(2, der.sequence(*parts))
+
+def _apReq(ticket, encAuthenticator, etype, apOptions=b"\x00\x00\x00\x00"):
+    return der.application(AP_REQ, der.sequence(
+        der.tagged(0, der.integer(PVNO)),
+        der.tagged(1, der.integer(AP_REQ)),
+        der.tagged(2, der.bitString(apOptions)),
+        der.tagged(3, ticket),                              # raw Ticket TLV (already [APPLICATION 1])
+        der.tagged(4, _encryptedData(etype, encAuthenticator)),
+    ))
+
+# ---- AS exchange (password -> TGT) ----------------------------------------------------------------
+def _asReq(realm, username, etypes, nonce, padata=None):
+    reqBody = _reqBody(realm, NT_SRV_INST, ["krbtgt", realm], etypes, nonce, cnameComponents=[username])
+    parts = [der.tagged(1, der.integer(PVNO)), der.tagged(2, der.integer(AS_REQ))]
+    if padata is not None:
+        parts.append(der.tagged(3, der.sequenceOf([padata])))
+    parts.append(der.tagged(4, reqBody))
+    return der.application(AS_REQ, der.sequence(*parts))
+
+def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES, salt=None):
+    """Run the AS exchange and return the TGT and its session key.
+
+    Follows the standard two-step flow: an initial request without pre-auth learns the KDC's expected
+    etype/salt/iteration-count from PA-ETYPE-INFO2 (so non-default salts and AES-128-only principals
+    work), then a PA-ENC-TIMESTAMP-authenticated request obtains the ticket. Returns
+    {'ticket': <raw Ticket TLV>, 'sessionKey': bytes, 'sessionKeyType': int, 'crealm': str}.
+    """
+
+    realm = realm.upper()
+    chosenSalt = salt if salt is not None else realm + username
+
+    # 1) probe without pre-auth to discover the etype/salt/iterations (or get the TGT outright)
+    nonce = _nonce()
+    response = _sendReceive(kdcHost, kdcPort, _asReq(realm, username, etypes, nonce))
+    tag = der.peel(response)[0]
+
+    if tag == der.applicationTag(AS_REP):                  # KDC issued the ticket without pre-auth
+        etype = _replyEtype(response)                      # derive the key for the etype the KDC actually used
+        clientKey = _enctype(etype).string2key(password, chosenSalt)
+        return _parseRep(response, clientKey, USAGE_AS_REP_ENCPART, nonce, AS_REP)
+
+    etype, iterations = etypes[0], None
+    if tag == der.applicationTag(KRB_ERROR):
+        errorFields = _fields(der.peel(der.peel(response)[1])[1])
+        code = _expInteger(errorFields[6]) if 6 in errorFields else -1
+        if code != KDC_ERR_PREAUTH_REQUIRED:
+            raise KerberosError(code, _expString(errorFields[11]) if 11 in errorFields else None)
+        for advertisedEtype, advertisedSalt, advertisedIters in _parseEtypeInfo2(errorFields):
+            if advertisedEtype in ENCTYPES:
+                etype = advertisedEtype
+                iterations = advertisedIters
+                if salt is None and advertisedSalt is not None:
+                    chosenSalt = advertisedSalt
+                break
+
+    enc = _enctype(etype)
+    clientKey = enc.string2key(password, chosenSalt, iterations)
+
+    # 2) authenticated request with PA-ENC-TIMESTAMP under the discovered etype/salt
+    paTsEnc = der.sequence(der.tagged(0, der.generalizedTime(_kerberosTime())), der.tagged(1, der.integer(0)))
+    cipher = enc.encrypt(clientKey, USAGE_AS_REQ_PA_ENC_TIMESTAMP, paTsEnc)
+    paData = der.sequence(
+        der.tagged(1, der.integer(PA_ENC_TIMESTAMP)),
+        der.tagged(2, der.octetString(_encryptedData(etype, cipher))),
+    )
+    nonce = _nonce()
+    response = _sendReceive(kdcHost, kdcPort, _asReq(realm, username, etypes, nonce, padata=paData))
+    return _parseRep(response, clientKey, USAGE_AS_REP_ENCPART, nonce, AS_REP)
+
+# ---- TGS exchange (TGT -> service ticket) ---------------------------------------------------------
+def getServiceTicket(tgt, realm, username, serviceComponents, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES):
+    """Present the TGT in a PA-TGS-REQ AP-REQ to obtain a ticket for the named service.
+
+    Returns the same shape as getTGT (the 'ticket' is now the service ticket).
+    """
+
+    realm = realm.upper()
+    enc = _enctype(tgt["sessionKeyType"])
+    nonce = _nonce()
+    reqBody = _reqBody(realm, NT_SRV_INST, serviceComponents, etypes, nonce)
+
+    cksum = (enc.cksumtype, enc.checksum(tgt["sessionKey"], USAGE_TGS_REQ_AUTH_CKSUM, reqBody))
+    authenticator = _authenticator(realm, [username], cksum=cksum)
+    encAuth = enc.encrypt(tgt["sessionKey"], USAGE_TGS_REQ_AUTH, authenticator)
+    apReq = _apReq(tgt["ticket"], encAuth, tgt["sessionKeyType"])
+
+    paTgs = der.sequence(der.tagged(1, der.integer(PA_TGS_REQ)), der.tagged(2, der.octetString(apReq)))
+    tgsReq = der.application(TGS_REQ, der.sequence(
+        der.tagged(1, der.integer(PVNO)),
+        der.tagged(2, der.integer(TGS_REQ)),
+        der.tagged(3, der.sequenceOf([paTgs])),
+        der.tagged(4, reqBody),
+    ))
+
+    return _parseRep(_sendReceive(kdcHost, kdcPort, tgsReq), tgt["sessionKey"], USAGE_TGS_REP_ENCPART, nonce, TGS_REP)
+
+# ---- SPNEGO "Negotiate" token (cached service ticket -> ready-to-send HTTP token) -----------------
+def spnegoFromTicket(service, realm, username):
+    """Build a fresh SPNEGO token from an already-obtained service ticket (no KDC round-trip). Each
+    call produces a new AP-REQ authenticator, as replay caches require, so a cached ticket can back
+    every request of a scan cheaply."""
+
+    enc = _enctype(service["sessionKeyType"])
+    gssChecksum = (GSS_CHECKSUM_TYPE, struct.pack("<I", 16) + b"\x00" * 16 + struct.pack("<I", GSS_CHECKSUM_FLAGS))
+    authenticator = _authenticator(realm.upper(), [username], cksum=gssChecksum)
+    encAuth = enc.encrypt(service["sessionKey"], USAGE_AP_REQ_AUTH, authenticator)
+    return spnego.negTokenInit(_apReq(service["ticket"], encAuth, service["sessionKeyType"]))

extra/kerberos/crypto.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Dependency-free Kerberos "simplified profile" crypto (RFC 3961) for the AES-CTS-HMAC-SHA1-96 etypes
+# (RFC 3962: aes128-cts-hmac-sha1-96 = etype 17, aes256-cts-hmac-sha1-96 = etype 18), built on the
+# pure-Python AES core. Provides n-fold, DK/DR key derivation, string-to-key (PBKDF2-HMAC-SHA1) and
+# authenticated encrypt/decrypt. Validated against the RFC 3961 A / RFC 3962 B test vectors.
+# Python 2.7 / 3.x.
+
+import binascii
+import hashlib
+import hmac
+import os
+import struct
+
+from extra.kerberos.aes import AES, _xor                   # _xor reused (no second copy) from the AES core
+from lib.request.ntlm import _md4                          # proven RFC 1320 MD4 (reused, not re-derived)
+
+# RFC 3962 string-to-key default work factor when the KDC advertises no explicit count
+DEFAULT_PBKDF2_ITERATIONS = 4096
+
+def _to_bytes(value):
+    return value if isinstance(value, bytes) else value.encode("utf-8")
+
+def _b2i(data):
+    data = bytearray(data)
+    return int(binascii.hexlify(bytes(data)), 16) if data else 0
+
+def _i2b(n, length):
+    if length <= 0:
+        return bytearray()
+    return bytearray(binascii.unhexlify(("%0*x" % (length * 2, n))[-length * 2:]))
+
+def _eq(a, b):
+    return hmac.compare_digest(bytes(a), bytes(b)) if hasattr(hmac, "compare_digest") else (bytes(a) == bytes(b))
+
+def _pbkdf2(password, salt, iterations, dklen):
+    """PBKDF2-HMAC-SHA1. Uses the stdlib primitive when present, with an RFC 2898 fallback for the
+    pre-2.7.8 interpreters that lack hashlib.pbkdf2_hmac."""
+
+    if hasattr(hashlib, "pbkdf2_hmac"):
+        return hashlib.pbkdf2_hmac("sha1", password, salt, iterations, dklen)
+
+    out = bytearray()
+    block = 1
+    while len(out) < dklen:
+        u = hmac.new(password, salt + struct.pack(">I", block), hashlib.sha1).digest()
+        acc = bytearray(u)
+        for _ in range(iterations - 1):
+            u = hmac.new(password, u, hashlib.sha1).digest()
+            acc = bytearray(x ^ y for x, y in zip(acc, bytearray(u)))
+        out += acc
+        block += 1
+    return bytes(out[:dklen])
+
+def _rotate_right(data, nbits):
+    """Rotate a byte string right by 'nbits' bits, preserving its length."""
+
+    data = bytearray(data)
+    if not data:
+        return data
+    total = len(data) * 8
+    nbits %= total
+    value = ((_b2i(data) >> nbits) | (_b2i(data) << (total - nbits))) & ((1 << total) - 1)
+    return _i2b(value, len(data))
+
+def nfold(data, nbytes):
+    """RFC 3961 n-fold: spread 'data' over 'nbytes' bytes via 13-bit rotated copies summed with an
+    end-around carry (ones-complement addition)."""
+
+    data = bytearray(data)
+
+    def gcd(a, b):
+        while b:
+            a, b = b, a % b
+        return a
+
+    lcm = len(data) * nbytes // gcd(len(data), nbytes)
+
+    buf = bytearray()
+    rotation = 0
+    while len(buf) < lcm:
+        buf += _rotate_right(data, rotation)
+        rotation += 13
+
+    bits = 8 * nbytes
+    mask = (1 << bits) - 1
+    acc = sum(_b2i(buf[off:off + nbytes]) for off in range(0, lcm, nbytes))
+    while acc > mask:
+        acc = (acc & mask) + (acc >> bits)
+    return bytes(_i2b(acc, nbytes))
+
+class AESEnctype(object):
+    """AES-CTS-HMAC-SHA1-96 simplified-profile enctype (RFC 3962). keysize 16 => etype 17, 32 => 18."""
+
+    blocksize = 16
+    macsize = 12
+
+    def __init__(self, keysize):
+        self.keysize = keysize
+        self.cksumtype = 16 if keysize == 32 else 15       # hmac-sha1-96-aes256 / -aes128
+
+    def checksum(self, key, usage, data):
+        """Keyed checksum (RFC 3961 get_mic): HMAC-SHA1-96 under the checksum key DK(key, usage|0x99)."""
+
+        kc = self.dk(key, struct.pack(">IB", usage, 0x99))
+        return hmac.new(kc, data, hashlib.sha1).digest()[:self.macsize]
+
+    # --- key schedule -------------------------------------------------------------------------------
+    def _dr(self, key, constant):
+        """RFC 3961 DR: iterate the single-block cipher over the (n-folded) constant to seedsize."""
+
+        aes = AES(key)
+        block = nfold(constant, self.blocksize)
+        out = bytearray()
+        while len(out) < self.keysize:
+            block = aes.encryptBlock(block)                     # single 16-byte block => CBC(iv=0) == ECB
+            out += bytearray(block)
+        return bytes(out[:self.keysize])
+
+    def dk(self, key, constant):
+        """RFC 3961 DK = random-to-key(DR(...)); random-to-key is the identity for AES."""
+
+        return self._dr(key, constant)
+
+    def string2key(self, password, salt, iterations=None):
+        """RFC 3962 string-to-key: DK(PBKDF2-HMAC-SHA1(password, salt), "kerberos")."""
+
+        iterations = iterations or DEFAULT_PBKDF2_ITERATIONS
+        tkey = _pbkdf2(_to_bytes(password), _to_bytes(salt), iterations, self.keysize)
+        return self.dk(tkey, b"kerberos")
+
+    # --- CBC ciphertext stealing (RFC 3962, CS3: always swap the final two blocks) ------------------
+    def _basicEncrypt(self, key, data):
+        aes = AES(key)
+        padded = data + b"\x00" * ((-len(data)) % self.blocksize)
+        ct = aes.cbcEncrypt(b"\x00" * self.blocksize, padded)
+        if len(data) > self.blocksize:
+            lastlen = len(data) % self.blocksize or self.blocksize
+            ct = ct[:-2 * self.blocksize] + ct[-self.blocksize:] + ct[-2 * self.blocksize:-self.blocksize][:lastlen]
+        return ct
+
+    def _basicDecrypt(self, key, data):
+        aes = AES(key)
+        if len(data) == self.blocksize:
+            return aes.decryptBlock(data)
+
+        blocks = [bytearray(data[p:p + self.blocksize]) for p in range(0, len(data), self.blocksize)]
+        lastlen = len(blocks[-1])
+        prev = bytearray(self.blocksize)
+        out = bytearray()
+        for block in blocks[:-2]:
+            out += bytearray(_xor(aes.decryptBlock(bytes(block)), prev))
+            prev = block
+
+        decrypted = bytearray(aes.decryptBlock(bytes(blocks[-2])))
+        lastPlain = _xor(decrypted[:lastlen], blocks[-1])
+        omitted = decrypted[lastlen:]
+        secondLast = _xor(aes.decryptBlock(bytes(blocks[-1] + omitted)), prev)
+        return bytes(out) + secondLast + lastPlain
+
+    # --- authenticated encryption (RFC 3961 section 5.3) --------------------------------------------
+    def _keys(self, key, usage):
+        ke = self.dk(key, struct.pack(">IB", usage, 0xAA))
+        ki = self.dk(key, struct.pack(">IB", usage, 0x55))
+        return ke, ki
+
+    def encrypt(self, key, usage, plaintext, confounder=None):
+        ke, ki = self._keys(key, usage)
+        if confounder is None:
+            confounder = os.urandom(self.blocksize)
+        basic = confounder + plaintext
+        return self._basicEncrypt(ke, basic) + hmac.new(ki, basic, hashlib.sha1).digest()[:self.macsize]
+
+    def decrypt(self, key, usage, ciphertext):
+        if len(ciphertext) < self.blocksize + self.macsize:   # confounder block + HMAC; guards a hostile short reply
+            raise ValueError("Kerberos ciphertext too short")
+        ke, ki = self._keys(key, usage)
+        ct, mac = ciphertext[:-self.macsize], ciphertext[-self.macsize:]
+        basic = self._basicDecrypt(ke, ct)
+        if not _eq(mac, hmac.new(ki, basic, hashlib.sha1).digest()[:self.macsize]):
+            raise ValueError("Kerberos integrity check failed (wrong key or corrupted ciphertext)")
+        return basic[self.blocksize:]
+
+def _rc4(key, data):
+    """RC4 (ARCFOUR) stream cipher."""
+
+    key, data = bytearray(key), bytearray(data)
+    if not key:
+        raise ValueError("RC4 requires a non-empty key")
+    s = list(range(256))
+    j = 0
+    for i in range(256):
+        j = (j + s[i] + key[i % len(key)]) & 0xff
+        s[i], s[j] = s[j], s[i]
+
+    out = bytearray(len(data))
+    i = j = 0
+    for n in range(len(data)):
+        i = (i + 1) & 0xff
+        j = (j + s[i]) & 0xff
+        s[i], s[j] = s[j], s[i]
+        out[n] = data[n] ^ s[(s[i] + s[j]) & 0xff]
+    return bytes(out)
+
+class RC4Enctype(object):
+    """rc4-hmac (etype 23, RFC 4757). The long-term key is the NT hash MD4(UTF-16LE(password)); the
+    salt and iteration count are unused. Legacy, but still enabled in many AD environments."""
+
+    keysize = 16
+    cksumtype = -138                                       # hmac-md5
+
+    def string2key(self, password, salt=None, iterations=None):
+        # the password is text; encode it UTF-16LE (in py2 a str is bytes, so decode to text first)
+        if isinstance(password, bytes):
+            password = password.decode("utf-8")
+        return _md4(password.encode("utf-16-le"))
+
+    @staticmethod
+    def _usage(usage):
+        # RFC 4757 section 3: a couple of Kerberos usages map to Microsoft-specific values
+        return struct.pack("<I", {3: 8, 9: 8}.get(usage, usage))
+
+    def encrypt(self, key, usage, plaintext, confounder=None):
+        if confounder is None:
+            confounder = os.urandom(8)
+        ki = hmac.new(key, self._usage(usage), hashlib.md5).digest()
+        cksum = hmac.new(ki, confounder + plaintext, hashlib.md5).digest()
+        ke = hmac.new(ki, cksum, hashlib.md5).digest()
+        return cksum + _rc4(ke, confounder + plaintext)
+
+    def decrypt(self, key, usage, ciphertext):
+        if len(ciphertext) < 24:
+            raise ValueError("rc4-hmac ciphertext too short")
+        cksum, data = ciphertext[:16], ciphertext[16:]
+        ki = hmac.new(key, self._usage(usage), hashlib.md5).digest()
+        ke = hmac.new(ki, cksum, hashlib.md5).digest()
+        plaintext = _rc4(ke, data)
+        if not _eq(cksum, hmac.new(ki, plaintext, hashlib.md5).digest()):
+            raise ValueError("Kerberos integrity check failed (wrong key or corrupted ciphertext)")
+        return plaintext[8:]                               # strip the 8-byte confounder
+
+    def checksum(self, key, usage, data):
+        ksign = hmac.new(key, b"signaturekey\x00", hashlib.md5).digest()
+        return hmac.new(ksign, hashlib.md5(self._usage(usage) + bytes(data)).digest(), hashlib.md5).digest()
+
+# etype number -> enctype implementation
+ENCTYPES = {
+    17: AESEnctype(16),
+    18: AESEnctype(32),
+    23: RC4Enctype(),
+}

extra/kerberos/der.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Minimal, dependency-free ASN.1 DER codec covering exactly the constructs Kerberos (RFC 4120) uses:
+# INTEGER, OCTET STRING, GeneralString, GeneralizedTime, BIT STRING, SEQUENCE / SEQUENCE OF, EXPLICIT
+# context tags [n] and [APPLICATION n]. All Kerberos tag numbers are <= 30, so only the low-tag-number
+# form is needed. Encoders return bytes; decoders accept bytes/bytearray. Python 2.7 / 3.x.
+
+# universal tag bytes
+INTEGER = 0x02
+BIT_STRING = 0x03
+OCTET_STRING = 0x04
+GENERAL_STRING = 0x1b
+GENERALIZED_TIME = 0x18
+SEQUENCE = 0x30                                             # 0x10 | constructed(0x20)
+
+def _encodeLength(length):
+    if length < 0x80:
+        return bytearray([length])
+    out = bytearray()
+    while length:
+        out.insert(0, length & 0xff)
+        length >>= 8
+    return bytearray([0x80 | len(out)]) + out
+
+def _tlv(tag, value):
+    value = bytearray(value)
+    return bytes(bytearray([tag]) + _encodeLength(len(value)) + value)
+
+# ---- context / application tags (EXPLICIT) --------------------------------------------------------
+def contextTag(number):
+    return 0x80 | 0x20 | number                            # context-specific, constructed
+
+def applicationTag(number):
+    return 0x40 | 0x20 | number                            # application, constructed
+
+def tagged(number, innerTLV):
+    """EXPLICIT [n] wrapper around an already-encoded inner TLV."""
+
+    return _tlv(contextTag(number), innerTLV)
+
+def application(number, innerTLV):
+    """[APPLICATION n] wrapper around an already-encoded inner TLV."""
+
+    return _tlv(applicationTag(number), innerTLV)
+
+# ---- primitive encoders ---------------------------------------------------------------------------
+def integer(value):
+    content = bytearray()
+    if value == 0:
+        content = bytearray([0])
+    elif value > 0:
+        n = value
+        while n:
+            content.insert(0, n & 0xff)
+            n >>= 8
+        if content[0] & 0x80:                              # keep the sign bit clear for a positive value
+            content.insert(0, 0x00)
+    else:
+        n = value
+        while True:
+            content.insert(0, n & 0xff)
+            n >>= 8
+            if n == -1 and (content[0] & 0x80):
+                break
+    return _tlv(INTEGER, content)
+
+def octetString(value):
+    return _tlv(OCTET_STRING, value)
+
+def generalString(value):
+    return _tlv(GENERAL_STRING, value if isinstance(value, bytes) else value.encode("utf-8"))
+
+def generalizedTime(value):
+    """'value' is a 'YYYYMMDDHHMMSSZ' UTC string."""
+
+    return _tlv(GENERALIZED_TIME, value if isinstance(value, bytes) else value.encode("ascii"))
+
+def bitString(value, unusedBits=0):
+    return _tlv(BIT_STRING, bytearray([unusedBits]) + bytearray(value))
+
+def sequence(*elements):
+    return _tlv(SEQUENCE, b"".join(bytes(_) for _ in elements))
+
+def sequenceOf(elements):
+    return _tlv(SEQUENCE, b"".join(bytes(_) for _ in elements))
+
+# ---- decoding -------------------------------------------------------------------------------------
+def peel(data, offset=0):
+    """Parse one TLV at 'offset'; return (tag, content_bytearray, next_offset). Raises ValueError on
+    truncated or indefinite-length input (the data may come from the network, so fail predictably)."""
+
+    data = bytearray(data)
+    if offset + 2 > len(data):
+        raise ValueError("truncated DER header")
+    tag = data[offset]
+    first = data[offset + 1]
+    offset += 2
+    if first < 0x80:
+        length = first
+    elif first == 0x80:
+        raise ValueError("indefinite-length DER is not permitted")
+    else:
+        count = first & 0x7f
+        if offset + count > len(data):
+            raise ValueError("truncated DER length")
+        length = 0
+        for _ in range(count):
+            length = (length << 8) | data[offset]
+            offset += 1
+    if offset + length > len(data):
+        raise ValueError("truncated DER content")
+    return tag, data[offset:offset + length], offset + length
+
+def children(content):
+    """Iterate the TLVs contained in a constructed value; yields (tag, content_bytearray)."""
+
+    content = bytearray(content)
+    offset = 0
+    out = []
+    while offset < len(content):
+        tag, inner, offset = peel(content, offset)
+        out.append((tag, inner))
+    return out
+
+def decodeInteger(content):
+    content = bytearray(content)
+    if not content:
+        return 0
+    value = 0
+    for b in content:
+        value = (value << 8) | b
+    if content[0] & 0x80:                                  # negative (two's complement)
+        value -= 1 << (8 * len(content))
+    return value
+
+def decodeGeneralString(content):
+    return bytes(bytearray(content)).decode("utf-8", "replace")

extra/kerberos/discovery.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Dependency-free KDC discovery for a realm, so '--auth-type=Negotiate' works without an explicit
+# KDC address. Resolution order: the 'SQLMAP_KERBEROS_KDC' environment variable, then the local
+# krb5.conf [realms] section, then a DNS SRV lookup (_kerberos._tcp.<realm>), then the realm name
+# itself as a host. Returns (host, port). Python 2.7 / 3.x.
+
+import os
+import re
+import socket
+import struct
+
+DEFAULT_KDC_PORT = 88
+_DNS_TIMEOUT = 3
+_SRV_TYPE = 33
+_IN_CLASS = 1
+
+def _splitHostPort(value, defaultPort=DEFAULT_KDC_PORT):
+    value = value.strip()
+    if value.startswith("["):                              # [IPv6] or [IPv6]:port
+        host, _, rest = value[1:].partition("]")
+        port = rest[1:] if rest.startswith(":") else ""
+    elif value.count(":") == 1:                            # host:port (a single colon rules out bare IPv6)
+        host, _, port = value.partition(":")
+    else:                                                  # bare host or bare IPv6 literal
+        host, port = value, ""
+    return host, (int(port) if port.isdigit() else defaultPort)
+
+# ---- krb5.conf --------------------------------------------------------------------------------
+def _fromKrb5Conf(realm):
+    path = os.environ.get("KRB5_CONFIG") or "/etc/krb5.conf"
+    try:
+        with open(path) as f:
+            content = f.read()
+    except (IOError, OSError):
+        return None
+
+    header = re.search(r"(?im)^\s*%s\s*=\s*\{" % re.escape(realm), content)
+    if not header:
+        return None
+    start = header.end()                                   # brace-match so a nested '{ }' block cannot truncate us
+    depth, i = 1, start
+    while i < len(content) and depth > 0:
+        if content[i] == "{":
+            depth += 1
+        elif content[i] == "}":
+            depth -= 1
+        i += 1
+    block = content[start:i - 1]
+    kdc = re.search(r"(?im)^\s*kdc\s*=\s*(\S+)", block)
+    return kdc.group(1) if kdc else None
+
+# ---- DNS SRV (_kerberos._tcp.<realm>) ---------------------------------------------------------
+def _nameservers():
+    servers = []
+    try:
+        with open("/etc/resolv.conf") as f:
+            for line in f:
+                parts = line.split()
+                if len(parts) >= 2 and parts[0] == "nameserver":
+                    servers.append(parts[1])
+    except (IOError, OSError):
+        pass
+    return servers
+
+def _encodeName(name):
+    out = bytearray()
+    for label in name.split("."):
+        out.append(len(label))
+        out += label.encode("ascii")
+    out.append(0)
+    return bytes(out)
+
+_MAX_NAME_JUMPS = 64                                       # guards against compression-pointer cycles
+
+def _skipName(data, offset):
+    while True:
+        if offset >= len(data):
+            raise ValueError("truncated DNS name")
+        length = data[offset]
+        if length == 0:
+            return offset + 1
+        if length & 0xc0 == 0xc0:                          # compression pointer ends the name
+            if offset + 2 > len(data):
+                raise ValueError("truncated DNS pointer")
+            return offset + 2
+        offset += 1 + length
+
+def _readName(data, offset):
+    labels = []
+    end = None
+    jumps = 0
+    while True:
+        if offset >= len(data):
+            raise ValueError("truncated DNS name")
+        length = data[offset]
+        if length == 0:
+            offset += 1
+            break
+        if length & 0xc0 == 0xc0:                          # follow compression pointer (bounded, cycle-safe)
+            if offset + 2 > len(data):
+                raise ValueError("truncated DNS pointer")
+            jumps += 1
+            if jumps > _MAX_NAME_JUMPS:
+                raise ValueError("too many DNS compression jumps")
+            if end is None:
+                end = offset + 2
+            offset = ((length & 0x3f) << 8) | data[offset + 1]
+            continue
+        if offset + 1 + length > len(data):
+            raise ValueError("truncated DNS label")
+        labels.append(bytes(data[offset + 1:offset + 1 + length]).decode("ascii", "replace"))
+        offset += 1 + length
+    return ".".join(labels), (end if end is not None else offset)
+
+def parseSrv(response):
+    """Parse SRV records from a (possibly hostile) DNS response into [(priority, weight, port,
+    target), ...]. Malformed input yields an empty list rather than raising."""
+
+    data = bytearray(response)
+    if len(data) < 12:
+        return []
+    try:
+        qdcount, ancount = struct.unpack(">HH", bytes(data[4:8]))
+        offset = 12
+        for _ in range(qdcount):
+            offset = _skipName(data, offset) + 4           # + qtype/qclass
+        records = []
+        for _ in range(ancount):
+            offset = _skipName(data, offset)
+            if offset + 10 > len(data):
+                break
+            rtype, _cls, _ttl, rdlength = struct.unpack(">HHIH", bytes(data[offset:offset + 10]))
+            offset += 10
+            if offset + rdlength > len(data):
+                break
+            if rtype == _SRV_TYPE and rdlength >= 6:
+                priority, weight, port = struct.unpack(">HHH", bytes(data[offset:offset + 6]))
+                target = _readName(data, offset + 6)[0].rstrip(".")
+                if target:
+                    records.append((priority, weight, port, target))
+            offset += rdlength
+        return records
+    except (ValueError, struct.error, IndexError):
+        return []
+
+def _fromDnsSrv(realm):
+    queryId = os.urandom(2)
+    query = (queryId + struct.pack(">HHHHH", 0x0100, 1, 0, 0, 0) +
+             _encodeName("_kerberos._tcp.%s" % realm) + struct.pack(">HH", _SRV_TYPE, _IN_CLASS))
+    for server in _nameservers():
+        family = socket.AF_INET6 if ":" in server else socket.AF_INET
+        sock = socket.socket(family, socket.SOCK_DGRAM)
+        sock.settimeout(_DNS_TIMEOUT)
+        try:
+            sock.connect((server, 53))                     # connect() so the kernel drops replies from any other source
+            sock.send(query)
+            response = sock.recv(4096)
+        except socket.error:
+            continue
+        finally:
+            sock.close()
+        if len(response) < 2 or response[:2] != queryId:   # ignore stray / spoofed replies
+            continue
+        records = parseSrv(response)
+        if records:
+            best = min(records, key=lambda r: (r[0], -r[1]))   # lowest priority, then highest weight
+            return best[3], best[2]
+    return None
+
+def discoverKdc(realm):
+    """Resolve (host, port) of a KDC for the realm; falls back to the realm name itself as a host."""
+
+    override = os.environ.get("SQLMAP_KERBEROS_KDC")
+    if override:
+        return _splitHostPort(override)
+
+    configured = _fromKrb5Conf(realm)
+    if configured:
+        return _splitHostPort(configured)
+
+    fromDns = _fromDnsSrv(realm)
+    if fromDns:
+        return fromDns
+
+    return realm.lower(), DEFAULT_KDC_PORT

extra/kerberos/spnego.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Minimal GSS-API / SPNEGO (RFC 2743, RFC 4178) wrapping of a Kerberos AP-REQ into the token carried
+# by the HTTP "Authorization: Negotiate <base64>" header. Only the initiator's NegTokenInit is built
+# (the one-shot token an HTTP client sends); the mechanism-specific OIDs are fixed constants.
+# Python 2.7 / 3.x.
+
+from extra.kerberos import der
+
+# fully-encoded OBJECT IDENTIFIER TLVs
+KRB5_OID = bytes(bytearray([0x06, 0x09, 0x2a, 0x86, 0x48, 0x86, 0xf7, 0x12, 0x01, 0x02, 0x02]))  # 1.2.840.113554.1.2.2
+SPNEGO_OID = bytes(bytearray([0x06, 0x06, 0x2b, 0x06, 0x01, 0x05, 0x05, 0x02]))                  # 1.3.6.1.5.5.2
+
+TOK_ID_AP_REQ = b"\x01\x00"                                # GSS Kerberos token id for KRB_AP_REQ
+
+def gssApReq(apReq):
+    """GSS InitialContextToken: [APPLICATION 0] { Kerberos OID, tok-id, AP-REQ }."""
+
+    return der.application(0, KRB5_OID + TOK_ID_AP_REQ + apReq)
+
+def negTokenInit(apReq):
+    """SPNEGO NegTokenInit wrapping the Kerberos GSS token (Kerberos advertised as the sole mech)."""
+
+    inner = der.sequence(
+        der.tagged(0, der.sequenceOf([KRB5_OID])),         # mechTypes
+        der.tagged(2, der.octetString(gssApReq(apReq))),   # mechToken
+    )
+    return der.application(0, SPNEGO_OID + der.tagged(0, inner))

lib/core/enums.py

@@ -439,6 +439,7 @@ class AUTH_TYPE(object):
     DIGEST = "digest"
     BEARER = "bearer"
     NTLM = "ntlm"
+    NEGOTIATE = "negotiate"
     PKI = "pki"
 
 class AUTOCOMPLETE_TYPE(object):

lib/core/option.py

@@ -1332,8 +1332,11 @@ def _setHTTPHandlers():
         # proxy - the keep-alive handler pools the proxy socket for plain HTTP and the CONNECT-tunnelled
         # socket per origin for HTTPS); '--no-keep-alive' opts out, and they are automatically disabled
         # when incompatible (authentication methods, or chunked transfer-encoding of the request body -
-        # handled by a dedicated, non-pooling handler)
+        # handled by a dedicated, non-pooling handler). Negotiate is the one auth exception: its token is
-        conf.keepAlive = not conf.noKeepAlive and not conf.authType and not conf.chunked
+        # a per-request, end-to-end header (minted fresh each request, no connection-bound handshake), so
+        # persistent connections remain safe and worthwhile.
+        negotiateAuth = (conf.authType or "").lower() == AUTH_TYPE.NEGOTIATE
+        conf.keepAlive = not conf.noKeepAlive and not conf.chunked and (not conf.authType or negotiateAuth)
 
         if conf.keepAlive:
             # persistent connections for both HTTP and HTTPS; the keep-alive HTTPS
@@ -1449,7 +1452,7 @@ def _setAuthCred():
 
 def _setHTTPAuthentication():
     """
-    Check and set the HTTP(s) authentication method (Basic, Digest, Bearer, NTLM or PKI),
+    Check and set the HTTP(s) authentication method (Basic, Digest, Bearer, NTLM, Negotiate or PKI),
     username and password for first three methods, or PEM private key file for
     PKI authentication
     """
@@ -1472,9 +1475,9 @@ def _setHTTPAuthentication():
         errMsg += "but did not provide the type (e.g. --auth-type=\"basic\")"
         raise SqlmapSyntaxException(errMsg)
 
-    elif (conf.authType or "").lower() not in (AUTH_TYPE.BASIC, AUTH_TYPE.DIGEST, AUTH_TYPE.BEARER, AUTH_TYPE.NTLM, AUTH_TYPE.PKI):
+    elif (conf.authType or "").lower() not in (AUTH_TYPE.BASIC, AUTH_TYPE.DIGEST, AUTH_TYPE.BEARER, AUTH_TYPE.NTLM, AUTH_TYPE.NEGOTIATE, AUTH_TYPE.PKI):
         errMsg = "HTTP authentication type value must be "
-        errMsg += "Basic, Digest, Bearer, NTLM or PKI"
+        errMsg += "Basic, Digest, Bearer, NTLM, Negotiate or PKI"
         raise SqlmapSyntaxException(errMsg)
 
     if not conf.authFile:
@@ -1490,11 +1493,12 @@ def _setHTTPAuthentication():
         elif authType == AUTH_TYPE.BEARER:
             conf.httpHeaders.append((HTTP_HEADER.AUTHORIZATION, "Bearer %s" % conf.authCred.strip()))
             return
-        elif authType == AUTH_TYPE.NTLM:
+        elif authType in (AUTH_TYPE.NTLM, AUTH_TYPE.NEGOTIATE):
             # Note: the DOMAIN\username part is colon-free, so the password group takes the full
-            # remainder (a greedy first group would otherwise swallow colons inside the password)
+            # remainder (a greedy first group would otherwise swallow colons inside the password).
+            # For Negotiate, DOMAIN is the Kerberos realm.
             regExp = "^([^:]*\\\\[^:]*):(.*)$"
-            errMsg = "HTTP NTLM authentication credentials value must "
+            errMsg = "HTTP %s authentication credentials value must " % authType
             errMsg += "be in format 'DOMAIN\\username:password'"
         elif authType == AUTH_TYPE.PKI:
             errMsg = "HTTP PKI authentication require "
@@ -1522,6 +1526,12 @@ def _setHTTPAuthentication():
         elif authType == AUTH_TYPE.NTLM:
             from lib.request.ntlm import HTTPNtlmAuthHandler
             authHandler = HTTPNtlmAuthHandler(kb.passwordMgr)
+
+        elif authType == AUTH_TYPE.NEGOTIATE:
+            from lib.request.kerberos import HTTPNegotiateAuthHandler
+            # DOMAIN is the Kerberos realm; the KDC is auto-discovered (env / krb5.conf / DNS SRV / realm)
+            realm, _, user = conf.authUsername.partition('\\')
+            authHandler = HTTPNegotiateAuthHandler(realm, user, conf.authPassword)
     else:
         debugMsg = "setting the HTTP(s) authentication PEM private key"
         logger.debug(debugMsg)

lib/core/settings.py

@@ -20,7 +20,7 @@ from lib.core.enums import OS
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.7.182"
+VERSION = "1.10.7.183"
 TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
 TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
 VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

lib/request/kerberos.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+# Native, dependency-free HTTP "Negotiate" (SPNEGO/Kerberos) authentication, built on the in-tree
+# pure-Python Kerberos client (extra/kerberos). No 'pykerberos'/'gssapi'/'requests-kerberos' needed.
+# The TGT and per-service tickets are obtained once and cached; a fresh AP-REQ is minted for every
+# request from the cached ticket (as replay caches require), so an entire scan costs a single AS+TGS
+# exchange and the token is sent pre-emptively (no extra 401 round-trip per request). Python 2.7 / 3.x.
+
+import base64
+import logging
+import threading
+
+from lib.core.common import getSafeExString
+from lib.core.common import singleTimeLogMessage
+from lib.core.convert import getText
+from lib.core.enums import HTTP_HEADER
+from thirdparty.six.moves import urllib as _urllib
+
+from extra.kerberos.client import getServiceTicket
+from extra.kerberos.client import getTGT
+from extra.kerberos.client import KerberosError
+from extra.kerberos.client import spnegoFromTicket
+from extra.kerberos.discovery import discoverKdc
+
+class HTTPNegotiateAuthHandler(_urllib.request.BaseHandler):
+    handler_order = 480
+
+    def __init__(self, realm, username, password, kdcHost=None, kdcPort=None):
+        self.realm = realm.upper()
+        self.username = username
+        self.password = password
+        self.kdcHost = kdcHost                             # None -> discovered from the realm on first use
+        self.kdcPort = kdcPort
+        self._tgt = None
+        self._tickets = {}                                 # target host -> service-ticket dict
+        self._tgtFailure = None                            # realm-wide failure (bad creds / KDC down)
+        self._hostFailures = {}                            # per-host failure (e.g. no HTTP/<host> SPN)
+        self._lock = threading.Lock()
+
+    def _serviceTicket(self, host):
+        with self._lock:
+            if self._tgtFailure is not None:               # TGT unobtainable -> nothing in the realm works
+                raise self._tgtFailure
+            if host in self._hostFailures:                 # this host already failed -> don't retry it
+                raise self._hostFailures[host]
+            if host not in self._tickets:
+                if self._tgt is None:
+                    if self.kdcHost is None:               # krb5.conf / DNS SRV / realm-name discovery
+                        self.kdcHost, self.kdcPort = discoverKdc(self.realm)
+                    try:
+                        self._tgt = getTGT(self.realm, self.username, self.password, self.kdcHost, self.kdcPort)
+                    except Exception as ex:                # cache so the AS exchange runs at most once
+                        self._tgtFailure = ex
+                        raise
+                try:
+                    self._tickets[host] = getServiceTicket(self._tgt, self.realm, self.username, ["HTTP", host], self.kdcHost, self.kdcPort)
+                except Exception as ex:                    # host-specific -> other hosts remain usable
+                    self._hostFailures[host] = ex
+                    raise
+            return self._tickets[host]
+
+    def _requestHandler(self, req):
+        host = _urllib.parse.urlsplit(req.get_full_url()).hostname
+        if host:
+            try:
+                token = spnegoFromTicket(self._serviceTicket(host), self.realm, self.username)
+                req.add_unredirected_header(HTTP_HEADER.AUTHORIZATION, "Negotiate %s" % getText(base64.b64encode(token)))
+            except KerberosError as ex:
+                # bad credentials / KDC-refused: log once, fall through unauthenticated (server 401s)
+                singleTimeLogMessage("Negotiate (Kerberos) authentication failed: %s" % getSafeExString(ex), logging.ERROR)
+            except Exception as ex:
+                # unreachable KDC, malformed reply, etc. - never let it crash the run
+                singleTimeLogMessage("could not obtain a Kerberos ticket (is the KDC '%s:%s' reachable?): %s" % (self.kdcHost, self.kdcPort, getSafeExString(ex)), logging.ERROR)
+        return req
+
+    def http_request(self, req):
+        return self._requestHandler(req)
+
+    https_request = http_request

tests/test_kerberos.py

@@ -0,0 +1,241 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+
+Tests for the dependency-free Kerberos stack under extra/kerberos: the AES core (FIPS-197), the
+RFC 3961/3962 etype crypto (n-fold, string-to-key, authenticated encryption) and the ASN.1 DER codec.
+All assertions use published FIPS/RFC test vectors, so they validate the crypto and encoding offline
+(the AS/TGS protocol and the HTTP Negotiate handler are exercised against a live KDC, not here).
+"""
+
+import binascii
+import os
+import struct
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _testutils import bootstrap
+bootstrap()
+
+from extra.kerberos import client
+from extra.kerberos import der
+from extra.kerberos import discovery
+from extra.kerberos.aes import AES
+from extra.kerberos.crypto import ENCTYPES, nfold
+
+
+def _dnsName(name):
+    out = bytearray()
+    for label in name.split("."):
+        out.append(len(label))
+        out += label.encode("ascii")
+    out.append(0)
+    return bytes(out)
+
+
+def _h(value):
+    return binascii.unhexlify(value)
+
+
+class TestKerberosAES(unittest.TestCase):
+    def test_fips197_known_answer(self):
+        # FIPS-197 Appendix C.1 (AES-128) and C.3 (AES-256)
+        for key, pt, ct in (
+            ("000102030405060708090a0b0c0d0e0f",
+             "00112233445566778899aabbccddeeff", "69c4e0d86a7b0430d8cdb78070b4c55a"),
+            ("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f",
+             "00112233445566778899aabbccddeeff", "8ea2b7ca516745bfeafc49904b496089"),
+        ):
+            aes = AES(_h(key))
+            self.assertEqual(aes.encryptBlock(_h(pt)), _h(ct))
+            self.assertEqual(aes.decryptBlock(_h(ct)), _h(pt))
+
+    def test_cbc_round_trip(self):
+        aes = AES(_h("00" * 32))
+        iv, data = _h("0f" * 16), os.urandom(64)
+        self.assertEqual(aes.cbcDecrypt(iv, aes.cbcEncrypt(iv, data)), data)
+
+
+class TestKerberosCrypto(unittest.TestCase):
+    def test_nfold_rfc3961(self):
+        # RFC 3961 Appendix A.1
+        for text, size, expected in (
+            ("012345", 8, "be072631276b1955"),
+            ("password", 7, "78a07b6caf85fa"),
+            ("Rough Consensus, and Running Code", 8, "bb6ed30870b7f0e0"),
+            ("password", 21, "59e4a8ca7c0385c3c37b3f6d2000247cb6e6bd5b3e"),
+            ("MASSACHVSETTS INSTITVTE OF TECHNOLOGY", 24,
+             "db3b0d8f0b061e603282b308a50841229ad798fab9540c1b"),
+        ):
+            self.assertEqual(binascii.hexlify(nfold(text.encode(), size)).decode(), expected)
+
+    def test_string2key_rfc3962(self):
+        # RFC 3962 Appendix B (pass 'password', salt 'ATHENA.MIT.EDUraeburn')
+        for iterations, keysize, expected in (
+            (1, 16, "42263c6e89f4fc28b8df68ee09799f15"),
+            (1, 32, "fe697b52bc0d3ce14432ba036a92e65bbb52280990a2fa27883998d72af30161"),
+            (1200, 16, "4c01cd46d632d01e6dbe230a01ed642a"),
+            (1200, 32, "55a6ac740ad17b4846941051e1e8b0a7548d93b0ab30a8bc3ff16280382b8c2a"),
+        ):
+            key = ENCTYPES[17 if keysize == 16 else 18].string2key("password", "ATHENA.MIT.EDUraeburn", iterations)
+            self.assertEqual(binascii.hexlify(key).decode(), expected)
+
+    def test_encrypt_decrypt_round_trip(self):
+        for etype in (17, 18):
+            enc = ENCTYPES[etype]
+            key = os.urandom(enc.keysize)
+            for length in (0, 1, 15, 16, 17, 31, 32, 100):
+                plaintext = os.urandom(length)
+                self.assertEqual(enc.decrypt(key, 1024, enc.encrypt(key, 1024, plaintext)), plaintext)
+
+    def test_integrity_check(self):
+        enc = ENCTYPES[18]
+        key = os.urandom(32)
+        ciphertext = bytearray(enc.encrypt(key, 3, b"secret"))
+        ciphertext[-1] ^= 1
+        self.assertRaises(ValueError, enc.decrypt, key, 3, bytes(ciphertext))
+
+    def test_decrypt_short_ciphertext(self):
+        # a hostile/truncated enc-part (< blocksize + macsize) must raise ValueError, not IndexError
+        enc = ENCTYPES[18]
+        key = os.urandom(32)
+        for length in (0, 1, 12, 27):
+            self.assertRaises(ValueError, enc.decrypt, key, 3, os.urandom(length))
+
+    def test_string2key_bytes_salt(self):
+        # the salt is opaque octets (RFC 3961): a bytes salt must derive the same key as the str form
+        enc = ENCTYPES[18]
+        self.assertEqual(enc.string2key("password", b"ATHENA.MIT.EDUraeburn", 1200),
+                         enc.string2key("password", "ATHENA.MIT.EDUraeburn", 1200))
+
+
+class TestKerberosRC4(unittest.TestCase):
+    def test_nt_hash_string2key(self):
+        # rc4-hmac long-term key is the NT hash: MD4(UTF-16LE(password))
+        self.assertEqual(binascii.hexlify(ENCTYPES[23].string2key("password")).decode(),
+                         "8846f7eaee8fb117ad06bdd830b7586c")
+
+    def test_encrypt_decrypt_round_trip(self):
+        enc = ENCTYPES[23]
+        key = enc.string2key("Secret123")
+        for length in (0, 1, 16, 100):
+            plaintext = os.urandom(length)
+            self.assertEqual(enc.decrypt(key, 1024, enc.encrypt(key, 1024, plaintext)), plaintext)
+
+    def test_integrity_check(self):
+        enc = ENCTYPES[23]
+        key = enc.string2key("x")
+        ciphertext = bytearray(enc.encrypt(key, 3, b"secret"))
+        ciphertext[-1] ^= 1
+        self.assertRaises(ValueError, enc.decrypt, key, 3, bytes(ciphertext))
+
+
+class TestKerberosDER(unittest.TestCase):
+    def test_integer_canonical(self):
+        for value, expected in ((0, "020100"), (127, "02017f"), (128, "02020080"),
+                                (256, "02020100"), (-1, "0201ff"), (-129, "0202ff7f")):
+            self.assertEqual(binascii.hexlify(der.integer(value)).decode(), expected)
+            self.assertEqual(der.decodeInteger(der.peel(der.integer(value))[1]), value)
+
+    def test_application_tags(self):
+        self.assertEqual(bytearray(der.application(10, der.sequence()))[0], 0x6a)   # AS-REQ
+        self.assertEqual(bytearray(der.application(14, der.sequence()))[0], 0x6e)   # AP-REQ
+        self.assertEqual(bytearray(der.tagged(0, der.integer(1)))[0], 0xa0)         # [0] EXPLICIT
+
+    def test_nested_round_trip(self):
+        pname = der.sequence(
+            der.tagged(0, der.integer(1)),
+            der.tagged(1, der.sequenceOf([der.generalString("HTTP"), der.generalString("web.example.com")])),
+        )
+        _, content, _ = der.peel(pname)
+        fields = dict(der.children(content))
+        components = [der.decodeGeneralString(c) for _, c in der.children(der.peel(fields[0xa1])[1])]
+        self.assertEqual(der.decodeInteger(der.peel(fields[0xa0])[1]), 1)
+        self.assertEqual(components, ["HTTP", "web.example.com"])
+
+
+class TestKerberosClient(unittest.TestCase):
+    def test_malformed_reply_raises_kerberoserror(self):
+        # a hostile/truncated KDC reply must surface as KerberosError, never a raw parse exception
+        key, nonce = os.urandom(32), 0x11223344
+        for blob in (b"", b"\x7e\x01", b"\x6b\x02\x30\x00", os.urandom(40)):
+            self.assertRaises(client.KerberosError, client._parseRep, blob, key, 3, nonce, client.AS_REP)
+            self.assertRaises(client.KerberosError, client._replyEtype, blob)
+
+    def test_etype_info2_best_effort(self):
+        # a malformed PA-ETYPE-INFO2 must yield no advertised info (fall back to defaults), not crash
+        self.assertEqual(client._parseEtypeInfo2({12: der.octetString(b"\xff\xff\xff")}), [])
+        self.assertEqual(client._parseEtypeInfo2({}), [])
+
+
+class TestKerberosDiscovery(unittest.TestCase):
+    def test_krb5conf(self):
+        content = ("[realms]\n"
+                   " EXAMPLE.COM = {\n  kdc = dc1.example.com:88\n  admin_server = dc1.example.com\n }\n"
+                   " OTHER.COM = { kdc = other-dc }\n"
+                   # a nested '{ }' block ahead of 'kdc =' must not truncate the realm section
+                   " NESTED.COM = {\n  auth_to_local_names = {\n    joe = joe\n  }\n  kdc = dc.nested.com\n }\n")
+        handle, path = tempfile.mkstemp()
+        os.write(handle, content.encode("utf-8"))
+        os.close(handle)
+        saved = os.environ.get("KRB5_CONFIG")
+        os.environ["KRB5_CONFIG"] = path
+        try:
+            self.assertEqual(discovery._fromKrb5Conf("EXAMPLE.COM"), "dc1.example.com:88")
+            self.assertEqual(discovery._fromKrb5Conf("OTHER.COM"), "other-dc")
+            self.assertEqual(discovery._fromKrb5Conf("NESTED.COM"), "dc.nested.com")
+            self.assertIsNone(discovery._fromKrb5Conf("MISSING.COM"))
+        finally:
+            os.remove(path)
+            os.environ.pop("KRB5_CONFIG", None) if saved is None else os.environ.__setitem__("KRB5_CONFIG", saved)
+
+    def test_split_host_port(self):
+        self.assertEqual(discovery._splitHostPort("dc.example.com"), ("dc.example.com", 88))
+        self.assertEqual(discovery._splitHostPort("dc.example.com:1088"), ("dc.example.com", 1088))
+        self.assertEqual(discovery._splitHostPort("[2001:db8::1]:1088"), ("2001:db8::1", 1088))
+        self.assertEqual(discovery._splitHostPort("[2001:db8::1]"), ("2001:db8::1", 88))
+        self.assertEqual(discovery._splitHostPort("2001:db8::1"), ("2001:db8::1", 88))
+
+    def test_srv_parse(self):
+        header = struct.pack(">HHHHHH", 0x2a2a, 0x8180, 1, 1, 0, 0)
+        question = _dnsName("_kerberos._tcp.EXAMPLE.COM") + struct.pack(">HH", 33, 1)
+        rdata = struct.pack(">HHH", 0, 100, 88) + _dnsName("dc.example.com")
+        answer = b"\xc0\x0c" + struct.pack(">HHIH", 33, 1, 300, len(rdata)) + rdata  # name = ptr to question
+        self.assertEqual(discovery.parseSrv(header + question + answer), [(0, 100, 88, "dc.example.com")])
+
+    def test_srv_parse_hostile_input(self):
+        # a compression-pointer cycle (name at offset 12 points to itself) must not hang or crash
+        cycle = struct.pack(">HHHHHH", 1, 0x8180, 0, 1, 0, 0) + b"\xc0\x0c"
+        self.assertEqual(discovery.parseSrv(cycle), [])
+        self.assertEqual(discovery.parseSrv(b""), [])
+        self.assertEqual(discovery.parseSrv(b"\x00\x00\x81\x80\x00\x00\x00\x05\xff\xff"), [])
+
+    def test_precedence_env_overrides(self):
+        saved = os.environ.get("SQLMAP_KERBEROS_KDC")
+        os.environ["SQLMAP_KERBEROS_KDC"] = "10.0.0.1:8888"
+        try:
+            self.assertEqual(discovery.discoverKdc("EXAMPLE.COM"), ("10.0.0.1", 8888))
+        finally:
+            os.environ.pop("SQLMAP_KERBEROS_KDC", None) if saved is None else os.environ.__setitem__("SQLMAP_KERBEROS_KDC", saved)
+
+    def test_fallback_to_realm(self):
+        savedEnv = os.environ.pop("SQLMAP_KERBEROS_KDC", None)
+        savedCfg = os.environ.get("KRB5_CONFIG")
+        os.environ["KRB5_CONFIG"] = "/nonexistent/sqlmap-krb5.conf"
+        savedDns = discovery._fromDnsSrv
+        discovery._fromDnsSrv = lambda realm: None         # avoid real DNS I/O in the test
+        try:
+            self.assertEqual(discovery.discoverKdc("CORP.EXAMPLE"), ("corp.example", 88))
+        finally:
+            discovery._fromDnsSrv = savedDns
+            if savedEnv is not None:
+                os.environ["SQLMAP_KERBEROS_KDC"] = savedEnv
+            os.environ.pop("KRB5_CONFIG", None) if savedCfg is None else os.environ.__setitem__("KRB5_CONFIG", savedCfg)
+
+
+if __name__ == "__main__":
+    unittest.main()