Added shared Interactsh poller with DNS/HTTP OOB support

lib/core/settings.py

@@ -20,7 +20,7 @@ from lib.core.enums import OS
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-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)

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.<query result>.suffix.<correlation domain>), 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__":

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.<hex>.suffix.<correlationId><nonce>'."""
-        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.<hex>.suffix.<correlationId><nonce>'.
+
+        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 []

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.<result>.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)