#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
Consolidated unit coverage for lib/core/common.py.
This module merges the previously separate test_common_utils.py,
test_common_parsers.py and the common.py-specific classes from
test_core_more.py, test_core_extra.py and test_core_final.py into a single
file. Test logic is unchanged from those sources.
Everything runs in isolation (no network, no DBMS, no persistent filesystem
mutation of the project). Any function that reads/writes global conf/kb/Backend
state has that state saved and restored around the call so test ordering stays
irrelevant. Temp files go to the session scratchpad and are removed.
stdlib unittest only (no pytest / no pip); works on Python 2.7 and 3.x.
"""
import atexit
import base64
import os
import shutil
import sys
import tempfile
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, paths
from lib.core.defaults import defaults
from lib.core.enums import (
CHARSET_TYPE,
DBMS,
EXPECTED,
HTTPMETHOD,
PLACE,
SORT_ORDER,
)
from lib.core.exception import (
SqlmapSystemException,
)
from lib.core.settings import (
NULL,
PAYLOAD_DELIMITER,
REFLECTED_VALUE_MARKER,
)
from lib.core.common import (
aliasToDbmsEnum,
applyFunctionRecursively,
arrayizeValue,
Backend,
boldifyMessage,
calculateDeltaSeconds,
checkFile,
checkOldOptions,
checkSystemEncoding,
cleanReplaceUnicode,
commonFinderOnly,
enumValueToNameLookup,
extractErrorMessage,
extractExpectedValue,
extractRegexResult,
extractTextTagContent,
filePathToSafeString,
filterListValue,
filterNone,
filterPairValues,
filterStringValue,
findMultipartPostBoundary,
findPageForms,
flattenValue,
Format,
getCharset,
getFilteredPageContent,
getHeader,
getLimitRange,
getPageWordSet,
getPartRun,
getRequestHeader,
getSQLSnippet,
getTechnique,
getText,
intersect,
isDBMSVersionAtLeast,
isListLike,
isNoneValue,
isNullValue,
isNumber,
isNumPosStrValue,
isWindowsDriveLetterPath,
isZipFile,
joinValue,
listToStrValue,
normalizeUnicode,
paramToDict,
parseJson,
parsePasswordHash,
parseRequestFile,
parseTargetDirect,
parseTargetUrl,
parseUnionPage,
removePostHintPrefix,
removeReflectiveValues,
resetCookieJar,
safeExpandUser,
safeFilepathEncode,
safeStringFormat,
safeSQLIdentificatorNaming,
saveConfig,
serializeObject,
setColor,
setTechnique,
splitFields,
trimAlphaNum,
unArrayizeValue,
unserializeObject,
urlencode,
zeroDepthSearch,
)
SCRATCH = tempfile.mkdtemp(prefix="sqlmap-tests-") # per-run temp dir (portable; replaces a stale hardcoded path)
atexit.register(lambda: shutil.rmtree(SCRATCH, ignore_errors=True))
def _write_temp(content, suffix):
"""Write `content` (str) to a scratchpad temp file, return its path."""
if not os.path.isdir(SCRATCH):
os.makedirs(SCRATCH)
handle, path = tempfile.mkstemp(suffix=suffix, dir=SCRATCH)
os.write(handle, content.encode("utf-8") if isinstance(content, str) else content)
os.close(handle)
return path
class _FakeRequest(object):
"""Minimal stand-in for urllib2.Request used by getRequestHeader()."""
def __init__(self, headers):
self.headers = headers
def header_items(self):
return self.headers.items()
# =========================================================================== #
# from tests/test_common_utils.py
# =========================================================================== #
class TestParamToDict(unittest.TestCase):
"""Parameter string -> OrderedDict for the various injection places."""
def test_get_two_params(self):
result = paramToDict(PLACE.GET, "id=1&name=foo")
self.assertEqual(list(result.items()), [("id", "1"), ("name", "foo")])
def test_get_preserves_order(self):
result = paramToDict(PLACE.GET, "c=3&a=1&b=2")
self.assertEqual(list(result.keys()), ["c", "a", "b"])
def test_post_place(self):
result = paramToDict(PLACE.POST, "user=admin&pass=secret")
self.assertEqual(result["user"], "admin")
self.assertEqual(result["pass"], "secret")
def test_empty_value(self):
result = paramToDict(PLACE.GET, "id=&name=x")
self.assertEqual(result["id"], "")
self.assertEqual(result["name"], "x")
def test_value_with_equal_signs(self):
# value is re-joined on '=' so embedded '=' survives
result = paramToDict(PLACE.GET, "token=a=b=c")
self.assertEqual(result["token"], "a=b=c")
def test_cookie_delimiter(self):
# COOKIE place splits on ';' rather than '&'
result = paramToDict(PLACE.COOKIE, "foo=bar;baz=qux")
self.assertEqual(list(result.items()), [("foo", "bar"), ("baz", "qux")])
def test_param_without_equals_ignored(self):
# an element with no '=' has len(parts) < 2 and is skipped
result = paramToDict(PLACE.GET, "lonely&id=1")
self.assertEqual(list(result.items()), [("id", "1")])
def test_html_entity_in_value_not_split(self):
# a genuine HTML entity in a value must not be split on its '&'/';' (any entity length)
result = paramToDict(PLACE.GET, "q=foo&bar&id=5")
self.assertEqual(list(result.items()), [("q", "foo&bar"), ("id", "5")])
result = paramToDict(PLACE.GET, "q=foo—bar&id=5")
self.assertEqual(list(result.items()), [("q", "foo—bar"), ("id", "5")])
def test_html_entity_in_cookie_value_not_corrupted(self):
# regression: the entity's own ';' must not act as the cookie delimiter
result = paramToDict(PLACE.COOKIE, "token=a—b; id=5")
self.assertEqual(list(result.items()), [("token", "a—b"), ("id", "5")])
def test_non_entity_ampersand_still_splits(self):
# "&nope;" is not a real entity, so '&' remains a genuine delimiter
result = paramToDict(PLACE.GET, "a=1&nope=2")
self.assertEqual(list(result.items()), [("a", "1"), ("nope", "2")])
class TestGetCharset(unittest.TestCase):
"""Inference charsets are fixed integer tables."""
def test_binary(self):
self.assertEqual(getCharset(CHARSET_TYPE.BINARY), [0, 1, 47, 48, 49])
def test_default_is_full_ascii(self):
self.assertEqual(getCharset(None), list(range(0, 128)))
def test_digits(self):
result = getCharset(CHARSET_TYPE.DIGITS)
self.assertEqual(result, list(range(0, 10)) + list(range(47, 58)))
def test_alpha_has_no_digits(self):
result = getCharset(CHARSET_TYPE.ALPHA)
# ASCII codes for '0'..'9' are 48..57; ALPHA must exclude them
self.assertFalse(any(48 <= _ <= 57 for _ in result))
self.assertIn(ord("A"), result)
self.assertIn(ord("z"), result)
def test_alphanum_superset_of_alpha(self):
alpha = set(getCharset(CHARSET_TYPE.ALPHA))
alphanum = set(getCharset(CHARSET_TYPE.ALPHANUM))
self.assertTrue(alpha.issubset(alphanum))
self.assertIn(ord("5"), alphanum)
def test_hexadecimal_contains_hex_letters(self):
result = getCharset(CHARSET_TYPE.HEXADECIMAL)
for ch in "0123456789abcdefABCDEF":
self.assertIn(ord(ch), result, msg="missing %r" % ch)
class TestGetLimitRange(unittest.TestCase):
def test_basic(self):
self.assertEqual(list(getLimitRange(10)), list(range(0, 10)))
def test_plus_one(self):
self.assertEqual(list(getLimitRange(3, plusOne=True)), [1, 2, 3])
def test_string_count_coerced(self):
# count is int()-coerced internally
self.assertEqual(list(getLimitRange("4")), [0, 1, 2, 3])
def test_length(self):
self.assertEqual(len(getLimitRange(7)), 7)
class TestParseUnionPage(unittest.TestCase):
def test_none(self):
self.assertIsNone(parseUnionPage(None))
def test_two_entries(self):
page = "%sfoo%s%sbar%s" % (kb.chars.start, kb.chars.stop, kb.chars.start, kb.chars.stop)
# returns a BigArray; compare element-wise
self.assertEqual(list(parseUnionPage(page)), ["foo", "bar"])
def test_single_entry_unwrapped(self):
# a lone wrapped string is returned as the bare string, not a 1-element list
page = "%shello%s" % (kb.chars.start, kb.chars.stop)
self.assertEqual(parseUnionPage(page), "hello")
def test_multi_column_row(self):
# a single row whose values are joined by kb.chars.delimiter becomes one
# nested list entry
page = "%sa%sb%s" % (kb.chars.start, kb.chars.delimiter, kb.chars.stop)
self.assertEqual(list(parseUnionPage(page)), [["a", "b"]])
def test_unmarked_page_returned_verbatim(self):
self.assertEqual(parseUnionPage("no markers here"), "no markers here")
class TestSafeStringFormat(unittest.TestCase):
def test_basic_tuple(self):
self.assertEqual(safeStringFormat("SELECT foo FROM %s LIMIT %d", ("bar", "1")),
"SELECT foo FROM bar LIMIT 1")
def test_literal_percent_preserved(self):
self.assertEqual(
safeStringFormat("SELECT foo FROM %s WHERE name LIKE '%susan%' LIMIT %d", ("bar", "1")),
"SELECT foo FROM bar WHERE name LIKE '%susan%' LIMIT 1")
def test_single_string_param(self):
self.assertEqual(safeStringFormat("a %s b", "X"), "a X b")
def test_scalar_non_string(self):
self.assertEqual(safeStringFormat("n=%d", 5), "n=5")
class TestUrlencode(unittest.TestCase):
def test_basic(self):
self.assertEqual(urlencode("AND 1>(2+3)#"), "AND%201%3E%282%2B3%29%23")
def test_none(self):
self.assertIsNone(urlencode(None))
def test_spaceplus(self):
self.assertEqual(urlencode("a b", spaceplus=True), "a+b")
def test_convall_encodes_safe_chars(self):
# with convall the explicit 'safe' set is dropped, so '/' gets encoded
self.assertEqual(urlencode("a/b", convall=True), "a%2Fb")
def test_safe_char_default_kept(self):
# by default '-' and '_' are in the safe set
self.assertEqual(urlencode("a-b_c"), "a-b_c")
class TestParseTargetUrl(unittest.TestCase):
"""parseTargetUrl mutates conf.* in place; save and restore everything touched."""
def _save(self):
return {k: conf.get(k) for k in
("url", "scheme", "path", "hostname", "port", "ipv6")}
def _restore(self, saved):
for k, v in saved.items():
conf[k] = v
def test_https_url(self):
saved = self._save()
orig_params = conf.parameters.get(PLACE.GET)
try:
conf.url = "https://www.test.com/?id=1"
parseTargetUrl()
self.assertEqual(conf.hostname, "www.test.com")
self.assertEqual(conf.scheme, "https")
self.assertEqual(conf.port, 443)
self.assertEqual(conf.parameters[PLACE.GET], "id=1")
finally:
self._restore(saved)
if orig_params is None:
conf.parameters.pop(PLACE.GET, None)
else:
conf.parameters[PLACE.GET] = orig_params
def test_scheme_defaulted_and_port(self):
saved = self._save()
try:
conf.url = "example.org:8080/app"
parseTargetUrl()
self.assertEqual(conf.hostname, "example.org")
self.assertEqual(conf.scheme, "http")
self.assertEqual(conf.port, 8080)
finally:
self._restore(saved)
def test_empty_url_returns_none(self):
saved = self._save()
try:
conf.url = ""
self.assertIsNone(parseTargetUrl())
finally:
self._restore(saved)
class TestParseTargetDirect(unittest.TestCase):
"""parseTargetDirect under smokeMode (early-returns before driver imports)."""
def _save(self):
return {k: conf.get(k) for k in
("direct", "dbms", "dbmsUser", "dbmsPass", "dbmsDb", "hostname", "port")}
def _restore(self, saved):
for k, v in saved.items():
conf[k] = v
def test_full_mysql_dsn(self):
saved = self._save()
orig_smoke = kb.smokeMode
orig_none = conf.parameters.get(None)
try:
kb.smokeMode = True
conf.direct = "mysql://root:testpass@127.0.0.1:3306/testdb"
parseTargetDirect()
self.assertEqual(conf.dbms, "mysql")
self.assertEqual(conf.dbmsUser, "root")
self.assertEqual(conf.dbmsPass, "testpass")
self.assertEqual(conf.dbmsDb, "testdb")
self.assertEqual(conf.hostname, "127.0.0.1")
self.assertEqual(conf.port, 3306)
finally:
self._restore(saved)
kb.smokeMode = orig_smoke
if orig_none is None:
conf.parameters.pop(None, None)
else:
conf.parameters[None] = orig_none
def test_quoted_password(self):
saved = self._save()
orig_smoke = kb.smokeMode
orig_none = conf.parameters.get(None)
try:
kb.smokeMode = True
conf.direct = "mysql://user:'P@ssw0rd'@127.0.0.1:3306/test"
parseTargetDirect()
self.assertEqual(conf.dbmsPass, "P@ssw0rd")
self.assertEqual(conf.hostname, "127.0.0.1")
finally:
self._restore(saved)
kb.smokeMode = orig_smoke
if orig_none is None:
conf.parameters.pop(None, None)
else:
conf.parameters[None] = orig_none
def test_empty_direct_returns_none(self):
saved = self._save()
try:
conf.direct = None
self.assertIsNone(parseTargetDirect())
finally:
self._restore(saved)
class TestSafeSQLIdentificatorNaming(unittest.TestCase):
"""Quoting of identifiers is DBMS-specific; drive it via kb.forcedDbms."""
def _run(self, dbms, name, **kw):
orig = kb.forcedDbms
try:
kb.forcedDbms = dbms
return getText(safeSQLIdentificatorNaming(name, **kw))
finally:
kb.forcedDbms = orig
def test_mssql_keyword_bracketed(self):
self.assertEqual(self._run(DBMS.MSSQL, "begin"), "[begin]")
def test_plain_name_unquoted(self):
self.assertEqual(self._run(DBMS.MSSQL, "foobar"), "foobar")
def test_firebird_name_with_space_double_quoted(self):
self.assertEqual(self._run(DBMS.FIREBIRD, "foo bar"), '"foo bar"')
def test_mysql_keyword_backticked(self):
self.assertEqual(self._run(DBMS.MYSQL, "select"), "`select`")
def test_oracle_keyword_uppercased(self):
# Oracle quotes AND uppercases reserved words
self.assertEqual(self._run(DBMS.ORACLE, "table"), '"TABLE"')
def test_unsafe_naming_passthrough(self):
orig = conf.unsafeNaming
try:
conf.unsafeNaming = True
self.assertEqual(self._run(DBMS.MYSQL, "select"), "select")
finally:
conf.unsafeNaming = orig
class TestGetPartRun(unittest.TestCase):
def test_no_dbms_handler_in_stack(self):
# called from a test (no conf.dbmsHandler.* on the stack) -> None
self.assertIsNone(getPartRun())
def test_non_alias_form_also_none(self):
self.assertIsNone(getPartRun(alias=False))
# =========================================================================== #
# from tests/test_common_parsers.py
# =========================================================================== #
class TestParseRequestFileBurp(unittest.TestCase):
"""_parseBurpLog via parseRequestFile (plain '=====' log + Burp XML history)."""
def setUp(self):
self._scope = conf.scope
self._method = conf.method
self._headers = conf.headers
conf.scope = None
conf.method = None # avoid a leaked conf.method overriding the parsed verb
def tearDown(self):
conf.scope = self._scope
conf.method = self._method
conf.headers = self._headers
def test_plain_burp_log_get(self):
content = (
"======================================================\n"
"GET http://www.target.com:80/vuln.php?id=1 HTTP/1.1\n"
"Host: www.target.com\n"
"Cookie: PHPSESSID=abc\n"
"======================================================\n"
)
path = _write_temp(content, ".log")
try:
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(len(targets), 1)
url, method, data, cookie, headers = targets[0]
self.assertEqual(url, "http://www.target.com:80/vuln.php?id=1")
self.assertEqual(method, HTTPMETHOD.GET)
self.assertIsNone(data)
self.assertEqual(cookie, "PHPSESSID=abc")
self.assertIn(("Host", "www.target.com"), headers)
def test_burp_xml_history_base64_request(self):
req = "GET /vuln.php?id=1 HTTP/1.1\r\nHost: www.target.com\r\nCookie: SID=xyz\r\n\r\n"
b64 = base64.b64encode(req.encode()).decode()
xml = ('- 80'
''
'
' % b64)
path = _write_temp(xml, ".xml")
try:
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(len(targets), 1)
url, method, data, cookie, headers = targets[0]
self.assertEqual(url, "http://www.target.com:80/vuln.php?id=1")
self.assertEqual(method, HTTPMETHOD.GET)
self.assertEqual(cookie, "SID=xyz")
def test_post_body_captured(self):
content = (
"======================================================\n"
"POST http://www.target.com:80/login HTTP/1.1\n"
"Host: www.target.com\n"
"Content-Length: 17\n"
"\n"
"user=admin&pw=1\n"
"======================================================\n"
)
path = _write_temp(content, ".log")
try:
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(len(targets), 1)
url, method, data, cookie, headers = targets[0]
self.assertEqual(method, HTTPMETHOD.POST)
self.assertEqual(data, "user=admin&pw=1")
def test_scope_filters_out_nonmatching(self):
content = (
"======================================================\n"
"GET http://www.target.com:80/vuln.php?id=1 HTTP/1.1\n"
"Host: www.target.com\n"
"======================================================\n"
)
path = _write_temp(content, ".log")
try:
conf.scope = r"example\.org" # does not match target.com
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(targets, [])
class TestParseRequestFileWebScarab(unittest.TestCase):
"""_parseWebScarabLog via parseRequestFile."""
def setUp(self):
self._scope = conf.scope
conf.scope = None
def tearDown(self):
conf.scope = self._scope
def test_get_conversation(self):
content = (
"### Conversation : 1\n"
"URL: http://www.target.com/vuln.php?id=1\n"
"METHOD: GET\n"
"COOKIE: SID=abc\n"
)
path = _write_temp(content, ".log")
try:
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(len(targets), 1)
url, method, data, cookie, headers = targets[0]
self.assertEqual(url, "http://www.target.com/vuln.php?id=1")
self.assertEqual(method, "GET")
self.assertIsNone(data)
self.assertEqual(cookie, "SID=abc")
self.assertEqual(headers, tuple())
def test_post_conversation_skipped(self):
# POST bodies live in separate files -> WebScarab POSTs are skipped
content = (
"### Conversation : 1\n"
"URL: http://www.target.com/login\n"
"METHOD: POST\n"
)
path = _write_temp(content, ".log")
try:
targets = list(parseRequestFile(path))
finally:
os.unlink(path)
self.assertEqual(targets, [])
class TestParseTargetDirectNonSmoke(unittest.TestCase):
"""parseTargetDirect() non-smoke branch: resolves the canonical DBMS name.
Uses SQLite because its driver (stdlib sqlite3) is always importable.
"""
_KEYS = ("direct", "dbms", "dbmsUser", "dbmsPass", "dbmsDb", "hostname", "port")
def setUp(self):
self._saved = {k: conf.get(k) for k in self._KEYS}
self._smoke = kb.smokeMode
self._params_none = conf.parameters.get(None)
def tearDown(self):
for k, v in self._saved.items():
conf[k] = v
kb.smokeMode = self._smoke
if self._params_none is None:
conf.parameters.pop(None, None)
else:
conf.parameters[None] = self._params_none
def test_sqlite_local_dsn(self):
kb.smokeMode = False
conf.direct = "sqlite://%s" % os.path.join(SCRATCH, "test.db")
parseTargetDirect()
# non-smoke path canonicalizes the DBMS name via DBMS_DICT
self.assertEqual(conf.dbms, DBMS.SQLITE)
# local file DBMS: hostname forced to localhost, port 0
self.assertEqual(conf.hostname, "localhost")
self.assertEqual(conf.port, 0)
self.assertEqual(conf.parameters[None], "direct connection")
class TestRemoveReflectiveValues(unittest.TestCase):
def setUp(self):
self._mech = kb.reflectiveMechanism
self._heur = kb.heuristicMode
kb.reflectiveMechanism = True
kb.heuristicMode = False
def tearDown(self):
kb.reflectiveMechanism = self._mech
kb.heuristicMode = self._heur
def test_reflected_payload_masked(self):
content = u"You searched for 1 AND 1=2 here"
out = removeReflectiveValues(content, "1 AND 1=2")
self.assertIn(REFLECTED_VALUE_MARKER, out)
self.assertNotIn("AND 1=2", out)
def test_no_reflection_returns_content_unchanged(self):
content = u"nothing interesting"
out = removeReflectiveValues(content, "1 AND 1=2")
self.assertEqual(out, content)
def test_none_payload_returns_content(self):
content = u"x"
self.assertEqual(removeReflectiveValues(content, None), content)
def test_bytes_content_returned_as_is(self):
# non-text content short-circuits (isinstance text_type check)
content = b"1 AND 1=2"
self.assertEqual(removeReflectiveValues(content, "1 AND 1=2"), content)
class TestFindPageForms(unittest.TestCase):
def setUp(self):
self._scope = conf.scope
self._crawlExclude = conf.crawlExclude
self._cookie = conf.cookie
conf.scope = None
conf.crawlExclude = None
conf.cookie = None
def tearDown(self):
conf.scope = self._scope
conf.crawlExclude = self._crawlExclude
conf.cookie = self._cookie
def test_post_form_discovered(self):
html = ('')
forms = findPageForms(html, "http://www.site.com")
self.assertEqual(forms, set([("http://www.site.com/input.php", "POST", "id=1", None, None)]))
def test_get_form_discovered(self):
html = ('')
forms = findPageForms(html, "http://www.site.com")
self.assertEqual(len(forms), 1)
url, method, data, _cookie, _ = list(forms)[0]
self.assertEqual(method, "GET")
self.assertIn("q=x", url)
def test_inline_js_post_discovered(self):
# the `.post('url', {k: v})` regex branch (independent of HTML form parsing)
html = ""
forms = findPageForms(html, "http://www.site.com")
self.assertTrue(any(m == HTTPMETHOD.POST and u.endswith("/api/save") for (u, m, d, c, e) in forms))
def test_blank_content_returns_empty_set(self):
self.assertEqual(findPageForms("", "http://www.site.com"), set())
class TestSaveConfig(unittest.TestCase):
def test_writes_ini_with_sections(self):
path = _write_temp("", ".ini")
try:
saveConfig(conf, path)
with open(path) as f:
data = f.read()
finally:
os.unlink(path)
# optDict families become [Section] headers
self.assertIn("[Target]", data)
self.assertIn("[Request]", data)
self.assertIn("[Enumeration]", data)
self.assertTrue(len(data) > 0)
class TestGetSQLSnippet(unittest.TestCase):
def test_mssql_proc_loaded(self):
snippet = getSQLSnippet(DBMS.MSSQL, "activate_sp_oacreate")
self.assertIn("RECONFIGURE", snippet)
def test_variable_substitution(self):
# %VAR% placeholders are substituted from kwargs (here %ENABLE%);
# supplying it avoids the interactive "provide substitution values" prompt.
snippet = getSQLSnippet(DBMS.MSSQL, "configure_xp_cmdshell", ENABLE="1")
self.assertIn("xp_cmdshell", snippet)
self.assertIn("RECONFIGURE", snippet)
# comments (#...) are stripped and the placeholder is fully resolved
self.assertNotIn("#", snippet)
self.assertNotIn("%ENABLE%", snippet)
class TestCheckSystemEncoding(unittest.TestCase):
def test_noop_on_normal_encoding(self):
# On a normal default encoding this is a no-op and must not raise.
self.assertIsNone(checkSystemEncoding())
class TestFormatGetOs(unittest.TestCase):
def setUp(self):
self._api = conf.api
conf.api = False
def tearDown(self):
conf.api = self._api
def test_humanizes_type_and_technology(self):
info = {
"type": set(["Linux"]),
"distrib": set(["Ubuntu"]),
"release": set(["8.10"]),
"technology": set(["PHP 5.2.6", "Apache 2.2.9"]),
}
out = Format.getOs("back-end DBMS", info)
self.assertTrue(out.startswith("back-end DBMS operating system: Linux"))
self.assertIn("Ubuntu", out)
self.assertIn("8.10", out)
self.assertIn("web application technology:", out)
def test_api_mode_returns_dict(self):
orig = conf.api
try:
conf.api = True
info = {"type": set(["Windows"]), "technology": set(["IIS"])}
out = Format.getOs("back-end DBMS", info)
self.assertIsInstance(out, dict)
self.assertIn("web application technology", out)
finally:
conf.api = orig
class TestBackendSetters(unittest.TestCase):
"""Backend OS/version setters write kb state; save and restore it."""
_KEYS = ("os", "osVersion", "osSP", "dbmsVersion")
def setUp(self):
self._saved = {k: kb.get(k) for k in self._KEYS}
def tearDown(self):
for k, v in self._saved.items():
kb[k] = v
def test_set_get_os(self):
kb.os = None
self.assertEqual(Backend.setOs("windows"), "Windows") # capitalized
self.assertEqual(Backend.getOs(), "Windows")
def test_set_os_none_returns_none(self):
self.assertIsNone(Backend.setOs(None))
def test_set_os_version(self):
kb.osVersion = None
Backend.setOsVersion("2008")
self.assertEqual(Backend.getOsVersion(), "2008")
def test_set_os_service_pack(self):
kb.osSP = None
Backend.setOsServicePack(3)
self.assertEqual(Backend.getOsServicePack(), 3)
def test_set_get_version(self):
kb.dbmsVersion = []
self.assertEqual(Backend.setVersion("5.7"), ["5.7"])
self.assertEqual(Backend.getVersion(), "5.7")
def test_set_version_list(self):
kb.dbmsVersion = []
Backend.setVersionList(["8.0", "8.1"])
self.assertEqual(Backend.getVersionList(), ["8.0", "8.1"])
class TestUrlencodeExtraBranches(unittest.TestCase):
def test_like_percent_encoded(self):
# '%' inside a LIKE '...' literal is encoded to %25
self.assertEqual(urlencode("AND name LIKE '%DBA%'"),
"AND%20name%20LIKE%20%27%25DBA%25%27")
def test_convall_drops_safe_set(self):
self.assertEqual(urlencode("a&b", convall=True), "a%26b")
def test_limit_does_not_crash_on_long_input(self):
out = urlencode("x " * 4000, limit=True)
self.assertTrue(len(out) > 0)
def test_direct_mode_returns_value_unchanged(self):
orig = conf.direct
try:
conf.direct = "mysql://u:p@h:3306/d"
self.assertEqual(urlencode("a b"), "a b")
finally:
conf.direct = orig
class TestSafeStringFormatExtraBranches(unittest.TestCase):
def test_percent_d_in_payload_region_becomes_string(self):
fmt = "SELECT %s" + PAYLOAD_DELIMITER + " AND %d " + PAYLOAD_DELIMITER
self.assertEqual(
safeStringFormat(fmt, ("a", "5")),
"SELECT a" + PAYLOAD_DELIMITER + " AND 5 " + PAYLOAD_DELIMITER)
def test_scalar_string_percent_preserved(self):
# single-string param path: plain replace, embedded '%' survives
self.assertEqual(safeStringFormat("LIKE %s", "100%done"), "LIKE 100%done")
def test_two_params_list(self):
self.assertEqual(safeStringFormat("%s/%s", ("a", "b")), "a/b")
# =========================================================================== #
# from tests/test_core_more.py (common.py classes)
# =========================================================================== #
class TestSmallPredicates(unittest.TestCase):
def test_is_none_value(self):
self.assertTrue(isNoneValue(None))
self.assertTrue(isNoneValue("None"))
self.assertTrue(isNoneValue(""))
self.assertTrue(isNoneValue([]))
self.assertTrue(isNoneValue(["None", ""]))
self.assertTrue(isNoneValue({}))
self.assertFalse(isNoneValue([2]))
self.assertFalse(isNoneValue("x"))
def test_is_null_value(self):
self.assertTrue(isNullValue(u"NULL"))
self.assertTrue(isNullValue(u"null"))
self.assertFalse(isNullValue(u"foobar"))
self.assertFalse(isNullValue(5))
def test_is_num_pos_str_value(self):
self.assertTrue(isNumPosStrValue(1))
self.assertTrue(isNumPosStrValue("1"))
self.assertFalse(isNumPosStrValue(0))
self.assertFalse(isNumPosStrValue("-2"))
self.assertFalse(isNumPosStrValue("100000000000000000000"))
self.assertFalse(isNumPosStrValue("abc"))
def test_is_number(self):
self.assertTrue(isNumber(1))
self.assertTrue(isNumber("0"))
self.assertTrue(isNumber("3.14"))
self.assertFalse(isNumber("foobar"))
self.assertFalse(isNumber(None))
def test_is_list_like(self):
self.assertTrue(isListLike([1]))
self.assertTrue(isListLike((1,)))
self.assertTrue(isListLike(set([1])))
self.assertFalse(isListLike("x"))
self.assertFalse(isListLike(5))
class TestValueShaping(unittest.TestCase):
def test_filter_pair_values(self):
self.assertEqual(filterPairValues([[1, 2], [3], 1, [4, 5]]), [[1, 2], [4, 5]])
self.assertEqual(filterPairValues(None), [])
def test_filter_list_value(self):
self.assertEqual(filterListValue(["users", "admins", "logs"], r"(users|admins)"),
["users", "admins"])
# non-list input returned unchanged
self.assertEqual(filterListValue("notlist", r"x"), "notlist")
# no regex returns input
self.assertEqual(filterListValue(["a"], None), ["a"])
def test_filter_none(self):
self.assertEqual(filterNone([1, 2, "", None, 3, 0]), [1, 2, 3, 0])
def test_filter_string_value(self):
self.assertEqual(filterStringValue("wzydeadbeef0123#", r"[0-9a-f]"), "deadbeef0123")
def test_un_arrayize_value(self):
self.assertEqual(unArrayizeValue(["1"]), "1")
self.assertEqual(unArrayizeValue("1"), "1")
self.assertEqual(unArrayizeValue(["1", "2"]), "1")
self.assertEqual(unArrayizeValue([["a", "b"], "c"]), "a")
self.assertIsNone(unArrayizeValue([]))
def test_flatten_value(self):
self.assertEqual(list(flattenValue([["1"], [["2"], "3"]])), ["1", "2", "3"])
def test_arrayize_value(self):
self.assertEqual(arrayizeValue("1"), ["1"])
self.assertEqual(arrayizeValue(["1"]), ["1"])
def test_join_value(self):
self.assertEqual(joinValue(["1", "2"]), "1,2")
self.assertEqual(joinValue("1"), "1")
self.assertEqual(joinValue(["1", None]), "1,None")
class TestZeroDepthAndSplit(unittest.TestCase):
def test_zero_depth_search_skips_parens(self):
expr = "SELECT (SELECT id FROM users WHERE 2>1) AS r FROM DUAL"
idx = zeroDepthSearch(expr, " FROM ")
# only the outer top-level FROM is found, not the one inside the subselect
self.assertEqual(len(idx), 1)
self.assertTrue(expr[idx[0]:].startswith(" FROM DUAL"))
def test_zero_depth_search_ignores_quoted(self):
expr = "a , 'b , c' , d"
# commas inside the quoted literal are not reported
self.assertEqual(len(zeroDepthSearch(expr, ",")), 2)
def test_split_fields_basic(self):
self.assertEqual(splitFields("foo, bar, max(foo, bar)"),
["foo", "bar", "max(foo,bar)"])
def test_split_fields_quoted(self):
self.assertEqual(splitFields("a, 'b, c', d"), ["a", "'b, c'", "d"])
def test_split_fields_custom_delimiter(self):
self.assertEqual(splitFields("a; b; max(c; d)", delimiter=";"),
["a", "b", "max(c;d)"])
class TestAliasToDbmsEnum(unittest.TestCase):
def test_known_aliases(self):
self.assertEqual(aliasToDbmsEnum("mssql"), DBMS.MSSQL)
self.assertEqual(aliasToDbmsEnum("mysql"), DBMS.MYSQL)
self.assertEqual(aliasToDbmsEnum("postgres"), DBMS.PGSQL)
def test_unknown_alias_returns_none(self):
self.assertIsNone(aliasToDbmsEnum("definitely_not_a_dbms"))
def test_empty_returns_none(self):
self.assertIsNone(aliasToDbmsEnum(""))
class TestIsDBMSVersionAtLeast(unittest.TestCase):
"""Version gating drives per-DBMS query selection; comparison must be component-wise."""
def setUp(self):
self._saved = kb.get("dbmsVersion")
def tearDown(self):
kb.dbmsVersion = self._saved
def _at_least(self, version, minimum):
kb.dbmsVersion = version
return isDBMSVersionAtLeast(minimum)
def test_single_major_thresholds(self):
self.assertTrue(self._at_least("5.4.3", "5"))
self.assertTrue(self._at_least("8.0.32", "8"))
self.assertFalse(self._at_least("5.7.44", "8"))
def test_multi_digit_minor_ordering(self):
# floats mis-sorted these (10.11->10.11 vs 10.5->10.5): component-wise fixes it
self.assertTrue(self._at_least("10.11", "10.5")) # MariaDB
self.assertFalse(self._at_least("10.6", "10.11"))
self.assertTrue(self._at_least("5.10.0", "5.5"))
self.assertTrue(self._at_least("9.10", "9.6")) # PostgreSQL
def test_presto_sequential_minor(self):
# Presto 0.NNN: release 99 is OLDER than release 178 (float made 0.99 > 0.178)
self.assertFalse(self._at_least("0.99", "0.178"))
self.assertTrue(self._at_least("0.180", "0.178"))
def test_range_and_prefix_semantics(self):
self.assertTrue(self._at_least("2", ">=2.0"))
self.assertFalse(self._at_least("2", ">2"))
self.assertFalse(self._at_least("<2", "2"))
self.assertTrue(self._at_least("<2", "1.5"))
def test_unknown_version_is_none(self):
from lib.core.settings import UNKNOWN_DBMS_VERSION
kb.dbmsVersion = UNKNOWN_DBMS_VERSION
self.assertIsNone(isDBMSVersionAtLeast("5"))
class TestGetPageWordSet(unittest.TestCase):
def test_word_extraction(self):
words = getPageWordSet(u"foobartest")
self.assertEqual(sorted(words), [u"foobar", u"test"])
def test_non_string_returns_empty(self):
self.assertEqual(getPageWordSet(None), set())
class TestNormalizeUnicode(unittest.TestCase):
def test_accents_stripped(self):
# normalizeUnicode collapses accented chars to their ASCII base
self.assertEqual(normalizeUnicode(u"\xe9\xe8"), "ee")
def test_plain_ascii_unchanged(self):
self.assertEqual(normalizeUnicode(u"abc123"), "abc123")
def test_none_returns_none(self):
self.assertIsNone(normalizeUnicode(None))
class TestResetCookieJar(unittest.TestCase):
"""resetCookieJar's clear branch (conf.loadCookies falsy)."""
def setUp(self):
self._loadCookies = conf.loadCookies
conf.loadCookies = None
def tearDown(self):
conf.loadCookies = self._loadCookies
def test_clear_branch(self):
try:
from http.cookiejar import CookieJar
except ImportError: # Python 2
from cookielib import CookieJar
jar = CookieJar()
cleared = {"called": False}
class _Jar(object):
def clear(self):
cleared["called"] = True
resetCookieJar(_Jar())
self.assertTrue(cleared["called"])
# also accepts a real jar without raising
self.assertIsNone(resetCookieJar(jar))
# =========================================================================== #
# from tests/test_core_extra.py (common.py classes)
# =========================================================================== #
class TestCommonStringHelpers(unittest.TestCase):
"""Small pure string/list/regex/encoding helpers in lib/core/common.py."""
def test_posix_to_nt_slashes(self):
from lib.core.common import posixToNtSlashes
self.assertEqual(posixToNtSlashes("C:/Windows"), "C:\\Windows")
self.assertEqual(posixToNtSlashes("a/b/c"), "a\\b\\c")
# falsy input returned unchanged
self.assertEqual(posixToNtSlashes(""), "")
self.assertIsNone(posixToNtSlashes(None))
def test_nt_to_posix_slashes(self):
from lib.core.common import ntToPosixSlashes
self.assertEqual(ntToPosixSlashes("C:\\Windows"), "C:/Windows")
self.assertEqual(ntToPosixSlashes("a\\b\\c"), "a/b/c")
self.assertEqual(ntToPosixSlashes(""), "")
def test_is_hex_encoded_string(self):
from lib.core.common import isHexEncodedString
self.assertTrue(isHexEncodedString("DEADBEEF"))
self.assertTrue(isHexEncodedString("0x1234")) # 'x' is allowed by the regex
self.assertFalse(isHexEncodedString("test"))
self.assertFalse(isHexEncodedString("12 34")) # space breaks it
def test_is_digit(self):
from lib.core.common import isDigit
self.assertTrue(isDigit("123456"))
self.assertFalse(isDigit("3b3"))
self.assertFalse(isDigit(u"\xb2")) # superscript-2: str.isdigit() True, isDigit False
self.assertFalse(isDigit("")) # empty -> no match
self.assertFalse(isDigit(None))
def test_sanitize_str(self):
from lib.core.common import sanitizeStr
self.assertEqual(sanitizeStr("foo\n\rbar"), "foo bar")
self.assertEqual(sanitizeStr("a\r\nb"), "a b")
self.assertEqual(sanitizeStr(None), "None")
def test_filter_control_chars(self):
from lib.core.common import filterControlChars
self.assertEqual(filterControlChars("AND 1>(2+3)\n--"), "AND 1>(2+3) --")
# custom replacement character
self.assertEqual(filterControlChars("a\tb", replacement="_"), "a_b")
def test_normalize_path(self):
from lib.core.common import normalizePath
self.assertEqual(normalizePath("//var///log/apache.log"), "/var/log/apache.log")
self.assertEqual(normalizePath("/a/b/../c"), "/a/c")
def test_directory_path(self):
from lib.core.common import directoryPath
self.assertEqual(directoryPath("/var/log/apache.log"), "/var/log")
# no extension -> returned unchanged
self.assertEqual(directoryPath("/var/log"), "/var/log")
def test_longest_common_prefix(self):
from lib.core.common import longestCommonPrefix
self.assertEqual(longestCommonPrefix("foobar", "fobar"), "fo")
self.assertEqual(longestCommonPrefix("abc", "abd", "abe"), "ab")
# single sequence returned verbatim
self.assertEqual(longestCommonPrefix("only"), "only")
def test_first_not_none(self):
from lib.core.common import firstNotNone
self.assertEqual(firstNotNone(None, None, 1, 2, 3), 1)
self.assertEqual(firstNotNone(None, 0), 0) # 0 is not None
self.assertIsNone(firstNotNone(None, None))
def test_decode_string_escape(self):
from lib.core.common import decodeStringEscape
self.assertEqual(decodeStringEscape("a\\tb"), "a\tb")
self.assertEqual(decodeStringEscape("a\\nb"), "a\nb")
# no backslash -> unchanged
self.assertEqual(decodeStringEscape("plain"), "plain")
def test_encode_string_escape(self):
from lib.core.common import encodeStringEscape
self.assertEqual(encodeStringEscape("a\tb"), "a\\tb")
self.assertEqual(encodeStringEscape("a\nb"), "a\\nb")
self.assertEqual(encodeStringEscape("plain"), "plain")
def test_decode_encode_string_escape_roundtrip(self):
from lib.core.common import decodeStringEscape, encodeStringEscape
self.assertEqual(decodeStringEscape(encodeStringEscape("x\ty\nz")), "x\ty\nz")
def test_escape_json_value(self):
from lib.core.common import escapeJsonValue
# newline gets escaped (literal '\n' becomes the two chars backslash+n)
self.assertNotIn("\n", escapeJsonValue("foo\nbar"))
self.assertIn("\\n", escapeJsonValue("foo\nbar"))
# tab gets escaped to '\t'
self.assertIn("\\t", escapeJsonValue("foo\tbar"))
# quote and backslash escaped
self.assertEqual(escapeJsonValue('a"b'), 'a\\"b')
self.assertEqual(escapeJsonValue("a\\b"), "a\\\\b")
# ordinary characters untouched
self.assertEqual(escapeJsonValue("plain text"), "plain text")
def test_clean_query(self):
from lib.core.common import cleanQuery
self.assertEqual(cleanQuery("select id from users"), "SELECT id FROM users")
# already-uppercase keywords stay; identifiers untouched
self.assertEqual(cleanQuery("SELECT a FROM t"), "SELECT a FROM t")
def test_json_minimize_canonical(self):
from lib.core.common import jsonMinimize
# key order / whitespace independence
self.assertEqual(jsonMinimize('{"b": 2, "a": 1}'), jsonMinimize('{"a":1, "b":2}'))
# nested leaf path
self.assertEqual(jsonMinimize('{"a": {"b": 1}}'), ".a.b=1")
# empty object
self.assertEqual(jsonMinimize("{}"), "")
# not parseable -> None (and only None)
self.assertIsNone(jsonMinimize("not json"))
def test_json_minimize_array_length_registers(self):
from lib.core.common import jsonMinimize
# array length change must perturb the projection
self.assertNotEqual(jsonMinimize('{"a": [1, 2]}'), jsonMinimize('{"a": [1, 2, 3]}'))
def test_list_to_str_value(self):
from lib.core.common import listToStrValue
self.assertEqual(listToStrValue([1, 2, 3]), "1, 2, 3")
# set/tuple/generator normalized via list first
self.assertEqual(listToStrValue((1, 2)), "1, 2")
# non-list passes through
self.assertEqual(listToStrValue("abc"), "abc")
def test_intersect(self):
from lib.core.common import intersect
self.assertEqual(intersect([1, 2, 3], set([1, 3])), [1, 3])
# order follows containerA
self.assertEqual(intersect([3, 2, 1], [1, 2]), [2, 1])
# case-insensitive option
self.assertEqual(intersect(["FOO", "bar"], ["foo"], lowerCase=True), ["foo"])
def test_priority_sort_columns(self):
from lib.core.common import prioritySortColumns
# 'id'-containing columns first, then by ascending length
self.assertEqual(
prioritySortColumns(["password", "userid", "name", "id"]),
["id", "userid", "name", "password"],
)
def test_safe_variable_naming(self):
from lib.core.common import safeVariableNaming
self.assertEqual(safeVariableNaming("class.id"), "EVAL_636c6173732e6964")
# plain identifier left untouched
self.assertEqual(safeVariableNaming("foobar"), "foobar")
def test_unsafe_variable_naming(self):
from lib.core.common import unsafeVariableNaming
self.assertEqual(unsafeVariableNaming("EVAL_636c6173732e6964"), "class.id")
self.assertEqual(unsafeVariableNaming("foobar"), "foobar")
def test_variable_naming_roundtrip(self):
from lib.core.common import safeVariableNaming, unsafeVariableNaming
self.assertEqual(unsafeVariableNaming(safeVariableNaming("a-b")), "a-b")
def test_average(self):
from lib.core.common import average
self.assertAlmostEqual(average([0.9, 0.9, 0.9, 1.0, 0.8, 0.9]), 0.9, places=6)
self.assertEqual(average([2, 4]), 3.0)
self.assertIsNone(average([]))
def test_stdev(self):
from lib.core.common import stdev
self.assertEqual("%.3f" % stdev([0.9, 0.9, 0.9, 1.0, 0.8, 0.9]), "0.063")
# fewer than 2 values -> None
self.assertIsNone(stdev([1.0]))
self.assertIsNone(stdev([]))
class TestCommonSafeCompare(unittest.TestCase):
"""Constant-time / checksum helpers."""
def test_safe_compare_strings(self):
from lib.core.common import safeCompareStrings
self.assertTrue(safeCompareStrings("test", "test"))
self.assertFalse(safeCompareStrings("test1", "test2"))
self.assertFalse(safeCompareStrings("test", None))
# both None compares equal (a == b path)
self.assertTrue(safeCompareStrings(None, None))
def test_safe_cs_value(self):
from lib.core.common import safeCSValue
# ensure deterministic delimiter
old = conf.get("csvDel")
conf.csvDel = defaults.csvDel
try:
self.assertEqual(safeCSValue("foo, bar"), '"foo, bar"')
self.assertEqual(safeCSValue("foobar"), "foobar")
self.assertEqual(safeCSValue("foo\rbar"), '"foo\rbar"')
self.assertEqual(safeCSValue('foo"bar'), '"foo""bar"')
finally:
conf.csvDel = old
class TestCommonSafeExString(unittest.TestCase):
def test_sqlmap_exception_message(self):
from lib.core.common import getSafeExString
from lib.core.exception import SqlmapBaseException
self.assertEqual(getSafeExString(SqlmapBaseException("foobar")), "foobar")
def test_oserror_prefixed_with_type(self):
from lib.core.common import getSafeExString
self.assertEqual(getSafeExString(OSError(0, "foobar")), "OSError: foobar")
def test_generic_value_error(self):
from lib.core.common import getSafeExString
self.assertEqual(getSafeExString(ValueError("bad input")), "ValueError: bad input")
class TestCommonHostHeader(unittest.TestCase):
def test_plain_host(self):
from lib.core.common import getHostHeader
self.assertEqual(getHostHeader("http://www.target.com/vuln.php?id=1"), "www.target.com")
def test_default_port_stripped(self):
from lib.core.common import getHostHeader
self.assertEqual(getHostHeader("http://www.target.com:80/x"), "www.target.com")
self.assertEqual(getHostHeader("https://www.target.com:443/x"), "www.target.com")
def test_nondefault_port_kept(self):
from lib.core.common import getHostHeader
self.assertEqual(getHostHeader("http://www.target.com:8080/x"), "www.target.com:8080")
def test_ipv6_brackets(self):
from lib.core.common import getHostHeader
self.assertEqual(getHostHeader("http://[::1]:8080/vuln.php?id=1"), "[::1]:8080")
self.assertEqual(getHostHeader("http://[::1]/vuln.php?id=1"), "[::1]")
class TestCommonCheckSameHost(unittest.TestCase):
def test_same_host(self):
from lib.core.common import checkSameHost
self.assertTrue(checkSameHost(
"http://www.target.com/page1.php?id=1",
"http://www.target.com/images/page2.php",
))
def test_different_host(self):
from lib.core.common import checkSameHost
self.assertFalse(checkSameHost(
"http://www.target.com/page1.php?id=1",
"http://www.target2.com/images/page2.php",
))
def test_www_prefix_ignored(self):
from lib.core.common import checkSameHost
# leading 'www.' is stripped before comparison
self.assertTrue(checkSameHost("http://www.target.com/a", "http://target.com/b"))
def test_single_url_true_and_empty_none(self):
from lib.core.common import checkSameHost
self.assertTrue(checkSameHost("http://only.com/a"))
self.assertIsNone(checkSameHost())
class TestCommonUrldecode(unittest.TestCase):
def test_convall_true(self):
from lib.core.common import urldecode
self.assertEqual(urldecode("AND%201%3E%282%2B3%29%23", convall=True), "AND 1>(2+3)#")
def test_convall_false_keeps_unsafe(self):
from lib.core.common import urldecode
# %2B (plus) is in the default 'unsafe' set so it stays encoded when convall=False
self.assertEqual(urldecode("AND%201%3E%282%2B3%29%23", convall=False), "AND 1>(2%2B3)#")
def test_bytes_input(self):
from lib.core.common import urldecode
self.assertEqual(urldecode(b"AND%201%3E%282%2B3%29%23", convall=False), "AND 1>(2%2B3)#")
def test_spaceplus(self):
from lib.core.common import urldecode
# with spaceplus the '+' becomes a space
self.assertEqual(urldecode("a+b", convall=False, spaceplus=True), "a b")
# without spaceplus the '+' stays
self.assertEqual(urldecode("a+b", convall=False, spaceplus=False), "a+b")
class TestCommonChunkSplit(unittest.TestCase):
def test_chunk_split_post_data(self):
import random
from lib.core.common import chunkSplitPostData
from lib.core.patch import unisonRandom
# The pinned docstring value is produced under sqlmap's cross-version PRNG; install it
# (then restore the stdlib functions) so the expectation is deterministic here too.
_saved = (random.choice, random.randint, random.sample, random.seed)
unisonRandom()
try:
random.seed(0)
expected = ('5;4Xe90\r\nSELEC\r\n3;irWlc\r\nT u\r\n1;eT4zO\r\ns\r\n'
'5;YB4hM\r\nernam\r\n9;2pUD8\r\ne,passwor\r\n3;mp07y\r\nd F\r\n'
'5;8RKXi\r\nROM u\r\n4;MvMhO\r\nsers\r\n0\r\n\r\n')
self.assertEqual(chunkSplitPostData("SELECT username,password FROM users"), expected)
finally:
random.choice, random.randint, random.sample, random.seed = _saved
def test_chunk_split_terminator(self):
from lib.core.common import chunkSplitPostData
# regardless of content, the chunked stream must end with the zero-length terminator
# (assertion is seed-independent, so don't touch the global RNG)
self.assertTrue(chunkSplitPostData("abc").endswith("0\r\n\r\n"))
class TestCommonDecodeIntToUnicode(unittest.TestCase):
def tearDown(self):
set_dbms(None)
def test_basic_ascii(self):
from lib.core.common import decodeIntToUnicode
self.assertEqual(decodeIntToUnicode(35), "#")
self.assertEqual(decodeIntToUnicode(64), "@")
self.assertEqual(decodeIntToUnicode(65), "A")
def test_non_int_passthrough(self):
from lib.core.common import decodeIntToUnicode
# non-int is returned unchanged
self.assertEqual(decodeIntToUnicode("x"), "x")
def test_pgsql_high_codepoint(self):
from lib.core.common import decodeIntToUnicode
set_dbms(DBMS.PGSQL)
# value > 255 on PGSQL takes the _unichr(value) branch
self.assertEqual(decodeIntToUnicode(0x2122), u"\u2122")
class TestCommonDecodeDbmsHex(unittest.TestCase):
def setUp(self):
self._old_binary = kb.binaryField
kb.binaryField = False
def tearDown(self):
kb.binaryField = self._old_binary
set_dbms(None)
def test_plain_hex(self):
from lib.core.common import decodeDbmsHexValue
self.assertEqual(decodeDbmsHexValue("3132332031"), u"123 1")
def test_odd_length_appends_question_mark(self):
from lib.core.common import decodeDbmsHexValue
self.assertEqual(decodeDbmsHexValue("313233203"), u"123 ?")
def test_list_input(self):
from lib.core.common import decodeDbmsHexValue
self.assertEqual(decodeDbmsHexValue(["0x31", "0x32"]), [u"1", u"2"])
def test_non_hex_passthrough(self):
from lib.core.common import decodeDbmsHexValue
self.assertEqual(decodeDbmsHexValue("5.1.41"), u"5.1.41")
class TestCommonUnsafeSQLIdentificator(unittest.TestCase):
def tearDown(self):
set_dbms(None)
def test_mssql_brackets(self):
from lib.core.common import unsafeSQLIdentificatorNaming
from lib.core.common import getText
set_dbms(DBMS.MSSQL)
self.assertEqual(getText(unsafeSQLIdentificatorNaming("[begin]")), "begin")
self.assertEqual(getText(unsafeSQLIdentificatorNaming("foobar")), "foobar")
def test_mysql_backticks(self):
from lib.core.common import unsafeSQLIdentificatorNaming, getText
set_dbms(DBMS.MYSQL)
self.assertEqual(getText(unsafeSQLIdentificatorNaming("`col`")), "col")
def test_oracle_uppercases(self):
from lib.core.common import unsafeSQLIdentificatorNaming, getText
set_dbms(DBMS.ORACLE)
# Oracle strips double quotes and uppercases
self.assertEqual(getText(unsafeSQLIdentificatorNaming('"name"')), "NAME")
class TestCommonParseSqliteSchema(unittest.TestCase):
def setUp(self):
self._old_cached = kb.data.get("cachedColumns")
self._old_db = conf.db
self._old_tbl = conf.tbl
kb.data.cachedColumns = {}
conf.db = "SQLITE_MASTER"
conf.tbl = "users"
def tearDown(self):
kb.data.cachedColumns = self._old_cached
conf.db = self._old_db
conf.tbl = self._old_tbl
def test_simple_schema(self):
from lib.core.common import parseSqliteTableSchema
self.assertTrue(parseSqliteTableSchema(
"CREATE TABLE users(\n\t\tid INTEGER,\n\t\tname TEXT\n);"))
cols = kb.data.cachedColumns[conf.db][conf.tbl]
self.assertEqual(tuple(cols.items()), (("id", "INTEGER"), ("name", "TEXT")))
def test_constraints_skipped(self):
from lib.core.common import parseSqliteTableSchema
self.assertTrue(parseSqliteTableSchema(
"CREATE TABLE suppliers(\n\tsupplier_id INTEGER PRIMARY KEY DESC,\n\tname TEXT NOT NULL\n);"))
cols = kb.data.cachedColumns[conf.db][conf.tbl]
self.assertEqual(tuple(cols.items()), (("supplier_id", "INTEGER"), ("name", "TEXT")))
# =========================================================================== #
# from tests/test_core_final.py (common.py classes)
# =========================================================================== #
class TestCommonPureHelpers(unittest.TestCase):
"""Pure string/encoding/list/regex helpers from lib/core/common.py."""
def test_boldify_message_marks_known_pattern(self):
self.assertEqual(
boldifyMessage("GET parameter id is not injectable", istty=True),
"\x1b[1mGET parameter id is not injectable\x1b[0m",
)
def test_boldify_message_leaves_plain_unchanged(self):
self.assertEqual(boldifyMessage("just a plain message", istty=True), "just a plain message")
def test_calculate_delta_seconds_from_epoch(self):
self.assertGreater(calculateDeltaSeconds(0), 1151721660)
def test_calculate_delta_seconds_nonnegative(self):
import time as _time
self.assertGreaterEqual(calculateDeltaSeconds(_time.time()), 0.0)
def test_common_finder_only_returns_longest_common_prefix(self):
self.assertEqual(commonFinderOnly("abcd", ["abcdefg", "foobar", "abcde"]), "abcde")
def test_enum_value_to_name_lookup_hit(self):
self.assertEqual(enumValueToNameLookup(SORT_ORDER, SORT_ORDER.LAST), "LAST")
def test_enum_value_to_name_lookup_miss(self):
self.assertIsNone(enumValueToNameLookup(SORT_ORDER, -987654321))
def test_file_path_to_safe_string(self):
self.assertEqual(filePathToSafeString("C:/Windows/system32"), "C__Windows_system32")
def test_file_path_to_safe_string_spaces_backslashes(self):
self.assertEqual(filePathToSafeString("a b\\c:d"), "a_b_c_d")
def test_is_windows_drive_letter_path_true(self):
self.assertTrue(isWindowsDriveLetterPath("C:\\boot.ini"))
def test_is_windows_drive_letter_path_false(self):
self.assertFalse(isWindowsDriveLetterPath("/var/log/apache.log"))
def test_clean_replace_unicode_list(self):
self.assertEqual(cleanReplaceUnicode(["a", "b"]), ["a", "b"])
def test_clean_replace_unicode_scalar(self):
self.assertEqual(cleanReplaceUnicode(u"plain"), u"plain")
def test_trim_alpha_num(self):
self.assertEqual(trimAlphaNum("AND 1>(2+3)-- foobar"), " 1>(2+3)-- ")
def test_trim_alpha_num_all_alnum(self):
self.assertEqual(trimAlphaNum("abc123"), "")
def test_trim_alpha_num_empty(self):
self.assertEqual(trimAlphaNum(""), "")
def test_list_to_str_value_list(self):
self.assertEqual(listToStrValue([1, 2, 3]), "1, 2, 3")
def test_list_to_str_value_tuple(self):
self.assertEqual(listToStrValue((4, 5)), "4, 5")
def test_list_to_str_value_scalar(self):
self.assertEqual(listToStrValue("foo"), "foo")
def test_intersect_lists(self):
self.assertEqual(intersect([1, 2, 3], set([1, 3])), [1, 3])
def test_intersect_lowercase(self):
self.assertEqual(intersect(["A", "B"], ["a"], lowerCase=True), ["a"])
def test_intersect_empty(self):
self.assertEqual(intersect([], [1, 2]), [])
def test_apply_function_recursively(self):
self.assertEqual(
applyFunctionRecursively([1, 2, [3, -9]], lambda _: _ > 0),
[True, True, [True, False]],
)
def test_apply_function_recursively_scalar(self):
self.assertEqual(applyFunctionRecursively(5, lambda _: _ + 1), 6)
class TestCommonRegexAndPage(unittest.TestCase):
"""Regex / page-content extraction helpers."""
def test_extract_regex_result_hit(self):
self.assertEqual(extractRegexResult(r"a(?P[^g]+)g", "abcdefg"), "bcdef")
def test_extract_regex_result_no_match(self):
self.assertIsNone(extractRegexResult(r"a(?P[^g]+)g", "xyz"))
def test_extract_regex_result_no_result_group(self):
self.assertIsNone(extractRegexResult(r"plain", "plain"))
def test_extract_regex_result_empty_content(self):
self.assertIsNone(extractRegexResult(r"a(?P.)b", ""))
def test_extract_text_tag_content(self):
self.assertEqual(
extractTextTagContent("Titlefoobar
"),
["Title", "foobar"],
)
def test_extract_text_tag_content_empty(self):
self.assertEqual(extractTextTagContent(""), [])
def test_get_filtered_page_content(self):
self.assertEqual(
getFilteredPageContent(u"foobartest"),
"foobar test",
)
def test_get_filtered_page_content_drops_script(self):
page = u"hello"
self.assertNotIn("var x", getFilteredPageContent(page))
self.assertIn("hello", getFilteredPageContent(page))
def test_get_filtered_page_content_nonstring_passthrough(self):
self.assertEqual(getFilteredPageContent(None), None)
def test_extract_error_message_oracle(self):
page = (u"Test\nWarning: oci_parse() "
u"[function.oci-parse]: ORA-01756: quoted string not properly "
u"terminated
Only a test page
")
self.assertEqual(
getText(extractErrorMessage(page)),
"oci_parse() [function.oci-parse]: ORA-01756: quoted string not properly terminated",
)
def test_extract_error_message_none_for_plain(self):
self.assertIsNone(extractErrorMessage("Warning: This is only a dummy foobar test"))
def test_extract_error_message_prose_like_dbms_signature(self):
# a specific DBMS signature must be extracted even when it reads like plain text (few
# non-writing chars) - the non-writing-char ratio guards only the generic keyword regexes
page = "Microsoft OLE DB Provider for SQL Server error '80040e14' Unclosed quotation mark after the character string ''."
self.assertEqual(extractErrorMessage(page), "Unclosed quotation mark after the character string ''.")
def test_extract_error_message_generic_prose_still_rejected(self):
# the generic '(fatal|error|warning): ...' path must still drop natural-language prose
self.assertIsNone(extractErrorMessage("Error: everything is working fine and nothing is wrong here"))
def test_extract_error_message_non_string(self):
self.assertIsNone(extractErrorMessage(None))
def test_find_multipart_post_boundary(self):
post = ("-----------------------------9051914041544843365972754266\n"
"Content-Disposition: form-data; name=text\n\ndefault")
self.assertEqual(findMultipartPostBoundary(post), "9051914041544843365972754266")
def test_find_multipart_post_boundary_none(self):
self.assertIsNone(findMultipartPostBoundary(""))
class TestCommonHeadersAndExpected(unittest.TestCase):
def test_get_header_case_insensitive(self):
self.assertEqual(getHeader({"Foo": "bar"}, "foo"), "bar")
def test_get_header_missing(self):
self.assertIsNone(getHeader({"Foo": "bar"}, "x"))
def test_get_header_empty_dict(self):
self.assertIsNone(getHeader({}, "anything"))
def test_get_request_header_hit(self):
self.assertEqual(getText(getRequestHeader(_FakeRequest({"FOO": "BAR"}), "foo")), "BAR")
def test_get_request_header_miss(self):
self.assertIsNone(getRequestHeader(_FakeRequest({"FOO": "BAR"}), "missing"))
def test_extract_expected_value_bool_true(self):
self.assertIs(extractExpectedValue(["1"], EXPECTED.BOOL), True)
def test_extract_expected_value_bool_false(self):
self.assertIs(extractExpectedValue(["0"], EXPECTED.BOOL), False)
def test_extract_expected_value_bool_word(self):
self.assertIs(extractExpectedValue(["true"], EXPECTED.BOOL), True)
self.assertIs(extractExpectedValue(["false"], EXPECTED.BOOL), False)
def test_extract_expected_value_int(self):
self.assertEqual(extractExpectedValue("5", EXPECTED.INT), 5)
def test_extract_expected_value_int_invalid(self):
self.assertIsNone(extractExpectedValue(u"7\xb9645", EXPECTED.INT))
def test_extract_expected_value_no_expected(self):
self.assertEqual(extractExpectedValue("foo", None), "foo")
class TestParseJsonAndHash(unittest.TestCase):
def test_parse_json_double_quotes(self):
self.assertEqual(parseJson('{"id":1}')["id"], 1)
def test_parse_json_single_quotes(self):
self.assertEqual(parseJson("{'id':1, 'foo':[2,3,4]}")["id"], 1)
def test_parse_json_not_json(self):
self.assertIsNone(parseJson("this is not json"))
def test_parse_password_hash_mssql(self):
saved = kb.forcedDbms
try:
kb.forcedDbms = DBMS.MSSQL
result = parsePasswordHash("0x01004086ceb60c90646a8ab9889fe3ed8e5c150b5460ece8425a")
self.assertIn("salt: 4086ceb6", result)
self.assertIn("header: 0x0100", result)
finally:
kb.forcedDbms = saved
def test_parse_password_hash_none(self):
self.assertEqual(parsePasswordHash(None), NULL)
def test_parse_password_hash_blank(self):
self.assertEqual(parsePasswordHash(" "), NULL)
class TestSerializeAndTechnique(unittest.TestCase):
def test_serialize_roundtrip(self):
self.assertEqual(unserializeObject(serializeObject([1, 2, 3])), [1, 2, 3])
def test_serialize_object_is_str(self):
self.assertIsInstance(serializeObject([1, 2, ("a", "b")]), str)
def test_unserialize_none(self):
self.assertIsNone(unserializeObject(None))
def test_set_get_technique_thread_local(self):
saved = getTechnique()
try:
setTechnique(5)
self.assertEqual(getTechnique(), 5)
finally:
setTechnique(saved)
def test_get_technique_falls_back_to_kb(self):
saved_thread = getTechnique()
saved_kb = kb.get("technique")
try:
setTechnique(None)
kb.technique = 7
self.assertEqual(getTechnique(), 7)
finally:
setTechnique(saved_thread)
kb.technique = saved_kb
class TestRemovePostHint(unittest.TestCase):
def test_removes_known_prefix(self):
self.assertEqual(removePostHintPrefix("JSON id"), "id")
def test_no_prefix_unchanged(self):
self.assertEqual(removePostHintPrefix("id"), "id")
class TestFileHelpers(unittest.TestCase):
def test_check_file_existing(self):
self.assertTrue(checkFile(__file__))
def test_check_file_missing_no_raise(self):
self.assertFalse(checkFile("/no/such/path_xyz_123", raiseOnError=False))
def test_check_file_missing_raises(self):
with self.assertRaises(SqlmapSystemException):
checkFile("/no/such/path_xyz_123", raiseOnError=True)
def test_is_zip_file_wordlist(self):
# paths.WORDLIST is a zip-compressed wordlist shipped with sqlmap
self.assertTrue(isZipFile(paths.WORDLIST))
def test_is_zip_file_plain_text(self):
self.assertFalse(isZipFile(paths.SQL_KEYWORDS))
def test_safe_filepath_encode_ascii_passthrough(self):
# On Python 3 the function returns the value unchanged for str input
self.assertEqual(safeFilepathEncode("/tmp/x"), "/tmp/x")
def test_safe_expand_user_basename_preserved(self):
self.assertIn(os.path.basename(__file__), safeExpandUser(__file__))
class TestCheckOldOptions(unittest.TestCase):
def test_no_old_options_is_noop(self):
# Returns None and does not raise when no deprecated options are present
self.assertIsNone(checkOldOptions(["-u", "http://test.invalid/?id=1", "--banner"]))
class TestSetColorHandlerMismatch(unittest.TestCase):
"""
Regression test for issue #6122: setColor() decided whether to colorize purely from
conf.disableColoring/IS_TTY, independent of whether the installed LOGGER_HANDLER actually
supports colorize() (a plain logging.StreamHandler - installed for --disable-coloring, or as
the ansistrm-unavailable fallback - never defines it). A multiprocessing hash-cracking worker
hit exactly this mismatch (conf.disableColoring not carried over into the worker) and crashed
with an AttributeError that got misreported as "there was a problem while hashing entry".
"""
def setUp(self):
import lib.core.common as common_mod
self._common_mod = common_mod
self._saved_handler = common_mod.LOGGER_HANDLER
self._saved_disableColoring = conf.get("disableColoring")
def tearDown(self):
self._common_mod.LOGGER_HANDLER = self._saved_handler
conf.disableColoring = self._saved_disableColoring
def test_plain_handler_without_colorize_does_not_raise(self):
import logging
self._common_mod.LOGGER_HANDLER = logging.StreamHandler() # no .colorize(), like the --disable-coloring handler
conf.disableColoring = False # the desync: coloring "should" apply, but handler can't
result = setColor("[INFO] current status: abcde", istty=True) # must not raise
self.assertIsInstance(result, str)
def test_colorizing_handler_still_used(self):
# sanity check: a handler that DOES define colorize() is unaffected by the guard
calls = []
class _FakeColorizingHandler(object):
def colorize(self, message, levelno, force=False):
calls.append((message, levelno, force))
return "COLORIZED"
self._common_mod.LOGGER_HANDLER = _FakeColorizingHandler()
conf.disableColoring = False
result = setColor("[INFO] current status: abcde", istty=True)
self.assertEqual(result, "COLORIZED")
self.assertEqual(len(calls), 1)
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