test: add unit test suite for dry-run mode, Nuclei export, tech-CVE mapping, and Burp import functionality

Author: 0

src/spring2shell/cli.py

@@ -163,7 +163,33 @@ def _cmd_encrypt(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_nuclei_export(args: argparse.Namespace) -> int:
+    """Export scan findings as Nuclei v3 YAML templates."""
+    import json
+    from pathlib import Path
+    from spring2shell.utils.nuclei_export import export_templates
+
+    report_path = Path(args.report_file)
+    if not report_path.exists():
+        print(f"[!] Report file not found: {report_path}")
+        return 1
+    try:
+        data = json.loads(report_path.read_text())
+        findings = data.get("findings", data) if isinstance(data, dict) else data
+    except Exception as exc:
+        print(f"[!] Could not parse report: {exc}")
+        return 1
+    created = export_templates(findings, args.output_dir)
+    print(f"[+] {len(created)} Nuclei templates written to {args.output_dir}")
+    for p in created[:10]:
+        print(f"    {p}")
+    if len(created) > 10:
+        print(f"    ... and {len(created) - 10} more")
+    return 0
+
+
 def _cmd_verify(args: argparse.Namespace) -> int:
+
     """Verify a potential RCE: echo-marker test + optional blind (time-delay + DNS)."""
     from spring2shell.core.verifier import check_real_rce, blind_rce_test
 
@@ -422,6 +448,11 @@ def build_parser() -> argparse.ArgumentParser:
         default="default",
         help="Runtime profile (timeout, retries, delay). Default: default.",
     )
+    parser.add_argument(
+        "--dry-run", action="store_true", default=False,
+        help="Print payloads and endpoints that WOULD be sent without making real HTTP requests.",
+    )
+
 
     # ── Auth flags ───────────────────────────────────────────────────────────
     auth_group = parser.add_argument_group("authentication")
@@ -559,24 +590,31 @@ def build_parser() -> argparse.ArgumentParser:
     p_enc.add_argument("--remove-original", action="store_true",
                        help="Delete plaintext file after encryption.")
 
+    # nuclei-export
+    p_nuclei = sub.add_parser("nuclei-export", help="Export findings as Nuclei v3 YAML templates.")
+    p_nuclei.add_argument("report_file", help="JSON report file to export from.")
+    p_nuclei.add_argument("output_dir", help="Directory to write Nuclei templates into.")
+
     return parser
 
 
 _COMMAND_MAP = {
-    "safe-audit": _cmd_safe_audit,
-    "log-audit":  _cmd_log_audit,
-    "direct":     _cmd_direct,
-    "scan":       _cmd_scan,
-    "cve-scan":   _cmd_cve_scan,
+    "safe-audit":     _cmd_safe_audit,
+    "log-audit":      _cmd_log_audit,
+    "direct":         _cmd_direct,
+    "scan":           _cmd_scan,
+    "cve-scan":       _cmd_cve_scan,
     "ssrf-scan":  _cmd_ssrf_scan,
-    "ssti-scan":  _cmd_ssti_scan,
-    "encrypt":    _cmd_encrypt,
-    "verify":     _cmd_verify,
-    "exploit":    _cmd_exploit,
-    "menu":       _cmd_menu,
+    "ssti-scan":      _cmd_ssti_scan,
+    "encrypt":        _cmd_encrypt,
+    "verify":         _cmd_verify,
+    "exploit":        _cmd_exploit,
+    "menu":           _cmd_menu,
+    "nuclei-export":  _cmd_nuclei_export,
 }
 
 
+
 # ---------------------------------------------------------------------------
 # Main entry point
 # ---------------------------------------------------------------------------
@@ -658,6 +696,16 @@ def _configure_subsystems(args: argparse.Namespace) -> None:
     except ImportError:
         pass
 
+    # Dry-run mode must be configured last (it overrides session creation)
+    if getattr(args, "dry_run", False):
+        try:
+            from spring2shell.utils.dry_run import enable_dry_run
+            enable_dry_run()
+            print("[DRY-RUN] Mode enabled — no real HTTP requests will be sent.")
+        except ImportError:
+            pass
+
+
 
 if __name__ == "__main__":
     main()

src/spring2shell/utils/burp_import.py

@@ -59,11 +59,11 @@ def load_targets(path: str | Path) -> list[str]:
 def _parse_burp_xml(content: str) -> list[str]:
     """Parse Burp Suite XML export — <items><item><url>…</url></item></items>."""
     urls: list[str] = []
