-
-
Notifications
You must be signed in to change notification settings - Fork 6.4k
Expand file tree
/
Copy pathtest_union_engine.py
More file actions
188 lines (155 loc) · 7.81 KB
/
Copy pathtest_union_engine.py
File metadata and controls
188 lines (155 loc) · 7.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
The UNION-based column-count detection engine (lib/techniques/union/test.py).
_findUnionCharCount discovers how many columns a UNION injection needs. Its
fastest path is the ORDER BY technique: a valid target accepts ORDER BY 1..N and
errors on ORDER BY N+1, so it binary-searches for N. We drive the REAL function
against a mock oracle (Request.queryPage replaced) that errors once the requested
column index exceeds a known true count - exercising the actual detection +
binary search with no live target.
This requires the full injection context (conf.parameters / conf.paramDict /
kb.injection) because column detection builds real payloads via agent.payload.
"""
import os
import re
import sys
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.agent import agent
from lib.core.data import conf, kb
from lib.core.datatype import AttribDict
from lib.core.enums import PAYLOAD, PLACE
from lib.request.connect import Connect
import lib.techniques.union.test as ut
import lib.techniques.union.use as uu
MARKER = "MARKER42"
VALID_PAGE = "<html>results %s</html>" % MARKER
_CONF = {"string": MARKER, "notString": None, "regexp": None, "code": None,
"uCols": None, "uColsStart": 1, "uColsStop": 50, "base64Parameter": ()}
_KB = {"heavilyDynamic": False, "errorIsNone": False, "futileUnion": False,
"uChar": "NULL", "forceWhere": None}
class TestOrderByColumnCount(unittest.TestCase):
def setUp(self):
self._sc = {k: conf.get(k) for k in _CONF}
self._sk = {k: kb.get(k) for k in _KB}
self._sp = (conf.get("parameters"), conf.get("paramDict"))
self._sqp = Connect.queryPage
self._stmpl = kb.get("pageTemplate")
self._sinj = (kb.injection.place, kb.injection.parameter)
for k, v in _CONF.items():
conf[k] = v
for k, v in _KB.items():
kb[k] = v
conf.parameters = {PLACE.GET: "id=1"}
conf.paramDict = {PLACE.GET: {"id": "1"}}
kb.pageTemplate = VALID_PAGE
kb.injection.place = None
kb.injection.parameter = None
set_dbms("MySQL")
def tearDown(self):
for k, v in self._sc.items():
conf[k] = v
for k, v in self._sk.items():
kb[k] = v
conf.parameters, conf.paramDict = self._sp
kb.pageTemplate = self._stmpl
kb.injection.place, kb.injection.parameter = self._sinj
Connect.queryPage = self._sqp
ut.Request.queryPage = self._sqp
def _detect(self, true_count):
# canary: a leaked agent.payload stub (another module patching the shared singleton and
# restoring it by assignment) makes every probe below identical, so the ORDER BY oracle turns
# unusable and this test used to fail as a bare 'None != 25'. Name the cause instead.
self.assertNotIn("payload", agent.__dict__,
"agent.payload is stubbed - a test module leaked it (see _testutils.save_attrs)")
def oracle(payload=None, place=None, content=False, raise404=True, **kwargs):
m = re.search(r"ORDER BY (\d+)", payload or "")
cols = int(m.group(1)) if m else 1
if cols <= true_count:
page = VALID_PAGE
else:
page = "<html>Unknown column '%d' in 'order clause'</html>" % cols
return (page, {}, 200) if content else True
Connect.queryPage = staticmethod(oracle)
ut.Request.queryPage = staticmethod(oracle)
kb.orderByColumns = None
return ut._findUnionCharCount("-- -", PLACE.GET, "id", "1", "", "", PAYLOAD.WHERE.ORIGINAL)
def test_detect_single_column(self):
self.assertEqual(self._detect(1), 1)
def test_detect_small(self):
self.assertEqual(self._detect(3), 3)
def test_detect_medium(self):
self.assertEqual(self._detect(7), 7)
def test_detect_larger(self):
self.assertEqual(self._detect(12), 12)
def test_detect_beyond_first_step(self):
# > ORDER_BY_STEP (10): forces the expand-then-bisect branch
self.assertEqual(self._detect(25), 25)
class TestMssqlJsonAggFalsyValues(unittest.TestCase):
"""Regression: MSSQL UNION dumps use FOR JSON (jsonAggMode), whose output carries native JSON
types. A real 0 / '' / false is a value; only JSON null is SQL NULL. The old `row.get(field) or
NULL` mapped every falsy value to the literal string 'NULL' - proven live (SELECT 0,'','ok'
dumped as NULL,NULL,ok). We drive the REAL _oneShotUnionUse against a mock FOR JSON page."""
def setUp(self):
self._s = {
"hexConvert": conf.get("hexConvert"), "parameters": conf.get("parameters"),
"paramDict": conf.get("paramDict"), "base64Parameter": conf.get("base64Parameter"),
"pageEncoding": conf.get("pageEncoding"), "hashDB": conf.get("hashDB"),
"inj": (kb.injection.place, kb.injection.parameter, kb.injection.data),
"jsonAggMode": kb.get("jsonAggMode"), "unionDuplicates": kb.get("unionDuplicates"),
"forcePartialUnion": kb.get("forcePartialUnion"), "tableFrom": kb.get("tableFrom"),
"unionTemplate": kb.get("unionTemplate"), "qp": Connect.queryPage,
}
conf.hexConvert = False
conf.parameters = {PLACE.GET: "id=1"}
conf.paramDict = {PLACE.GET: {"id": "1"}}
conf.base64Parameter = ()
conf.pageEncoding = None
conf.hashDB = None
v = AttribDict()
v.vector = (0, 4, "", "", "", "NULL", PAYLOAD.WHERE.NEGATIVE, False, False, None, None)
kb.injection.place = PLACE.GET
kb.injection.parameter = "id"
kb.injection.data = {PAYLOAD.TECHNIQUE.UNION: v}
kb.jsonAggMode = True
kb.unionDuplicates = kb.forcePartialUnion = False
kb.tableFrom = kb.unionTemplate = None
set_dbms("MSSQL")
def tearDown(self):
conf.hexConvert = self._s["hexConvert"]
conf.parameters = self._s["parameters"]
conf.paramDict = self._s["paramDict"]
conf.base64Parameter = self._s["base64Parameter"]
conf.pageEncoding = self._s["pageEncoding"]
conf.hashDB = self._s["hashDB"]
kb.injection.place, kb.injection.parameter, kb.injection.data = self._s["inj"]
kb.jsonAggMode = self._s["jsonAggMode"]
kb.unionDuplicates = self._s["unionDuplicates"]
kb.forcePartialUnion = self._s["forcePartialUnion"]
kb.tableFrom = self._s["tableFrom"]
kb.unionTemplate = self._s["unionTemplate"]
Connect.queryPage = self._s["qp"]
uu.Request.queryPage = self._s["qp"]
def _dump(self, jsonstr):
# jsonstr is exactly what MSSQL FOR JSON AUTO, INCLUDE_NULL_VALUES emits (literal to keep key
# order stable across py2/py3); object_pairs_hook=OrderedDict preserves that order downstream
page = "%s%s%s" % (kb.chars.start, jsonstr, kb.chars.stop)
def oracle(payload=None, content=False, raise404=True, **kwargs):
return (page, {}, 200) if content else True
Connect.queryPage = staticmethod(oracle)
uu.Request.queryPage = staticmethod(oracle)
kb.jsonAggMode = True
out = uu._oneShotUnionUse("SELECT a,b,c,d FROM users", False)
firstRow = out.replace(kb.chars.start, "").split(kb.chars.stop)[0]
return firstRow.split(kb.chars.delimiter)
def test_falsy_values_preserved(self):
fields = self._dump('[{"a":0,"b":"","c":"ok","d":null}]')
self.assertEqual(fields, ["0", "", "ok", "NULL"]) # only JSON null -> NULL
if __name__ == "__main__":
unittest.main(verbosity=2)
def tearDownModule():
reset_dbms() # clear any DBMS forced via set_dbms() so it can't leak into later test modules