From af6e843b80909d71fe65a448334b99df49063f50 Mon Sep 17 00:00:00 2001
From: liangzaihuanyv <144539158+liangzaihuanyv@users.noreply.github.com>
Date: Thu, 29 Oct 2026 18:41:23 +0800
Subject: [PATCH] fix: preserve protocol metadata in Burp XML imports
---
lib/core/common.py | 16 +++-
tests/test_common.py | 181 +++++++++++++++++++++++++++++++++++++++++++
2 files changed, 195 insertions(+), 2 deletions(-)
diff --git a/lib/core/common.py b/lib/core/common.py
index ee3a26e70a7..7a9dd8a82a5 100644
--- a/lib/core/common.py
+++ b/lib/core/common.py
@@ -5694,6 +5694,8 @@ def _parseBurpLog(content):
reqResList = []
for match in re.finditer(BURP_XML_HISTORY_REGEX, content, re.I | re.S):
port, request = match.groups()
+ protocol = extractRegexResult(r"\s*(?Phttps?)\s*",
+ match.group(0), re.I)
try:
request = decodeBase64(request, binary=False)
except (binascii.Error, TypeError):
@@ -5703,14 +5705,20 @@ def _parseBurpLog(content):
host = _.group(0).strip()
if not re.search(r":\d+\Z", host) and int(port) != 80:
request = request.replace(host, "%s:%d" % (host, int(port)))
- reqResList.append(request)
+ reqResList.append((request, protocol.lower() if protocol else None, port))
else:
reqResList = [content]
else:
reqResList = re.finditer(BURP_REQUEST_REGEX, content, re.I | re.S)
for match in reqResList:
- request = match if isinstance(match, six.string_types) else match.group(1)
+ xmlScheme, xmlPort = None, None
+
+ if isinstance(match, tuple):
+ request, xmlScheme, xmlPort = match
+ else:
+ request = match if isinstance(match, six.string_types) else match.group(1)
+
request = re.sub(r"\A[^\w]+", "", request)
schemePort = re.search(r"(http[\w]*)\:\/\/.*?\:([\d]+).+?={10,}", request, re.I | re.S)
@@ -5814,6 +5822,10 @@ def _parseBurpLog(content):
data = data.rstrip("\r\n") if data else data
+ if xmlScheme:
+ scheme = scheme or xmlScheme
+ port = port or xmlPort
+
if getPostReq and (params or cookie or not checkParams):
if not port and hasattr(scheme, "lower") and scheme.lower() == "https":
port = "443"
diff --git a/tests/test_common.py b/tests/test_common.py
index ddede42ab29..45ffe0e8b33 100644
--- a/tests/test_common.py
+++ b/tests/test_common.py
@@ -561,6 +561,187 @@ def test_scope_filters_out_nonmatching(self):
self.assertEqual(targets, [])
+class TestParseRequestFileBurpXmlProtocol(unittest.TestCase):
+ """Burp XML is kept per item; metadata only fills missing fields."""
+
+ def setUp(self):
+ self._scope = conf.scope
+ self._method = conf.method
+ self._headers = conf.headers
+ self._force_ssl = conf.forceSSL
+ conf.scope = None
+ conf.method = None
+ conf.headers = None
+ conf.forceSSL = False
+
+ def tearDown(self):
+ conf.scope = self._scope
+ conf.method = self._method
+ conf.headers = self._headers
+ conf.forceSSL = self._force_ssl
+
+ @staticmethod
+ def _request(method, target, host="example.test", headers=(), body=None):
+ lines = ["%s %s HTTP/1.1" % (method, target), "Host: %s" % host]
+ lines.extend(headers)
+ if body is not None:
+ lines.append("Content-Length: %d" % len(body))
+ lines.extend(("", body or ""))
+ return "\r\n".join(lines)
+
+ @staticmethod
+ def _item(port, protocol, request):
+ tag = "" if protocol is None else "%s" % protocol
+ encoded = base64.b64encode(request.encode("utf-8")).decode("ascii")
+ return ('- %s%s'
+ '
' % (port, tag, encoded))
+
+ def _parse(self, items):
+ xml = "%s" % "".join(items)
+ path = _write_temp(xml, ".xml")
+ try:
+ return list(parseRequestFile(path))
+ finally:
+ os.unlink(path)
+
+ def _url(self, port, protocol, request):
+ targets = self._parse([self._item(port, protocol, request)])
+ self.assertEqual(len(targets), 1)
+ return targets[0][0]
+
+ def test_https_on_nondefault_port(self):
+ url = self._url(8443, "https", self._request("GET", "/api?id=1"))
+ self.assertEqual(url, "https://example.test:8443/api?id=1")
+
+ def test_https_on_port_80(self):
+ url = self._url(80, "https", self._request("GET", "/api?id=1"))
+ self.assertEqual(url, "https://example.test:80/api?id=1")
+
+ def test_http_on_port_443(self):
+ url = self._url(443, "http", self._request("GET", "/api?id=1"))
+ self.assertEqual(url, "http://example.test:443/api?id=1")
+
+ def test_standard_protocol_controls(self):
+ cases = ((80, "http", "http://example.test:80/api?id=1"),
+ (443, "https", "https://example.test:443/api?id=1"),
+ (8080, "http", "http://example.test:8080/api?id=1"))
+ for port, protocol, expected in cases:
+ self.assertEqual(self._url(port, protocol, self._request("GET", "/api?id=1")), expected)
+
+ def test_multiple_items_reset_protocol_context(self):
+ items = [
+ self._item(8443, "https", self._request("GET", "/one?id=1")),
+ self._item(8080, None, self._request("GET", "/two?id=1")),
+ self._item(443, "http", self._request("GET", "/three?id=1")),
+ self._item(443, "https", self._request("GET", "/four?id=1")),
+ ]
+ urls = [target[0] for target in self._parse(items)]
+ self.assertEqual(urls, ["https://example.test:8443/one?id=1",
+ "http://example.test:8080/two?id=1",
+ "http://example.test:443/three?id=1",
+ "https://example.test:443/four?id=1"])
+
+ def test_missing_or_unknown_protocol_keeps_legacy(self):
+ items = [
+ self._item(8443, None, self._request("GET", "/missing?id=1")),
+ self._item(8443, "ftp", self._request("GET", "/unknown?id=1")),
+ ]
+ urls = [target[0] for target in self._parse(items)]
+ self.assertEqual(urls, ["http://example.test:8443/missing?id=1",
+ "http://example.test:8443/unknown?id=1"])
+
+ def test_missing_protocol_does_not_fill_xml_port(self):
+ # Legacy: a Host-provided scheme implies 443; XML port 80 must not override it.
+ request = self._request("GET", "/index?id=1", host="https://example.test")
+ self.assertEqual(self._url(80, None, request), "https://example.test:443/index?id=1")
+
+ def test_host_explicit_port_wins_over_xml_port(self):
+ request = self._request("GET", "/index?id=1", host="example.test:9443")
+ self.assertEqual(self._url(8443, "https", request), "https://example.test:9443/index?id=1")
+
+ def test_metadata_case_and_whitespace(self):
+ self.assertEqual(self._url(8443, " HTTPS ", self._request("GET", "/index?id=1")),
+ "https://example.test:8443/index?id=1")
+
+ def test_force_ssl_remains_final_override(self):
+ conf.forceSSL = True
+ self.assertEqual(self._url(8080, "http", self._request("GET", "/index?id=1")),
+ "https://example.test:8080/index?id=1")
+
+ def test_absolute_request_target_takes_precedence(self):
+ request = self._request("GET", "http://absolute.example.test:8080/abs?id=1")
+ self.assertEqual(self._url(8443, "https", request),
+ "http://absolute.example.test:8080/abs?id=1")
+
+ def test_host_scheme_takes_precedence(self):
+ request = self._request("GET", "/index?id=1", host="http://example.test")
+ self.assertEqual(self._url(8443, "https", request), "http://example.test:8443/index?id=1")
+
+ def test_method_body_cookie_and_headers_retained(self):
+ request = self._request("POST", "/login",
+ headers=("Cookie: SID=abc", "X-Test: keep"),
+ body="user=admin&pw=1")
+ targets = self._parse([self._item(8443, "https", request)])
+ self.assertEqual(len(targets), 1)
+ url, method, data, cookie, headers = targets[0]
+ self.assertEqual(url, "https://example.test:8443/login")
+ self.assertEqual(method, HTTPMETHOD.POST)
+ self.assertEqual(data, "user=admin&pw=1")
+ self.assertEqual(cookie, "SID=abc")
+ self.assertIn(("X-Test", "keep"), headers)
+
+ def test_scope_uses_final_protocol_url(self):
+ item = self._item(8443, "https", self._request("GET", "/index?id=1"))
+ path = _write_temp("%s" % item, ".xml")
+ try:
+ conf.scope = r"^http://example\.test:8443/"
+ self.assertEqual(list(parseRequestFile(path)), [])
+ conf.scope = r"^https://example\.test:8443/"
+ self.assertEqual(len(list(parseRequestFile(path))), 1)
+ finally:
+ os.unlink(path)
+
+ def test_plain_raw_request_without_hint_unchanged(self):
+ for host, expected in (("example.test", "http://example.test:80/index?id=1"),
+ ("example.test:443", "https://example.test:443/index?id=1")):
+ content = "GET /index?id=1 HTTP/1.1\r\nHost: %s\r\n\r\n" % host
+ path = _write_temp(content, ".log")
+ try:
+ targets = list(parseRequestFile(path))
+ finally:
+ os.unlink(path)
+ self.assertEqual(len(targets), 1)
+ self.assertEqual(targets[0][0], expected)
+
+ def test_set_multiple_targets_mixed_xml(self):
+ from lib.core.option import _setMultipleTargets
+
+ items = [
+ self._item(8080, "http", self._request("GET", "/plain?id=1")),
+ self._item(8443, "https", self._request("GET", "/secure?id=1")),
+ self._item(443, "https", self._request("GET", "/standard?id=1")),
+ ]
+ path = _write_temp("%s" % "".join(items), ".xml")
+ saved_targets = kb.targets
+ saved_log_file = conf.logFile
+ saved_param_del = conf.paramDel
+ try:
+ kb.targets = type(saved_targets)()
+ conf.logFile = path
+ conf.paramDel = None
+ _setMultipleTargets()
+ urls = sorted(target[0] for target in kb.targets)
+ self.assertEqual(len(kb.targets), 3)
+ self.assertEqual(urls, ["http://example.test:8080/plain?id=1",
+ "https://example.test:443/standard?id=1",
+ "https://example.test:8443/secure?id=1"])
+ finally:
+ kb.targets = saved_targets
+ conf.logFile = saved_log_file
+ conf.paramDel = saved_param_del
+ os.unlink(path)
+
+
class TestParseRequestFileWebScarab(unittest.TestCase):
"""_parseWebScarabLog via parseRequestFile."""