-    # Use regex to avoid requiring xml.etree on all platforms
+    # Strip CDATA wrappers: <![CDATA[...]]> → content inside
+    content = re.sub(r"<!\[CDATA\[(.*?)]]>", r"\1", content, flags=re.DOTALL)
     pattern = re.compile(r"<url>(.*?)</url>", re.DOTALL | re.IGNORECASE)
     for match in pattern.finditer(content):
         url = match.group(1).strip()
-        # Burp sometimes CDATA-encodes or entity-encodes URLs
         url = url.replace("&amp;", "&").replace("&lt;", "<").replace("&gt;", ">")
         if url.startswith(("http://", "https://")):
             urls.append(url)

tests/unit/test_burp_import.py

@@ -0,0 +1,132 @@
+"""Unit tests for burp_import.py — B1: Burp Suite target import."""
+from __future__ import annotations
+
+from pathlib import Path
+import pytest
+
+from spring2shell.utils.burp_import import (
+    _parse_burp_xml,
+    _parse_plain_txt,
+    _to_base_urls,
+    load_targets,
+)
+
+
+_BURP_XML = """<?xml version="1.0" ?>
+<!DOCTYPE items [<!ENTITY xxe SYSTEM "">]>
+<items burpVersion="2023.1" exportTime="Mon Jan 01 00:00:00 UTC 2023">
+  <item>
+    <url><![CDATA[https://target.example/api/graphql?query=test]]></url>
+  </item>
+  <item>
+    <url>https://target.example/actuator/env</url>
+  </item>
+  <item>
+    <url>https://other.example:8443/login</url>
+  </item>
+  <item>
+    <url>ftp://skip.example/file</url>
+  </item>
+</items>
+"""
+
+_PLAIN_TXT = """# My test targets
+https://alpha.example/api
+https://alpha.example/api/v2
+http://beta.example:8080/health
+# comment line
+invalid-line-no-scheme
+gamma.example
+"""
+
+
+class TestParseBurpXml:
+    def test_extracts_https_urls(self):
+        urls = _parse_burp_xml(_BURP_XML)
+        assert any("target.example" in u for u in urls)
+        assert any("other.example" in u for u in urls)
+
+    def test_skips_non_http_urls(self):
+        urls = _parse_burp_xml(_BURP_XML)
+        assert not any("ftp://" in u for u in urls)
+
+    def test_handles_cdata_urls(self):
+        """URLs inside CDATA sections should still be extracted."""
+        xml = """<items>
+  <item><url><![CDATA[https://target.example/api/graphql?query=test]]></url></item>
+</items>"""
+        urls = _parse_burp_xml(xml)
+        # CDATA markers are removed — the actual URL should be present
+        assert any("graphql" in u and "target.example" in u for u in urls), \
+            f"Expected URL with 'graphql' in parsed URLs, got: {urls}"
+
+
+
+class TestParsePlainTxt:
+    def test_returns_http_urls(self):
+        urls = _parse_plain_txt(_PLAIN_TXT)
+        assert "https://alpha.example/api" in urls
+        assert "http://beta.example:8080/health" in urls
+
+    def test_skips_comments(self):
+        urls = _parse_plain_txt(_PLAIN_TXT)
+        assert not any(u.startswith("#") for u in urls)
+
+    def test_handles_bare_hostname(self):
+        urls = _parse_plain_txt(_PLAIN_TXT)
+        # gamma.example gets https:// prepended
+        assert any("gamma.example" in u for u in urls)
+
+    def test_skips_invalid_lines(self):
+        urls = _parse_plain_txt(_PLAIN_TXT)
+        assert "invalid-line-no-scheme" not in urls
+
+
+class TestToBaseUrls:
+    def test_deduplicates_same_host_different_paths(self):
+        raw = [
+            "https://example.com/api/v1",
+            "https://example.com/api/v2",
+            "https://example.com/actuator",
+        ]
+        bases = _to_base_urls(raw)
+        assert bases == ["https://example.com"]
+
+    def test_keeps_non_standard_port(self):
+        raw = ["https://example.com:8443/api"]
+        bases = _to_base_urls(raw)
+        assert "https://example.com:8443" in bases
+
+    def test_strips_default_ports(self):
+        raw = ["https://example.com:443/api", "http://example.com:80/api"]
+        bases = _to_base_urls(raw)
+        assert "https://example.com" in bases
+        assert "http://example.com" in bases
+
+    def test_filters_non_http_urls(self):
+        raw = ["ftp://example.com/file", "https://ok.example"]
+        bases = _to_base_urls(raw)
+        assert not any("ftp" in b for b in bases)
+
+
+class TestLoadTargets:
+    def test_xml_file(self, tmp_path):
+        xml_file = tmp_path / "burp.xml"
+        xml_file.write_text(_BURP_XML)
+        targets = load_targets(xml_file)
+        assert len(targets) >= 2
+        assert all(t.startswith(("http://", "https://")) for t in targets)
+
+    def test_plain_text_file(self, tmp_path):
+        txt_file = tmp_path / "targets.txt"
+        txt_file.write_text(_PLAIN_TXT)
+        targets = load_targets(txt_file)
+        assert len(targets) >= 2
+
+    def test_deduplication_across_full_load(self, tmp_path):
+        txt_file = tmp_path / "dup_targets.txt"
+        txt_file.write_text(
+            "https://example.com/api\nhttps://example.com/login\nhttps://example.com/health\n"
+        )
+        targets = load_targets(txt_file)
+        assert targets == ["https://example.com"]

