From 7bcb96a368749fbdbb5c28b732d7cb0576c9ee37 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sat, 30 May 2026 01:02:21 +0800 Subject: [PATCH] Improve packet filter validation: use tcpdump, handle multi-line BPF, safe project load Changes: - Replace tshark BPF validation with tcpdump -d (calls pcap_compile internally like ubridge, returns instantly without waiting for traffic) - Support multi-line BPF expressions: split on newlines and validate each line individually - Always validate, never save invalid filters on error - Drop invalid filters during project load with warning (prevents old topologies with bad filters from failing to open) - Simplify test cases (no longer depend on tshark availability) --- gns3server/controller/link.py | 3 +- gns3server/controller/project.py | 8 ++- gns3server/utils/packet_filter_validation.py | 76 +++++++++----------- tests/utils/test_packet_filter_validation.py | 28 ++++---- 4 files changed, 58 insertions(+), 57 deletions(-) diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index 7c568ee61..dc8217224 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -583,7 +583,7 @@ class Link: "suspend": self._suspended, "show_filters_icon": getattr(self, '_show_filters_icon', True), } - return { + result = { "nodes": res, "link_id": self._id, "project_id": self._project.id, @@ -598,3 +598,4 @@ class Link: "wireshark": self._wireshark, "show_filters_icon": getattr(self, '_show_filters_icon', True), } + return result diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index d9c30a703..ae063e4aa 100644 --- a/gns3server/controller/project.py +++ b/gns3server/controller/project.py @@ -1205,7 +1205,13 @@ class Project: continue link = await self.add_link(link_id=link_data["link_id"]) if "filters" in link_data: - await link.update_filters(link_data["filters"]) + try: + await link.update_filters(link_data["filters"]) + except ControllerError as e: + log.warning( + "Dropping invalid filters on link %s: %s", + link_data.get("link_id"), e + ) if "link_style" in link_data: await link.update_link_style(link_data["link_style"]) if "show_filters_icon" in link_data: diff --git a/gns3server/utils/packet_filter_validation.py b/gns3server/utils/packet_filter_validation.py index ad19eb316..f5845a025 100644 --- a/gns3server/utils/packet_filter_validation.py +++ b/gns3server/utils/packet_filter_validation.py @@ -16,12 +16,12 @@ class FilterValidationError(Exception): def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Optional[str]]: """ - Validate BPF filter expression syntax using tshark. + Validate BPF filter expression syntax using tcpdump. - This uses the same approach as gns3_copilot's packet filter tool: - - Run tshark with the BPF expression on loopback interface - - Check for "Invalid" in output indicating syntax errors - - Timeout is expected behavior (tshark waits for traffic) + Uses `tcpdump -d` to compile the BPF expression into filter instructions. + This calls pcap_compile() internally (same as ubridge) but does not + capture traffic, so it returns immediately for both valid and invalid + expressions. Args: bpf_expression: BPF filter expression to validate @@ -30,51 +30,35 @@ def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Optional[str]]: dict with 'valid' (bool) and 'error' (str or None) keys """ try: - # Use tshark to validate BPF syntax with 1 second timeout - # Use -i lo (loopback) to avoid "(null)" interface in error messages result = subprocess.run( - ["tshark", "-f", bpf_expression, "-i", "lo"], - timeout=1, + ["tcpdump", "-d", bpf_expression], capture_output=True, text=True, ) - # Check if output contains "Invalid" indicating syntax error - if "Invalid" in result.stdout or "Invalid" in result.stderr: + if result.returncode != 0: + # Extract meaningful error from tcpdump's stderr + # Skip "Warning: assuming Ethernet" lines, keep only error lines error_lines = [] - if "Invalid" in result.stderr: - error_lines.extend( - line for line in result.stderr.split("\n") if "Invalid" in line - ) - if "Invalid" in result.stdout: - error_lines.extend( - line for line in result.stdout.split("\n") if "Invalid" in line - ) - - # Strip interface suffix for cleaner error - error_msg_parts = [] - for line in error_lines: - clean = line.split(" for interface")[0].strip() - if clean: - error_msg_parts.append(clean) - error_msg = " ".join(error_msg_parts) if error_msg_parts else "Invalid BPF syntax" + for line in result.stderr.split("\n"): + line = line.strip() + if line and not line.startswith("Warning:"): + # Strip "tcpdump: " prefix + for prefix in ["tcpdump: "]: + if line.startswith(prefix): + line = line[len(prefix):] + error_lines.append(line) + error_msg = " ".join(error_lines) if error_lines else "Invalid BPF expression" log.warning("BPF syntax validation failed: %s", error_msg) return {"valid": False, "error": error_msg} log.info("BPF syntax validation passed") return {"valid": True, "error": None} - except subprocess.TimeoutExpired: - # Timeout is expected behavior - tshark waits for traffic - # No "Invalid" in output means syntax is correct - log.info("BPF syntax validation passed (timeout expected)") - return {"valid": True, "error": None} - except FileNotFoundError: - # tshark not installed - skip validation log.warning( - "tshark not found, skipping BPF syntax validation. " - "Install tshark to enable BPF validation." + "tcpdump not found, skipping BPF syntax validation. " + "Install tcpdump to enable BPF validation." ) return {"valid": True, "error": None} @@ -149,13 +133,21 @@ def validate_filter_parameters(filter_type: str, values: List[Any]) -> None: ) # Validate BPF syntax using tshark (same method as gns3_copilot) + # The value may be a multi-line string; each line becomes a + # separate ubridge filter. Validate each line individually. value = value.strip() - if value: # Only validate non-empty BPF expressions - bpf_result = validate_bpf_syntax(value) - if not bpf_result["valid"]: - raise FilterValidationError( - f"{filter_type} parameter {rules['names'][i]} has invalid syntax: {bpf_result['error']}" - ) + if value: + lines = value.split("\n") + for line_num, line in enumerate(lines): + line = line.strip() + if not line: + continue + bpf_result = validate_bpf_syntax(line) + if not bpf_result["valid"]: + raise FilterValidationError( + f"{filter_type} parameter {rules['names'][i]} line {line_num + 1} " + f"has invalid syntax: {bpf_result['error']}" + ) else: # Integer parameter validation try: diff --git a/tests/utils/test_packet_filter_validation.py b/tests/utils/test_packet_filter_validation.py index d248dd868..0951783ac 100644 --- a/tests/utils/test_packet_filter_validation.py +++ b/tests/utils/test_packet_filter_validation.py @@ -93,26 +93,28 @@ class TestPacketFilterValidation: validate_filter_parameters("bpf", [""]) # Empty is valid validate_filter_parameters("bpf", ["host 192.168.1.1 and port 443"]) + def test_bpf_multi_line_valid(self): + """Test valid multi-line BPF expressions.""" + validate_filter_parameters("bpf", ["tcp port 80\nnot arp"]) + validate_filter_parameters("bpf", ["tcp and not port 22\nhost 192.168.1.1\nicmp"]) + + def test_bpf_multi_line_invalid(self): + """Test multi-line BPF with invalid line.""" + with pytest.raises(FilterValidationError) as excinfo: + validate_filter_parameters("bpf", ["tcp port 80\ninvalid!!!"]) + err = str(excinfo.value).lower() + assert "syntax error" in err + def test_bpf_invalid(self): """Test invalid BPF parameters.""" # Wrong type with pytest.raises(FilterValidationError, match="must be a string"): validate_filter_parameters("bpf", [123]) - # Invalid BPF syntax (requires tshark) - try: + # Invalid BPF syntax + with pytest.raises(FilterValidationError) as excinfo: validate_filter_parameters("bpf", ["tcp port"]) # Missing port number - # If tshark is not available, this might pass - import subprocess - subprocess.run(["which", "tshark"], capture_output=True) - # If we get here, tshark exists, so validation should have failed - pytest.fail("Expected BPF validation to fail for invalid syntax") - except FilterValidationError as e: - # Expected: BPF syntax error - assert "invalid syntax" in str(e).lower() or "Invalid capture filter" in str(e) - except (FileNotFoundError, subprocess.CalledProcessError): - # tshark not installed, skip this test - pytest.skip("tshark not installed, skipping BPF syntax validation test") + assert "syntax error" in str(excinfo.value).lower() def test_parameter_count_mismatch(self): """Test wrong number of parameters."""