diff --git a/data/txt/brotli-dictionary.tx_ b/data/txt/brotli-dictionary.tx_ new file mode 100644 index 00000000000..b8354d0bdb5 Binary files /dev/null and b/data/txt/brotli-dictionary.tx_ differ diff --git a/extra/kerberos/__init__.py b/extra/kerberos/__init__.py new file mode 100644 index 00000000000..2c772879a4f --- /dev/null +++ b/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 +""" diff --git a/extra/kerberos/aes.py b/extra/kerberos/aes.py new file mode 100644 index 00000000000..2cce2db762b --- /dev/null +++ b/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) diff --git a/extra/kerberos/client.py b/extra/kerberos/client.py new file mode 100644 index 00000000000..f8fe44e7f06 --- /dev/null +++ b/extra/kerberos/client.py @@ -0,0 +1,405 @@ +#!/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 calendar +import os +import socket +import struct +import threading +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) +KERBEROS_TIME_FORMAT = "%Y%m%d%H%M%SZ" # RFC 4120 KerberosTime (always UTC) + +# Bounds on the string-to-key work factor a KDC may ask for. The PA-ETYPE-INFO2 hint carrying it +# arrives on an *unauthenticated* KRB-ERROR, and the field is a full 32 bits, so an absurd value would +# either weaken the derived key against offline guessing or burn hours of CPU (RFC 3962 warns about +# both and recommends configurable bounds). A count of 0 nominally means 2**32, which we cannot honour. +MIN_PBKDF2_ITERATIONS = 4096 # the RFC 3962 default; nothing legitimate is lower +MAX_PBKDF2_ITERATIONS = 1000000 + +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(KERBEROS_TIME_FORMAT, time.gmtime(time.time() + offsetSeconds)) + +_timestampLock = threading.Lock() +_lastMicros = -1 + +def _timestamp(): + """(KerberosTime, microseconds) taken from a single clock reading and unique within the process. + + An acceptor's replay cache rejects a repeated (ctime, cusec) for the same principal and service, + and a threaded scan mints an authenticator per request, so the pair must never repeat; a strictly + increasing microsecond counter also keeps cusec inside its INTEGER (0..999999) range by construction. + """ + + global _lastMicros + + with _timestampLock: + micros = max(int(time.time() * 1000000), _lastMicros + 1) + _lastMicros = micros + return time.strftime(KERBEROS_TIME_FORMAT, time.gmtime(micros // 1000000)), micros % 1000000 + +def _expTime(field): + """An [n]-wrapped KerberosTime as epoch seconds (None when absent or unparsable, so an unusual + time format degrades ticket-expiry tracking rather than failing the exchange).""" + + try: + return calendar.timegm(time.strptime(der.decodeGeneralString(der.peel(field)[1]), KERBEROS_TIME_FORMAT)) + except ValueError: + return None + +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 _etypeHints(methodData): + """Parse a METHOD-DATA TLV (SEQUENCE OF PA-DATA) into PA-ETYPE-INFO2 hints as + {etype: (salt, iterations)}, telling us which etype/salt/s2kparams the KDC expects for the + long-term key. The first entry for an etype wins; a malformed hint yields none (so the caller + falls back to its defaults) rather than raising.""" + + hints = {} + try: + 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) + 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 + hints.setdefault(_expInteger(fields[0]), (salt, iterations)) + except (KeyError, IndexError, ValueError, struct.error): + hints.clear() # malformed hint -> fall back to the default etype/salt + return hints + +def _preauthHints(errorFields): + """The etype hints carried by a KDC_ERR_PREAUTH_REQUIRED error's e-data (best effort).""" + + if 12 not in errorFields: # no e-data + return {} + try: + return _etypeHints(der.peel(errorFields[12])[1]) # e-data OCTET STRING -> METHOD-DATA + except (KeyError, IndexError, ValueError, struct.error): + return {} + +def _validatedIterations(iterations): + """Refuse a string-to-key work factor outside local policy. The hint is unauthenticated, so a + spoofed count could either cheapen an offline attack on the PA-ENC-TIMESTAMP we are about to send + or stall the scan for hours; failing loudly beats doing either silently.""" + + if iterations is not None and not MIN_PBKDF2_ITERATIONS <= iterations <= MAX_PBKDF2_ITERATIONS: + raise KerberosError(-1, "KDC advertised an out-of-policy string-to-key iteration count (%d)" % iterations) + return iterations + +def _hintFor(hints, etype, salt, chosenSalt): + """Apply the hint for 'etype': its salt (unless the caller pinned one) and its work factor.""" + + advertisedSalt, iterations = hints.get(etype, (None, None)) + if salt is None and advertisedSalt is not None: + chosenSalt = advertisedSalt + return chosenSalt, _validatedIterations(iterations) + +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]), + # EncKDCRepPart endtime [7]; a scan can outlive the ticket, so the caller can re-fetch + "endtime": _expTime(encKdcRep[7]) if 7 in encKdcRep else None, + } + 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, seqNumber=None): + ctime, cusec = _timestamp() # both from one clock reading, never repeating + 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(cusec))) + parts.append(der.tagged(5, der.generalizedTime(ctime))) + if seqNumber is not None: # [7] seq-number, expected of a GSS AP-REQ + parts.append(der.tagged(7, der.integer(seqNumber))) + 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': , 'sessionKey': bytes, 'sessionKeyType': int, 'crealm': str, + 'endtime': epoch seconds}. 'realm' is used exactly as given (RFC 4120 realms are case-sensitive). + """ + + 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 + rep = _fields(der.peel(der.peel(response)[1])[1]) + # the reply's own padata can still carry the salt/iterations of a non-default principal + chosenSalt, iterations = _hintFor(_etypeHints(rep[2]) if 2 in rep else {}, etype, salt, chosenSalt) + clientKey = _enctype(etype).string2key(password, chosenSalt, iterations) + 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) + # the hint is unauthenticated, so it may only choose among the etypes we actually offered, and + # in *our* order of preference rather than the KDC's (otherwise it could force a downgrade) + hints = _preauthHints(errorFields) + for offered in etypes: + if offered in hints and offered in ENCTYPES: + etype = offered + break + chosenSalt, iterations = _hintFor(hints, etype, salt, chosenSalt) + + enc = _enctype(etype) + clientKey = enc.string2key(password, chosenSalt, iterations) + + # 2) authenticated request with PA-ENC-TIMESTAMP under the discovered etype/salt + patime, pausec = _timestamp() + paTsEnc = der.sequence(der.tagged(0, der.generalizedTime(patime)), der.tagged(1, der.integer(pausec))) + 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). Cross-realm referrals + are not followed, so 'serviceComponents' must name a service inside 'realm'. + """ + + 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", 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]) + + @cachedmethod + def dk(self, key, constant): + """RFC 3961 DK = random-to-key(DR(...)); random-to-key is the identity for AES. + + Cached: it is a pure function of (key, constant), while a scan mints an authenticator per + request from the same handful of long-lived keys, so the pure-Python DR would otherwise be + recomputed for every single one.""" + + 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 (per the + # published errata, usage 9 is NOT folded into 8 - only 3->8 and 23->13 apply) + return struct.pack(" enctype implementation +ENCTYPES = { + 17: AESEnctype(16), + 18: AESEnctype(32), + 23: RC4Enctype(), +} diff --git a/extra/kerberos/der.py b/extra/kerberos/der.py new file mode 100644 index 00000000000..650c78fb799 --- /dev/null +++ b/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") diff --git a/extra/kerberos/discovery.py b/extra/kerberos/discovery.py new file mode 100644 index 00000000000..bed96a63769 --- /dev/null +++ b/extra/kerberos/discovery.py @@ -0,0 +1,200 @@ +#!/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.), 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 + + # scope the search to the [realms] section itself: '[capaths]' uses the identical + # 'realm = { ... }' syntax, so a same-named capath block must not shadow the real one + section = re.search(r"(?im)^[ \t]*\[realms\][ \t]*$", content) + if not section: + return None + nextSection = re.search(r"(?m)^[ \t]*\[", content[section.end():]) + sectionEnd = section.end() + nextSection.start() if nextSection else len(content) + realms = content[section.end():sectionEnd] + + header = re.search(r"(?im)^\s*%s\s*=\s*\{" % re.escape(realm), realms) + if not header: + return None + start = header.end() # brace-match so a nested '{ }' block cannot truncate us + depth, i = 1, start + while i < len(realms) and depth > 0: + if realms[i] == "{": + depth += 1 + elif realms[i] == "}": + depth -= 1 + i += 1 + block = realms[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.) --------------------------------------------------------- +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, rclass, _ttl, rdlength = struct.unpack(">HHIH", bytes(data[offset:offset + 10])) + offset += 10 + if offset + rdlength > len(data): + break + if rtype == _SRV_TYPE and rclass == _IN_CLASS 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 diff --git a/extra/kerberos/spnego.py b/extra/kerberos/spnego.py new file mode 100644 index 00000000000..e4285c45a80 --- /dev/null +++ b/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 " 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)) diff --git a/lib/core/common.py b/lib/core/common.py index 9220dcb9d81..ab5edf895df 100644 --- a/lib/core/common.py +++ b/lib/core/common.py @@ -1594,6 +1594,7 @@ def setPaths(rootPath): paths.SMALL_DICT = os.path.join(paths.SQLMAP_TXT_PATH, "smalldict.txt") paths.USER_AGENTS = os.path.join(paths.SQLMAP_TXT_PATH, "user-agents.txt") paths.WORDLIST = os.path.join(paths.SQLMAP_TXT_PATH, "wordlist.tx_") + paths.BROTLI_DICTIONARY = os.path.join(paths.SQLMAP_TXT_PATH, "brotli-dictionary.tx_") paths.ERRORS_XML = os.path.join(paths.SQLMAP_XML_PATH, "errors.xml") paths.BOUNDARIES_XML = os.path.join(paths.SQLMAP_XML_PATH, "boundaries.xml") paths.QUERIES_XML = os.path.join(paths.SQLMAP_XML_PATH, "queries.xml") diff --git a/lib/core/enums.py b/lib/core/enums.py index 7973d48b3f8..aa8cc4e6561 100644 --- a/lib/core/enums.py +++ b/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): diff --git a/lib/core/option.py b/lib/core/option.py index 816c698891a..5d9163e0c62 100644 --- a/lib/core/option.py +++ b/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) - conf.keepAlive = not conf.noKeepAlive and not conf.authType and not conf.chunked + # handled by a dedicated, non-pooling handler). Negotiate is the one auth exception: its token is + # 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) diff --git a/lib/core/settings.py b/lib/core/settings.py index f4abc3823ff..3cde0d5cad0 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.182" +VERSION = "1.10.7.186" 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) @@ -250,8 +250,16 @@ # Default value for HTTP Accept header HTTP_ACCEPT_HEADER_VALUE = "*/*" -# Default value for HTTP Accept-Encoding header -HTTP_ACCEPT_ENCODING_HEADER_VALUE = "gzip,deflate" +# Whether the interpreter can decode Zstandard responses (stdlib 'compression.zstd', Python 3.14+ / PEP 784) +try: + import compression.zstd as _zstdModule +except ImportError: + _zstdModule = None +HTTP_ZSTD_AVAILABLE = _zstdModule is not None + +# Default value for HTTP Accept-Encoding header (browser-realistic; 'br' via the in-tree decoder, 'zstd' +# only when the stdlib provides it - never advertise a content-coding we cannot decode) +HTTP_ACCEPT_ENCODING_HEADER_VALUE = "gzip, deflate, br%s" % (", zstd" if HTTP_ZSTD_AVAILABLE else "") # Default timeout for running commands over backdoor BACKDOOR_RUN_CMD_TIMEOUT = 5 diff --git a/lib/request/basic.py b/lib/request/basic.py index f3541b60c70..8ed3c06ded4 100644 --- a/lib/request/basic.py +++ b/lib/request/basic.py @@ -12,6 +12,13 @@ import re import zlib +from lib.utils import brotli as _brotli + +try: + from compression import zstd as _zstd # Python 3.14+ stdlib (PEP 784); no third-party dependency +except ImportError: + _zstd = None + from lib.core.common import Backend from lib.core.common import extractErrorMessage from lib.core.common import extractRegexResult @@ -297,7 +304,7 @@ def decodePage(page, contentEncoding, contentType, percentDecode=True): contentEncoding = getText(contentEncoding).lower() if contentEncoding else "" contentType = getText(contentType).lower() if contentType else "" - if contentEncoding in ("gzip", "x-gzip", "deflate"): + if contentEncoding in ("gzip", "x-gzip", "deflate", "br", "zstd"): if not kb.pageCompress: return None @@ -313,6 +320,14 @@ def decodePage(page, contentEncoding, contentType, percentDecode=True): page += obj.flush() if len(page) > MAX_CONNECTION_TOTAL_SIZE: raise Exception("size too large") + elif contentEncoding == "br": + page = _brotli.decompress(page, MAX_CONNECTION_TOTAL_SIZE) # in-tree RFC 7932 decoder (bomb-capped) + elif contentEncoding == "zstd": + if _zstd is None: + raise Exception("no Zstandard decoder available") + page = _zstd.decompress(page) + if len(page) > MAX_CONNECTION_TOTAL_SIZE: + raise Exception("size too large") else: data = gzip.GzipFile("", "rb", 9, io.BytesIO(page)) page = data.read(MAX_CONNECTION_TOTAL_SIZE + 1) diff --git a/lib/request/connect.py b/lib/request/connect.py index e3cd05b470a..de374ed407b 100644 --- a/lib/request/connect.py +++ b/lib/request/connect.py @@ -102,6 +102,7 @@ from lib.core.settings import DEFAULT_USER_AGENT from lib.core.settings import EVALCODE_ENCODED_PREFIX from lib.core.settings import HTTP_ACCEPT_ENCODING_HEADER_VALUE +from lib.core.settings import HTTP_ZSTD_AVAILABLE from lib.core.settings import HTTP_ACCEPT_HEADER_VALUE from lib.core.settings import IPS_WAF_CHECK_PAYLOAD from lib.core.settings import IS_WIN @@ -557,9 +558,11 @@ def getPage(**kwargs): for key, value in list(headers.items()): if key.upper() == HTTP_HEADER.ACCEPT_ENCODING.upper(): # keep only content-codings sqlmap can actually decode (see decodePage): a browser-pasted - # 'Accept-Encoding' (e.g. "gzip, deflate, br, zstd") must not make the server return a body - # we cannot read. Anything else (br, zstd, *, ...) is dropped, falling back to "identity". - value = ','.join(_ for _ in re.split(r"\s*,\s*", value) if _.split(';', 1)[0].strip().lower() in ("gzip", "x-gzip", "deflate", "identity")) or "identity" + # 'Accept-Encoding' must not make the server return a body we cannot read. 'br' is always + # decodable (in-tree decoder), 'zstd' only on interpreters that ship it; anything else is + # dropped, falling back to "identity". + decodable = ["gzip", "x-gzip", "deflate", "br", "identity"] + (["zstd"] if HTTP_ZSTD_AVAILABLE else []) + value = ','.join(_ for _ in re.split(r"\s*,\s*", value) if _.split(';', 1)[0].strip().lower() in decodable) or "identity" del headers[key] if isinstance(value, six.string_types): diff --git a/lib/request/kerberos.py b/lib/request/kerberos.py new file mode 100644 index 00000000000..66cb6112ac2 --- /dev/null +++ b/lib/request/kerberos.py @@ -0,0 +1,99 @@ +#!/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 +import time + +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 + +TICKET_REFRESH_SKEW = 300 # re-fetch a ticket this long before it expires + +def _expiring(ticket): + """True for a cached ticket close enough to its expiry to be worth replacing (a scan can easily + run longer than the ticket lifetime, and an expired AP-REQ is rejected by every acceptor).""" + + return ticket is not None and ticket.get("endtime") is not None and time.time() + TICKET_REFRESH_SKEW >= ticket["endtime"] + +class HTTPNegotiateAuthHandler(_urllib.request.BaseHandler): + handler_order = 480 + + def __init__(self, realm, username, password, kdcHost=None, kdcPort=None): + # Kerberos realms are case-sensitive, but the credentials arrive in the Windows 'DOMAIN\\user' + # form where the domain is not, so normalize here rather than inside the protocol client + 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/ 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 _expiring(self._tickets.get(host)): # drop a ticket that a long scan has outlived + del self._tickets[host] + if _expiring(self._tgt): + self._tgt = None + 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 diff --git a/lib/utils/brotli.py b/lib/utils/brotli.py new file mode 100644 index 00000000000..631a97ae1fc --- /dev/null +++ b/lib/utils/brotli.py @@ -0,0 +1,652 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission +""" + +# Native, dependency-free Brotli (RFC 7932) decompressor, so sqlmap can advertise a browser-realistic +# 'Accept-Encoding: gzip, deflate, br' and read 'Content-Encoding: br' responses (common behind CDNs) +# without pulling in the 'brotli'/'brotlicffi' third-party module. Decode-only: it is used solely to +# inflate server responses (see lib/request/basic.py::decodePage). Validated byte-for-byte against the +# reference encoder across every quality/window/size. The 122 KB static dictionary + context-lookup +# table live ZIP-packed in data/txt/brotli-dictionary.tx_ (same convention as wordlist.tx_). Py 2.7 / 3.x. + +import os +import zipfile + +_DICTIONARY = None # 122784-byte static dictionary (lazy-loaded) +_CONTEXT = None # 2048-byte context-lookup table (4 modes x 2 halves x 256) + +# RFC 7932 Appendix A: words are bucketed by length (4..24); size_bits gives the index width per bucket, +# offsets the cumulative byte offset of each bucket (derived from size_bits; last bucket end == 122784). +_SIZE_BITS = [0, 0, 0, 0, 10, 10, 11, 11, 10, 10, 10, 10, 10, 9, 9, 8, 7, 7, 8, 7, 7, 6, 6, 5, 5] +_OFFSETS = [0] * 25 +for _i in range(24): + _OFFSETS[_i + 1] = _OFFSETS[_i] + ((_i << _SIZE_BITS[_i]) if _SIZE_BITS[_i] else 0) + +# insert-length and copy-length codes (RFC 7932 section 5): (extra bits, base) per code 0..23 +_INS_EXTRA = [0, 0, 0, 0, 0, 0, 1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 7, 8, 9, 10, 12, 14, 24] +_INS_BASE = [0, 1, 2, 3, 4, 5, 6, 8, 10, 14, 18, 26, 34, 50, 66, 98, 130, 194, 322, 578, 1090, 2114, 6210, 22594] +_COPY_EXTRA = [0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 7, 8, 9, 10, 24] +_COPY_BASE = [2, 3, 4, 5, 6, 7, 8, 9, 10, 12, 14, 18, 22, 30, 38, 54, 70, 102, 134, 198, 326, 582, 1094, 2118] +# block-length code (RFC 7932 section 6): (extra bits, base) per code 0..25 +_BLEN_EXTRA = [2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 4, 5, 5, 5, 5, 6, 6, 7, 8, 9, 10, 11, 12, 13, 24] +_BLEN_BASE = [1, 5, 9, 13, 17, 25, 33, 41, 49, 65, 81, 97, 113, 145, 177, 209, 241, 305, 369, 497, 753, 1265, 2289, 4337, 8433, 16625] + +# insert-and-copy command split (RFC 7932 section 5): per command range (code >> 6), the insert/copy +# sub-code base and whether the distance is implicit (codes 0..127 reuse the last distance) +_CMD_RANGE = [(0, 0, True), (0, 8, True), (0, 0, False), (0, 8, False), (8, 0, False), (8, 8, False), + (0, 16, False), (16, 0, False), (8, 16, False), (16, 8, False), (16, 16, False)] + +# code-length-code order and the fixed prefix used to read the 18 code-length code lengths (section 3.5) +_CL_ORDER = [1, 2, 3, 4, 0, 5, 17, 6, 16, 7, 8, 9, 10, 11, 12, 13, 14, 15] +_CLP_LEN = [2, 2, 2, 3, 2, 2, 2, 4, 2, 2, 2, 3, 2, 2, 2, 4] +_CLP_VAL = [0, 4, 3, 2, 0, 4, 3, 1, 0, 4, 3, 2, 0, 4, 3, 5] + +# distance short codes (RFC 7932 section 4): index into the 4-entry distance ring + a signed delta +_DIST_IDX_OFF = [3, 2, 1, 0, 3, 3, 3, 3, 3, 3, 2, 2, 2, 2, 2, 2] +_DIST_VAL_OFF = [0, 0, 0, 0, -1, 1, -2, 2, -3, 3, -1, 1, -2, 2, -3, 3] + +# transform table (RFC 7932 Appendix B): (prefix, transform id, suffix); ids 0=identity, 1..9=omit-last-N, +# 10=uppercase-first, 11=uppercase-all, 12..20=omit-first-N +_TRANSFORMS = [ + (b"", 0, b""), + (b"", 0, b" "), + (b" ", 0, b" "), + (b"", 12, b""), + (b"", 10, b" "), + (b"", 0, b" the "), + (b" ", 0, b""), + (b"s ", 0, b" "), + (b"", 0, b" of "), + (b"", 10, b""), + (b"", 0, b" and "), + (b"", 13, b""), + (b"", 1, b""), + (b", ", 0, b" "), + (b"", 0, b", "), + (b" ", 10, b" "), + (b"", 0, b" in "), + (b"", 0, b" to "), + (b"e ", 0, b" "), + (b"", 0, b"\""), + (b"", 0, b"."), + (b"", 0, b"\">"), + (b"", 0, b"\x0a"), + (b"", 3, b""), + (b"", 0, b"]"), + (b"", 0, b" for "), + (b"", 14, b""), + (b"", 2, b""), + (b"", 0, b" a "), + (b"", 0, b" that "), + (b" ", 10, b""), + (b"", 0, b". "), + (b".", 0, b""), + (b" ", 0, b", "), + (b"", 15, b""), + (b"", 0, b" with "), + (b"", 0, b"'"), + (b"", 0, b" from "), + (b"", 0, b" by "), + (b"", 16, b""), + (b"", 17, b""), + (b" the ", 0, b""), + (b"", 4, b""), + (b"", 0, b". The "), + (b"", 11, b""), + (b"", 0, b" on "), + (b"", 0, b" as "), + (b"", 0, b" is "), + (b"", 7, b""), + (b"", 1, b"ing "), + (b"", 0, b"\x0a\x09"), + (b"", 0, b":"), + (b" ", 0, b". "), + (b"", 0, b"ed "), + (b"", 20, b""), + (b"", 18, b""), + (b"", 6, b""), + (b"", 0, b"("), + (b"", 10, b", "), + (b"", 8, b""), + (b"", 0, b" at "), + (b"", 0, b"ly "), + (b" the ", 0, b" of "), + (b"", 5, b""), + (b"", 9, b""), + (b" ", 10, b", "), + (b"", 10, b"\""), + (b".", 0, b"("), + (b"", 11, b" "), + (b"", 10, b"\">"), + (b"", 0, b"=\""), + (b" ", 0, b"."), + (b".com/", 0, b""), + (b" the ", 0, b" of the "), + (b"", 10, b"'"), + (b"", 0, b". This "), + (b"", 0, b","), + (b".", 0, b" "), + (b"", 10, b"("), + (b"", 10, b"."), + (b"", 0, b" not "), + (b" ", 0, b"=\""), + (b"", 0, b"er "), + (b" ", 11, b" "), + (b"", 0, b"al "), + (b" ", 11, b""), + (b"", 0, b"='"), + (b"", 11, b"\""), + (b"", 10, b". "), + (b" ", 0, b"("), + (b"", 0, b"ful "), + (b" ", 10, b". "), + (b"", 0, b"ive "), + (b"", 0, b"less "), + (b"", 11, b"'"), + (b"", 0, b"est "), + (b" ", 10, b"."), + (b"", 11, b"\">"), + (b" ", 0, b"='"), + (b"", 10, b","), + (b"", 0, b"ize "), + (b"", 11, b"."), + (b"\xc2\xa0", 0, b""), + (b" ", 0, b","), + (b"", 10, b"=\""), + (b"", 11, b"=\""), + (b"", 0, b"ous "), + (b"", 11, b", "), + (b"", 10, b"='"), + (b" ", 10, b","), + (b" ", 11, b"=\""), + (b" ", 11, b", "), + (b"", 11, b","), + (b"", 11, b"("), + (b"", 11, b". "), + (b" ", 11, b"."), + (b"", 11, b"='"), + (b" ", 11, b". "), + (b" ", 10, b"=\""), + (b" ", 11, b"='"), + (b" ", 10, b"='"), +] + + +class BrotliError(Exception): + pass + + +def _loadTables(): + global _DICTIONARY, _CONTEXT + if _DICTIONARY is not None: + return + + path = None + try: + from lib.core.data import paths + path = getattr(paths, "BROTLI_DICTIONARY", None) + except ImportError: + pass + if not path or not os.path.isfile(path): + path = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, "data", "txt", "brotli-dictionary.tx_") + + archive = zipfile.ZipFile(path) # ZIP-packed like wordlist.tx_ / catalog-identifiers.tx_ + try: + raw = archive.read(archive.namelist()[0]) + finally: + archive.close() + if len(raw) != 122784 + 2048: + raise BrotliError("invalid Brotli dictionary table") + _DICTIONARY = raw[:122784] + _CONTEXT = bytearray(raw[122784:]) + + +class _BitReader(object): + __slots__ = ("data", "size", "pos", "acc", "bits") + + def __init__(self, data): + self.data = bytearray(data) + self.size = len(self.data) + self.pos = 0 + self.acc = 0 + self.bits = 0 + + def _fill(self): + while self.bits <= 24 and self.pos < self.size: + self.acc |= self.data[self.pos] << self.bits + self.pos += 1 + self.bits += 8 + + def readBits(self, count): + if count == 0: + return 0 + if self.bits < count: + self._fill() + value = self.acc & ((1 << count) - 1) + self.acc >>= count + self.bits -= count + return value + + def peek(self, count): + if self.bits < count: + self._fill() + return self.acc & ((1 << count) - 1) + + def drop(self, count): + self.acc >>= count + self.bits -= count + + def alignToByte(self): + drop = self.bits & 7 + if drop: + self.acc >>= drop + self.bits -= drop + + def readBytes(self, count): + out = bytearray() + while count > 0 and self.bits >= 8: + out.append(self.acc & 0xff) + self.acc >>= 8 + self.bits -= 8 + count -= 1 + if count > 0: + out += self.data[self.pos:self.pos + count] + self.pos += count + return bytes(out) + + +def _reverseBits(value, count): + result = 0 + for _ in range(count): + result = (result << 1) | (value & 1) + value >>= 1 + return result + + +class _Huffman(object): + __slots__ = ("maxLength", "table", "single") + + def __init__(self, lengths): + self.single = None + self.table = None + self.maxLength = max(lengths) if lengths else 0 + if self.maxLength == 0: + self.single = 0 + for symbol, length in enumerate(lengths): + if length > 0: + self.single = symbol + return + + counts = [0] * (self.maxLength + 1) + for length in lengths: + if length: + counts[length] += 1 + nextCode = [0] * (self.maxLength + 2) + code = 0 + for bits in range(1, self.maxLength + 1): + code = (code + counts[bits - 1]) << 1 + nextCode[bits] = code + + self.table = [(0, 0)] * (1 << self.maxLength) + for symbol, length in enumerate(lengths): + if length: + reversed_ = _reverseBits(nextCode[length], length) + nextCode[length] += 1 + step = 1 << length + for index in range(reversed_, 1 << self.maxLength, step): + self.table[index] = (symbol, length) + + def decode(self, reader): + if self.table is None: + return self.single + symbol, length = self.table[reader.peek(self.maxLength)] + reader.drop(length) + return symbol + + +def _readSimplePrefix(reader, alphabetSize): + count = reader.readBits(2) + 1 + symbolBits = (alphabetSize - 1).bit_length() or 1 + symbols = [reader.readBits(symbolBits) for _ in range(count)] + lengths = [0] * alphabetSize + if count == 1: + huffman = _Huffman([]) + huffman.single = symbols[0] + return huffman + if count == 2: + pairs = [(symbols[0], 1), (symbols[1], 1)] + elif count == 3: + pairs = [(symbols[0], 1), (symbols[1], 2), (symbols[2], 2)] + elif reader.readBits(1): + pairs = [(symbols[0], 1), (symbols[1], 2), (symbols[2], 3), (symbols[3], 3)] + else: + pairs = [(symbols[0], 2), (symbols[1], 2), (symbols[2], 2), (symbols[3], 2)] + for symbol, length in pairs: + lengths[symbol] = length + return _Huffman(lengths) + + +def _readComplexPrefix(reader, alphabetSize, skip): + codeLengths = [0] * 18 + space = 32 + for symbol in _CL_ORDER[skip:]: + index = reader.peek(4) + codeLengths[symbol] = _CLP_VAL[index] + reader.drop(_CLP_LEN[index]) + if codeLengths[symbol]: + space -= 32 >> codeLengths[symbol] + if space <= 0: + break + codeLengthHuffman = _Huffman(codeLengths) + + lengths = [0] * alphabetSize + symbol = 0 + previous = 8 + repeat = 0 + repeatLength = 0 + space = 32768 + while symbol < alphabetSize and space > 0: + code = codeLengthHuffman.decode(reader) + if code < 16: + lengths[symbol] = code + symbol += 1 + if code: + previous = code + space -= 32768 >> code + repeat = 0 + else: + extra = 2 if code == 16 else 3 + newLength = previous if code == 16 else 0 + if repeatLength != newLength: + repeat = 0 + repeatLength = newLength + old = repeat + delta = reader.readBits(extra) + if repeat > 0: + repeat = (repeat - 2) << extra + repeat += delta + 3 + emit = repeat - old + for _ in range(emit): + if symbol >= alphabetSize: + break + lengths[symbol] = repeatLength + symbol += 1 + if repeatLength: + space -= emit << (15 - repeatLength) + return _Huffman(lengths) + + +def _readPrefix(reader, alphabetSize): + header = reader.readBits(2) + if header == 1: + return _readSimplePrefix(reader, alphabetSize) + return _readComplexPrefix(reader, alphabetSize, header) + + +def _readBlockTypeCount(reader): + if not reader.readBits(1): + return 1 + bits = reader.readBits(3) + return (1 << bits) + 1 + reader.readBits(bits) + + +def _readContextMap(reader, treeCount, size): + maxRun = reader.readBits(4) + 1 if reader.readBits(1) else 0 + huffman = _readPrefix(reader, treeCount + maxRun) + contextMap = [] + while len(contextMap) < size: + code = huffman.decode(reader) + if code == 0: + contextMap.append(0) + elif code <= maxRun: + contextMap.extend([0] * ((1 << code) + reader.readBits(code))) + else: + contextMap.append(code - maxRun) + del contextMap[size:] + if reader.readBits(1): # inverse move-to-front + moveToFront = list(range(256)) + for i in range(len(contextMap)): + index = contextMap[i] + value = moveToFront[index] + contextMap[i] = value + del moveToFront[index] + moveToFront.insert(0, value) + return contextMap + + +def _toUpperCase(word, offset): + char = word[offset] + if char < 0xc0: # ASCII: flip case of a-z + if 97 <= char <= 122: + word[offset] = char ^ 32 + return 1 + if char < 0xe0: # 2-byte UTF-8 + if offset + 1 < len(word): + word[offset + 1] ^= 32 + return 2 + if offset + 2 < len(word): # 3-byte UTF-8 + word[offset + 2] ^= 5 + return 3 + + +def _applyTransform(transformId, word): + prefix, kind, suffix = _TRANSFORMS[transformId] + result = bytearray(word) + if kind == 0: + pass + elif 1 <= kind <= 9: # omit last N + result = result[:len(result) - kind] if len(result) >= kind else bytearray() + elif 12 <= kind <= 20: # omit first N + count = kind - 11 + result = result[count:] if len(result) >= count else bytearray() + elif kind == 10: # uppercase first + if result: + _toUpperCase(result, 0) + elif kind == 11: # uppercase all + offset = 0 + while offset < len(result): + offset += _toUpperCase(result, offset) + return prefix + bytes(result) + suffix + + +def decompress(data, maxOutput=100 * 1024 * 1024): + """Decompress a Brotli (RFC 7932) stream, returning the original bytes. Raises BrotliError on a + malformed stream or if the output would exceed 'maxOutput' (an anti-decompression-bomb cap).""" + + _loadTables() + dictionary = _DICTIONARY + context = _CONTEXT + + try: + reader = _BitReader(data) + header = reader.readBits(1) + if header == 0: + windowBits = 16 + else: + header = reader.readBits(3) + if header: + windowBits = 17 + header + else: + header = reader.readBits(3) + windowBits = (8 + header) if header else 17 + maxBackward = (1 << windowBits) - 16 + + out = bytearray() + distRing = [16, 15, 11, 4] + distIndex = 0 + + while True: + isLast = reader.readBits(1) + if isLast and reader.readBits(1): # ISLASTEMPTY + break + + nibbles = reader.readBits(2) + if nibbles == 3: # metadata block (no output) + if reader.readBits(1): + raise BrotliError("reserved bit set") + skipBytes = reader.readBits(2) + if skipBytes: + skipLength = reader.readBits(skipBytes * 8) + 1 + reader.alignToByte() + reader.readBytes(skipLength) + if isLast: + break + continue + + metaLength = reader.readBits((nibbles + 4) * 4) + 1 + if len(out) + metaLength > maxOutput: # reject an over-large block up front (anti-bomb) + raise BrotliError("output too large") + if not isLast and reader.readBits(1): # ISUNCOMPRESSED + reader.alignToByte() + out += reader.readBytes(metaLength) + if len(out) > maxOutput: + raise BrotliError("output too large") + continue + + typesL = _readBlockTypeCount(reader) + blockL, typeHuffmanL, lengthHuffmanL, prevTypeL = 1 << 28, None, None, 1 + typeL = 0 + if typesL >= 2: + typeHuffmanL = _readPrefix(reader, typesL + 2) + lengthHuffmanL = _readPrefix(reader, 26) + code = lengthHuffmanL.decode(reader) + blockL = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + + typesI = _readBlockTypeCount(reader) + blockI, typeHuffmanI, lengthHuffmanI, prevTypeI = 1 << 28, None, None, 1 + typeI = 0 + if typesI >= 2: + typeHuffmanI = _readPrefix(reader, typesI + 2) + lengthHuffmanI = _readPrefix(reader, 26) + code = lengthHuffmanI.decode(reader) + blockI = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + + typesD = _readBlockTypeCount(reader) + blockD, typeHuffmanD, lengthHuffmanD, prevTypeD = 1 << 28, None, None, 1 + typeD = 0 + if typesD >= 2: + typeHuffmanD = _readPrefix(reader, typesD + 2) + lengthHuffmanD = _readPrefix(reader, 26) + code = lengthHuffmanD.decode(reader) + blockD = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + + postfix = reader.readBits(2) + direct = reader.readBits(4) << postfix + contextModes = [reader.readBits(2) for _ in range(typesL)] + + treesL = _readBlockTypeCount(reader) + contextMapL = _readContextMap(reader, treesL, typesL * 64) if treesL >= 2 else [0] * (typesL * 64) + treesD = _readBlockTypeCount(reader) + contextMapD = _readContextMap(reader, treesD, typesD * 4) if treesD >= 2 else [0] * (typesD * 4) + + huffmanL = [_readPrefix(reader, 256) for _ in range(treesL)] + huffmanI = [_readPrefix(reader, 704) for _ in range(typesI)] + distanceAlphabet = 16 + direct + (48 << postfix) + huffmanD = [_readPrefix(reader, distanceAlphabet) for _ in range(treesD)] + + produced = 0 + while produced < metaLength: + if blockI == 0: + code = typeHuffmanI.decode(reader) + nextType = prevTypeI if code == 0 else ((typeI + 1) % typesI if code == 1 else code - 2) + prevTypeI, typeI = typeI, nextType + code = lengthHuffmanI.decode(reader) + blockI = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + blockI -= 1 + + command = huffmanI[typeI].decode(reader) + insertBase, copyBase, implicit = _CMD_RANGE[command >> 6] + insertCode = insertBase + ((command >> 3) & 7) + copyCode = copyBase + (command & 7) + insertLength = _INS_BASE[insertCode] + reader.readBits(_INS_EXTRA[insertCode]) + copyLength = _COPY_BASE[copyCode] + reader.readBits(_COPY_EXTRA[copyCode]) + # a well-formed command never inserts beyond the meta-block; bounding here keeps a hostile + # stream from spinning the literal loop far past the output cap before it is caught (the + # copy length is checked in the back-reference branch, and dictionary copies are <= 24) + if produced + insertLength > metaLength: + raise BrotliError("insert exceeds meta-block length") + + for _ in range(insertLength): + if blockL == 0: + code = typeHuffmanL.decode(reader) + nextType = prevTypeL if code == 0 else ((typeL + 1) % typesL if code == 1 else code - 2) + prevTypeL, typeL = typeL, nextType + code = lengthHuffmanL.decode(reader) + blockL = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + blockL -= 1 + mode = contextModes[typeL] * 512 + p1 = out[-1] if out else 0 + p2 = out[-2] if len(out) >= 2 else 0 + contextId = context[mode + p1] | context[mode + 256 + p2] + out.append(huffmanL[contextMapL[64 * typeL + contextId]].decode(reader)) + produced += 1 + + if produced >= metaLength: + break + + if implicit: + distanceCode = 0 + else: + if blockD == 0: + code = typeHuffmanD.decode(reader) + nextType = prevTypeD if code == 0 else ((typeD + 1) % typesD if code == 1 else code - 2) + prevTypeD, typeD = typeD, nextType + code = lengthHuffmanD.decode(reader) + blockD = _BLEN_BASE[code] + reader.readBits(_BLEN_EXTRA[code]) + blockD -= 1 + distanceContext = min(copyLength - 2, 3) if copyLength >= 2 else 0 + distanceCode = huffmanD[contextMapD[4 * typeD + distanceContext]].decode(reader) + + if distanceCode < 16: + distance = distRing[(distIndex + _DIST_IDX_OFF[distanceCode]) & 3] + _DIST_VAL_OFF[distanceCode] + else: + value = distanceCode - 16 + if value < direct: + distance = value + 1 + else: + value -= direct + extraBits = 1 + (value >> (postfix + 1)) + extra = reader.readBits(extraBits) + high = value >> postfix + low = value & ((1 << postfix) - 1) + distance = ((((2 + (high & 1)) << extraBits) - 4 + extra) << postfix) + low + direct + 1 + + maxDistance = min(len(out), maxBackward) + if distanceCode != 0 and distance <= maxDistance: + distRing[distIndex & 3] = distance + distIndex += 1 + + if distance <= maxDistance: # ordinary back-reference (may overlap) + if produced + copyLength > metaLength: # can't copy past the block (also bounds the loop) + raise BrotliError("copy exceeds meta-block length") + source = len(out) - distance + for i in range(copyLength): + out.append(out[source + i]) + produced += 1 + else: # static-dictionary reference + offset = distance - maxDistance - 1 + if not (4 <= copyLength <= 24) or _SIZE_BITS[copyLength] == 0: + raise BrotliError("invalid dictionary reference") + bits = _SIZE_BITS[copyLength] + index = offset & ((1 << bits) - 1) + transformId = offset >> bits + if transformId >= len(_TRANSFORMS): + raise BrotliError("invalid dictionary transform") + start = _OFFSETS[copyLength] + index * copyLength + word = _applyTransform(transformId, dictionary[start:start + copyLength]) + out += word + produced += len(word) + + if len(out) > maxOutput: + raise BrotliError("output too large") + + if isLast: + break + return bytes(out) + except BrotliError: + raise + except Exception as ex: + raise BrotliError("malformed Brotli stream (%s)" % ex) diff --git a/tests/test_brotli.py b/tests/test_brotli.py new file mode 100644 index 00000000000..42525036661 --- /dev/null +++ b/tests/test_brotli.py @@ -0,0 +1,69 @@ +#!/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 Brotli (RFC 7932) decompressor under lib/utils/brotli.py. The compressed +fixtures were produced by the reference encoder at various quality levels; the expected plaintext is +reconstructed here by construction, so the suite validates the decoder fully offline (no third-party +'brotli' module at test time) on Python 2.7 / 3.x. Cases deliberately exercise the static dictionary + +word transforms, long overlapping copies, UTF-8, the low-quality (near-uniform tree) path and the +repetitive content that relies on the higher insert-and-copy command ranges. +""" + +import binascii +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _testutils import bootstrap +bootstrap() + +from lib.utils.brotli import decompress +from lib.utils.brotli import BrotliError + + +# (expected plaintext, reference-compressed stream in hex) +_CASES = [ + (b"", "3b"), + (b"The quick brown fox jumps over the lazy dog.", + "8b158054686520717569636b2062726f776e20666f78206a756d7073206f76657220746865206c617a7920646f672e03"), + (b"the time of the data on the site is now and the code" * 3, + "1b9b000004e164a9be171b85c00636e080bd799109f64571f090852e78334d90cb20ac9c346c190ff171965a3d2f7a90171c"), + (b"the quick brown fox " * 30, + "1b570200047463a92ee78362f22082d628041695d90acef9a3f135e9c701"), + (b"AB" * 400, "1b1f0300a48284a2b230b009"), + ((u"caf\xe9 na\xefve \u4f60\u597d ").encode("utf-8") * 12, + "1bef00004427477ad6d60ac38c93200a288ab462c2a06461d22d186dbbe0263e0707"), + (b"hello hello hello world world foo bar baz " * 6, + "8b7d000080aaaaaaeaff74e5f355048415f8c0000c201701d0ffbbeadf736f75cfa82e6f63b82b5e2c2c2c6c6cacea654675f0e1c38fc160308e33595583c16030180ce65067442a4aa370586827d97b828968074727f5b21e97eebd045d8baeefef94c3fca4fb1e"), +] + + +class TestBrotli(unittest.TestCase): + def test_known_fixtures(self): + for expected, hexstream in _CASES: + self.assertEqual(decompress(binascii.unhexlify(hexstream)), expected) + + def test_empty_stream(self): + self.assertEqual(decompress(binascii.unhexlify("3b")), b"") + + def test_malformed_raises(self): + # a hostile/truncated stream must surface as BrotliError, never a raw exception + for blob in (b"\xff", b"\x00\x00\x00", b"\x1b\xff\xff\xff\xff", os.urandom(32)): + try: + decompress(blob) + except BrotliError: + pass + except Exception as ex: + self.fail("non-BrotliError on malformed input: %s" % ex) + + def test_bomb_cap(self): + # a small stream must not be allowed to expand past the output cap + self.assertRaises(BrotliError, decompress, binascii.unhexlify("1b1f0300a48284a2b230b009"), 16) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_kerberos.py b/tests/test_kerberos.py new file mode 100644 index 00000000000..86971695e8f --- /dev/null +++ b/tests/test_kerberos.py @@ -0,0 +1,324 @@ +#!/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 time +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 +from lib.request.kerberos import _expiring + + +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) + + +def _etypeInfo2Entry(etype, salt=None, iterations=None): + parts = [der.tagged(0, der.integer(etype))] + if salt is not None: + parts.append(der.tagged(1, der.generalString(salt))) + if iterations is not None: + parts.append(der.tagged(2, der.octetString(struct.pack(">I", iterations)))) + return der.sequence(*parts) + + +def _preauthError(entries): + """The error-field map a KDC_ERR_PREAUTH_REQUIRED reply advertising 'entries' would produce.""" + + paData = der.sequence( + der.tagged(1, der.integer(19)), # PA-ETYPE-INFO2 + der.tagged(2, der.octetString(der.sequenceOf(entries))), + ) + return {12: der.octetString(der.sequenceOf([paData]))} + + +def _selectEtype(offered, hints): + """getTGT's etype choice: the client's own preference order, restricted to what it offered.""" + + return next((_ for _ in offered if _ in hints and _ in ENCTYPES), None) + + +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._preauthHints({12: der.octetString(b"\xff\xff\xff")}), {}) + self.assertEqual(client._preauthHints({}), {}) + self.assertEqual(client._etypeHints(der.octetString(b"\xff\xff\xff")), {}) + + def test_etype_info2_hints(self): + hints = client._preauthHints(_preauthError([_etypeInfo2Entry(18, "SALT", 4096), + _etypeInfo2Entry(23)])) + self.assertEqual(hints, {18: (b"SALT", 4096), 23: (None, None)}) + + def test_iteration_count_policy(self): + # the hint is unauthenticated: a count that would cheapen an offline attack or stall the scan + # for hours must be refused, and 0 (nominally 2**32) is not silently taken as the default + self.assertIsNone(client._validatedIterations(None)) + self.assertEqual(client._validatedIterations(4096), 4096) + for bogus in (0, 1, 1000, client.MAX_PBKDF2_ITERATIONS + 1, 0xFFFFFFFF): + self.assertRaises(client.KerberosError, client._validatedIterations, bogus) + + def test_hint_cannot_override_pinned_salt(self): + hints = {18: (b"KDCSALT", 4096)} + self.assertEqual(client._hintFor(hints, 18, "PINNED", "PINNED"), ("PINNED", 4096)) + self.assertEqual(client._hintFor(hints, 18, None, "DEFAULT"), (b"KDCSALT", 4096)) + self.assertEqual(client._hintFor(hints, 17, None, "DEFAULT"), ("DEFAULT", None)) + + def test_etype_selection_honours_client_preference(self): + # a spoofed hint must not be able to pull the client onto an etype it never offered, and the + # client's own preference order wins over the KDC's + hints = client._preauthHints(_preauthError([_etypeInfo2Entry(23), _etypeInfo2Entry(18)])) + self.assertEqual(_selectEtype((18, 17), hints), 18) # KDC listed rc4 first + self.assertEqual(_selectEtype((17, 18), hints), 18) # only 18 is hinted + self.assertIsNone(_selectEtype((18, 17), client._preauthHints(_preauthError([_etypeInfo2Entry(23)])))) + + def test_authenticator_timestamps_are_unique(self): + # an acceptor's replay cache keys on (ctime, cusec), and a threaded scan mints one per request + stamps = [client._timestamp() for _ in range(2000)] + self.assertEqual(len(set(stamps)), len(stamps)) + self.assertTrue(all(0 <= cusec <= 999999 for _, cusec in stamps)) + + def test_authenticator_carries_seq_number(self): + # RFC 4121 expects a sequence number in the GSS mechanism's initial AP-REQ authenticator + fields = client._fields(der.peel(der.peel( + client._authenticator("EXAMPLE.COM", ["user"], seqNumber=0x11223344))[1])[1]) + self.assertEqual(client._expInteger(fields[7]), 0x11223344) + self.assertNotIn(7, client._fields(der.peel(der.peel( + client._authenticator("EXAMPLE.COM", ["user"]))[1])[1])) + + def test_kerberos_time_round_trip(self): + self.assertEqual(client._expTime(der.generalizedTime("19700101000010Z")), 10) + self.assertIsNone(client._expTime(der.generalizedTime("not-a-time"))) + + +class TestKerberosTicketCache(unittest.TestCase): + def test_expiring(self): + now = time.time() + self.assertFalse(_expiring(None)) + self.assertFalse(_expiring({"endtime": None})) # a KDC that sent no parsable endtime + self.assertFalse(_expiring({"endtime": now + 36000})) + self.assertTrue(_expiring({"endtime": now - 1})) # already expired + self.assertTrue(_expiring({"endtime": now + 60})) # inside the refresh skew + + +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()