Skip to content
Switch branches/tags
Go to file
Cannot retrieve contributors at this time
5487 lines (4295 sloc) 185 KB
#!/usr/bin/env python
Copyright (c) 2006-2021 sqlmap developers (
See the file 'LICENSE' for copying permission
from __future__ import division
import binascii
import codecs
import contextlib
import copy
import distutils.version
import functools
import getpass
import hashlib
import inspect
import io
import json
import keyword
import locale
import logging
import ntpath
import os
import platform
import posixpath
import random
import re
import socket
import string
import subprocess
import sys
import tempfile
import threading
import time
import types
import unicodedata
from difflib import SequenceMatcher
from math import sqrt
from optparse import OptionValueError
from xml.sax import parse
from xml.sax import SAXParseException
from extra.beep.beep import beep
from extra.cloak.cloak import decloak
from lib.core.bigarray import BigArray
from lib.core.compat import cmp
from lib.core.compat import round
from lib.core.compat import xrange
from lib.core.convert import base64pickle
from lib.core.convert import base64unpickle
from lib.core.convert import decodeBase64
from lib.core.convert import decodeHex
from lib.core.convert import getBytes
from lib.core.convert import getText
from lib.core.convert import getUnicode
from lib.core.convert import htmlUnescape
from lib.core.convert import stdoutEncode
from import cmdLineOptions
from import conf
from import kb
from import logger
from import paths
from lib.core.datatype import OrderedSet
from lib.core.decorators import cachedmethod
from lib.core.defaults import defaults
from lib.core.dicts import DBMS_DICT
from lib.core.dicts import DEFAULT_DOC_ROOTS
from lib.core.dicts import DEPRECATED_OPTIONS
from lib.core.dicts import OBSOLETE_OPTIONS
from lib.core.dicts import SQL_STATEMENTS
from lib.core.enums import ADJUST_TIME_DELAY
from lib.core.enums import CHARSET_TYPE
from lib.core.enums import CONTENT_STATUS
from lib.core.enums import DBMS
from lib.core.enums import EXPECTED
from lib.core.enums import HASHDB_KEYS
from lib.core.enums import HEURISTIC_TEST
from lib.core.enums import HTTP_HEADER
from lib.core.enums import HTTPMETHOD
from lib.core.enums import LOGGING_LEVELS
from lib.core.enums import MKSTEMP_PREFIX
from lib.core.enums import OPTION_TYPE
from lib.core.enums import OS
from lib.core.enums import PAYLOAD
from lib.core.enums import PLACE
from lib.core.enums import POST_HINT
from lib.core.enums import REFLECTIVE_COUNTER
from lib.core.enums import SORT_ORDER
from lib.core.exception import SqlmapBaseException
from lib.core.exception import SqlmapDataException
from lib.core.exception import SqlmapGenericException
from lib.core.exception import SqlmapInstallationException
from lib.core.exception import SqlmapMissingDependence
from lib.core.exception import SqlmapNoneDataException
from lib.core.exception import SqlmapSilentQuitException
from lib.core.exception import SqlmapSyntaxException
from lib.core.exception import SqlmapSystemException
from lib.core.exception import SqlmapUserQuitException
from lib.core.exception import SqlmapValueException
from lib.core.log import LOGGER_HANDLER
from lib.core.optiondict import optDict
from lib.core.settings import BANNER
from lib.core.settings import BOLD_PATTERNS
from lib.core.settings import BOUNDED_INJECTION_MARKER
from lib.core.settings import BRUTE_DOC_ROOT_PREFIXES
from lib.core.settings import BRUTE_DOC_ROOT_SUFFIXES
from lib.core.settings import BRUTE_DOC_ROOT_TARGET_MARK
from lib.core.settings import BURP_REQUEST_REGEX
from lib.core.settings import BURP_XML_HISTORY_REGEX
from lib.core.settings import CRAWL_EXCLUDE_EXTENSIONS
from lib.core.settings import CUSTOM_INJECTION_MARK_CHAR
from lib.core.settings import DBMS_DIRECTORY_DICT
from lib.core.settings import DEFAULT_COOKIE_DELIMITER
from lib.core.settings import DEFAULT_GET_POST_DELIMITER
from lib.core.settings import DEFAULT_MSSQL_SCHEMA
from lib.core.settings import DEV_EMAIL_ADDRESS
from lib.core.settings import DOLLAR_MARKER
from lib.core.settings import DUMMY_USER_INJECTION
from lib.core.settings import DYNAMICITY_BOUNDARY_LENGTH
from lib.core.settings import ERROR_PARSING_REGEXES
from lib.core.settings import EVALCODE_ENCODED_PREFIX
from lib.core.settings import FILE_PATH_REGEXES
from lib.core.settings import FORCE_COOKIE_EXPIRATION_TIME
from lib.core.settings import FORM_SEARCH_REGEX
from lib.core.settings import GENERIC_DOC_ROOT_DIRECTORY_NAMES
from lib.core.settings import GIT_PAGE
from lib.core.settings import GITHUB_REPORT_OAUTH_TOKEN
from lib.core.settings import GOOGLE_ANALYTICS_COOKIE_PREFIX
from lib.core.settings import HASHDB_MILESTONE_VALUE
from lib.core.settings import HOST_ALIASES
from lib.core.settings import HTTP_CHUNKED_SPLIT_KEYWORDS
from lib.core.settings import IGNORE_PARAMETERS
from lib.core.settings import IGNORE_SAVE_OPTIONS
from lib.core.settings import INFERENCE_UNKNOWN_CHAR
from lib.core.settings import IP_ADDRESS_REGEX
from lib.core.settings import ISSUES_PAGE
from lib.core.settings import IS_TTY
from lib.core.settings import IS_WIN
from lib.core.settings import LARGE_OUTPUT_THRESHOLD
from lib.core.settings import LOCALHOST
from lib.core.settings import MAX_INT
from lib.core.settings import MIN_ENCODED_LEN_CHECK
from lib.core.settings import MIN_ERROR_PARSING_NON_WRITING_RATIO
from lib.core.settings import MIN_TIME_RESPONSES
from lib.core.settings import MIN_VALID_DELAYED_RESPONSE
from lib.core.settings import NETSCAPE_FORMAT_HEADER_COOKIES
from lib.core.settings import NULL
from lib.core.settings import PARAMETER_AMP_MARKER
from lib.core.settings import PARAMETER_SEMICOLON_MARKER
from lib.core.settings import PARAMETER_PERCENTAGE_MARKER
from lib.core.settings import PARTIAL_HEX_VALUE_MARKER
from lib.core.settings import PARTIAL_VALUE_MARKER
from lib.core.settings import PAYLOAD_DELIMITER
from lib.core.settings import PLATFORM
from lib.core.settings import PRINTABLE_CHAR_REGEX
from lib.core.settings import PUSH_VALUE_EXCEPTION_RETRY_COUNT
from lib.core.settings import PYVERSION
from lib.core.settings import RANDOMIZATION_TLDS
from lib.core.settings import REFERER_ALIASES
from lib.core.settings import REFLECTED_BORDER_REGEX
from lib.core.settings import REFLECTED_MAX_REGEX_PARTS
from lib.core.settings import REFLECTED_REPLACEMENT_REGEX
from lib.core.settings import REFLECTED_REPLACEMENT_TIMEOUT
from lib.core.settings import REFLECTED_VALUE_MARKER
from lib.core.settings import REFLECTIVE_MISS_THRESHOLD
from lib.core.settings import SENSITIVE_DATA_REGEX
from lib.core.settings import SENSITIVE_OPTIONS
from lib.core.settings import STDIN_PIPE_DASH
from lib.core.settings import SUPPORTED_DBMS
from lib.core.settings import TEXT_TAG_REGEX
from lib.core.settings import TIME_STDEV_COEFF
from lib.core.settings import UNICODE_ENCODING
from lib.core.settings import UNKNOWN_DBMS_VERSION
from lib.core.settings import URI_QUESTION_MARKER
from lib.core.settings import URLENCODE_CHAR_LIMIT
from lib.core.settings import URLENCODE_FAILSAFE_CHARS
from lib.core.settings import USER_AGENT_ALIASES
from lib.core.settings import VERSION_COMPARISON_CORRECTION
from lib.core.settings import VERSION_STRING
from lib.core.settings import ZIP_HEADER
from lib.core.settings import WEBSCARAB_SPLITTER
from lib.core.threads import getCurrentThreadData
from lib.utils.safe2bin import safecharencode
from lib.utils.sqlalchemy import _sqlalchemy
from thirdparty import six
from thirdparty.clientform.clientform import ParseResponse
from thirdparty.clientform.clientform import ParseError
from thirdparty.colorama.initialise import init as coloramainit
from thirdparty.magic import magic
from thirdparty.odict import OrderedDict
from thirdparty.six import unichr as _unichr
from thirdparty.six.moves import collections_abc as _collections
from thirdparty.six.moves import configparser as _configparser
from thirdparty.six.moves import http_client as _http_client
from thirdparty.six.moves import input as _input
from thirdparty.six.moves import reload_module as _reload_module
from thirdparty.six.moves import urllib as _urllib
from thirdparty.six.moves import zip as _zip
from thirdparty.termcolor.termcolor import colored
class UnicodeRawConfigParser(_configparser.RawConfigParser):
RawConfigParser with unicode writing support
def write(self, fp):
Write an .ini-format representation of the configuration state.
if self._defaults:
fp.write("[%s]\n" % _configparser.DEFAULTSECT)
for (key, value) in self._defaults.items():
fp.write("%s = %s" % (key, getUnicode(value, UNICODE_ENCODING)))
for section in self._sections:
fp.write("[%s]\n" % section)
for (key, value) in self._sections[section].items():
if key != "__name__":
if value is None:
fp.write("%s\n" % (key))
elif not isListLike(value):
fp.write("%s = %s\n" % (key, getUnicode(value, UNICODE_ENCODING)))
class Format(object):
def humanize(values, chain=" or "):
return chain.join(values)
# Get methods
def getDbms(versions=None):
Format the back-end DBMS fingerprint value and return its
values formatted as a human readable string.
@return: detected back-end DBMS based upon fingerprint techniques.
@rtype: C{str}
if versions is None and Backend.getVersionList():
versions = Backend.getVersionList()
return Backend.getDbms() if versions is None else "%s %s" % (Backend.getDbms(), " and ".join(filterNone(versions)))
def getErrorParsedDBMSes():
Parses the knowledge base htmlFp list and return its values
formatted as a human readable string.
@return: list of possible back-end DBMS based upon error messages
@rtype: C{str}
htmlParsed = None
if len(kb.htmlFp) == 0 or kb.heuristicTest != HEURISTIC_TEST.POSITIVE:
elif len(kb.htmlFp) == 1:
htmlParsed = kb.htmlFp[0]
elif len(kb.htmlFp) > 1:
htmlParsed = " or ".join(kb.htmlFp)
return htmlParsed
def getOs(target, info):
Formats the back-end operating system fingerprint value
and return its values formatted as a human readable string.
Example of info (kb.headersFp) dictionary:
'distrib': set(['Ubuntu']),
'type': set(['Linux']),
'technology': set(['PHP 5.2.6', 'Apache 2.2.9']),
'release': set(['8.10'])
Example of info (kb.bannerFp) dictionary:
'sp': set(['Service Pack 4']),
'dbmsVersion': '8.00.194',
'dbmsServicePack': '0',
'distrib': set(['2000']),
'dbmsRelease': '2000',
'type': set(['Windows'])
@return: detected back-end operating system based upon fingerprint
@rtype: C{str}
infoStr = ""
infoApi = {}
if info and "type" in info:
if conf.api:
infoApi["%s operating system" % target] = info
infoStr += "%s operating system: %s" % (target, Format.humanize(info["type"]))
if "distrib" in info:
infoStr += " %s" % Format.humanize(info["distrib"])
if "release" in info:
infoStr += " %s" % Format.humanize(info["release"])
if "sp" in info:
infoStr += " %s" % Format.humanize(info["sp"])
if "codename" in info:
infoStr += " (%s)" % Format.humanize(info["codename"])
if "technology" in info:
if conf.api:
infoApi["web application technology"] = Format.humanize(info["technology"], ", ")
infoStr += "\nweb application technology: %s" % Format.humanize(info["technology"], ", ")
if conf.api:
return infoApi
return infoStr.lstrip()
class Backend(object):
def setDbms(dbms):
dbms = aliasToDbmsEnum(dbms)
if dbms is None:
return None
# Little precaution, in theory this condition should always be false
elif kb.dbms is not None and kb.dbms != dbms:
warnMsg = "there appears to be a high probability that "
warnMsg += "this could be a false positive case"
msg = "sqlmap previously fingerprinted back-end DBMS as "
msg += "%s. However now it has been fingerprinted " % kb.dbms
msg += "as %s. " % dbms
msg += "Please, specify which DBMS should be "
msg += "correct [%s (default)/%s] " % (kb.dbms, dbms)
while True:
choice = readInput(msg, default=kb.dbms)
if aliasToDbmsEnum(choice) == kb.dbms:
kb.dbmsVersion = []
kb.resolutionDbms = kb.dbms
elif aliasToDbmsEnum(choice) == dbms:
kb.dbms = aliasToDbmsEnum(choice)
warnMsg = "invalid value"
elif kb.dbms is None:
kb.dbms = aliasToDbmsEnum(dbms)
return kb.dbms
def setVersion(version):
if isinstance(version, six.string_types):
kb.dbmsVersion = [version]
return kb.dbmsVersion
def setVersionList(versionsList):
if isinstance(versionsList, list):
kb.dbmsVersion = versionsList
elif isinstance(versionsList, six.string_types):
logger.error("invalid format of versionsList")
def forceDbms(dbms, sticky=False):
if not kb.stickyDBMS:
kb.forcedDbms = aliasToDbmsEnum(dbms)
kb.stickyDBMS = sticky
def flushForcedDbms(force=False):
if not kb.stickyDBMS or force:
kb.forcedDbms = None
kb.stickyDBMS = False
def setOs(os):
if os is None:
return None
# Little precaution, in theory this condition should always be false
elif kb.os is not None and isinstance(os, six.string_types) and kb.os.lower() != os.lower():
msg = "sqlmap previously fingerprinted back-end DBMS "
msg += "operating system %s. However now it has " % kb.os
msg += "been fingerprinted to be %s. " % os
msg += "Please, specify which OS is "
msg += "correct [%s (default)/%s] " % (kb.os, os)
while True:
choice = readInput(msg, default=kb.os)
if choice == kb.os:
elif choice == os:
kb.os = choice.capitalize()
warnMsg = "invalid value"
elif kb.os is None and isinstance(os, six.string_types):
kb.os = os.capitalize()
return kb.os
def setOsVersion(version):
if version is None:
return None
elif kb.osVersion is None and isinstance(version, six.string_types):
kb.osVersion = version
def setOsServicePack(sp):
if sp is None:
return None
elif kb.osSP is None and isinstance(sp, int):
kb.osSP = sp
def setArch():
msg = "what is the back-end database management system architecture?"
msg += "\n[1] 32-bit (default)"
msg += "\n[2] 64-bit"
while True:
choice = readInput(msg, default='1')
if hasattr(choice, "isdigit") and choice.isdigit() and int(choice) in (1, 2):
kb.arch = 32 if int(choice) == 1 else 64
warnMsg = "invalid value. Valid values are 1 and 2"
return kb.arch
# Get methods
def getForcedDbms():
return aliasToDbmsEnum(conf.get("forceDbms")) or aliasToDbmsEnum(kb.get("forcedDbms"))
def getDbms():
return aliasToDbmsEnum(kb.get("dbms"))
def getErrorParsedDBMSes():
Returns array with parsed DBMS names till now
This functions is called to:
1. Ask user whether or not skip specific DBMS tests in detection phase,
lib/controller/ - detection phase.
2. Sort the fingerprint of the DBMS, lib/controller/ -
fingerprint phase.
return kb.htmlFp if kb.get("heuristicTest") == HEURISTIC_TEST.POSITIVE else []
def getIdentifiedDbms():
This functions is called to:
1. Sort the tests, getSortedInjectionTests() - detection phase.
2. Etc.
dbms = None
if not kb:
elif not kb.get("testMode") and conf.get("dbmsHandler") and getattr(conf.dbmsHandler, "_dbms", None):
dbms = conf.dbmsHandler._dbms
elif Backend.getForcedDbms() is not None:
dbms = Backend.getForcedDbms()
elif Backend.getDbms() is not None:
dbms = Backend.getDbms()
elif kb.get("injection") and kb.injection.dbms:
dbms = unArrayizeValue(kb.injection.dbms)
elif Backend.getErrorParsedDBMSes():
dbms = unArrayizeValue(Backend.getErrorParsedDBMSes())
elif conf.get("dbms"):
dbms = conf.get("dbms")
return aliasToDbmsEnum(dbms)
def getVersion():
versions = filterNone(flattenValue(kb.dbmsVersion)) if not isinstance(kb.dbmsVersion, six.string_types) else [kb.dbmsVersion]
if not isNoneValue(versions):
return versions[0]
return None
def getVersionList():
versions = filterNone(flattenValue(kb.dbmsVersion)) if not isinstance(kb.dbmsVersion, six.string_types) else [kb.dbmsVersion]
if not isNoneValue(versions):
return versions
return None
def getOs():
return kb.os
def getOsVersion():
return kb.osVersion
def getOsServicePack():
return kb.osSP
def getArch():
if kb.arch is None:
return kb.arch
# Comparison methods
def isDbms(dbms):
if not kb.get("testMode") and all((Backend.getDbms(), Backend.getIdentifiedDbms())) and Backend.getDbms() != Backend.getIdentifiedDbms():
singleTimeWarnMessage("identified ('%s') and fingerprinted ('%s') DBMSes differ. If you experience problems in enumeration phase please rerun with '--flush-session'" % (Backend.getIdentifiedDbms(), Backend.getDbms()))
return Backend.getIdentifiedDbms() == aliasToDbmsEnum(dbms)
def isFork(fork):
return hashDBRetrieve(HASHDB_KEYS.DBMS_FORK) == fork
def isDbmsWithin(aliases):
return Backend.getDbms() is not None and Backend.getDbms().lower() in aliases
def isVersion(version):
return Backend.getVersion() is not None and Backend.getVersion() == version
def isVersionWithin(versionList):
if Backend.getVersionList() is None:
return False
for _ in Backend.getVersionList():
if _ != UNKNOWN_DBMS_VERSION and _ in versionList:
return True
return False
def isVersionGreaterOrEqualThan(version):
retVal = False
if Backend.getVersion() is not None and version is not None:
retVal = distutils.version.LooseVersion(Backend.getVersion()) >= distutils.version.LooseVersion(version)
retVal = str(Backend.getVersion()) >= str(version)
return retVal
def isOs(os):
return Backend.getOs() is not None and Backend.getOs().lower() == os.lower()
def paramToDict(place, parameters=None):
Split the parameters into names and values, check if these parameters
are within the testable parameters and return in a dictionary.
testableParameters = OrderedDict()
if place in conf.parameters and not parameters:
parameters = conf.parameters[place]
parameters = re.sub(r"&(\w{1,4});", r"%s\g<1>%s" % (PARAMETER_AMP_MARKER, PARAMETER_SEMICOLON_MARKER), parameters)
if place == PLACE.COOKIE:
splitParams = parameters.split(conf.cookieDel or DEFAULT_COOKIE_DELIMITER)
splitParams = parameters.split(conf.paramDel or DEFAULT_GET_POST_DELIMITER)
for element in splitParams:
element = re.sub(r"%s(.+?)%s" % (PARAMETER_AMP_MARKER, PARAMETER_SEMICOLON_MARKER), r"&\g<1>;", element)
parts = element.split("=")
if len(parts) >= 2:
parameter = urldecode(parts[0].replace(" ", ""))
if not parameter:
if conf.paramDel and conf.paramDel == '\n':
parts[-1] = parts[-1].rstrip()
condition = not conf.testParameter
condition |= conf.testParameter is not None and parameter in conf.testParameter
condition |= place == PLACE.COOKIE and len(intersect((PLACE.COOKIE,), conf.testParameter, True)) > 0
if condition:
value = "=".join(parts[1:])
if parameter in (conf.base64Parameter or []):
kb.base64Originals[parameter] = oldValue = value
value = urldecode(value, convall=True)
value = decodeBase64(value, binary=False, encoding=conf.encoding or UNICODE_ENCODING)
parameters = re.sub(r"\b%s(\b|\Z)" % re.escape(oldValue), value, parameters)
errMsg = "parameter '%s' does not contain " % parameter
errMsg += "valid Base64 encoded value ('%s')" % value
raise SqlmapValueException(errMsg)
testableParameters[parameter] = value
if not conf.multipleTargets and not (conf.csrfToken and, parameter, re.I)):
_ = urldecode(testableParameters[parameter], convall=True)
if (_.endswith("'") and _.count("'") == 1 or'\A9{3,}', _) or'\A-\d+\Z', _) or, _)) and not parameter.upper().startswith(GOOGLE_ANALYTICS_COOKIE_PREFIX):
warnMsg = "it appears that you have provided tainted parameter values "
warnMsg += "('%s') with most likely leftover " % element
warnMsg += "chars/statements from manual SQL injection test(s). "
warnMsg += "Please, always use only valid parameter values "
warnMsg += "so sqlmap could be able to run properly"
message = "are you really sure that you want to continue (sqlmap could have problems)? [y/N] "
if not readInput(message, default='N', boolean=True):
raise SqlmapSilentQuitException
elif not _:
warnMsg = "provided value for parameter '%s' is empty. " % parameter
warnMsg += "Please, always use only valid parameter values "
warnMsg += "so sqlmap could be able to run properly"
if place in (PLACE.POST, PLACE.GET):
for regex in (r"\A((?:<[^>]+>)+\w+)((?:<[^>]+>)+)\Z", r"\A([^\w]+.*\w+)([^\w]+)\Z"):
match =, testableParameters[parameter])
if match:
candidates = OrderedDict()
def walk(head, current=None):
if current is None:
current = head
if isListLike(current):
for _ in current:
walk(head, _)
elif isinstance(current, dict):
for key in current.keys():
value = current[key]
if isinstance(value, (bool, int, float, six.string_types)) or value in (None, []):
original = current[key]
if isinstance(value, bool):
current[key] = "%s%s" % (getUnicode(value).lower(), BOUNDED_INJECTION_MARKER)
elif value is None:
current[key] = "%s%s" % (randomInt(), BOUNDED_INJECTION_MARKER)
elif value == []:
current[key] = ["%s%s" % (randomInt(), BOUNDED_INJECTION_MARKER)]
current[key] = "%s%s" % (value, BOUNDED_INJECTION_MARKER)
candidates["%s (%s)" % (parameter, key)] = re.sub(r"\b(%s\s*=\s*)%s" % (re.escape(parameter), re.escape(testableParameters[parameter])), r"\g<1>%s" % json.dumps(deserialized, separators=(',', ':') if ", " not in testableParameters[parameter] else None), parameters)
current[key] = original
elif isinstance(value, (list, tuple, set, dict)):
if value:
walk(head, value)
deserialized = json.loads(testableParameters[parameter])
if candidates:
message = "it appears that provided value for %sparameter '%s' " % ("%s " % place if place != parameter else "", parameter)
message += "is JSON deserializable. Do you want to inject inside? [y/N] "
if readInput(message, default='N', boolean=True):
del testableParameters[parameter]
except (KeyboardInterrupt, SqlmapUserQuitException):
except Exception:
_ = re.sub(regex, r"\g<1>%s\g<%d>" % (kb.customInjectionMark, len(match.groups())), testableParameters[parameter])
message = "it appears that provided value for %sparameter '%s' " % ("%s " % place if place != parameter else "", parameter)
message += "has boundaries. Do you want to inject inside? ('%s') [y/N] " % getUnicode(_)
if readInput(message, default='N', boolean=True):
testableParameters[parameter] = re.sub(r"\b(%s\s*=\s*)%s" % (re.escape(parameter), re.escape(testableParameters[parameter])), (r"\g<1>%s" % re.sub(regex, r"\g<1>%s\g<2>" % BOUNDED_INJECTION_MARKER, testableParameters[parameter].replace("\\", r"\\"))), parameters)
if conf.testParameter:
if not testableParameters:
paramStr = ", ".join(test for test in conf.testParameter)
if len(conf.testParameter) > 1:
warnMsg = "provided parameters '%s' " % paramStr
warnMsg += "are not inside the %s" % place
parameter = conf.testParameter[0]
if not intersect(USER_AGENT_ALIASES + REFERER_ALIASES + HOST_ALIASES, parameter, True):
debugMsg = "provided parameter '%s' " % paramStr
debugMsg += "is not inside the %s" % place
elif len(conf.testParameter) != len(testableParameters):
for parameter in conf.testParameter:
if parameter not in testableParameters:
debugMsg = "provided parameter '%s' " % parameter
debugMsg += "is not inside the %s" % place
if testableParameters:
for parameter, value in testableParameters.items():
if value and not value.isdigit():
for encoding in ("hex", "base64"):
decoded = codecs.decode(value, encoding)
if len(decoded) > MIN_ENCODED_LEN_CHECK and all(_ in getBytes(string.printable) for _ in decoded):
warnMsg = "provided parameter '%s' " % parameter
warnMsg += "appears to be '%s' encoded" % encoding
return testableParameters
def getManualDirectories():
directories = None
defaultDocRoot = DEFAULT_DOC_ROOTS.get(Backend.getOs(), DEFAULT_DOC_ROOTS[OS.LINUX])
if kb.absFilePaths:
for absFilePath in kb.absFilePaths:
if directories:
if directoryPath(absFilePath) == '/':
absFilePath = normalizePath(absFilePath)
windowsDriveLetter = None
if isWindowsDriveLetterPath(absFilePath):
windowsDriveLetter, absFilePath = absFilePath[:2], absFilePath[2:]
absFilePath = ntToPosixSlashes(posixToNtSlashes(absFilePath))
for _ in list(GENERIC_DOC_ROOT_DIRECTORY_NAMES) + [conf.hostname]:
_ = "/%s/" % _
if _ in absFilePath:
directories = "%s%s" % (absFilePath.split(_)[0], _)
if not directories and conf.path.strip('/') and conf.path in absFilePath:
directories = absFilePath.split(conf.path)[0]
if directories and windowsDriveLetter:
directories = "%s/%s" % (windowsDriveLetter, ntToPosixSlashes(directories))
directories = normalizePath(directories)
if conf.webRoot:
directories = [conf.webRoot]
infoMsg = "using '%s' as web server document root" % conf.webRoot
elif directories:
infoMsg = "retrieved the web server document root: '%s'" % directories
warnMsg = "unable to automatically retrieve the web server "
warnMsg += "document root"
directories = []
message = "what do you want to use for writable directory?\n"
message += "[1] common location(s) ('%s') (default)\n" % ", ".join(root for root in defaultDocRoot)
message += "[2] custom location(s)\n"
message += "[3] custom directory list file\n"
message += "[4] brute force search"
choice = readInput(message, default='1')
if choice == '2':
message = "please provide a comma separate list of absolute directory paths: "
directories = readInput(message, default="").split(',')
elif choice == '3':
message = "what's the list file location?\n"
listPath = readInput(message, default="")
directories = getFileItems(listPath)
elif choice == '4':
targets = set([conf.hostname])
_ = conf.hostname.split('.')
if _[0] == "www":
targets = filterNone(targets)
for prefix in BRUTE_DOC_ROOT_PREFIXES.get(Backend.getOs(), DEFAULT_DOC_ROOTS[OS.LINUX]):
if BRUTE_DOC_ROOT_TARGET_MARK in prefix and re.match(IP_ADDRESS_REGEX, conf.hostname):
for target in targets:
if not prefix.endswith("/%s" % suffix):
item = "%s/%s" % (prefix, suffix)
item = prefix
item = item.replace(BRUTE_DOC_ROOT_TARGET_MARK, target).replace("//", '/').rstrip('/')
if item not in directories:
if BRUTE_DOC_ROOT_TARGET_MARK not in prefix:
infoMsg = "using generated directory list: %s" % ','.join(directories)
msg = "use any additional custom directories [Enter for None]: "
answer = readInput(msg)
if answer:
directories = defaultDocRoot
return directories
def getAutoDirectories():
>>> pushValue(kb.absFilePaths)
>>> kb.absFilePaths = ["C:\\inetpub\\wwwroot\\index.asp", "/var/www/html"]
>>> getAutoDirectories()
['C:/inetpub/wwwroot', '/var/www/html']
>>> kb.absFilePaths = popValue()
retVal = OrderedSet()
if kb.absFilePaths:
infoMsg = "retrieved web server absolute paths: "
infoMsg += "'%s'" % ", ".join(ntToPosixSlashes(path) for path in kb.absFilePaths)
for absFilePath in kb.absFilePaths:
if absFilePath:
directory = directoryPath(absFilePath)
directory = ntToPosixSlashes(directory)
warnMsg = "unable to automatically parse any web server path"
return list(retVal)
def filePathToSafeString(filePath):
Returns string representation of a given filepath safe for a single filename usage
>>> filePathToSafeString('C:/Windows/system32')
retVal = filePath.replace("/", "_").replace("\\", "_")
retVal = retVal.replace(" ", "_").replace(":", "_")
return retVal
def singleTimeDebugMessage(message):
singleTimeLogMessage(message, logging.DEBUG)
def singleTimeWarnMessage(message):
singleTimeLogMessage(message, logging.WARN)
def singleTimeLogMessage(message, level=logging.INFO, flag=None):
if flag is None:
flag = hash(message)
if not conf.smokeTest and flag not in kb.singleLogFlags:
logger.log(level, message)
def boldifyMessage(message, istty=None):
Sets ANSI bold marking on entire message if parts found in predefined BOLD_PATTERNS
>>> boldifyMessage("Hello World", istty=True)
'Hello World'
>>> boldifyMessage("GET parameter id is not injectable", istty=True)
'\\x1b[1mGET parameter id is not injectable\\x1b[0m'
retVal = message
if any(_ in message for _ in BOLD_PATTERNS):
retVal = setColor(message, bold=True, istty=istty)
return retVal
def setColor(message, color=None, bold=False, level=None, istty=None):
Sets ANSI color codes
>>> setColor("Hello World", color="red", istty=True)
'\\x1b[31mHello World\\x1b[0m'
>>> setColor("[INFO] Hello World", istty=True)
'[\\x1b[32mINFO\\x1b[0m] Hello World'
>>> setColor("[INFO] Hello [CRITICAL] World", istty=True)
'[INFO] Hello [CRITICAL] World'
retVal = message
if message:
if (IS_TTY or istty) and not conf.get("disableColoring"): # colorizing handler
if level is None:
levels = re.findall(r"\[(?P<result>%s)\]" % '|'.join(_[0] for _ in getPublicTypeMembers(LOGGING_LEVELS)), message)
if len(levels) == 1:
level = levels[0]
if bold or color:
retVal = colored(message, color=color, on_color=None, attrs=("bold",) if bold else None)
elif level:
level = getattr(logging, level, None)
level = None
retVal = LOGGER_HANDLER.colorize(message, level, True)
match ="\(([^)]*)\s*fork\)", message)
if match:
retVal = retVal.replace(, colored(, color="lightgrey"))
if not any(_ in message for _ in ("Payload: ",)):
for match in re.finditer(r"([^\w])'([^\n']+)'", message): # single-quoted (Note: watch-out for the banner)
retVal = retVal.replace(, "%s'%s'" % (, colored(, color="lightgrey")))
message = message.strip()
return retVal
def clearColors(message):
Clears ANSI color codes
>>> clearColors("\x1b[38;5;82mHello \x1b[38;5;198mWorld")
'Hello World'
retVal = message
if isinstance(message, str):
retVal = re.sub(r"\x1b\[[\d;]+m", "", message)
return retVal
def dataToStdout(data, forceOutput=False, bold=False, contentType=None, status=CONTENT_STATUS.IN_PROGRESS, coloring=True):
Writes text to the stdout (console) stream
if not IS_TTY and isinstance(data, six.string_types) and data.startswith("\r"):
if"\(\d+%\)", data):
data = ""
data = "\n%s" % data.strip("\r")
if not kb.get("threadException"):
if forceOutput or not (getCurrentThreadData().disableStdOut or kb.get("wizardMode")):
multiThreadMode = isMultiThreadMode()
if multiThreadMode:
if conf.get("api"):
sys.stdout.write(stdoutEncode(clearColors(data)), status, contentType)
sys.stdout.write(stdoutEncode(setColor(data, bold=bold) if coloring else clearColors(data)))
except IOError:
if multiThreadMode:
kb.prependFlag = isinstance(data, six.string_types) and (len(data) == 1 and data not in ('\n', '\r') or len(data) > 2 and data[0] == '\r' and data[-1] != '\n')
def dataToTrafficFile(data):
if not conf.trafficFile:
except IOError as ex:
errMsg = "something went wrong while trying "
errMsg += "to write to the traffic file '%s' ('%s')" % (conf.trafficFile, getSafeExString(ex))
raise SqlmapSystemException(errMsg)
def dataToDumpFile(dumpFile, data):
except IOError as ex:
if "No space left" in getUnicode(ex):
errMsg = "no space left on output device"
elif "Permission denied" in getUnicode(ex):
errMsg = "permission denied when flushing dump data"
errMsg = "error occurred when writing dump data to file ('%s')" % getUnicode(ex)
def dataToOutFile(filename, data):
Saves data to filename
>>> pushValue(conf.get("filePath"))
>>> conf.filePath = tempfile.gettempdir()
>>> "_etc_passwd" in dataToOutFile("/etc/passwd", b":::*")
>>> conf.filePath = popValue()
retVal = None
if data:
while True:
retVal = os.path.join(conf.filePath, filePathToSafeString(filename))
with open(retVal, "w+b") as f: # has to stay as non-codecs because data is raw ASCII encoded data
except UnicodeEncodeError as ex:
_ = normalizeUnicode(filename)
if filename != _:
filename = _
errMsg = "couldn't write to the "
errMsg += "output file ('%s')" % getSafeExString(ex)
raise SqlmapGenericException(errMsg)
except IOError as ex:
errMsg = "something went wrong while trying to write "
errMsg += "to the output file ('%s')" % getSafeExString(ex)
raise SqlmapGenericException(errMsg)
return retVal
def readInput(message, default=None, checkBatch=True, boolean=False):
Reads input from terminal
retVal = None
message = getUnicode(message)
if "\n" in message:
message += "%s> " % ("\n" if message.count("\n") > 1 else "")
elif message[-1] == ']':
message += " "
if kb.get("prependFlag"):
message = "\n%s" % message
kb.prependFlag = False
if conf.get("answers"):
if not any(_ in conf.answers for _ in ",="):
return conf.answers
for item in conf.answers.split(','):
question = item.split('=')[0].strip()
answer = item.split('=')[1] if len(item.split('=')) > 1 else None
if answer and question.lower() in message.lower():
retVal = getUnicode(answer, UNICODE_ENCODING)
elif answer is None and retVal:
retVal = "%s,%s" % (retVal, getUnicode(item, UNICODE_ENCODING))
if message and IS_TTY:
message = "\r%s" % message
if retVal:
dataToStdout("%s%s\n" % (message, retVal), forceOutput=not kb.wizardMode, bold=True)
debugMsg = "used the given answer"
if retVal is None:
if checkBatch and conf.get("batch") or any(conf.get(_) for _ in ("api", "nonInteractive")):
if isListLike(default):
options = ','.join(getUnicode(opt, UNICODE_ENCODING) for opt in default)
elif default:
options = getUnicode(default, UNICODE_ENCODING)
options = six.text_type()
dataToStdout("%s%s\n" % (message, options), forceOutput=not kb.wizardMode, bold=True)
debugMsg = "used the default behavior, running in batch mode"
retVal = default
if conf.get("beep"):
dataToStdout("%s" % message, forceOutput=not kb.wizardMode, bold=True)
kb.prependFlag = False
retVal = _input()
if not retVal: # Note: Python doesn't print newline on empty input
retVal = retVal.strip() or default
retVal = getUnicode(retVal, encoding=getattr(sys.stdin, "encoding", None)) if retVal else retVal
time.sleep(0.05) # Reference:
kb.prependFlag = True
raise SqlmapUserQuitException
if retVal and default and isinstance(default, six.string_types) and len(default) == 1:
retVal = retVal.strip()
if boolean:
retVal = retVal.strip().upper() == 'Y'
retVal = retVal or ""
return retVal
def setTechnique(technique):
Thread-safe setting of currently used technique (Note: dealing with cases of per-thread technique switching)
getCurrentThreadData().technique = technique
def getTechnique():
Thread-safe getting of currently used technique
return getCurrentThreadData().technique or kb.get("technique")
def randomRange(start=0, stop=1000, seed=None):
Returns random integer value in given range
>>> random.seed(0)
>>> randomRange(1, 500)
if seed is not None:
_ = getCurrentThreadData().random
randint = _.randint
randint = random.randint
return int(randint(start, stop))
def randomInt(length=4, seed=None):
Returns random integer value with provided number of digits
>>> random.seed(0)
>>> randomInt(6)
if seed is not None:
_ = getCurrentThreadData().random
choice = _.choice
choice = random.choice
return int("".join(choice(string.digits if _ != 0 else string.digits.replace('0', '')) for _ in xrange(0, length)))
def randomStr(length=4, lowercase=False, alphabet=None, seed=None):
Returns random string value with provided number of characters
>>> random.seed(0)
>>> randomStr(6)
if seed is not None:
_random = getCurrentThreadData().random
choice = _random.choice
choice = random.choice
if alphabet:
retVal = "".join(choice(alphabet) for _ in xrange(0, length))
elif lowercase:
retVal = "".join(choice(string.ascii_lowercase) for _ in xrange(0, length))
retVal = "".join(choice(string.ascii_letters) for _ in xrange(0, length))
return retVal
def sanitizeStr(value):
Sanitizes string value in respect to newline and line-feed characters
>>> sanitizeStr('foo\\n\\rbar') == 'foo bar'
>>> sanitizeStr(None) == 'None'
return getUnicode(value).replace("\n", " ").replace("\r", "")
def getHeader(headers, key):
Returns header value ignoring the letter case
>>> getHeader({"Foo": "bar"}, "foo")
retVal = None
for header in (headers or {}):
if header.upper() == key.upper():
retVal = headers[header]
return retVal
def checkPipedInput():
Checks whether input to program has been provided via standard input (e.g. cat /tmp/req.txt | python -r -)
# Reference:
return hasattr(sys.stdin, "fileno") and not os.isatty(sys.stdin.fileno())
def isZipFile(filename):
Checks if file contains zip compressed content
>>> isZipFile(paths.WORDLIST)
return openFile(filename, "rb", encoding=None).read(len(ZIP_HEADER)) == ZIP_HEADER
def isDigit(value):
Checks if provided (string) value consists of digits (Note: Python's isdigit() is problematic)
>>> u'\xb2'.isdigit()
>>> isDigit(u'\xb2')
>>> isDigit('123456')
>>> isDigit('3b3')
return"\A[0-9]+\Z", value or "") is not None
def checkFile(filename, raiseOnError=True):
Checks for file existence and readability
>>> checkFile(__file__)
valid = True
if filename:
filename = filename.strip('"\'')
if filename == STDIN_PIPE_DASH:
return checkPipedInput()
if filename is None or not os.path.isfile(filename):
valid = False
valid = False
if valid:
with open(filename, "rb"):
valid = False
if not valid and raiseOnError:
raise SqlmapSystemException("unable to read file '%s'" % filename)
return valid
def banner():
This function prints sqlmap banner with its version
if not any(_ in sys.argv for _ in ("--version", "--api")) and not conf.get("disableBanner"):
result = BANNER
if not IS_TTY or any(_ in sys.argv for _ in ("--disable-coloring", "--disable-colouring")):
result = clearColors(result)
elif IS_WIN:
dataToStdout(result, forceOutput=True)
def parsePasswordHash(password):
In case of Microsoft SQL Server password hash value is expanded to its components
>>> pushValue(kb.forcedDbms)
>>> kb.forcedDbms = DBMS.MSSQL
>>> "salt: 4086ceb6" in parsePasswordHash("0x01004086ceb60c90646a8ab9889fe3ed8e5c150b5460ece8425a")
>>> kb.forcedDbms = popValue()
blank = ' ' * 8
if isNoneValue(password) or password == ' ':
retVal = NULL
retVal = password
if Backend.isDbms(DBMS.MSSQL) and retVal != NULL and isHexEncodedString(password):
retVal = "%s\n" % password
retVal += "%sheader: %s\n" % (blank, password[:6])
retVal += "%ssalt: %s\n" % (blank, password[6:14])
retVal += "%smixedcase: %s\n" % (blank, password[14:54])
if password[54:]:
retVal += "%suppercase: %s" % (blank, password[54:])
return retVal
def cleanQuery(query):
Switch all SQL statement (alike) keywords to upper case
>>> cleanQuery("select id from users")
'SELECT id FROM users'
retVal = query
for sqlStatements in SQL_STATEMENTS.values():
for sqlStatement in sqlStatements:
candidate = sqlStatement.replace("(", "").replace(")", "").strip()
queryMatch ="(?i)\b(%s)\b" % candidate, query)
if queryMatch and "sys_exec" not in query:
retVal = retVal.replace(, candidate.upper())
return retVal
def setPaths(rootPath):
Sets absolute paths for project directories and files
paths.SQLMAP_ROOT_PATH = rootPath
# sqlmap paths
paths.SQLMAP_DATA_PATH = os.path.join(paths.SQLMAP_ROOT_PATH, "data")
paths.SQLMAP_EXTRAS_PATH = os.path.join(paths.SQLMAP_ROOT_PATH, "extra")
paths.SQLMAP_SETTINGS_PATH = os.path.join(paths.SQLMAP_ROOT_PATH, "lib", "core", "")
paths.SQLMAP_TAMPER_PATH = os.path.join(paths.SQLMAP_ROOT_PATH, "tamper")
paths.SQLMAP_PROCS_PATH = os.path.join(paths.SQLMAP_DATA_PATH, "procs")
paths.SQLMAP_SHELL_PATH = os.path.join(paths.SQLMAP_DATA_PATH, "shell")
paths.SQLMAP_TXT_PATH = os.path.join(paths.SQLMAP_DATA_PATH, "txt")
paths.SQLMAP_UDF_PATH = os.path.join(paths.SQLMAP_DATA_PATH, "udf")
paths.SQLMAP_XML_PATH = os.path.join(paths.SQLMAP_DATA_PATH, "xml")
paths.SQLMAP_XML_BANNER_PATH = os.path.join(paths.SQLMAP_XML_PATH, "banner")
paths.SQLMAP_XML_PAYLOADS_PATH = os.path.join(paths.SQLMAP_XML_PATH, "payloads")
# sqlmap files
paths.COMMON_COLUMNS = os.path.join(paths.SQLMAP_TXT_PATH, "common-columns.txt")
paths.COMMON_FILES = os.path.join(paths.SQLMAP_TXT_PATH, "common-files.txt")
paths.COMMON_TABLES = os.path.join(paths.SQLMAP_TXT_PATH, "common-tables.txt")
paths.COMMON_OUTPUTS = os.path.join(paths.SQLMAP_TXT_PATH, 'common-outputs.txt')
paths.SQL_KEYWORDS = os.path.join(paths.SQLMAP_TXT_PATH, "keywords.txt")
paths.SMALL_DICT = os.path.join(paths.SQLMAP_TXT_PATH, "smalldict.txt")
paths.USER_AGENTS = os.path.join(paths.SQLMAP_TXT_PATH, "user-agents.txt")
paths.WORDLIST = os.path.join(paths.SQLMAP_TXT_PATH, "wordlist.tx_")
paths.ERRORS_XML = os.path.join(paths.SQLMAP_XML_PATH, "errors.xml")
paths.BOUNDARIES_XML = os.path.join(paths.SQLMAP_XML_PATH, "boundaries.xml")
paths.QUERIES_XML = os.path.join(paths.SQLMAP_XML_PATH, "queries.xml")
paths.GENERIC_XML = os.path.join(paths.SQLMAP_XML_BANNER_PATH, "generic.xml")
paths.MSSQL_XML = os.path.join(paths.SQLMAP_XML_BANNER_PATH, "mssql.xml")
paths.MYSQL_XML = os.path.join(paths.SQLMAP_XML_BANNER_PATH, "mysql.xml")
paths.ORACLE_XML = os.path.join(paths.SQLMAP_XML_BANNER_PATH, "oracle.xml")
paths.PGSQL_XML = os.path.join(paths.SQLMAP_XML_BANNER_PATH, "postgresql.xml")
for path in paths.values():
if any(path.endswith(_) for _ in (".txt", ".xml", ".tx_")):
if IS_WIN:
# Reference:
if os.getenv("LOCALAPPDATA"):
paths.SQLMAP_HOME_PATH = os.path.expandvars("%LOCALAPPDATA%\\sqlmap")
elif os.getenv("USERPROFILE"):
paths.SQLMAP_HOME_PATH = os.path.expandvars("%USERPROFILE%\\Local Settings\\sqlmap")
paths.SQLMAP_HOME_PATH = os.path.join(os.path.expandvars(os.path.expanduser("~")), "sqlmap")
paths.SQLMAP_HOME_PATH = os.path.join(os.path.expandvars(os.path.expanduser("~")), ".sqlmap")
if not os.path.isdir(paths.SQLMAP_HOME_PATH):
if "XDG_DATA_HOME" in os.environ:
paths.SQLMAP_HOME_PATH = os.path.join(os.environ["XDG_DATA_HOME"], "sqlmap")
paths.SQLMAP_HOME_PATH = os.path.join(os.path.expandvars(os.path.expanduser("~")), ".local", "share", "sqlmap")
paths.SQLMAP_OUTPUT_PATH = getUnicode(paths.get("SQLMAP_OUTPUT_PATH", os.path.join(paths.SQLMAP_HOME_PATH, "output")), encoding=sys.getfilesystemencoding() or UNICODE_ENCODING)
paths.SQLMAP_DUMP_PATH = os.path.join(paths.SQLMAP_OUTPUT_PATH, "%s", "dump")
paths.SQLMAP_FILES_PATH = os.path.join(paths.SQLMAP_OUTPUT_PATH, "%s", "files")
# History files
paths.SQLMAP_HISTORY_PATH = getUnicode(os.path.join(paths.SQLMAP_HOME_PATH, "history"), encoding=sys.getfilesystemencoding() or UNICODE_ENCODING)
paths.API_SHELL_HISTORY = os.path.join(paths.SQLMAP_HISTORY_PATH, "api.hst")
paths.OS_SHELL_HISTORY = os.path.join(paths.SQLMAP_HISTORY_PATH, "os.hst")
paths.SQL_SHELL_HISTORY = os.path.join(paths.SQLMAP_HISTORY_PATH, "sql.hst")
paths.SQLMAP_SHELL_HISTORY = os.path.join(paths.SQLMAP_HISTORY_PATH, "sqlmap.hst")
paths.GITHUB_HISTORY = os.path.join(paths.SQLMAP_HISTORY_PATH, "github.hst")
def weAreFrozen():
Returns whether we are frozen via py2exe.
This will affect how we find out where we are located.
# Reference:
return hasattr(sys, "frozen")
def parseTargetDirect():
Parse target dbms and set some attributes into the configuration singleton
>>> pushValue(
>>> = "mysql://root:testpass@"
>>> parseTargetDirect()
>>> conf.dbmsDb
>>> conf.dbmsPass
>>> = popValue()
if not
details = None
remote = False
for dbms in SUPPORTED_DBMS:
details ="^(?P<dbms>%s)://(?P<credentials>(?P<user>.*?)\:(?P<pass>.*)\@)?(?P<remote>(?P<hostname>[\w.-]+?)\:(?P<port>[\d]+)\/)?(?P<db>[\w\d\ \:\.\_\-\/\\]*)$" % dbms,, re.I)
if details:
conf.dbms ="dbms")
conf.dbmsUser ="user")
conf.dbmsPass ="pass")
if conf.dbmsCred:
conf.dbmsUser, conf.dbmsPass = conf.dbmsCred.split(':')
conf.dbmsUser = ""
conf.dbmsPass = ""
if not conf.dbmsPass:
conf.dbmsPass = None
remote = True
conf.hostname ="hostname").strip()
conf.port = int("port"))
conf.hostname = "localhost"
conf.port = 0
conf.dbmsDb ="db").strip() if"db") is not None else None
conf.parameters[None] = "direct connection"
if kb.smokeMode:
if not details:
errMsg = "invalid target details, valid syntax is for instance "
errMsg += "or 'access://DATABASE_FILEPATH'"
raise SqlmapSyntaxException(errMsg)
for dbmsName, data in DBMS_DICT.items():
if dbmsName == conf.dbms or conf.dbms.lower() in data[0]:
conf.dbms = dbmsName
if remote:
warnMsg = "direct connection over the network for "
warnMsg += "%s DBMS is not supported" % dbmsName
conf.hostname = "localhost"
conf.port = 0
elif not remote:
errMsg = "missing remote connection details (e.g. "
errMsg += "or 'access://DATABASE_FILEPATH')"
raise SqlmapSyntaxException(errMsg)
if dbmsName in (DBMS.MSSQL, DBMS.SYBASE):
pymssql = __import__("pymssql")
if not hasattr(pymssql, "__version__") or pymssql.__version__ < "1.0.2":
errMsg = "'%s' third-party library must be " % data[1]
errMsg += "version >= 1.0.2 to work properly. "
errMsg += "Download from '%s'" % data[2]
raise SqlmapMissingDependence(errMsg)
elif dbmsName == DBMS.MYSQL:
elif dbmsName == DBMS.PGSQL:
elif dbmsName == DBMS.ORACLE:
# Reference:
if (conf.dbmsUser or "").upper() == "SYS": = "%s?mode=SYSDBA" %
elif dbmsName == DBMS.SQLITE:
elif dbmsName == DBMS.ACCESS:
elif dbmsName == DBMS.FIREBIRD:
except (SqlmapSyntaxException, SqlmapMissingDependence):
if _sqlalchemy and data[3] and any(_ in _sqlalchemy.dialects.__all__ for _ in (data[3], data[3].split('+')[0])):
errMsg = "sqlmap requires '%s' third-party library " % data[1]
errMsg += "in order to directly connect to the DBMS "
errMsg += "'%s'. You can download it from '%s'" % (dbmsName, data[2])
errMsg += ". Alternative is to use a package 'python-sqlalchemy' "
errMsg += "with support for dialect '%s' installed" % data[3]
raise SqlmapMissingDependence(errMsg)
def parseTargetUrl():
Parse target URL and set some attributes into the configuration singleton
>>> pushValue(conf.url)
>>> conf.url = ""
>>> parseTargetUrl()
>>> conf.hostname
>>> conf.scheme
>>> conf.url = popValue()
if not conf.url:
originalUrl = conf.url
if"://\[.+\]", conf.url) and not socket.has_ipv6:
errMsg = "IPv6 communication is not supported "
errMsg += "on this platform"
raise SqlmapGenericException(errMsg)
if not"^(http|ws)s?://", conf.url, re.I):
if":443\b", conf.url):
conf.url = "https://%s" % conf.url
conf.url = "http://%s" % conf.url
if kb.customInjectionMark in conf.url:
conf.url = conf.url.replace('?', URI_QUESTION_MARKER)
urlSplit = _urllib.parse.urlsplit(conf.url)
except ValueError as ex:
errMsg = "invalid URL '%s' has been given ('%s'). " % (conf.url, getSafeExString(ex))
errMsg += "Please be sure that you don't have any leftover characters (e.g. '[' or ']') "
errMsg += "in the hostname part"
raise SqlmapGenericException(errMsg)
hostnamePort = urlSplit.netloc.split(":") if not"\[.+\]", urlSplit.netloc) else filterNone(("\[.+\]", urlSplit.netloc).group(0),"\](:(?P<port>\d+))?", urlSplit.netloc).group("port")))
conf.scheme = (urlSplit.scheme.strip().lower() or "http")
conf.path = urlSplit.path.strip()
conf.hostname = hostnamePort[0].strip()
if conf.forceSSL:
conf.scheme = re.sub(r"(?i)\A(http|ws)\Z", r"\g<1>s", conf.scheme)
conf.ipv6 = conf.hostname != conf.hostname.strip("[]")
conf.hostname = conf.hostname.strip("[]").replace(kb.customInjectionMark, "")
except (LookupError, UnicodeError):
invalid = True
invalid = False
if any((invalid,"\s", conf.hostname), '..' in conf.hostname, conf.hostname.startswith('.'), '\n' in originalUrl)):
errMsg = "invalid target URL ('%s')" % originalUrl
raise SqlmapSyntaxException(errMsg)
if len(hostnamePort) == 2:
conf.port = int(hostnamePort[1])
errMsg = "invalid target URL"
raise SqlmapSyntaxException(errMsg)
elif conf.scheme in ("https", "wss"):
conf.port = 443
conf.port = 80
if conf.port < 1 or conf.port > 65535:
errMsg = "invalid target URL port (%d)" % conf.port
raise SqlmapSyntaxException(errMsg)
conf.url = getUnicode("%s://%s:%d%s" % (conf.scheme, ("[%s]" % conf.hostname) if conf.ipv6 else conf.hostname, conf.port, conf.path))
conf.url = conf.url.replace(URI_QUESTION_MARKER, '?')
if urlSplit.query:
if '=' not in urlSplit.query:
conf.url = "%s?%s" % (conf.url, getUnicode(urlSplit.query))
conf.parameters[PLACE.GET] = urldecode(urlSplit.query, spaceplus=not conf.base64Parameter) if urlSplit.query and urlencode(DEFAULT_GET_POST_DELIMITER, None) not in urlSplit.query else urlSplit.query
if (intersect(REFERER_ALIASES, conf.testParameter, True) or conf.level >= 3) and not any(_[0].upper() == HTTP_HEADER.REFERER.upper() for _ in conf.httpHeaders):
debugMsg = "setting the HTTP Referer header to the target URL"
conf.httpHeaders = [_ for _ in conf.httpHeaders if _[0] != HTTP_HEADER.REFERER]
conf.httpHeaders.append((HTTP_HEADER.REFERER, conf.url.replace(kb.customInjectionMark, "")))
if (intersect(HOST_ALIASES, conf.testParameter, True) or conf.level >= 5) and not any(_[0].upper() == HTTP_HEADER.HOST.upper() for _ in conf.httpHeaders):
debugMsg = "setting the HTTP Host header to the target URL"
conf.httpHeaders = [_ for _ in conf.httpHeaders if _[0] != HTTP_HEADER.HOST]
conf.httpHeaders.append((HTTP_HEADER.HOST, getHostHeader(conf.url)))
if conf.url != originalUrl:
kb.originalUrls[conf.url] = originalUrl
def escapeJsonValue(value):
Escapes JSON value (used in payloads)
# Reference:
>>> "\\n" in escapeJsonValue("foo\\nbar")
>>> "\\\\t" in escapeJsonValue("foo\\tbar")
retVal = ""
for char in value:
if char < ' ' or char == '"':
retVal += json.dumps(char)[1:-1]
retVal += char
return retVal
def expandAsteriskForColumns(expression):
If the user provided an asterisk rather than the column(s)
name, sqlmap will retrieve the columns itself and reprocess
the SQL query string (expression)
match ="(?i)\ASELECT(\s+TOP\s+[\d]+)?\s+\*\s+FROM\s+((`[^`]+`|[^\s]+)+)", expression)
if match:
infoMsg = "you did not provide the fields in your query. "
infoMsg += "sqlmap will retrieve the column names itself"
_ ="..", '.').replace(".dbo.", '.')
db, conf.tbl = _.split('.', 1) if '.' in _ else (None, _)
if db is None:
if expression != conf.sqlQuery:
conf.db = db
elif conf.db:
expression = re.sub(r"([^\w])%s" % re.escape(conf.tbl), r"\g<1>%s.%s" % (conf.db, conf.tbl), expression)
conf.db = db
conf.db = safeSQLIdentificatorNaming(conf.db)
conf.tbl = safeSQLIdentificatorNaming(conf.tbl, True)
columnsDict = conf.dbmsHandler.getColumns(onlyColNames=True)
if columnsDict and conf.db in columnsDict and conf.tbl in columnsDict[conf.db]:
columns = list(columnsDict[conf.db][conf.tbl].keys())
columnsStr = ", ".join(column for column in columns)
expression = expression.replace('*', columnsStr, 1)
infoMsg = "the query with expanded column name(s) is: "
infoMsg += "%s" % expression
return expression
def getLimitRange(count, plusOne=False):
Returns range of values used in limit/offset constructs
>>> [_ for _ in getLimitRange(10)]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
retVal = None
count = int(count)
limitStart, limitStop = 1, count
reverse = False
if kb.dumpTable:
if conf.limitStart and conf.limitStop and conf.limitStart > conf.limitStop:
limitStop = conf.limitStart
limitStart = conf.limitStop
reverse = True
if isinstance(conf.limitStop, int) and conf.limitStop > 0 and conf.limitStop < limitStop:
limitStop = conf.limitStop
if isinstance(conf.limitStart, int) and conf.limitStart > 0 and conf.limitStart <= limitStop:
limitStart = conf.limitStart
retVal = xrange(limitStart, limitStop + 1) if plusOne else xrange(limitStart - 1, limitStop)
if reverse:
retVal = xrange(retVal[-1], retVal[0] - 1, -1)
return retVal
def parseUnionPage(page):
Returns resulting items from UNION query inside provided page content
>>> parseUnionPage("%sfoo%s%sbar%s" % (kb.chars.start, kb.chars.stop, kb.chars.start, kb.chars.stop))
['foo', 'bar']
if page is None:
return None
if"(?si)\A%s.*%s\Z" % (kb.chars.start, kb.chars.stop), page):
warnMsg = "large output detected. This might take a while"
data = BigArray()
keys = set()
for match in re.finditer(r"%s(.*?)%s" % (kb.chars.start, kb.chars.stop), page, re.DOTALL | re.IGNORECASE):
entry =
if kb.chars.start in entry:
entry = entry.split(kb.chars.start)[-1]
if kb.unionDuplicates:
key = entry.lower()
if key not in keys:
entry = entry.split(kb.chars.delimiter)
if conf.hexConvert:
entry = applyFunctionRecursively(entry, decodeDbmsHexValue)
if kb.safeCharEncode:
entry = applyFunctionRecursively(entry, safecharencode)
data.append(entry[0] if len(entry) == 1 else entry)
data = page
if len(data) == 1 and isinstance(data[0], six.string_types):
data = data[0]
return data
def parseFilePaths(page):
Detects (possible) absolute system paths inside the provided page content
>>> _ = "/var/www/html/index.php"; parseFilePaths("<html>Error occurred at line 207 of: %s<br>Please contact your administrator</html>" % _); _ in kb.absFilePaths
if page:
for regex in FILE_PATH_REGEXES:
for match in re.finditer(regex, page):
absFilePath ="result").strip()
page = page.replace(absFilePath, "")
if isWindowsDriveLetterPath(absFilePath):
absFilePath = posixToNtSlashes(absFilePath)
if absFilePath not in kb.absFilePaths:
def getLocalIP():
Get local IP address (exposed to the remote/target)
retVal = None
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((conf.hostname, conf.port))
retVal, _ = s.getsockname()
debugMsg = "there was an error in opening socket "
debugMsg += "connection toward '%s'" % conf.hostname
return retVal
def getRemoteIP():
Get remote/target IP address
>>> pushValue(conf.hostname)
>>> conf.hostname = "localhost"
>>> getRemoteIP() == ""
>>> conf.hostname = popValue()
retVal = None
retVal = socket.gethostbyname(conf.hostname)
except socket.gaierror:
errMsg = "address resolution problem "
errMsg += "occurred for hostname '%s'" % conf.hostname
singleTimeLogMessage(errMsg, logging.ERROR)
return retVal
def getFileType(filePath):
Returns "magic" file type for given file path
>>> getFileType(__file__)
>>> getFileType(sys.executable)
desc = magic.from_file(filePath) or magic.MAGIC_UNKNOWN_FILETYPE
desc = getText(desc)
if desc == getText(magic.MAGIC_UNKNOWN_FILETYPE):
content = openFile(filePath, "rb", encoding=None).read()
desc = "ascii"
return "text" if any(_ in desc.lower() for _ in ("ascii", "text")) else "binary"
def getCharset(charsetType=None):
Returns list with integers representing characters of a given
charset type appropriate for inference techniques
[0, 1, 47, 48, 49]
asciiTbl = []
if charsetType is None:
asciiTbl.extend(xrange(0, 128))
# Binary
elif charsetType == CHARSET_TYPE.BINARY:
asciiTbl.extend((0, 1))
asciiTbl.extend(xrange(47, 50))
# Digits
elif charsetType == CHARSET_TYPE.DIGITS:
asciiTbl.extend((0, 9))
asciiTbl.extend(xrange(47, 58))
# Hexadecimal
asciiTbl.extend((0, 1))
asciiTbl.extend(xrange(47, 58))
asciiTbl.extend(xrange(64, 71))
asciiTbl.extend((87, 88)) # X
asciiTbl.extend(xrange(96, 103))
asciiTbl.extend((119, 120)) # x
# Characters
elif charsetType == CHARSET_TYPE.ALPHA:
asciiTbl.extend((0, 1))
asciiTbl.extend(xrange(64, 91))
asciiTbl.extend(xrange(96, 123))
# Characters and digits
elif charsetType == CHARSET_TYPE.ALPHANUM:
asciiTbl.extend((0, 1))
asciiTbl.extend(xrange(47, 58))
asciiTbl.extend(xrange(64, 91))
asciiTbl.extend(xrange(96, 123))
return asciiTbl
def directoryPath(filepath):
Returns directory path for a given filepath
>>> directoryPath('/var/log/apache.log')
>>> directoryPath('/var/log')
retVal = filepath
if filepath and os.path.splitext(filepath)[-1]:
retVal = ntpath.dirname(filepath) if isWindowsDriveLetterPath(filepath) else posixpath.dirname(filepath)
return retVal
def normalizePath(filepath):
Returns normalized string representation of a given filepath
>>> normalizePath('//var///log/apache.log')
retVal = filepath
if retVal:
retVal = retVal.strip("\r\n")
retVal = ntpath.normpath(retVal) if isWindowsDriveLetterPath(retVal) else re.sub(r"\A/{2,}", "/", posixpath.normpath(retVal))
return retVal
def safeFilepathEncode(filepath):
Returns filepath in (ASCII) format acceptable for OS handling (e.g. reading)
>>> 'sqlmap' in safeFilepathEncode(paths.SQLMAP_HOME_PATH)
retVal = filepath
if filepath and six.PY2 and isinstance(filepath, six.text_type):
retVal = getBytes(filepath, sys.getfilesystemencoding() or UNICODE_ENCODING)
return retVal
def safeExpandUser(filepath):
Patch for a Python Issue18171 (
>>> os.path.basename(__file__) in safeExpandUser(__file__)
retVal = filepath
retVal = os.path.expanduser(filepath)
except UnicodeError:
_ = locale.getdefaultlocale()
encoding = _[1] if _ and len(_) > 1 else UNICODE_ENCODING
retVal = getUnicode(os.path.expanduser(filepath.encode(encoding)), encoding=encoding)
return retVal
def safeStringFormat(format_, params):
Avoids problems with inappropriate string format strings
>>> safeStringFormat('SELECT foo FROM %s LIMIT %d', ('bar', '1'))
>>> safeStringFormat("SELECT foo FROM %s WHERE name LIKE '%susan%' LIMIT %d", ('bar', '1'))
"SELECT foo FROM bar WHERE name LIKE '%susan%' LIMIT 1"
if format_.count(PAYLOAD_DELIMITER) == 2:
_ = format_.split(PAYLOAD_DELIMITER)
_[1] = re.sub(r"(\A|[^A-Za-z0-9])(%d)([^A-Za-z0-9]|\Z)", r"\g<1>%s\g<3>", _[1])
retVal = PAYLOAD_DELIMITER.join(_)
retVal = re.sub(r"(\A|[^A-Za-z0-9])(%d)([^A-Za-z0-9]|\Z)", r"\g<1>%s\g<3>", format_)
if isinstance(params, six.string_types):
retVal = retVal.replace("%s", params, 1)
elif not isListLike(params):
retVal = retVal.replace("%s", getUnicode(params), 1)
start, end = 0, len(retVal)
match ="%s(.+)%s" % (PAYLOAD_DELIMITER, PAYLOAD_DELIMITER), retVal)
if match and PAYLOAD_DELIMITER not in
start, end = match.start(), match.end()
if retVal.count("%s", start, end) == len(params):
for param in params:
index = retVal.find("%s", start)
if isinstance(param, six.string_types):
param = param.replace('%', PARAMETER_PERCENTAGE_MARKER)
retVal = retVal[:index] + getUnicode(param) + retVal[index + 2:]
if any('%s' in _ for _ in conf.parameters.values()):
parts = format_.split(' ')
for i in xrange(len(parts)):
if PAYLOAD_DELIMITER in parts[i]:
parts[i] = parts[i].replace(PAYLOAD_DELIMITER, "")
parts[i] = "%s%s" % (parts[i], PAYLOAD_DELIMITER)
format_ = ' '.join(parts)
count = 0
while True:
match ="(\A|[^A-Za-z0-9])(%s)([^A-Za-z0-9]|\Z)", retVal)
if match:
if count >= len(params):
warnMsg = "wrong number of parameters during string formatting. "
warnMsg += "Please report by e-mail content \"%r | %r | %r\" to '%s'" % (format_, params, retVal, DEV_EMAIL_ADDRESS)
raise SqlmapValueException(warnMsg)
retVal = re.sub(r"(\A|[^A-Za-z0-9])(%s)([^A-Za-z0-9]|\Z)", r"\g<1>%s\g<3>" % params[count], retVal, 1)
except re.error:
retVal = retVal.replace(, % params[count], 1)
count += 1
retVal = getText(retVal).replace(PARAMETER_PERCENTAGE_MARKER, '%')
return retVal
def getFilteredPageContent(page, onlyText=True, split=" "):
Returns filtered page content without script, style and/or comments
or all HTML tags
>>> getFilteredPageContent(u'<html><title>foobar</title><body>test</body></html>') == "foobar test"
retVal = page
# only if the page's charset has been successfully identified
if isinstance(page, six.text_type):
retVal = re.sub(r"(?si)<script.+?</script>|<!--.+?-->|<style.+?</style>%s" % (r"|<[^>]+>|\t|\n|\r" if onlyText else ""), split, page)
retVal = re.sub(r"%s{2,}" % split, split, retVal)
retVal = htmlUnescape(retVal.strip().strip(split))
return retVal
def getPageWordSet(page):
Returns word set used in page content
>>> sorted(getPageWordSet(u'<html><title>foobar</title><body>test</body></html>')) == [u'foobar', u'test']
retVal = set()
# only if the page's charset has been successfully identified
if isinstance(page, six.string_types):
retVal = set( for _ in re.finditer(r"\w+", getFilteredPageContent(page)))
return retVal
def showStaticWords(firstPage, secondPage, minLength=3):
Prints words appearing in two different response pages
>>> showStaticWords("this is a test", "this is another test")