Fixes #6143

Every tests/test_*.py file provides an

    if __name__ == "__main__":
        unittest.main(...)

entrypoint so it can be run directly, except test_ssti.py and
test_xpath.py. Running either of those directly produced no output and
executed none of its tests; they only ran under `unittest discover`.

Add the missing entrypoint to both so they behave like the rest of the
suite. No test logic changed (62 and 48 tests respectively still pass).

data/xml/errors.xml

@@ -145,7 +145,7 @@
     </dbms>
 
     <dbms value="SAP MaxDB">
-        <error regexp="SQL error.*?POS([0-9]+)"/>
+        <error regexp="SQL[- ]error.*?POS\(\d+\)"/>
         <error regexp="Warning.*?\Wmaxdb_"/>
         <error regexp="DriverSapDB"/>
         <error regexp="-3014.*?Invalid end of SQL statement"/>

doc/ARCHITECTURE.md

@@ -19,6 +19,7 @@ non-ASCII byte is unavoidable. Compatibility shims live in `lib/core/compat.py`
 |-------|------|---------|
 | CLI | `sqlmap.py` -> `main()` | the scanner. Applies runtime patches, parses options, runs a scan. |
 | REST API | `sqlmapapi.py` | `-s` server / `-c` client wrappers around `lib/utils/api.py`. |
+| Library | `import sqlmap` -> `lib/utils/library.py` | `scan()` / `scanFromRequest()` for programmatic callers; like the API, it drives the engine as a subprocess. |
 
 `main()` (sqlmap.py) does, in order: `dirtyPatches()` (monkey-patches stdlib for
 quirks/security - see below), `setPaths()`, `init()` (option parsing + environment
@@ -58,7 +59,7 @@ Identifiers in the codebase are camelCase.
 | `lib/core/` | conf/kb model, common helpers, settings, enums, dump, session, agent, option parsing |
 | `lib/controller/` | the scan orchestrator (`controller.py`), detection checks (`checks.py`), enumeration dispatch (`action.py`), DBMS handler selection (`handler.py`) |
 | `lib/request/` | HTTP layer: `connect.py` (sending), `comparison.py` (the true/false oracle), `inject.py` (value extraction), protocol handlers, response processing |
-| `lib/techniques/` | the exploitation engines: `blind/inference.py`, `error/use.py`, `union/{test,use}.py`, `dns/` |
+| `lib/techniques/` | the exploitation engines: `blind/{inference,multibit}.py`, `error/use.py`, `union/{test,use}.py`, `dns/`, plus one directory per non-SQL family (`ssti/`, `nosql/`, `xpath/`, ...) |
 | `lib/parse/` | parsing of inputs: CLI, config, HTTP request/log files, HTML, sitemap, and the XML payload/boundary loader (`payloads.py`) |
 | `lib/utils/` | feature modules: `api.py` (REST), `hashdb.py` (session), `crawler.py`, `hash.py` (cracking), `har.py`, `brute.py`, `search.py`, ... |
 | `lib/takeover/` | OS-level takeover: shells, file access, UDF, registry, Metasploit, `xp_cmdshell` |
@@ -130,6 +131,7 @@ Once a parameter is injectable, value extraction is dispatched by
 | Technique | Engine | Mechanism |
 |-----------|--------|-----------|
 | boolean-based blind | `blind/inference.py: bisection()` | binary-search each character via true/false oracle |
+| multi-bit blind (`--multi-bit`) | `blind/multibit.py` | opt-in; reads several bits per request out of the rows a listing page renders |
 | time-based blind / stacked | `blind/inference.py` (time compare) | same bisection, oracle is a measured delay |
 | error-based | `error/use.py: errorUse()` | parse the value straight out of a provoked DB error |
 | UNION query | `union/{test,use}.py` | column-count detection then `UNION SELECT` extraction |
@@ -141,6 +143,14 @@ Once a parameter is injectable, value extraction is dispatched by
 (intentional). Multi-threaded extraction is coordinated via `kb.locks` and
 `getCurrentThreadData()` (`lib/core/threads.py`).
 
+**Non-SQL engines.** `--nosql`, `--xpath`, `--ldap`, `--ssti`, `--graphql`, `--hql`, `--sparql`,
+`--odata`, `--xslt`, `--xxe` and `--jwt` do not go through the pipeline above at all. Each one is a
+self-contained scanner in `lib/techniques/<family>/inject.py`, dispatched from `controller.py` and
+registered once in `NONSQL_TECHNIQUES` (`lib/core/settings.py`) - that tuple is what the rest of the
+code tests against, so a new family is declared there and nowhere else. The response comparison,
+reflection stripping and blind-bit classification they all need live in `lib/utils/nonsql.py`
+instead of being copied per family.
+
 ---
 
 ## 7. DBMS abstraction
@@ -155,7 +165,10 @@ Enumeration is DBMS-agnostic at the top and specialized underneath:
   pieces and supplying dialect specifics.
 - **`data/xml/queries.xml`** - per-DBMS SQL query templates (banner, current user, table
   enumeration, casting, etc.) keyed by DBMS. The generic code asks for a query by name;
-  the dialect comes from XML.
+  the dialect comes from XML. It also carries the per-DBMS `<gadgets>` - side-effecting scalar
+  expressions (e.g. PostgreSQL `dblink_exec`) that `getGadget()` (`lib/request/inject.py`) probes
+  once and caches in `kb.gadget`, so `--sql-query` / `--file-write` / `--os-cmd` still work from an
+  injection point that has no stacked queries.
 
 `conf.dbmsHandler` (set in `handler.py`) is the live object that `action()` calls into.
 
@@ -180,12 +193,16 @@ Enumeration is DBMS-agnostic at the top and specialized underneath:
 `lib/request/connect.py` (`Connect.getPage`) is the single HTTP chokepoint. Around it:
 protocol handlers (`httpshandler`, `redirecthandler`, `chunkedhandler`, `rangehandler`,
 persistent connections via `lib/request/keepalive.py`), response processing (`basic.py`), and the
-comparison oracle (`comparison.py`).
+comparison oracle (`comparison.py`). Alternative transports hang off the same chokepoint: the
+standard-library HTTP/2 client (`http2.py`, `--http2`, which also carries the `--timeless` oracle in
+`timeless.py`) and WebSocket targets (`websocket.py`).
 
 **Tamper scripts** (`tamper/`) mutate the payload just before sending to evade WAF/IPS.
 Each file exposes a `tamper(payload, **kwargs)` and a `__priority__`; `--tamper=a,b,c`
 chains them in priority order. They are payload-string transforms only (no engine
-coupling), which is why they compose freely.
+coupling), which is why they compose freely. When a WAF/IPS is identified, `lib/utils/wafbypass.py`
+ranks the plausible candidates and `_autoWafBypass()` (`controller.py`) trials them one at a time -
+a candidate is adopted only if re-running the detection through it brings the injection back.
 
 ---
 
@@ -227,6 +244,7 @@ Two complementary layers:
 | a constant/threshold | `lib/core/settings.py` |
 | how injection is *detected* | `data/xml/boundaries.xml` + `data/xml/payloads/*.xml`, then `lib/controller/checks.py` |
 | how a value is *extracted* | `lib/request/inject.py` + the relevant `lib/techniques/` engine |
+| a non-SQL technique (SSTI, NoSQL, XPath, ...) | `lib/techniques/<family>/inject.py` (+ `NONSQL_TECHNIQUES` in `settings.py`) |
 | the true/false decision | `lib/request/comparison.py` |
 | a per-DBMS query/dialect | `data/xml/queries.xml` + `plugins/dbms/<dbms>/` |
 | enumeration behavior | `plugins/generic/*.py` |

doc/CHANGELOG.md

@@ -15,10 +15,12 @@
 * Added the switch `--xslt`. It tests for XSLT injection. The engine names itself in the response. sqlmap then dumps the XML document that the stylesheet transforms. It also reads the files that the engine can reach. When the engine exposes an extension bridge (PHP `php:function` or the Xalan `java:` namespace), sqlmap reads any file through it, and with `--os-cmd` or `--os-shell` it runs operating system commands.
 * Added the switch `--xxe`. It tests for XML External Entity injection. It uses in-band, error-based, and out-of-band channels.
 * Added the switch `--jwt`. It examines JSON Web Tokens for weak keys and for injection in the claims.
+* sqlmap now offers to get around a WAF/IPS on its own. Once it identifies one and the ordinary payloads come back blocked, it drops the scanner fingerprint and tries the tamper scripts that suit that WAF, and it keeps the first one that brings the injection back.
 
 ## Speed
 
 * Added the switch `--timeless`. It reads each blind bit from the HTTP/2 response order. It does not use a delay. sqlmap calibrates the target first, and it uses the usual time-based technique if the target is not applicable.
+* Added the switch `--multi-bit`. On a page that renders rows, one request reads a bit from each row that comes back instead of a single bit altogether.
 * Added set-membership (Huffman) retrieval for blind dumps. It needs fewer requests for each character. Use `--no-huffman` to stop it.
 * Added keyset (seek) pagination for blind table dumps. Use `--no-keyset` to stop it.
 * Added parallel retrieval of values in blind mode. Each thread retrieves a different value.
@@ -53,10 +55,13 @@
 * Added time-based payloads for CUBRID.
 * Added out-of-band DNS channels for H2 and ClickHouse.
 * Added PostgreSQL command execution through a PL extension.
-* Added the running of non-query statements without stacked queries through a gadget. On PostgreSQL, when the `dblink` extension is present, `--sql-query`, `--file-write`, `--os-cmd`, and `--os-shell` now work from a plain (e.g. boolean-based) injection point.
+* Added the running of non-query statements without stacked queries through a gadget. On PostgreSQL, when the `dblink` extension is there and the current user can run it, `--sql-query`, `--file-write`, `--os-cmd`, and `--os-shell` now work from a plain (e.g. boolean-based) injection point.
 * Added file read and file write support for SQLite through the `fileio` extension functions `readfile` and `writefile`.
 * Added the data access and the security type of a routine to the output of `--procs` on MySQL and PostgreSQL, so a routine that runs as its definer or that modifies data stands out.
-* Added the tamper scripts `blindbinary`, `dollarquote`, `infoschema2innodb`, `oraclequote`, and `sign`.
+* Added stacked query payloads for Microsoft SQL Server and Sybase that carry no semicolon, so a filtered semicolon alone no longer hides the technique.
+* Added the error signatures that tell the MySQL and the PostgreSQL forks apart: Doris, StarRocks, CockroachDB, YugabyteDB, OpenGauss, DuckDB, and Trino. These forks keep the wording of the engine that they come from, so only a leaked driver package or an engine-internal source reference gives them away.
+* Made the fingerprinting payloads friendlier to a WAF, so the version detection survives where it used to get blocked.
+* Added the tamper scripts `blindbinary`, `castprefix`, `dollarquote`, `infoschema2innodb`, `mid2leftright`, `mssqlnosemicolon`, `odbcbrace`, `oraclequote`, `quote2ltat`, `sign`, `sleep2hex`, and `uniontable`.
 
 ## Fewer dependencies
 
