mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-08-04 14:55:40 +00:00
361 lines
17 KiB
Python
361 lines
17 KiB
Python
#!/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")
|