mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-08-04 14:55:40 +00:00
Add experimental gRPC-Web (text) support
This commit is contained in:
Miroslav Štampar
parent
76b75e95f0
commit
ab6ce22593
9 changed files with 619 additions and 10 deletions
|
|
@ -133,7 +133,7 @@ class Agent(object):
|
|||
origValue = origValue.split(kb.customInjectionMark)[0]
|
||||
if kb.postHint in (POST_HINT.SOAP, POST_HINT.XML):
|
||||
origValue = re.split(r"['\">]", origValue)[-1]
|
||||
elif kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE):
|
||||
elif kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE, POST_HINT.GRPC_WEB):
|
||||
match = re.search(r"['\"]", origValue)
|
||||
quote = match.group(0) if match else '"'
|
||||
origValue = extractRegexResult(r"%s\s*:\s*(?P<result>\d+)\Z" % quote, origValue) or extractRegexResult(r"(?P<result>[^%s]*)\Z" % quote, origValue)
|
||||
|
|
@ -206,7 +206,7 @@ class Agent(object):
|
|||
if place in (PLACE.URI, PLACE.CUSTOM_POST, PLACE.CUSTOM_HEADER):
|
||||
_ = "%s%s" % (_origValue if base64Encoding else origValue, kb.customInjectionMark)
|
||||
|
||||
if kb.postHint == POST_HINT.JSON and isNumber(origValue) and not isNumber(newValue) and '"%s"' % _ not in paramString:
|
||||
if kb.postHint in (POST_HINT.JSON, POST_HINT.GRPC_WEB) and isNumber(origValue) and not isNumber(newValue) and '"%s"' % _ not in paramString:
|
||||
newValue = '"%s"' % self.addPayloadDelimiters(newValue)
|
||||
elif kb.postHint == POST_HINT.JSON_LIKE and isNumber(origValue) and not isNumber(newValue) and re.search(r"['\"]%s['\"]" % re.escape(_), paramString) is None:
|
||||
newValue = "'%s'" % self.addPayloadDelimiters(newValue)
|
||||
|
|
|
|||
|
|
@ -385,6 +385,7 @@ SQL_STATEMENTS = {
|
|||
POST_HINT_CONTENT_TYPES = {
|
||||
POST_HINT.JSON: "application/json",
|
||||
POST_HINT.JSON_LIKE: "application/json",
|
||||
POST_HINT.GRPC_WEB: "application/grpc-web-text",
|
||||
POST_HINT.MULTIPART: "multipart/form-data",
|
||||
POST_HINT.SOAP: "application/soap+xml",
|
||||
POST_HINT.XML: "application/xml",
|
||||
|
|
|
|||
|
|
@ -146,6 +146,7 @@ class POST_HINT(object):
|
|||
MULTIPART = "MULTIPART"
|
||||
XML = "XML (generic)"
|
||||
ARRAY_LIKE = "Array-like"
|
||||
GRPC_WEB = "gRPC-Web"
|
||||
|
||||
class HTTPMETHOD(object):
|
||||
GET = "GET"
|
||||
|
|
|
|||
|
|
@ -2295,6 +2295,7 @@ def _setKnowledgeBaseAttributes(flushAll=True):
|
|||
kb.permissionFlag = False
|
||||
kb.place = None
|
||||
kb.postHint = None
|
||||
kb.grpcWeb = None
|
||||
kb.postSpaceToPlus = False
|
||||
kb.postUrlEncode = True
|
||||
kb.prependFlag = False
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ from lib.core.enums import OS
|
|||
from thirdparty import six
|
||||
|
||||
# sqlmap version (<major>.<minor>.<month>.<monthly commit>)
|
||||
VERSION = "1.10.7.112"
|
||||
VERSION = "1.10.7.113"
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -77,6 +77,8 @@ from lib.core.settings import URI_INJECTABLE_REGEX
|
|||
from lib.core.settings import USER_AGENT_ALIASES
|
||||
from lib.core.settings import XML_RECOGNITION_REGEX
|
||||
from lib.core.threads import getCurrentThreadData
|
||||
from lib.utils.grpcweb import CONTENT_TYPE as grpcWebContentType
|
||||
from lib.utils.grpcweb import decodeBody as grpcDecodeBody
|
||||
from lib.utils.hashdb import HashDB
|
||||
from thirdparty import six
|
||||
from collections import OrderedDict
|
||||
|
|
@ -112,6 +114,16 @@ def _setRequestParams():
|
|||
if conf.data is not None:
|
||||
conf.method = conf.method or HTTPMETHOD.POST
|
||||
|
||||
# probe (side-effect-free) whether this is a gRPC-Web body; if so, expose its string fields as a
|
||||
# JSON view so the standard JSON injection path can process it. The skeleton is committed to
|
||||
# kb.grpcWeb (and the JSON view kept) ONLY on user acceptance below; if declined, the original
|
||||
# body is restored so it is sent verbatim (never the JSON surrogate)
|
||||
kb.grpcWeb = None
|
||||
_grpcOriginalData = conf.data
|
||||
_grpcView, _grpcSkeleton = grpcDecodeBody(conf.data)
|
||||
if _grpcView is not None:
|
||||
conf.data = _grpcView
|
||||
|
||||
def process(match, repl):
|
||||
retVal = match.group(0)
|
||||
|
||||
|
|
@ -145,14 +157,26 @@ def _setRequestParams():
|
|||
kb.testOnlyCustom = True
|
||||
|
||||
if re.search(JSON_RECOGNITION_REGEX, conf.data):
|
||||
message = "JSON data found in %s body. " % conf.method
|
||||
message += "Do you want to process it? [Y/n/q] "
|
||||
if _grpcSkeleton:
|
||||
message = "gRPC-Web text data found in %s body. Do you want to process its detected string fields? [Y/n/q] " % conf.method
|
||||
else:
|
||||
message = "JSON data found in %s body. Do you want to process it? [Y/n/q] " % conf.method
|
||||
choice = readInput(message, default='Y').upper()
|
||||
|
||||
if choice == 'Q':
|
||||
raise SqlmapUserQuitException
|
||||
elif choice == 'Y':
|
||||
kb.postHint = POST_HINT.JSON
|
||||
kb.postHint = POST_HINT.GRPC_WEB if _grpcSkeleton else POST_HINT.JSON
|
||||
if _grpcSkeleton:
|
||||
kb.grpcWeb = _grpcSkeleton # commit only on acceptance
|
||||
# force a grpc-web-text response (the only kind this cut decodes) by REPLACING any
|
||||
# existing Accept - a captured 'Accept: */*', or an explicit 'q=0' we cannot honor
|
||||
for _index, (_header, _value) in enumerate(conf.httpHeaders or []):
|
||||
if _header.lower() == HTTP_HEADER.ACCEPT.lower():
|
||||
conf.httpHeaders[_index] = (_header, grpcWebContentType)
|
||||
break
|
||||
else:
|
||||
conf.httpHeaders.append((HTTP_HEADER.ACCEPT, grpcWebContentType))
|
||||
if not (kb.processUserMarks and kb.customInjectionMark in conf.data):
|
||||
conf.data = getattr(conf.data, UNENCODED_ORIGINAL_VALUE, conf.data)
|
||||
conf.data = conf.data.replace(kb.customInjectionMark, ASTERISK_MARKER)
|
||||
|
|
@ -245,6 +269,11 @@ def _setRequestParams():
|
|||
conf.data = re.sub(r"(?si)(Content-Disposition:[^\n]+\s+name=\"(?P<name>[^\"]+)\"(?:[^f|^b]|f(?!ilename=)|b(?!oundary=))*?)((%s)--)" % ("\r\n" if "\r\n" in conf.data else '\n'),
|
||||
functools.partial(process, repl=r"\g<1>%s\g<3>" % kb.customInjectionMark), conf.data)
|
||||
|
||||
# a gRPC-Web body that was NOT accepted as GRPC_WEB (declined prompt) must be sent verbatim,
|
||||
# not as the JSON surrogate that was temporarily swapped into conf.data for detection
|
||||
if _grpcSkeleton is not None and kb.postHint != POST_HINT.GRPC_WEB:
|
||||
conf.data = _grpcOriginalData
|
||||
|
||||
if not kb.postHint:
|
||||
if kb.customInjectionMark in conf.data: # later processed
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -134,6 +134,8 @@ from lib.request.basic import processResponse
|
|||
from lib.request.comparison import comparison
|
||||
from lib.request.direct import direct
|
||||
from lib.request.methodrequest import MethodRequest
|
||||
from lib.utils.grpcweb import decodeResponse as grpcDecodeResponse
|
||||
from lib.utils.grpcweb import encodeBody as grpcEncodeBody
|
||||
from lib.utils.safe2bin import safecharencode
|
||||
from lib.utils.sqllint import checkSanity
|
||||
from thirdparty import six
|
||||
|
|
@ -569,6 +571,10 @@ class Connect(object):
|
|||
|
||||
logger.log(CUSTOM_LOGGING.TRAFFIC_OUT, requestMsg)
|
||||
else:
|
||||
# re-encode a gRPC-Web body from its injected JSON view, fixing length prefixes
|
||||
if kb.postHint == POST_HINT.GRPC_WEB and post is not None:
|
||||
post = grpcEncodeBody(post)
|
||||
|
||||
post = getBytes(post)
|
||||
|
||||
# Reference: https://github.com/sqlmapproject/sqlmap/issues/6049
|
||||
|
|
@ -996,6 +1002,12 @@ class Connect(object):
|
|||
singleTimeLogMessage(errMsg, logging.CRITICAL)
|
||||
raise SystemExit
|
||||
|
||||
# decode a gRPC-Web response to readable text so the oracle/error/inband paths see the
|
||||
# message fields + trailer (grpc-message) instead of an opaque base64 blob. Headers are
|
||||
# passed so a trailers-only error (empty body, grpc-status/message in headers) is still surfaced
|
||||
if kb.postHint == POST_HINT.GRPC_WEB:
|
||||
page = grpcDecodeResponse(page, responseHeaders)
|
||||
|
||||
threadData.lastPage = page
|
||||
threadData.lastCode = code
|
||||
|
||||
|
|
@ -1149,7 +1161,7 @@ class Connect(object):
|
|||
payload = payload.replace("&#", SAFE_HEX_MARKER)
|
||||
payload = payload.replace('&', "&").replace('>', ">").replace('<', "<").replace('"', """).replace("'", "'") # Reference: https://stackoverflow.com/a/1091953
|
||||
payload = payload.replace(SAFE_HEX_MARKER, "&#")
|
||||
elif kb.postHint == POST_HINT.JSON:
|
||||
elif kb.postHint in (POST_HINT.JSON, POST_HINT.GRPC_WEB):
|
||||
payload = escapeJsonValue(payload)
|
||||
elif kb.postHint == POST_HINT.JSON_LIKE:
|
||||
payload = payload.replace("'", REPLACEMENT_MARKER).replace('"', "'").replace(REPLACEMENT_MARKER, '"')
|
||||
|
|
@ -1391,7 +1403,7 @@ class Connect(object):
|
|||
value = urldecode(value, convall=True, spaceplus=(item == post and kb.postSpaceToPlus))
|
||||
variables[name] = value
|
||||
|
||||
if post and kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE):
|
||||
if post and kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE, POST_HINT.GRPC_WEB):
|
||||
json_ = parseJson(post)
|
||||
for name, value in (json_ if isinstance(json_, dict) else {}).items():
|
||||
if safeVariableNaming(name) != name:
|
||||
|
|
@ -1485,7 +1497,7 @@ class Connect(object):
|
|||
found = True
|
||||
post = re.sub(r"(?s)(\b%s>)(.*?)(</[^<]*\b%s>)" % (re.escape(name), re.escape(name)), r"\g<1>%s\g<3>" % entry.replace('\\', r'\\'), post)
|
||||
|
||||
elif kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE):
|
||||
elif kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE, POST_HINT.GRPC_WEB):
|
||||
match = re.search(r"['\"]%s['\"]:" % re.escape(name), post)
|
||||
if match:
|
||||
quote = match.group(0)[0]
|
||||
|
|
@ -1521,7 +1533,7 @@ class Connect(object):
|
|||
|
||||
if not found:
|
||||
if post is not None:
|
||||
if kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE):
|
||||
if kb.postHint in (POST_HINT.JSON, POST_HINT.JSON_LIKE, POST_HINT.GRPC_WEB):
|
||||
match = re.search(r"['\"]", post)
|
||||
if match:
|
||||
quote = match.group(0)
|
||||
|
|
|
|||
297
lib/utils/grpcweb.py
Normal file
297
lib/utils/grpcweb.py
Normal file
|
|
@ -0,0 +1,297 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import re
|
||||
import struct
|
||||
|
||||
from lib.core.data import conf
|
||||
from lib.core.data import kb
|
||||
from lib.core.enums import HTTP_HEADER
|
||||
from thirdparty.six.moves.urllib.parse import unquote
|
||||
|
||||
# gRPC-Web body support (grpc-web-text / base64, unary only). A gRPC-Web message is a length-prefixed
|
||||
# protobuf frame; grpc-web-text is that frame base64-encoded. Because protobuf is length-prefixed and
|
||||
# sqlmap injects by appending, the body is decoded to a JSON view of its (heuristically-detected) string
|
||||
# fields so the existing JSON injection engine handles marking/placement, then re-encoded with corrected
|
||||
# length prefixes at send time (see connect.py). NOT supported (deliberately not detected, never
|
||||
# corrupted): binary application/grpc-web+proto, compression (grpc-encoding), and streaming.
|
||||
|
||||
CONTENT_TYPE = "application/grpc-web-text"
|
||||
CONTENT_TYPE_PROTO = "application/grpc-web-text+proto" # equivalent spelling (message format hint)
|
||||
_MAX_VARINT_BYTES = 10 # a 64-bit varint is at most 10 bytes
|
||||
_MAX_FIELD_NUMBER = 0x1fffffff # protobuf maximum field number (2**29 - 1)
|
||||
_BASE64_QUANTUM_REGEX = re.compile(r"\A(?:[A-Za-z0-9+/]{4}|[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)\Z")
|
||||
|
||||
def _b64decode(text):
|
||||
# strict, py2/py3-safe (no 'validate=' kwarg): reject whitespace/invalid chars, but decode per 4-char
|
||||
# quantum so INDEPENDENTLY-padded base64 chunks (allowed even for a unary text response) reconstruct
|
||||
# correctly - a mid-stream padded quantum is fine, arbitrary mid-stream padding chars are not
|
||||
if isinstance(text, bytes):
|
||||
text = text.decode("ascii")
|
||||
if len(text) % 4 != 0:
|
||||
raise ValueError("invalid base64 length")
|
||||
out = bytearray()
|
||||
for offset in range(0, len(text), 4):
|
||||
quantum = text[offset:offset + 4]
|
||||
if not _BASE64_QUANTUM_REGEX.match(quantum):
|
||||
raise ValueError("invalid base64")
|
||||
out += base64.b64decode(quantum)
|
||||
return bytes(out)
|
||||
|
||||
def _readVarint(buf, pos):
|
||||
shift = result = 0
|
||||
start = pos
|
||||
while True:
|
||||
if pos >= len(buf):
|
||||
raise ValueError("truncated varint")
|
||||
b = buf[pos] if isinstance(buf[pos], int) else ord(buf[pos])
|
||||
pos += 1
|
||||
result |= (b & 0x7f) << shift
|
||||
if not (b & 0x80):
|
||||
if result >= (1 << 64):
|
||||
raise ValueError("varint exceeds 64 bits")
|
||||
return result, pos
|
||||
shift += 7
|
||||
if pos - start >= _MAX_VARINT_BYTES:
|
||||
raise ValueError("overlong varint")
|
||||
|
||||
def _writeVarint(n):
|
||||
out = bytearray()
|
||||
while True:
|
||||
b = n & 0x7f
|
||||
n >>= 7
|
||||
out.append(b | 0x80 if n else b)
|
||||
if not n:
|
||||
return bytes(out)
|
||||
|
||||
def _readExact(buf, pos, length):
|
||||
end = pos + length
|
||||
if length < 0 or end > len(buf):
|
||||
raise ValueError("truncated protobuf field")
|
||||
return buf[pos:end], end
|
||||
|
||||
def _decode(buf):
|
||||
buf = bytes(buf)
|
||||
pos = 0
|
||||
out = []
|
||||
while pos < len(buf):
|
||||
tag, pos = _readVarint(buf, pos)
|
||||
fn, wt = tag >> 3, tag & 7
|
||||
if fn == 0 or fn > _MAX_FIELD_NUMBER:
|
||||
raise ValueError("invalid field number %d" % fn)
|
||||
if wt == 0:
|
||||
val, pos = _readVarint(buf, pos)
|
||||
elif wt == 1:
|
||||
val, pos = _readExact(buf, pos, 8)
|
||||
elif wt == 2:
|
||||
ln, pos = _readVarint(buf, pos)
|
||||
val, pos = _readExact(buf, pos, ln)
|
||||
elif wt == 5:
|
||||
val, pos = _readExact(buf, pos, 4)
|
||||
else:
|
||||
raise ValueError("unsupported wire type %d" % wt)
|
||||
out.append([fn, wt, val])
|
||||
return out
|
||||
|
||||
def _encode(fields):
|
||||
out = bytearray()
|
||||
for fn, wt, val in fields:
|
||||
out += _writeVarint((fn << 3) | wt)
|
||||
if wt == 0:
|
||||
out += _writeVarint(val)
|
||||
elif wt == 2:
|
||||
val = val if isinstance(val, bytes) else bytes(val)
|
||||
out += _writeVarint(len(val)) + val
|
||||
else:
|
||||
out += val if isinstance(val, bytes) else bytes(val)
|
||||
return bytes(out)
|
||||
|
||||
def _frame(msg):
|
||||
return b"\x00" + struct.pack(">I", len(msg)) + msg
|
||||
|
||||
def _unframe(data):
|
||||
# strict: exactly one uncompressed data frame (flag 0x00), nothing trailing (unary; no
|
||||
# compression/streaming/reserved flag bits)
|
||||
if len(data) < 5:
|
||||
raise ValueError("truncated gRPC-Web frame")
|
||||
flag = data[0] if isinstance(data[0], int) else ord(data[0])
|
||||
if flag != 0x00:
|
||||
raise ValueError("unsupported request frame flags 0x%02x" % flag)
|
||||
length = struct.unpack(">I", data[1:5])[0]
|
||||
if len(data) != 5 + length:
|
||||
raise ValueError("incomplete/oversized gRPC-Web frame")
|
||||
return data[5:5 + length]
|
||||
|
||||
def _isTextContentType(value):
|
||||
return (value or "").split(";", 1)[0].strip().lower() in (CONTENT_TYPE, CONTENT_TYPE_PROTO)
|
||||
|
||||
def acceptsTextContentType(value):
|
||||
# True if a (possibly comma-separated) Accept header already negotiates a grpc-web-text response
|
||||
return any(_isTextContentType(_) for _ in (value or "").split(","))
|
||||
|
||||
def _stringFields(fields):
|
||||
# Descriptorless: wire type 2 is string / bytes / embedded-message / packed - indistinguishable
|
||||
# without the .proto. Offer every printable-UTF-8 length-delimited field as a candidate; do NOT try
|
||||
# to exclude values that merely also parse as protobuf (ordinary strings like "A12345678"/"M1234" do,
|
||||
# so excluding them silently drops real injection points). Non-selected fields stay in the skeleton
|
||||
# untouched, so a mis-picked embedded message just fails to inject and is skipped - the safe failure.
|
||||
retVal = []
|
||||
for index, (fn, wt, val) in enumerate(fields):
|
||||
if wt != 2:
|
||||
continue
|
||||
try:
|
||||
value = bytes(val).decode("utf-8")
|
||||
except Exception:
|
||||
continue
|
||||
if all(char in "\t\n" or ord(char) > 31 for char in value):
|
||||
retVal.append((index, fn, value))
|
||||
return retVal
|
||||
|
||||
def _requestContentType():
|
||||
for header, value in (conf.httpHeaders or []):
|
||||
if header.lower() == HTTP_HEADER.CONTENT_TYPE.lower():
|
||||
return value
|
||||
return ""
|
||||
|
||||
def _headerValue(headers, key):
|
||||
if not headers:
|
||||
return None
|
||||
key = key.lower()
|
||||
try:
|
||||
items = headers.items()
|
||||
except AttributeError:
|
||||
items = headers
|
||||
for header, value in items:
|
||||
if header.lower() == key:
|
||||
return value
|
||||
return None
|
||||
|
||||
def decodeBody(data):
|
||||
"""
|
||||
Probe 'data' for a gRPC-Web (grpc-web-text) body WITHOUT any side effects. Returns a
|
||||
(jsonView, skeleton) tuple - the JSON string of injectable string fields plus the message skeleton
|
||||
to re-encode with - or (None, None). The caller commits the skeleton to kb.grpcWeb only on acceptance.
|
||||
"""
|
||||
|
||||
if not _isTextContentType(_requestContentType()):
|
||||
return None, None
|
||||
|
||||
try:
|
||||
fields = _decode(_unframe(_b64decode(data)))
|
||||
except Exception:
|
||||
return None, None
|
||||
|
||||
strings = _stringFields(fields)
|
||||
if not strings:
|
||||
return None, None
|
||||
|
||||
counts = {}
|
||||
for _, fn, _value in strings:
|
||||
counts[fn] = counts.get(fn, 0) + 1
|
||||
|
||||
occurrence = {}
|
||||
mapping = {}
|
||||
view = {}
|
||||
for index, fn, value in strings:
|
||||
if counts[fn] == 1:
|
||||
key = "f%d" % fn
|
||||
else:
|
||||
key = "f%d_%d" % (fn, occurrence.get(fn, 0))
|
||||
occurrence[fn] = occurrence.get(fn, 0) + 1
|
||||
mapping[key] = index
|
||||
view[key] = value
|
||||
|
||||
skeleton = {"fields": [list(_) for _ in fields], "map": mapping}
|
||||
|
||||
return json.dumps(view), skeleton
|
||||
|
||||
def encodeBody(jsonBody):
|
||||
"""
|
||||
Inverse of decodeBody(): overlay the (possibly injected) JSON string values onto the skeleton and
|
||||
re-encode a grpc-web-text (base64) body, recomputing the length prefixes. Called at send time.
|
||||
|
||||
Only a body whose keys are EXACTLY the gRPC surrogate keys is transformed - so unrelated JSON bodies
|
||||
on the shared request path (CSRF/second-order/redirect/safe requests) pass through untouched.
|
||||
"""
|
||||
|
||||
if not kb.grpcWeb:
|
||||
return jsonBody
|
||||
|
||||
try:
|
||||
parsed = json.loads(jsonBody)
|
||||
except Exception:
|
||||
return jsonBody
|
||||
|
||||
if not isinstance(parsed, dict) or set(parsed) != set(kb.grpcWeb["map"]):
|
||||
return jsonBody
|
||||
|
||||
fields = [list(_) for _ in kb.grpcWeb["fields"]]
|
||||
|
||||
for key, index in kb.grpcWeb["map"].items():
|
||||
value = parsed[key]
|
||||
fields[index][2] = (value if hasattr(value, "encode") else str(value)).encode("utf-8")
|
||||
|
||||
return base64.b64encode(_frame(_encode(fields))).decode("ascii")
|
||||
|
||||
def decodeResponse(page, responseHeaders=None):
|
||||
"""
|
||||
Render a gRPC-Web response as readable text (message-frame fields + the trailer frame's
|
||||
grpc-status/grpc-message = the back-end error) so the oracle/error-regex/in-band paths see content,
|
||||
not an opaque blob. Handles trailers-only errors (status/message in response headers, empty body).
|
||||
Unary only: at most one data frame, an optional final trailer, exact frame flags. Only a
|
||||
grpc-web-text response body is decoded; anything else is left unchanged.
|
||||
"""
|
||||
|
||||
if not kb.grpcWeb:
|
||||
return page
|
||||
|
||||
out = []
|
||||
|
||||
status = _headerValue(responseHeaders, "grpc-status")
|
||||
if status is not None:
|
||||
out.append("grpc-status:%s" % status)
|
||||
message = _headerValue(responseHeaders, "grpc-message")
|
||||
if message:
|
||||
out.append("grpc-message:%s" % unquote(message))
|
||||
|
||||
if page and _isTextContentType(_headerValue(responseHeaders, HTTP_HEADER.CONTENT_TYPE)):
|
||||
try:
|
||||
raw = _b64decode(page)
|
||||
bodyOut = [] # separate so a mid-parse failure never leaks partial frame renders
|
||||
pos = 0
|
||||
dataFrames = 0
|
||||
seenTrailer = False
|
||||
while pos + 5 <= len(raw):
|
||||
flag = raw[pos] if isinstance(raw[pos], int) else ord(raw[pos])
|
||||
length = struct.unpack(">I", raw[pos + 1:pos + 5])[0]
|
||||
payload, pos = _readExact(raw, pos + 5, length)
|
||||
if seenTrailer:
|
||||
raise ValueError("frame after trailer")
|
||||
if flag == 0x00: # data frame
|
||||
dataFrames += 1
|
||||
if dataFrames > 1:
|
||||
raise ValueError("streaming response not supported")
|
||||
for _fn, wt, val in _decode(payload):
|
||||
bodyOut.append(bytes(val).decode("utf-8", "replace") if wt == 2 else str(val))
|
||||
elif flag == 0x80: # trailer frame (grpc-status / grpc-message), must be last
|
||||
seenTrailer = True
|
||||
bodyOut.append(unquote(payload.decode("latin-1")))
|
||||
else:
|
||||
raise ValueError("unsupported response frame flags 0x%02x" % flag)
|
||||
if pos != len(raw):
|
||||
raise ValueError("trailing bytes after final frame")
|
||||
out.extend(bodyOut) # commit body renders only on a fully-valid parse
|
||||
except Exception:
|
||||
# keep status/message recovered from HEADERS (discard any partial body); append the raw page
|
||||
out.append(page)
|
||||
return "\n".join(out)
|
||||
elif page:
|
||||
out.append(page) # unsupported/absent response Content-Type: leave the body for the oracle
|
||||
|
||||
return "\n".join(out) if out else page
|
||||
268
tests/test_grpcweb.py
Normal file
268
tests/test_grpcweb.py
Normal file
|
|
@ -0,0 +1,268 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
"""
|
||||
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
||||
See the file 'LICENSE' for copying permission
|
||||
|
||||
gRPC-Web body support (lib/utils/grpcweb.py, grpc-web-text/unary). Covers the peer-review merge gate:
|
||||
exact round-trip, injection with length recomputation, strict framing validation (truncation / length
|
||||
mismatch / trailing bytes / compression / bad varint / field 0), repeated-field distinct injection
|
||||
points, response decoding incl. trailers-only-via-headers and response-Content-Type gating, and the
|
||||
side-effect-free probe that lets a declined prompt send the original body.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import struct
|
||||
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.data import kb
|
||||
from lib.utils import grpcweb
|
||||
|
||||
TEXT_CT = [("Content-Type", "application/grpc-web-text")]
|
||||
|
||||
|
||||
def _grpcBody(fields):
|
||||
return base64.b64encode(grpcweb._frame(grpcweb._encode(fields))).decode("ascii")
|
||||
|
||||
|
||||
class TestCodec(unittest.TestCase):
|
||||
def test_wire_roundtrip_exact(self):
|
||||
inner = grpcweb._encode([[1, 2, b"deep"], [2, 0, 7]])
|
||||
msg = grpcweb._encode([[1, 2, b"alice"], [2, 0, 10], [3, 2, inner], [4, 5, b"\x01\x02\x03\x04"]])
|
||||
self.assertEqual(grpcweb._encode(grpcweb._decode(msg)), msg)
|
||||
|
||||
def test_varint_boundaries(self):
|
||||
for n in (0, 1, 127, 128, 300, 16384, 2 ** 31, 2 ** 63):
|
||||
self.assertEqual(grpcweb._readVarint(grpcweb._writeVarint(n), 0)[0], n)
|
||||
|
||||
def test_overlong_varint_rejected(self):
|
||||
self.assertRaises(ValueError, grpcweb._readVarint, b"\x80" * 11, 0)
|
||||
|
||||
def test_field_zero_rejected(self):
|
||||
self.assertRaises(ValueError, grpcweb._decode, b"\x02\x01a") # tag 0x02 -> field 0, wire 2
|
||||
|
||||
def test_truncated_field_rejected(self):
|
||||
self.assertRaises(ValueError, grpcweb._decode, b"\x0a\x05abc") # declares 5, has 3
|
||||
|
||||
def test_unframe_strict(self):
|
||||
good = grpcweb._frame(b"\x08\x01")
|
||||
self.assertEqual(grpcweb._unframe(good), b"\x08\x01")
|
||||
self.assertRaises(ValueError, grpcweb._unframe, good[:4]) # truncated header
|
||||
self.assertRaises(ValueError, grpcweb._unframe, good + b"\x00") # trailing bytes
|
||||
self.assertRaises(ValueError, grpcweb._unframe, good[:-1]) # declared > available
|
||||
self.assertRaises(ValueError, grpcweb._unframe, b"\x01" + good[1:]) # compressed flag
|
||||
|
||||
|
||||
class TestTranscode(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._h, self._g = conf.httpHeaders, kb.get("grpcWeb")
|
||||
conf.httpHeaders = list(TEXT_CT)
|
||||
kb.grpcWeb = None
|
||||
|
||||
def tearDown(self):
|
||||
conf.httpHeaders, kb.grpcWeb = self._h, self._g
|
||||
|
||||
def test_decode_is_side_effect_free(self):
|
||||
# probe must NOT touch kb.grpcWeb (that is what lets a declined prompt restore the original)
|
||||
view, skeleton = grpcweb.decodeBody(_grpcBody([[1, 2, b"alice"], [2, 0, 10]]))
|
||||
self.assertEqual(json.loads(view), {"f1": "alice"})
|
||||
self.assertIsNotNone(skeleton)
|
||||
self.assertIsNone(kb.grpcWeb)
|
||||
|
||||
def test_exact_roundtrip_no_injection(self):
|
||||
body = _grpcBody([[1, 2, b"alice"], [2, 0, 10]])
|
||||
view, skeleton = grpcweb.decodeBody(body)
|
||||
kb.grpcWeb = skeleton
|
||||
self.assertEqual(grpcweb.encodeBody(view), body) # merge-gate #11: encodeBody(decodeBody(x)) == x
|
||||
|
||||
def test_injection_reencodes_with_correct_length(self):
|
||||
view, skeleton = grpcweb.decodeBody(_grpcBody([[1, 2, b"alice"], [2, 0, 10]]))
|
||||
kb.grpcWeb = skeleton
|
||||
wire = grpcweb.encodeBody(json.dumps({"f1": "alice' OR '1'='1"}))
|
||||
fields = grpcweb._decode(grpcweb._unframe(base64.b64decode(wire)))
|
||||
self.assertEqual(bytes(fields[0][2]).decode(), "alice' OR '1'='1")
|
||||
self.assertEqual(fields[1][2], 10) # non-string field preserved
|
||||
|
||||
def test_repeated_fields_distinct_points(self):
|
||||
view, _ = grpcweb.decodeBody(_grpcBody([[3, 2, b"first"], [3, 2, b"second"]]))
|
||||
self.assertEqual(json.loads(view), {"f3_0": "first", "f3_1": "second"})
|
||||
|
||||
def test_parseable_strings_are_offered(self):
|
||||
# ordinary strings that ALSO happen to parse as protobuf wire data must NOT be silently dropped
|
||||
# ("A12345678" -> fixed64 tag, "M1234" -> fixed32 tag); descriptorless can't tell, so offer them
|
||||
view, _ = grpcweb.decodeBody(_grpcBody([[1, 2, b"A12345678"], [2, 2, b"M1234"]]))
|
||||
self.assertEqual(json.loads(view), {"f1": "A12345678", "f2": "M1234"})
|
||||
|
||||
def test_non_grpc_and_binary_not_detected(self):
|
||||
conf.httpHeaders = [("Content-Type", "application/json")]
|
||||
self.assertEqual(grpcweb.decodeBody('{"a":"b"}'), (None, None))
|
||||
conf.httpHeaders = [("Content-Type", "application/grpc-web+proto")] # binary: deliberately out of scope
|
||||
self.assertEqual(grpcweb.decodeBody(_grpcBody([[1, 2, b"x"]])), (None, None))
|
||||
|
||||
def test_encode_guards_bad_surrogate(self):
|
||||
_, skeleton = grpcweb.decodeBody(_grpcBody([[1, 2, b"alice"]]))
|
||||
kb.grpcWeb = skeleton
|
||||
self.assertEqual(grpcweb.encodeBody("[1,2,3]"), "[1,2,3]") # JSON scalar/array -> unchanged
|
||||
self.assertEqual(grpcweb.encodeBody("not json"), "not json")
|
||||
|
||||
|
||||
class TestResponse(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._g = kb.get("grpcWeb")
|
||||
kb.grpcWeb = {"fields": [], "map": {}} # any truthy skeleton enables response decoding
|
||||
|
||||
def tearDown(self):
|
||||
kb.grpcWeb = self._g
|
||||
|
||||
def _resp(self, frames, ct="application/grpc-web-text", extra=None):
|
||||
page = base64.b64encode(b"".join(frames)).decode("ascii") if frames else ""
|
||||
headers = {"Content-Type": ct}
|
||||
if extra:
|
||||
headers.update(extra)
|
||||
return grpcweb.decodeResponse(page, headers)
|
||||
|
||||
def test_message_and_trailer_rendered(self):
|
||||
msg = grpcweb._frame(grpcweb._encode([[1, 0, 3], [2, 2, b"luther"]]))
|
||||
trailer = b"\x80" + struct.pack(">I", len(b"grpc-status:0")) + b"grpc-status:0"
|
||||
decoded = self._resp([msg, trailer])
|
||||
self.assertIn("3", decoded)
|
||||
self.assertIn("luther", decoded)
|
||||
self.assertIn("grpc-status:0", decoded)
|
||||
|
||||
def test_backend_error_in_trailer(self):
|
||||
tmsg = b"grpc-status:13\r\ngrpc-message:SQLite%20error%3A%20near%20syntax"
|
||||
decoded = self._resp([b"\x80" + struct.pack(">I", len(tmsg)) + tmsg])
|
||||
self.assertIn("SQLite error: near syntax", decoded) # unquoted -> matchable by errors.xml
|
||||
|
||||
def test_trailers_only_via_headers(self):
|
||||
# empty body, status/message carried in response HEADERS (protocol-allowed trailers-only)
|
||||
decoded = grpcweb.decodeResponse("", {"Content-Type": "application/grpc-web-text",
|
||||
"grpc-status": "13", "grpc-message": "boom%20here"})
|
||||
self.assertIn("grpc-status:13", decoded)
|
||||
self.assertIn("boom here", decoded)
|
||||
|
||||
def test_response_content_type_gating(self):
|
||||
# a non-grpc-web response (e.g. an HTML error page) must be left untouched
|
||||
page = base64.b64encode(grpcweb._frame(b"\x08\x01")).decode("ascii")
|
||||
self.assertEqual(grpcweb.decodeResponse(page, {"Content-Type": "text/html"}), page)
|
||||
|
||||
def test_compressed_response_frame_falls_back(self):
|
||||
page = base64.b64encode(b"\x01" + struct.pack(">I", 2) + b"\x08\x01").decode("ascii")
|
||||
self.assertEqual(grpcweb.decodeResponse(page, {"Content-Type": "application/grpc-web-text"}), page)
|
||||
|
||||
|
||||
class TestReviewRound2(unittest.TestCase):
|
||||
"""The three second-round blockers + smaller hardening."""
|
||||
|
||||
def setUp(self):
|
||||
self._h, self._g = conf.httpHeaders, kb.get("grpcWeb")
|
||||
conf.httpHeaders = list(TEXT_CT)
|
||||
kb.grpcWeb = None
|
||||
|
||||
def tearDown(self):
|
||||
conf.httpHeaders, kb.grpcWeb = self._h, self._g
|
||||
|
||||
def test_unrelated_json_body_not_converted(self):
|
||||
# blocker #1: an unrelated JSON body on the shared request path must pass through untouched
|
||||
_, skeleton = grpcweb.decodeBody(_grpcBody([[1, 2, b"alice"]]))
|
||||
kb.grpcWeb = skeleton
|
||||
self.assertEqual(grpcweb.encodeBody('{"foo":"bar"}'), '{"foo":"bar"}') # different keys
|
||||
self.assertEqual(grpcweb.encodeBody('{"f1":"x","extra":"y"}'), '{"f1":"x","extra":"y"}') # superset
|
||||
# but the genuine surrogate (exact keys) IS transformed
|
||||
self.assertNotEqual(grpcweb.encodeBody('{"f1":"x"}'), '{"f1":"x"}')
|
||||
|
||||
def test_streaming_response_rejected(self):
|
||||
kb.grpcWeb = {"fields": [], "map": {}}
|
||||
two = grpcweb._frame(grpcweb._encode([[1, 2, b"a"]])) + grpcweb._frame(grpcweb._encode([[1, 2, b"b"]]))
|
||||
page = base64.b64encode(two).decode("ascii")
|
||||
self.assertEqual(grpcweb.decodeResponse(page, {"Content-Type": TEXT_CT[0][1]}), page) # falls back, no corruption
|
||||
|
||||
def test_reserved_frame_flags_rejected(self):
|
||||
# request: reserved flag byte -> not a valid gRPC-Web frame -> not detected
|
||||
bad = b"\x02" + struct.pack(">I", 3) + grpcweb._encode([[1, 2, b"x"]])
|
||||
self.assertRaises(ValueError, grpcweb._unframe, bad)
|
||||
# response: 0x82 (trailer + reserved bit) rejected -> fall back
|
||||
kb.grpcWeb = {"fields": [], "map": {}}
|
||||
page = base64.b64encode(b"\x82" + struct.pack(">I", 3) + b"a=0").decode("ascii")
|
||||
self.assertEqual(grpcweb.decodeResponse(page, {"Content-Type": TEXT_CT[0][1]}), page)
|
||||
|
||||
def test_strict_base64(self):
|
||||
self.assertRaises(ValueError, grpcweb._b64decode, "!!!!") # invalid chars
|
||||
self.assertRaises(ValueError, grpcweb._b64decode, "AAA") # bad length
|
||||
self.assertRaises(ValueError, grpcweb._b64decode, "AAAA=BCD") # stray mid-quantum pad
|
||||
# independently-padded chunks ARE accepted (reconstruct the concatenation) - a unary text
|
||||
# response may legitimately be flushed as separate padded base64 segments
|
||||
a, b = b"hello", b"world!!"
|
||||
chunks = base64.b64encode(a).decode("ascii") + base64.b64encode(b).decode("ascii")
|
||||
self.assertEqual(grpcweb._b64decode(chunks), a + b)
|
||||
|
||||
def test_strict_media_type(self):
|
||||
conf.httpHeaders = [("Content-Type", "application/grpc-web-textual")] # not the real type
|
||||
self.assertEqual(grpcweb.decodeBody(_grpcBody([[1, 2, b"x"]])), (None, None))
|
||||
conf.httpHeaders = [("Content-Type", "application/grpc-web-text; charset=utf-8")] # params OK
|
||||
view, _ = grpcweb.decodeBody(_grpcBody([[1, 2, b"x"]]))
|
||||
self.assertIsNotNone(view)
|
||||
|
||||
def test_header_status_preserved_on_body_failure(self):
|
||||
kb.grpcWeb = {"fields": [], "map": {}}
|
||||
raw = b"\x00" + struct.pack(">I", 5) + b"ab" # declares 5, has 2 -> body parse fails
|
||||
page = base64.b64encode(raw).decode("ascii")
|
||||
decoded = grpcweb.decodeResponse(page, {"Content-Type": TEXT_CT[0][1], "grpc-status": "13"})
|
||||
self.assertIn("grpc-status:13", decoded) # header status not lost
|
||||
self.assertIn(page, decoded) # raw page still available to the oracle
|
||||
|
||||
def test_varint_and_field_number_bounds(self):
|
||||
self.assertRaises(ValueError, grpcweb._readVarint, b"\xff" * 9 + b"\x02", 0) # > 64 bits
|
||||
# field number above the protobuf max (2**29 - 1)
|
||||
big = grpcweb._writeVarint(((0x1fffffff + 1) << 3) | 2) + b"\x01a"
|
||||
self.assertRaises(ValueError, grpcweb._decode, big)
|
||||
|
||||
|
||||
class TestReviewRound3(unittest.TestCase):
|
||||
"""Media-type +proto acceptance, Accept negotiation helper, unsupported-CT body preservation."""
|
||||
|
||||
def setUp(self):
|
||||
self._h, self._g = conf.httpHeaders, kb.get("grpcWeb")
|
||||
kb.grpcWeb = None
|
||||
|
||||
def tearDown(self):
|
||||
conf.httpHeaders, kb.grpcWeb = self._h, self._g
|
||||
|
||||
def test_proto_media_type_accepted(self):
|
||||
for ct in ("application/grpc-web-text+proto", "application/grpc-web-text+proto; charset=utf-8"):
|
||||
conf.httpHeaders = [("Content-Type", ct)]
|
||||
view, _ = grpcweb.decodeBody(_grpcBody([[1, 2, b"alice"]]))
|
||||
self.assertEqual(json.loads(view), {"f1": "alice"}, "CT %r not accepted" % ct)
|
||||
|
||||
def test_accepts_text_content_type_helper(self):
|
||||
self.assertFalse(grpcweb.acceptsTextContentType("*/*"))
|
||||
self.assertFalse(grpcweb.acceptsTextContentType("application/json"))
|
||||
self.assertTrue(grpcweb.acceptsTextContentType("application/grpc-web-text"))
|
||||
self.assertTrue(grpcweb.acceptsTextContentType("application/json, application/grpc-web-text+proto"))
|
||||
|
||||
def test_unsupported_response_ct_preserves_body_and_status(self):
|
||||
kb.grpcWeb = {"fields": [], "map": {}}
|
||||
page = base64.b64encode(grpcweb._frame(b"\x08\x01")).decode("ascii")
|
||||
decoded = grpcweb.decodeResponse(page, {"Content-Type": "text/html", "grpc-status": "2"})
|
||||
self.assertIn("grpc-status:2", decoded) # header status kept
|
||||
self.assertIn(page, decoded) # body not dropped
|
||||
# and with no status, an unsupported-CT body is returned unchanged
|
||||
self.assertEqual(grpcweb.decodeResponse(page, {"Content-Type": "text/html"}), page)
|
||||
|
||||
def test_proto_response_ct_decoded(self):
|
||||
kb.grpcWeb = {"fields": [], "map": {}}
|
||||
msg = grpcweb._frame(grpcweb._encode([[2, 2, b"luther"]]))
|
||||
page = base64.b64encode(msg).decode("ascii")
|
||||
self.assertIn("luther", grpcweb.decodeResponse(page, {"Content-Type": "application/grpc-web-text+proto"}))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Loading…
Add table
Add a link
Reference in a new issue