@@ -93,6 +98,11 @@
 * Made the heuristic hints of the non-SQL switches exclusive. A signature no longer matches the errors of a different engine, an ordinary SQL error, or a page that only contains the name of a template engine.
 * Corrected the GraphQL validation signatures. They now match the quotes in the way that the JSON body escapes them.
 * Added the error signatures of Mako and of DynamoDB. sqlmap did not recognise the errors of these two back-ends.
+* sqlmap no longer stops with an exception on a JWT whose signature is not valid base64url. Such a signature can not be an HMAC that sqlmap could verify, so it is reported as not crackable instead.
+* Corrected the uppercasing of the keywords in the shown queries. A word that merely contains a keyword (e.g. a table named `selected`) and a keyword inside a quoted string are now left alone.
+* Corrected the cracking of the old Oracle password hashes when the user name or the password does not come in as text.
+* sqlmap now skips a UNION test whose request holds no character instead of stopping with an exception. Only a hand-edited `payloads/union_query.xml` can hold such a test.
+* The gadget check now runs the gadget itself instead of only looking for the extension. A user that can not use it no longer gets a false success.
 
 ## Quality
 

extra/esperanto/handler.py

@@ -31,6 +31,7 @@ def buildHandler():
     from lib.core.enums import CHARSET_TYPE
     from lib.core.enums import EXPECTED
     from lib.core.exception import SqlmapDataException
+    from lib.core.exception import SqlmapUnsupportedFeatureException
     from lib.request.inject import checkBooleanExpression
     from lib.request.inject import getValue
     from plugins.generic.enumeration import Enumeration
@@ -53,6 +54,15 @@ def buildHandler():
                                         # reuses what --columns already enumerated
             self._scopeCache = {}       # table -> resolved schema (see _scopeFor)
 
+        def __getattr__(self, name):
+            # this handler only mixes in Enumeration/Miscellaneous (read-only, boolean-oracle
+            # extraction); any DBMS-specific capability (--os-*/--file-*/--sql-shell/--udf-*/--reg-*/
+            # --cleanup) has no generic equivalent - fail cleanly instead of an AttributeError deep
+            # in action(). Leading underscore excluded so internal/dunder lookups behave normally.
+            if name.startswith("_"):
+                raise AttributeError(name)
+            raise SqlmapUnsupportedFeatureException("Esperanto does not support '%s' (DBMS-agnostic engine covers enumeration/dump only)" % name)
+
         def _engine(self):
             if self._esp is None:
                 # esperanto is a PURE boolean-oracle engine: every probe is one true/false

lib/controller/checks.py

@@ -15,6 +15,7 @@ import time
 from extra.beep.beep import beep
 from lib.core.agent import agent
 from lib.core.common import Backend
+from lib.core.common import arrayizeValue
 from lib.core.common import extractRegexResult
 from lib.core.common import extractStructuralTokens
 from lib.core.common import extractTextTagContent
@@ -183,7 +184,11 @@ def checkSqlInjection(place, parameter, value):
                 if kb.reduceTests is None and not conf.testFilter and (intersect(Backend.getErrorParsedDBMSes(), SUPPORTED_DBMS, True) or kb.heuristicDbms or injection.dbms):
                     msg = "it looks like the back-end DBMS is '%s'. " % (Format.getErrorParsedDBMSes() or kb.heuristicDbms or joinValue(injection.dbms, '/'))
                     msg += "Do you want to skip test payloads specific for other DBMSes? [Y/n]"
-                    kb.reduceTests = (Backend.getErrorParsedDBMSes() or [kb.heuristicDbms]) if readInput(msg, default='Y', boolean=True) else []
+                    # mirror msg's fallback chain (error-parsed -> heuristic -> injection.dbms) - falling
+                    # back only through the first two (as before) left kb.reduceTests as [None] whenever
+                    # injection.dbms alone satisfied the 'if' above, silently skipping every DBMS-specific
+                    # test payload for the rest of the scan
+                    kb.reduceTests = (Backend.getErrorParsedDBMSes() or ([kb.heuristicDbms] if kb.heuristicDbms else arrayizeValue(injection.dbms))) if readInput(msg, default='Y', boolean=True) else []
 
             # If the DBMS has been fingerprinted (via DBMS-specific error
             # message, via simple heuristic check or via DBMS-specific
@@ -925,7 +930,9 @@ def heuristicCheckDbms(injection):
     may be
     """
 
-    retVal = False
+    # None (not False) on failure - every caller gates re-running this on 'kb.heuristicDbms is
+    # None', and a stale False from an earlier parameter's failed check would wrongly block that
+    retVal = None
 
     if conf.skipHeuristics:
         return retVal
@@ -1223,6 +1230,14 @@ def heuristicCheckSqlInjection(place, parameter):
         if conf.beep:
             beep()
 
+    def _search(regex):
+        # Note: on a rare (e.g. huge) response the regex engine itself can fail, and losing one
+        # advisory heuristic beats losing the whole run (e.g. #5994 and #6105)
+        try:
+            return re.search(regex, page or "")
+        except (SystemError, RuntimeError) as ex:
+            logger.debug("skipping heuristic check because of a regex engine failure ('%s')" % getSafeExString(ex))
+
     try:
         for match in re.finditer(FI_ERROR_REGEX, page or ""):
             if randStr1.lower() in match.group(0).lower():
@@ -1234,71 +1249,71 @@ def heuristicCheckSqlInjection(place, parameter):
 
                 break
     except (SystemError, RuntimeError) as ex:
-        logger.debug("Skipping FI heuristic due to regex failure: %s", getSafeExString(ex))
+        logger.debug("skipping heuristic check because of a regex engine failure ('%s')" % getSafeExString(ex))
 
-    if not conf.nosql and re.search(NOSQL_ERROR_REGEX, page or ""):
+    if not conf.nosql and _search(NOSQL_ERROR_REGEX):
         infoMsg = "heuristic (NoSQL) test shows that %sparameter '%s' might be vulnerable to NoSQL injection attacks (rerun with switch '--nosql')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.graphql and re.search(GRAPHQL_ERROR_REGEX, page or ""):
+    if not conf.graphql and _search(GRAPHQL_ERROR_REGEX):
         infoMsg = "heuristic (GraphQL) test shows that %sparameter '%s' appears to be a GraphQL endpoint (rerun with switch '--graphql')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.ldap and re.search(LDAP_ERROR_REGEX, page or ""):
+    if not conf.ldap and _search(LDAP_ERROR_REGEX):
         infoMsg = "heuristic (LDAP) test shows that %sparameter '%s' might be vulnerable to LDAP injection (rerun with switch '--ldap')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.xpath and re.search(XPATH_ERROR_REGEX, page or ""):
+    if not conf.xpath and _search(XPATH_ERROR_REGEX):
         infoMsg = "heuristic (XPath) test shows that %sparameter '%s' might be vulnerable to XPath injection (rerun with switch '--xpath')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.ssti and re.search(SSTI_ERROR_REGEX, page or ""):
+    if not conf.ssti and _search(SSTI_ERROR_REGEX):
         infoMsg = "heuristic (SSTI) test shows that %sparameter '%s' might be vulnerable to server-side template injection (rerun with switch '--ssti')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.hql and re.search(HQL_ERROR_REGEX, page or ""):
+    if not conf.hql and _search(HQL_ERROR_REGEX):
         infoMsg = "heuristic (HQL) test shows that %sparameter '%s' might be vulnerable to HQL/JPQL (Hibernate ORM) injection (rerun with switch '--hql')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.xslt and re.search(XSLT_ERROR_REGEX, page or ""):
+    if not conf.xslt and _search(XSLT_ERROR_REGEX):
         infoMsg = "heuristic (XSLT) test shows that %sparameter '%s' might be vulnerable to XSLT injection (rerun with switch '--xslt')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
         if conf.beep:
             beep()
 
-    if not conf.sparql and re.search(SPARQL_ERROR_REGEX, page or ""):
+    if not conf.sparql and _search(SPARQL_ERROR_REGEX):
         infoMsg = "heuristic (SPARQL) test shows that %sparameter '%s' might be vulnerable to SPARQL injection (rerun with switch '--sparql')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.odata and re.search(ODATA_ERROR_REGEX, page or ""):
+    if not conf.odata and _search(ODATA_ERROR_REGEX):
         infoMsg = "heuristic (OData) test shows that %sparameter '%s' might be vulnerable to OData $filter injection (rerun with switch '--odata')" % ("%s " % paramType if paramType != parameter else "", parameter)
         logger.info(infoMsg)
 
         if conf.beep:
             beep()
 
-    if not conf.xxe and kb.postHint in (POST_HINT.XML, POST_HINT.SOAP) and re.search(XXE_ERROR_REGEX, page or ""):
+    if not conf.xxe and kb.postHint in (POST_HINT.XML, POST_HINT.SOAP) and _search(XXE_ERROR_REGEX):
         infoMsg = "heuristic (XXE) test shows that the XML request body might be vulnerable to XML External Entity injection (rerun with switch '--xxe')"
         logger.info(infoMsg)
 
@@ -1317,7 +1332,7 @@ def checkJWT():
     headers, missing expiry), hinting at '--jwt' for active confirmation and injection.
     """
 
-    if kb.jwtChecked or conf.jwt:
+    if kb.get("jwtChecked") or conf.jwt:
         return
 
     kb.jwtChecked = True

lib/controller/controller.py

@@ -376,6 +376,11 @@ def start():
     for targetUrl, targetMethod, targetData, targetCookie, targetHeaders in kb.targets:
         targetCount += 1
 
+        # --report-json: give each target its own taskid in the shared collector, so a multi-target
+        # run (e.g. '-m' bulk file) doesn't have later targets overwrite earlier ones (see _reportData)
+        if conf.reportJson:
+            kb.reportTaskId = targetCount
+
         try:
             if conf.checkInternet:
                 infoMsg = "checking for Internet connection"

lib/core/agent.py

@@ -113,7 +113,11 @@ class Agent(object):
         paramDict = conf.paramDict[place]
         origValue = getUnicode(paramDict[parameter])
         newValue = getUnicode(newValue) if newValue else newValue
-        base64Encoding = re.sub(r" \(.+", "", parameter) in conf.base64Parameter
+        try:
+            # Dirty patch for Python 3.14.7 re.sub() regression (e.g. https://github.com/sqlmapproject/sqlmap/issues/6128)
+            base64Encoding = re.sub(r" \(.+", "", parameter) in conf.base64Parameter
+        except TypeError:
+            base64Encoding = False
 
         if place == PLACE.URI or BOUNDED_INJECTION_MARKER in origValue:
             paramString = origValue

lib/core/common.py

@@ -27,6 +27,7 @@ import platform
 import posixpath
 import random
 import re
