From cb8298d55ae9b8eb4f05b6153c158d23479958a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Tue, 4 Aug 2026 17:02:57 +0200 Subject: [PATCH] Couple of trivial bug fixes --- data/xml/queries.xml | 2 +- lib/core/common.py | 16 ++++++------ lib/core/dump.py | 9 ++++--- lib/core/settings.py | 5 +--- lib/request/comparison.py | 18 ++++++++----- lib/request/redirecthandler.py | 8 ++++-- lib/techniques/nosql/inject.py | 2 +- lib/utils/har.py | 16 ++++++++---- plugins/generic/users.py | 2 +- tests/test_datafiles.py | 16 ++++++++++++ tests/test_dump_format.py | 24 +++++++++++++++++ tests/test_har.py | 13 ++++++++++ tests/test_nosql.py | 28 ++++++++++++++++++++ tests/test_request_basic.py | 47 ++++++++++++++++++++++++++++++++++ tests/test_users_enum.py | 21 +++++++++++++++ 15 files changed, 195 insertions(+), 32 deletions(-) diff --git a/data/xml/queries.xml b/data/xml/queries.xml index 9bf22bac1..c6e1d6345 100644 --- a/data/xml/queries.xml +++ b/data/xml/queries.xml @@ -1050,7 +1050,7 @@ - + diff --git a/lib/core/common.py b/lib/core/common.py index 9580d2bac..7bc0a488a 100644 --- a/lib/core/common.py +++ b/lib/core/common.py @@ -4300,19 +4300,19 @@ def decodeStringEscape(value): >>> decodeStringEscape("a" + chr(92) + "tb") == "a" + chr(9) + "b" True + >>> decodeStringEscape(chr(92) + chr(0)) == chr(92) + chr(0) # a NUL in the data must be preserved, not rewritten to a backslash + True """ retVal = value if value and '\\' in value: - # Note: shield an escaped backslash ('\\\\') behind a marker BEFORE decoding the whitespace - # escapes, then restore it - otherwise decoding '\\\\' -> '\\' first turns a literal '\\n' - # into a newline (i.e. the round-trip with encodeStringEscape was not lossless) - _marker = "\x00" - retVal = retVal.replace("\\\\", _marker) - for _ in string.whitespace.replace(" ", ""): - retVal = retVal.replace(repr(_).strip("'"), _) - retVal = retVal.replace(_marker, "\\") + # Note: single left-to-right pass so an escaped backslash ('\\\\') shields the next char + # (a literal '\\n' stays '\\n', not a newline) WITHOUT a sentinel that could collide with a + # pre-existing byte (e.g. a NUL in the data) and get rewritten on restore + _mapping = dict((repr(_).strip("'"), _) for _ in string.whitespace.replace(" ", "")) + _mapping["\\\\"] = "\\" + retVal = re.sub("|".join(re.escape(_) for _ in ["\\\\"] + list(_mapping)), lambda match: _mapping[match.group(0)], retVal) return retVal diff --git a/lib/core/dump.py b/lib/core/dump.py index cff211659..f8b134a54 100644 --- a/lib/core/dump.py +++ b/lib/core/dump.py @@ -48,7 +48,6 @@ from lib.core.exception import SqlmapGenericException from lib.core.exception import SqlmapSystemException from lib.core.exception import SqlmapValueException from lib.core.replication import Replication -from lib.core.settings import CHECK_SQLITE_TYPE_THRESHOLD from lib.core.settings import DUMP_FILE_BUFFER_SIZE from lib.core.settings import HTML_DUMP_CSS_STYLE from lib.core.settings import IS_WIN @@ -554,7 +553,11 @@ class Dump(object): if column != "__infos__": colType = Replication.INTEGER - for i in xrange(min(CHECK_SQLITE_TYPE_THRESHOLD, len(tableValues[column]['values']))): + # Note: the type must hold for EVERY value that will be inserted - sampling only a + # prefix would type the column INTEGER/REAL while a later leading-zero/signed/overflow + # value gets silently rewritten by SQLite's affinity (the INTEGER scan breaks early on + # the first non-conforming value, so a genuine TEXT column costs almost nothing) + for i in xrange(len(tableValues[column]['values'])): value = tableValues[column]['values'][i] try: if not value or value == " ": # NULL @@ -571,7 +574,7 @@ class Dump(object): if colType is None: colType = Replication.REAL - for i in xrange(min(CHECK_SQLITE_TYPE_THRESHOLD, len(tableValues[column]['values']))): + for i in xrange(len(tableValues[column]['values'])): value = tableValues[column]['values'][i] try: if not value or value == " ": # NULL diff --git a/lib/core/settings.py b/lib/core/settings.py index 93273ac4c..4424ed251 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from lib.core.enums import OS from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.253" +VERSION = "1.10.8.0" 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) @@ -1407,9 +1407,6 @@ VALID_TIME_CHARS_RUN_THRESHOLD = 100 # Check for empty columns only if table is sufficiently large CHECK_ZERO_COLUMNS_THRESHOLD = 10 -# Threshold for checking types of columns in case of SQLite dump format -CHECK_SQLITE_TYPE_THRESHOLD = 100 - # 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") diff --git a/lib/request/comparison.py b/lib/request/comparison.py index c9a4dcc1f..27cd39f9e 100644 --- a/lib/request/comparison.py +++ b/lib/request/comparison.py @@ -230,16 +230,20 @@ def _comparison(page, headers, code, getRatioValue, pageLength): else: key = (hash(seq1), hash(seq2)) - try: - seqMatcher.set_seq1(seq1) - seqMatcher.set_seq2(seq2) - except: - seqMatcher.set_seq1(repr(seq1)) - seqMatcher.set_seq2(repr(seq2)) - ratio = kb.cache.comparison.get(key) if key else None if ratio is None: + # Note: populate the matcher only on a cache MISS - set_seq2() eagerly builds difflib's + # O(len(page)) b2j index, and since each response is a fresh string that whole build was + # thrown away on every cache hit (the common case after warmup: responses cluster into a + # few distinct pages). seqMatcher carries no state across calls that a hit would read. + try: + seqMatcher.set_seq1(seq1) + seqMatcher.set_seq2(seq2) + except: + seqMatcher.set_seq1(repr(seq1)) + seqMatcher.set_seq2(repr(seq2)) + try: try: ratio = seqMatcher.quick_ratio() if not kb.heavilyDynamic else seqMatcher.ratio() diff --git a/lib/request/redirecthandler.py b/lib/request/redirecthandler.py index 2f1472212..f7a6fce81 100644 --- a/lib/request/redirecthandler.py +++ b/lib/request/redirecthandler.py @@ -157,8 +157,12 @@ class SmartRedirectHandler(_urllib.request.HTTPRedirectHandler): elif last: cookies[last] += "%s%s" % (delimiter, part) - if HTTP_HEADER.SET_COOKIE in headers: - for match in re.finditer(r"(?:^|,\s*)([^=;,]+)=([^;,]+)", headers[HTTP_HEADER.SET_COOKIE]): + # Note: multiple cookies arrive as SEPARATE Set-Cookie headers (RFC-6265 forbids folding + # them into one comma-joined value), and __getitem__ returns only the FIRST - iterate all + # values so 2nd+ cookies (e.g. a CSRF token) are not silently dropped across the redirect + setCookies = headers.get_all(HTTP_HEADER.SET_COOKIE) if hasattr(headers, "get_all") else [headers[HTTP_HEADER.SET_COOKIE]] + for setCookie in setCookies: + for match in re.finditer(r"(?:^|,\s*)([^=;,]+)=([^;,]+)", setCookie): key = match.group(1).strip() if key.lower() not in ("expires", "path", "domain", "max-age", "secure", "httponly", "samesite"): cookies[key] = match.group(2).strip() diff --git a/lib/techniques/nosql/inject.py b/lib/techniques/nosql/inject.py index 6b41d349d..8b98c4485 100644 --- a/lib/techniques/nosql/inject.py +++ b/lib/techniques/nosql/inject.py @@ -990,7 +990,7 @@ def _resolve(place, parameter, key): falseModel = _reproduced(lambda: _fetch(place, parameter, "$in", NOSQL_SENTINEL, isArray=True)) # matches nothing return Vector(_fingerprintMongo(place, parameter), lambda value: _fetch(place, parameter, "$regex", value), - lambda n: "^.{%d,}$" % n, + lambda n: "(?s)^.{%d,}$" % n, # (?s): a value containing '\n' must still match its own length (else the $-anchored probe fails for every n -> empty result) lambda known, klass: "^%s%s" % (re.escape(known), klass), template=template, bypass='{"$ne": null}', falseModel=falseModel) diff --git a/lib/utils/har.py b/lib/utils/har.py index c9a17b25a..0eb31a1c1 100644 --- a/lib/utils/har.py +++ b/lib/utils/har.py @@ -125,11 +125,17 @@ class Request(object): } if self.postBody: - contentType = self.headers.get("Content-Type") - out["postData"] = { - "mimeType": contentType, - "text": getText(self.postBody).rstrip("\r\n"), - } + out["postData"] = {"mimeType": self.headers.get("Content-Type")} + + # HAR text must be UTF-8: a binary POST body (e.g. a file upload) that does not decode is + # base64-encoded losslessly rather than mangled through a lossy text decode - mirroring the + # Response.toDict() contract below (otherwise the exported HAR cannot reproduce the request) + raw = self.postBody if isinstance(self.postBody, bytes) else getBytes(self.postBody) + try: + out["postData"]["text"] = raw.decode("utf-8").rstrip("\r\n") + except UnicodeDecodeError: + out["postData"]["encoding"] = "base64" + out["postData"]["text"] = getText(base64.b64encode(raw)) return out diff --git a/plugins/generic/users.py b/plugins/generic/users.py index 9ce87db65..1a6025695 100644 --- a/plugins/generic/users.py +++ b/plugins/generic/users.py @@ -405,7 +405,7 @@ class Users(object): # Set containing the list of DBMS administrators areAdmins = set() - if not kb.data.cachedUsersPrivileges and any(isTechniqueAvailable(_) for _ in (PAYLOAD.TECHNIQUE.UNION, PAYLOAD.TECHNIQUE.ERROR, PAYLOAD.TECHNIQUE.QUERY)) or conf.direct: + if not kb.data.cachedUsersPrivileges and (any(isTechniqueAvailable(_) for _ in (PAYLOAD.TECHNIQUE.UNION, PAYLOAD.TECHNIQUE.ERROR, PAYLOAD.TECHNIQUE.QUERY)) or conf.direct): if Backend.isDbms(DBMS.MYSQL) and not kb.data.has_information_schema: query = rootQuery.inband.query2 condition = rootQuery.inband.condition2 diff --git a/tests/test_datafiles.py b/tests/test_datafiles.py index f8dbcabe9..4816c21e9 100644 --- a/tests/test_datafiles.py +++ b/tests/test_datafiles.py @@ -62,6 +62,22 @@ class TestQueriesXmlCoverage(unittest.TestCase): missing = [t for t in self.CORE_TAGS if t not in present] self.assertEqual(missing, [], msg="%s missing core tags: %s" % (dbms.get("value"), missing)) + def test_column_comment_queries_format_with_three_args(self): + # Regression: getColumns() formats every column_comment query with exactly (db, tbl, name) + # via '%'-formatting (plugins/generic/databases.py). A literal LIKE wildcard that is not + # escaped to '%%' (or a wrong placeholder count) raises at format time and aborts + # '--columns --comments' before any request. Vertica's entry had 'LIKE '%.%s'' (ValueError). + tree = ET.parse(os.path.join(ROOT, "data", "xml", "queries.xml")) + for dbms in tree.findall(".//dbms"): + for node in dbms.iter("column_comment"): + query = node.get("query") + if query: + try: + query % ("db", "tbl", "col") + except (ValueError, TypeError) as ex: + self.fail("%s column_comment query cannot be formatted with (db, tbl, name): %r (%s)" + % (dbms.get("value"), query, ex)) + class TestErrorsXmlCompile(unittest.TestCase): def test_all_error_regexes_compile(self): diff --git a/tests/test_dump_format.py b/tests/test_dump_format.py index 2fb052f2b..34d50558b 100644 --- a/tests/test_dump_format.py +++ b/tests/test_dump_format.py @@ -389,6 +389,30 @@ class TestSqliteDump(_FileDumpCase): finally: conn.close() + def test_type_breaking_value_past_sampling_prefix_stays_text(self): + # Regression: type inference once sampled only the first 100 values, so a leading-zero / + # signed / overflow value at index >= 100 was missed and the column got typed INTEGER, + # silently corrupting that value via SQLite affinity on insert. The whole column must be scanned. + values = [str(i) for i in range(1, 101)] + ["007"] # 100 clean ints, then a leading-zero at index 100 + tv = _PlainOrderedDict([ + ("__infos__", {"count": len(values), "db": "testdb", "table": "big"}), + ("code", {"length": 3, "values": values}), + ]) + conf.dumpFormat = DUMP_FORMAT.SQLITE + self.d.dbTableValues(tv) + + import sqlite3 + conn = sqlite3.connect(os.path.join(self.tmp, "testdb.sqlite3")) + try: + cur = conn.cursor() + cur.execute("PRAGMA table_info(big)") + types = {name: ctype for (_cid, name, ctype, _nn, _dv, _pk) in cur.fetchall()} + self.assertEqual(types["code"], "TEXT") # a single non-round-trip value anywhere forces TEXT + cur.execute("SELECT code FROM big WHERE code = '007'") + self.assertEqual(cur.fetchone(), ("007",)) # stored verbatim, not rewritten to integer 7 + finally: + conn.close() + # --- replication backend tests (pure sqlite3, no network/DBMS) ----------------------------------- diff --git a/tests/test_har.py b/tests/test_har.py index bf98f2608..9b6142624 100644 --- a/tests/test_har.py +++ b/tests/test_har.py @@ -114,6 +114,19 @@ class TestRequest(unittest.TestCase): self.assertEqual(d["postData"]["mimeType"], "application/json") self.assertIn('{"a":1}', d["postData"]["text"]) + def test_toDict_binary_postbody_base64(self): + # Regression: a non-UTF-8 (binary) POST body - e.g. a raw file upload - must be base64-encoded + # losslessly, not mangled through a lossy text decode, so the exported HAR reproduces the request. + # Mirrors the Response.toDict() contract (see TestResponse.test_toDict_binary_content_encoded). + import base64 as _b64 + payload = b"\xff\xd8\xff\xe0\x00\x10JFIF" # JPEG header: invalid UTF-8, contains a NUL + req = H.Request("POST", "/upload", "HTTP/1.1", + {"Host": "test.com", "Content-Type": "application/octet-stream"}, + postBody=payload) + d = req.toDict() + self.assertEqual(d["postData"]["encoding"], "base64") + self.assertEqual(_b64.b64decode(d["postData"]["text"]), payload) # losslessly reconstructable + def test_url_property(self): req = H.Request("GET", "/path?q=1", "HTTP/1.0", {"Host": "example.com"}) diff --git a/tests/test_nosql.py b/tests/test_nosql.py index 952af925e..deb842ce8 100644 --- a/tests/test_nosql.py +++ b/tests/test_nosql.py @@ -88,6 +88,34 @@ class TestNoSqlMongo(unittest.TestCase): lambda known, klass: "^" + re.escape(known) + klass) self.assertEqual(value, SECRET) + def test_extract_value_with_newline_not_truncated(self): + # Regression: the length probe once used '.{n,}', and PCRE '.' does not match '\n', so a + # value with an embedded newline made the $-anchored '^.{n,}$' probe fail for every n -> + # empty result (total data loss). The '(?s)' DOTALL fix counts the newline toward the length, + # so the recovered value is length-correct (the unreadable newline itself shows as '?', but + # extraction no longer collapses to ""). + secret = "ab\ncd" + + def mongo_nl(place, parameter, op, value, isArray=False): + if op == "$ne": + return MATCH + if op == "$in": + return NOMATCH + if op == "$regex": + try: + return MATCH if re.match(value, secret) is not None else NOMATCH + except re.error: + return "error" + return "" + + ni._fetch = mongo_nl + vector = ni._resolve("GET", "password", "password") + template = ni._fetch("GET", "password", "$ne", ni.NOSQL_SENTINEL) + value = ni._extract(template, vector.fetch, vector.lengthValue, vector.charValue, falseModel=vector.falseModel) + self.assertIsNotNone(value) + self.assertEqual(len(value), len(secret)) # honest length, not truncated / not empty + self.assertTrue(value.startswith("ab")) + def test_not_injectable(self): ni._fetch = lambda *args, **kwargs: MATCH self.assertIsNone(ni._detectMongo("GET", "password")) diff --git a/tests/test_request_basic.py b/tests/test_request_basic.py index 29dc53c2a..c9e69be24 100644 --- a/tests/test_request_basic.py +++ b/tests/test_request_basic.py @@ -84,6 +84,53 @@ class TestBasicDecodePage(unittest.TestCase): self.assertEqual(getText(decodePage(b"", None, "text/html")), "") +class TestRedirectSetCookieMerge(unittest.TestCase): + """A 302 that sets more than one cookie sends SEPARATE Set-Cookie headers (RFC-6265 forbids + comma-folding them). The handler must merge ALL of them into the follow-up request's Cookie + header; a __getitem__ read returns only the first, silently dropping the 2nd+ (e.g. a CSRF token).""" + + _CONF = ("cookieDel", "scope") + + def setUp(self): + self._c = dict((k, conf.get(k)) for k in self._CONF) + self._redirect = kb.choices.get("redirect") if kb.get("choices") else None + + def tearDown(self): + for k, v in self._c.items(): + conf[k] = v + if kb.get("choices"): + kb.choices.redirect = self._redirect + + def test_all_set_cookies_merged_across_redirect(self): + import email, io + from http.client import HTTPMessage + from lib.core.enums import HTTP_HEADER, REDIRECTION + from thirdparty.six.moves import urllib as _urllib + import lib.request.redirecthandler as rh + + conf.cookieDel = None + conf.scope = None + kb.choices.redirect = REDIRECTION.YES + + # stub the network-following parent so the test touches no socket + saved = _urllib.request.HTTPRedirectHandler.http_error_302 + _urllib.request.HTTPRedirectHandler.http_error_302 = lambda self, req, fp, code, msg, headers: fp + try: + req = _urllib.request.Request("http://example.com/login", headers={"Cookie": "sid=OLD"}) + raw = ("Location: http://example.com/home\r\n" + "Set-Cookie: sid=NEW; Path=/; HttpOnly\r\n" + "Set-Cookie: csrf=XYZ; Path=/\r\n\r\n") + headers = email.message_from_string(raw, _class=HTTPMessage) + fp = _urllib.response.addinfourl(io.BytesIO(b""), headers, req.get_full_url()) + rh.SmartRedirectHandler().http_error_302(req, fp, 302, "Found", headers) + finally: + _urllib.request.HTTPRedirectHandler.http_error_302 = saved + + merged = req.headers.get(HTTP_HEADER.COOKIE) or req.headers.get("Cookie") or "" + self.assertIn("sid=NEW", merged) + self.assertIn("csrf=XYZ", merged) # the 2nd Set-Cookie must survive the redirect + + class TestForgeHeadersCookieMerge(unittest.TestCase): """A domain-scoped jar cookie (Domain=example.com -> '.example.com') must merge into the request for the apex host, not be dropped by a naive endswith() domain check.""" diff --git a/tests/test_users_enum.py b/tests/test_users_enum.py index f831108b1..8191161f6 100644 --- a/tests/test_users_enum.py +++ b/tests/test_users_enum.py @@ -267,6 +267,27 @@ class TestUsersEnum(unittest.TestCase): privileges, areAdmins = users.getPrivileges() self.assertIn("root", privileges) + def test_get_privileges_cache_guard_under_direct(self): + # Regression: the inband guard must short-circuit when the cache is already + # populated, even under conf.direct. A precedence bug ('not cached and any(...) + # or conf.direct') once made the trailing 'or conf.direct' override the guard, + # so a second getPrivileges() (e.g. --privileges then --roles under -d) re-ran + # the whole enumeration. Assert the second call issues zero injection queries. + calls = {"n": 0} + + def counting_gv(query, *a, **k): + calls["n"] += 1 + return [["root", "SUPER"], ["guest", "SELECT"]] + + umod.inject.getValue = counting_gv + users = Users() + kb.data.cachedUsersPrivileges = {} + users.getPrivileges() + first = calls["n"] + self.assertGreater(first, 0) + users.getPrivileges() + self.assertEqual(calls["n"] - first, 0, "populated cache must suppress re-enumeration under --direct") + # --- getRoles (delegates to getPrivileges) ------------------------------ def test_get_roles(self):