mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-10-12 02:22:55 +00:00
344 lines
14 KiB
Python
344 lines
14 KiB
Python
#!/usr/bin/env python
|
|
|
|
"""
|
|
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
|
See the file 'LICENSE' for copying permission
|
|
|
|
Offline, deterministic tests for the OData '$filter' injection engine. A mock oracle mirrors the
|
|
boolean-blind semantics of a real OData service (a broken-out filter reduced to its injected predicate),
|
|
so detection, OData-only confirmation, version fingerprinting and per-entity blind extraction are
|
|
exercised without a live OData endpoint.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from _testutils import bootstrap
|
|
bootstrap()
|
|
|
|
import lib.techniques.odata.inject as odata
|
|
from lib.core.settings import ODATA_ERROR_REGEX
|
|
|
|
_ENTITIES = (
|
|
{"Id": 1, "Name": "luther", "Secret": "S3CR3Tvalue"},
|
|
{"Id": 2, "Name": "fluffy", "Secret": "hunter2"},
|
|
{"Id": 3, "Name": "wu", "Secret": "letmein"},
|
|
)
|
|
_FIELDS = ("Id", "Name", "Secret")
|
|
|
|
|
|
def _wrapped(expr):
|
|
if not (expr.startswith("(") and expr.endswith(")")):
|
|
return False
|
|
depth = 0
|
|
for i, ch in enumerate(expr):
|
|
depth += (ch == "(") - (ch == ")")
|
|
if depth == 0 and i < len(expr) - 1:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _split(expr, sep):
|
|
parts, buf = [], []
|
|
for token in expr.split(sep):
|
|
buf.append(token)
|
|
chunk = sep.join(buf)
|
|
if chunk.count("(") == chunk.count(")"):
|
|
parts.append(chunk)
|
|
buf = []
|
|
if buf:
|
|
parts.append(sep.join(buf))
|
|
return parts
|
|
|
|
|
|
class _ODErr(Exception):
|
|
pass
|
|
|
|
|
|
def _atom(e, a):
|
|
a = a.strip()
|
|
while _wrapped(a):
|
|
a = a[1:-1].strip()
|
|
m = re.match(r"^length\('([^']*)'\) eq (\d+)$", a)
|
|
if m:
|
|
return len(m.group(1)) == int(m.group(2))
|
|
m = re.match(r"^startswith\('([^']*)','([^']*)'\)$", a)
|
|
if m:
|
|
return m.group(1).startswith(m.group(2))
|
|
m = re.match(r"^contains\('([^']*)','([^']*)'\)$", a)
|
|
if m:
|
|
return m.group(2) in m.group(1)
|
|
if a.startswith("substringof("):
|
|
raise _ODErr("v4 has no substringof")
|
|
m = re.match(r"^'([^']*)' eq '([^']*)'$", a)
|
|
if m:
|
|
return m.group(1) == m.group(2)
|
|
m = re.match(r"^(\d+) eq (\d+)$", a)
|
|
if m:
|
|
return m.group(1) == m.group(2)
|
|
m = re.match(r"^(\w+) eq '([^']*)'$", a)
|
|
if m:
|
|
if m.group(1) not in _FIELDS:
|
|
raise _ODErr("unknown property")
|
|
return "%s" % e.get(m.group(1)) == m.group(2)
|
|
m = re.match(r"^(\w+) ne null$", a)
|
|
if m:
|
|
if m.group(1) not in _FIELDS:
|
|
raise _ODErr("unknown property")
|
|
return e.get(m.group(1)) is not None
|
|
m = re.match(r"^(\w+) (eq|ge|gt|le|lt) (-?\d+)$", a)
|
|
if m:
|
|
p, op, num = m.group(1), m.group(2), int(m.group(3))
|
|
if p not in _FIELDS:
|
|
raise _ODErr("unknown property")
|
|
v = e.get(p)
|
|
if not isinstance(v, int):
|
|
return False
|
|
return {"eq": v == num, "ge": v >= num, "gt": v > num, "le": v <= num, "lt": v < num}[op]
|
|
m = re.match(r"^length\((\w+)\) (eq|ge) (\d+)$", a)
|
|
if m:
|
|
p, op, num = m.group(1), m.group(2), int(m.group(3))
|
|
if p not in _FIELDS:
|
|
raise _ODErr("unknown property")
|
|
n = len("%s" % e.get(p, ""))
|
|
return n == num if op == "eq" else n >= num
|
|
# an inner single quote arrives DOUBLED, the way the OData spec escapes it
|
|
m = re.match(r"^substring\((\w+),(\d+),1\) eq '(''|.)'$", a)
|
|
if m:
|
|
p, pos, ch = m.group(1), int(m.group(2)), m.group(3)
|
|
ch = "'" if ch == "''" else ch
|
|
if p not in _FIELDS:
|
|
raise _ODErr("unknown property")
|
|
t = "%s" % e.get(p, "")
|
|
return pos < len(t) and t[pos] == ch
|
|
raise _ODErr("syntax error")
|
|
|
|
|
|
def _eval(e, expr):
|
|
expr = expr.strip()
|
|
while _wrapped(expr):
|
|
expr = expr[1:-1].strip()
|
|
ors = _split(expr, " or ")
|
|
if len(ors) > 1:
|
|
return any(_eval(e, o) for o in ors)
|
|
ands = _split(expr, " and ")
|
|
if len(ands) > 1:
|
|
return all(_eval(e, a) for a in ands)
|
|
return _atom(e, expr)
|
|
|
|
|
|
_EMPTY = "<html><body><ul></ul></body></html>"
|
|
|
|
|
|
def _render(matched):
|
|
return "<html><body><ul>%s</ul></body></html>" % "".join("<li>%s: %s</li>" % (e["Id"], e["Name"]) for e in matched)
|
|
|
|
|
|
def _mockSend(place, parameter, value, raw=False):
|
|
expr = "Name eq '%s'" % value
|
|
if expr.count("'") % 2:
|
|
return "<pre>The query specified in the URI is not valid. There is an unterminated string literal at position 8</pre>" if raw else None
|
|
try:
|
|
matched = [e for e in _ENTITIES if _eval(e, expr)]
|
|
except _ODErr:
|
|
return "<pre>Could not find a property named 'x' on type 'Default.Product'. Microsoft.OData</pre>" if raw else None
|
|
return _render(matched)
|
|
|
|
|
|
class TestHelpers(unittest.TestCase):
|
|
def test_is_error_and_backend(self):
|
|
self.assertTrue(odata._isError("The query specified in the URI is not valid. Microsoft.OData"))
|
|
self.assertFalse(odata._isError(_EMPTY))
|
|
self.assertEqual(odata._backendFromError("Could not find a property named 'X'"), "Microsoft OData (WebAPI/.NET)")
|
|
|
|
def test_error_regex_matches_real(self):
|
|
self.assertIsNotNone(re.search(ODATA_ERROR_REGEX, "There is an unterminated string literal at position 17 in 'Name eq'"))
|
|
|
|
|
|
class TestNoAmpersandInvariant(unittest.TestCase):
|
|
"""A raw '&' in a GET value is the parameter delimiter; OData uses the 'or'/'and' keywords, so no
|
|
payload may carry one."""
|
|
|
|
def test_boundaries_and_predicates_have_no_ampersand(self):
|
|
for row in odata._BOUNDARY_TABLE:
|
|
for field in row:
|
|
self.assertNotIn("&", field)
|
|
for t, f in odata._ODATA_PREDICATES:
|
|
self.assertNotIn("&", t)
|
|
self.assertNotIn("&", f)
|
|
|
|
def test_send_url_encodes_the_payload(self):
|
|
"""The boundaries being '&'-free was never enough: the character scan emits '&' (and '+') as
|
|
DATA. _send() must URL-encode, or those probes split the request, the truncated $filter 400s,
|
|
and the resulting InconclusiveError aborts the WHOLE property (it dumped as '?')."""
|
|
sent = []
|
|
savedParams, odata.conf.parameters = odata.conf.parameters, {odata.PLACE.GET: "name=luther"}
|
|
savedGet = odata.Request.getPage
|
|
odata.Request.getPage = staticmethod(
|
|
lambda **kwargs: (sent.append(odata.conf.parameters[odata.PLACE.GET]), "", None, 200)[1:])
|
|
try:
|
|
odata._send(odata.PLACE.GET, "name", odata._literal(ord("&")))
|
|
finally:
|
|
odata.Request.getPage = savedGet
|
|
odata.conf.parameters = savedParams
|
|
self.assertEqual(len(sent), 1)
|
|
self.assertNotIn("&", sent[0].split("=", 1)[1], sent[0])
|
|
self.assertIn("%26", sent[0])
|
|
|
|
|
|
class TestCharsetCoverage(unittest.TestCase):
|
|
"""The scan is EXACT equality, not a bisection, so the ORDER is free but the COVERAGE is not: an
|
|
excluded codepoint is simply never recoverable. The one character an OData literal cannot carry
|
|
raw, the single quote, is doubled per the spec rather than dropped.
|
|
|
|
(This replaces an earlier assertion pinning 0x27/0x5c as EXCLUDED, which cost coverage of two
|
|
characters common in real names and paths for no correctness gain.)"""
|
|
|
|
def test_charset_covers_every_printable(self):
|
|
self.assertEqual(sorted(odata._CS_ORDS),
|
|
list(range(odata.ODATA_CHAR_MIN, odata.ODATA_CHAR_MAX + 1)))
|
|
|
|
def test_quote_is_doubled(self):
|
|
self.assertEqual(odata._literal(0x27), "''''")
|
|
self.assertEqual(odata._literal(0x5c), "'\\'")
|
|
self.assertEqual(odata._literal(ord("a")), "'a'")
|
|
|
|
|
|
def _mockSendV23(place, parameter, value, raw=False):
|
|
"""A v2/v3 service: contains() is an unknown function (400), substringof() is the one that parses."""
|
|
if "contains(" in value:
|
|
return "<pre>Syntax error at position 0. Microsoft.OData</pre>" if raw else None
|
|
return _mockSend(place, parameter, value.replace("substringof('sql','sqlmap')", "1 eq 1"), raw)
|
|
|
|
|
|
def _mockSendQuiet(place, parameter, value, raw=False):
|
|
"""An endpoint that SWALLOWS the service's 400: same boolean oracle, no error surface at all - a
|
|
failed $filter is rendered as the ordinary empty result page, error body and status included."""
|
|
expr = "Name eq '%s'" % value
|
|
if expr.count("'") % 2:
|
|
return _EMPTY
|
|
try:
|
|
return _render([e for e in _ENTITIES if _eval(e, expr)])
|
|
except _ODErr:
|
|
return _EMPTY
|
|
|
|
|
|
class TestVersionFingerprintDoesNotCrash(unittest.TestCase):
|
|
"""Both version probes are EXPECTED to fail on the dialect that does not own them. An unknown
|
|
$filter function is a 400 the oracle can only call inconclusive, and that must read as a negative
|
|
answer - not as an exception that aborts the scan before the finding is even reported, on exactly
|
|
the v2/v3 services the second branch exists to name."""
|
|
|
|
def setUp(self):
|
|
self.saved, self.savedParams = odata._send, odata.conf.parameters
|
|
odata.conf.parameters = {odata.PLACE.GET: "name=luther"}
|
|
odata.SENTINEL = "zzsentinelzz"
|
|
|
|
def tearDown(self):
|
|
odata._send, odata.conf.parameters = self.saved, self.savedParams
|
|
|
|
def test_v23_service_is_named_not_fatal(self):
|
|
odata._send = _mockSendV23
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
oracle = odata._makeOracle(odata.PLACE.GET, "name", boundary)
|
|
self.assertIsNotNone(oracle)
|
|
self.assertEqual(odata._fingerprintVersion(oracle), "v2/v3")
|
|
|
|
def test_service_speaking_neither_returns_none(self):
|
|
odata._send = lambda place, parameter, value, raw=False: _mockSend(
|
|
place, parameter, value.replace("contains('sqlmap','sql')", "unknownfn()"), raw)
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
oracle = odata._makeOracle(odata.PLACE.GET, "name", boundary)
|
|
self.assertIsNone(odata._fingerprintVersion(oracle))
|
|
|
|
|
|
class TestErrorSurfaceGate(unittest.TestCase):
|
|
"""_fieldExists() is an error/no-error split, so on an endpoint that swallows the service's 400 it
|
|
answers 'exists' for EVERY name - which reported all 37 candidate properties as reachable and
|
|
dumped a wall of empty columns. _hasErrorSurface() tells the two apart in one request."""
|
|
|
|
def setUp(self):
|
|
self.saved, self.savedParams = odata._send, odata.conf.parameters
|
|
odata.conf.parameters = {odata.PLACE.GET: "name=luther"}
|
|
odata.SENTINEL = "zzsentinelzz"
|
|
|
|
def tearDown(self):
|
|
odata._send, odata.conf.parameters = self.saved, self.savedParams
|
|
|
|
def _boundary(self):
|
|
return odata._detectBoolean(odata.PLACE.GET, "name")[2]
|
|
|
|
def test_error_surface_present(self):
|
|
odata._send = _mockSend
|
|
boundary = self._boundary()
|
|
self.assertTrue(odata._hasErrorSurface(odata.PLACE.GET, "name", boundary))
|
|
|
|
def test_error_surface_absent(self):
|
|
odata._send = _mockSendQuiet
|
|
boundary = self._boundary()
|
|
self.assertFalse(odata._hasErrorSurface(odata.PLACE.GET, "name", boundary))
|
|
# the error-based check is useless here - every unknown name looks real
|
|
self.assertTrue(odata._fieldExists(odata.PLACE.GET, "name", boundary, "TotallyMadeUp"))
|
|
# ...so existence falls through to the boolean oracle, which still separates them. The dump
|
|
# must keep working on a blind target, not be surrendered because one probe went blind.
|
|
oracle = odata._makeOracle(odata.PLACE.GET, "name", boundary, truePredicate="(Id eq 1)")
|
|
self.assertIsNotNone(oracle)
|
|
self.assertTrue(odata._fieldExistsBlind(oracle, "Id", 1, "Secret"))
|
|
self.assertFalse(odata._fieldExistsBlind(oracle, "Id", 1, "TotallyMadeUp"))
|
|
|
|
|
|
class TestDetectionAndExtraction(unittest.TestCase):
|
|
def setUp(self):
|
|
self.saved = odata._send
|
|
odata._send = _mockSend
|
|
self.savedParams = odata.conf.parameters
|
|
odata.conf.parameters = {odata.PLACE.GET: "name=luther"}
|
|
odata.SENTINEL = "zzsentinelzz"
|
|
|
|
def tearDown(self):
|
|
odata._send = self.saved
|
|
odata.conf.parameters = self.savedParams
|
|
|
|
def test_boolean_detection(self):
|
|
template, payload, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
self.assertIsNotNone(template)
|
|
|
|
def test_confirms_odata(self):
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
self.assertTrue(odata._confirmOData(odata.PLACE.GET, "name", boundary))
|
|
|
|
def test_version_fingerprint_v4(self):
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
oracle = odata._makeOracle(odata.PLACE.GET, "name", boundary)
|
|
self.assertEqual(odata._fingerprintVersion(oracle), "v4")
|
|
|
|
def test_field_existence(self):
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
self.assertTrue(odata._fieldExists(odata.PLACE.GET, "name", boundary, "Secret"))
|
|
self.assertFalse(odata._fieldExists(odata.PLACE.GET, "name", boundary, "Nope"))
|
|
|
|
def test_key_and_entities(self):
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
key, keys = odata._findKeyAndEntities(odata.PLACE.GET, "name", boundary, _EMPTY)
|
|
self.assertEqual(key, "Id")
|
|
self.assertEqual(keys, [1, 2, 3])
|
|
|
|
def test_blind_field_extraction_reaches_secret(self):
|
|
_t, _p, boundary = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
oracle = odata._makeOracle(odata.PLACE.GET, "name", boundary, truePredicate="(Id eq 1)")
|
|
self.assertIsNotNone(oracle)
|
|
self.assertEqual(odata._inferField(oracle, "Id", 1, "Name"), "luther")
|
|
self.assertEqual(odata._inferField(oracle, "Id", 1, "Secret"), "S3CR3Tvalue")
|
|
|
|
def test_plain_sql_endpoint_not_confirmed(self):
|
|
odata._send = lambda place, parameter, value, raw=False: _render(_ENTITIES) # always same
|
|
template, _p, _b = odata._detectBoolean(odata.PLACE.GET, "name")
|
|
self.assertIsNone(template)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|