+import signal
 import socket
 import string
 import subprocess
@@ -997,7 +998,7 @@ def setColor(message, color=None, bold=False, level=None, istty=None):
 
             if bold or color:
                 retVal = colored(message, color=color, on_color=None, attrs=("bold",) if bold else None)
-            elif level:
+            elif level and hasattr(LOGGER_HANDLER, "colorize"):
                 try:
                     level = getattr(logging, level, None)
                 except:
@@ -2476,22 +2477,58 @@ def getConsoleWidth(default=80):
 
     return width or default
 
-def shellExec(cmd):
+def shellExec(cmd, timeout=None):
     """
-    Executes arbitrary shell command
+    Executes arbitrary shell command, optionally bounded by 'timeout' seconds - killing (and
+    flagging) a hung child instead of blocking forever, as callers otherwise have no other
+    watchdog around this call (e.g. --vuln-test runs one such call per entry, unattended)
 
     >>> shellExec('echo 1').strip() == '1'
     True
     """
 
     retVal = ""
+    timedOut = []
 
     try:
-        retVal = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).communicate()[0] or ""
+        popenKwargs = {"shell": True, "stdout": subprocess.PIPE, "stderr": subprocess.STDOUT}
+        if timeout:
+            # shell=True's Popen.pid is the shell, not the (possibly grandchild) command it runs -
+            # killing just that pid leaves the real child holding the stdout pipe open, so
+            # communicate() keeps blocking past the deadline; run it in its own group/session instead
+            # so the whole tree can be killed at once
+            if IS_WIN:
+                popenKwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+            else:
+                popenKwargs["preexec_fn"] = os.setsid
+
+        process = subprocess.Popen(cmd, **popenKwargs)
+
+        def _kill():
+            timedOut.append(True)
+            try:
+                if IS_WIN:
+                    subprocess.call(["taskkill", "/F", "/T", "/PID", str(process.pid)])
+                else:
+                    os.killpg(os.getpgid(process.pid), signal.SIGKILL)
+            except Exception:
+                pass
+
+        timer = threading.Timer(timeout, _kill) if timeout else None
+        if timer:
+            timer.daemon = True
+            timer.start()
+
+        retVal = process.communicate()[0] or ""
+
+        if timer:
+            timer.cancel()
     except Exception as ex:
         retVal = getSafeExString(ex)
     finally:
         retVal = getText(retVal)
+        if timedOut:
+            retVal += "\n[shellExec] child process tree killed after exceeding %d-second timeout" % timeout
 
     return retVal
 

lib/core/dump.py

@@ -108,11 +108,15 @@ class Dump(object):
         collector is active - which is only ever the case for a CLI --report-json run, never under
         --api - so this never double-captures alongside StdDbOut. A None content_type is resolved
         via the kb.partRun fallback (e.g. the fingerprint line), mirroring the API exactly.
+
+        Keyed by kb.reportTaskId rather than the fixed REPORT_TASKID so that a multi-target run
+        (e.g. '-m' bulk file) keeps each target's results separate instead of later targets
+        overwriting earlier ones under the same content_type.
         """
 
         if conf.get("reportCollector") is not None:
             from lib.utils.api import _storeData, REPORT_TASKID
-            _storeData(conf.reportCollector, REPORT_TASKID, stdoutEncode(clearColors(data)), CONTENT_STATUS.COMPLETE, content_type)
+            _storeData(conf.reportCollector, kb.get("reportTaskId", REPORT_TASKID), stdoutEncode(clearColors(data)), CONTENT_STATUS.COMPLETE, content_type)
 
     def flush(self):
         if self._outputFP:

lib/core/option.py

@@ -847,6 +847,24 @@ def _listTamperingFunctions():
                 comment = match.group(1).strip()
                 dataToStdout("* %s - %s\n" % (setColor(os.path.basename(script), "yellow"), re.sub(r" *\n *", " ", comment.split("\n\n")[0].strip())))
 
+def _argSpec(function):
+    """
+    Cross Py2/Py3 argument-name introspection - inspect.getargspec() is deprecated since
+    Python 3.0 and removed entirely since 3.11, while inspect.getfullargspec() (used here
+    whenever available) covers the same 'args'/'keywords' need without the warning/removal
+    """
+
+    return inspect.getfullargspec(function) if hasattr(inspect, "getfullargspec") else inspect.getargspec(function)
+
+def _kwargsName(function):
+    """
+    Name of 'function's **kwargs-style catch-all parameter, or None - the field holding it is
+    called 'varkw' on a getfullargspec() result and 'keywords' on a (Python 2) getargspec() one
+    """
+
+    spec = _argSpec(function)
+    return getattr(spec, "varkw", None) or getattr(spec, "keywords", None)
+
 def _setTamperingFunctions():
     """
     Loads tampering functions from given script(s)
@@ -905,8 +923,15 @@ def _setTamperingFunctions():
             priority = PRIORITY.NORMAL if not hasattr(module, "__priority__") else module.__priority__
             priority = priority if priority is not None else PRIORITY.LOWEST
 
+            if not isinstance(priority, int):
+                warnMsg = "tamper module '%s' has an invalid value for '__priority__' " % filename[:-3]
+                warnMsg += "(assuming '%d')" % PRIORITY.NORMAL
+                logger.warning(warnMsg)
+
+                priority = PRIORITY.NORMAL
+
             for name, function in inspect.getmembers(module, inspect.isfunction):
-                if name == "tamper" and (hasattr(inspect, "signature") and all(_ in inspect.signature(function).parameters for _ in ("payload", "kwargs")) or inspect.getargspec(function).args and inspect.getargspec(function).keywords == "kwargs"):
+                if name == "tamper" and (hasattr(inspect, "signature") and all(_ in inspect.signature(function).parameters for _ in ("payload", "kwargs")) or _argSpec(function).args and _kwargsName(function) == "kwargs"):
                     found = True
                     kb.tamperFunctions.append(function)
                     function.__name__ = module.__name__
@@ -1016,7 +1041,7 @@ def _setPreprocessFunctions():
 
             for name, function in inspect.getmembers(module, inspect.isfunction):
                 try:
-                    if name == "preprocess" and inspect.getargspec(function).args and all(_ in inspect.getargspec(function).args for _ in ("req",)):
+                    if name == "preprocess" and _argSpec(function).args and all(_ in _argSpec(function).args for _ in ("req",)):
                         found = True
 
                         kb.preprocessFunctions.append(function)
@@ -1099,7 +1124,7 @@ def _setPostprocessFunctions():
                 raise SqlmapSyntaxException("cannot import postprocess module '%s' (%s)" % (getUnicode(filename[:-3]), getSafeExString(ex)))
 
             for name, function in inspect.getmembers(module, inspect.isfunction):
-                if name == "postprocess" and inspect.getargspec(function).args and all(_ in inspect.getargspec(function).args for _ in ("page", "headers", "code")):
+                if name == "postprocess" and _argSpec(function).args and all(_ in _argSpec(function).args for _ in ("page", "headers", "code")):
                     found = True
 
                     kb.postprocessFunctions.append(function)

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.0"
+VERSION = "1.10.10.2"
 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)
@@ -1193,7 +1193,10 @@ XPATH_ERROR_SIGNATURES = (
     ("Python ElementTree", r"xml\.etree\.ElementTree\.(?:ParseError|Element)"),
     # NOT XSLT: a dedicated '--xslt' engine owns those errors now, and claiming them here made every
     # XSLT parser error suggest '--xpath' as well
-    ("Generic XPath", r"XPath.*?(?:error|exception|syntax)"),
+    # NOTE: the gap has to stay bounded (like in 'Handlebars' below). An unbounded '.*?' turns this
+    # into a quadratic scan of every long line that merely carries the word 'xpath' (e.g. minified
+    # JS/JSON), which took ~25s on a 400KB response - and blew up the regex engine itself (#6105)
+    ("Generic XPath", r"XPath[^\n]{0,100}?(?:error|exception|syntax)"),
     ("Generic XPath", r"Invalid XPath|XPath evaluation failed"),
 )
 
