From 2e3e4385ca569055fb4cecc8626c15c20fd0d85e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Mon, 28 Sep 2026 00:38:03 +0200 Subject: [PATCH] Added shared Interactsh poller with DNS/HTTP OOB support --- lib/core/settings.py | 2 +- lib/request/dns.py | 77 ++++-- lib/request/interactsh.py | 390 ++++++++++++++++++++++++++++- tests/test_dns_server.py | 511 ++++++++++++++++++++++++++++++++++++-- 4 files changed, 936 insertions(+), 44 deletions(-) diff --git a/lib/core/settings.py b/lib/core/settings.py index 8699ac548..1afbb74c2 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.9.25" +VERSION = "1.10.9.26" 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) diff --git a/lib/request/dns.py b/lib/request/dns.py index 8dc1f0514..3b4a15b0d 100644 --- a/lib/request/dns.py +++ b/lib/request/dns.py @@ -235,8 +235,9 @@ class InteractshDNSServer(object): is a drop-in for conf.dnsServer. """ - _POLL_TRIES = 6 # a triggered lookup surfaces at interactsh within a couple of seconds; - _POLL_DELAY = 1.0 # poll up to ~6s per retrieval before treating the channel as silent + # Default time budget (seconds) pop() spends waiting on the poller queue for a + # not-yet-arriving captured lookup. Overridable per instance for fast tests. + _POP_DEADLINE = 6 def __init__(self, server=None): from lib.request.interactsh import Interactsh, hasCrypto @@ -249,42 +250,74 @@ class InteractshDNSServer(object): if not self._client.registered: raise socket.error("could not register with an interactsh interaction server") + # One background poller drains the remote session; pop() consumes from its in-memory + # queue with zero network calls. + self._client._ensurePoller() self.domain = self._client.dnsDomain() self._seen = set() self._running = True self._initialized = True + def stop(self): + """Best-effort shutdown: stop the single background poller (join it) then deregister. + + stop() also deregisters, and the poller is the only code that drains the remote + session, so it must be stopped before deregistering.""" + self._running = False + client = getattr(self, "_client", None) + if client is not None: + client.stop() + def run(self): """No background listener is needed - interactsh does the receiving.""" pass - def pop(self, prefix=None, suffix=None): + def _matchesDns(self, record, prefix, suffix): + # Protocol gate: the shared poller queues dns/http/https together, so a record can land in + # the queue for another consumer. Never hand an HTTP record to the DNS channel - doing so + # would make pop() return a request line as though it were a DNS lookup, and the HTTP + # consumer would then lose that interaction (the exact cross-consumer loss the shared queue + # was introduced to prevent). + if record.get("protocol") != "dns": + return False + name = record.get("full-id") + if not name: + return False + if name in self._seen: # already handed out once, never return it again + return False + if prefix is None and suffix is None: + return True + if prefix and suffix and re.search( + r"%s\..+\.%s" % (re.escape(prefix), re.escape(suffix)), name, re.I): + return True + return False + + def pop(self, prefix=None, suffix=None, deadline=None, sync=False): """ Returns a captured DNS lookup name matching the given prefix/suffix (prefix..suffix.), mirroring DNSServer.pop(). Unlike the synchronous local DNSServer (which reads a query captured during the very request), interactsh is remote and eventually-consistent: a just-triggered - lookup takes a moment to reach the collector and surface via its poll API. So we - poll a few times before giving up, instead of reading once. + lookup takes a moment to reach the collector and surface via its poll API. The + background poller stages matching names in a queue, so pop() waits on that queue + (no network call) until the ~6s deadline, then returns None. + + ``deadline`` is the number of seconds to wait for a not-yet-arriving lookup + (``deadline=0`` is a snapshot of the current queue with no wait); ``None`` selects the + class default (``_POP_DEADLINE``). This None semantics differs from ``_consume``, where + ``None`` means "don't wait" - here it means "wait the default". It is resolved here to + an absolute timestamp for the poller's Condition. ``sync`` forces the direct-poll + fallback path. """ - - for attempt in range(self._POLL_TRIES): - for name in self._client.dnsNames(): - if name in self._seen: - continue - - if prefix is None and suffix is None: - self._seen.add(name) - return name - - if prefix and suffix and re.search(r"%s\..+\.%s" % (re.escape(prefix), re.escape(suffix)), name, re.I): - self._seen.add(name) - return name - - if attempt < self._POLL_TRIES - 1: - time.sleep(self._POLL_DELAY) - + selector = lambda record: self._matchesDns(record, prefix, suffix) + # deadline=None -> use the class default budget; a number -> that many seconds. + budget = self._POP_DEADLINE if deadline is None else deadline + matched = self._client._consume(selector, deadline=time.time() + budget, sync=sync) + if matched: + matched_id = matched[0].get("full-id") + self._seen.add(matched_id) + return matched_id return None if __name__ == "__main__": diff --git a/lib/request/interactsh.py b/lib/request/interactsh.py index e57a09537..e81fc8e00 100644 --- a/lib/request/interactsh.py +++ b/lib/request/interactsh.py @@ -6,8 +6,11 @@ See the file 'LICENSE' for copying permission """ import base64 +import collections +import threading import json import time +import hashlib from lib.core.common import randomStr from lib.core.convert import getBytes @@ -51,6 +54,13 @@ class Interactsh(object): self.registered = False self._key = None self._dnsNonce = None + self._poller = None + # Guards the poller create/start decision so two callers racing on an unstarted + # session cannot each build their own poller and destructively poll one session. + self._pollerLock = threading.Lock() + # Set under _pollerLock while stop() tears down: blocks _ensurePoller from restarting the + # poller in the gap between stop() and close()/deregister(). + self._stopping = False if not _HAS_CRYPTO: return @@ -102,12 +112,20 @@ class Interactsh(object): handlers.append(ProxyHandler({"http": conf.proxy, "https": conf.proxy})) request = _Request(url, data=getBytes(post) if post is not None else None, headers=headers) - response = build_opener(*handlers).open(request, timeout=conf.get("timeout") or 30) + # request_timeout() is the single source of truth for every interactsh network op so + # InteractshPoller.stop() can join for exactly one /poll network call. + response = build_opener(*handlers).open(request, timeout=self.request_timeout()) return getText(response.read()) except Exception as ex: logger.debug("OOB request to '%s' failed: %s" % (url, getText(ex))) return None + def request_timeout(self): + """Network timeout (seconds) applied to every /poll and /deregister request. Interactsh + derives its /poll network timeout identically, so InteractshPoller.stop() can join for + exactly one of these before the thread is guaranteed to have exited.""" + return conf.get("timeout") or _DEFAULT_NETWORK_TIMEOUT + def url(self): """Return a fresh unique callback URL (host = correlationId + nonce).""" nonce = randomStr(OOB_NONCE_LENGTH, lowercase=True) @@ -122,9 +140,120 @@ class Interactsh(object): return "%s%s.%s" % (self.correlationId, self._dnsNonce, self.server) def dnsNames(self): - """Poll and return the fully-qualified names (minus the server suffix) of the - DNS lookups captured so far, e.g. 'prefix..suffix.'.""" - return [_.get("full-id") for _ in self.poll() if _.get("protocol") == "dns" and _.get("full-id")] + """Return the fully-qualified names (minus the server suffix) of the DNS lookups + captured so far, e.g. 'prefix..suffix.'. + + This is a *snapshot* over the single background poller's in-memory queue: it ensures the + poller is running and drains whatever it has already staged, so it does not itself issue + a fresh /poll. On the very first call after registration the poller may not have finished + its first round-trip yet and can therefore return [] even though interactions exist - + retry the call rather than treating that first call as authoritative.""" + return [_.get("full-id") + for _ in self._consume( + lambda _: _.get("protocol") == "dns" and _.get("full-id"), + None)] + + def httpRequests(self): + """Return the captured HTTP interactions (collector half of an HTTP-based out-of-band + data channel). Unlike DNS, where the value rides the queried hostname, an HTTP callback + carries its payload in the request line/body - so the interaction itself is the evidence. + + interactsh stores each record with a ``raw-request`` (the full dumped request), a + ``protocol`` (``http`` or ``https``) and a ``remote-address`` - it does not pre-parse the + path/query. A value with no safe-in-URL character set is therefore embedded in a + query-string *parameter name* (the path stays a constant request line and only the + attacker-controlled name varies) and the decoded names are returned here. + """ + retVal = [] + + for _ in self._consume(lambda _: _.get("protocol") in ("http", "https"), None): + retVal.append(self._renderHttpRequest(_)) + + return retVal + + def _renderHttpRequest(self, record): + interaction = { + "full-id": record.get("full-id") or "", + "client": record.get("remote-address") or record.get("client") or "", + "request": record.get("raw-request") or "", + "query-param-names": [], + } + + raw = interaction["request"] + target = _parseHttpRequestTarget(raw) if raw else (record.get("full-id") or "") + if target: + interaction["query-param-names"] = _extractQueryStringNames(target) + + return interaction + + def _ensurePoller(self): + """Start the single background /poll consumer (idempotent). The poller is the only code + that calls the remote endpoint; DNS/HTTP consumers then read the same records from + memory, so they never drain each other. No-op without crypto, since /poll cannot then be + decrypted.""" + if not _HAS_CRYPTO: + return + + # The whole create-or-reuse-and-start sequence is under one lock, so exactly one + # background poller is ever created and started for this session. A poller must only run + # for a live, registered session: skip if the session was never registered, and skip if + # stop() is in progress (so no poller can be restarted between stop() and deregister()). + with self._pollerLock: + if not self.registered or self._stopping: + return + poller = getattr(self, "_poller", None) + if poller is None: + self._poller = poller = InteractshPoller(self) + if not poller.isRunning(): + poller.start() + + def _consume(self, selector, deadline=None, sync=False): + """Return records matching selector. + + For normal registered operation the single background poller drains the remote session + (zero extra network calls from the consumer, and exactly one caller ever drains it), so + we ensure it is running and read its in-memory queue. ``sync=True`` forces the direct + ``poll()`` path for the deprecated fallback / explicit testing. + + ``deadline`` is an absolute timestamp: ``None`` drains the current queue only (no wait), + used by snapshot-style callers such as ``httpRequests``/``dnsNames``. + + Outside ``sync=True`` only the background poller ever calls the remote ``poll()`` - the + invariant this method enforces. A stale poller (never started, or started then stopped by + ``stop()``) yields ``[]`` rather than a second, unsynchronised ``poll()`` that could + destructively drain the shared session from the wrong caller. + + ``sync=True`` forces the direct-poll path, but only for a live, registered session that + has no background poller already draining it - the poller is the only code that is allowed + to call the remote endpoint, so a spurious ``sync=True`` during normal operation drains the + running poller's queue instead of adding a second destructive ``poll()`` caller. Once + ``stop()`` has begun, no caller initiates another remote ``poll()`` - not this + compatibility path either.""" + if sync: + with self._pollerLock: + if self._stopping or not self.registered: + return [] + poller = getattr(self, "_poller", None) + if poller is None or not poller.isRunning(): + # Keep the lock across this compatibility poll so _ensurePoller() cannot start + # a background poller concurrently in the gap between the check and poll() + # (which would then become a second /poll caller). + return [record for record in self.poll() if self._safeSelector(selector, record)] + # Releasing the lock before draining the poller's queue keeps _ensurePoller() free too. + return poller.drain_matching(selector, deadline) + + self._ensurePoller() + poller = getattr(self, "_poller", None) + if poller is not None and poller.isRunning(): + return poller.drain_matching(selector, deadline) + return [] + + def _safeSelector(self, selector, record): + try: + return bool(selector(record)) + except Exception as ex: + logger.debug("interactsh selector error: %s" % getText(ex)) + return False def poll(self): """Return the list of decrypted interaction records captured so far.""" @@ -160,16 +289,263 @@ class Interactsh(object): return retVal def pollUntil(self, attempts, delay): - """Poll repeatedly, returning as soon as any interaction is captured.""" + """Wait for and return captured interactions, up to attempts * delay seconds. + + The background poller remains the sole remote /poll consumer. Each attempt waits on its + in-memory queue for at most ``delay`` seconds instead of issuing an independent destructive + poll() request. + """ for _ in range(attempts): - time.sleep(delay) - interactions = self.poll() + interactions = self._consume(lambda record: True, deadline=time.time() + delay) if interactions: return interactions return [] + def stop(self): + """Stop the background poller (join it) then deregister the interaction. + + The poller is the only code that drains the remote /poll session, so it must be + stopped before deregistering; stopping it first also prevents it from re-staging + interactions after close() is called (best-effort on a dead/broken poller). + + Normal completion leaves no active polling thread. If the poller cannot be joined within + one request-timeout budget, this interaction is deliberately left registered so deregister + never races a still-active /poll drain; a later stop()/close() retries the join and + deregisters once the thread has exited.""" + # Set _stopping and capture the poller under _pollerLock (the start/stop race lives under + # that lock); release the lock before joining so a concurrent _ensurePoller() caller is + # not blocked for a whole join. The lock prevents a restart of this same poller between here + # and deregister(); only the (now-dead) poller is joined. + poller = None + with self._pollerLock: + self._stopping = True + poller = getattr(self, "_poller", None) + stopped = True + if poller is not None: + try: + stopped = poller.stop() + except Exception: + logger.debug("error while stopping interactsh poller") + stopped = False + if not stopped: + # A /poll network call is still draining the session; skip deregister this time and let + # the next stop()/close() deregister once the thread has actually exited. + return + self._deregister() + def close(self): + """Canonical complete teardown: stop the poller AND deregister. + + close() predates the background-poller addition, so existing callers can keep using it + without learning about stop(). If the current /poll cannot be joined within its bounded + network-timeout budget, teardown remains pending and the interaction stays registered so + a later close()/stop() can retry safely.""" + self.stop() + + def _deregister(self): if self.registered: body = json.dumps({"correlation-id": self.correlationId, "secret-key": self.secret}) self._request("https://%s/deregister" % self.server, post=body) self.registered = False + + +# Delay between two remote /poll round-trips in the background poller. +_POP_POLL_DELAY = 1.0 + +# Network timeout (seconds) applied to each interactsh /poll and /deregister request (see +# Interactsh._request). InteractshPoller.stop() joins for one of these (with a little slack) so a +# /poll wedged at the remote endpoint cannot keep a dead poller's thread alive past stop(). +_DEFAULT_NETWORK_TIMEOUT = 30 + + +def _parseHttpRequestTarget(raw_request): + """Return the request target (undecoded) from the first line of a raw HTTP request. + + It is deliberately left raw: decoding before splitting on '&'/'=' would corrupt encoded + delimiters (a value with an unsafe-in-URL character set may hit the same host as its own + name, e.g. ``?aa%26bb=1`` must parse to one name ``aa&bb``, not two). Downstream code decodes + each name individually. + """ + if not raw_request: + return "" + first_line = raw_request.splitlines()[0] + parts = first_line.split(None, 2) + if len(parts) >= 2: + target = parts[1] + else: + target = first_line + return target + + +def _unquote(value): + try: + from urllib import unquote + except ImportError: + from urllib.parse import unquote + return unquote(value) + + +def _extractQueryStringNames(raw_target): + """Return the ordered, de-duplicated query-string parameter names from a raw (undecoded) target. + + Split the raw target on literal '&' and the first raw '=', then percent-decode only each + parameter name individually so encoded delimiters (``%26`` '&', ``%3D`` '=', ``%25`` '%') are + preserved in the decoded name rather than misinterpreted as structure. + """ + try: + query = raw_target.split("?", 1)[1] + except IndexError: + return [] + names = [] + for pair in query.split("&"): + if not pair: + continue + encoded_name = pair.split("=", 1)[0] + name = _unquote(encoded_name) + if name and name not in names: + names.append(name) + return names + + +class InteractshPoller(object): + """A single background consumer of one interactsh session: one thread polls /poll and queues + complete interaction records under a Condition, so the DNS and HTTP consumers read the same + records from memory without ever calling the remote endpoint twice or draining each other.""" + + def __init__(self, client): + self._client = client + self._queue = collections.deque() + self._lock = threading.RLock() + self._cond = threading.Condition(self._lock) + self._seen = set() + self._running = False + self._thread = None + + def start(self): + """Start the poller (idempotent).""" + with self._lock: + if self._running: + return + self._running = True + self._thread = threading.Thread(target=self._loop) + self._thread.daemon = True + self._thread.start() + + def stop(self): + """Stop the poller and join the thread until it actually exits. + + Returns True if the poll thread has exited, False if it could not be stopped within the + budget (its in-flight /poll request is still running). stop() flips _running and wakes the + loop, so the thread exits after its current /poll returns - join therefore only has to + outlast one /poll network operation, whose timeout is derived from the client's request + timeout. Joining is retried on every call, so a second stop() after a failed first still + joins the (now-sleeping-loop) thread instead of bailing out before _running was read. + + Only a genuinely hung /poll network call exceeds the budget; the common case exits as soon + as the loop wakes. On timeout this method returns False and the owner leaves the interaction + registered; a later stop() retries joining the same thread.""" + with self._lock: + self._running = False + self._cond.notify_all() + thread = self._thread + if not thread or not thread.is_alive(): + return True + # stop() has already woken the loop, so the thread exits after its current /poll returns. + # One join that outlasts one /poll network call is enough; the slack covers the loop's own + # work before it re-checks _running. A hung request is guarded by retrying stop(). + fn = getattr(self._client, "request_timeout", None) + network_timeout = fn() if callable(fn) else _DEFAULT_NETWORK_TIMEOUT + thread.join(network_timeout + _POP_POLL_DELAY) + return not thread.is_alive() + + def isRunning(self): + return self._running and self._thread is not None and self._thread.is_alive() + + def _loop(self): + while self._running: + try: + records = self._client.poll() + with self._cond: + # Recheck _running after poll() returns: stop() may have set it False while the + # network call was in flight. Drop the just-received records without staging + # them, so a stopped poller can never repopulate its queue after shutdown. + if not self._running: + break + for record in records: + # Only queue interaction protocols sqlmap consumes (dns/http/https). Other + # interaction types (smtp/ldap/...) would never be drained by sqlmap's + # consumers and would otherwise accumulate in _queue for the whole session; + # bounding the queue to what we actually use keeps it from growing. + if record.get("protocol") not in ("dns", "http", "https"): + continue + # Fingerprint the *interaction*, not the payload host ("full-id"). + # protocol + full-id + unique-id only identify the receiving subdomain + # (a single host drives many interactions); timestamp is a separate, per- + # interaction field that keeps two identical requests to that host distinct. + # Per-record try: a malformed interaction (a non-string field making the + # join raise, or a bytes value getBytes() cannot decode) throws here. Keeping + # it inside the per-record handler means one odd interaction skips itself + # without aborting every later interaction in the already-drained batch, which + # the outer batch-level except would otherwise do. + try: + rid_str = "|".join((record.get("protocol") or "", + record.get("full-id") or "", + record.get("unique-id") or "", + record.get("timestamp") or "", + record.get("raw-request") or "", + record.get("remote-address") or "")) + # Hash before storing so _seen never retains a full raw-request (bounded + # to a fixed-length hexdigest); the review noted raw-request retention + # grew _seen for the whole session. Hashlib is only consulted when we + # have a record to fingerprint. + rid = hashlib.sha256(getBytes(rid_str)).hexdigest() + except Exception as ex: + logger.debug("interactsh fingerprint error: %s" % getText(ex)) + continue + if rid in self._seen: + continue + self._seen.add(rid) + self._queue.append(record) + self._cond.notify_all() + except Exception as ex: + logger.debug("interactsh poller poll error: %s" % getText(ex)) + if self._running: + time.sleep(_POP_POLL_DELAY) + + def _safeSelector(self, selector, record): + try: + return bool(selector(record)) + except Exception as ex: + logger.debug("interactsh selector error: %s" % getText(ex)) + return False + + def drain_matching(self, selector, deadline=None): + """Return all currently-queued records matching selector (removing them). + + ``deadline=None`` drains the current queue with no wait (snapshot-style callers such as + ``httpRequests``/``dnsNames``); only ``InteractshDNSServer.pop`` passes an absolute + deadline to wait for new arrivals. The wait and the scan share one condition critical + section so a producer cannot enqueue+notify between releasing and reacquiring the lock, + which would otherwise stall another 0.5 s even when a match already exists.""" + while True: + with self._cond: + matched = [record for record in self._queue if self._safeSelector(selector, record)] + if matched: + for record in matched: + self._queue.remove(record) + return matched + # Check _running before the first wait as well: stop() may have fired between the + # scan and a wait() call, so the matching notification has already happened - a bare + # wait() would otherwise sleep the whole budget for nothing. + if not self._running: + return [] + if deadline is None: + return [] + remaining = deadline - time.time() + if remaining <= 0: + return [] + self._cond.wait(timeout=remaining) + # Wake-up was not for a match; if stop() fired in the meantime no producer can + # ever enqueue one, so exit promptly instead of looping for the full deadline. + if not self._running: + return [] diff --git a/tests/test_dns_server.py b/tests/test_dns_server.py index 918e54659..ee549bd65 100644 --- a/tests/test_dns_server.py +++ b/tests/test_dns_server.py @@ -17,6 +17,7 @@ import struct import sys import threading import time +import types import unittest sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -100,6 +101,17 @@ def _wait_popped(srv, prefix, suffix, timeout=WAIT_TIMEOUT): return None +def _stop_client_poller(client): + """Best-effort shutdown of a client's single background poller, used by stubbed close() + so test teardown never joins a live thread or reaches the network.""" + poller = getattr(client, "_poller", None) + if poller is not None: + try: + poller.stop() + except Exception: + pass + + class _SendFailOnceSocket(object): """Wraps a real UDP socket; first sendto() raises (simulated transient failure)""" def __init__(self, real): @@ -322,27 +334,64 @@ class TestDNSServerConcurrency(unittest.TestCase): self.assertEqual(self.srv.pop("gg" + chr(ord("g")), "mm" + chr(ord("g"))), None) -if __name__ == "__main__": - unittest.main(verbosity=2) - - class TestInteractshDNSServer(unittest.TestCase): """The interactsh-backed DNS collector must present the same pop(prefix, suffix) accounting as DNSServer, matching only prefix..suffix names and never returning the same captured lookup twice.""" - def _collector(self, names): - class _FakeClient(object): - registered = True - def dnsDomain(self): return "corr0000000000000nnc.oast.fun" - def dnsNames(self): return list(names) + def _collector(self, names, deadline=None, start_poller=True): + """Build an InteractshDNSServer (bypassing __init__) over a fake client. + + The fake drives the real poller-aware consume path (production code, not a stub), returns + its records ONCE destructively - exactly as a real interactsh endpoint drains storage - + and, because close()/_request() are stubbed, can never reach the network at teardown. + ``deadline`` sets this instance's pop() search budget (seconds); ``start_poller`` runs the + single background consumer, as normal registered operation would. + """ + from lib.request.interactsh import Interactsh + client = Interactsh.__new__(Interactsh) # bypass network __init__ + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = False # __init__ sets it; set it here too + client._dnsNonce = None # lets the real dnsDomain() run (lazy-initialises it) + + names_batch = list(names) + + def poll(): + # return the whole batch once, then nothing - mirrors an interactsh endpoint + # draining its pending storage in a single /poll response + if names_batch: + batch = [{"protocol": "dns", "full-id": n, "unique-id": n} for n in names_batch] + names_batch.clear() + return batch + return [] + client.poll = poll + + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + + # teardown (srv.stop() -> client.stop() -> close) must never touch the network + client.close = lambda: _stop_client_poller(client) + client._request = lambda *a, **k: None + srv = InteractshDNSServer.__new__(InteractshDNSServer) - srv._client = _FakeClient() + srv._client = client srv.domain = srv._client.dnsDomain() srv._seen = set() srv._running = True srv._initialized = True - srv._POLL_TRIES = 1 # no real sleeps in unit tests + srv._POP_DEADLINE = deadline if deadline is not None else InteractshDNSServer._POP_DEADLINE + self.addCleanup(srv.stop) # stops the poller then deregisters (close() is network-stubbed) + if start_poller: + srv._client._ensurePoller() + poller = srv._client._poller + if poller is not None: + t = time.time() + 2.0 + while not poller._queue and time.time() < t: # wait for the destructive batch to stage + time.sleep(0.01) return srv def test_pop_matches_prefix_suffix_and_dedups(self): @@ -350,16 +399,450 @@ class TestInteractshDNSServer(unittest.TestCase): srv = self._collector(names) got = srv.pop("aaa", "zzz") self.assertEqual(got, "aaa.5345435245540a.zzz.corr0000000000000nnc") - self.assertIsNone(srv.pop("aaa", "zzz")) # already consumed + self.assertIsNone(srv.pop("aaa", "zzz", deadline=0)) # consumed only once (snapshot) + srv.stop() + + def test_pop_does_not_consume_a_sharing_http_record(self): + """Regression for the cross-consumer loss the shared poller was introduced to prevent. + + pop() and httpRequests() draw from the same background poller queue. If pop() does not + verify the record's protocol, a queued HTTP interaction whose full-id matches the DNS + window would be returned as though it were a DNS lookup, and the HTTP consumer would + then lose it. Both records here share the SAME full-id; pop() must return the DNS name + only and leave the HTTP record for httpRequests().""" + from lib.request.interactsh import Interactsh + dns_full = "aaa.deadbeef.bbb.corr0000000000000nnc" + http_full = "aaa.deadbeef.bbb.corr0000000000000nnc" # shared full-id with the DNS record + recs = [ + {"protocol": "dns", "full-id": dns_full, "unique-id": dns_full, + "remote-address": "1.1.1.1", "raw-request": "QUERY"}, + {"protocol": "http", "full-id": http_full, + "raw-request": "GET /?cafebabe=1 HTTP/1.1", "remote-address": "2.2.2.2"}, + ] + client = Interactsh.__new__(Interactsh) # bypass network __init__ + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = False # __init__ sets it; set it here too + client._dnsNonce = None + + rec_batch = list(recs) + + def poll(): + # return the whole batch once, then nothing - mirrors an interactsh endpoint + # draining its pending storage in a single /poll response + if rec_batch: + batch = rec_batch[:] + rec_batch.clear() + return batch + return [] + client.poll = poll + + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + client.close = lambda: _stop_client_poller(client) + client._request = lambda *a, **k: None + + srv = InteractshDNSServer.__new__(InteractshDNSServer) + srv._client = client + srv.domain = srv._client.dnsDomain() + srv._seen = set() + srv._running = True + srv._initialized = True + srv._POP_DEADLINE = 1.0 + self.addCleanup(srv.stop) + + srv._client._ensurePoller() # start the single shared background poller + poller = srv._client._poller + if poller is not None: + t = time.time() + 2.0 + while not poller._queue and time.time() < t: # wait for the destructive batch to stage + time.sleep(0.01) + + self.assertEqual(srv.pop("aaa", "bbb"), dns_full) # DNS record claimed + self.assertFalse(srv.pop("aaa", "bbb", deadline=0)) # snapshot after the DNS record is gone + # the HTTP record with the same full-id is still there for the HTTP consumer + out = client.httpRequests() + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["full-id"], http_full) + self.assertEqual(out[0]["query-param-names"], ["cafebabe"]) def test_pop_no_match(self): - srv = self._collector(["aaa.deadbeef.qqq.corr0000000000000nnc"]) + srv = self._collector(["aaa.deadbeef.qqq.corr0000000000000nnc"], deadline=1.0, start_poller=False) self.assertIsNone(srv.pop("aaa", "zzz")) def test_pop_any(self): srv = self._collector(["whatever.corr0000000000000nnc"]) self.assertEqual(srv.pop(), "whatever.corr0000000000000nnc") + srv.stop() def test_run_is_noop(self): - self._collector([]).run() # must not raise + srv = self._collector([]) + srv.run() # must not raise + srv.stop() + def test_pop_with_no_poller_installed(self): + """Backwards-compatible: a server without a running poller resolves pop() via the direct + (synchronous) poll path.""" + names = ["mmm.cafebabefff.ccc.corr0000000000000nnc"] + srv = self._collector(names, start_poller=False) + self.assertEqual(srv.pop("mmm", "ccc", sync=True), names[0]) + srv.stop() + + def test_background_poller_stages_names_instantly(self): + """With the single background poller running, pop() reads a staged name from its in-memory + queue with no extra /poll round-trip.""" + names = ["aaa.deadbeef.bbb.corr0000000000000nnc"] + srv = self._collector(names) # poller started and its destructive batch staged + try: + self.assertTrue(srv._client._poller.isRunning()) + self.assertEqual(_wait_popped(srv, "aaa", "bbb"), names[0]) + self.assertIsNone(srv.pop("aaa", "bbb", deadline=0)) # only one lookup exists (snapshot) + finally: + srv.stop() + + def test_background_poller_preserves_foreign_window_names(self): + """A name for a *different* window survives its own pop() scan - pop(zzz,yyy) must not drain + the aaa/bbb record, which pop(aaa,bbb) then returns.""" + names = ["aaa.deadbeef.bbb.corr0000000000000nnc"] + srv = self._collector(names, deadline=1.0) + try: + self.assertFalse(srv.pop("zzz", "yyy")) # wrong window yields nothing (~1s) + self.assertEqual(srv.pop("aaa", "bbb"), names[0]) + self.assertIsNone(srv.pop("aaa", "bbb", deadline=0)) # only one lookup (snapshot) + finally: + srv.stop() + + def test_ensure_poller_starts_only_one_background_thread(self): + """Concurrent _ensurePoller() callers must converge on a single background poller. + + The create-and-start decision runs under _pollerLock, so two threads racing on an + unstarted session cannot each build their own poller and destructively poll the same + interaction session concurrently - precisely what this redesign exists to prevent.""" + from lib.request.interactsh import Interactsh, InteractshPoller, hasCrypto + if not hasCrypto(): + self.skipTest("requires pycryptodome") + client = Interactsh.__new__(Interactsh) + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._dnsNonce = None + client._poller = None + client._pollerLock = threading.Lock() + client._stopping = False # set by __init__ in the real client; set it here too + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + # stub the remote endpoint and deregister path so the poll loop and client.stop() stay + # offline while we probe the create/start synchronisation + client.poll = lambda: [] + client._request = lambda *a, **k: "[]" + + started = [] + orig_start = InteractshPoller.start + + def counting_start(self): + started.append(self) # record every poller whose start() is invoked + orig_start(self) + + client._poller = None + InteractshPoller.start = counting_start + try: + threads = [threading.Thread(target=client._ensurePoller) for _ in range(16)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + InteractshPoller.start = orig_start + # the single poller was created exactly once and is still running (only one thread + # ever polled the session) + self.assertEqual(len(started), 1) # start() invoked exactly once + self.assertEqual(len(set(id(p) for p in started)), 1) # exactly one InteractshPoller + self.assertIn(client._poller, started) + self.assertTrue(client._poller._running) + # stop the daemon poller before this test returns so its /poll loop (stubbed to []) + # does not keep looping once per second for the rest of the test process + client._poller.stop() + + def test_close_stops_poller_and_deregisters(self): + """close() must remain the canonical complete teardown: after it returns this object owns + no active polling thread. + + Interactsh.close() predates the background poller and only deregistered, so a running + poller could keep polling (and race the remote session during deregistration). close() + now routes through stop().""" + from lib.request.interactsh import Interactsh, hasCrypto + if not hasCrypto(): + self.skipTest("requires pycryptodome") + client = Interactsh.__new__(Interactsh) + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._dnsNonce = None + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = False # __init__ sets it; set it here too + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + # stub remote endpoint and deregister path so the poll loop and client.stop() stay offline + client.poll = lambda: [] + client._request = lambda *a, **k: "[]" + + client._ensurePoller() # start the single background poller + self.assertTrue(client._poller.isRunning()) + + client.close() # close() must stop the poller and deregister + # isRunning() would be False even while the thread still runs when poll() wedged past the + # join budget; assert the underlying thread is actually gone so the invariant is real. + self.assertFalse(client._poller._thread.is_alive()) # no active polling thread remains + self.assertFalse(client._poller.isRunning()) + self.assertFalse(client.registered) + # cleanup: idempotent - already stopped/deregistered + self.addCleanup(client.stop) + + def test_consume_does_not_poll_while_stopping(self): + """_consume() must not do its own destructive /poll during shutdown. + + _consume() used to fall back to self.poll() whenever the poller was not running. Even + though _ensurePoller() refuses to start a new poller once _stopping is set, pop() can still + reach that fallback: _stopping is set, the poller is stopped, but _consume() could then + directly call self.poll() (a destructive /poll) just before stop()'s close(). A direct + poll during shutdown is exactly the unsynchronised drain that breaks the shared-session + invariant, so a stopped poller must yield [] - never a direct poll().""" + from lib.request.interactsh import Interactsh, hasCrypto + if not hasCrypto(): + self.skipTest("requires pycryptodome") + client = Interactsh.__new__(Interactsh) # bypass network __init__ + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._poller = None + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = True # simulate an active stop() + client._dnsNonce = None + + calls = {"poll": 0} + + def spy_poll(): + calls["poll"] += 1 + return [] + + client.poll = spy_poll + + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + + # even with no poller at all, a stopped session must not call poll() from _consume + # (drain_matching on a non-running poller yields []) + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns"), []) + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns"), []) + self.assertEqual(calls["poll"], 0) # sync=False: ensurePoller refused while stopping + + # sync=True must NOT initiate a remote poll once shutdown has begun: the direct-poll path + # is gated behind _stopping==False, so a stopped session drains the (empty) queue instead + # - no caller initiates another /poll after stop(), not even this compatibility path. + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns", sync=True), []) + self.assertEqual(calls["poll"], 0) + + # sanity: sync=True still drives poll() on a live (not-stopping) session - this guards + # against over-closing the safety check to sync itself. _stopping stays False, poller + # never starts (sync drains directly), so nothing needs cleaning up. + client._stopping = False + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns", sync=True), []) + self.assertEqual(calls["poll"], 1) + + def test_consume_does_not_second_poll_while_poller_running(self): + """sync=True must not add a second /poll caller while the background poller is already running. + + The shared-poller invariant is that only ONE code path ever calls the remote /poll endpoint. + A spurious sync=True during normal operation used to be the footgun: it called self.poll() + even when a live poller was already draining the session, yielding a second destructive + /poll. With a running poller sync=True must drain the poller's in-memory queue instead - + no additional poll() call - and the decision must be serialized under _pollerLock so a + concurrent _ensurePoller() cannot start a new poller in the gap between the check and the + (absent) poll().""" + from lib.request.interactsh import Interactsh, hasCrypto + if not hasCrypto(): + self.skipTest("requires pycryptodome") + client = Interactsh.__new__(Interactsh) # bypass network __init__ + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = False + client._dnsNonce = None + client._poller = None + # stub remote + deregister paths so any direct poll() stays offline + client.poll = lambda: [] + client._request = lambda *a, **k: "[]" + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + # start the single background poller, then replace poll with a counting spy AFTER staging. + # Count only calls made from THIS (foreground) thread: the legitimate background poller + # thread is permitted to poll(), so thread identity - not a bare call count - isolates the + # footgun (sync=True issuing a second /poll) from routine background polling. + main_thread = threading.current_thread() + foreground_calls = [] + client._ensurePoller() + self.assertTrue(client._poller.isRunning()) + + def spy_poll(): + if threading.current_thread() is main_thread: + foreground_calls.append(1) + return [] + client.poll = spy_poll + + # sync=True on a live, poller-running session drains the queue; the running poller is the + # only code that issues /poll, so the foreground must never call poll() directly. + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns", sync=True), []) + self.assertEqual(client._consume(lambda r: r.get("protocol") == "dns", sync=True), []) + self.assertEqual(foreground_calls, []) # sync=True issued no direct /poll + client.stop() # join the single (still-running) poll thread + + + +class TestInteractshHTTPRequests(unittest.TestCase): + """HTTP-based out-of-band retrieval is the collector half of an HTTP-OOB SQLi exfil channel. + + interactsh returns full 'http' protocol records with a verbatim 'raw-request', a + 'protocol' ("http" or "https") and a 'remote-address' - the request line, not a parsed + path/query. Because an HTTP callback can be blocked (e.g. firewalled query string) and a + value may contain characters unsafe in a query-string name, the payload is embedded in a + query-string *parameter name* and httpRequests() decodes and returns the distinct names. + + The request target is split on the *raw* '?'/'&'/first '=' and only each name is percent- + decoded individually, so encoded delimiter bytes survive (``%26``->'&', ``%3D``->'=', + ``%25``->'%'); decoding the whole target first would corrupt them.""" + + def _client(self, records): + """Fake client carrying the records in poll(), wired to the real poller-aware consume path. + + Records are returned once (destructive, like a draining interactsh endpoint) and the + poller is started so httpRequests() exercises the normal registered-poller path; teardown + cannot reach the network.""" + from lib.request.interactsh import Interactsh + client = Interactsh.__new__(Interactsh) # bypass network __init__ + client.registered = True + client.correlationId = "corr0000000000000nnc" + client.server = "oast.fun" + client.secret = "test-secret" + client._pollerLock = threading.Lock() # __init__ is bypassed, so set it explicitly + client._stopping = False # __init__ sets it; set it here too + + rec_batch = list(records) + + def poll(): + # return the whole batch once, then nothing - mirrors an interactsh endpoint + # draining its pending storage in a single /poll response + if rec_batch: + batch = rec_batch[:] + rec_batch.clear() + return batch + return [] + client.poll = poll + + for meth in ("_ensurePoller", "_consume", "_safeSelector"): + setattr(client, meth, types.MethodType(getattr(Interactsh, meth), client)) + + client.close = lambda: _stop_client_poller(client) + client._request = lambda *a, **k: None + + client._ensurePoller() + poller = client._poller + if poller is not None: + t = time.time() + 2.0 + while not poller._queue and time.time() < t: + time.sleep(0.01) + # client.stop() (registered as cleanup) stops the single background poller then + # deregisters, so the poll thread never outlives this test and never reaches the network + self.addCleanup(client.stop) + return client + + def test_extracts_query_param_names_from_raw_request(self): + records = [{"protocol": "http", "full-id": "a.b.c", "raw-request": "GET /?deadbeef=cafebabe HTTP/1.1", "remote-address": "1.2.3.4"}] + client = self._client(records) + out = client.httpRequests() + self.assertEqual(len(out), 1) + # only the query-string *name* is the attacker-controlled exfil slot; the value rides on it + self.assertEqual(out[0]["query-param-names"], ["deadbeef"]) + self.assertEqual(out[0]["client"], "1.2.3.4") + self.assertEqual(out[0]["request"], "GET /?deadbeef=cafebabe HTTP/1.1") + + def test_decodes_each_name_individually(self): + """Encoded delimiter bytes must survive split: each name is decoded on its own.""" + records = [ + {"protocol": "https", "full-id": "a", "raw-request": "GET /?aa%26bb=1 HTTP/1.1"}, # %26 -> '&' + {"protocol": "https", "full-id": "b", "raw-request": "GET /?cc%3Ddd=1 HTTP/1.1"}, # %3D -> '=' + {"protocol": "https", "full-id": "c", "raw-request": "GET /?ee%25ff=1 HTTP/1.1"}, # %25 -> '%' + ] + client = self._client(records) + out = client.httpRequests() + self.assertEqual([r["query-param-names"] for r in out], [["aa&bb"], ["cc=dd"], ["ee%ff"]]) + + def test_skips_non_http_records(self): + records = [{"protocol": "dns", "full-id": "x.y"}] + client = self._client(records) + self.assertEqual(client.httpRequests(), []) + + def test_distinct_names_are_not_duplicated(self): + records = [{"protocol": "http", "full-id": "a", "raw-request": "GET /?deadbeef=1 HTTP/1.1"}, + {"protocol": "http", "full-id": "b", "raw-request": "GET /?cafebabe=2 HTTP/1.1"}] + client = self._client(records) + out = client.httpRequests() + names = [name for r in out for name in r["query-param-names"]] + self.assertEqual(set(names), {"deadbeef", "cafebabe"}) + + def test_empty_poll_returns_empty(self): + client = self._client([]) + self.assertEqual(client.httpRequests(), []) + + def test_selector_exception_is_absorbed(self): + """A selector that raises must be caught (not propagated out of httpRequests/_consume)""" + records = [{"protocol": "http", "full-id": "a", "raw-request": "GET /?x=1 HTTP/1.1"}] + client = self._client(records) + + def boom(record): + raise ValueError("boom") + + self.assertEqual(client._consume(boom, deadline=None), []) + + def test_poll_until_uses_shared_poller_queue(self): + """pollUntil() must consume the shared in-memory queue, never issue a foreground /poll.""" + records = [{"protocol": "http", "full-id": "a", "raw-request": "GET /?queued=1 HTTP/1.1"}] + client = self._client(records) + main_thread = threading.current_thread() + foreground_calls = [] + + def spy_poll(): + if threading.current_thread() is main_thread: + foreground_calls.append(1) + return [] + + client.poll = spy_poll + out = client.pollUntil(1, 0) + self.assertEqual(len(out), 1) + self.assertEqual(out[0].get("full-id"), "a") + self.assertEqual(foreground_calls, []) + + def test_multiple_http_same_host_are_not_dropped(self): + """The reviewer's core case: several HTTP requests hitting the same interactsh host - i.e. + sharing 'full-id' - must all survive the poller's dedup. Dedup is by the interaction + itself (protocol+full-id+id/timestamp+raw-request), never by the shared host.""" + records = [ + {"protocol": "dns", "full-id": "abc..."}, + {"protocol": "http", "full-id": "abc...", "raw-request": "GET /?deadbeef=1 HTTP/1.1"}, + {"protocol": "http", "full-id": "abc...", "raw-request": "GET /?cafebabe=1 HTTP/1.1"}, + ] + client = self._client(records) + out = client.httpRequests() + names = [name for r in out for name in r["query-param-names"]] + self.assertEqual(set(names), {"deadbeef", "cafebabe"}) + + +if __name__ == "__main__": + unittest.main(verbosity=2)