Tons of stabilization of non-SQLi techniques

tests/test_graphql.py

@@ -264,15 +264,63 @@ class TestGraphqlBooleanDetection(unittest.TestCase):
 
     def test_boolean_detected(self):
         slot = _slot("query", "Query", "user", "username", "string")
-        oracleType, template = gi._detectBoolean(slot, "http://test/graphql")
+        oracleType, template, _win = gi._detectBoolean(slot, "http://test/graphql")
         self.assertIsNotNone(oracleType)
         self.assertIn("boolean-based", oracleType)
 
     def test_numeric_skipped(self):
         slot = _slot("query", "Query", "byId", "id", "numeric")
-        oracleType, template = gi._detectBoolean(slot, "http://test/graphql")
+        oracleType, template, _win = gi._detectBoolean(slot, "http://test/graphql")
         self.assertIsNone(oracleType)
 
+    def test_graphql_two_true_transport_failures_do_not_confirm(self):
+        # the TRUE query fails transport (-> None), the FALSE succeeds: two None trues must NOT be
+        # read as a reproducible page that "differs" from false (the classic fabricated confirmation)
+        def fakeSend(endpoint, query, variables=None):
+            if "'1'='1" in query:
+                return None, 0            # transport failure on the true payload
+            return NOMATCH, 200
+        gi._gqlSend = fakeSend
+        slot = _slot("query", "Query", "user", "username", "string")
+        oracleType, _, _win = gi._detectBoolean(slot, "http://test/graphql")
+        self.assertIsNone(oracleType)
+
+    def test_graphql_false_page_is_replayed(self):
+        # a FALSE page that does not reproduce (jitter) must not establish an oracle
+        state = {"n": 0}
+        def fakeSend(endpoint, query, variables=None):
+            if "'1'='1" in query:
+                return MATCH, 200
+            state["n"] += 1
+            if state["n"] % 2:                                    # false response is unstable (jitter)
+                return '{"data":{"user":{"id":1,"name":"alpha"}}}', 200
+            return '{"data":{"user":{"totally":"different","shape":"here","x":12345,"y":67890}}}', 200
+        gi._gqlSend = fakeSend
+        slot = _slot("query", "Query", "user", "username", "string")
+        oracleType, _, _win = gi._detectBoolean(slot, "http://test/graphql")
+        self.assertIsNone(oracleType)
+
+    def test_graphql_resolver_error_false_is_not_a_boolean_oracle(self):
+        # P0-3: the FALSE payload trips a stable HTTP-200 resolver error ({data:null, errors:[...]}),
+        # which yields the same {"user":null} observation as a genuine false. That is NOT a boolean
+        # oracle (it belongs to error-based detection) - _detectBoolean must reject the errored pair.
+        def fakeSend(endpoint, query, variables=None):
+            if "'1'='1" in query:                                 # true -> real rows
+                return MATCH, 200
+            return '{"data":{"user":null},"errors":[{"message":"resolver failed","path":["user"]}]}', 200
+        gi._gqlSend = fakeSend
+        slot = _slot("query", "Query", "user", "username", "string")
+        oracleType, _, _win = gi._detectBoolean(slot, "http://test/graphql")
+        self.assertIsNone(oracleType)
+
+    def test_has_errors_and_alias_errored(self):
+        self.assertTrue(gi._hasErrors('{"data":{"user":null},"errors":[{"message":"x"}]}'))
+        self.assertFalse(gi._hasErrors('{"data":{"user":null}}'))
+        # an alias with an error path, or an absent alias, is errored/unknown
+        self.assertTrue(gi._aliasErrored('{"data":{"a0":null},"errors":[{"message":"e","path":["a0"]}]}', "a0"))
+        self.assertTrue(gi._aliasErrored('{"data":{"a1":true}}', "a0"))     # a0 absent
+        self.assertFalse(gi._aliasErrored('{"data":{"a0":true}}', "a0"))
+
 
 class TestGraphqlErrorDetection(unittest.TestCase):
     """Error-based detection via mock oracle"""
@@ -294,9 +342,30 @@ class TestGraphqlErrorDetection(unittest.TestCase):
 
     def test_error_detected(self):
         slot = _slot("query", "Query", "user", "username", "string")
-        oracleType, detail = gi._detectError(slot, "http://test/graphql")
+        oracleType, detail, _win = gi._detectError(slot, "http://test/graphql")
         self.assertEqual(oracleType, "error-based")
 
+    def test_report_shows_winning_error_payload_not_boolean(self):
+        # boolean payloads do NOT diverge (both -> NOMATCH) so boolean detection fails; only the error
+        # payloads trip a DB error. The reported reproducer must be the WINNING error payload, never the
+        # generic ' OR '1'='1 boolean payload.
+        def fakeSend(endpoint, query, variables=None):
+            if "'1'='" in query:               # both boolean payloads ('1'='1 / '1'='2) -> identical, no oracle
+                return NOMATCH, 200
+            if "'" in query:                    # error payloads (', '', '") -> DB error
+                return DB_ERROR, 500
+            return NOMATCH, 200
+        gi._gqlSend = fakeSend
+        reports = []
+        gi.conf.dumper = type("D", (), {"singleString": lambda self, m: reports.append(m)})()
+        gi.conf.beep = False
+        slot = _slot("query", "Query", "user", "username", "string")
+        oracleType, _oracle, _detail = gi._testSlot(slot, "http://test/graphql")
+        self.assertEqual(oracleType, "error-based")
+        report = next(r for r in reports if "Payload:" in r)
+        self.assertNotIn("'1'='1", report)                 # not the boolean payload
+        self.assertIn("error-based", report)
+
 
 class TestGraphqlParseRows(unittest.TestCase):
     """JSON data row parsing for in-band dumps"""
@@ -443,8 +512,23 @@ class TestGraphqlDialects(unittest.TestCase):
         d = gi.DIALECTS["SQLite"]
         sql = d.row(["name", "surname"], "users", 3)
         self.assertIn("LIMIT 1 OFFSET 3", sql)                       # per-row, not a whole-table GROUP_CONCAT
-        self.assertIn("COALESCE(CAST(name AS TEXT),'NULL')", sql)
-        self.assertIn("FROM users", sql)
+        self.assertIn('COALESCE(CAST("name" AS TEXT),\'NULL\')', sql)   # column identifier quoted
+        self.assertIn('FROM "users"', sql)                              # table identifier quoted
+
+    def test_row_quotes_reserved_and_mixedcase_identifiers(self):
+        # a reserved word / mixed-case / spaced / quote-bearing name must be quoted, not interpolated raw
+        d = gi.DIALECTS["PostgreSQL"]
+        sql = d.row(["order", 'we"ird'], "myTable", 0)
+        self.assertIn('CAST("order" AS TEXT)', sql)
+        self.assertIn('CAST("we""ird" AS TEXT)', sql)                  # embedded quote doubled
+        d2 = gi.DIALECTS["Microsoft SQL Server"]
+        self.assertIn("[order]", d2.row(["order"], "dbo.t", 0))
+
+    def test_pgsql_schema_qualified_from(self):
+        # a "schema.table" catalog name qualifies AND quotes both parts for the dump FROM
+        d = gi.DIALECTS["PostgreSQL"]
+        self.assertIn('FROM "secret"."users"', d.row(["id"], "secret.users", 0))
+        self.assertEqual(d.fromIdent("secret.users"), '"secret"."users"')
 
     def test_mysql_uses_sleep_delay(self):
         d = gi.DIALECTS["MySQL"]
@@ -529,7 +613,7 @@ def _mockOracle(target):
         tableFrom=None, tableCol=None, columnFrom=None, columnCol=None, paginate=None,
         length=lambda expr: "LEN(%s)" % expr,
         ordinal=lambda expr, pos: "ORD(%s,%d)" % (expr, pos),
-        row=None)
+        row=None, fromIdent=lambda table: table)
 
     def _value(cond):
         pos = None
@@ -579,6 +663,43 @@ class TestGraphqlInference(unittest.TestCase):
         dialect, truth, truthBatch = _mockOracle("")
         self.assertEqual(gi._inferExprBatched(truthBatch, truth, dialect, "EXPR"), "")
 
+    def test_inconclusive_truth_aborts_value_not_fabricates(self):
+        # a persistently-inconclusive oracle must abort the value (None), never coerce to false bits
+        dialect = gi.DIALECTS["SQLite"]
+
+        def truth(cond):
+            raise gi.InconclusiveError()
+
+        def truthBatch(conds):
+            raise gi.InconclusiveError()
+
+        self.assertIsNone(gi._inferExpr(truth, dialect, "EXPR"))
+        self.assertIsNone(gi._inferExprBatched(truthBatch, truth, dialect, "EXPR"))
+
+    def test_make_oracle_batch_transport_failure_raises_not_false(self):
+        # a FAILED batch request must raise InconclusiveError, NOT decay into a list of False bits
+        # (which would silently corrupt every value extracted through the batch path)
+        slot = _slot("query", "Query", "user", "username", "string")
+        MATCHV = '{"data":{"user":{"id":1,"name":"luther"}}}'
+        NOMATCHV = '{"data":{"user":null}}'
+        saved = gi._gqlSend
+        try:
+            def fakeSend(endpoint, query, variables=None):
+                if "1=1" in query:
+                    return MATCHV, 200
+                if "1=2" in query:
+                    return NOMATCHV, 200
+                return MATCHV, 200
+            gi._gqlSend = fakeSend
+            truth, truthBatch = gi._makeOracle(slot, "http://test/graphql")
+            self.assertIsNotNone(truth)
+
+            # now make the batch endpoint fail transport -> must raise, not return [False, ...]
+            gi._gqlSend = lambda endpoint, query, variables=None: (None, 0)
+            self.assertRaises(gi.InconclusiveError, truthBatch, ["1=1", "1=2"])
+        finally:
+            gi._gqlSend = saved
+
 
 class TestGraphqlDumpTable(unittest.TestCase):
     """Whole-table dump: column list + COUNT(*) + one row-scalar per ordinal offset"""
@@ -591,7 +712,7 @@ class TestGraphqlDumpTable(unittest.TestCase):
             "(SELECT COUNT(*) %s)" % colFrom: "2",
             "(SELECT %s %s %s)" % (d.columnCol, colFrom, d.paginate(d.columnCol, 0)): "id",
             "(SELECT %s %s %s)" % (d.columnCol, colFrom, d.paginate(d.columnCol, 1)): "name",
-            "(SELECT COUNT(*) FROM users)": "2",
+            "(SELECT COUNT(*) FROM %s)" % d.fromIdent("users"): "2",
             d.row(["id", "name"], "users", 0): "1~~~null",
             d.row(["id", "name"], "users", 1): "2~~~luther",
         }
@@ -669,6 +790,70 @@ class TestVulnserverGraphqlParser(unittest.TestCase):
         self.assertEqual(sels[0][1], "login")
 
 