@@ -1337,7 +1340,7 @@ SSTI_ERROR_SIGNATURES = (
     ("Velocity", r"org\.apache\.velocity\.(?:runtime|exception)\.\w+|ParseErrorException|MethodInvocationException|ResourceNotFoundException"),
     ("Spring EL / Thymeleaf", r"org\.springframework\.expression\.\w+|org\.thymeleaf\.\w+|SpelEvaluationException|TemplateProcessingException|ExpressionParsingException"),
     ("Struts2 (OGNL)", r"ognl\.(?:OgnlException|NoSuchPropertyException|MethodFailedException|InappropriateExpressionException|ExpressionSyntaxException)|com\.opensymphony\.xwork2|org\.apache\.struts2|There is no Action mapped for|Struts (?:Problem Report|has detected an unhandled exception)"),
-    ("ERB", r"\(erb\):\d+|NameError.*undefined local variable"),
+    ("ERB", r"\(erb\):\d+|NameError[^\n]{0,100}?undefined local variable"),
     # NOTE: these must stay anchored to a diagnostic. The bare product names matched any page that
     # carries the word 'pug'/'jade'/'handlebars' (a surname, a colour, a <script src=> of the runtime),
     # and the bare 'ParseError' matched lxml.etree.XSLTParseError and ElementTree.ParseError
@@ -1361,7 +1364,7 @@ XXE_ERROR_SIGNATURES = (
     ("Java (Xerces/JAXP)", r"(?:org\.xml\.sax\.SAXParseException|com\.sun\.org\.apache\.xerces|javax\.xml\.stream\.XMLStreamException|The (?:entity|element type) \"[^\"]*\" was referenced|DOCTYPE is disallowed when the feature|External (?:DTD|parsed entities|Entity): failed|\"[^\"]*\" must be declared|had to be read but the maximum)"),
     (".NET System.Xml", r"(?:System\.Xml\.XmlException|For security reasons DTD is prohibited|Reference to undeclared entity|An error occurred while parsing EntityName|XmlTextReaderImpl)"),
     ("Python expat", r"(?:xml\.parsers\.expat\.ExpatError|undefined entity|not well-formed \(invalid token\)|ExpatError)"),
-    ("Ruby Nokogiri/REXML", r"(?:Nokogiri::XML::SyntaxError|REXML::ParseException|Entity .* not defined)"),
+    ("Ruby Nokogiri/REXML", r"(?:Nokogiri::XML::SyntaxError|REXML::ParseException|Entity [^\n]{0,100}? not defined)"),
     ("Go encoding/xml", r"XML syntax error on line \d+"),
     # NOTE: 'unexpected end of ...' is what every parser says, not what an XML parser says. It matched
     # the "Unexpected end of query" of BaseX, the "Unexpected <EOF>" of GraphQL and the "Unexpected end
@@ -1417,7 +1420,7 @@ HQL_ERROR_SIGNATURES = (
     ("Hibernate", r"(?:unexpected (?:token:|end of subtree|AST node)|Could not (?:resolve|interpret) (?:attribute|root entity|path|property))"),
     ("EclipseLink / JPQL", r"(?:org\.eclipse\.persistence\.exceptions\.JPQLException|Exception \[EclipseLink|Problem compiling \[|An exception occurred while creating a query)"),
     ("JPA / JPQL", r"(?:javax|jakarta)\.persistence\.(?:PersistenceException|Query(?:Syntax|Timeout)?Exception)"),
-    ("Generic HQL/JPQL", r"(?:HQL|JPQL|EJBQL)\b.*?(?:error|exception|syntax|not (?:mapped|resolve))"),
+    ("Generic HQL/JPQL", r"(?:HQL|JPQL|EJBQL)\b[^\n]{0,100}?(?:error|exception|syntax|not (?:mapped|resolve))"),
 )
 
 HQL_ERROR_REGEX = r"(?i)(?:%s)" % '|'.join(regex for _, regex in HQL_ERROR_SIGNATURES)
@@ -1484,7 +1487,7 @@ SPARQL_ERROR_SIGNATURES = (
     # too - matching on it alone mislabelled them as RDF4J, so only the package name is kept
     ("RDF4J / GraphDB", r"org\.eclipse\.rdf4j|org\.openrdf\.query"),
     ("Blazegraph", r"com\.bigdata\.rdf|\bBlazegraph\b"),
-    ("rdflib", r"rdflib\.plugins\.sparql|\bParseException\b.*?(?:SPARQL|sparql)"),
+    ("rdflib", r"rdflib\.plugins\.sparql|\bParseException\b[^\n]{0,100}?(?:SPARQL|sparql)"),
     ("Stardog", r"com\.(?:complexible\.)?stardog"),
 )
 

lib/core/target.py

@@ -561,7 +561,11 @@ def _resumeDBMS():
     dbms = value.lower()
     dbmsVersion = [UNKNOWN_DBMS_VERSION]
     _ = "(%s)" % ('|'.join(SUPPORTED_DBMS))
-    _ = re.search(r"\A%s (.*)" % _, dbms, re.I)
+    try:
+        # Dirty patch for a Python 3.13/3.14 re module regression (e.g. https://github.com/sqlmapproject/sqlmap/issues/6128, https://github.com/sqlmapproject/sqlmap/issues/6137)
+        _ = re.search(r"\A%s (.*)" % _, dbms, re.I)
+    except TypeError:
+        _ = None
 
     if _:
         dbms = _.group(1).lower()
@@ -666,8 +670,12 @@ def _createFilesDir():
     if not any((conf.fileRead, conf.commonFiles, conf.xxe)):
         return
 
-    # Note: normalize the hostname consistently with conf.outputPath / conf.dumpPath (see _createDumpDir)
-    conf.filePath = paths.SQLMAP_FILES_PATH % normalizeUnicode(getUnicode(conf.hostname))
+    # Note: normalize the hostname consistently with conf.outputPath / conf.dumpPath (see _createDumpDir).
+    # safeStringFormat() (not raw '%') because SQLMAP_FILES_PATH embeds SQLMAP_OUTPUT_PATH, which comes
+    # straight from --output-dir - a literal '%' anywhere in that user-supplied path (e.g. an unexpanded
+    # '$(date +%Y%m%d)' on a shell that doesn't support command substitution) would otherwise make the
+    # '%' operator try to parse it as a format directive and crash
+    conf.filePath = safeStringFormat(paths.SQLMAP_FILES_PATH, normalizeUnicode(getUnicode(conf.hostname)))
 
     if not os.path.isdir(conf.filePath):
         try:

lib/core/testing.py

@@ -268,7 +268,9 @@ def vulnTest(tests=None, label="vuln"):
 
         os.environ["SQLMAP_UNSAFE_EVAL"] = '1'
 
-        output = shellExec(cmd)
+        # bounded well above the slowest known entry (GraphQL, ~96s) - a hung entry fails fast and
+        # visibly instead of silently burning the whole CI job's timeout (see #6129 CI investigation)
+        output = shellExec(cmd, timeout=180)
 
         if not all((check in output if not check.startswith('~') else check[1:] not in output) for check in checks) or "unhandled exception" in output:
             dataToStdout("---\n\n$ %s\n" % cmd)

lib/request/basic.py

@@ -442,8 +442,8 @@ def processResponse(page, responseHeaders, code=None, status=None):
         rawResponse = "%s %s %s\n%s\n%s" % (_http_client.HTTPConnection._http_vsn_str, code or "", status or "", "".join(getUnicode(responseHeaders.headers if responseHeaders else [])), page[:IDENTYWAF_PARSE_PAGE_LIMIT] if not kb.checkWafMode else page[:HEURISTIC_PAGE_SIZE_THRESHOLD])
 
         with kb.locks.identYwaf:
-            identYwaf.non_blind.clear()
             try:
+                identYwaf.non_blind.clear()
                 if identYwaf.non_blind_check(rawResponse, silent=True):
                     for waf in set(identYwaf.non_blind):
                         if waf not in kb.identifiedWafs:

lib/request/connect.py

@@ -1072,6 +1072,10 @@ class Connect(object):
                     errMsg += "function '%s' ('%s')" % (function.__name__, getSafeExString(ex))
                     raise SqlmapGenericException(errMsg)
 
+                # a postprocess function may replace responseHeaders with a plain dict lacking
+                # the internal '.headers' attribute expected further down (e.g. in processResponse())
+                responseHeaders = patchHeaders(responseHeaders)
+
             if isinstance(page, six.binary_type):
                 if HTTP_HEADER.CONTENT_TYPE in (responseHeaders or {}) and not re.search(TEXT_CONTENT_TYPE_REGEX, responseHeaders[HTTP_HEADER.CONTENT_TYPE]):
                     page = six.text_type(page, errors="ignore")
@@ -1634,7 +1638,7 @@ class Connect(object):
                                     match = re.search(r"['\"]", post)
                                     if match:
                                         quote = match.group(0)
-                                        post = re.sub(r"\}\Z", "%s%s}" % (',' if re.search(r"\w", post) else "", "%s%s%s:%s" % (quote, name, quote, entry if entry.isdigit() else "%s%s%s" % (quote, entry, quote))), post)
+                                        post = re.sub(r"\}\Z", "%s%s}" % (',' if re.search(r"\w", post) else "", "%s%s%s:%s" % (quote, name, quote, entry if entry.isdigit() else "%s%s%s" % (quote, entry.replace('\\', r'\\'), quote))), post)
                                 else:
                                     post += "%s%s=%s" % (delimiter, name, entry)
                             elif get is not None:

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/httpshandler.py

@@ -91,7 +91,8 @@ class HTTPSConnection(_http_client.HTTPSConnection):
                     else:
                         for header, value in conf.httpHeaders:
                             if header.lower() == "host":
-                                hostname = value
+                                match = re.search(r"\A\[(?P<hostname>.+)\](:\d+)?\Z", value.strip())
+                                hostname = match.group("hostname") if match else value.rsplit(":", 1)[0] if ":" in value else value
                                 break
                     hostname = hostname if re.search(r"\A[\d.]+\Z", hostname or "") is None else None
                     result = _contexts[protocol].wrap_socket(sock, do_handshake_on_connect=True, server_hostname=hostname)

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."""
@@ -168,8 +297,251 @@ class Interactsh(object):
                 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).
+
+        Invariant: after stop()/close() returns, this object owns no active polling thread. If the
+        poller cannot be stopped within the budget it is deliberately left running (and this
+        interaction left registered) so a deregister can never race a still-active /poll drain -
+        a later stop()/close() re-runs it 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 every existing/future caller can
+        keep using it without learning about stop(); making it do the full shutdown means the
+        invariant holds regardless of which entry point is taken: after close() returns this
+        object owns no active polling thread."""
+        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 the
+        instant the loop wakes. This is why the stated invariant holds strictly - it is retried,
+        not capped at a guessed timeout."""
+        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 []

lib/techniques/blind/inference.py

@@ -67,6 +67,7 @@ from lib.core.settings import INFERENCE_MARKER
 from lib.core.settings import INFERENCE_NOT_EQUALS_CHAR
 from lib.core.settings import INFERENCE_UNKNOWN_CHAR
 from lib.core.settings import MAX_BISECTION_LENGTH
+from lib.core.settings import MAX_INT
 from lib.core.settings import MAX_REVALIDATION_STEPS
 from lib.core.settings import NULL
 from lib.core.settings import PARTIAL_HEX_VALUE_MARKER
@@ -864,7 +865,9 @@ def bisection(payload, expression, length=None, charsetType=None, firstChar=None
                             # elements we use a xrange, which is a virtual
                             # list
                             if expand and shiftTable:
-                                charTbl = xrange(maxChar + 1, (maxChar + 1) << shiftTable.pop())
+                                # Cap the expansion so its length never exceeds what len() can report
+                                # (a run of all-true probes would otherwise overflow on 32-bit builds)
+                                charTbl = xrange(maxChar + 1, min((maxChar + 1) << shiftTable.pop(), maxChar + 1 + MAX_INT))
                                 originalTbl = xrange(charTbl[0], charTbl[-1] + 1)
                                 maxChar = maxValue = charTbl[-1]
                                 minValue = charTbl[0]

lib/utils/api.py

@@ -265,9 +265,19 @@ def writeReportJson(collector, filepath):
     """
     Writes the collected results to filepath as JSON, in the same shape as the REST API's
     /scan/<id>/data response, wrapped with a small 'meta' block for standalone consumers.
+
+    A multi-target run (e.g. '-m' bulk file) stores each target under its own taskid (see
+    kb.reportTaskId), so every taskid present in the collector is assembled here; a single-target
+    run still yields exactly one taskid and keeps the flat {success, data, error} shape.
     """
 
-    result = _assembleData(collector, REPORT_TASKID)
+    taskids = sorted(row[0] for row in collector.execute("SELECT DISTINCT taskid FROM data")) or [REPORT_TASKID]
+
+    if len(taskids) > 1:
+        result = {"success": True, "targets": [_assembleData(collector, taskid) for taskid in taskids]}
+    else:
+        result = _assembleData(collector, taskids[0])
+
     result["meta"] = {
         "api_version": int(RESTAPI_VERSION.split(".")[0]),   # MAJOR only - the part that matters for client compatibility
         "sqlmap_version": VERSION_STRING,
@@ -467,7 +477,8 @@ class ReportErrorRecorder(logging.Handler):
 
     def emit(self, record):
         try:
-            self.collector.execute("INSERT INTO errors VALUES(NULL, ?, ?)", (REPORT_TASKID, str(record.msg % record.args if record.args else record.msg)))
+            taskid = kb.get("reportTaskId", REPORT_TASKID)
+            self.collector.execute("INSERT INTO errors VALUES(NULL, ?, ?)", (taskid, str(record.msg % record.args if record.args else record.msg)))
         except Exception:
             pass
 

lib/utils/hash.py

@@ -79,6 +79,7 @@ from lib.core.enums import HASH
 from lib.core.enums import MKSTEMP_PREFIX
 from lib.core.exception import SqlmapDataException
 from lib.core.exception import SqlmapUserQuitException
+from lib.core.patch import dirtyPatches
 from lib.core.settings import COMMON_PASSWORD_SUFFIXES
 from lib.core.settings import COMMON_PASSWORDS
 from lib.core.settings import COMMON_USER_COLUMNS
@@ -101,6 +102,16 @@ from thirdparty import six
 from thirdparty.colorama.initialise import init as coloramainit
 from thirdparty.six.moves import queue as _queue
 
+# On Windows, multiprocessing's 'spawn' start method boots a fresh interpreter per cracking
+# worker (_bruteProcessVariantA/B/Salted below) to unpickle its target function, which re-imports
+# this module from scratch without ever running the main process's startup dirtyPatches() - so
+# process-wide monkeypatches it installs (e.g. the 'reversible' codec error handler used by
+# getUnicode()/openFile()) are missing in the worker, raising LookupError on every hashed entry
+# (#6143). Placing the (idempotent) call here means that very re-import applies it. POSIX fork()
+# workers inherit the parent's patches already, so this only needs to run on Windows.
+if IS_WIN:
+    dirtyPatches()
+
 def mysql_passwd(password, uppercase=True):
     """
     Reference(s):
@@ -406,6 +417,8 @@ def crypt_generic_passwd(password, salt, **kwargs):
     'rl.3StKT.4T8M'
     """
 
+    password = getText(password)  # crypt.crypt() requires native str (unicode on Py3, bytes on Py2/PyPy), unlike the hashlib-based functions here
+
     return getText(crypt(password, salt))
 
 def unix_md5_passwd(password, salt, magic="$1$", **kwargs):

lib/utils/jwt.py

@@ -89,6 +89,8 @@ def crackHMAC(token, secrets, limit=None):
     's3cr3t'
     >>> crackHMAC(token, ["admin", "letmein"]) is None
     True
+    >>> crackHMAC("eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyIjoiYWRtaW4ifQ.x", ["secret"]) is None
+    True
     """
 
     data = parseJWT(token)
@@ -97,7 +99,13 @@ def crackHMAC(token, secrets, limit=None):
 
     fn = HMAC_ALGORITHMS[data["header"]["alg"].upper()]
     signingInput = getBytes(data["signingInput"])
-    target = decodeBase64(data["signature"], binary=True)
+
+    try:
+        target = decodeBase64(data["signature"], binary=True)
+    except Exception:
+        # a signature segment that is not valid base64url (e.g. a truncated/malformed token that still
+        # matches the JWT pattern) cannot be an HMAC we could verify, so it is simply not crackable
+        return None
 
     for index, secret in enumerate(secrets):
         if limit is not None and index >= limit:

plugins/dbms/mysql/connector.py

@@ -34,7 +34,7 @@ class Connector(GenericConnector):
 
         try:
             self.connector = pymysql.connect(host=self.hostname, user=self.user, passwd=self.password, db=self.db, port=self.port, connect_timeout=conf.timeout, use_unicode=True)
-        except (pymysql.OperationalError, pymysql.InternalError, pymysql.ProgrammingError, struct.error) as ex:
+        except (pymysql.OperationalError, pymysql.InternalError, pymysql.ProgrammingError, struct.error, UnicodeEncodeError) as ex:
             raise SqlmapConnectionException(getSafeExString(ex))
 
         self.initCursor()

sqlmap.py

@@ -8,10 +8,24 @@ See the file 'LICENSE' for copying permission
 from __future__ import print_function
 
 try:
+    import os
     import sys
 
     sys.dont_write_bytecode = True
 
+    # Reference: https://github.com/python/cpython/issues/156319 - CPython's tier-2 optimizer can
+    # corrupt frame locals (raising exceptions the executed code cannot produce), but only when the
+    # experimental JIT is on AND a sys.monitoring tool (debugger/profiler/coverage run) is active at
+    # the same time - neither alone triggers it. So only silence monitoring when the JIT is also on;
+    # otherwise leave it alone (e.g. a legitimate `coverage run sqlmap.py ...` must keep working).
+    if hasattr(sys, "monitoring") and (sys._jit.is_enabled() if hasattr(sys, "_jit") else (sys.version_info >= (3, 13) and os.environ.get("PYTHON_JIT") == '1')):
+        for _ in range(6):  # valid tool id range (Reference: https://docs.python.org/3/library/sys.monitoring.html)
+            try:
+                if sys.monitoring.get_tool(_) is not None:
+                    sys.monitoring.set_events(_, sys.monitoring.events.NO_EVENTS)
+            except Exception:
+                pass
+
     try:
         __import__("lib.utils.versioncheck")  # this has to be the first non-standard import
     except ImportError:
@@ -558,6 +572,17 @@ def main():
             logger.critical(errMsg)
             raise SystemExit
 
+        elif (sys._jit.is_enabled() if hasattr(sys, "_jit") else (sys.version_info >= (3, 13) and os.environ.get("PYTHON_JIT") == '1')):
+            errMsg = "the experimental Python JIT compiler is turned on. CPython has an open defect where "
+            errMsg += "the tier-2 optimizer runs the wrong instruction stream against a correct frame, raising "
+            errMsg += "exceptions that the executed code cannot produce "
+            errMsg += "(Reference: 'https://github.com/python/cpython/issues/156319'). Please rerun with it "
+            errMsg += "turned off (e.g. 'PYTHON_JIT=0') and report the problem only if it still occurs"
+            logger.critical(errMsg)
+            print()
+            dataToStdout(excMsg)
+            raise SystemExit
+
         for match in re.finditer(r'File "(.+?)", line', excMsg):
             file_ = match.group(1)
             try:

tests/test_common.py

@@ -113,6 +113,7 @@ from lib.core.common import (
     safeSQLIdentificatorNaming,
     saveConfig,
     serializeObject,
+    setColor,
     setTechnique,
     splitFields,
     trimAlphaNum,
@@ -1784,6 +1785,49 @@ class TestCheckOldOptions(unittest.TestCase):
         self.assertIsNone(checkOldOptions(["-u", "http://test.invalid/?id=1", "--banner"]))
 
 
+class TestSetColorHandlerMismatch(unittest.TestCase):
+    """
+    Regression test for issue #6122: setColor() decided whether to colorize purely from
+    conf.disableColoring/IS_TTY, independent of whether the installed LOGGER_HANDLER actually
+    supports colorize() (a plain logging.StreamHandler - installed for --disable-coloring, or as
+    the ansistrm-unavailable fallback - never defines it). A multiprocessing hash-cracking worker
+    hit exactly this mismatch (conf.disableColoring not carried over into the worker) and crashed
+    with an AttributeError that got misreported as "there was a problem while hashing entry".
+    """
+
+    def setUp(self):
+        import lib.core.common as common_mod
+        self._common_mod = common_mod
+        self._saved_handler = common_mod.LOGGER_HANDLER
+        self._saved_disableColoring = conf.get("disableColoring")
+
+    def tearDown(self):
+        self._common_mod.LOGGER_HANDLER = self._saved_handler
+        conf.disableColoring = self._saved_disableColoring
+
+    def test_plain_handler_without_colorize_does_not_raise(self):
+        import logging
+        self._common_mod.LOGGER_HANDLER = logging.StreamHandler()   # no .colorize(), like the --disable-coloring handler
+        conf.disableColoring = False    # the desync: coloring "should" apply, but handler can't
+        result = setColor("[INFO] current status: abcde", istty=True)   # must not raise
+        self.assertIsInstance(result, str)
+
+    def test_colorizing_handler_still_used(self):
+        # sanity check: a handler that DOES define colorize() is unaffected by the guard
+        calls = []
+
+        class _FakeColorizingHandler(object):
+            def colorize(self, message, levelno, force=False):
+                calls.append((message, levelno, force))
+                return "COLORIZED"
+
+        self._common_mod.LOGGER_HANDLER = _FakeColorizingHandler()
+        conf.disableColoring = False
+        result = setColor("[INFO] current status: abcde", istty=True)
+        self.assertEqual(result, "COLORIZED")
+        self.assertEqual(len(calls), 1)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)
 

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,33 @@ 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
+
+
+def _start_test_poller(client):
+    """Start a real InteractshPoller around a fake poll() without requiring pycryptodome.
+
+    Interactsh._ensurePoller() intentionally no-ops when crypto is unavailable because a real
+    interactsh /poll response cannot then be decrypted. These tests replace poll() with an
+    already-decoded in-memory fake, so their queue/poller behavior should not depend on whether
+    the test interpreter happens to have Crypto installed (notably PyPy 2.7 in CI).
+    """
+    from lib.request.interactsh import InteractshPoller
+    poller = getattr(client, "_poller", None)
+    if poller is None:
+        client._poller = poller = InteractshPoller(client)
+    poller.start()
+    return poller
+
+
 class _SendFailOnceSocket(object):
     """Wraps a real UDP socket; first sendto() raises (simulated transient failure)"""
     def __init__(self, real):
@@ -322,27 +350,62 @@ 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]
+                del names_batch[:]
+                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:
+            poller = _start_test_poller(srv._client)
+            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 +413,428 @@ 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[:]
+                del rec_batch[:]
+                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)
+
+        poller = _start_test_poller(srv._client)   # real queue/poller, fake already-decoded poll()
+        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[:]
+                del rec_batch[:]
+                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
+
+        poller = _start_test_poller(client)
+        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_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)

