CERBERUS/sqlmap
1
0
Fork 0
mirror of https://github.com/sqlmapproject/sqlmap.git synced 2026-08-04 14:55:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
sqlmap/tests/test_esperanto.py
Miroslav Štampar a5a7f7203a Tons of improvements for Esperanto engine
2026-07-22 01:06:57 +02:00

1095 lines
53 KiB
Python
Raw Blame History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
Regression suite for the DBMS-agnostic engine (extra/esperanto/), in two parts:
1. Semantic corpus - round-trips awkward values through an in-memory SQLite boolean
oracle and asserts the correctness invariants the engine must never break: no silent
character substitution or deletion, length never exceeds maxlen (over-length is
flagged), limit=0 is an incomplete empty prefix, NULL stays distinct from '',
truncation sets complete=False, bytes-first extraction is byte-exact, and the
capability fallbacks (pattern-match floor, length-from-substring, LEFT/RIGHT
composition) and the frozen InferenceStrategy hand-off all still extract.
2. Adversarial dialect scenarios - an intermediate oracle disguises the same SQLite as
a different / hostile back-end by BLOCKING the SQL forms the fake engine lacks and
REWRITING its forms into the SQLite equivalent, so the engine must discover and adapt
the dialect it is handed (substring=SUBSTRING, length=LEN, charcode=ASCII,
concat=CONCAT, no usable '>'/BETWEEN, no readable catalog) rather than only working
on dialects it already knows.
stdlib unittest only; Python 2.7 and 3.x.
"""
import binascii
import os
import re
import sqlite3
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from extra.esperanto import Cap
from extra.esperanto import Esperanto
from extra.esperanto import hostExtract
from extra.esperanto import Integrity
EXPR = "(SELECT v FROM t)"
# awkward values (NULL handled separately - it is not a string value)
CORPUS = [
u"", u"A", u"AB", u"A ", u" leading", u"trailing ", u"two spaces",
u"O'Reilly", u'double"quote', u"comma,value", u"back\\slash",
u"line\nbreak", u"\t", u"é", u"€", u"中文",
u"\U00010348", u"Aé€\U00010348Z",
]
def _oracle(value=None, is_null=False):
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES (?)", (None if is_null else value,))
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
return ask
# -- adversarial dialect scenarios: one SQLite disguised as many hostile back-ends -----
# ground truth seeded into every disguised scenario
_SEED = (
"CREATE TABLE users (id INTEGER, name TEXT, email TEXT)",
"INSERT INTO users VALUES (1,'luther','a@b.c'),(2,'admin','p@w.n'),(3,'wu',NULL)",
)
_SECRET_EXPR = "(SELECT name FROM users WHERE id=2)" # -> 'admin'
def _concatFn(*args):
return "".join(u"" if a is None else (a if isinstance(a, type(u"")) else str(a)) for a in args)
def _disguisedOracle(blocked=None, rewrites=(), funcs=()):
"""Boolean oracle over a disguised SQLite. `blocked` (regex) forms read False (the
fake engine lacks them); `rewrites` translate the fake engine's forms into SQLite;
`funcs` registers extra SQL functions (e.g. a variadic CONCAT SQLite lacks). The
engine never sees SQLite - it must adapt to whatever back-end the scenario emulates."""
con = sqlite3.connect(":memory:")
for name, narg, fn in funcs:
con.create_function(name, narg, fn)
for stmt in _SEED:
con.execute(stmt)
con.commit()
blk = re.compile(blocked, re.I) if blocked else None
rw = [(re.compile(p, re.I), r) for p, r in rewrites]
def ask(cond):
if blk and blk.search(cond):
return False # fake engine doesn't support it
sql = cond
for rx, rep in rw:
sql = rx.sub(rep, sql)
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % sql).fetchone()[0] == 1
except Exception:
return False
return ask
# coherent whole-back-end disguises: each blocks the forms its fake DBMS lacks and maps
# the forms it "has" onto SQLite. All must discover the dialect and dump every row.
# (name, blocked, rewrites, funcs)
_DIALECTS = (
("mysql-ish: MID / CHAR_LENGTH / ORD, backtick idents, || is OR -> CONCAT",
r"\bSUBSTR\(|\bSUBSTRING\(|\bLEN\(|\bLENGTH\(|\bASCII\(|\bUNICODE\(|\|\|",
[(r"\bMID\(", "substr("), (r"CHAR_LENGTH\(", "length("), (r"\bORD\(", "unicode("), (r"`", '"')],
[("CONCAT", -1, _concatFn)]),
("mssql-ish: SUBSTRING / LEN / UNICODE, '+' concat, [..] idents",
r"\bSUBSTR\(|\bMID\(|CHAR_LENGTH\(|\bLENGTH\(|\bASCII\(|\bORD\(|\|\||\bCONCAT\(|\"|`",
[(r"SUBSTRING\(", "substr("), (r"\bLEN\(", "length("), (r"\)\+\(", ")||(")],
()),
("postgres-ish: SUBSTRING / LENGTH / ASCII, || concat",
r"\bSUBSTR\(|\bMID\(|\bLEN\(|CHAR_LENGTH\(|\bUNICODE\(|\bORD\(",
[(r"SUBSTRING\(", "substr("), (r"\bASCII\(", "unicode(")],
()),
("oracle-ish: SUBSTR / LENGTH / ASCII / CHR, || concat",
r"\bMID\(|\bLEN\(|CHAR_LENGTH\(|\bUNICODE\(|\bORD\(|\bCHAR\(",
[(r"\bASCII\(", "unicode("), (r"\bCHR\(", "char(")],
()),
("waf: code fns and collation stripped -> forced hex extraction",
r"\bASCII\(|\bUNICODE\(|\bORD\(|CODEPOINT\(|COLLATE|AS\s+BLOB",
(), ()),
("waf: '>'/'<' stripped on top of a MID/CHAR_LENGTH dialect",
r">|<|\bSUBSTR\(|\bSUBSTRING\(|\bLENGTH\(|\bLEN\(|\|\|",
[(r"\bMID\(", "substr("), (r"CHAR_LENGTH\(", "length(")],
[("CONCAT", -1, _concatFn)]),
)
class TestEsperanto(unittest.TestCase):
def test_bytes_roundtrip(self):
# bytes-first round-trips EVERY value exactly (the fundamental guarantee)
for v in CORPUS:
esp = Esperanto(_oracle(v))
esp.discover()
self.assertEqual(esp.extractBytes(EXPR), v.encode("utf-8"), "extractBytes %r" % v)
self.assertEqual(esp.extractText(EXPR), v, "extractText %r" % v)
def test_no_silent_corruption(self):
# any deviation must be surfaced (replacement marker + warning), never a
# wrong-but-plausible character silently substituted
for v in CORPUS:
for mode in ("code", "ordinal", "collation", "hex"):
esp = Esperanto(_oracle(v))
esp.discover()
esp.dialect.compare = mode
if mode == "hex":
esp._ensureHexfn()
elif mode == "collation": # SQLite is byte-ordered via BLOB cast
esp.dialect.binwrap = Cap("cast_blob", "CAST(({x}) AS BLOB)")
res = esp.extractResult(EXPR)
if res.value == v:
continue
self.assertTrue(not res.complete and res.warnings,
"silent corruption in %s mode for %r -> %r" % (mode, v, res.value))
self.assertTrue(u"<EFBFBD>" in res.value or res.value == u"",
"%s mode invented a value for %r -> %r" % (mode, v, res.value))
def test_null_vs_empty(self):
esp = Esperanto(_oracle(is_null=True))
esp.discover()
rnull = esp.extractResult(EXPR)
self.assertTrue(rnull.is_null and rnull.value is None, "NULL not null: %r" % rnull)
esp = Esperanto(_oracle(u""))
esp.discover()
rempty = esp.extractResult(EXPR)
self.assertTrue((not rempty.is_null) and rempty.value == u"" and rempty.complete,
"empty string mis-reported: %r" % rempty)
def test_limit_and_maxlen(self):
# limit=0 on a non-empty value -> empty prefix, but INCOMPLETE/TRUNCATED
esp = Esperanto(_oracle(u"hello"))
esp.discover()
r0 = esp.extractResult(EXPR, limit=0)
self.assertTrue(r0.value == u"" and r0.truncated and not r0.complete, "limit=0: %r" % r0)
# over-maxlen -> truncated flagged, length bounded
esp = Esperanto(_oracle(u"0123456789ABCDEF"), maxlen=4)
esp.discover()
rt = esp.extractResult(EXPR)
self.assertTrue(rt.truncated and not rt.complete and len(rt.value) <= 4, "over-maxlen: %r" % rt)
def test_integer(self):
for n in (0, 1, 42, 255, 65535, 1000000, 2147483648, -1, -42):
esp = Esperanto(_oracle())
esp.discover()
self.assertEqual(esp.extractInteger("(%d)" % n), n, "extractInteger(%d)" % n)
def test_between_no_saturation(self):
# #1: BETWEEN caps range probes at the ceiling, so an out-of-range value must be
# flagged (truncated length / OverflowError), NEVER converge to a wrong SMALL value
# (the old bug read a len-16 value as length 1 and an int 100/max-10 as 0).
esp = Esperanto(_oracle(u"0123456789ABCDEF"), maxlen=8) # true length 16 > ceiling 8
esp.discover()
esp._comparator = "between"
n, trunc = esp._measureLength(EXPR, ceiling=8)
self.assertTrue(trunc and n == 8, "between over-length saturated wrong: (%r,%r)" % (n, trunc))
esp2 = Esperanto(_oracle())
esp2.discover()
esp2._comparator = "between"
self.assertEqual(esp2.extractInteger("(100)", maximum=1000), 100)
self.assertRaises(OverflowError, esp2.extractInteger, "(100)", 10)
def test_framing_requires_terminal_marker(self):
# a length-framed token MUST carry its terminal ';' - a truncated final token (no ';')
# must be rejected, not accepted as valid
esp = Esperanto(lambda c: False)
esp._rowLenFramed = True
self.assertEqual(esp._splitRow("V3:414243", 1), (None, False)) # no terminator -> invalid
self.assertEqual(esp._splitRow("V3:414243;", 1), (["ABC"], True)) # terminated -> valid
self.assertEqual(esp._splitRow("V4:414243;", 1), (None, False)) # declared 4 != decoded 3
def test_host_native_integrity_parity(self):
# hostExtract must mirror the native EXACT/AMBIGUOUS verdict: in a mode with no byte-exact
# witness (ordinal), both must be WHOLE_BUT_AMBIGUOUS, not native-ambiguous/host-exact.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('Admin-42')")
con.commit()
blk = re.compile(r"(?i)\bHEX\(|RAWTOHEX|ENCODE\(|(ASCII|UNICODE|ORD)\(|COLLATE|BLOB|BINARY")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
if esp._byteFaithful():
self.skipTest("dialect still byte-faithful")
native = esp.extractResult("(SELECT v FROM t)")
host = hostExtract(ask, esp.strategy(), "(SELECT v FROM t)")
self.assertEqual(host.value, native.value)
self.assertFalse(host.exact, "host claimed exact without a witness: %r" % host)
self.assertEqual(host.integrity, native.integrity)
def test_framing_length_witness_catches_capped_hex(self):
# a HEX fn that silently caps its output (correct up to a point, then truncates) would
# shorten a framed cell. The independent <charlen> witness in the row token (V<len>:<hex>;)
# must reject the shortened token -> fall back to cell-by-cell (which reads the true
# length via substring), so a 400-char value is recovered whole, not silently as 300.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
big = "A" * 400
con.execute("INSERT INTO t VALUES (1, ?)", (big,))
con.commit()
con.create_function("CHEX", 1, lambda s: binascii.hexlify((s or "").encode("utf-8")[:300]).decode().upper())
def ask(cond):
c = re.sub(r"UPPER\(HEX\(([^()]*)\)\)", r"CHEX(\1)", cond) # HEX -> a 300-byte-capped hex
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % c).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=8192)
esp.discover()
d = esp.dump("t")
val = d["rows"][0][1] if d and d["rows"] else None
self.assertTrue((val == big and d["complete"]) or (d and not d["complete"]),
"capped hex slipped through: %r len=%s" % ((val or "")[:8], len(val) if val else None))
def test_integer_final_equality(self):
# comparator proven on literals, but the backend rewrites '>' to '>=' for scalar/fn
# operands -> bisection converges off-by-one. The final `expr = recovered` check must
# reject it (fail closed) rather than return a silently-wrong integer.
con = sqlite3.connect(":memory:")
def ask(cond):
c = cond
if re.search(r"(SELECT|LENGTH|UNICODE|SUBSTR|\+)", c, re.I):
c = c.replace(">", ">=")
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % c).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertRaises(Exception, esp.extractInteger, "(SELECT 5)", 10) # OracleUndecided (fail closed)
def test_charset_codec_quirks(self):
from extra.esperanto.atlas import _charsetCodec
self.assertEqual(_charsetCodec("latin1"), "cp1252") # MySQL 'latin1' is Windows-1252
self.assertEqual(_charsetCodec("utf8mb4"), "utf-8")
self.assertEqual(_charsetCodec("utf8mb4_general_ci"), "utf-8") # collation suffix peeled
self.assertEqual(_charsetCodec("WE8MSWIN1252"), "cp1252") # Oracle
self.assertEqual(_charsetCodec("1252"), "cp1252") # SQL Server code page
self.assertEqual(_charsetCodec("AL32UTF8"), "utf-8")
self.assertIsNone(_charsetCodec("some_unknown_set"))
def test_host_length_code_final_equality(self):
# Round 8 #1: hostExtract must apply the native reader's final-equality to the recovered
# LENGTH and each per-char CODE - a backend that rewrites '>' to '>=' for computed operands
# (LENGTH(..)>n, UNICODE(..)>n) converges off-by-one; the '=' confirmation must fail closed
# rather than hand back silently-wrong code-point text as exact.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('Admin-42')")
con.commit()
def ask(cond):
c = cond
if re.search(r"(LENGTH|UNICODE|SUBSTR)\(", c, re.I): # '>' rewritten for computed operands
c = c.replace(">", ">=")
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % c).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
if esp.dialect.compare != "code":
self.skipTest("dialect not in code mode")
host = hostExtract(ask, esp.strategy(), "(SELECT v FROM t)")
# the invariant: anything claimed EXACT must be CORRECT. An off-by-one read is caught by
# the final equality and must surface as non-exact/failed, never as a wrong exact value.
self.assertTrue(host.value == "Admin-42" or not host.exact,
"host handed back a wrong value as exact: %r" % host)
def test_hex_unknown_codec_non_ascii_not_exact(self):
# Round 8 #2: scalar hex recovery under an UNKNOWN codec must not certify non-ASCII text
# (e.g. utf-8 bytes with an embedded NUL heuristically read as UTF-16). Bytes are faithful;
# only a PROVEN codec makes the decoded text exact.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES (?)", (u"é",)) # utf-8 C3 A9 -> non-ASCII bytes
con.execute("CREATE TABLE a (v TEXT)")
con.execute("INSERT INTO a VALUES ('plain')")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
if not esp._ensureHexfn():
self.skipTest("no hex fn")
esp.dialect.charset = None # force UNKNOWN codec
hx = esp.dialect.hexfn
esp.dialect.hexfn = Cap(hx.name, hx[1], encoding=None)
self.assertIsNone(esp._extractViaHex("(SELECT v FROM t)"), # non-ASCII + unknown codec -> refuse
"unknown-codec non-ASCII hex certified as exact")
ascii_r = esp._extractViaHex("(SELECT v FROM a)") # ASCII is codec-invariant -> still OK
self.assertTrue(ascii_r is None or ascii_r.value == "plain")
def test_bytelen_witness_fails_closed(self):
# Round 8 #5: once a byte-length witness is selected, an UNDECIDED reading must reject the
# bytes (fail closed), never treat "couldn't verify" as "matched".
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('abc')")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
if not esp._ensureHexfn():
self.skipTest("no hex fn")
esp.dialect.bytelen = Cap("blen", "OCTET_NOSUCH({expr})") # a byte-length fn that never resolves
self.assertIsNone(esp.extractBytes("(SELECT v FROM t)"),
"undecided byte-length witness slipped through")
def test_unknown_codec_nul_not_shortened(self):
# Round 8 (final) blocker #1: unknown codec + embedded NUL is genuinely ambiguous - bytes
# 41 00 are valid UTF-8 ("A"+NUL) AND valid UTF-16LE ("A"). _decodeHexToken must refuse
# rather than collapse to a shortened ASCII "A" that then reads as exact. The framed dump
# calls _decodeHexToken directly, so the guard has to live there (not just _extractViaHex).
esp = Esperanto(lambda c: False)
self.assertIsNone(esp._decodeHexToken("4100", None)) # ambiguous -> refuse
self.assertEqual(esp._decodeHexToken("4100", "utf-8"), u"A\x00") # proven codec keeps NUL
self.assertEqual(esp._decodeHexToken("41", None), u"A") # pure ASCII still fine
def test_lossy_cast_marked_inexact(self):
# Round 8 (final) blocker #2: a text cast that FOLDS non-ASCII to "?" (café -> caf?) yields
# an ASCII-only result, so gating the cast-safety canary on _hasNonAscii(rows) meant it
# never ran on the very failure it defends against. The canary must run whenever a textcast
# was applied, and an unproven cast must not certify source identity.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
con.execute(u"INSERT INTO t VALUES (1, 'café')")
con.commit()
con.create_function("LOSSY", 1, lambda s: "".join(c if ord(c) < 128 else "?" for c in (s or "")))
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
esp.dialect.charset = None # unknown -> exercise the canary path
esp.dialect.textcast = Cap("lossy", "LOSSY({expr})") # a narrowing/folding cast
esp._castPreserves = None
self.assertFalse(esp._castPreservesAccents(), "folding cast wrongly certified as preserving")
d = esp.dump("t")
self.assertTrue(d and d["rows"], "dump returned nothing")
self.assertFalse(d["exact"], "folded (café->caf?) dump reported source-exact: %r" % d["rows"])
def test_repl_chars_escalate_to_hex(self):
# a code fn lossy for a column type (Oracle NVARCHAR2 read in code mode marks non-ASCII
# chars outside its alphabet -> _REPL). Rather than return the marked-incomplete value,
# the engine must escalate to the cast+hex path (which the framed dump proves works) and
# recover the value exactly. Without hex, it stays honestly incomplete (no false-complete).
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute(u"INSERT INTO t VALUES ('café-€')")
con.commit()
con.create_function("UNICODE", 1, lambda s: (ord(s[0]) if s and ord(s[0]) < 128 else None))
def make(block_hex):
pat = r"\bASCII\(|\bORD\(" + (r"|RAWTOHEX|\bHEX\(|ENCODE\(" if block_hex else "")
blk = re.compile(pat, re.I)
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
return ask
with_hex = Esperanto(make(False)); with_hex.discover()
r = with_hex.extractResult("(SELECT v FROM t)")
self.assertEqual(r.value, u"café-€", "hex escalation failed to recover _REPL value: %r" % r.value)
no_hex = Esperanto(make(True)); no_hex.discover()
r2 = no_hex.extractResult("(SELECT v FROM t)")
self.assertFalse(r2.complete, "no-hex _REPL value must stay incomplete, got complete: %r" % r2.value)
def test_keyset_cycle_detection(self):
# Round 8 (keyset): paging and read-back can resolve under different collations, so the
# walk can revisit an earlier name (f,e,f,e...). Full cycle detection (not just the
# immediate predecessor) must stop with a partial, de-duplicated list, never loop.
esp = Esperanto(lambda c: True) # _keysetWalk only, no discovery
seq = iter(["f", "e", "f", "e", "f"])
class _R(object):
def __init__(self, v):
self.value, self.complete, self.exact, self.is_null = v, True, True, False
esp.extractResult = lambda *a, **k: _R(next(seq))
esp._ask = lambda c: True
esp._beyondSql = lambda *a, **k: "1=1"
names = esp._keysetWalk("name", "cat", "", 10)
self.assertEqual(names, ["f", "e"], "cycle not detected/deduped: %r" % names)
def test_terminal_sanitized(self):
from extra.esperanto.__main__ import _safeterm
self.assertEqual(_safeterm(u"ok"), u"ok")
self.assertEqual(_safeterm(u"a\x1b[2Jb"), u"a\\x1b[2Jb") # ANSI clear-screen neutralized
self.assertEqual(_safeterm(u"x\ny"), u"x\\x0ay") # embedded newline neutralized
def test_comparator_rejects_gte_rewrite(self):
# a '>' -> '>=' rewrite passes 2>1 / !2>3 but FAILS the equality boundary (2>2 must be
# False). the full truth table must reject 'gt' (and fall to BETWEEN) so counts/lengths
# aren't read off-by-one. reproduced: extractInteger('5',max=10) used to return 6.
con = sqlite3.connect(":memory:")
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond.replace(">", ">=")).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertNotEqual(esp._comparator, "gt", "'>'->'>=' rewrite wrongly accepted as gt")
self.assertEqual(esp.extractInteger("(5)", maximum=10), 5)
def test_exact_requires_byte_witness(self):
# without a proven hex/binary witness (or code-codepoint), a value is WHOLE_BUT_AMBIGUOUS,
# never EXACT - plain '=' is collation-dependent and can't certify byte-exactness.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('Zz')")
con.commit()
blk = re.compile(r"(?i)\bHEX\(|RAWTOHEX|ENCODE\(|(ASCII|UNICODE|ORD)\(|GLOB|COLLATE|BLOB|BINARY")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
r = esp.extractResult("(SELECT v FROM t)")
if esp._byteFaithful():
self.skipTest("dialect still has a byte-faithful primitive")
self.assertEqual(r.value, "Zz")
self.assertFalse(r.exact, "value marked exact without a byte-faithful witness: %r" % r)
def test_extractresult_no_contradiction(self):
# a hard defect (incomplete/truncated) is classified before null-exactness
from extra.esperanto import ExtractResult, Integrity
self.assertEqual(ExtractResult(None, is_null=True, complete=False).integrity, Integrity.FAILED)
self.assertEqual(ExtractResult(None, is_null=True, complete=True).integrity, Integrity.EXACT)
self.assertFalse(ExtractResult(None, is_null=True, complete=False).exact)
def test_host_maxlen_zero(self):
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('abcdef')")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
r = hostExtract(ask, esp.strategy(), "(SELECT v FROM t)", maxlen=0)
self.assertEqual(r.value, "")
self.assertTrue(r.truncated and not r.complete)
def test_between_overflow_magnitude(self):
# a positive value beyond BETWEEN's global ceiling fails closed WITHOUT claiming "below
# minimum" (the direction is genuinely unknown to a bounded range predicate)
esp = Esperanto(_oracle())
esp.discover()
esp._comparator = "between"
try:
esp.extractInteger("(%d)" % ((1 << 62) + 5))
self.fail("expected OverflowError")
except OverflowError as ex:
self.assertNotIn("below minimum", str(ex))
def test_integrity_semantics(self):
# `complete` (walk finished) must be distinct from `exact` (bytes proven identical).
# a case-insensitive collation recovers a WHOLE value that is NOT exact.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('A')")
con.commit()
blk = re.compile(r"(?i)\bHEX\(|RAWTOHEX|ENCODE\(|(ASCII|UNICODE|ORD)\(|GLOB|COLLATE|BLOB")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s COLLATE NOCASE) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
self.assertEqual(esp.dialect.compare, "equality-ci")
r = esp.extractResult(EXPR.replace("v FROM t", "v FROM t"))
self.assertTrue(r.complete, "case-ci value should be WHOLE: %r" % r)
self.assertFalse(r.exact, "case-ci value must NOT be exact: %r" % r)
self.assertEqual(r.integrity, Integrity.WHOLE_BUT_AMBIGUOUS)
def test_hostextract_structured_and_tristate(self):
# hostExtract returns an ExtractResult: a bounded read is TRUNCATED (not a bare string),
# and an undecided (None) host observation degrades to a FAILED result, never a fake bit.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute("INSERT INTO t VALUES ('abcdef')")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask, maxlen=64)
esp.discover()
strat = esp.strategy()
r = hostExtract(ask, strat, "(SELECT v FROM t)", maxlen=3)
self.assertEqual(r.value, "abc")
self.assertTrue(r.truncated and not r.exact and r.integrity == Integrity.TRUNCATED)
undecided = hostExtract(lambda c: None, strat, "(SELECT v FROM t)")
self.assertTrue(undecided.value is None and not undecided.complete)
def test_between_overflow_direction(self):
# a positive value above the cap in BETWEEN mode must report ABOVE maximum (not below)
esp = Esperanto(_oracle())
esp.discover()
esp._comparator = "between"
try:
esp.extractInteger("(100)", maximum=10)
self.fail("expected OverflowError")
except OverflowError as ex:
self.assertIn("exceeds maximum", str(ex))
def test_key_uniqueness_gate(self):
# #2: a keyset ordering column MUST be single-column unique + non-NULL. a composite
# key's first column repeats -> `> prev` paging silently drops rows sharing it, so it
# must be REJECTED (COUNT(*) != COUNT(DISTINCT col)); only a unique column is accepted.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE k (a INT, b INT, u INT, nul INT)")
con.executemany("INSERT INTO k VALUES (?,?,?,?)", [(1, 1, 10, 1), (1, 2, 20, None), (2, 1, 30, 3)])
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask)
esp.discover()
self.assertFalse(esp._keyIsUnique("a", "k"), "non-unique composite-first column accepted")
self.assertFalse(esp._keyIsUnique("nul", "k"), "nullable column accepted as key")
self.assertTrue(esp._keyIsUnique("u", "k"), "unique non-null column rejected")
def test_fixup_length(self):
# a trailing-space-trimming length fn is rebuilt as LEN(x||'.')-1
esp = Esperanto(_oracle(u"x"))
esp.discover()
esp.dialect.length = Cap("LEN", "LEN({expr})", unit="characters", trailing=False, empty_is_null=False)
esp.dialect.concat = Cap("plus", "({a})+({b})")
esp._fixupLength()
self.assertTrue(esp.dialect.length.name.endswith("+dot") and esp.dialect.length.get("trailing"),
"_fixupLength did not rebuild: %r" % esp.dialect.length)
def test_lit_escaping(self):
esp = Esperanto(_oracle(u"x"))
esp.discover()
esp._backslashEscape = True
self.assertEqual(esp._lit("\\"), "'\\\\'")
esp._backslashEscape = False
self.assertEqual(esp._lit("\\"), "'\\'")
self.assertEqual(esp._lit("'"), "''''")
def test_build_literal(self):
esp = Esperanto(_oracle(u"x"))
esp.discover()
lit = esp.buildLiteral("a\\b'c")
self.assertTrue(esp._ask("%s IS NOT NULL" % lit), "buildLiteral invalid SQL: %s" % lit)
self.assertEqual(esp.extract(lit), "a\\b'c")
def test_coalesce(self):
esp = Esperanto(_oracle(u"x"))
esp.discover()
if esp.dialect.coalesce:
self.assertEqual(esp.extract(esp.coalesce("NULL", "'Z'")), "Z")
def test_enumerate(self):
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE alpha (x)")
con.execute("CREATE TABLE beta (y)")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask)
esp.discover()
tabs = set(esp.enumerate("table", limit=10) or [])
self.assertTrue(set(["alpha", "beta"]).issubset(tabs), "enumerate tables: %r" % tabs)
def test_dump(self):
# row DATA byte-exact, including commas / quotes / unicode (hex framing keeps intact)
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE users (id, uname, note)")
truth = [(1, u"admin", u"all,good"), (2, u"o'brien", u"café,€"), (3, u"x", u"")]
for row in truth:
con.execute("INSERT INTO users VALUES (?,?,?)", row)
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask)
esp.discover()
d = esp.dump("users", limit=10)
got = set(frozenset(dict(zip(d["columns"], r)).items()) for r in d["rows"])
want = set(frozenset({"id": str(i), "uname": u, "note": n}.items()) for i, u, n in truth)
self.assertTrue(d["complete"] and got == want, "dump mismatch: %r" % d["rows"])
def test_dump_scavenges_without_hex_or_concat(self):
# DOOMSDAY: a back-end with NO hex function AND NO concat operator (nothing to
# frame a whole row with) must still be dumped - cell by cell, one value per
# extraction. Quotes/commas/NULLs in the data must survive (a single value has
# no framing ambiguity). This is the scavenger's whole reason to exist.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE users (id INTEGER, name TEXT, email TEXT)")
for row in ((1, "luther", "a@b.c"), (2, "o'brien", "x,y@z"), (3, "wu", None)):
con.execute("INSERT INTO users VALUES (?,?,?)", row)
con.commit()
blk = re.compile(r"(?i)\bHEX\(|RAWTOHEX|ENCODE\(|BINTOHEX|HEX_ENCODE|TO_HEX|BINTOSTR|"
r"\|\||\bCONCAT\(|CONCAT_WS|\)\+\(|\)&\(") # no hex, no concat at all
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertIsNone(esp.dialect.hexfn)
self.assertIsNone(esp.dialect.concat)
d = esp.dump("users")
got = set(frozenset(dict(zip(d["columns"], r)).items()) for r in d["rows"])
want = set(frozenset(v.items()) for v in (
{"id": "1", "name": "luther", "email": "a@b.c"},
{"id": "2", "name": "o'brien", "email": "x,y@z"},
{"id": "3", "name": "wu", "email": None}))
self.assertTrue(d["complete"] and got == want, "cell-by-cell dump: %r" % d["rows"])
def test_enumerate_without_count(self):
# COUNT() filtered (a WAF, or an exotic engine) must not kill discovery:
# catalog detection, brute-force existence probing, and identifier quoting are
# all COUNT-free (scalar-subquery existence), so tables/columns/dump still work.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE users (id INTEGER, name TEXT)")
con.execute("INSERT INTO users VALUES (1, 'admin'), (2, 'root')")
con.commit()
blk = re.compile(r"(?i)\bCOUNT\s*\(")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertEqual(esp.enumerate("table", limit=10), ["users"])
self.assertEqual(sorted(esp.columns("users")), ["id", "name"])
rows = (esp.dump("users") or {}).get("rows") or []
got = set(frozenset(dict(zip(esp.dump("users")["columns"], r)).items()) for r in rows)
want = set(frozenset(v.items()) for v in ({"id": "1", "name": "admin"}, {"id": "2", "name": "root"}))
self.assertEqual(got, want)
def test_quoting(self):
# reserved-word / spaced column names must be quoted, not interpolated raw
con = sqlite3.connect(":memory:")
con.execute('CREATE TABLE q (id INTEGER, "order" TEXT, "group by" TEXT)')
con.execute('INSERT INTO q VALUES (1, ?, ?)', ("a'b", "x,y"))
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
dq = esp.dump("q")
got = set(frozenset(dict(zip(dq["columns"], r)).items()) for r in dq["rows"])
want = set([frozenset({"id": "1", "order": "a'b", "group by": "x,y"}.items())])
self.assertTrue(dq["complete"] and got == want and esp.dialect.identQuote,
"quoted-identifier dump failed: %r" % dq["rows"])
def test_strategy_handoff(self):
# the frozen InferenceStrategy is a *sufficient* host interface, and immutable
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE k (v TEXT)")
con.execute("INSERT INTO k VALUES ('Str4t3gy!')")
con.commit()
def ask(cond):
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
strat = esp.strategy()
self.assertRaises(AttributeError, setattr, strat, "compare_mode", "x")
hr = hostExtract(ask, strat, "(SELECT v FROM k)")
self.assertEqual(hr.value, "Str4t3gy!")
self.assertTrue(hr.exact)
self.assertTrue(strat.asQueriesRow()["substring"])
def test_pattern_match_fallback(self):
# SUBSTR + LENGTH + hex + code fns ALL blacklisted -> pure GLOB/LIKE floor
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE flag (id INTEGER, v TEXT)")
con.execute("INSERT INTO flag VALUES (1, 'FLAG{no_SUBSTR_%_needed}')")
con.commit()
blk = re.compile(r"(?i)\bSUBSTR|SUBSTRING|\bMID\(|\bLENGTH|CHAR_LENGTH|\bLEN\(|"
r"OCTET_LENGTH|LENGTHB|DATALENGTH|(ASCII|UNICODE|ORD)\(|\bHEX\(|"
r"RAWTOHEX|ENCODE\(")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertIsNotNone(esp.dialect.prefix)
self.assertEqual(esp.extract("(SELECT v FROM flag WHERE id=1)"), "FLAG{no_SUBSTR_%_needed}")
def test_like_floor_escapes_wildcard_chars(self):
# the classic trap: on the LIKE floor '_' and '%' ARE the wildcards, so a value
# like 'all_products' must escape them (\_ ESCAPE '\'), not treat them as
# match-anything. GLOB is blocked here to force LIKE (where '_'/'%' are magic).
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE v (val TEXT)")
con.execute("INSERT INTO v VALUES ('all_products')")
for t in ("all_products", "wp_users", "sales%2024"): # '_'/'%' in identifiers
con.execute('CREATE TABLE "%s" (id)' % t)
con.commit()
blk = re.compile(r"(?i)\bSUBSTR|SUBSTRING|\bMID\(|\bLENGTH|CHAR_LENGTH|\bLEN\(|"
r"OCTET_LENGTH|LENGTHB|DATALENGTH|(ASCII|UNICODE|ORD|CODEPOINT)\(|"
r"\bHEX\(|RAWTOHEX|ENCODE\(|GLOB") # +GLOB -> only LIKE left
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
esp.discover()
self.assertEqual(esp.dialect.prefix.name, "LIKE") # GLOB gone -> LIKE floor
self.assertEqual(esp.extract("(SELECT val FROM v)"), "all_products")
tabs = esp.enumerate("table", limit=10) or []
self.assertEqual(sorted(tabs), ["all_products", "sales%2024", "v", "wp_users"])
def test_length_from_substring(self):
# every length fn blacklisted but SUBSTR present -> length derived from the end
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t6 (v TEXT)")
con.execute(u"INSERT INTO t6 VALUES ('Admin-42€')")
con.commit()
blk = re.compile(r"(?i)\b(CHAR_LENGTH|LENGTH|LEN)\s*\(")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
d = esp.discover()
self.assertIsNone(d.length)
self.assertEqual(esp.extract("(SELECT v FROM t6)"), u"Admin-42€")
def test_enumeration_degrades_not_crashes(self):
# a permission/charset wall mid-walk (oracle can't decide the keyset bound)
# must STOP with partial results, never crash with OracleUndecided
con = sqlite3.connect(":memory:")
for t in ("alpha", "beta", "gamma"):
con.execute("CREATE TABLE %s (x)" % t)
con.commit()
def ask(cond):
if "name>" in cond.replace(" ", "").lower(): # the keyset bound
raise RuntimeError("simulated permission/charset wall")
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask)
esp.discover()
names = esp.enumerate("table", limit=10) # must not raise
self.assertEqual(names, ["alpha"])
self.assertTrue(any("stopped early" in n for n in esp.dialect.notes))
def test_left_right_rung(self):
# SUBSTR/SUBSTRING/MID blocked but LEFT+RIGHT present -> RIGHT(LEFT(x,p),1).
# LEFT/RIGHT are reserved JOIN keywords in SQLite; some builds reject LEFT(/RIGHT(
# as function calls (syntax error, keyword) rather than invoking a same-named UDF,
# so register the shims under non-keyword names and translate the engine's LEFT(/
# RIGHT( onto them (the same rewrite the disguised-dialect oracles use). Keeps the
# test portable across SQLite builds while still exercising the real left_right rung.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t7 (v TEXT)")
con.execute("INSERT INTO t7 VALUES ('Zagreb-42')")
con.commit()
con.create_function("esp_left", 2, lambda s, n: (s or "")[:max(n, 0)])
con.create_function("esp_right", 2, lambda s, n: (s or "")[len(s or "") - n:] if n > 0 else "")
blk = re.compile(r"(?i)\b(SUBSTR|SUBSTRING|SUBSTRC|MID)\s*\(")
rwLeft = re.compile(r"(?i)\bLEFT\s*\(")
rwRight = re.compile(r"(?i)\bRIGHT\s*\(")
def ask(cond):
if blk.search(cond):
return False
sql = rwRight.sub("esp_right(", rwLeft.sub("esp_left(", cond))
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % sql).fetchone()[0] == 1
except sqlite3.DatabaseError:
return False
esp = Esperanto(ask)
d = esp.discover()
self.assertTrue(d.substring is not None and d.substring[0] == "left_right")
self.assertEqual(esp.extract("(SELECT v FROM t7)"), "Zagreb-42")
# -- adversarial dialect scenarios ---------------------------------------------
def _adapt(self, oracle):
esp = Esperanto(oracle)
esp.discover()
return esp, esp.extract(_SECRET_EXPR)
def test_disguise_substring_is_SUBSTRING(self):
# SUBSTR/MID blocked; the engine "has" SUBSTRING -> mapped to SQLite substr
esp, got = self._adapt(_disguisedOracle(
blocked=r"\bSUBSTR\(|\bMID\(|\bSUBSTRC\(",
rewrites=[(r"SUBSTRING\(", "substr(")]))
self.assertEqual(esp.dialect.substring[0], "SUBSTRING")
self.assertEqual(got, "admin")
def test_disguise_length_is_LEN(self):
# CHAR_LENGTH/LENGTH blocked; the engine "has" LEN -> mapped to SQLite length
esp, got = self._adapt(_disguisedOracle(
blocked=r"CHAR_LENGTH\(|\bLENGTH\(",
rewrites=[(r"\bLEN\(", "length(")]))
self.assertEqual(esp.dialect.length[0], "LEN")
self.assertEqual(got, "admin")
def test_disguise_charcode_is_ASCII(self):
# UNICODE/ORD blocked; the engine "has" ASCII -> mapped to SQLite unicode()
esp, got = self._adapt(_disguisedOracle(
blocked=r"\bUNICODE\(|\bORD\(|CODEPOINT\(",
rewrites=[(r"\bASCII\(", "unicode(")]))
self.assertEqual(esp.dialect.charcode[0], "ASCII")
self.assertEqual(esp.dialect.compare, "code")
self.assertEqual(got, "admin")
def test_disguise_concat_is_CONCAT_not_pipes(self):
# || blocked (as if it were logical OR, MySQL-style); engine has variadic CONCAT
esp, got = self._adapt(_disguisedOracle(
blocked=r"\|\|",
funcs=[("CONCAT", -1, _concatFn)]))
self.assertEqual(esp.dialect.concat[0], "concat")
self.assertEqual(got, "admin")
def test_disguise_no_catalog_brute_forces_schema(self):
# every system catalog denied: the engine knows NOTHING about the schema and
# must brute-force the table + columns, then dump
esp = Esperanto(_disguisedOracle(
blocked=r"sqlite_master|INFORMATION_SCHEMA|SYS\.|SYSIBM|pg_catalog|pg_tables|"
r"syscat|RDB\$|master\.\.|sys\.tables|sys\.schemas"))
esp.discover()
self.assertIsNone(esp.dialect.catalog)
self.assertEqual(esp.enumerate("table"), ["users"])
self.assertEqual(esp.columns("users"), ["id", "name", "email"])
rows = (esp.dump("users") or {}).get("rows")
self.assertEqual(rows, [["1", "luther", "a@b.c"], ["2", "admin", "p@w.n"], ["3", "wu", None]])
def test_disguise_gt_blocked_falls_to_between(self):
# a WAF that strips '<'/'>' must not break bisection - retry via BETWEEN
esp = Esperanto(_disguisedOracle(blocked=r">|<"))
esp.discover()
self.assertEqual(esp._comparator, "between")
self.assertEqual(esp.extract(_SECRET_EXPR), "admin")
self.assertEqual(len((esp.dump("users") or {}).get("rows") or []), 3) # BETWEEN keyset pages all
def test_disguise_gt_and_between_blocked_use_operator_free(self):
# '<'/'>' AND BETWEEN gone: rather than drop to slow order-free membership, the ladder
# finds an ORDERED operator-free rung (SIGN((e)-(n))=1 etc.) that needs no comparison
# operator - keeping efficient log2 bisection alive.
esp = Esperanto(_disguisedOracle(blocked=r">|<|BETWEEN"))
esp.discover()
self.assertEqual(esp._comparator, "sign")
self.assertEqual(esp.extract(_SECRET_EXPR), "admin")
self.assertEqual(len((esp.dump("users") or {}).get("rows") or []), 3) # ordered rung still pages all
def test_disguise_no_ordering_no_in_still_extracts(self):
# the hard floor: no '<'/'>', no BETWEEN, no IN, and no operator-free ordered rung
# (SIGN/ABS/LEAST/GREATEST/NULLIF/WIDTH_BUCKET/INTERVAL) - only '=' equality. Values
# still extract (linear scan); multi-row dump honestly degrades (can't page)
esp = Esperanto(_disguisedOracle(
blocked=r">|<|BETWEEN|\bIN\s*\(|SIGN\(|ABS\(|LEAST\(|GREATEST\(|NULLIF\(|WIDTH_BUCKET\(|INTERVAL\("))
esp.discover()
self.assertEqual(esp.extract(_SECRET_EXPR), "admin")
rows = (esp.dump("users") or {}).get("rows")
self.assertTrue(rows and rows[0] == ["1", "luther", "a@b.c"])
def test_disguise_everything_at_once(self):
# the monster: SUBSTRING + LEN + ASCII(->unicode) + variadic CONCAT (no ||) +
# NO catalog, all at once - the engine must adapt to every disguise and dump
esp = Esperanto(_disguisedOracle(
blocked=(r"\bSUBSTR\(|\bMID\(|\bSUBSTRC\(|CHAR_LENGTH\(|\bLENGTH\(|"
r"\bUNICODE\(|\bORD\(|CODEPOINT\(|\|\||"
r"sqlite_master|INFORMATION_SCHEMA|SYS\.|SYSIBM|pg_catalog|pg_tables|"
r"syscat|RDB\$|master\.\.|sys\.tables|sys\.schemas"),
rewrites=[(r"SUBSTRING\(", "substr("), (r"\bLEN\(", "length("), (r"\bASCII\(", "unicode(")],
funcs=[("CONCAT", -1, _concatFn)]))
esp.discover()
self.assertEqual(esp.dialect.substring[0], "SUBSTRING")
self.assertEqual(esp.dialect.length[0], "LEN")
self.assertEqual(esp.dialect.charcode[0], "ASCII")
self.assertEqual(esp.dialect.concat[0], "concat")
self.assertIsNone(esp.dialect.catalog)
rows = (esp.dump("users") or {}).get("rows")
self.assertEqual(rows, [["1", "luther", "a@b.c"], ["2", "admin", "p@w.n"], ["3", "wu", None]])
def test_disguise_dialect_profiles(self):
# each coherent fake back-end must be discovered from scratch and yield every
# row byte-exact - not just the one capability the disguise touches (column
# ORDER is a dialect detail, so compare rows as column->value maps)
want = set(frozenset(d.items()) for d in (
{"id": "1", "name": "luther", "email": "a@b.c"},
{"id": "2", "name": "admin", "email": "p@w.n"},
{"id": "3", "name": "wu", "email": None}))
for name, blocked, rewrites, funcs in _DIALECTS:
esp = Esperanto(_disguisedOracle(blocked, rewrites, funcs))
esp.discover()
self.assertEqual(esp.extract(_SECRET_EXPR), "admin", "%s: extract" % name)
d = esp.dump("users") or {}
got = set(frozenset(dict(zip(d.get("columns") or [], r)).items()) for r in (d.get("rows") or []))
self.assertEqual(got, want, "%s: dump %r" % (name, d.get("rows")))
def test_disguise_bracket_quoting_reserved_words(self):
# '"' and backtick idents blocked -> the engine must fall to [..] quoting to
# reach reserved-word columns; commas/quotes in the data must survive framing
seed = ('CREATE TABLE q (id INTEGER, "order" TEXT, "group" TEXT)',
"INSERT INTO q VALUES (1, ?, ?)")
con = sqlite3.connect(":memory:")
con.execute(seed[0]); con.execute(seed[1], ("a,b", "x'y")); con.commit()
blk = re.compile(r'"|`')
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask); esp.discover()
dq = esp.dump("q")
self.assertEqual(esp.dialect.identQuote, ("[", "]")) # forced past " and ` to [..]
got = set(frozenset(dict(zip(dq["columns"], r)).items()) for r in dq["rows"])
want = set([frozenset({"id": "1", "order": "a,b", "group": "x'y"}.items())])
self.assertTrue(dq["complete"] and got == want, "bracket-quoted dump: %r" % dq["rows"])
def test_disguise_lossy_charcode_escalates_to_hex(self):
# a first-byte charcode fn (MySQL-style ASCII) is lossy for non-ASCII; the
# engine must detect that and escalate to hex, recovering the bytes exactly
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (v TEXT)")
con.execute(u"INSERT INTO t VALUES ('café-€')") # cafe-EUR
con.commit()
con.create_function("ASCII", 1, lambda s: (bytearray(s.encode("utf-8"))[0] if s else None))
blk = re.compile(r"\bUNICODE\(|\bORD\(|CODEPOINT\(")
def ask(cond):
if blk.search(cond):
return False
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % cond).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask); esp.discover()
self.assertEqual(esp.dialect.charcode[0], "ASCII")
self.assertEqual(esp.extractText("(SELECT v FROM t)"), u"café-€")
def test_disguise_unicode_through_dialect(self):
# non-ASCII data recovered byte-exact even while the dialect is disguised
# (SUBSTR/MID/CHAR_LENGTH/LENGTH blocked -> SUBSTRING/LEN mapped to SQLite)
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE s (v TEXT)")
con.execute(u"INSERT INTO s VALUES ('Zagreb-župa-€42')") # Zagreb-zupa-EUR42
con.commit()
blk = re.compile(r"\bSUBSTR\(|\bMID\(|CHAR_LENGTH\(|\bLENGTH\(", re.I)
rw = [(re.compile(r"SUBSTRING\(", re.I), "substr("), (re.compile(r"\bLEN\(", re.I), "length(")]
def ask(cond):
if blk.search(cond):
return False
sql = cond
for rx, rep in rw:
sql = rx.sub(rep, sql)
try:
return con.execute("SELECT CASE WHEN (%s) THEN 1 ELSE 0 END" % sql).fetchone()[0] == 1
except Exception:
return False
esp = Esperanto(ask); esp.discover()
self.assertEqual(esp.extractText("(SELECT v FROM s)"), u"Zagreb-župa-€42")
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in a new issue View git blame Copy permalink