+class TestGraphqlMutationPlanner(unittest.TestCase):
+    """Mutation slots are auto-tested (read-like ranked first), impact-classified, dry-run preferred."""
+
+    def test_impact_classification(self):
+        self.assertEqual(gi._mutationImpact("login"), "read-like")
+        self.assertEqual(gi._mutationImpact("verifyToken"), "read-like")
+        self.assertEqual(gi._mutationImpact("createUser"), "write-like")
+        self.assertEqual(gi._mutationImpact("deletePost"), "write-like")
+        self.assertEqual(gi._mutationImpact("frobnicate"), "unknown")
+
+    def test_mixed_names_are_write_like_not_read_like(self):
+        # a read-like substring must NOT mask a write token in the same (camelCase/snake) name
+        for name in ("updateUserPreview", "previewDeleteUser", "getAndDeleteUser", "createSession",
+                     "validateAndRemoveUser", "preview_delete_user", "get-and-delete-user"):
+            self.assertEqual(gi._mutationImpact(name), "write-like", name)
+        # genuine read-like names stay read-like
+        for name in ("login", "verifyToken", "previewReport", "checkSession", "fetchToken"):
+            self.assertEqual(gi._mutationImpact(name), "read-like", name)
+
+    def test_ranking_puts_read_like_first_write_like_last(self):
+        slots = [_slot("mutation", "Mutation", "deleteUser", "id"),
+                 _slot("mutation", "Mutation", "frobnicate", "x"),
+                 _slot("mutation", "Mutation", "login", "username")]
+        ranked = [s.fieldName for s in gi._rankMutations(slots)]
+        self.assertEqual(ranked[0], "login")            # read-like first
+        self.assertEqual(ranked[-1], "deleteUser")      # write-like last
+
+    def test_write_like_mutation_not_auto_enumerated(self):
+        # a write-like mutation is NOT eligible as the bulk-enumeration oracle (non-persistence
+        # unverified), whereas a read-like one is. This is the gate graphqlScan applies.
+        createSlot = _slot("mutation", "Mutation", "createUser", "name")
+        loginSlot = _slot("mutation", "Mutation", "login", "username")
+        self.assertFalse(gi._mutationImpact("createUser") == "read-like" or gi._dryRunVerified(createSlot, "http://x"))
+        self.assertTrue(gi._mutationImpact("login") == "read-like" or gi._dryRunVerified(loginSlot, "http://x"))
+        self.assertFalse(gi._dryRunVerified(createSlot, "http://x"))   # no automatic non-persistence proof
+
+    def test_mutation_oracle_is_never_batched(self):
+        # a mutation must NOT return truthBatch: aliased batching executes the write resolver once per
+        # alias (many writes per request). A boolean-diverging mutation slot yields (truth, None).
+        MATCHV = '{"data":{"createUser":{"id":1,"name":"luther"}}}'
+        NOMATCHV = '{"data":{"createUser":null}}'
+        saved = gi._gqlSend
+        try:
+            gi._gqlSend = lambda endpoint, query, variables=None: (MATCHV if "1=1" in query else NOMATCHV, 200)
+            slot = _slot("mutation", "Mutation", "createUser", "name", "string")
+            truth, truthBatch = gi._makeOracle(slot, "http://test/graphql")
+            self.assertIsNotNone(truth)
+            self.assertIsNone(truthBatch)               # batching disabled for mutations
+        finally:
+            gi._gqlSend = saved
+
+    def test_dryrun_flag_forced_true_even_when_optional(self):
+        # an optional Boolean 'dryRun' sibling is normally omitted; for a mutation probe it is forced
+        # true so the write does not commit
+        allArgs = [
+            ("name", {"kind": "NON_NULL", "name": None, "ofType": {"kind": "SCALAR", "name": "String", "ofType": None}}, None),
+            ("dryRun", {"kind": "SCALAR", "name": "Boolean"}, None),
+        ]
+        slot = gi.Slot("mutation", "Mutation", "createUser", allArgs, "name", "string",
+                        "OBJECT", "User", "{ id }")
+        q = gi._buildQuery(slot, "x")
+        self.assertIn("dryRun:true", q)
+
+
 class TestGraphqlSiblingDefaults(unittest.TestCase):
     """Required sibling arguments must use their real type, not be hardcoded as strings"""
 
@@ -684,8 +869,9 @@ class TestGraphqlSiblingDefaults(unittest.TestCase):
         self.assertIn("limit:0", q)
         self.assertNotIn('limit:"0"', q)
 