tests/test_esperanto.py

@@ -49,9 +49,15 @@ CORPUS = [
     u"\U00010348", u"Aé€\U00010348Z",
 ]
 
+# every connection _oracle()/_disguisedOracle() open outlives the helper call (the returned
+# closure keeps using it), so callers can't close it themselves - tracked here and closed once
+# in tearDownModule() instead of leaking to GC (ResourceWarning: unclosed database)
+_OPEN_CONNECTIONS = []
+
 
 def _oracle(value=None, is_null=False):
     con = sqlite3.connect(":memory:")
+    _OPEN_CONNECTIONS.append(con)
     con.execute("CREATE TABLE t (v TEXT)")
     con.execute("INSERT INTO t VALUES (?)", (None if is_null else value,))
     con.commit()
@@ -84,6 +90,7 @@ def _disguisedOracle(blocked=None, rewrites=(), funcs=()):
     `funcs` registers extra SQL functions (e.g. a variadic CONCAT SQLite lacks). The
     engine never sees SQLite - it must adapt to whatever back-end the scenario emulates."""
     con = sqlite3.connect(":memory:")
+    _OPEN_CONNECTIONS.append(con)
     for name, narg, fn in funcs:
         con.create_function(name, narg, fn)
     for stmt in _SEED:
@@ -221,6 +228,7 @@ class TestEsperanto(unittest.TestCase):
         # hostExtract must mirror the native EXACT/AMBIGUOUS verdict: in a mode with no byte-exact
         # witness (ordinal), both must be WHOLE_BUT_AMBIGUOUS, not native-ambiguous/host-exact.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('Admin-42')")
         con.commit()
@@ -249,6 +257,7 @@ class TestEsperanto(unittest.TestCase):
         # must reject the shortened token -> fall back to cell-by-cell (which reads the true
         # length via substring), so a 400-char value is recovered whole, not silently as 300.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
         big = "A" * 400
         con.execute("INSERT INTO t VALUES (1, ?)", (big,))
@@ -273,6 +282,7 @@ class TestEsperanto(unittest.TestCase):
         # operands -> bisection converges off-by-one. The final `expr = recovered` check must
         # reject it (fail closed) rather than return a silently-wrong integer.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
 
         def ask(cond):
             c = cond
@@ -302,6 +312,7 @@ class TestEsperanto(unittest.TestCase):
         # (LENGTH(..)>n, UNICODE(..)>n) converges off-by-one; the '=' confirmation must fail closed
         # rather than hand back silently-wrong code-point text as exact.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('Admin-42')")
         con.commit()
@@ -329,6 +340,7 @@ class TestEsperanto(unittest.TestCase):
         # (e.g. utf-8 bytes with an embedded NUL heuristically read as UTF-16). Bytes are faithful;
         # only a PROVEN codec makes the decoded text exact.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES (?)", (u"é",))            # utf-8 C3 A9 -> non-ASCII bytes
         con.execute("CREATE TABLE a (v TEXT)")
@@ -356,6 +368,7 @@ class TestEsperanto(unittest.TestCase):
         # Round 8 #5: once a byte-length witness is selected, an UNDECIDED reading must reject the
         # bytes (fail closed), never treat "couldn't verify" as "matched".
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('abc')")
         con.commit()
@@ -389,6 +402,7 @@ class TestEsperanto(unittest.TestCase):
         # never ran on the very failure it defends against. The canary must run whenever a textcast
         # was applied, and an unproven cast must not certify source identity.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
         con.execute(u"INSERT INTO t VALUES (1, 'café')")
         con.commit()
@@ -415,6 +429,7 @@ class TestEsperanto(unittest.TestCase):
         # the engine must escalate to the cast+hex path (which the framed dump proves works) and
         # recover the value exactly. Without hex, it stays honestly incomplete (no false-complete).
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute(u"INSERT INTO t VALUES ('café-€')")
         con.commit()
@@ -466,6 +481,7 @@ class TestEsperanto(unittest.TestCase):
         # False). the full truth table must reject 'gt' (and fall to BETWEEN) so counts/lengths
         # aren't read off-by-one. reproduced: extractInteger('5',max=10) used to return 6.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
 
         def ask(cond):
             try:
@@ -481,6 +497,7 @@ class TestEsperanto(unittest.TestCase):
         # without a proven hex/binary witness (or code-codepoint), a value is WHOLE_BUT_AMBIGUOUS,
         # never EXACT - plain '=' is collation-dependent and can't certify byte-exactness.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('Zz')")
         con.commit()
@@ -510,6 +527,7 @@ class TestEsperanto(unittest.TestCase):
 
     def test_host_maxlen_zero(self):
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('abcdef')")
         con.commit()
@@ -541,6 +559,7 @@ class TestEsperanto(unittest.TestCase):
         # `complete` (walk finished) must be distinct from `exact` (bytes proven identical).
         # a case-insensitive collation recovers a WHOLE value that is NOT exact.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('A')")
         con.commit()
@@ -568,6 +587,7 @@ class TestEsperanto(unittest.TestCase):
         # hostExtract returns an ExtractResult: a bounded read is TRUNCATED (not a bare string),
         # and an undecided (None) host observation degrades to a FAILED result, never a fake bit.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute("INSERT INTO t VALUES ('abcdef')")
         con.commit()
@@ -602,6 +622,7 @@ class TestEsperanto(unittest.TestCase):
         # key's first column repeats -> `> prev` paging silently drops rows sharing it, so it
         # must be REJECTED (COUNT(*) != COUNT(DISTINCT col)); only a unique column is accepted.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE k (a INT, b INT, u INT, nul INT)")
         con.executemany("INSERT INTO k VALUES (?,?,?,?)", [(1, 1, 10, 1), (1, 2, 20, None), (2, 1, 30, 3)])
         con.commit()
@@ -651,6 +672,7 @@ class TestEsperanto(unittest.TestCase):
 
     def test_enumerate(self):
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE alpha (x)")
         con.execute("CREATE TABLE beta (y)")
         con.commit()
@@ -668,6 +690,7 @@ class TestEsperanto(unittest.TestCase):
     def test_dump(self):
         # row DATA byte-exact, including commas / quotes / unicode (hex framing keeps intact)
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE users (id, uname, note)")
         truth = [(1, u"admin", u"all,good"), (2, u"o'brien", u"café,€"), (3, u"x", u"")]
         for row in truth:
@@ -692,6 +715,7 @@ class TestEsperanto(unittest.TestCase):
         # extraction. Quotes/commas/NULLs in the data must survive (a single value has
         # no framing ambiguity). This is the scavenger's whole reason to exist.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE users (id INTEGER, name TEXT, email TEXT)")
         for row in ((1, "luther", "[email protected]"), (2, "o'brien", "x,y@z"), (3, "wu", None)):
             con.execute("INSERT INTO users VALUES (?,?,?)", row)
@@ -723,6 +747,7 @@ class TestEsperanto(unittest.TestCase):
         # catalog detection, brute-force existence probing, and identifier quoting are
         # all COUNT-free (scalar-subquery existence), so tables/columns/dump still work.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE users (id INTEGER, name TEXT)")
         con.execute("INSERT INTO users VALUES (1, 'admin'), (2, 'root')")
         con.commit()
@@ -747,6 +772,7 @@ class TestEsperanto(unittest.TestCase):
     def test_quoting(self):
         # reserved-word / spaced column names must be quoted, not interpolated raw
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute('CREATE TABLE q (id INTEGER, "order" TEXT, "group by" TEXT)')
         con.execute('INSERT INTO q VALUES (1, ?, ?)', ("a'b", "x,y"))
         con.commit()
@@ -767,6 +793,7 @@ class TestEsperanto(unittest.TestCase):
     def test_strategy_handoff(self):
         # the frozen InferenceStrategy is a *sufficient* host interface, and immutable
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE k (v TEXT)")
         con.execute("INSERT INTO k VALUES ('Str4t3gy!')")
         con.commit()
@@ -788,6 +815,7 @@ class TestEsperanto(unittest.TestCase):
     def test_pattern_match_fallback(self):
         # SUBSTR + LENGTH + hex + code fns ALL blacklisted -> pure GLOB/LIKE floor
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE flag (id INTEGER, v TEXT)")
         con.execute("INSERT INTO flag VALUES (1, 'FLAG{no_SUBSTR_%_needed}')")
         con.commit()
@@ -812,6 +840,7 @@ class TestEsperanto(unittest.TestCase):
         # like 'all_products' must escape them (\_ ESCAPE '\'), not treat them as
         # match-anything. GLOB is blocked here to force LIKE (where '_'/'%' are magic).
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE v (val TEXT)")
         con.execute("INSERT INTO v VALUES ('all_products')")
         for t in ("all_products", "wp_users", "sales%2024"):     # '_'/'%' in identifiers
@@ -838,6 +867,7 @@ class TestEsperanto(unittest.TestCase):
     def test_length_from_substring(self):
         # every length fn blacklisted but SUBSTR present -> length derived from the end
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t6 (v TEXT)")
         con.execute(u"INSERT INTO t6 VALUES ('Admin-42€')")
         con.commit()
@@ -859,6 +889,7 @@ class TestEsperanto(unittest.TestCase):
         # a permission/charset wall mid-walk (oracle can't decide the keyset bound)
         # must STOP with partial results, never crash with OracleUndecided
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         for t in ("alpha", "beta", "gamma"):
             con.execute("CREATE TABLE %s (x)" % t)
         con.commit()
@@ -884,6 +915,7 @@ class TestEsperanto(unittest.TestCase):
         # RIGHT( onto them (the same rewrite the disguised-dialect oracles use). Keeps the
         # test portable across SQLite builds while still exercising the real left_right rung.
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t7 (v TEXT)")
         con.execute("INSERT INTO t7 VALUES ('Zagreb-42')")
         con.commit()
@@ -1029,6 +1061,7 @@ class TestEsperanto(unittest.TestCase):
         seed = ('CREATE TABLE q (id INTEGER, "order" TEXT, "group" TEXT)',
                 "INSERT INTO q VALUES (1, ?, ?)")
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute(seed[0]); con.execute(seed[1], ("a,b", "x'y")); con.commit()
         blk = re.compile(r'"|`')
 
@@ -1050,6 +1083,7 @@ class TestEsperanto(unittest.TestCase):
         # a first-byte charcode fn (MySQL-style ASCII) is lossy for non-ASCII; the
         # engine must detect that and escalate to hex, recovering the bytes exactly
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE t (v TEXT)")
         con.execute(u"INSERT INTO t VALUES ('café-€')")   # cafe-EUR
         con.commit()
@@ -1071,6 +1105,7 @@ class TestEsperanto(unittest.TestCase):
         # non-ASCII data recovered byte-exact even while the dialect is disguised
         # (SUBSTR/MID/CHAR_LENGTH/LENGTH blocked -> SUBSTRING/LEN mapped to SQLite)
         con = sqlite3.connect(":memory:")
+        self.addCleanup(con.close)
         con.execute("CREATE TABLE s (v TEXT)")
         con.execute(u"INSERT INTO s VALUES ('Zagreb-župa-€42')")   # Zagreb-zupa-EUR42
         con.commit()
@@ -1091,5 +1126,13 @@ class TestEsperanto(unittest.TestCase):
         self.assertEqual(esp.extractText("(SELECT v FROM s)"), u"Zagreb-župa-€42")
 
 
+def tearDownModule():
+    while _OPEN_CONNECTIONS:
+        try:
+            _OPEN_CONNECTIONS.pop().close()
+        except Exception:
+            pass
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

tests/test_hash.py

@@ -87,6 +87,22 @@ class TestDbmsSpecificVectors(unittest.TestCase):
         self.assertEqual(H.oracle_old_passwd("tiger", "scott", uppercase=True), "F894844C34402B67")
 
 
+class TestCryptGenericBytesInput(unittest.TestCase):
+    """
+    The dictionary-attack workers (_bruteProcessVariantA/B in lib/utils/hash.py) always pass
+    'password' as bytes (getBytes(word)), regardless of which hash function gets dispatched. Every
+    other function here hashes via hashlib (bytes-friendly), but crypt_generic_passwd forwards
+    straight to the stdlib crypt.crypt(), which requires str - raising 'TypeError: crypt() argument
+    1 must be str, not bytes' on Linux/macOS (reported via a user's cracked-hash session).
+    """
+
+    def test_bytes_password_does_not_raise(self):
+        self.assertEqual(H.crypt_generic_passwd(password=b"rasmuslerdorf", salt="rl", uppercase=False), "rl.3StKT.4T8M")
+
+    def test_str_password_unaffected(self):
+        self.assertEqual(H.crypt_generic_passwd(password="rasmuslerdorf", salt="rl", uppercase=False), "rl.3StKT.4T8M")
+
+
 class TestMssqlUnicodePassword(unittest.TestCase):
     """MSSQL hashes the password as UCS-2/UTF-16LE. A per-char 'utf-8 + NUL' approximation is only
     correct for ASCII, so non-ASCII passwords (cafe, etc.) hashed WRONG and were uncrackable. The

