Refactoring for --xxe

tests/test_xxe.py

@@ -77,6 +77,17 @@ class TestBuildDoctype(unittest.TestCase):
         self.assertIn(self.SUBSET, out)
         self.assertIn("[", out)
 
+    def test_existing_internal_subset_with_gt_in_entity_value(self):
+        # a '>' inside a quoted entity value must not fool the internal-subset splice
+        out = xxe._buildDoctype('<!DOCTYPE r [<!ENTITY a "x>y">]><r>z</r>', "r", self.SUBSET)
+        self.assertEqual(out.count("<!DOCTYPE"), 1)
+        self.assertIn(self.SUBSET, out)
+
+    def test_existing_external_subset(self):
+        out = xxe._buildDoctype('<!DOCTYPE r SYSTEM "http://x/r.dtd"><r>z</r>', "r", self.SUBSET)
+        self.assertEqual(out.count("<!DOCTYPE"), 1)
+        self.assertIn("[", out)
+
 
 class TestPlaceRef(unittest.TestCase):
     def test_all_text_nodes(self):
@@ -110,6 +121,14 @@ class TestGuards(unittest.TestCase):
         self.assertTrue(xxe._echoed("mirror <!ENTITY x ..."))
         self.assertFalse(xxe._echoed("Hello, expanded value!"))
 
+    def test_echoed_escaped_forms(self):
+        # a debug endpoint that reflects the body in escaped form is still an echo
+        self.assertTrue(xxe._echoed("You sent: &lt;!DOCTYPE r [&lt;!ENTITY e ..."))   # HTML-entity
+        self.assertTrue(xxe._echoed("You sent: %3C!ENTITY e ..."))                     # percent-encoded
+        self.assertTrue(xxe._echoed("You sent: &#x3c;!DOCTYPE ..."))                   # hex numeric
+        self.assertTrue(xxe._echoed("json {\\u003c!ENTITY e ...}"))                    # JSON \\u escape
+        self.assertFalse(xxe._echoed("result: sentineltoken1"))                        # pure expanded value
+
     def test_fingerprint(self):
         self.assertIsNotNone(xxe._fingerprint('failed to load external entity "file:///x"'))
         self.assertIsNotNone(xxe._fingerprint('failed to load "file:///x": No such file'))  # PHP form
@@ -141,6 +160,103 @@ class TestOobToggle(unittest.TestCase):
         conf.oobServer = None
 
 
+class TestOobConsent(unittest.TestCase):
+    """No third-party OOB without explicit opt-in: --oob-server none disables it,
+    an explicit server opts in, and an unset server under --batch defaults to NO."""
+
+    def setUp(self):
+        self._batch = conf.get("batch")
+        conf.batch = True
+        xxe._OOB_CONSENT = None
+
+    def tearDown(self):
+        conf.batch = self._batch
+        conf.oobServer = None
+        xxe._OOB_CONSENT = None
+
+    def test_none_denies(self):
+        conf.oobServer = "none"
+        self.assertFalse(xxe._oobConsent())
+
+    def test_explicit_server_opts_in(self):
+        conf.oobServer = "my.interactsh.example"
+        self.assertTrue(xxe._oobConsent())
+
+    def test_unset_under_batch_defaults_no(self):
+        conf.oobServer = None
+        self.assertFalse(xxe._oobConsent())   # readInput returns the 'N' default under --batch
+
+
+class TestPathNorm(unittest.TestCase):
+    def test_unix(self):
+        self.assertEqual(xxe._toSystemId("/etc/passwd"), "file:///etc/passwd")
+        self.assertEqual(xxe._toResource("/etc/passwd"), "/etc/passwd")
+
+    def test_windows(self):
+        self.assertEqual(xxe._toSystemId("C:\\Windows\\win.ini"), "file:///C:/Windows/win.ini")
+        self.assertEqual(xxe._toResource("C:\\Windows\\win.ini"), "C:/Windows/win.ini")
+
+    def test_already_uri(self):
+        self.assertEqual(xxe._toSystemId("file:///etc/passwd"), "file:///etc/passwd")
+        self.assertEqual(xxe._toSystemId("php://filter/x"), "php://filter/x")
+        self.assertEqual(xxe._toResource("file:///etc/passwd"), "/etc/passwd")
+
+
+class TestMarkerHonored(unittest.TestCase):
+    def tearDown(self):
+        xxe._MARKER = None
+
+    def test_place_ref_honors_marker(self):
+        xxe._MARKER = "xxemarkzzzz"
+        out = xxe._placeRef("<u><n>xxemarkzzzz</n><m>other</m></u>", "&e;")
+        self.assertIn("<n>&e;</n>", out)
+        self.assertIn("<m>other</m>", out)          # other node left intact
+        self.assertNotIn("xxemarkzzzz", out)
+
+    def test_clean_body_sets_marker_on_user_marks(self):
+        conf.data = "<u><n>luther%s</n></u>" % (kb.customInjectionMark or "*")
+        kb.processUserMarks = True
+        try:
+            cleaned = xxe._cleanBody()
+            self.assertIsNotNone(xxe._MARKER)
+            self.assertIn(xxe._MARKER, cleaned)
+        finally:
+            kb.processUserMarks = False
+            xxe._MARKER = None
+
+
+class TestReportMethod(unittest.TestCase):
+    def test_report_uses_conf_method(self):
+        captured = []
+
+        class _Dumper(object):
+            def singleString(self, data, content_type=None):
+                captured.append(data)
+
+        old_dumper, old_method, old_beep = conf.get("dumper"), conf.get("method"), conf.get("beep")
+        conf.dumper, conf.method, conf.beep = _Dumper(), "PUT", False
+        try:
+            xxe._report("Title", "Payload")
+        finally:
+            conf.dumper, conf.method, conf.beep = old_dumper, old_method, old_beep
+        self.assertIn("PUT XML body", captured[0])
+
+
+class TestOobBase64Capture(unittest.TestCase):
+    def test_path_capture_survives_plus_slash_equals(self):
+        import base64
+        from lib.core.convert import getText, decodeBase64
+        marker = "mk12345678"
+        raw = b">>>\xff\xfe some + / = data =="
+        blob = getText(base64.b64encode(raw))
+        self.assertTrue(any(c in blob for c in "+/="))   # ensure the risky chars are present
+        url = "http://webhook.site/tok/%s/%s" % (marker, blob)   # base64 in the PATH
+        m = re.search(r"/%s/([A-Za-z0-9+/=]+)" % re.escape(marker), url)
+        self.assertIsNotNone(m)
+        self.assertEqual(m.group(1), blob)
+        self.assertEqual(decodeBase64(m.group(1)), raw)
+
+
 class TestDetectionMocked(unittest.TestCase):
     def setUp(self):
         self._send = xxe._send