-    def test_boolean_sibling_gets_default_string(self):
-        """field(name: String!, active: Boolean!) -- Boolean gets \"x\" since there is no Boolean strategy"""
+    def test_boolean_sibling_uses_native_syntax(self):
+        """field(name: String!, active: Boolean!) -- a required Boolean renders as the native `false`
+        literal, NOT the quoted string "x" (which would make the whole query fail to parse)"""
         allArgs = [
             ("name", {"kind": "NON_NULL", "name": None, "ofType": {"kind": "SCALAR", "name": "String", "ofType": None}}, None),
             ("active", {"kind": "NON_NULL", "name": None, "ofType": {"kind": "SCALAR", "name": "Boolean", "ofType": None}}, None),
@@ -693,7 +879,20 @@ class TestGraphqlSiblingDefaults(unittest.TestCase):
         slot = gi.Slot("query", "Query", "toggle", allArgs, "name", "string",
                         "OBJECT", "User", "{ id }")
         q = gi._buildQuery(slot, "test")
-        self.assertIn('active:"x"', q)
+        self.assertIn('active:false', q)
+        self.assertNotIn('active:"x"', q)
+
+    def test_optional_sibling_is_omitted(self):
+        """field(name: String!, verbose: Boolean) -- an OPTIONAL sibling with no default is omitted,
+        not filled with a bogus sentinel that would invalidate the query"""
+        allArgs = [
+            ("name", {"kind": "NON_NULL", "name": None, "ofType": {"kind": "SCALAR", "name": "String", "ofType": None}}, None),
+            ("verbose", {"kind": "SCALAR", "name": "Boolean"}, None),
+        ]
+        slot = gi.Slot("query", "Query", "toggle", allArgs, "name", "string",
+                        "OBJECT", "User", "{ id }")
+        q = gi._buildQuery(slot, "test")
+        self.assertNotIn("verbose", q)
 
 
 class TestGraphqlScalarReturnSelection(unittest.TestCase):

tests/test_hql.py

@@ -88,6 +88,39 @@ class TestDetection(unittest.TestCase):
         template, _, _ = hql._detectBoolean("GET", "name")
         self.assertIsNone(template)
 
+    def test_confirm_hql_battery_on_orm(self):
+        # a Hibernate back-end evaluates str(): str(1)='1' true, str(1)='2' false -> diverges -> HQL
+        # confirmed with no error leakage
+        def mock(place, parameter, value):
+            return "<html><div>row</div></html>" if "str(1)='1'" in value else "<html></html>"
+        hql._send = mock
+        boundary = hql.Boundary("' OR ", " OR '1'='2", True)
+        self.assertTrue(hql._confirmHql("GET", "name", boundary, "x"))
+
+    def test_confirm_hql_battery_rejects_plain_sql(self):
+        # a raw-SQL back-end has no str() function -> the payload ERRORS on both sides -> no divergence
+        def mock(place, parameter, value):
+            if "str(" in value:
+                return "You have an error in your SQL syntax; no such function: str"
+            return "<html><div>row</div></html>"
+        hql._send = mock
+        boundary = hql.Boundary("' OR ", " OR '1'='2", True)
+        self.assertFalse(hql._confirmHql("GET", "name", boundary, "x"))
+
+    def test_confirm_hql_battery_rejects_sqlite_flexible_cast(self):
+        # SQLite accepts arbitrary CAST type names, so CAST(1 AS string)='1' is TRUE on plain SQLite;
+        # the battery must NOT use cast aliases and must NOT confirm HQL here (str() has no SQLite fn ->
+        # errors -> no divergence). This is the exact P0-3 false-positive being guarded against.
+        def mock(place, parameter, value):
+            if "str(" in value:                                   # SQLite: no such function -> error
+                return "SQLite error: no such function: str"
+            if "CAST(1 AS string)='1'" in value:                  # SQLite WOULD accept this as true...
+                return "<html><div>row</div></html>"
+            return "<html></html>"
+        hql._send = mock
+        boundary = hql.Boundary("' OR ", " OR '1'='2", True)
+        self.assertFalse(hql._confirmHql("GET", "name", boundary, "x"))   # ...but the battery no longer uses casts
+
 
 def _recordOracle(record, entity="Member"):
     """Build a truth(predicate) that answers the LENGTH/SUBSTRING/EXISTS predicates
@@ -150,6 +183,16 @@ class TestExtraction(unittest.TestCase):
     def test_infer_absent_attribute_empty(self):
         self.assertEqual(hql._inferValue(self.truth, "Member", "nope", "id"), "")
 
+    def test_infer_inconclusive_aborts_value(self):
+        """A truth() that stays INCONCLUSIVE must abort the value (return None) rather than emit a
+        length/char chosen from an ambiguous bit."""
+        from lib.utils.nonsql import InconclusiveError
+
+        def inconclusiveTruth(predicate):
+            raise InconclusiveError()
+
+        self.assertIsNone(hql._inferValue(inconclusiveTruth, "Member", "name", "id"))
+
 
 def _multiOracle(records):
     """Row-aware oracle: honors the "_h2.<pin> > <after>" walk bound by selecting the

tests/test_ldap.py

@@ -281,6 +281,35 @@ class TestBooleanDetection(unittest.TestCase):
         self.assertEqual(breakout, "*)")
         self.assertIn("*)(objectClass=*", bypass)
 
+    def test_ldap_breakout_uses_matched_false_filter(self):
+        # the false control must share the true control's breakout+attribute+open-fragment shape,
+        # differing ONLY in the assertion value: (attr=*) vs (attr=<sentinel>). It must NEVER be a
+        # bare original+SENTINEL string (an unmatched control a validation layer could diverge on).
+        sent = []
+
+        def spy(place, param, value):
+            sent.append(value)
+            return '{"count":15}' if value.startswith("x*)(objectClass=*") else '{"count":0}'
+
+        ldap._send = spy
+        from lib.core.enums import PLACE
+        template, _, _ = ldap._detectBoolean(PLACE.GET, 'q')
+        self.assertIsNotNone(template)
+        self.assertTrue(any(v.endswith("=%s" % SENTINEL) and "(" in v for v in sent),
+                        "no syntax-matched false LDAP filter control was sent: %r" % sent[:8])
+        self.assertNotIn("x%s" % SENTINEL, sent)   # the discredited bare original+SENTINEL is gone
+
+    def test_ldap_403_is_inconclusive(self):
+        # a 403 (WAF / rate-limit) must NOT enter the oracle as a page - _send returns None
+        from lib.request.connect import Connect
+        from lib.core.enums import PLACE
+        orig = Connect.getPage
+        Connect.getPage = staticmethod(lambda **kw: ("blocked by WAF", {}, 403))
+        try:
+            self.assertIsNone(ldap._send(PLACE.GET, 'q', 'x'))
+        finally:
+            Connect.getPage = orig
+
 
 class TestExtraction(unittest.TestCase):
     def test_inferAttribute_simple(self):
@@ -311,6 +340,47 @@ class TestExtraction(unittest.TestCase):
         value = ldap._inferAttribute(oracle, builder, "mail")
         self.assertEqual(value, "admin@example.com")
 
+    def test_inferAttribute_inconclusive_aborts_not_truncates(self):
+        """An oracle that stays INCONCLUSIVE must abort the attribute (return None) rather than
+        truncate it to whatever prefix was recovered before the ambiguous bit."""
+        from lib.utils.nonsql import InconclusiveError
+
+        class InconclusiveOracle(object):
+            def extract(self, payload):
+                raise InconclusiveError()
+
+        builder = ldap._ProbeBuilder(")")
+        self.assertIsNone(ldap._inferAttribute(InconclusiveOracle(), builder, "uid"))
+
+
+class TestMultiValueDump(unittest.TestCase):
+    """Multi-valued LDAP attributes must NOT be 'enumerated' via entry-scoped negation (which excludes
+    the whole entry and mixes entries) - recover ONE matching value and label it honestly."""
+
+    def setUp(self):
+        self._exists, self._infer, self._dumpTable = ldap._exists, ldap._inferAttribute, ldap._dumpTable
+
+    def tearDown(self):
+        ldap._exists, ldap._inferAttribute, ldap._dumpTable = self._exists, self._infer, self._dumpTable
+
+    def test_reports_one_value_and_never_excludes(self):
+        captured = {}
+        exclusionsSeen = []
+
+        ldap._exists = lambda oracle, builder, attr, **kw: attr == "member"
+        def fakeInfer(oracle, builder, attr, constraint=None, exclusions=None, **kw):
+            exclusionsSeen.append(exclusions)
+            return "cn=alice,dc=x" if attr == "member" else None
+        ldap._inferAttribute = fakeInfer
+        ldap._dumpTable = lambda title, cols, rows: captured.update(title=title, cols=cols, rows=rows)
+
+        dumped = ldap._dumpMultiValues(object(), ldap._ProbeBuilder(")"), "GET", "q")
+        self.assertTrue(dumped)
+        self.assertEqual(captured["rows"], [("cn=alice,dc=x",)])          # exactly one value
+        self.assertIn("one matching value", captured["title"].lower())    # honest label
+        # the broken exclusion walk must be gone: _inferAttribute is called WITHOUT exclusions
+        self.assertTrue(all(e in (None, [], ()) for e in exclusionsSeen))
+
 
 class TestIsError(unittest.TestCase):
     def test_isError_positive(self):

tests/test_nosql.py

@@ -11,6 +11,7 @@ Neo4j Cypher and ArangoDB AQL string break-out.
 """
 
 import re
+import time
 import unittest
 
 from _testutils import bootstrap
@@ -54,6 +55,10 @@ def _mongo(place, parameter, op, value, isArray=False):
 def _es(place, parameter, value):
     if value == "*":
         return MATCH
+    if "AND NOT" in value:                                      # Lucene (rand AND NOT rand) -> nothing
+        return NOMATCH
+    if value.startswith("(NOT ") and value.endswith(")"):       # Lucene (NOT rand) -> everything
+        return MATCH
     if value == ni.NOSQL_SENTINEL:
         return NOMATCH
     if value.startswith("/") and value.endswith("/"):           # Lucene regexp is full-anchored
@@ -87,6 +92,25 @@ class TestNoSqlMongo(unittest.TestCase):
         ni._fetch = lambda *args, **kwargs: MATCH
         self.assertIsNone(ni._detectMongo("GET", "password"))
 
+    def test_resolve_vector_carries_false_model(self):
+        # the LIVE vector must carry a calibrated false model so extraction is dual-model, not one-sided
+        vector = ni._resolve("GET", "password", "password")
+        self.assertIsNotNone(vector)
+        self.assertEqual(vector.falseModel, NOMATCH)          # $in[sentinel] no-match page
+
+    def test_dual_model_extraction_and_unrelated_page_inconclusive(self):
+        template = MATCH
+        falseModel = NOMATCH
+        value = ni._extract(template,
+                            lambda v: ni._fetch("GET", "password", "$regex", v),
+                            lambda n: "^.{%d,}$" % n,
+                            lambda known, klass: "^" + re.escape(known) + klass,
+                            falseModel=falseModel)
+        self.assertEqual(value, SECRET)
+        # an unrelated usable page (neither true nor false model) must be inconclusive, not a false bit
+        self.assertRaises(ni.InconclusiveError,
+                          ni._contentBit, lambda v: "CCCCC unrelated captcha page", "A", MATCH, NOMATCH)
+
 
 class TestNoSqlElasticsearch(unittest.TestCase):
     def setUp(self):
@@ -113,16 +137,19 @@ class TestNoSqlElasticsearch(unittest.TestCase):
 
 
 def _cypher(place, parameter, value):
-    if "'1'='1" in value:
-        return MATCH
-    if "'1'='2" in value:
-        return NOMATCH
+    m = re.search(r"STARTS WITH '([^']*)'", value)              # Cypher-only prefix predicate on 'ab'
+    if m:
+        return MATCH if "ab".startswith(m.group(1)) else NOMATCH
     m = re.search(r"=~ '\^(.*)$", value)                        # the regex body after the =~ operator
     if m:
         try:
             return MATCH if re.match("^(?:%s)$" % m.group(1), SECRET) is not None else NOMATCH
         except re.error:
             return NOMATCH
+    if "'1'='1" in value:
+        return MATCH
+    if "'1'='2" in value:
+        return NOMATCH
     return NOMATCH
 
 
@@ -239,6 +266,16 @@ class TestNoSqlWhere(unittest.TestCase):
         charValue = lambda known, klass: ni._whereDelay("d.%s&&/^%s%s/.test(d.%s)" % (key, ni._javaEscape(known), klass, key))
         self.assertEqual(ni._extract(None, None, lengthValue, charValue, _whereTruth), SECRET)
 
+    def test_where_delay_is_dos_bounded(self):
+        # the per-document busy-loop must be capped to ONE document per query (shared-scope counter),
+        # so a loose/unconditional condition on a large collection cannot block for docCount*timeSec.
+        # The condition and delay budget must still be embedded verbatim (oracle unchanged).
+        payload = ni._whereDelay("true")
+        self.assertIn("__c", payload)                       # shared-scope counter present
+        self.assertIn("__c<1", payload)                     # loop gated on the one-shot cap
+        self.assertIn("(true)", payload)                    # original condition preserved
+        self.assertIn(str(int(ni.conf.timeSec * 1000)), payload)   # delay budget preserved
+
 
 def _jswhere(place, parameter, value):
     # emulate a content-bearing MongoDB $where (server-side JavaScript) endpoint
@@ -295,7 +332,7 @@ class TestNoSqlWhereDump(unittest.TestCase):
         names = [name for name, _ in self.DOC]
         values = dict(self.DOC)
 
-        def fake(place, parameter, bound, expr, threshold):
+        def fake(place, parameter, bound, expr, threshold, strict=False):
             m = re.search(r"Object\.keys\(d\)\[(\d+)\]", expr)
             if m:
                 index = int(m.group(1))
@@ -311,7 +348,7 @@ class TestNoSqlWhereDump(unittest.TestCase):
         ni._whereField = self._orig
 
     def test_dump(self):
-        columns, rows = ni._whereDump("GET", "password", "", 0)
+        columns, rows, bound, complete = ni._whereDump("GET", "password", "", 0)
         self.assertEqual(columns, ["id", "username", "password", "role"])
         self.assertEqual(rows, [["1", "luther", "s3cr3t", "admin"]])
 
@@ -320,6 +357,88 @@ class TestNoSqlWhereDump(unittest.TestCase):
         self.assertIsNone(ni._whereDump("GET", "password", "", 0))
 
 
+class TestNoSqlRecordBinding(unittest.TestCase):
+    """A whole-document dump must be flagged bound only when a unique-record constraint pins it; an
+    unbound dump (no distinguishing sibling) is representative, not one coherent document."""
+
+    def test_vector_bound_defaults_true(self):
+        self.assertTrue(ni.Vector("X", None, None, None).bound)
+        self.assertFalse(ni.Vector("X", None, None, None, bound=False).bound)
+
+    def test_where_vector_unbound_without_sibling(self):
+        # single injected param, no sibling -> _constraint is "" -> $where dump is representative
+        ni.conf.parameters = {ni.PLACE.GET: "name=luther"}
+        ni.conf.paramDict = {ni.PLACE.GET: {"name": "luther"}}
+        self.assertEqual(ni._constraint(ni.PLACE.GET, "name", "==", "&&", prefix="d."), "")
+
+    def test_where_vector_bound_with_sibling(self):
+        # a distinguishing sibling pins the record -> bound constraint is non-empty
+        ni.conf.parameters = {ni.PLACE.GET: "id=7&name=luther"}
+        ni.conf.paramDict = {ni.PLACE.GET: {"name": "luther"}}
+        bound = ni._constraint(ni.PLACE.GET, "name", "==", "&&", prefix="d.")
+        self.assertIn("d.id=='7'", bound)
+        self.assertTrue(bool(bound))
+
+
+class TestNoSqlTriStateOracle(unittest.TestCase):
+    """A failed/blocked NoSQL response is UNKNOWN, retried, then aborts - never a silent false bit."""
+
+    def test_content_bit_retries_transient_then_recovers(self):
+        state = {"n": 0}
+        def fetch(value):
+            state["n"] += 1
+            return None if state["n"] == 1 else "TEMPLATE"      # first send fails, retry recovers
+        self.assertTrue(ni._contentBit(fetch, "A", "TEMPLATE"))
+
+    def test_content_bit_persistent_failure_raises(self):
+        self.assertRaises(ni.InconclusiveError, ni._contentBit, lambda v: None, "A", "TEMPLATE")
+
+    def test_content_bit_error_page_is_not_true(self):
+        # an error page is unusable -> retried -> InconclusiveError, never classified true/false
+        self.assertRaises(ni.InconclusiveError, ni._contentBit,
+                          lambda v: "MongoServerError: unknown operator: $foo", "A", "TEMPLATE")
+
+    def test_extract_aborts_value_on_inconclusive(self):
+        # a persistently failing oracle aborts the value (None), never fabricates a length/char
+        self.assertIsNone(ni._extract("TMPL", lambda v: None,
+                                      lambda n: "len>=%d" % n, lambda k, c: "char", truthFn=None))
+
+    def test_timed_bit_rejects_blocked_slow_response(self):
+        # a slow response that is BLOCKED (WAF/5xx) must not count as a true timing bit
+        self._fv = ni._fetchValue
+        try:
+            ni._lastCode = 429
+            ni._fetchValue = lambda *a, **k: (time.sleep(0.01) or "")     # slow but blocked (_isError via 429)
+            self.assertRaises(ni.InconclusiveError, ni._timedBit, "GET", "q", "payload", 0.0)
+        finally:
+            ni._fetchValue = self._fv
+            ni._lastCode = None
+
+    def test_content_bit_unrelated_page_is_inconclusive_not_false(self):
+        # P0-2: a usable page matching NEITHER the true nor the false model is UNKNOWN, not false -
+        # with both models supplied it must abort (InconclusiveError), never silently return False
+        self.assertRaises(ni.InconclusiveError,
+                          ni._contentBit, lambda v: "CCCCC unrelated soft-WAF page", "A", "AAAAA", "BBBBB")
+
+    def test_content_bit_both_models_classify_true_and_false(self):
+        self.assertTrue(ni._contentBit(lambda v: "AAAAA", "A", "AAAAA", "BBBBB"))
+        self.assertFalse(ni._contentBit(lambda v: "BBBBB", "A", "AAAAA", "BBBBB"))
+
+    def test_detect_where_rejects_blocked_slow_responses(self):
+        # P0-2: delayed BLOCKED responses (zero usable) must NOT establish a $where timing threshold
+        self._fv = ni._fetchValue
+        try:
+            ni.conf.timeSec = 5
+            ni.conf.parameters = {ni.PLACE.GET: "q=1"}
+            ni.conf.paramDict = {ni.PLACE.GET: {"q": "1"}}
+            ni._lastCode = 503
+            ni._fetchValue = lambda *a, **k: (time.sleep(0.02) or None)   # slow AND blocked/failed
+            self.assertIsNone(ni._detectWhere("GET", "q"))
+        finally:
+            ni._fetchValue = self._fv
+            ni._lastCode = None
+
+
 class TestNoSqlEnumDump(unittest.TestCase):
     """Content-based whole-document dump (e.g. Neo4j keys(u)): enumerate field names then values"""
 
@@ -327,11 +446,13 @@ class TestNoSqlEnumDump(unittest.TestCase):
 
     def setUp(self):
         self._ef, self._fv = ni._enumField, ni._fetchValue
-        ni._fetchValue = lambda *args, **kwargs: "<b>Welcome</b>"        # non-error single-record template
+        # true (any-match '.*') vs false (never-match sentinel) template must be SEPARABLE so _enumDump
+        # can calibrate both models; a constant page would (correctly) disable the dump
+        ni._fetchValue = lambda place, parameter, value: (NOMATCH if ni.NOSQL_SENTINEL in value else MATCH)
         names = [name for name, _ in self.DOC]
         values = dict(self.DOC)
 
-        def fake(place, parameter, template, payloadFor):
+        def fake(place, parameter, template, payloadFor, strict=False, falseModel=None):
             probe = payloadFor("X")                                     # render to inspect the target expression
             m = re.search(r"\(u\)\[(\d+)\]", probe)                     # keys/ATTRIBUTES/OBJECT_NAMES(u)[i]
             if m:
@@ -349,9 +470,11 @@ class TestNoSqlEnumDump(unittest.TestCase):
 
     def _check(self, keysExpr, valueExpr):
         makePayload = lambda expr, rb: "X' OR %s =~ '^%s.*" % (expr, rb)
-        columns, rows = ni._enumDump("GET", "password", makePayload, keysExpr, valueExpr)
+        columns, rows, bound, complete = ni._enumDump("GET", "password", makePayload, keysExpr, valueExpr)
         self.assertEqual(columns, ["id", "username", "password", "role"])
         self.assertEqual(rows, [["1", "luther", "s3cr3t", "admin"]])
+        # a constraint-only enum dump is NOT proven single-record -> the dump reports itself unbound
+        self.assertFalse(bound)
 
     def test_cypher(self):
         self._check(lambda i: "keys(u)[%d]" % i, lambda n: "toString(u[%s])" % ni._propLiteral(n))
@@ -428,7 +551,11 @@ class TestNoSqlRecords(unittest.TestCase):
 
 
 def _numeric(place, parameter, value):
-    # numeric-context oracle: 'OR 1=1' is always-true (rows), 'AND 1=2' is false (no rows)
+    # numeric-context Neo4j: 'OR 1=1' is always-true (rows), 'AND 1=2' is false, PLUS the Cypher-only
+    # STARTS WITH prefix predicate the detector now requires to attribute Neo4j (vs plain SQL)
+    m = re.search(r"STARTS WITH '([^']*)'", value)
+    if m:
+        return MATCH if "ab".startswith(m.group(1)) else NOMATCH
     if "OR 1=1" in value:
         return MATCH
     if "AND 1=2" in value:
@@ -492,7 +619,11 @@ class TestNoSqlNumericN1QL(unittest.TestCase):
 
 
 def _numericAql(place, parameter, value):
-    # numeric-context ArangoDB: only the ||/&& family diverges (OR/AND and REGEXP_CONTAINS do not)
+    # numeric-context ArangoDB: the ||/&& family diverges, PLUS the AQL-only two-arg LIKE(text, search)
+    # function the detector now requires to attribute ArangoDB (SQL's LIKE is an operator, not a function)
+    m = re.search(r"LIKE\('ab', '([^%]*)%'\)", value)
+    if m:
+        return MATCH if "ab".startswith(m.group(1)) else NOMATCH
     return MATCH if "|| 1==1" in value else NOMATCH
 
 
@@ -545,7 +676,7 @@ class TestNoSqlPartiQL(unittest.TestCase):
         self.assertEqual(value, SECRET)
 
     def test_dump_binds_sibling(self):
-        columns, rows = ni._partiqlDump("GET", "password", "password")
+        columns, rows, bound, complete = ni._partiqlDump("GET", "password", "password")
         self.assertEqual(columns, ["password"])
         self.assertEqual(rows, [[SECRET]])
 
@@ -613,6 +744,118 @@ class TestNoSqlCookiePlace(unittest.TestCase):
         self.assertIn("u.session='abc'", constraint)
         self.assertIn("u.username='luther'", constraint)
 
+    def test_constraint_escapes_literal_and_skips_non_identifiers(self):
+        # a quote/backslash in a sibling value must be escaped (not break out of the string literal),
+        # and a non-identifier field name must be skipped rather than alter the predicate structure
+        ni.conf.parameters = {ni.PLACE.GET: "q=x&name=o'brien&weird.field=v&password=p"}
+        ni.conf.paramDict = {ni.PLACE.GET: {"q": "x"}}
+        constraint = ni._constraint(ni.PLACE.GET, "q")
+        self.assertIn("u.name='o\\'brien'", constraint)          # single quote escaped
+        self.assertNotIn("weird.field", constraint)              # dotted (non-identifier) name skipped
+        self.assertIn("u.password='p'", constraint)
+
+
+class TestNoSqlJsonRawReplace(unittest.TestCase):
+    """Parse-failure JSON fallback: mutate ONLY the target key's value span in a JSON-like body, never
+    reconstruct it with a form serializer (which would produce unrelated 'name=value&...' content)."""
+
+    def test_double_quoted_value_replaced_in_place(self):
+        body = '{"name": "luther", "role": "user"}'
+        out = ni._jsonRawReplace(body, "name", {"$ne": None})
+        self.assertEqual(out, '{"name": {"$ne": null}, "role": "user"}')
+        self.assertIn('"role": "user"', out)                    # sibling preserved verbatim
+
+    def test_json_like_single_quotes_not_form_serialized(self):
+        # single-quoted -> json.loads() fails in the real flow; the span replace still works and the
+        # body stays JSON-shaped (no '&', no 'name=value' reconstruction)
+        body = "{'name': 'luther', 'active': true}"
+        out = ni._jsonRawReplace(body, "name", "payload")
+        self.assertIsNotNone(out)
+        self.assertIn("'active': true", out)                    # sibling + JS literal preserved
+        self.assertNotIn("&", out)
+        self.assertTrue(out.strip().startswith("{"))            # still a JSON object, not form content
+
+    def test_bareword_and_numeric_values(self):
+        self.assertEqual(ni._jsonRawReplace('{"age": 42}', "age", 7), '{"age": 7}')
+        self.assertEqual(ni._jsonRawReplace('{"ok": true}', "ok", "x"), '{"ok": "x"}')
+
+    def test_missing_key_returns_none(self):
+        # key absent -> None so the caller SKIPS the probe rather than corrupt the body
+        self.assertIsNone(ni._jsonRawReplace('{"other": "v"}', "name", "x"))
+
+    def test_key_like_text_inside_string_is_not_matched(self):
+        # the reviewer's reproduction: 'name: old' inside the "note" STRING must NOT be mutated - only
+        # the real "name" property is replaced
+        body = '{"note":"name: old", "name":"real"}'
+        out = ni._jsonRawReplace(body, "name", {"$ne": None})
+        self.assertEqual(out, '{"note":"name: old", "name":{"$ne": null}}')
+        self.assertIn('"note":"name: old"', out)                 # decoy string untouched
+
+    def test_key_like_text_inside_single_quoted_string(self):
+        body = "{'note':'name: trap', 'name':'real'}"
+        out = ni._jsonRawReplace(body, "name", "P")
+        self.assertIn("'note':'name: trap'", out)                # decoy untouched
+        self.assertTrue(out.endswith('"P"}'))                    # real value replaced
+
+    def test_key_like_text_inside_comment_is_not_matched(self):
+        body = '{/* name: not here */ "name": "real"}'
+        out = ni._jsonRawReplace(body, "name", "P")
+        self.assertIn("/* name: not here */", out)               # comment untouched
+        self.assertIn('"name": "P"', out)
+
+    def test_object_and_array_values_replaced_whole(self):
+        # an object/array as the original value must be replaced in full, not partially
+        self.assertEqual(ni._jsonRawReplace('{"f": {"a": 1, "b": [2, 3]}, "g": 9}', "f", "X"),
+                         '{"f": "X", "g": 9}')
+        self.assertEqual(ni._jsonRawReplace('{"f": [1, {"x": "}"}, 2], "g": 9}', "f", 0),
+                         '{"f": 0, "g": 9}')
+
+    def test_nested_property_located_at_depth(self):
+        # a nested property (not top-level) is located and replaced without disturbing structure
+        out = ni._jsonRawReplace('{"outer": {"name": "luther"}}', "name", {"$ne": None})
+        self.assertEqual(out, '{"outer": {"name": {"$ne": null}}}')
+
+    def test_brace_inside_string_value_does_not_close_object(self):
+        # a '}' inside a string value must not end the value token early
+        out = ni._jsonRawReplace('{"a": "va}lue", "name": "x"}', "name", "P")
+        self.assertIn('"a": "va}lue"', out)
+        self.assertIn('"name": "P"', out)
+
+    def test_brace_inside_comment_in_value_does_not_close_object(self):
+        # reviewer P0-2 reproduction: a '}' inside a COMMENT within an object value closed it early
+        out = ni._jsonRawReplace('{"f": {/* } */ "a": 1}, "g": 9}', "f", "X")
+        self.assertEqual(out, '{"f": "X", "g": 9}')
+
+    def test_key_like_text_inside_regex_literal_is_not_matched(self):
+        # reviewer P0-2 reproduction: 'name:' inside a /regex/ literal must not be taken as the property
+        out = ni._jsonRawReplace('{pattern: /name: trap/, name: "real"}', "name", "P")
+        self.assertEqual(out, '{pattern: /name: trap/, name: "P"}')
+
+    def test_regex_with_slash_in_char_class(self):
+        # a regex value containing '/' inside a [..] class and a '}' must be skipped whole
+        out = ni._jsonRawReplace('{"re": /[a/}]x/, "name": "y"}', "name", "P")
+        self.assertIn("/[a/}]x/", out)
+        self.assertIn('"name": "P"', out)
+
+    def test_backtick_string_is_not_a_key(self):
+        # a backtick template value containing 'name:' must not be mistaken for the property
+        out = ni._jsonRawReplace('{"tpl": `name: ${x}`, "name": "z"}', "name", "P")
+        self.assertIn("`name: ${x}`", out)
+        self.assertIn('"name": "P"', out)
+
+    def test_regex_value_flags_consumed(self):
+        # P0-6: a regex value's span must include trailing flags, else a dangling 'i' is left behind
+        self.assertEqual(ni._jsonRawReplace('{re: /abc/i, name: "x"}', "re", "P"),
+                         '{re: "P", name: "x"}')
+
+    def test_duplicate_key_at_different_depths_is_skipped(self):
+        # P0-6: with only the leaf key name, an ambiguous body (same key at 2 places) must be SKIPPED,
+        # never guessed - the payload could otherwise reach the wrong field
+        self.assertIsNone(ni._jsonRawReplace('{"outer":{"name":"first"},"name":"second"}', "name", "P"))
+
+    def test_single_occurrence_still_mutates(self):
+        self.assertEqual(ni._jsonRawReplace('{"a":1,"name":"real"}', "name", "P"), '{"a":1,"name":"P"}')
+
 
 class TestNoSqlErrorRegex(unittest.TestCase):
     """The heuristic regex must match real back-end error structures, not bare product names (so an
@@ -660,6 +903,27 @@ class TestNoSqlErrorRegex(unittest.TestCase):
             self.assertIsNone(re.search(self.NOSQL_ERROR_REGEX, sample), "should NOT match: %s" % sample)
 
 
+class TestNoSqlNoneSafety(unittest.TestCase):
+    """A blocked/failed request makes _send() return None; the fingerprint/error helpers must not
+    crash calling .lower() on it."""
+
+    def setUp(self):
+        self._f, self._fv = ni._fetch, ni._fetchValue
+        ni._fetch = lambda *a, **k: None
+        ni._fetchValue = lambda *a, **k: None
+        ni.conf.parameters = {"GET": "q=x"}
+        ni.conf.paramDict = {"GET": {"q": "x"}}
+
+    def tearDown(self):
+        ni._fetch, ni._fetchValue = self._f, self._fv
+
+    def test_nosql_failed_fingerprint_does_not_crash(self):
+        # None responses must not raise (was: 'NoneType' has no attribute 'lower')
+        self.assertIn(ni._fingerprintMongo("GET", "q"), ("CouchDB", "MongoDB", "MongoDB/CouchDB-compatible operator back-end"))
+        self.assertIn(ni._fingerprintLucene("GET", "q"), ("Solr", "OpenSearch", "Lucene query_string-compatible back-end"))
+        self.assertIsNone(ni._detectError("GET", "q"))
+
+
 if __name__ == "__main__":
     unittest.main()
 

tests/test_ssti.py

@@ -145,6 +145,15 @@ class TestErrorDetection(unittest.TestCase):
         page = ssti._probeError("GET", "q", engine)
         self.assertIsNone(page)
 
+    def test_ssti_static_engine_error_is_not_confirmation(self):
+        # a static template-parser error (present for every value, no arithmetic/boolean evaluation
+        # proof) must NOT confirm SSTI - only reflect the payload and always leak an engine name
+        def mock(place, parameter, value):
+            return "debug: jinja2.exceptions.TemplateSyntaxError (cached). you sent: " + value
+        ssti._send = mock
+        engine, evidence = ssti._fingerprint("GET", "q")
+        self.assertIsNone(engine)   # no evaluation proof -> not a confirmed SSTI
+
     def test_backend_from_error(self):
         page = "jinja2.exceptions.UndefinedError: 'foo' is undefined"
         backend = ssti._backendFromError(page)
@@ -222,6 +231,36 @@ class TestBooleanDetection(unittest.TestCase):
         template = ssti._detectBoolean("GET", "q", engine)
         self.assertIsNone(template)
 
+    def test_true_marker_in_baseline_rejected(self):
+        """When the true marker ('True') is already present in the untouched baseline it is page
+        furniture, not our evaluated output, so its appearance cannot confirm a boolean oracle."""
+        engine = ssti._ENGINE_TABLE[0]  # Jinja2, trueRendered='True'
+
+        def mock(place, parameter, value):
+            if "{{ True }}" in value:
+                return "flag=True ok"
+            if "{{ False }}" in value:
+                return "flag=True no"       # diverges, but 'True' still shown
+            return "flag=True baseline"     # 'True' already in the baseline
+
+        ssti._send = mock
+        self.assertIsNone(ssti._detectBoolean("GET", "q", engine))
+
+    def test_error_pages_are_not_a_boolean_oracle(self):
+        """Two syntactically invalid true/false payloads that merely trip DIFFERENT engine error
+        messages diverge, but an error page is not a rendered boolean -> no oracle."""
+        engine = ssti._ENGINE_TABLE[0]  # Jinja2
+
+        def mock(place, parameter, value):
+            if "{{ True }}" in value:
+                return "jinja2.exceptions.UndefinedError: x"
+            if "{{ False }}" in value:
+                return "TemplateSyntaxError: y"
+            return "baseline"
+
+        ssti._send = mock
+        self.assertIsNone(ssti._detectBoolean("GET", "q", engine))
+
 
 class TestFingerprint(unittest.TestCase):
     def setUp(self):
@@ -393,6 +432,100 @@ class TestRequestMutation(unittest.TestCase):
         self.assertEqual(result, "a=1&b=xx")
 
 
+class TestRceProof(unittest.TestCase):
+    """Proof-of-execution via a DERIVED challenge: reflection (raw OR transformed) must NOT be accepted."""
+
+    def setUp(self):
+        self.original_send = ssti._send
+
+    def tearDown(self):
+        ssti._send = self.original_send
+
+    def test_derived_executed_needs_product_absent_from_request(self):
+        # the product proves execution: present in the page, absent from baseline
+        self.assertTrue(ssti._derivedExecuted("page 6772561 footer", "baseline", "6772561"))
+        self.assertIsNone(ssti._derivedExecuted("page without it", "baseline", "6772561"))
+        # product already in baseline -> not attributable
+        self.assertIsNone(ssti._derivedExecuted("x 6772561 x", "seen 6772561 here", "6772561"))
+
+    def test_probe_rce_rejects_raw_reflection(self):
+        # an app that echoes the whole request body verbatim: the product ($((A*B)) result) is NEVER in
+        # the request, so it cannot appear in a reflected response -> not RCE-capable
+        engine = ssti._ENGINE_TABLE[0]  # Jinja2 (no _FILE_RCE spec -> file path is a no-op)
+        ssti._send = lambda place, parameter, value: "Hello, %s!" % value   # pure reflection
+        self.assertFalse(ssti._probeRce("GET", "q", engine))
+
+    def test_probe_rce_rejects_url_encoded_reflection(self):
+        # KEY P0-4 case: the app reflects the URL-ENCODED payload. A marker-in-payload check would pass;
+        # the derived product is still absent from any reflected form -> correctly NOT RCE-capable
+        from thirdparty.six.moves.urllib.parse import quote
+        engine = ssti._ENGINE_TABLE[0]
+        ssti._send = lambda place, parameter, value: "reflected: %s" % quote(value, safe="")
+        self.assertFalse(ssti._probeRce("GET", "q", engine))
+
+    def test_probe_rce_all_collisions_do_not_confirm(self):
+        # every generated product collides with the baseline -> every challenge is skipped; the loop
+        # must NOT fall through to success with zero executed payloads (counts confirmations, not iters)
+        engine = ssti._ENGINE_TABLE[0]
+        import lib.techniques.ssti.inject as _m
+        orig = _m.randomInt
+        try:
+            _m.randomInt = lambda n: 2                       # product is always 4
+            ssti._send = lambda place, parameter, value: "result is 4 everywhere"   # baseline contains "4"
+            self.assertFalse(ssti._probeRce("GET", "q", engine))
+        finally:
+            _m.randomInt = orig
+
+    def test_probe_rce_confirms_real_execution(self):
+        # a backend that actually evaluates `echo $((A*B))` returns the PRODUCT as command output
+        engine = ssti._ENGINE_TABLE[0]
+        import re as _re
+
+        def mock(place, parameter, value):
+            m = _re.search(r"echo \$\(\((\d+)\*(\d+)\)\)", value)
+            if m:
+                return "<html>%d</html>" % (int(m.group(1)) * int(m.group(2)))   # shell-evaluated product
+            return "baseline"
+
+        ssti._send = mock
+        self.assertTrue(ssti._probeRce("GET", "q", engine))
+
+    def test_framed_output_markers_are_reflection_proof(self):
+        # markers are shell-concatenated fragments: the completed 'startABstartCD' never appears in the
+        # request, so only genuine execution places them in the page
+        start, end = "aaaaaabbbbbb", "ccccccdddddd"
+        executed = "junk %suid=0(root) gid=0(root)%s junk" % (start, end)
+        self.assertEqual(ssti._framedOutput(executed, start, end), "uid=0(root) gid=0(root)")
+        # a response that lacks the concatenated markers (e.g. reflected 'aaaaaa bbbbbb' separated) -> None
+        self.assertIsNone(ssti._framedOutput("printf %s%s aaaaaa bbbbbb ...", start, end))
+
+    def test_probe_rce_confirms_on_windows_backend(self):
+        # a Windows-hosted engine evaluates `cmd /c set /a A*B` (Unix `$((...))` does nothing) - the
+        # derived product still proves execution, so capability detection works on Windows too
+        engine = ssti._ENGINE_TABLE[0]
+        import re as _re
+
+        def mock(place, parameter, value):
+            m = _re.search(r"set /a (\d+)\*(\d+)", value)        # cmd.exe set /a arithmetic
+            if m and "$((" not in value:
+                return "<html>%d</html>" % (int(m.group(1)) * int(m.group(2)))
+            return "baseline"                                    # the Unix $(( )) family produces nothing here
+
+        ssti._send = mock
+        self.assertTrue(ssti._probeRce("GET", "q", engine))
+
+    def test_windows_framed_builder_shape(self):
+        # the Windows framed command concatenates the marker fragments at runtime via `echo|set /p=`
+        cmd = ssti._winFramed("whoami", "SA", "SB", "EA", "EB")
+        self.assertIn("cmd /c", cmd)
+        self.assertIn("set /p=SA", cmd)
+        self.assertIn("set /p=SB", cmd)
+        self.assertIn("whoami", cmd)
+        # the concatenated markers 'SASB'/'EAEB' are NOT present literally (only the separate fragments)
+        self.assertNotIn("SASB", cmd)
+        self.assertNotIn("EAEB", cmd)
+
+
 class TestExecuteCommand(unittest.TestCase):
     def setUp(self):
         self.original_send = ssti._send
@@ -426,9 +559,12 @@ class TestExecuteCommand(unittest.TestCase):
             "Should have tried the second payload after error skip")
 
     def test_all_error_pages_produce_warning(self):
-        """When all RCE payloads produce template errors, no success is reported.
-        _executeCommand sends baseline + one request per fallback payload."""
+        """When all RCE payloads produce template errors, no success is reported. _executeCommand sends
+        a baseline, then TWO passes over the payloads: a reflection-proof boundary-marker capture pass
+        a framed pass PER OS family (unix + windows), and an unframed baseline-diff fallback pass (the
+        file-based pass sends nothing without a _FILE_RCE spec, as for Jinja2)."""
         engine = ssti._ENGINE_TABLE[0]
+        self.assertNotIn(engine.name, ssti._FILE_RCE)   # guard the arithmetic below
         calls = []
 
         def mock(place, parameter, value):
@@ -437,9 +573,10 @@ class TestExecuteCommand(unittest.TestCase):
 
         ssti._send = mock
         ssti._executeCommand("GET", "q", engine, "test")
-        # 1 baseline + N payload attempts = N+1 calls
-        self.assertEqual(len(calls), len(engine.rcePayloads) + 1,
-            "Should have tried all payloads (baseline + one per fallback) before giving up")
+        # 1 baseline + (families framed + 1 unframed) passes, each over N payloads
+        passes = len(ssti._SHELL_FAMILIES) + 1
+        self.assertEqual(len(calls), 1 + passes * len(engine.rcePayloads),
+            "Should have tried the framed pass per OS family then the unframed pass before giving up")
 
 
 class TestCommandEscaping(unittest.TestCase):
@@ -642,27 +779,51 @@ class TestStruts2Header(unittest.TestCase):
     def test_struts2_wired_for_file_rce(self):
         self.assertIn("Struts2 (OGNL)", ssti._FILE_RCE)   # modern-JDK file-based fallback wired
 
-    def test_s2045_detection_marker_echo(self):
+    def test_s2045_detection_derived_product(self):
         import re
-        # a vulnerable Struts2 evaluates the OGNL and writes the printed marker into the response
+        # a vulnerable Struts2 EVALUATES the OGNL arithmetic and writes the PRODUCT (absent from the header)
         def mock(url, action):
-            m = re.search(r"#w\.print\('([a-z0-9]+)'\)", action)
-            return "<html> %s </html>" % m.group(1) if m else "<html/>"
+            m = re.search(r"#w\.print\((\d+)\*(\d+)\)", action)
+            return "<html> %d </html>" % (int(m.group(1)) * int(m.group(2))) if m else "<html/>"
         ssti._s2045Send = mock
-        self.assertIsNotNone(ssti._probeStruts2Header("http://target"))
+        self.assertTrue(ssti._probeStruts2Header("http://target"))
+
+    def test_s2045_detection_rejects_reflected_header(self):
+        # KEY P0-6 case: the server REFLECTS the Content-Type header verbatim. The literal operands are
+        # echoed but never their product, so no reflection can satisfy the derived challenge -> not vuln
+        ssti._s2045Send = lambda url, action: "You sent Content-Type: %s" % action
+        self.assertIsNone(ssti._probeStruts2Header("http://target"))
 
     def test_s2045_not_vulnerable(self):
         ssti._s2045Send = lambda url, action: "<html>ordinary Struts page, no eval</html>"
         self.assertIsNone(ssti._probeStruts2Header("http://target"))
 
-    def test_s2045_command_output_sliced_from_markers(self):
-        # the shell echoes start/end markers around stdout; the response also carries the action HTML
+    def test_s2045_all_collisions_do_not_confirm(self):
+        # every product collides with the baseline -> no challenge is ever evaluated -> not confirmed
+        import lib.techniques.ssti.inject as _m
+        orig = _m.randomInt
+        try:
+            _m.randomInt = lambda n: 2                       # product is always 4
+            ssti._s2045Send = lambda url, action: "page containing 4"
+            self.assertIsNone(ssti._probeStruts2Header("http://target"))
+        finally:
+            _m.randomInt = orig
+
+    def test_s2045_command_output_sliced_from_derived_markers(self):
+        import re
+        # the shell CONCATENATES the marker fragments (printf %s%s A B -> AB); the concatenation never
+        # appears in the header, so it can only come from execution
         def mock(url, action):
-            m = re.search(r"echo ([a-z0-9]+); .* 2>&1; echo ([a-z0-9]+)", action)
+            m = re.search(r"printf %s%s (\w+) (\w+); .* 2>&1; printf %s%s (\w+) (\w+)", action)
             if not m:
                 return "<html/>"
-            start, end = m.group(1), m.group(2)
+            start, end = m.group(1) + m.group(2), m.group(3) + m.group(4)
             return "<html>%s\nuid=0(root) gid=0(root)\n%s</html>" % (start, end)
-        import re
         ssti._s2045Send = mock
         self.assertEqual(ssti._executeStruts2Header("http://target", "id"), "uid=0(root) gid=0(root)")
+
+    def test_s2045_command_rejects_reflected_header(self):
+        # raw header reflection: the fragments appear separated ('printf %s%s A B'), never concatenated,
+        # 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"))

tests/test_techniques.py

@@ -839,11 +839,18 @@ class TestLdapPureHelpers(unittest.TestCase):
         # header + 2 rows + 4 separators (top, under-header, ... actually 3 borders + n rows)
         self.assertEqual(grid.count("+----+----+"), 3)
 
-    def test_charset_excludes_metachars(self):
+    def test_charset_includes_metachars_escaped(self):
+        # filter metacharacters ARE extractable - _ldapLiteral() escapes them, so a value containing
+        # '*'/'('/')'/'\\' is recovered in full rather than truncated at the first one
         for meta in ("*", "(", ")", "\\"):
-            self.assertNotIn(ord(meta), ldap._CHARSET)
+            self.assertIn(ord(meta), ldap._CHARSET)
         self.assertIn(ord("a"), ldap._CHARSET)
         self.assertIn(ord("0"), ldap._CHARSET)
+        # common characters are still tried before the (rare) metacharacters
+        self.assertLess(ldap._CHARSET.index(ord("a")), ldap._CHARSET.index(ord("*")))
+        # the escaping the extractor relies on
+        self.assertEqual(ldap._ldapLiteral("abc*def"), "abc\\2adef")
+        self.assertIn("\\28", ldap._ldapLiteral("x(y)"))
 
     def test_probe_builder_shapes(self):
         b = ldap._ProbeBuilder("*)")
@@ -865,7 +872,7 @@ class _LdapOracleCase(unittest.TestCase):
     match: a payload's trailing assertion '(attr=value*' matches when the directory holds
     `attr` whose value starts with `value`."""
 
-    DIRECTORY = {"uid": "admin", "mail": "bob", "cn": "Administrator"}
+    DIRECTORY = {"objectClass": "top", "uid": "admin", "mail": "bob", "cn": "Administrator"}
 
     def setUp(self):
         self._sparams = conf.get("parameters")
@@ -876,6 +883,9 @@ class _LdapOracleCase(unittest.TestCase):
         conf.parameters = {PLACE.GET: "user=admin"}
         conf.paramDict = {PLACE.GET: {"user": "admin"}}
         conf.cookieDel = None
+        # the boolean tests exercise the content-similarity path; null any explicit user oracle that
+        # an earlier test module may have left set (the engines now honor --string/--regexp globally)
+        conf.string = conf.notString = conf.regexp = conf.code = None
 
         directory = self.DIRECTORY
 
@@ -911,7 +921,9 @@ class TestLdapParamSegment(_LdapOracleCase):
 
 class TestLdapOracle(_LdapOracleCase):
     def _oracle(self):
-        return ldap._makeOracle(PLACE.GET, "user", "TRUE-CONTENT-stable-match-uid")
+        # _makeOracle now recalibrates its own true/false models on the winning breakout + SENTINEL
+        # base (matched (objectClass=*) vs (objectClass=<sentinel>)); pass the breakout, not a template
+        return ldap._makeOracle(PLACE.GET, "user", ")")
 
     def test_exists_true(self):
         oracle, builder = self._oracle(), ldap._ProbeBuilder(")")
@@ -935,9 +947,10 @@ class TestLdapOracle(_LdapOracleCase):
 
     def test_enumerate_entry_keys(self):
         oracle, builder = self._oracle(), ldap._ProbeBuilder(")")
-        keyAttr, values = ldap._enumerateEntryKeys(oracle, builder)
+        keyAttr, values, partial = ldap._enumerateEntryKeys(oracle, builder)
         self.assertEqual(keyAttr, "uid")
         self.assertEqual(values, ["admin"])
+        self.assertFalse(partial)                   # clean end, not an inconclusive abort
 
 
 class TestLdapBoolean(_LdapOracleCase):
@@ -1036,10 +1049,111 @@ class TestGraphqlPureHelpers(unittest.TestCase):
         # non-graphql passes through unchanged
         self.assertEqual(gql._slotValue("raw"), "raw")
 
-    def test_default_for_arg(self):
-        self.assertEqual(gql._defaultForArg({"kind": "SCALAR", "name": "Int"}, None), 0)
-        self.assertEqual(gql._defaultForArg({"kind": "SCALAR", "name": "String"}, None), "x")
-        self.assertEqual(gql._defaultForArg({"kind": "SCALAR", "name": "String"}, "given"), "given")
+    def _nn(self, inner):
+        return {"kind": "NON_NULL", "ofType": inner}
+
+    def test_render_sibling_omits_optionals(self):
+        # OPTIONAL argument (not NON_NULL) with no default -> OMITTED (None), never a bogus sentinel
+        # that would invalidate the query and cause a false negative
+        self.assertIsNone(gql._renderSibling("limit", {"kind": "SCALAR", "name": "Int"}, None))
+        self.assertIsNone(gql._renderSibling("active", {"kind": "SCALAR", "name": "Boolean"}, None))
+        self.assertIsNone(gql._renderSibling("tags", {"kind": "LIST"}, None))
+
+    def test_render_sibling_required_native_syntax(self):
+        # REQUIRED (NON_NULL) argument with no default -> synthesize NATIVE syntax per kind
+        self.assertEqual(gql._renderSibling("limit", self._nn({"kind": "SCALAR", "name": "Int"}), None), "limit:0")
+        self.assertEqual(gql._renderSibling("q", self._nn({"kind": "SCALAR", "name": "String"}), None), 'q:"x"')
+        self.assertEqual(gql._renderSibling("active", self._nn({"kind": "SCALAR", "name": "Boolean"}), None), "active:false")
+        self.assertEqual(gql._renderSibling("ids", self._nn({"kind": "LIST", "ofType": {"kind": "SCALAR", "name": "Int"}}), None), "ids:[]")
+        self.assertEqual(gql._renderSibling("cfg", self._nn({"kind": "INPUT_OBJECT", "name": "Cfg"}), None), "cfg:{}")
+
+    def test_required_nested_input_object_is_recursively_populated(self):
+        # SearchInput!{ filter: FilterInput!{ term: String! (req), note: String (opt) } }: a required
+        # nested input must populate its REQUIRED inner fields recursively, not emit a bare {} the
+        # server rejects; optional inner fields are omitted
+        gql._inputFields.clear()
+        gql._inputFields["SearchInput"] = [("filter", self._nn({"kind": "INPUT_OBJECT", "name": "FilterInput"}), None)]
+        gql._inputFields["FilterInput"] = [
+            ("term", self._nn({"kind": "SCALAR", "name": "String"}), None),
+            ("note", {"kind": "SCALAR", "name": "String"}, None),
+        ]
+        try:
+            out = gql._renderSibling("input", self._nn({"kind": "INPUT_OBJECT", "name": "SearchInput"}), None)
+            self.assertEqual(out, 'input:{filter:{term:"x"}}')       # required term populated, optional note omitted
+        finally:
+            gql._inputFields.clear()
+
+    def test_recursive_input_cycle_is_bounded(self):
+        # a self-referential required input must not recurse forever - it terminates at {}
+        gql._inputFields.clear()
+        gql._inputFields["Node"] = [("child", self._nn({"kind": "INPUT_OBJECT", "name": "Node"}), None)]
+        try:
+            out = gql._renderSibling("n", self._nn({"kind": "INPUT_OBJECT", "name": "Node"}), None)
+            self.assertTrue(out.startswith("n:{child:"))
+            self.assertIn("{}", out)                                 # cycle broken with a bare {}
+        finally:
+            gql._inputFields.clear()
+
+    def test_render_sibling_required_enum_uses_bare_identifier(self):
+        gql._enumValues.clear()
+        gql._enumValues["Role"] = ["ADMIN", "USER"]
+        try:
+            self.assertEqual(gql._renderSibling("role", self._nn({"kind": "ENUM", "name": "Role"}), None), "role:ADMIN")
+        finally:
+            gql._enumValues.clear()
+
+    def test_render_sibling_default_emitted_verbatim(self):
+        # a schema defaultValue is ALREADY a serialized GraphQL literal -> emit VERBATIM, never re-quote
+        self.assertEqual(gql._renderSibling("active", {"kind": "SCALAR", "name": "Boolean"}, "true"), "active:true")
+        self.assertEqual(gql._renderSibling("role", {"kind": "ENUM", "name": "Role"}, "ADMIN"), "role:ADMIN")
+        self.assertEqual(gql._renderSibling("ids", {"kind": "LIST"}, "[1, 2]"), "ids:[1, 2]")
+        self.assertEqual(gql._renderSibling("filter", {"kind": "INPUT_OBJECT"}, "{a: 1}"), "filter:{a: 1}")
+        self.assertEqual(gql._renderSibling("n", {"kind": "SCALAR", "name": "Int"}, "5"), "n:5")
+        self.assertEqual(gql._renderSibling("q", {"kind": "SCALAR", "name": "String"}, '"hello"'), 'q:"hello"')
+
+    def _nnInput(self, name):
+        return {"kind": "NON_NULL", "ofType": {"kind": "INPUT_OBJECT", "name": name}}
+
+    def test_deep_nested_input_slot_discovered_and_rendered(self):
+        # search(input: SearchInput!{ filter: FilterInput!{ credentials: Creds!{ username: String! } } })
+        # the injectable leaf is input.filter.credentials.username, THREE levels deep - it must be both
+        # DISCOVERED as a slot and RENDERED as the full nested literal
+        gql._inputFields.clear()
+        gql._inputFields["SearchInput"] = [("filter", self._nnInput("FilterInput"), None)]
+        gql._inputFields["FilterInput"] = [("credentials", self._nnInput("Creds"), None)]
+        gql._inputFields["Creds"] = [("username", self._nn({"kind": "SCALAR", "name": "String"}), None)]
+        try:
+            slots = []
+            gql._inputSlots("query", "Query", "search",
+                            [("input", self._nnInput("SearchInput"), None)],
+                            "input", self._nnInput("SearchInput"),
+                            "OBJECT", "User", "{ id }",
+                            {"SearchInput": {"kind": "INPUT_OBJECT", "name": "SearchInput", "inputFields": [{"name": "filter", "type": self._nnInput("FilterInput")}]},
+                             "FilterInput": {"kind": "INPUT_OBJECT", "name": "FilterInput", "inputFields": [{"name": "credentials", "type": self._nnInput("Creds")}]},
+                             "Creds": {"kind": "INPUT_OBJECT", "name": "Creds", "inputFields": [{"name": "username", "type": self._nn({"kind": "SCALAR", "name": "String"})}]}},
+                            slots)
+            paths = [s.targetArg for s in slots]
+            self.assertIn("input.filter.credentials.username", paths)
+
+            slot = [s for s in slots if s.targetArg == "input.filter.credentials.username"][0]
+            q = gql._buildQuery(slot, "PWN")
+            self.assertIn('input: {filter:{credentials:{username:"PWN"}}}', q)
+        finally:
+            gql._inputFields.clear()
+
+    def test_recursive_input_slot_cycle_bounded(self):
+        # a self-referential input object must not loop forever during slot discovery
+        gql._inputFields.clear()
+        gql._inputFields["Node"] = [("child", self._nnInput("Node"), None), ("val", self._nn({"kind": "SCALAR", "name": "String"}), None)]
+        try:
+            slots = []
+            tbn = {"Node": {"kind": "INPUT_OBJECT", "name": "Node", "inputFields": [
+                {"name": "child", "type": self._nnInput("Node")}, {"name": "val", "type": self._nn({"kind": "SCALAR", "name": "String"})}]}}
+            gql._inputSlots("mutation", "Mutation", "f", [("n", self._nnInput("Node"), None)],
+                            "n", self._nnInput("Node"), "OBJECT", "R", "{ id }", tbn, slots)
+            self.assertTrue(any(s.targetArg.endswith(".val") for s in slots))   # terminates + finds a leaf
+        finally:
+            gql._inputFields.clear()
 
 
 # A minimal but realistic introspection schema: query user(id: String, limit: Int): User
@@ -1121,7 +1235,7 @@ class TestGraphqlQueryBuilding(unittest.TestCase):
         q = gql._buildQuery(self.strSlot, "x' OR '1'='1")
         self.assertTrue(q.startswith("{user:user("))
         self.assertIn('id:"x\' OR \'1\'=\'1"', q)
-        self.assertIn("limit:0", q)             # required-ish sibling defaulted
+        self.assertNotIn("limit", q)            # optional sibling with no default is OMITTED (P0-2)
         self.assertIn("{ name uid }", q)
 
     def test_build_query_numeric_rejects_non_numeric(self):
@@ -1246,7 +1360,7 @@ class TestGraphqlDumpTable(unittest.TestCase):
             "(SELECT COUNT(*) %s)" % colFrom: "2",
             "(SELECT %s %s %s)" % (d.columnCol, colFrom, d.paginate(d.columnCol, 0)): "id",
             "(SELECT %s %s %s)" % (d.columnCol, colFrom, d.paginate(d.columnCol, 1)): "name",
-            "(SELECT COUNT(*) FROM users)": "2",
+            "(SELECT COUNT(*) FROM %s)" % d.fromIdent("users"): "2",
             d.row(["id", "name"], "users", 0): gql.COL_SEP.join(("1", "alice")),
             d.row(["id", "name"], "users", 1): gql.COL_SEP.join(("2", "bob")),
         }

tests/test_xpath.py

@@ -10,6 +10,7 @@ formatting can be exercised without a live target.
 """
 
 import os
+import re
 import sys
 import unittest
 
@@ -181,6 +182,11 @@ class TestBooleanDetection(unittest.TestCase):
 
     def test_detection_returns_extractable_boundary(self):
         def mock(place, parameter, value):
+            # faithful XPath engine: string-length('ab')=2 holds (XPath-only confirm the detector
+            # now requires), =3 does not; plus the true()/false() the break-out probes with
+            m = re.search(r"string-length\('ab'\)=(\d+)", value)
+            if m:
+                return '{"count":7,"entries":[{...}]}' if int(m.group(1)) == 2 else '{"count":0,"entries":[],"error":null}'
             if "true()" in value:
                 return '{"count":7,"entries":[{...}]}'
             elif "false()" in value:
@@ -218,6 +224,71 @@ class TestGridAndTable(unittest.TestCase):
         self.assertGreater(len(rows), 0)
 
 
+class TestExtractionCalibration(unittest.TestCase):
+    def test_xpath_or_boundary_calibrates_with_sentinel_base(self):
+        # for an OR-style boundary the extraction base is SENTINEL (not the original), and _makeOracle
+        # must calibrate its true()/false() models on THAT base so they match the extraction payloads
+        from lib.core.enums import PLACE
+        orBoundary = xpath.Boundary("' or ", " and '1'='1", True)
+        self.assertEqual(xpath._extractionBase("origvalue", orBoundary), xpath.SENTINEL)
+
+        sent = []
+
+        def spy(place, parameter, value):
+            sent.append(value)
+            return "TRUE-model-page" if "true()" in value else "FALSE-model-page"
+
+        old = xpath._send
+        xpath._send = spy
+        try:
+            xpath.conf.parameters = {PLACE.GET: "q=x"}
+            xpath.conf.paramDict = {PLACE.GET: {"q": "x"}}
+            oracle = xpath._makeOracle(PLACE.GET, "q", orBoundary, xpath._extractionBase("origvalue", orBoundary))
+        finally:
+            xpath._send = old
+        self.assertIsNotNone(oracle)
+        self.assertTrue(sent)
+        self.assertTrue(all(xpath.SENTINEL in p for p in sent), "calibration used a non-sentinel base: %r" % sent)
+        self.assertFalse(any("origvalue" in p for p in sent))
+
+    def test_transient_failure_on_true_bit_is_not_a_false_bit(self):
+        # A timeout/5xx on a TRUE predicate must NOT be cached as a false bit: resolveBit re-sends and
+        # recovers the correct TRUE, and a PERSISTENT failure aborts (InconclusiveError), never False.
+        from lib.core.enums import PLACE
+        from lib.utils.nonsql import InconclusiveError
+        orBoundary = xpath.Boundary("' or ", " and '1'='1", True)
+        base = xpath._extractionBase("x", orBoundary)
+
+        state = {"armed": False, "failed": set()}
+
+        def flaky(place, parameter, value):
+            page = "TRUE-model-page" if "true()" in value else "FALSE-model-page"
+            # once armed (post-build), the FIRST send of each probe fails transiently, then recovers
+            if state["armed"] and value not in state["failed"]:
+                state["failed"].add(value)
+                return None
+            return page
+
+        old = xpath._send
+        xpath._send = flaky
+        try:
+            xpath.conf.parameters = {PLACE.GET: "q=x"}
+            xpath.conf.paramDict = {PLACE.GET: {"q": "x"}}
+            oracle = xpath._makeOracle(PLACE.GET, "q", orBoundary, base)   # calibrates cleanly
+            self.assertIsNotNone(oracle)
+
+            # a fresh TRUE probe whose FIRST send fails transiently must resolve to True (retry), never False
+            probe = xpath._makePayload(base, orBoundary, "true()") + "[1]"
+            state["armed"] = True
+            self.assertTrue(oracle.extract(probe))
+
+            # a PERSISTENTLY failing probe must raise InconclusiveError, never return False
+            xpath._send = lambda place, parameter, value: None
+            self.assertRaises(InconclusiveError, oracle.extract, probe + "Z")
+        finally:
+            xpath._send = old
+
+
 class TestExtraction(unittest.TestCase):
     def test_infer_value_mock(self):
         expected = "directory"
@@ -289,6 +360,48 @@ class TestExtraction(unittest.TestCase):
                                    maxLen=32)
         self.assertEqual(fast, linear)
 
+    def test_inconclusive_oracle_aborts_value_not_fabricates(self):
+        # An oracle that stays INCONCLUSIVE (raises InconclusiveError, as resolveBit does after
+        # retries) must abort the value cleanly - _inferString returns None and _inferCount returns
+        # None (unknown) - rather than emitting a length/char/count chosen from an ambiguous bit.
+        from lib.utils.nonsql import InconclusiveError
+        boundary = xpath._BREAKOUT_BOUNDARY["') or true() or ('"]
+        builder = xpath._XPathPayloadBuilder("x", boundary)
+
+        class InconclusiveOracle(object):
+            def extract(self, payload):
+                raise InconclusiveError()
+
+        oracle = InconclusiveOracle()
+        self.assertIsNone(xpath._inferString(oracle, builder, "name(/*)", maxLen=32))
+        # inconclusive count must be None (unknown), NEVER 0 - 0 would read as a leaf and fabricate text
+        self.assertIsNone(xpath._inferCount(oracle, builder, "/*",
+                                            lambda b, p, c: b.childCount(p, c), maxCount=8))
+        self.assertIsNone(xpath._inferValue(oracle, builder, "/*",
+                                            lambda b, p, prefix: b.nameStartsWith(p, prefix), maxLen=32))
+
+    def test_walk_tree_marks_partial_and_does_not_fabricate_text_on_unknown_count(self):
+        # name resolves (real lxml eval), but every child/attribute COUNT probe is inconclusive: the
+        # node must be marked partial, must NOT be treated as a leaf (no fabricated scalar text), and
+        # must not iterate phantom children
+        from lib.utils.nonsql import InconclusiveError
+        boundary = xpath._BREAKOUT_BOUNDARY["') or true() or ('"]
+        builder = xpath._XPathPayloadBuilder("x", boundary)
+        template = _XPATH_TEMPLATES["function_arg"]
+
+        class PartialCountOracle(object):
+            def extract(self, payload):
+                if "count(" in payload:
+                    raise InconclusiveError()               # child/attribute counts are ambiguous
+                return _xpath_eval(template, payload) > 0    # names/strings resolve normally
+
+        node = xpath._walkTree(PartialCountOracle(), builder, "/*")
+        self.assertIsNotNone(node)
+        self.assertEqual(node["name"], "directory")
+        self.assertTrue(node["partial"])
+        self.assertIsNone(node["text"])       # unknown child count must NOT fabricate leaf text
+        self.assertEqual(node["children"], [])
+
 
 class TestBackendFingerprint(unittest.TestCase):
     def test_lxml(self):

tests/test_xxe.py

@@ -88,18 +88,62 @@ class TestBuildDoctype(unittest.TestCase):
         self.assertEqual(out.count("<!DOCTYPE"), 1)
         self.assertIn("[", out)
 
+    def test_splice_survives_quoted_bracket_close_in_subset(self):
+        # a ']>' sequence inside a quoted entity value must NOT be taken as the subset close - the
+        # splice still lands inside the real internal subset and produces a single valid DOCTYPE
+        xml = '<!DOCTYPE r [<!ENTITY a "x]>y">]><r>z</r>'
+        out = xxe._buildDoctype(xml, "r", self.SUBSET)
+        self.assertEqual(out.count("<!DOCTYPE"), 1)
+        self.assertIn(self.SUBSET, out)
+        # our subset must be spliced BEFORE the real subset close, i.e. before the document element
+        self.assertLess(out.index(self.SUBSET), out.index("<r>z</r>"))
+
+
+class TestScanDoctype(unittest.TestCase):
+    """Lexical DOCTYPE scanner: boundaries must be immune to quoted '>', comments and ']>' in values."""
+
+    def test_no_doctype(self):
+        self.assertIsNone(xxe._scanDoctype("<?xml version='1.0'?><r>x</r>"))
+
+    def test_content_start_skips_doctype_with_quoted_bracket(self):
+        # ']>' inside the entity value is NOT the subset end; content starts after the REAL '>'
+        xml = '<!DOCTYPE r [<!ENTITY a "]>trap">]><r>real</r>'
+        cs = xxe._contentStart(xml)
+        self.assertEqual(xml[cs:], "<r>real</r>")
+
+    def test_content_start_skips_doctype_with_comment(self):
+        xml = '<!DOCTYPE r [<!-- ]> not the end --><!ELEMENT r ANY>]><r>real</r>'
+        self.assertEqual(xml[xxe._contentStart(xml):], "<r>real</r>")
+
+    def test_text_node_count_ignores_dtd_bracket_in_value(self):
+        # the '>text<'-looking fragment is inside the DTD entity value, not a body text node
+        xml = '<!DOCTYPE r [<!ENTITY a "]>x">]><r><n>luther</n></r>'
+        self.assertEqual(xxe._textNodeCount(xml), 1)   # only <n>luther</n>
+
 
 class TestPlaceRef(unittest.TestCase):
-    def test_all_text_nodes(self):
+    def test_single_node_preserves_others(self):
+        # ONE location per call - every OTHER value stays intact (no whole-document destruction)
         out = xxe._placeRef("<p><a>one</a><b>two</b></p>", "&e;")
-        self.assertEqual(out.count("&e;"), 2)
-        self.assertNotIn("one", out)
-        self.assertNotIn("two", out)
+        self.assertEqual(out.count("&e;"), 1)
+        self.assertIn("&e;</a>", out)      # default: first text node
+        self.assertIn("two", out)          # second field preserved
 
-    def test_attributes_only_when_requested(self):
-        text = '<u id="1"><n>luther</n></u>'
-        self.assertNotIn('id="&e;"', xxe._placeRef(text, "&e;"))            # attrs off by default
-        self.assertIn('id="&e;"', xxe._placeRef(text, "&e;", attrs=True))   # attrs on
+    def test_index_sweeps_each_node(self):
+        xml = "<p><a>one</a><b>two</b></p>"
+        self.assertEqual(xxe._textNodeCount(xml), 2)
+        out1 = xxe._placeRef(xml, "&e;", index=1)
+        self.assertIn("&e;</b>", out1)     # second text node targeted
+        self.assertIn("one", out1)         # first field preserved
+
+    def test_attribute_seeded_only_as_fallback(self):
+        noText = '<u id="1"><c k="v"/></u>'                                 # no leaf text node
+        self.assertNotIn('="&e;"', xxe._placeRef(noText, "&e;"))            # attrs off -> no seeding
+        self.assertIn('="&e;"', xxe._placeRef(noText, "&e;", attrs=True))   # attrs on -> one attr seeded
+        withText = '<u id="1"><n>luther</n></u>'
+        seeded = xxe._placeRef(withText, "&e;", attrs=True)
+        self.assertIn(">&e;<", seeded)                                      # text node preferred over attr
+        self.assertIn('id="1"', seeded)                                     # attribute preserved
 
     def test_xmlns_preserved(self):
         out = xxe._placeRef('<soap:E xmlns:soap="ns"><b>x</b></soap:E>', "&e;", attrs=True)
@@ -240,6 +284,25 @@ class TestReportMethod(unittest.TestCase):
         finally:
             conf.dumper, conf.method, conf.beep = old_dumper, old_method, old_beep
         self.assertIn("Parameter: XML body (PUT)", captured[0])
+        self.assertIn("Type: XXE injection", captured[0])   # default vuln type
+
+    def test_xxe_internal_entity_is_not_reported_as_xxe(self):
+        # internal-only general-entity expansion is a parser-configuration weakness, NOT confirmed
+        # XXE (which needs external resolution) - it must carry a distinct, weaker vuln type
+        captured = []
+
+        class _Dumper(object):
+            def singleString(self, data, content_type=None):
+                captured.append(data)
+
+        old_dumper, old_method, old_beep = conf.get("dumper"), conf.get("method"), conf.get("beep")
+        conf.dumper, conf.method, conf.beep = _Dumper(), "POST", False
+        try:
+            xxe._report("DTD/internal general entity expansion enabled", "&e;", vulnType="XML parser configuration")
+        finally:
+            conf.dumper, conf.method, conf.beep = old_dumper, old_method, old_beep
+        self.assertIn("Type: XML parser configuration", captured[0])
+        self.assertNotIn("Type: XXE injection", captured[0])
 
 
 class TestHarvestFiles(unittest.TestCase):
@@ -298,6 +361,57 @@ class TestDetectionMocked(unittest.TestCase):
         payload, _ = xxe._tryInternal("<u><n>luther</n></u>", "u", baseline="Hello, luther!")
         self.assertIsNotNone(payload)
 
+    def test_inband_read_rejects_html_escaped_entity_reflection(self):
+        # the app HTML-escapes the reflected entity reference (&amp;<ent>;) between the markers: that is
+        # reflection, NOT an expanded file read - the random entity name survives de-escaping, so it
+        # must be rejected (the P0-5 false positive that fabricated 'file contents')
+        import re as _re
+
+        def mock(body):
+            m = _re.search(r"<!ENTITY (\w+) SYSTEM", body)
+            ent = m.group(1) if m else "e"
+            mk = _re.search(r'(\w{8})&' + _re.escape(ent) + r';(\w{8})', body)   # markers around the ref
+            if mk:
+                m1, m2 = mk.group(1), mk.group(2)
+                return "%s&amp;%s;%s" % (m1, ent, m2)     # ESCAPED reflection, not file content
+            return "no match"
+
+        xxe._send = mock
+        content, _ = xxe._tryInbandFileRead("<u><n>x</n></u>", "u", "/etc/passwd")
+        self.assertIsNone(content)
+
+    def test_inband_read_accepts_genuine_expansion(self):
+        # a genuine file read: the requested path returns real content, a NONEXISTENT path returns
+        # something different -> the matched control passes and the content is accepted
+        import re as _re
+
+        def mock(body):
+            mk = _re.search(r'(\w{8})&\w+;(\w{8})', body)
+            if not (mk and "SYSTEM" in body and "php://filter" not in body):
+                return "nope"
+            if "/etc/passwd" in body:
+                return "%sroot:x:0:0:root:/root:/bin/bash%s" % (mk.group(1), mk.group(2))
+            return "%s%s" % (mk.group(1), mk.group(2))       # nonexistent path -> empty between markers
+
+        xxe._send = mock
+        content, _ = xxe._tryInbandFileRead("<u><n>x</n></u>", "u", "/etc/passwd")
+        self.assertEqual(content, "root:x:0:0:root:/root:/bin/bash")
+
+    def test_inband_read_rejects_path_independent_placeholder(self):
+        # P0-2: a gateway returns a FIXED placeholder for every path (real + nonexistent). The matched
+        # control sees identical content and rejects it as not-genuine file contents.
+        import re as _re
+
+        def mock(body):
+            mk = _re.search(r'(\w{8})&\w+;(\w{8})', body)
+            if not (mk and "SYSTEM" in body and "php://filter" not in body):
+                return "nope"
+            return "%s[external entity disabled]%s" % (mk.group(1), mk.group(2))   # same for ANY path
+
+        xxe._send = mock
+        content, _ = xxe._tryInbandFileRead("<u><n>x</n></u>", "u", "/etc/passwd")
+        self.assertIsNone(content)
+
     def test_internal_echo_rejected(self):
         # endpoint mirrors the raw body back (never parses) -> must NOT be a hit
         xxe._send = lambda body: "You sent: %s" % body
@@ -309,6 +423,30 @@ class TestDetectionMocked(unittest.TestCase):
         payload, _ = xxe._tryInternal("<u><n>luther</n></u>", "u", baseline="already %s here" % xxe.SENTINEL)
         self.assertIsNone(payload)
 
+    def test_location_sweep_finds_non_first_leaf(self):
+        # The first leaf is inside <id> which the (mock) app validates and strips; only the entity ref
+        # placed in the SECOND leaf (<n>) survives and reflects. The sweep must try location #1 and the
+        # engine must latch it so downstream read tiers reuse it - a fixed index=0 would be a false neg.
+        xml = "<u><id>7</id><n>luther</n></u>"
+        self.assertEqual(xxe._textNodeCount(xml), 2)
+
+        def mock(body):
+            # reflect the sentinel only when the entity ref sits in the second leaf (<n>&ent;</n>);
+            # a ref in the first leaf (<id>&ent;</id>) is validated away and never reflects
+            return ("Hello, %s!" % xxe.SENTINEL) if re.search(r"<n>\s*&\w+;", body) else "Hello, !"
+
+        xxe._send = mock
+        xxe._PLACE_INDEX = 0
+        hit = None
+        for i in xxe._sweepLocations(xml):
+            payload, _ = xxe._tryInternal(xml, "u", baseline="Hello, luther!", index=i)
+            if payload:
+                hit = i
+                break
+        self.assertEqual(hit, 1)
+        # location #0 alone (the default) must NOT reflect -> proves the sweep was necessary
+        self.assertIsNone(xxe._tryInternal(xml, "u", baseline="Hello, luther!", index=0)[0])
+
     def test_error_based_positive(self):
         xxe._send = lambda body: 'XML error: failed to load external entity "file:///%s/nonexistent"' % xxe.SENTINEL
         payload, page = xxe._tryError("<u><n>x</n></u>", "u")