tests/test_heuristic_signatures.py

@@ -26,6 +26,7 @@ stdlib unittest only (no pytest / no pip); works on Python 2.7 and 3.x.
 import os
 import re
 import sys
+import time
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -223,6 +224,26 @@ class HeuristicSignatureTest(unittest.TestCase):
             self.assertEqual(fired, (),
                              msg="%s output suggests %s: %r" % (backend, '/'.join("'--%s'" % _ for _ in fired), text))
 
+    def test_no_signature_has_an_unbounded_gap(self):
+        # an unbounded '.*' between a literal and a keyword is quadratic on a long line (minified
+        # JS/JSON, a one-line JSON error body), which is how a stray 'xpath' in a 400KB response
+        # cost ~25s per parameter - and made the regex engine itself blow up (#5994, #6105).
+        # The bounded form ('[^\n]{0,100}?') matches the same real errors in constant work
+        for name, regex in ENGINES:
+            found = re.search(r"(?<!\\)\.[*+]", regex)
+            self.assertIsNone(found, msg="'--%s' signatures carry an unbounded '%s' gap" % (name, found.group(0) if found else ""))
+
+    def test_signatures_stay_linear_on_a_long_line(self):
+        # the same invariant, measured: every engine has to survive a single-line response that
+        # carries the words its signatures start with, without any error actually being present
+        page = ("<div class=\"x\">lorem ipsum dolor sit amet consectetur adipiscing elit</div>" + ''.join("%s " % _ for _ in ("xpath", "HQL", "ParseException", "NameError", "Entity x", "handlebars", "no such file", "Twig"))) * 2000
+
+        for name, regex in ENGINES:
+            start = time.time()
+            re.search(regex, page)
+            elapsed = time.time() - start
+            self.assertLess(elapsed, 2, msg="'--%s' signatures took %.1fs on a %dKB single-line response" % (name, elapsed, len(page) // 1024))
+
     def test_every_engine_is_covered(self):
         # a new switch must arrive here with its own errors, or the matrix above proves nothing about it
         owners = set(owner for owner, _, _ in CORPUS)

tests/test_html_parser_forks.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _testutils import bootstrap
+bootstrap()
+
+from lib.core.data import kb
+from lib.core.enums import DBMS
+from lib.core.enums import FORK
+from lib.parse.html import htmlParser
+
+
+class TestHtmlParserForks(unittest.TestCase):
+    def setUp(self):
+        kb.forkNote = None
+        kb.cache.parsedDbms.clear()
+
+    # NOTE: these forks all speak the PostgreSQL wire protocol and inherit its error wording, so
+    # the brand name itself never appears next to 'ERROR:' - only a leaked driver package or an
+    # engine-internal source reference tells them apart from stock PostgreSQL
+    def test_cockroachdb_error_fingerprint(self):
+        for page in (
+            "ERROR: relation \"crdb_internal.tables\" does not exist",
+            "DETAIL: stack trace: github.com/cockroachdb/cockroach/pkg/sql/conn_executor.go:733",
+        ):
+            kb.forkNote = None
+            kb.cache.parsedDbms.clear()
+            self.assertEqual(htmlParser(page), DBMS.PGSQL)
+            self.assertEqual(kb.forkNote, FORK.COCKROACHDB)
+
+    def test_yugabytedb_error_fingerprint(self):
+        for page in (
+            "com.yugabyte.util.PSQLException: ERROR: syntax error at or near \"'\"",
+            "ERROR:  Query error: [Query error (yb/tserver/read_query.cc:265): MISMATCHED_SCHEMA]",
+        ):
+            kb.forkNote = None
+            kb.cache.parsedDbms.clear()
+            self.assertEqual(htmlParser(page), DBMS.PGSQL)
+            self.assertEqual(kb.forkNote, FORK.YUGABYTEDB)
+
+    def test_opengauss_error_fingerprint(self):
+        page = "org.opengauss.util.PSQLException: ERROR: syntax error at or near \"'\""
+        dbms = htmlParser(page)
+        self.assertEqual(dbms, DBMS.PGSQL)
+        self.assertEqual(kb.forkNote, FORK.OPENGAUSS)
+
+    def test_duckdb_error_fingerprint(self):
+        # Messages below are verbatim from DuckDB 1.1.3/1.5.5 (the module was renamed
+        #'duckdb.duckdb' -> '_duckdb')
+        for page in (
+            "_duckdb.ParserException: Parser Error: unterminated quoted string at or near \"'''\"",
+            "duckdb.duckdb.CatalogException: Catalog Error: Table with name t does not exist!",
+            "_duckdb.BinderException: Binder Error: Referenced column \"x\" not found in FROM clause!",
+            "at org.duckdb.DuckDBPreparedStatement.execute(DuckDBPreparedStatement.java:180)",
+        ):
+            kb.forkNote = None
+            kb.cache.parsedDbms.clear()
+            self.assertEqual(htmlParser(page), DBMS.PGSQL)
+            self.assertEqual(kb.forkNote, FORK.DUCKDB)
+
+    def test_trino_error_fingerprint(self):
+        # the generic Presto 'mismatched input' entry also matches, so this pins the ordering
+        # that keeps the Trino fork note from being swallowed by it
+        page = "io.trino.jdbc.TrinoSQLException: line 1:15: mismatched input 'FROM'. Expecting: <expression>"
+        dbms = htmlParser(page)
+        self.assertEqual(dbms, DBMS.PRESTO)
+        self.assertEqual(kb.forkNote, FORK.TRINO)
+
+    def test_doris_error_fingerprint(self):
+        # Doris speaks the MySQL wire protocol, so the client-side exception is MySQL's own -
+        # only the leaked FE package name tells the fork apart
+        page = "com.mysql.jdbc.exceptions.MySQLSyntaxErrorException at org.apache.doris.qe.ConnectProcessor"
+        dbms = htmlParser(page)
+        self.assertEqual(dbms, DBMS.MYSQL)
+        self.assertEqual(kb.forkNote, FORK.DORIS)
+
+    def test_starrocks_error_fingerprint(self):
+        for page in (
+            "com.mysql.jdbc.exceptions.MySQLSyntaxErrorException at com.starrocks.qe.ConnectProcessor",
+            "(1064, 'Getting syntax error from line 1, column 7 to line 1, column 62. Detail message: x.')",
+        ):
+            kb.forkNote = None
+            kb.cache.parsedDbms.clear()
+            self.assertEqual(htmlParser(page), DBMS.MYSQL)
+            self.assertEqual(kb.forkNote, FORK.STARROCKS)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_jit_guard.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+
+The last branch of main()'s unhandled-exception chain in sqlmap.py: when the
+experimental Python JIT is turned on, a crash gets the tier-2 advisory (Reference:
+'https://github.com/python/cpython/issues/156319') instead of the automatic
+issue-creation prompt - but the traceback is still printed, so a genuine sqlmap
+bug is not swallowed.
+
+Driven through a subprocess because it is main()'s own except chain under test.
+"""
+
+import os
+import subprocess
+import sys
+import unittest
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# Forces a crash inside init(), i.e. the same place #6117 landed, without needing a target.
+# NOTE: the injected fault must raise on Python 2 as well - 'a' < 128 does NOT (Python 2 orders
+# mismatched types by type name and quietly returns a bool), which let a broken run reach the
+# network on the PyPy-2.7 job instead of the exception handler under test.
+# codeIsModified() is neutralized because its branch sits earlier in the chain and would
+# otherwise win on any working tree with uncommitted edits (or a stale txt/checksum.md5).
+DRIVER = """
+import sys
+sys.path.insert(0, %r)
+sys.argv = ["sqlmap.py", "-u", "http://127.0.0.1/?id=1", "--batch"]
+import sqlmap
+import lib.core.option
+sqlmap.codeIsModified = lambda: False
+lib.core.option.loadPayloads = lambda: iter([])()
+sqlmap.main()
+""" % ROOT
+
+ADVISORY = "experimental Python JIT compiler is turned on"
+TRACEBACK = "TypeError"
+PROMPT = "automatically create a new (anonymized) issue"
+
+
+def _run(jit):
+    env = dict(os.environ)
+    env["PYTHON_JIT"] = jit
+    proc = subprocess.Popen([sys.executable, "-c", DRIVER], cwd=ROOT, env=env,
+                            stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                            stderr=subprocess.STDOUT)
+    out = proc.communicate(input=b"")[0]
+    return out.decode("utf-8", "replace")
+
+
+def _jitEnabled(jit):
+    """Whether the guard in sqlmap.py would consider the JIT enabled (mirrors its condition)."""
+    code = ("import sys, os; print(sys._jit.is_enabled() if hasattr(sys, '_jit') "
+            "else (sys.version_info >= (3, 13) and os.environ.get('PYTHON_JIT') == '1'))")
+    env = dict(os.environ)
+    env["PYTHON_JIT"] = jit
+    return subprocess.check_output([sys.executable, "-c", code], env=env).strip() == b"True"
+
+
+class TestJitGuard(unittest.TestCase):
+    def test_advisory_replaces_issue_prompt_but_keeps_traceback(self):
+        if not _jitEnabled('1'):
+            self.skipTest("interpreter does not report the JIT as enabled")
+
+        out = _run('1')
+        self.assertIn(ADVISORY, out)
+        self.assertIn(TRACEBACK, out)          # a real bug must still be visible in full
+        self.assertNotIn(PROMPT, out)          # ... but not auto-filed while the JIT is on
+
+    def test_normal_path_is_untouched_without_jit(self):
+        if _jitEnabled('0'):
+            self.skipTest("JIT stays enabled with PYTHON_JIT=0 on this build")
+
+        out = _run('0')
+        self.assertNotIn(ADVISORY, out)
+        self.assertIn(TRACEBACK, out)
+        self.assertIn(PROMPT, out)
+
+
+# Tool id 3 is unused by CPython's own well-known ids (DEBUGGER=0, COVERAGE=1, PROFILER=2, OPTIMIZER=5)
+_TEST_TOOL_ID = 3
+
+_MONITORING_DRIVER = """
+import sys
+sys.path.insert(0, %r)
+sys.monitoring.use_tool_id(%d, "test-tool")
+sys.monitoring.set_events(%d, sys.monitoring.events.PY_START)
+import sqlmap
+print(sys.monitoring.get_tool(%d))
+print(sys.monitoring.get_events(%d))
+""" % (ROOT, _TEST_TOOL_ID, _TEST_TOOL_ID, _TEST_TOOL_ID, _TEST_TOOL_ID)
+
+
+def _monitoringStateAfterImport(jit):
+    env = dict(os.environ)
+    env["PYTHON_JIT"] = jit
+    out = subprocess.check_output([sys.executable, "-c", _MONITORING_DRIVER], cwd=ROOT, env=env)
+    tool, events = out.decode("utf-8").strip().splitlines()
+    return tool, int(events)
+
+
+class TestMonitoringGuard(unittest.TestCase):
+    """
+    The other half of the cpython#156319 mitigation: the tier-2 corruption needs BOTH the JIT and
+    an active sys.monitoring tool (debugger/profiler/coverage) at once - neither alone triggers it.
+    So sqlmap only silences an already-registered tool's events when the JIT is also on, and leaves
+    it alone otherwise. A plain `coverage run sqlmap.py ...` (JIT off) must keep working - which is
+    exactly what broke CI the first time this guard shipped unconditionally.
+    """
+
+    def test_active_tool_is_silenced_when_jit_is_on(self):
+        if not hasattr(sys, "monitoring"):
+            self.skipTest("interpreter has no sys.monitoring (needs 3.12+)")
+        if not _jitEnabled('1'):
+            self.skipTest("interpreter does not report the JIT as enabled")
+
+        tool, events = _monitoringStateAfterImport('1')
+        self.assertEqual(tool, "test-tool")    # still registered - its own owner can still free it
+        self.assertEqual(events, 0)            # but events cleared to NO_EVENTS
+
+    def test_active_tool_is_left_alone_without_jit(self):
+        if not hasattr(sys, "monitoring"):
+            self.skipTest("interpreter has no sys.monitoring (needs 3.12+)")
+        if _jitEnabled('0'):
+            self.skipTest("JIT stays enabled with PYTHON_JIT=0 on this build")
+
+        tool, events = _monitoringStateAfterImport('0')
+        self.assertEqual(tool, "test-tool")
+        self.assertNotEqual(events, 0)         # untouched - e.g. a real `coverage run` must keep working
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report.py

@@ -222,5 +222,109 @@ class TestWriteReportJson(_CollectorCase):
             os.remove(path)
 
 
+class TestMultiTargetTaskId(_CollectorCase):
+    """
+    Regression coverage for issue #6123: a '-m' bulk-file run shares ONE process (and thus one
+    report collector) across many targets. _storeData()'s COMPLETE-status branch deletes any
+    existing row for a (taskid, content_type) key before inserting the new one, so if every
+    target reused the same fixed REPORT_TASKID, a later target's TARGET/TECHNIQUES write would
+    silently delete an earlier target's row of the same content_type. The fix keys each target's
+    writes by kb.reportTaskId instead.
+    """
+
+    def setUp(self):
+        super(TestMultiTargetTaskId, self).setUp()
+        from lib.core.dump import Dump
+        self._had_reportTaskId = "reportTaskId" in kb
+        self._saved_reportTaskId = kb.get("reportTaskId")
+        self._saved_dumper = conf.get("dumper")
+        self._saved_reportCollector = conf.get("reportCollector")
+        conf.dumper = Dump()
+        conf.reportCollector = self.c
+
+    def tearDown(self):
+        # restore to the PRIOR STATE exactly - if the key was absent before (the common case, since
+        # kb.reportTaskId is normally set lazily by a multi-target run), setting it to the saved None
+        # would instead leave it explicitly present with value None, which breaks the
+        # kb.get("reportTaskId", REPORT_TASKID) fallback used elsewhere (e.g. ReportErrorRecorder.emit())
+        # for the rest of the test process - order-dependent flakiness, since Python 2's unittest
+        # discover() does not sort tests/ (unlike Python 3's), so this class can run before
+        # TestReportErrorCapture depending on the checkout's raw directory order
+        if self._had_reportTaskId:
+            kb.reportTaskId = self._saved_reportTaskId
+        else:
+            del kb.reportTaskId
+        conf.dumper = self._saved_dumper
+        conf.reportCollector = self._saved_reportCollector
+        super(TestMultiTargetTaskId, self).tearDown()
+
+    def test_second_target_does_not_overwrite_first(self):
+        kb.reportTaskId = 1
+        conf.dumper._reportData({"url": "http://host1/?id=1"}, CONTENT_TYPE.TARGET)
+
+        kb.reportTaskId = 2
+        conf.dumper._reportData({"url": "http://host2/?id=1"}, CONTENT_TYPE.TARGET)
+
+        first = api._assembleData(self.c, 1)["data"]
+        second = api._assembleData(self.c, 2)["data"]
+        self.assertEqual(first[0]["value"]["url"], "http://host1/?id=1")    # not clobbered by target #2
+        self.assertEqual(second[0]["value"]["url"], "http://host2/?id=1")
+
+    def test_write_report_json_wraps_multiple_targets(self):
+        kb.reportTaskId = 1
+        conf.dumper._reportData({"url": "http://host1/?id=1"}, CONTENT_TYPE.TARGET)
+        kb.reportTaskId = 2
+        conf.dumper._reportData({"url": "http://host2/?id=1"}, CONTENT_TYPE.TARGET)
+
+        fd, path = tempfile.mkstemp(suffix=".json")
+        os.close(fd)
+        try:
+            api.writeReportJson(self.c, path)
+            with io.open(path, encoding="utf-8") as f:
+                loaded = json.load(f)
+            self.assertIn("targets", loaded)
+            self.assertEqual(len(loaded["targets"]), 2)
+            urls = [t["data"][0]["value"]["url"] for t in loaded["targets"]]
+            self.assertEqual(urls, ["http://host1/?id=1", "http://host2/?id=1"])
+        finally:
+            os.remove(path)
+
+    def test_single_target_report_keeps_flat_shape(self):
+        # backward compatibility: exactly one taskid -> no 'targets' wrapper, same shape as before
+        kb.reportTaskId = 1
+        conf.dumper._reportData({"url": "http://host1/?id=1"}, CONTENT_TYPE.TARGET)
+
+        fd, path = tempfile.mkstemp(suffix=".json")
+        os.close(fd)
+        try:
+            api.writeReportJson(self.c, path)
+            with io.open(path, encoding="utf-8") as f:
+                loaded = json.load(f)
+            self.assertNotIn("targets", loaded)
+            self.assertEqual(loaded["data"][0]["value"]["url"], "http://host1/?id=1")
+        finally:
+            os.remove(path)
+
+    def test_error_recorded_under_active_target(self):
+        import logging
+        from lib.core.data import logger
+
+        saved_level = logger.level
+        logger.setLevel(logging.ERROR)
+        # mute pre-existing handlers (e.g. console) but not the ReportErrorRecorder added by setUp
+        muted = [(handler, handler.level) for handler in logger.handlers if not isinstance(handler, api.ReportErrorRecorder)]
+        for handler, _ in muted:
+            handler.setLevel(logging.CRITICAL + 1)
+        try:
+            kb.reportTaskId = 2
+            logger.error("boom for target 2")
+            self.assertTrue(any("boom for target 2" in _ for _ in api._assembleData(self.c, 2)["error"]))
+            self.assertEqual(api._assembleData(self.c, 1)["error"], [])    # not attributed to target #1
+        finally:
+            logger.setLevel(saved_level)
+            for handler, level in muted:
+                handler.setLevel(level)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

tests/test_ssti.py

@@ -827,3 +827,7 @@ class TestStruts2Header(unittest.TestCase):
         # so no start/end marker is found -> no fabricated 'output'
         ssti._s2045Send = lambda url, action: "reflected: %s" % action
         self.assertIsNone(ssti._executeStruts2Header("http://target", "id"))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/test_target_parsing.py

@@ -51,6 +51,7 @@ from lib.core.settings import RESTORE_MERGED_OPTIONS
 from lib.core.settings import UNENCODED_ORIGINAL_VALUE
 from lib.core.threads import getCurrentThreadData
 from lib.utils.hashdb import HashDB
+import lib.core.target as target
 from lib.core.target import _createDumpDir
 from lib.core.target import _createFilesDir
 from lib.core.target import _createTargetDirs
@@ -354,6 +355,31 @@ class TestResumeDBMS(_TargetTestBase):
         with self.assertRaises(SqlmapNoneDataException):
             _resumeDBMS()
 
+    def test_survives_broken_re_module(self):
+        # a Python 3.13/3.14 're' module regression can make re.search() raise TypeError on an
+        # otherwise-valid pattern (#6128, #6137) - must fall back to "no match" (the stored value
+        # used as-is) instead of crashing the whole resume
+        self._new_hashdb()
+        conf.dbms = None
+        conf.offline = False
+        hashDBWrite(HASHDB_KEYS.DBMS, "MySQL 5.0")
+
+        originalRe = target.re
+
+        class _BrokenRe(object):
+            I = originalRe.I
+
+            def search(self, *args, **kwargs):
+                raise TypeError("list.append() takes exactly one argument (2 given)")
+
+        target.re = _BrokenRe()
+        try:
+            _resumeDBMS()
+        finally:
+            target.re = originalRe
+
+        self.assertIsNone(Backend.getIdentifiedDbms())
+
 
 class TestResumeOS(_TargetTestBase):
     def test_resumes_os(self):
@@ -483,6 +509,18 @@ class TestCreateDirs(_TargetTestBase):
         self.assertTrue(os.path.isdir(conf.filePath))
         self.assertIn("example.com", conf.filePath)
 
+    def test_files_dir_survives_percent_in_output_dir(self):
+        # --output-dir carries a literal '%' straight into SQLMAP_FILES_PATH (e.g. an unexpanded
+        # '$(date +%Y%m%d)' on a shell without command substitution, as in #6133) - the '%' operator
+        # would try to parse '%Y' as a format directive and crash; safeStringFormat() must not
+        self._outdir("f_out_%Y%m%d_%H%M%S")
+        conf.hostname = "example.com"
+        conf.fileRead = "/etc/passwd"
+        conf.commonFiles = None
+        _createFilesDir()
+        self.assertTrue(os.path.isdir(conf.filePath))
+        self.assertIn("example.com", conf.filePath)
+
     def test_target_dir_and_target_txt(self):
         self._outdir("t_out")
         conf.hostname = "example.com"

tests/test_xpath.py

@@ -557,3 +557,7 @@ class TestRealXPathSyntax(unittest.TestCase):
                 self.fail("Boundary '%s' in '%s' with orig='%s' invalid: %s\n  payload: %s" % (bk, tkey, original, e, payload))
             self.assertIsInstance(count, int,
                 "Boundary '%s' in '%s' produced no count" % (bk, tkey))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)