tests/unit/test_dry_run.py

@@ -0,0 +1,110 @@
+"""Unit tests for dry_run.py — C3: dry-run mode."""
+from __future__ import annotations
+
+import pytest
+
+import spring2shell.utils.dry_run as dr_module
+from spring2shell.utils.dry_run import (
+    DryRunSession,
+    disable_dry_run,
+    enable_dry_run,
+    is_dry_run,
+)
+
+
+@pytest.fixture(autouse=True)
+def reset_dry_run():
+    """Always reset dry-run state between tests."""
+    disable_dry_run()
+    yield
+    disable_dry_run()
+
+
+class TestDryRunFlag:
+    def test_disabled_by_default(self):
+        assert not is_dry_run()
+
+    def test_enable_sets_flag(self):
+        enable_dry_run()
+        assert is_dry_run()
+
+    def test_disable_clears_flag(self):
+        enable_dry_run()
+        disable_dry_run()
+        assert not is_dry_run()
+
+
+class TestDryRunSession:
+    def test_get_returns_fake_response(self):
+        session = DryRunSession()
+        resp = session.get("http://example.com/test")
+        assert resp.status_code == 200
+        assert resp.text == "DRY_RUN_RESPONSE"
+
+    def test_post_returns_fake_response(self):
+        session = DryRunSession()
+        resp = session.post("http://example.com/api", data='{"key": "value"}')
+        assert resp.status_code == 200
+
+    def test_all_http_methods_work(self):
+        session = DryRunSession()
+        for method in [session.get, session.post, session.put,
+                       session.patch, session.delete, session.head]:
+            resp = method("http://example.com/")
+            assert resp.status_code == 200
+
+    def test_prints_method_and_url(self, capsys):
+        session = DryRunSession()
+        session.get("http://target.example/api/graphql")
+        captured = capsys.readouterr()
+        assert "[DRY-RUN]" in captured.out
+        assert "GET" in captured.out
+        assert "target.example" in captured.out
+
+    def test_prints_post_body(self, capsys):
+        session = DryRunSession()
+        session.post("http://example.com/api", data='{"query": "test"}')
+        captured = capsys.readouterr()
+        assert "query" in captured.out
+
+    def test_prints_headers(self, capsys):
+        session = DryRunSession()
+        session.get("http://example.com", headers={"Content-Type": "application/json"})
+        captured = capsys.readouterr()
+        assert "Content-Type" in captured.out
+
+    def test_redacts_authorization_header(self, capsys):
+        session = DryRunSession()
+        session.get("http://example.com", headers={"Authorization": "Bearer secret-token"})
+        captured = capsys.readouterr()
+        # Authorization header should NOT appear in output
+        assert "secret-token" not in captured.out
+
+    def test_prints_query_params(self, capsys):
+        session = DryRunSession()
+        session.get("http://example.com/search", params={"q": "test", "page": "1"})
+        captured = capsys.readouterr()
+        assert "q=test" in captured.out or "q" in captured.out
+
+    def test_mount_does_not_crash(self):
+        session = DryRunSession()
+        session.mount("https://", None)  # should not raise
+
+    def test_fake_response_json_returns_empty_dict(self):
+        session = DryRunSession()
+        resp = session.post("http://example.com")
+        assert resp.json() == {}
+
+    def test_fake_response_raise_for_status_no_op(self):
+        session = DryRunSession()
+        resp = session.get("http://example.com")
+        resp.raise_for_status()  # should not raise
+
+
+class TestSessionIntegration:
+    def test_create_stealth_session_returns_dry_run_session(self):
+        """When dry-run is active, create_stealth_session() returns DryRunSession."""
+        enable_dry_run()
+        from spring2shell.core.session import create_stealth_session
+        session = create_stealth_session()
+        assert isinstance(session, DryRunSession)