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)
This commit is contained in:
YueGuobin 2026-05-30 01:02:21 +08:00
parent 87f38b1507
commit 7bcb96a368
No known key found for this signature in database
4 changed files with 58 additions and 57 deletions

View File

@ -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

View File

@ -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:

View File

@ -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:

View File

@ -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."""