mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-07-10 18:43:07 +00:00
Switching from pickle to json for serialization
This commit is contained in:
Miroslav Štampar
parent
e2c55c1ddc
commit
4504edaeb1
12 changed files with 933 additions and 273 deletions
|
|
@ -21,7 +21,7 @@ bootstrap()
|
|||
|
||||
from lib.core.convert import (decodeHex, encodeHex, decodeBase64, encodeBase64,
|
||||
getBytes, getText, getUnicode, getOrds,
|
||||
jsonize, dejsonize, base64pickle, base64unpickle)
|
||||
jsonize, dejsonize, serializeValue, deserializeValue)
|
||||
from lib.core.common import decodeDbmsHexValue
|
||||
|
||||
try:
|
||||
|
|
@ -109,36 +109,31 @@ class TestJson(unittest.TestCase):
|
|||
self.assertEqual(jsonize(123), "123") # int -> textual "123"
|
||||
|
||||
|
||||
class TestBase64Pickle(unittest.TestCase):
|
||||
# Types sqlmap actually serializes (injection objects, cached values, BigArray).
|
||||
class TestSerialize(unittest.TestCase):
|
||||
# Smoke coverage of the safe (JSON-based) session serializer; the exhaustive corner-case
|
||||
# and security suite lives in tests/test_serialize.py.
|
||||
def test_roundtrip_allowed_types(self):
|
||||
for obj in [[1, 2, 3], {"a": 1}, (1, 2), "text", 42, 3.14, True, None, {"k": [1, {"n": "v"}]}]:
|
||||
self.assertEqual(base64unpickle(base64pickle(obj)), obj)
|
||||
self.assertEqual(deserializeValue(serializeValue(obj)), obj)
|
||||
|
||||
# REGRESSION: under Python 3 + PICKLE_PROTOCOL=2 a raw `bytes` value is pickled via the
|
||||
# `_codecs.encode` global. The RestrictedUnpickler allowlist (patch.py) once rejected that,
|
||||
# so any serialized session value containing bytes failed to load on py3. The fix allows
|
||||
# exactly `_codecs.encode` (a benign codec call). Bytes MUST round-trip on both py2 and py3.
|
||||
def test_bytes_roundtrip(self):
|
||||
for raw in [b"x", b"\x00\x01\xff", b"\xde\xad\xbe\xef"]:
|
||||
self.assertEqual(base64unpickle(base64pickle(raw)), raw, msg="bytes round-trip %r" % raw)
|
||||
self.assertEqual(deserializeValue(serializeValue(raw)), raw, msg="bytes round-trip %r" % raw)
|
||||
|
||||
def test_bytes_nested_in_container_roundtrip(self):
|
||||
for obj in [{"a": b"bytes"}, [b"ab", "s", 1, None], ("t", b"\xde\xad")]:
|
||||
self.assertEqual(base64unpickle(base64pickle(obj)), obj, msg="nested-bytes round-trip %r" % (obj,))
|
||||
self.assertEqual(deserializeValue(serializeValue(obj)), obj, msg="nested-bytes round-trip %r" % (obj,))
|
||||
|
||||
def test_dangerous_globals_still_blocked(self):
|
||||
# bootstrap() installs sqlmap's RestrictedUnpickler over pickle.loads. These are VALID
|
||||
# pickles that reference os.system / builtins.eval - stdlib would import them happily; the
|
||||
# allowlist must reject them. Assert the SPECIFIC "forbidden" ValueError (not just any
|
||||
# error) so the test proves the allowlist fired, not that the bytes failed to parse.
|
||||
import pickle
|
||||
for payload in (b"cos\nsystem\n.", b"c__builtin__\neval\n."):
|
||||
try:
|
||||
pickle.loads(payload)
|
||||
self.fail("dangerous global was NOT blocked: %r" % payload)
|
||||
except ValueError as ex:
|
||||
self.assertIn("forbidden", str(ex), msg="unexpected error for %r: %s" % (payload, ex))
|
||||
def test_output_is_plain_ascii_text_no_base64(self):
|
||||
# the serialized form must be readable JSON text (not Base64 / binary) so it lands verbatim in
|
||||
# the TEXT session column with zero wrapping overhead
|
||||
out = serializeValue({"k": [1, (2, 3)]})
|
||||
self.assertIsInstance(out, str)
|
||||
self.assertTrue(out.startswith("{") and out.endswith("}"), out)
|
||||
|
||||
def test_deserialize_accepts_bytes(self):
|
||||
# BigArray hands the serialized data back as bytes (off a compressed disk chunk)
|
||||
self.assertEqual(deserializeValue(getBytes(serializeValue([1, (2, 3)]))), [1, (2, 3)])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
462
tests/test_serialize.py
Normal file
462
tests/test_serialize.py
Normal file
|
|
@ -0,0 +1,462 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
|
||||
Exhaustive lock on the safe (JSON-based, no code execution) serializer that backs the
|
||||
session store (HashDB) and BigArray disk chunks - the replacement for the former
|
||||
code-executing serializer.
|
||||
|
||||
Two properties must hold forever, on BOTH Python 2.7 and 3.x:
|
||||
|
||||
1. CORRECTNESS - every value sqlmap actually persists must round-trip losslessly, with
|
||||
its exact type. The historically fragile cases are covered explicitly: integer/tuple
|
||||
dict keys (naive JSON turns them into strings), tuple-vs-list, set/frozenset, bytes,
|
||||
the AttribDict/InjectionDict/RawPair classes and their internal state, and the native
|
||||
DB-driver scalars (Decimal/datetime/...) that '-d' direct-mode output can contain.
|
||||
A regression here silently corrupts a user's saved session.
|
||||
|
||||
2. SECURITY - deserialization must NEVER execute code and must reconstruct ONLY the small
|
||||
explicit class allowlist. Any other class name (os.system, subprocess.Popen, eval,
|
||||
lib.core.common.shellExec, ...) must be refused with a "forbidden" ValueError, and a
|
||||
crafted legacy payload must fail inertly (no side effects).
|
||||
|
||||
Kept deliberately verbose - each case maps to a real session/BigArray value or a real report.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import decimal
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _testutils import bootstrap
|
||||
bootstrap()
|
||||
|
||||
from lib.core.bigarray import BigArray
|
||||
from lib.core.convert import deserializeValue, encodeBase64, getUnicode, serializeValue
|
||||
from lib.core.convert import _SERIALIZE_TAG
|
||||
from lib.core.datatype import AttribDict, InjectionDict, LRUDict
|
||||
from lib.utils.har import RawPair
|
||||
from thirdparty import six
|
||||
|
||||
INTS = six.integer_types
|
||||
_unichr = six.unichr
|
||||
|
||||
|
||||
def rt(value):
|
||||
"""Full session round-trip: value -> JSON text -> value."""
|
||||
return deserializeValue(serializeValue(value))
|
||||
|
||||
|
||||
def _import_all_modules():
|
||||
"""Import every sqlmap module (like --smoke-test) so class-subclass reflection sees them all."""
|
||||
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
for base in ("lib", "plugins", "tamper"):
|
||||
for dirpath, _, filenames in os.walk(os.path.join(root, base)):
|
||||
if any(skip in dirpath for skip in ("thirdparty", "extra", "tests", "__pycache__")):
|
||||
continue
|
||||
for filename in filenames:
|
||||
if filename.endswith(".py") and filename not in ("__init__.py", "gui.py"):
|
||||
dotted = os.path.join(dirpath, filename[:-3]).replace(root + os.sep, "").replace(os.sep, ".")
|
||||
try:
|
||||
__import__(dotted)
|
||||
except Exception:
|
||||
pass # import health is --smoke-test's job; here we only need the classes that DO load
|
||||
|
||||
|
||||
def _all_subclasses(cls):
|
||||
for sub in cls.__subclasses__():
|
||||
yield sub
|
||||
for _ in _all_subclasses(sub):
|
||||
yield _
|
||||
|
||||
|
||||
class TestScalars(unittest.TestCase):
|
||||
def test_simple(self):
|
||||
for value in [None, True, False, 0, 1, -1, 42, 3.14, -0.5, 0.0]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIs(type(restored) is bool, type(value) is bool) # bool never collapses to int
|
||||
|
||||
def test_big_ints(self):
|
||||
for value in [2 ** 64, 2 ** 200, -(2 ** 128)]:
|
||||
self.assertEqual(rt(value), value)
|
||||
|
||||
def test_text(self):
|
||||
# empty, ascii, non-BMP, sqlmap's reversible-codec private-use-area char, control chars
|
||||
for value in [u"", u"plain", _unichr(0x2299) + u" mid " + _unichr(0xFF),
|
||||
_unichr(0x1F600) if sys.maxunicode > 0xFFFF else u"x",
|
||||
_unichr(0xF0055), u"tab\tnewline\nquote\"backslash\\"]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, six.text_type)
|
||||
|
||||
|
||||
class TestBinary(unittest.TestCase):
|
||||
def test_bytes(self):
|
||||
for value in [b"", b"abc", b"\x00\x01\x02\xff\xfe", bytes(bytearray(range(256)))]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, bytes)
|
||||
|
||||
def test_bytearray(self):
|
||||
value = bytearray(b"\x00binary\xff")
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, bytearray)
|
||||
|
||||
def test_memoryview(self):
|
||||
# some drivers (e.g. psycopg2 on py3) return memoryview for BLOB columns
|
||||
restored = rt(memoryview(b"\x00\x01blob"))
|
||||
self.assertEqual(bytes(restored) if isinstance(restored, (bytearray, memoryview)) else restored, b"\x00\x01blob")
|
||||
|
||||
|
||||
class TestContainers(unittest.TestCase):
|
||||
def test_list_nested(self):
|
||||
value = [1, [2, [3, [4]]], "x", None]
|
||||
self.assertEqual(rt(value), value)
|
||||
|
||||
def test_tuple_preserved(self):
|
||||
for value in [(), (1,), (1, 2, "x"), ((1, 2), (3,))]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, tuple) # must NOT degrade to list
|
||||
|
||||
def test_tuple_not_confused_with_list(self):
|
||||
restored = rt([(1, 2), [1, 2]])
|
||||
self.assertIsInstance(restored[0], tuple)
|
||||
self.assertIsInstance(restored[1], list)
|
||||
|
||||
def test_set_and_frozenset(self):
|
||||
for value in [set(), {1, 2, 3}, {u"a", u"b"}]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, set)
|
||||
fs = frozenset([1, 2])
|
||||
restored = rt(fs)
|
||||
self.assertEqual(restored, fs)
|
||||
self.assertIsInstance(restored, frozenset)
|
||||
|
||||
|
||||
class TestDicts(unittest.TestCase):
|
||||
def test_string_keys(self):
|
||||
value = {u"a": 1, u"b": {u"c": [1, 2]}}
|
||||
self.assertEqual(rt(value), value)
|
||||
|
||||
def test_int_keys_preserved(self):
|
||||
# THE landmine: naive json.dumps turns {1: ...} into {"1": ...}; injection.data is int-keyed
|
||||
restored = rt({1: u"one", 2: u"two", 100: u"hundred"})
|
||||
self.assertEqual(restored, {1: u"one", 2: u"two", 100: u"hundred"})
|
||||
self.assertTrue(all(isinstance(k, INTS) for k in restored), list(restored))
|
||||
|
||||
def test_tuple_and_mixed_keys(self):
|
||||
value = {(1, 2): u"tuple-key", u"s": 1, 7: u"int-key"}
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIn((1, 2), restored)
|
||||
self.assertTrue(any(isinstance(k, INTS) and not isinstance(k, bool) for k in restored))
|
||||
|
||||
def test_empty_dict(self):
|
||||
self.assertEqual(rt({}), {})
|
||||
|
||||
|
||||
class TestSqlmapTypes(unittest.TestCase):
|
||||
def test_attribdict(self):
|
||||
value = AttribDict()
|
||||
value.foo = u"bar"
|
||||
value["n"] = {u"k": [1, (2, 3)]}
|
||||
restored = rt(value)
|
||||
self.assertIsInstance(restored, AttribDict)
|
||||
self.assertEqual(restored.foo, u"bar")
|
||||
self.assertEqual(restored["n"], {u"k": [1, (2, 3)]})
|
||||
|
||||
def test_attribdict_keycheck_flag_preserved(self):
|
||||
strict = AttribDict(keycheck=True)
|
||||
lax = AttribDict(keycheck=False)
|
||||
self.assertTrue(rt(strict).__dict__.get("_keycheck"))
|
||||
self.assertFalse(rt(lax).__dict__.get("_keycheck"))
|
||||
# a lax AttribDict returns None for a missing attribute instead of raising - behaviour must survive
|
||||
self.assertIsNone(rt(lax).nonexistent_attribute)
|
||||
|
||||
def test_injectiondict_full(self):
|
||||
inj = InjectionDict()
|
||||
inj.place = "GET"
|
||||
inj.parameter = "id"
|
||||
inj.ptype = 1
|
||||
inj.prefix = ""
|
||||
inj.suffix = ""
|
||||
inj.dbms = "MySQL"
|
||||
inj.notes = ["note1"]
|
||||
# int-keyed .data (PAYLOAD.TECHNIQUE.* are ints), each a nested AttribDict
|
||||
for stype in (1, 5):
|
||||
data = AttribDict()
|
||||
data.title = "technique %d" % stype
|
||||
data.payload = u"id=1 AND %d=%d" % (stype, stype)
|
||||
data.where = 1
|
||||
data.vector = None
|
||||
data.matchRatio = 0.987
|
||||
data.trueCode = 200
|
||||
data.falseCode = 500
|
||||
inj.data[stype] = data
|
||||
inj.conf = AttribDict()
|
||||
inj.conf.textOnly = False
|
||||
|
||||
restored = rt([inj])[0]
|
||||
self.assertIsInstance(restored, InjectionDict)
|
||||
self.assertEqual(restored.place, "GET")
|
||||
self.assertEqual(restored.parameter, "id")
|
||||
self.assertEqual(restored.notes, ["note1"])
|
||||
self.assertEqual(set(restored.data.keys()), set((1, 5)))
|
||||
self.assertTrue(all(isinstance(k, INTS) for k in restored.data), list(restored.data))
|
||||
self.assertIsInstance(restored.data[1], AttribDict)
|
||||
self.assertEqual(restored.data[1].title, "technique 1")
|
||||
self.assertEqual(restored.data[1].matchRatio, 0.987)
|
||||
self.assertIsNone(restored.data[1].vector)
|
||||
self.assertIsInstance(restored.conf, AttribDict)
|
||||
self.assertFalse(restored.conf.textOnly)
|
||||
|
||||
def test_bigarray_type_preserved(self):
|
||||
# BigArray is a list subclass; it must round-trip AS a BigArray, not degrade to a plain list
|
||||
restored = rt(BigArray([1, 2, (3, 4), u"x"]))
|
||||
self.assertIsInstance(restored, BigArray)
|
||||
self.assertEqual(list(restored), [1, 2, (3, 4), u"x"])
|
||||
|
||||
def test_rawpair(self):
|
||||
# the class stored in a BigArray by the HAR collector
|
||||
pair = RawPair(b"GET / HTTP/1.1", b"HTTP/1.1 200 OK", startTime=1.5, endTime=2.5, extendedArguments={u"k": u"v"})
|
||||
restored = rt(pair)
|
||||
self.assertIsInstance(restored, RawPair)
|
||||
self.assertEqual(restored.request, b"GET / HTTP/1.1")
|
||||
self.assertEqual(restored.response, b"HTTP/1.1 200 OK")
|
||||
self.assertEqual(restored.startTime, 1.5)
|
||||
self.assertEqual(restored.extendedArguments, {u"k": u"v"})
|
||||
|
||||
|
||||
class TestDbmsScalars(unittest.TestCase):
|
||||
# '-d' direct-mode query output (and dump BigArray) can hold native driver types
|
||||
def test_decimal(self):
|
||||
for value in [decimal.Decimal("0"), decimal.Decimal("1.50"), decimal.Decimal("-0.0001"), decimal.Decimal("123456789.987654321")]:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIsInstance(restored, decimal.Decimal)
|
||||
|
||||
def test_datetime_family(self):
|
||||
cases = [
|
||||
datetime.datetime(2024, 1, 2, 3, 4, 5, 6),
|
||||
datetime.datetime(1970, 1, 1, 0, 0, 0),
|
||||
datetime.date(1999, 12, 31),
|
||||
datetime.time(23, 59, 58, 123456),
|
||||
datetime.timedelta(days=3, seconds=7, microseconds=9),
|
||||
datetime.timedelta(0),
|
||||
]
|
||||
for value in cases:
|
||||
restored = rt(value)
|
||||
self.assertEqual(restored, value)
|
||||
self.assertIs(type(restored), type(value))
|
||||
|
||||
|
||||
class TestRealSessionObjects(unittest.TestCase):
|
||||
# the exact shapes written with serialize=True across the codebase
|
||||
def test_brute_tables_columns(self):
|
||||
tables = [("mydb", "users"), ("mydb", "logs")]
|
||||
columns = [("mydb", "users", "id", "numeric"), ("mydb", "users", "name", "non-numeric")]
|
||||
self.assertEqual(rt(tables), tables)
|
||||
self.assertEqual(rt(columns), columns)
|
||||
self.assertTrue(all(isinstance(_, tuple) for _ in rt(tables)))
|
||||
|
||||
def test_dynamic_markings(self):
|
||||
markings = [(u"<b>", None), (None, u"</div>"), (u"pre", u"suf")]
|
||||
self.assertEqual(rt(markings), markings)
|
||||
self.assertTrue(all(isinstance(_, tuple) for _ in rt(markings)))
|
||||
|
||||
def test_abs_file_paths_set(self):
|
||||
paths = set([u"/var/www/index.php", u"/etc/passwd"])
|
||||
restored = rt(paths)
|
||||
self.assertEqual(restored, paths)
|
||||
self.assertIsInstance(restored, set) # resume code does an explicit isinstance(_, set) union
|
||||
|
||||
def test_kb_chars_attribdict(self):
|
||||
chars = AttribDict()
|
||||
chars.delimiter = u"abcdef"
|
||||
chars.start = u"//start//"
|
||||
chars.stop = u"//stop//"
|
||||
restored = rt(chars)
|
||||
self.assertIsInstance(restored, AttribDict)
|
||||
self.assertEqual(restored.delimiter, u"abcdef")
|
||||
|
||||
|
||||
class TestSecurity(unittest.TestCase):
|
||||
def _forged(self, class_name):
|
||||
# the raw on-wire object wrapper as a hostile session file would hold it (a plain JSON
|
||||
# object tag; NOT produced via the encoder, which would re-tag a dict as a harmless map)
|
||||
return json.dumps({_SERIALIZE_TAG: "o", "c": class_name, "s": {}})
|
||||
|
||||
def test_forbidden_classes_rejected(self):
|
||||
for name in ("os.system", "subprocess.Popen", "builtins.eval", "__builtin__.eval",
|
||||
"lib.core.common.shellExec", "lib.core.common.evaluateCode", "lib.core.common.openFile"):
|
||||
try:
|
||||
deserializeValue(self._forged(name))
|
||||
self.fail("class %r was NOT rejected" % name)
|
||||
except ValueError as ex:
|
||||
self.assertIn("forbidden", str(ex), msg="unexpected error for %r: %s" % (name, ex))
|
||||
|
||||
def test_legacy_payload_fails_inertly(self):
|
||||
# an OLD session stored base64-of-pickle as TEXT; the new text reader must fail on it
|
||||
# WITHOUT executing anything (and, in practice, the bumped HASHDB_MILESTONE_VALUE means such
|
||||
# a value is never even looked up). Use a classic protocol-0 os.system pickle as the payload.
|
||||
sentinel = os.path.join(tempfile.gettempdir(), "sqlmap_serialize_sentinel_%d" % os.getpid())
|
||||
if os.path.exists(sentinel):
|
||||
os.remove(sentinel)
|
||||
legacy_pickle = b"cos\nsystem\n(S'touch " + sentinel.encode("ascii", "ignore") + b"'\ntR."
|
||||
legacy_stored = encodeBase64(legacy_pickle, binary=False) # exactly what old sqlmap wrote
|
||||
try:
|
||||
deserializeValue(legacy_stored) # base64 blob is not valid JSON -> raises
|
||||
except Exception:
|
||||
pass # any failure is fine; the point is no execution
|
||||
self.assertFalse(os.path.exists(sentinel), "legacy payload EXECUTED (sentinel created)")
|
||||
|
||||
def test_hashdb_ignores_undecodable_old_value(self):
|
||||
# the belt-and-suspenders guarantee: even if an un-deserializable (e.g. legacy) value IS
|
||||
# looked up, HashDB.retrieve() swallows it and returns None - a stale session never crashes
|
||||
from lib.utils.hashdb import HashDB
|
||||
|
||||
handle, path = tempfile.mkstemp(suffix=".sqlite")
|
||||
os.close(handle)
|
||||
os.remove(path)
|
||||
try:
|
||||
db = HashDB(path)
|
||||
db.write("legacy", encodeBase64(b"cos\nsystem\n(S'x'\ntR.", binary=False)) # stored, NOT serialized
|
||||
db.flush()
|
||||
db._write_cache.clear()
|
||||
db._read_cache.cache.clear()
|
||||
self.assertIsNone(db.retrieve("legacy", unserialize=True)) # must be None, not an exception
|
||||
db.closeAll()
|
||||
finally:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_sqlmap_type_not_in_allowlist_fails_loud(self):
|
||||
# a sqlmap-own class that is not allowlisted must raise on SERIALIZE (dev-visible), never be
|
||||
# silently dropped - LRUDict lives in lib.* and is not serializable
|
||||
self.assertRaises(TypeError, serializeValue, LRUDict(capacity=2))
|
||||
|
||||
def test_foreign_exotic_value_degrades_to_text(self):
|
||||
# a non-sqlmap, non-handled scalar degrades to its textual form rather than crashing a session
|
||||
self.assertEqual(rt(complex(1, 2)), getUnicode(complex(1, 2)))
|
||||
|
||||
|
||||
class TestBigArrayDisk(unittest.TestCase):
|
||||
def test_scalar_chunks_through_disk(self):
|
||||
ba = BigArray(chunk_size=1) # force one chunk per item -> exercises the on-disk path
|
||||
source = []
|
||||
for i in range(300):
|
||||
row = (i, u"name%d" % i, decimal.Decimal("%d.25" % i), None, b"\x00\xff")
|
||||
source.append(row)
|
||||
ba.append(row)
|
||||
self.assertGreater(len(ba.chunks), 1)
|
||||
self.assertEqual(len(ba), 300)
|
||||
self.assertEqual(list(ba), source)
|
||||
self.assertEqual(ba[150], source[150])
|
||||
self.assertEqual(ba[-1], source[-1])
|
||||
|
||||
def test_rawpair_chunks_through_disk(self):
|
||||
ba = BigArray(chunk_size=1)
|
||||
for i in range(40):
|
||||
ba.append(RawPair(b"REQ%d" % i, b"RESP%d" % i, startTime=float(i), endTime=float(i) + 1, extendedArguments={u"i": i}))
|
||||
got = list(ba)
|
||||
self.assertEqual(len(got), 40)
|
||||
self.assertTrue(all(isinstance(_, RawPair) for _ in got))
|
||||
self.assertEqual(got[7].request, b"REQ7")
|
||||
self.assertEqual(got[7].extendedArguments, {u"i": 7})
|
||||
|
||||
|
||||
class TestAllowlistGuard(unittest.TestCase):
|
||||
"""CI guard: catch a NEW class becoming serializable before it reaches a user's session."""
|
||||
|
||||
def test_allowlist_names_resolve_and_stay_in_sync(self):
|
||||
# every allowlisted name must resolve to a class whose real module.qualname matches - keeps
|
||||
# convert._SERIALIZE_CLASSES and convert._serializeResolveClass in lockstep (a rename/move/
|
||||
# typo of an allowlisted class fails here instead of silently at a user's session load)
|
||||
from lib.core.convert import _SERIALIZE_CLASSES, _serializeResolveClass
|
||||
for name in _SERIALIZE_CLASSES:
|
||||
cls = _serializeResolveClass(name)
|
||||
self.assertEqual("%s.%s" % (cls.__module__, cls.__name__), name)
|
||||
|
||||
def test_no_undeclared_attribdict_subclass(self):
|
||||
# AttribDict/InjectionDict carry the session's structured data. A NEW AttribDict subclass that
|
||||
# ends up stored in the session would be REJECTED by the serializer at a user's runtime. Catch
|
||||
# it here coverage-INDEPENDENTLY: import the whole tree, then require every AttribDict subclass
|
||||
# to be declared serializable in convert._SERIALIZE_CLASSES.
|
||||
from lib.core.convert import _SERIALIZE_CLASSES
|
||||
from lib.core.datatype import AttribDict
|
||||
|
||||
_import_all_modules()
|
||||
|
||||
offenders = sorted(set(
|
||||
"%s.%s" % (cls.__module__, cls.__name__)
|
||||
for cls in _all_subclasses(AttribDict)
|
||||
) - set(_SERIALIZE_CLASSES))
|
||||
|
||||
self.assertEqual(offenders, [], (
|
||||
"undeclared AttribDict subclass(es): %s -- if stored in the session, add it to BOTH "
|
||||
"convert._SERIALIZE_CLASSES and convert._serializeResolveClass (else the safe serializer "
|
||||
"raises on it at a user's runtime). If it is a runtime-only registry, make it subclass a "
|
||||
"plain dict instead of AttribDict (see lib.core.unescaper.Unescaper) so it never lands "
|
||||
"here." % offenders
|
||||
))
|
||||
|
||||
def test_known_serializable_classes_present(self):
|
||||
# lock the intended set: exactly the dict-like data types plus the HAR RawPair. If this list
|
||||
# legitimately grows, update it here deliberately (a conscious review point).
|
||||
from lib.core.convert import _SERIALIZE_CLASSES
|
||||
self.assertEqual(set(_SERIALIZE_CLASSES), set((
|
||||
"lib.core.datatype.AttribDict",
|
||||
"lib.core.datatype.InjectionDict",
|
||||
"lib.utils.har.RawPair",
|
||||
)))
|
||||
|
||||
|
||||
class TestHashDBIntegration(unittest.TestCase):
|
||||
def test_write_retrieve_roundtrip(self):
|
||||
from lib.utils.hashdb import HashDB
|
||||
|
||||
handle, path = tempfile.mkstemp(suffix=".sqlite")
|
||||
os.close(handle)
|
||||
os.remove(path)
|
||||
try:
|
||||
inj = InjectionDict()
|
||||
inj.place = "GET"
|
||||
inj.parameter = "id"
|
||||
inj.data[1] = AttribDict()
|
||||
inj.data[1].title = "boolean-based blind"
|
||||
inj.data[1].matchRatio = 0.9
|
||||
value = [inj]
|
||||
|
||||
db = HashDB(path)
|
||||
db.write("KB_INJECTIONS", value, serialize=True)
|
||||
db.flush()
|
||||
# drop the in-memory caches so retrieve() must SELECT the serialized blob back off
|
||||
# disk and deserialize it (the real resume path), rather than returning a cached object
|
||||
db._write_cache.clear()
|
||||
db._read_cache.cache.clear()
|
||||
restored = db.retrieve("KB_INJECTIONS", unserialize=True)
|
||||
db.closeAll()
|
||||
|
||||
self.assertIsInstance(restored, list)
|
||||
self.assertIsInstance(restored[0], InjectionDict)
|
||||
self.assertEqual(restored[0].place, "GET")
|
||||
self.assertTrue(all(isinstance(k, INTS) for k in restored[0].data))
|
||||
self.assertEqual(restored[0].data[1].title, "boolean-based blind")
|
||||
self.assertEqual(restored[0].data[1].matchRatio, 0.9)
|
||||
finally:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
191
tests/test_sqllint.py
Normal file
191
tests/test_sqllint.py
Normal file
|
|
@ -0,0 +1,191 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
|
||||
Unit tests + atom-level SQL coverage for lib/utils/sqllint.py.
|
||||
|
||||
Two concerns:
|
||||
|
||||
1. Linter self-test - a curated set of well-formed fragments/statements that
|
||||
must NOT flag, malformed ones that MUST flag, and the cross-dialect edge
|
||||
cases the linter has learned (==, ::, $/backtick identifiers, LIKE() as a
|
||||
function, HSQLDB "LIMIT off lim", ...). Guards the linter from regressions.
|
||||
|
||||
2. Atom-level SQL coverage - every SQL-bearing template in data/xml/queries.xml
|
||||
and data/xml/payloads.xml, across ALL back-end DBMSes, must lint structurally
|
||||
clean. This is a coverage gate over the *building blocks* of every query
|
||||
sqlmap emits. The composed/runtime layer (agent.py wrapping these atoms into
|
||||
wire payloads) is exercised faithfully by the separate payload-lint walk; a
|
||||
0-flag result here means the catalog itself is sound in all 30 dialects.
|
||||
|
||||
A regression (a newly malformed catalog entry, OR a genuinely new valid dialect
|
||||
construct the linter does not yet understand) makes the relevant test fail with a
|
||||
pointer to the offending template.
|
||||
|
||||
stdlib unittest only; Python 2.7 and 3.x; pure-ASCII.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import unittest
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from lib.utils.sqllint import checkSanity
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
# well-formed fragments/statements that must NOT flag
|
||||
GOOD = (
|
||||
"1 AND 1=1",
|
||||
"1) AND 5108=5108 AND (7936=7936",
|
||||
"1)) OR 1=1-- -",
|
||||
"1' AND '1'='1",
|
||||
"1 UNION ALL SELECT NULL,NULL,CONCAT(0x71,0x62,0x71)-- -",
|
||||
"1 AND (SELECT 1 FROM (SELECT COUNT(*),CONCAT(0x7176,(SELECT database()),0x71)x FROM information_schema.tables GROUP BY x)a)",
|
||||
"1 AND ORD(MID((SELECT IFNULL(CAST(username AS NCHAR),0x20) FROM users ORDER BY id LIMIT 0,1),1,1))>64",
|
||||
"1 AND EXTRACTVALUE(1,CONCAT(0x5c,0x7e,(SELECT version())))",
|
||||
"SELECT name FROM users WHERE id=1 ORDER BY id",
|
||||
"SELECT count(*) FROM t WHERE x=1",
|
||||
"1 WAITFOR DELAY '0:0:5'",
|
||||
)
|
||||
|
||||
# cross-dialect valid constructs the linter must accept (regression guards for
|
||||
# the exact false positives that adversarial/catalog runs exposed and fixed)
|
||||
DIALECT_GOOD = (
|
||||
"1 AND id==1", # SQLite '==' equality
|
||||
"SELECT 1 WHERE (SELECT 1)::text = '1'", # PostgreSQL '::' cast
|
||||
"SELECT 1 WHERE 9223=LIKE(CHAR(65),UPPER(HEX(1)))", # SQLite LIKE() function
|
||||
"SELECT NAME FROM SYSMASTER:SYSDATABASES", # Informix 'db:table'
|
||||
"SELECT $ZVERSION", # InterSystems Cache system var
|
||||
"SELECT `col` FROM `directory` LIMIT 0,1", # MySQL backtick identifiers
|
||||
"SELECT LIMIT 0 1 DISTINCT(user) FROM INFORMATION_SCHEMA.SYSTEM_USERS", # HSQLDB space-LIMIT
|
||||
"SELECT TOP 1 name FROM master..sysdatabases", # MSSQL TOP + db..table
|
||||
"CONCAT('\\',0x71,(SELECT 1))", # backslash-literal string (ANSI)
|
||||
)
|
||||
|
||||
# malformed fragments/statements that MUST flag
|
||||
BAD = (
|
||||
"(SELECT id 1 FROM users)", # missing separator
|
||||
"1 AND 1=(SELECT id 1 FROM users' WHERE tablename='foobar')", # stray quote in scope
|
||||
"1UNION SELECT NULL", # digit glued to word
|
||||
"1 AND 5108=5108AND 1=1", # digit glued to keyword
|
||||
"1 AND 1 = = 1", # doubled operator
|
||||
"1 AND AND 1=1", # doubled keyword operator
|
||||
"1 UNION SELECT NULL,,NULL", # empty list item
|
||||
"1 AND (1=1)(2=2)", # adjacent groups
|
||||
"SELECT count() FROM t WHERE id=)", # operator before ')'
|
||||
)
|
||||
|
||||
# queries.xml: attributes that carry SQL (not regexes/markers)
|
||||
_SQL_ATTRS = ("query", "query2", "query3", "count", "count2", "count3",
|
||||
"condition", "condition2", "condition3",
|
||||
"keyset_where", "keyset_next", "keyset_first", "keyset_by", "keyset_ordered", "rowid")
|
||||
_SKIP_TAGS = ("limitregexp", "comment")
|
||||
|
||||
|
||||
def _materialize(s):
|
||||
"""Substitute sqlmap's template markers with structurally-neutral values so a
|
||||
catalog template becomes lintable SQL (mirrors what agent.py fills in)."""
|
||||
s = s.replace("[QUERY]", "SELECT 1").replace("[UNION]", "UNION SELECT NULL")
|
||||
s = s.replace("[INFERENCE]", "1=1")
|
||||
s = re.sub(r"\[RANDNUM\d*\]", "1", s)
|
||||
s = re.sub(r"\[RANDSTR\d*\]", "abc", s)
|
||||
s = s.replace("[SLEEPTIME]", "1").replace("[DELAYED]", "1")
|
||||
for _ in ("[DELIMITER_START]", "[DELIMITER_STOP]", "[COLSTART]", "[COLSTOP]"):
|
||||
s = s.replace(_, "0x71")
|
||||
s = s.replace("[ORIGVALUE]", "1").replace("[CHAR]", "NULL").replace("[GENERIC_SQL_COMMENT]", "-- -")
|
||||
s = s.replace("[SINGLE_QUOTE]", "'").replace("[DOUBLE_QUOTE]", '"').replace("[DB]", "db")
|
||||
s = s.replace("[SPACE_REPLACE]", " ").replace("[HASH_REPLACE]", "#")
|
||||
s = s.replace("[DOLLAR_REPLACE]", "$").replace("[AT_REPLACE]", "@")
|
||||
s = re.sub(r"\[[A-Z][A-Z0-9_]*\]", "1", s)
|
||||
s = s.replace("%d", "1").replace("%s", "col").replace("%%", "%")
|
||||
return s
|
||||
|
||||
|
||||
def _queries_atoms():
|
||||
"""{dbms: sorted list of SQL atom strings} from queries.xml."""
|
||||
retVal = {}
|
||||
root = ET.parse(os.path.join(ROOT, "data", "xml", "queries.xml")).getroot()
|
||||
for dbms in root.iter("dbms"):
|
||||
atoms = set()
|
||||
for el in dbms.iter():
|
||||
if el.tag in _SKIP_TAGS:
|
||||
continue
|
||||
for attr in _SQL_ATTRS:
|
||||
raw = el.get(attr)
|
||||
if raw and not raw.strip().isdigit():
|
||||
atoms.add(raw)
|
||||
retVal[dbms.get("value")] = sorted(atoms)
|
||||
return retVal
|
||||
|
||||
|
||||
def _payload_atoms():
|
||||
"""[(file, stype-title, sql)] from data/xml/payloads/*.xml."""
|
||||
import glob
|
||||
retVal = []
|
||||
for path in sorted(glob.glob(os.path.join(ROOT, "data", "xml", "payloads", "*.xml"))):
|
||||
name = os.path.basename(path)
|
||||
for test in ET.parse(path).getroot().iter("test"):
|
||||
title = (test.findtext("title") or "").strip()
|
||||
for tag in ("payload", "vector", "comparison"):
|
||||
for el in test.iter(tag):
|
||||
raw = (el.text or "").strip()
|
||||
if raw:
|
||||
retVal.append((name, title, raw))
|
||||
return retVal
|
||||
|
||||
|
||||
class TestLinterSelf(unittest.TestCase):
|
||||
def test_well_formed_pass(self):
|
||||
for sql in GOOD + DIALECT_GOOD:
|
||||
self.assertEqual(checkSanity(sql), [], "false positive on valid SQL: %r" % sql)
|
||||
|
||||
def test_malformed_flag(self):
|
||||
for sql in BAD:
|
||||
self.assertTrue(checkSanity(sql), "missed malformed SQL: %r" % sql)
|
||||
|
||||
|
||||
# module-level coverage accounting (printed in tearDownModule)
|
||||
_COVERAGE = {"queries_dbms": 0, "queries_atoms": 0, "payload_atoms": 0}
|
||||
|
||||
|
||||
class TestCatalogCoverage(unittest.TestCase):
|
||||
def test_queries_xml_clean(self):
|
||||
atoms = _queries_atoms()
|
||||
_COVERAGE["queries_dbms"] = len(atoms)
|
||||
total = 0
|
||||
for dbms, items in atoms.items():
|
||||
for raw in items:
|
||||
total += 1
|
||||
issues = checkSanity(_materialize(raw))
|
||||
self.assertEqual(issues, [],
|
||||
"queries.xml [%s] malformed atom: %r -> %s" % (dbms, raw, issues))
|
||||
_COVERAGE["queries_atoms"] = total
|
||||
self.assertGreater(total, 1000) # sanity: the catalog was actually walked
|
||||
|
||||
def test_payloads_xml_clean(self):
|
||||
items = _payload_atoms()
|
||||
_COVERAGE["payload_atoms"] = len(items)
|
||||
for name, title, raw in items:
|
||||
issues = checkSanity(_materialize(raw))
|
||||
self.assertEqual(issues, [],
|
||||
"%s malformed payload atom (%s): %r -> %s" % (name, title, raw, issues))
|
||||
self.assertGreater(len(items), 500)
|
||||
|
||||
|
||||
def tearDownModule():
|
||||
q = _COVERAGE
|
||||
total = q["queries_atoms"] + q["payload_atoms"]
|
||||
sys.stderr.write(
|
||||
"\n[sqllint coverage] atom-level: %d SQL atoms linted clean "
|
||||
"(%d queries.xml across %d/30 DBMSes + %d payloads.xml vectors)\n"
|
||||
% (total, q["queries_atoms"], q["queries_dbms"], q["payload_atoms"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
|
|
@ -1,121 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
|
||||
Locks the RestrictedUnpickler security control (lib/core/patch.py, installed over
|
||||
pickle.loads by dirtyPatches()). sqlmap deserializes pickled blobs out of its own
|
||||
session DB / cache, so the unpickler is an ALLOWLIST: only safe builtin data types
|
||||
and sqlmap's own (lib/plugins/thirdparty) classes may be reconstructed.
|
||||
|
||||
Two directions, both of which must keep holding:
|
||||
- LEGIT round-trips sqlmap actually relies on (AttribDict, BigArray, nested
|
||||
builtins, and - the easy-to-regress one - bytes under PICKLE_PROTOCOL=2, which
|
||||
emits a _codecs.encode global) must survive base64pickle -> base64unpickle.
|
||||
- MALICIOUS / exotic globals (eval, os.system, subprocess.Popen, importlib,
|
||||
operator.attrgetter, and even the non-whitelisted _codecs.lookup) must be
|
||||
REJECTED at find_class time, before the object is ever built.
|
||||
|
||||
A regression in either direction is a security or a data-loss bug, hence the test.
|
||||
"""
|
||||
|
||||
import os
|
||||
import pickle
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _testutils import bootstrap
|
||||
bootstrap() # installs dirtyPatches(), i.e. the RestrictedUnpickler over pickle.loads
|
||||
|
||||
from lib.core.bigarray import BigArray
|
||||
from lib.core.convert import base64pickle, base64unpickle, encodeBase64
|
||||
from lib.core.datatype import AttribDict
|
||||
from lib.core.settings import PICKLE_PROTOCOL
|
||||
|
||||
|
||||
class _EvilReduce(object):
|
||||
"""On unpickling, __reduce__ asks the loader to resolve (and would call) an arbitrary global."""
|
||||
def __init__(self, func, args):
|
||||
self._func = func
|
||||
self._args = args
|
||||
|
||||
def __reduce__(self):
|
||||
return (self._func, self._args)
|
||||
|
||||
|
||||
def _payload(func, *args):
|
||||
# built with the REAL pickler (only pickle.loads is restricted, not dumps); base64 to mirror
|
||||
# exactly what base64unpickle() consumes from sqlmap's session store
|
||||
return encodeBase64(pickle.dumps(_EvilReduce(func, args), PICKLE_PROTOCOL), binary=False)
|
||||
|
||||
|
||||
class TestUnpicklerIsInstalled(unittest.TestCase):
|
||||
def test_patch_active(self):
|
||||
# if this is False the whole allowlist is bypassed and the negative tests would pass vacuously
|
||||
self.assertTrue(getattr(pickle, "_patched", False))
|
||||
|
||||
|
||||
class TestLegitRoundTrips(unittest.TestCase):
|
||||
def _roundtrip(self, value):
|
||||
return base64unpickle(base64pickle(value))
|
||||
|
||||
def test_nested_builtins(self):
|
||||
value = {"a": [1, 2.5, True, None, complex(1, 2)], "b": (u"x", b"y"), "c": {3, 4}, "d": frozenset([5])}
|
||||
self.assertEqual(self._roundtrip(value), value)
|
||||
|
||||
def test_bytes_protocol2(self):
|
||||
# protocol-2 pickling of bytes on Python 3 emits a _codecs.encode global; this is the
|
||||
# exact case the allowlist explicitly permits, and the one most likely to silently break
|
||||
for value in (b"", b"\x00\x01\x02binary\xff", bytearray(b"abc")):
|
||||
self.assertEqual(self._roundtrip(value), value)
|
||||
|
||||
def test_attribdict(self):
|
||||
value = AttribDict()
|
||||
value.foo = "bar"
|
||||
value.nested = {"k": [1, 2]}
|
||||
restored = self._roundtrip(value)
|
||||
self.assertIsInstance(restored, AttribDict)
|
||||
self.assertEqual(restored.foo, "bar")
|
||||
self.assertEqual(restored.nested, {"k": [1, 2]})
|
||||
|
||||
def test_bigarray(self):
|
||||
restored = self._roundtrip(BigArray([1, 2, 3]))
|
||||
self.assertIsInstance(restored, BigArray)
|
||||
self.assertEqual(list(restored), [1, 2, 3])
|
||||
|
||||
|
||||
class TestMaliciousRejected(unittest.TestCase):
|
||||
def _assert_blocked(self, payload):
|
||||
# find_class() raises ValueError; base64unpickle only swallows TypeError, so it propagates
|
||||
self.assertRaises(ValueError, base64unpickle, payload)
|
||||
|
||||
def test_dangerous_builtins(self):
|
||||
# builtins are allowed ONLY for the safe data-type subset; callables must be refused
|
||||
for func in (eval, getattr, __import__):
|
||||
self._assert_blocked(_payload(func, "1+1") if func is eval else _payload(func, "x"))
|
||||
|
||||
def test_os_system(self):
|
||||
self._assert_blocked(_payload(os.system, "echo pwned"))
|
||||
|
||||
def test_subprocess_popen(self):
|
||||
self._assert_blocked(_payload(subprocess.Popen, "echo pwned"))
|
||||
|
||||
def test_importlib(self):
|
||||
import importlib
|
||||
self._assert_blocked(_payload(importlib.import_module, "os"))
|
||||
|
||||
def test_operator_attrgetter(self):
|
||||
import operator
|
||||
self._assert_blocked(_payload(operator.attrgetter, "system"))
|
||||
|
||||
def test_codecs_lookup_not_whitelisted(self):
|
||||
# only _codecs.encode is allowed (for the bytes round-trip); every other _codecs name stays blocked
|
||||
import codecs
|
||||
self._assert_blocked(_payload(codecs.lookup, "utf-8"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue