mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-08-04 14:55:40 +00:00
324 lines
15 KiB
Python
324 lines
15 KiB
Python
#!/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()
|