CERBERUS/sqlmap
1
0
Fork 0
mirror of https://github.com/sqlmapproject/sqlmap.git synced 2026-09-27 13:34:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
sqlmap/tests/test_sparql.py
Miroslav Štampar 90b14df46c Minor update
2026-08-11 23:07:22 +02:00

384 lines
17 KiB
Python
Raw Permalink Blame History

#!/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 SPARQL injection engine. A mock oracle mirrors the boolean-blind
semantics of a real triple store (a broken-out FILTER reduced to its injected predicate), so detection,
SPARQL-only confirmation, error fingerprinting and schema-agnostic blind extraction are exercised
without a live SPARQL 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.sparql.inject as sparql
from lib.core.settings import SPARQL_ERROR_REGEX
# a tiny fixed graph the mock evaluates against
_TRIPLES = (
("http://example.org/p1", "http://xmlns.com/foaf/0.1/name", "luther"),
("http://example.org/p1", "http://xmlns.com/foaf/0.1/mbox", "luther@example.org"),
("http://example.org/secret", "http://example.org/flag", "S3CR3Tvalue"),
# every character class the transport used to mangle or the charset used to alias:
# '&' split the request, '+' arrived as a space, '"' and '\\' were holes in the bisection
("http://example.org/edge", "http://example.org/raw", 'a&b+c"d\\e()'),
)
_PREDICATES = sorted(set(_[1] for _ in _TRIPLES))
_OBJECTS = sorted(_[2] for _ in _TRIPLES)
def _bind(inner, offset):
if "COUNT(*)" in inner:
return len(_TRIPLES)
if "COUNT(DISTINCT ?p)" in inner:
return len(_PREDICATES)
if "DISTINCT ?p" in inner:
return _PREDICATES[offset] if offset < len(_PREDICATES) else None
if "SELECT ?o" in inner:
return _OBJECTS[offset] if offset < len(_OBJECTS) else None
return None
def _cmp(value, expr):
match = re.match(r"^\?v >= (\d+)$", expr)
if match:
return isinstance(value, int) and value >= int(match.group(1))
match = re.match(r"^STRLEN\(STR\(\?v\)\) >= (\d+)$", expr)
if match:
return len("%s" % value) >= int(match.group(1))
# the literal may be escaped ('\"' / '\\'), the way a real store receives it
match = re.match(r'^SUBSTR\(STR\(\?v\),(\d+),1\) >= "(\\.|.)"$', expr)
if match:
pos, ch = int(match.group(1)), match.group(2)
ch = {'\\"': '"', "\\\\": "\\"}.get(ch, ch)
text = "%s" % value
return (text[pos - 1] if pos <= len(text) else "") >= ch
return False
def _predicate(pred):
pred = pred.strip()
if pred in ("1=1", "(1=1)"):
return True
if pred in ("1=2", "(1=2)"):
return False
if "FILTER(!isIRI(?zo))" in pred:
return False
if pred == "EXISTS { ?zs ?zp ?zo }":
return True
match = re.match(r"^EXISTS \{ SELECT \?v WHERE \{ (.*) FILTER\((.*)\) \} \}$", pred)
if match:
inner, expr = match.group(1).strip(), match.group(2).strip()
off = re.search(r"OFFSET (\d+)", inner)
value = _bind(inner, int(off.group(1)) if off else 0)
return value is not None and _cmp(value, expr)
return False
def _evaluate(value):
for quote, tail in (('"', '""!="'), ("'", "''!='")):
marker, suffix = '%s || (' % quote, ') || %s' % tail
if marker in value and value.endswith(suffix):
return _predicate(value.split(marker, 1)[1][:-len(suffix)])
match = re.match(r"^\d+ \|\| \((.*)\)$", value) # numeric / unquoted slot
if match:
return _predicate(match.group(1))
if value.count('"') % 2 or value.rstrip().endswith(("'", ")", ".")):
return "ERROR"
return any(o == value for _s, p, o in _TRIPLES if p.endswith("name"))
_ROWS = "<html><body><ul><li>luther</li></ul></body></html>"
_EMPTY = "<html><body><ul></ul></body></html>"
_ERROR = "<html><body><pre>Parse error: Lexical error at line 1, column 42.</pre></body></html>"
def _mockSend(place, parameter, value, raw=False):
verdict = _evaluate(value)
if verdict == "ERROR":
return _ERROR if raw else None # a 500 is nulled for the oracle, kept for the error probe
return _ROWS if verdict else _EMPTY
class TestHelpers(unittest.TestCase):
def test_is_error(self):
self.assertTrue(sparql._isError("Parse error: Lexical error at line 1, column 42."))
self.assertFalse(sparql._isError("<html><body><ul><li>luther</li></ul></body></html>"))
def test_backend_from_error(self):
self.assertEqual(sparql._backendFromError("Lexical error at line 1, column 8."), "Apache Jena / Fuseki")
self.assertEqual(sparql._backendFromError("org.eclipse.rdf4j.query.parser.ParseException"), "RDF4J / GraphDB")
self.assertIsNone(sparql._backendFromError("plain results page"))
def test_error_regex_matches_real_jena(self):
self.assertIsNotNone(re.search(SPARQL_ERROR_REGEX, "Parse error: Lexical error at line 1, column 136."))
class TestNoAmpersandInvariant(unittest.TestCase):
"""A raw '&' in a GET value is the parameter delimiter, so any payload carrying '&&' would be split
in transit and arrive as a truncated query (a 500 that mimics a dead oracle). Every boundary and
attribution predicate must therefore use SPARQL's '||' / two-FILTER conjunction, never '&&'."""
def test_boundaries_have_no_ampersand(self):
for row in sparql._BOUNDARY_TABLE:
for field in row[:4]:
self.assertNotIn("&", field, row)
def test_confirm_predicates_have_no_ampersand(self):
for truePred, falsePred in sparql._SPARQL_PREDICATES:
self.assertNotIn("&", truePred)
self.assertNotIn("&", falsePred)
def test_send_url_encodes_the_payload(self):
"""The boundaries being '&'-free was never enough: the character probes emit '&' (and '+') as
DATA. _send() must URL-encode, or those probes split the request and the bisection is answered
by a truncated query."""
sent = []
savedParams, sparql.conf.parameters = sparql.conf.parameters, {sparql.PLACE.GET: "q=luther"}
savedGet = sparql.Request.getPage
sparql.Request.getPage = staticmethod(
lambda **kwargs: (sent.append(sparql.conf.parameters[sparql.PLACE.GET]), "", None, 200)[1:])
try:
sparql._send(sparql.PLACE.GET, "q", sparql._cmpChar(1, ord("&")))
finally:
sparql.Request.getPage = savedGet
sparql.conf.parameters = savedParams
self.assertEqual(len(sent), 1)
self.assertNotIn("&", sent[0].split("=", 1)[1], sent[0])
self.assertIn("%26", sent[0])
class TestCharsetContiguity(unittest.TestCase):
"""The character recovery is a lexicographic '>=' BISECTION, so the charset must have no holes.
Excluding a codepoint does not make it come back as '?' - it makes the bisection converge on the
hole's neighbour and report a DIFFERENT character with no warning. '"' and '\\' consequently stay in
the charset and travel as their SPARQL escapes.
(This replaces an earlier assertion that pinned the opposite - that 0x22/0x5c were EXCLUDED. That
contract was wrong: live against Jena Fuseki it silently decoded 'A&x', 'A)x' and 'A+x' all as
'A%x'. The gap is the bug, not the fix.)"""
def test_charset_is_contiguous(self):
self.assertEqual(sparql._CS_ORDS,
list(range(sparql.SPARQL_CHAR_MIN, sparql.SPARQL_CHAR_MAX + 1)))
def test_meta_characters_travel_escaped(self):
self.assertEqual(sparql._literal(0x22), '"\\""')
self.assertEqual(sparql._literal(0x5c), '"\\\\"')
self.assertEqual(sparql._literal(ord('a')), '"a"')
def test_every_char_probe_is_a_well_formed_literal(self):
# exactly one opening and one closing quote, everything inside either plain or backslash-escaped
for ordinal in sparql._CS_ORDS:
probe = sparql._cmpChar(1, ordinal)
body = probe.split(">= ", 1)[1]
self.assertTrue(body.startswith('"') and body.endswith('"'), probe)
inner = body[1:-1]
self.assertEqual(inner.replace('\\"', "").replace("\\\\", "").count('"'), 0, probe)
def _mockSendNumeric(place, parameter, value, raw=False):
"""An UNQUOTED numeric slot: FILTER(?age = <value>). A string boundary lands a bare word / stray
quote in a term position, which a real store rejects outright - only the numeric shape parses."""
match = re.match(r"^\d+ \|\| \((.*)\)$", value)
if match:
return _ROWS if _predicate(match.group(1)) else _EMPTY
if value.isdigit():
return _ROWS
return _ERROR if raw else None # syntax error: nulled for the oracle, kept for the probe
class TestNumericBoundary(unittest.TestCase):
"""The unquoted numeric/term slot. The base has to be a valid SPARQL TERM there - a random word is
not one, it is a syntax error, so the row as first written could never fire (verified live: the old
shape returned 500 'Lexical error ... after prefix "zzsentinelzz"' from Jena, the new one 200)."""
def setUp(self):
self.saved, self.savedParams = sparql._send, sparql.conf.parameters
sparql._send = _mockSendNumeric
sparql.conf.parameters = {sparql.PLACE.GET: "age=30"}
sparql.SENTINEL, sparql.NUMBER_SENTINEL = "zzsentinelzz", "961962811"
def tearDown(self):
sparql._send, sparql.conf.parameters = self.saved, self.savedParams
def test_numeric_row_uses_a_numeric_base(self):
row = [_ for _ in sparql._BOUNDARY_TABLE if _[4] == sparql._BASE_NUMBER]
self.assertEqual(len(row), 1)
trueBreak, falseBreak, prefix, suffix, _kind = row[0]
boundary = sparql.Boundary(prefix, suffix, sparql._BASE_NUMBER)
self.assertEqual(sparql._base(boundary), "961962811")
# the payload must be a valid term followed by an OR - no quote to close, no paren to re-balance
self.assertEqual(sparql._base(boundary) + trueBreak, "961962811 || (1=1)")
self.assertEqual(sparql._base(boundary) + falseBreak, "961962811 || (1=2)")
self.assertEqual(sparql._wrap(sparql._base(boundary), boundary, "PRED"), "961962811 || (PRED)")
def test_string_rows_keep_the_word_base(self):
for row in [_ for _ in sparql._BOUNDARY_TABLE if _[4] == sparql._BASE_STRING]:
boundary = sparql.Boundary(row[2], row[3], sparql._BASE_STRING)
self.assertEqual(sparql._base(boundary), "zzsentinelzz")
def test_numeric_slot_detects_and_extracts(self):
template, payload, boundary = sparql._detectBoolean(sparql.PLACE.GET, "age")
self.assertIsNotNone(template)
self.assertEqual(boundary.base, sparql._BASE_NUMBER)
self.assertTrue(payload.startswith("961962811"))
truth = sparql._makeOracle(sparql.PLACE.GET, "age", boundary)
self.assertIsNotNone(truth)
self.assertEqual(sparql._inferString(truth, sparql._nthObject(0)), _OBJECTS[0])
class TestErrorStatusIsNeverAnOracleSample(unittest.TestCase):
"""A 4xx body must be a NON-ANSWER, not a cheap false. Nulling only 5xx let a front-end that serves
its parser failure as a generic 400 feed that body to the oracle, and when it resembled the FALSE
model the bit was decided FALSE instead of INCONCLUSIVE. Measured live: a stored 'A(x' came back as
'A\'x'. The error probe still needs the body, hence raw=True."""
def _sendWithCode(self, code, raw):
savedParams, sparql.conf.parameters = sparql.conf.parameters, {sparql.PLACE.GET: "q=luther"}
savedGet = sparql.Request.getPage
sparql.Request.getPage = staticmethod(lambda **kwargs: ("BODY", None, code))
try:
return sparql._send(sparql.PLACE.GET, "q", "x", raw=raw)
finally:
sparql.Request.getPage = savedGet
sparql.conf.parameters = savedParams
def test_4xx_and_5xx_are_nulled_for_the_oracle(self):
for code in (400, 403, 404, 429, 500, 503):
self.assertIsNone(self._sendWithCode(code, raw=False), code)
def test_2xx_and_3xx_are_usable(self):
for code in (200, 204, 302):
self.assertEqual(self._sendWithCode(code, raw=False), "BODY", code)
def test_raw_keeps_the_body_for_the_error_probe(self):
for code in (400, 500):
self.assertEqual(self._sendWithCode(code, raw=True), "BODY", code)
class TestOriginalValue(unittest.TestCase):
def setUp(self):
self.savedParams = sparql.conf.parameters
self.savedDict = sparql.conf.paramDict
sparql.conf.parameters = {sparql.PLACE.GET: "q=luther"}
def tearDown(self):
sparql.conf.parameters = self.savedParams
sparql.conf.paramDict = self.savedDict
def test_reads_and_replaces(self):
self.assertEqual(sparql._originalValue(sparql.PLACE.GET, "q"), "luther")
self.assertEqual(sparql._replaceSegment(sparql.PLACE.GET, "q", "X"), "q=X")
class TestDetectionAndExtraction(unittest.TestCase):
def setUp(self):
self.saved = sparql._send
sparql._send = _mockSend
self.savedParams = sparql.conf.parameters
sparql.conf.parameters = {sparql.PLACE.GET: "q=luther"}
sparql.SENTINEL = "zzsentinelzz"
def tearDown(self):
sparql._send = self.saved
sparql.conf.parameters = self.savedParams
def test_boolean_detection(self):
template, payload, boundary = sparql._detectBoolean(sparql.PLACE.GET, "q")
self.assertIsNotNone(template)
self.assertEqual(boundary.base, sparql._BASE_STRING)
def test_confirms_sparql(self):
_t, _p, boundary = sparql._detectBoolean(sparql.PLACE.GET, "q")
self.assertTrue(sparql._confirmSparql(sparql.PLACE.GET, "q", boundary))
def test_error_probe_fingerprints_jena(self):
backend, _page = sparql._probeError(sparql.PLACE.GET, "q")
self.assertEqual(backend, "Apache Jena / Fuseki")
def test_blind_extraction_recovers_objects(self):
_t, _p, boundary = sparql._detectBoolean(sparql.PLACE.GET, "q")
truth = sparql._makeOracle(sparql.PLACE.GET, "q", boundary)
self.assertIsNotNone(truth)
self.assertEqual(sparql._inferCount(truth, sparql._COUNT_TRIPLES, 10 ** 9), len(_TRIPLES))
self.assertEqual(sparql._inferCount(truth, sparql._COUNT_PREDICATES, 64), len(_PREDICATES))
# every seeded object comes back BYTE-FOR-BYTE, including the edge value whose '&', '+', '"'
# and '\' used to be silently rewritten (transport split / charset hole)
for offset, expected in enumerate(_OBJECTS):
self.assertEqual(sparql._inferString(truth, sparql._nthObject(offset)), expected)
self.assertEqual(sparql._inferString(truth, sparql._nthPredicate(0)), _PREDICATES[0])
def test_plain_sql_endpoint_is_not_confirmed(self):
# an endpoint where the injected predicate has no effect (always same page) must not confirm
sparql._send = lambda place, parameter, value, raw=False: _ROWS
template, _p, boundary = sparql._detectBoolean(sparql.PLACE.GET, "q")
self.assertIsNone(template) # no true/false divergence -> not even boolean-detected
class _Dumper(object):
"""Stands in for conf.dumper, which the offline bootstrap does not build."""
def __init__(self):
self.output = []
def singleString(self, value):
self.output.append(value)
class TestCapsAreDisclosed(unittest.TestCase):
"""Both bounds here are bisection bounds, so a result sitting ON one is a FLOOR, not a measurement.
A capped IRI reads exactly like a whole one, and a capped predicate count reads like the size of
the graph - the reader has no way to tell unless it is said."""
def setUp(self):
self.saved, self.savedParams = sparql._send, sparql.conf.parameters
sparql._send = _mockSend
sparql.conf.parameters = {sparql.PLACE.GET: "q=luther"}
sparql.SENTINEL = "zzsentinelzz"
self.savedDumper, sparql.conf.dumper = sparql.conf.get("dumper"), _Dumper()
self.savedWarning, self.warnings = sparql.logger.warning, []
sparql.logger.warning = lambda message, *args: self.warnings.append(message % args if args else message)
def tearDown(self):
sparql._send, sparql.conf.parameters = self.saved, self.savedParams
sparql.conf.dumper = self.savedDumper
sparql.logger.warning = self.savedWarning
def _truth(self):
_t, _p, boundary = sparql._detectBoolean(sparql.PLACE.GET, "q")
truth = sparql._makeOracle(sparql.PLACE.GET, "q", boundary)
self.assertIsNotNone(truth)
return truth
def test_truncated_value_is_reported(self):
value = sparql._inferString(self._truth(), sparql._nthObject(0), maxLen=4)
self.assertEqual(value, _OBJECTS[0][:4])
self.assertTrue(any("at least 4 characters" in _ for _ in self.warnings), self.warnings)
def test_value_that_fits_is_silent(self):
value = sparql._inferString(self._truth(), sparql._nthObject(0), maxLen=64)
self.assertEqual(value, _OBJECTS[0])
self.assertEqual(self.warnings, [])
def test_saturated_predicate_count_is_reported_as_a_floor(self):
saved, sparql.SPARQL_MAX_PREDICATES = sparql.SPARQL_MAX_PREDICATES, 2
try:
sparql._dumpGraph(self._truth())
finally:
sparql.SPARQL_MAX_PREDICATES = saved
self.assertTrue(any("hit the 2 cap" in _ for _ in self.warnings), self.warnings)
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in a new issue View git blame Copy permalink