mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-08-04 14:55:40 +00:00
Hardening Negotiate/Kerberos authentication (#3404)
This commit is contained in:
Miroslav Štampar
parent
1bd1b457d7
commit
ab881132b8
5 changed files with 227 additions and 40 deletions
|
|
@ -10,9 +10,11 @@ See the file 'LICENSE' for copying permission
|
|||
# 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
|
||||
|
|
@ -47,6 +49,14 @@ 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:
|
||||
|
|
@ -87,7 +97,34 @@ def _nonce():
|
|||
return struct.unpack(">I", os.urandom(4))[0] & 0x7fffffff
|
||||
|
||||
def _kerberosTime(offsetSeconds=0):
|
||||
return time.strftime("%Y%m%d%H%M%SZ", time.gmtime(time.time() + offsetSeconds))
|
||||
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(
|
||||
|
|
@ -131,31 +168,56 @@ def _raiseIfError(message):
|
|||
_expString(fields[11]) if 11 in fields else None)
|
||||
return tag, content
|
||||
|
||||
def _parseEtypeInfo2(errorFields):
|
||||
"""Extract [(etype, salt, iterations), ...] from a KDC_ERR_PREAUTH_REQUIRED error's PA-ETYPE-INFO2,
|
||||
telling us which etype/salt/s2kparams the KDC expects for the long-term key."""
|
||||
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."""
|
||||
|
||||
out = []
|
||||
if 12 not in errorFields: # no e-data
|
||||
return out
|
||||
hints = {}
|
||||
try:
|
||||
methodData = der.peel(errorFields[12])[1] # e-data OCTET STRING -> METHOD-DATA (SEQ OF PA-DATA)
|
||||
for _, paData in der.children(der.peel(methodData)[1]):
|
||||
pa = _fields(paData)
|
||||
if 1 in pa and 2 in pa and _expInteger(pa[1]) == PA_ETYPE_INFO2:
|
||||
info = der.peel(pa[2])[1] # padata-value OCTET STRING -> ETYPE-INFO2 (SEQ OF entry)
|
||||
for _, entry in der.children(der.peel(info)[1]):
|
||||
fields = _fields(entry)
|
||||
etype = _expInteger(fields[0])
|
||||
salt = _expOctet(fields[1]) if 1 in fields else None # opaque octets for string2key (RFC 3961), not UTF-8
|
||||
iterations = None
|
||||
if 2 in fields:
|
||||
raw = bytes(der.peel(fields[2])[1]) # s2kparams: 4-byte BE iteration count for AES
|
||||
iterations = struct.unpack(">I", raw)[0] if len(raw) == 4 else None
|
||||
out.append((etype, salt, iterations))
|
||||
hints.setdefault(_expInteger(fields[0]), (salt, iterations))
|
||||
except (KeyError, IndexError, ValueError, struct.error):
|
||||
del out[:] # malformed hint -> fall back to the default etype/salt
|
||||
return out
|
||||
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)."""
|
||||
|
|
@ -196,6 +258,8 @@ def _parseRep(response, key, usage, expectedNonce, expectedType):
|
|||
"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")
|
||||
|
|
@ -211,7 +275,8 @@ def _reqBody(realm, snameType, snameComponents, etypes, nonce, cnameComponents=N
|
|||
parts.append(der.tagged(8, der.sequenceOf([der.integer(_) for _ in etypes]))) # etype
|
||||
return der.sequence(*parts)
|
||||
|
||||
def _authenticator(crealm, cnameComponents, cksum=None):
|
||||
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)),
|
||||
|
|
@ -220,8 +285,10 @@ def _authenticator(crealm, cnameComponents, cksum=None):
|
|||
if cksum is not None:
|
||||
parts.append(der.tagged(3, der.sequence(der.tagged(0, der.integer(cksum[0])),
|
||||
der.tagged(1, der.octetString(cksum[1])))))
|
||||
parts.append(der.tagged(4, der.integer(struct.unpack(">I", os.urandom(4))[0] % 1000000))) # cusec
|
||||
parts.append(der.tagged(5, der.generalizedTime(_kerberosTime()))) # ctime
|
||||
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"):
|
||||
|
|
@ -248,10 +315,10 @@ def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES
|
|||
Follows the standard two-step flow: an initial request without pre-auth learns the KDC's expected
|
||||
etype/salt/iteration-count from PA-ETYPE-INFO2 (so non-default salts and AES-128-only principals
|
||||
work), then a PA-ENC-TIMESTAMP-authenticated request obtains the ticket. Returns
|
||||
{'ticket': <raw Ticket TLV>, 'sessionKey': bytes, 'sessionKeyType': int, 'crealm': str}.
|
||||
{'ticket': <raw Ticket TLV>, 'sessionKey': bytes, 'sessionKeyType': int, 'crealm': str,
|
||||
'endtime': epoch seconds}. 'realm' is used exactly as given (RFC 4120 realms are case-sensitive).
|
||||
"""
|
||||
|
||||
realm = realm.upper()
|
||||
chosenSalt = salt if salt is not None else realm + username
|
||||
|
||||
# 1) probe without pre-auth to discover the etype/salt/iterations (or get the TGT outright)
|
||||
|
|
@ -261,7 +328,10 @@ def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES
|
|||
|
||||
if tag == der.applicationTag(AS_REP): # KDC issued the ticket without pre-auth
|
||||
etype = _replyEtype(response) # derive the key for the etype the KDC actually used
|
||||
clientKey = _enctype(etype).string2key(password, chosenSalt)
|
||||
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
|
||||
|
|
@ -270,19 +340,21 @@ def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES
|
|||
code = _expInteger(errorFields[6]) if 6 in errorFields else -1
|
||||
if code != KDC_ERR_PREAUTH_REQUIRED:
|
||||
raise KerberosError(code, _expString(errorFields[11]) if 11 in errorFields else None)
|
||||
for advertisedEtype, advertisedSalt, advertisedIters in _parseEtypeInfo2(errorFields):
|
||||
if advertisedEtype in ENCTYPES:
|
||||
etype = advertisedEtype
|
||||
iterations = advertisedIters
|
||||
if salt is None and advertisedSalt is not None:
|
||||
chosenSalt = advertisedSalt
|
||||
# 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
|
||||
paTsEnc = der.sequence(der.tagged(0, der.generalizedTime(_kerberosTime())), der.tagged(1, der.integer(0)))
|
||||
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)),
|
||||
|
|
@ -296,10 +368,10 @@ def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES
|
|||
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).
|
||||
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'.
|
||||
"""
|
||||
|
||||
realm = realm.upper()
|
||||
enc = _enctype(tgt["sessionKeyType"])
|
||||
nonce = _nonce()
|
||||
reqBody = _reqBody(realm, NT_SRV_INST, serviceComponents, etypes, nonce)
|
||||
|
|
@ -327,6 +399,7 @@ def spnegoFromTicket(service, realm, username):
|
|||
|
||||
enc = _enctype(service["sessionKeyType"])
|
||||
gssChecksum = (GSS_CHECKSUM_TYPE, struct.pack("<I", 16) + b"\x00" * 16 + struct.pack("<I", GSS_CHECKSUM_FLAGS))
|
||||
authenticator = _authenticator(realm.upper(), [username], cksum=gssChecksum)
|
||||
# seq-number is expected of the GSS mechanism's initial AP-REQ (RFC 4121), so always send one
|
||||
authenticator = _authenticator(realm, [username], cksum=gssChecksum, seqNumber=_nonce())
|
||||
encAuth = enc.encrypt(service["sessionKey"], USAGE_AP_REQ_AUTH, authenticator)
|
||||
return spnego.negTokenInit(_apReq(service["ticket"], encAuth, service["sessionKeyType"]))
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import os
|
|||
import struct
|
||||
|
||||
from extra.kerberos.aes import AES, _xor # _xor reused (no second copy) from the AES core
|
||||
from lib.core.decorators import cachedmethod
|
||||
from lib.request.ntlm import _md4 # proven RFC 1320 MD4 (reused, not re-derived)
|
||||
|
||||
# RFC 3962 string-to-key default work factor when the KDC advertises no explicit count
|
||||
|
|
@ -122,8 +123,13 @@ class AESEnctype(object):
|
|||
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."""
|
||||
"""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)
|
||||
|
||||
|
|
@ -222,8 +228,9 @@ class RC4Enctype(object):
|
|||
|
||||
@staticmethod
|
||||
def _usage(usage):
|
||||
# RFC 4757 section 3: a couple of Kerberos usages map to Microsoft-specific values
|
||||
return struct.pack("<I", {3: 8, 9: 8}.get(usage, usage))
|
||||
# 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("<I", {3: 8, 23: 13}.get(usage, usage))
|
||||
|
||||
def encrypt(self, key, usage, plaintext, confounder=None):
|
||||
if confounder is None:
|
||||
|
|
|
|||
|
|
@ -40,18 +40,27 @@ def _fromKrb5Conf(realm):
|
|||
except (IOError, OSError):
|
||||
return None
|
||||
|
||||
header = re.search(r"(?im)^\s*%s\s*=\s*\{" % re.escape(realm), content)
|
||||
# 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(content) and depth > 0:
|
||||
if content[i] == "{":
|
||||
while i < len(realms) and depth > 0:
|
||||
if realms[i] == "{":
|
||||
depth += 1
|
||||
elif content[i] == "}":
|
||||
elif realms[i] == "}":
|
||||
depth -= 1
|
||||
i += 1
|
||||
block = content[start: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
|
||||
|
||||
|
|
@ -135,11 +144,11 @@ def parseSrv(response):
|
|||
offset = _skipName(data, offset)
|
||||
if offset + 10 > len(data):
|
||||
break
|
||||
rtype, _cls, _ttl, rdlength = struct.unpack(">HHIH", bytes(data[offset:offset + 10]))
|
||||
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 rdlength >= 6:
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ See the file 'LICENSE' for copying permission
|
|||
import base64
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
from lib.core.common import getSafeExString
|
||||
from lib.core.common import singleTimeLogMessage
|
||||
|
|
@ -27,10 +28,20 @@ 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
|
||||
|
|
@ -48,6 +59,10 @@ class HTTPNegotiateAuthHandler(_urllib.request.BaseHandler):
|
|||
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
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import os
|
|||
import struct
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
|
@ -26,6 +27,7 @@ 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):
|
||||
|
|
@ -41,6 +43,31 @@ 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)
|
||||
|
|
@ -168,8 +195,64 @@ class TestKerberosClient(unittest.TestCase):
|
|||
|
||||
def test_etype_info2_best_effort(self):
|
||||
# a malformed PA-ETYPE-INFO2 must yield no advertised info (fall back to defaults), not crash
|
||||
self.assertEqual(client._parseEtypeInfo2({12: der.octetString(b"\xff\xff\xff")}), [])
|
||||
self.assertEqual(client._parseEtypeInfo2({}), [])
|
||||
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):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue