diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index 14ff447ea..7c568ee61 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -23,6 +23,7 @@ import html from .controller_error import ControllerError, ControllerNotFoundError from gns3server.agent.web_wireshark.manager import WebWiresharkManager from gns3server.config import Config +from gns3server.utils.packet_filter_validation import validate_all_filters, FilterValidationError import logging @@ -161,6 +162,12 @@ class Link: if len(values) != 0 and values[0] != 0 and values[0] != "": new_filters[filter] = values + # Validate filter parameters before applying + try: + validate_all_filters(new_filters) + except FilterValidationError as e: + raise ControllerError(f"Invalid packet filter parameters: {str(e)}") + if new_filters != self.filters: self._filters = new_filters if self._created: diff --git a/gns3server/utils/packet_filter_validation.py b/gns3server/utils/packet_filter_validation.py new file mode 100644 index 000000000..ad19eb316 --- /dev/null +++ b/gns3server/utils/packet_filter_validation.py @@ -0,0 +1,199 @@ +""" +Packet filter parameter validation utilities. +""" + +import logging +import subprocess +from typing import Dict, List, Any, Optional + +log = logging.getLogger(__name__) + + +class FilterValidationError(Exception): + """Raised when packet filter parameters fail validation.""" + pass + + +def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Optional[str]]: + """ + Validate BPF filter expression syntax using tshark. + + 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) + + Args: + bpf_expression: BPF filter expression to validate + + Returns: + 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, + capture_output=True, + text=True, + ) + + # Check if output contains "Invalid" indicating syntax error + if "Invalid" in result.stdout or "Invalid" in result.stderr: + 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" + 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." + ) + return {"valid": True, "error": None} + + except Exception as e: + log.error("Unexpected error during BPF validation: %s", e) + return {"valid": False, "error": f"BPF validation error: {str(e)}"} + + +def validate_filter_parameters(filter_type: str, values: List[Any]) -> None: + """ + Validate packet filter parameters. + + Args: + filter_type: Type of packet filter + values: List of parameter values + + Raises: + FilterValidationError: If parameters are invalid + """ + + # Define validation rules based on ubridge implementation + VALIDATION_RULES = { + "frequency_drop": { + "params_count": 1, + "ranges": [(-1, 32767)], # min, max + "names": ["Frequency"], + "units": ["th packet"] + }, + "packet_loss": { + "params_count": 1, + "ranges": [(0, 100)], + "names": ["Chance"], + "units": ["%"] + }, + "delay": { + "params_count": 2, # latency, jitter + "ranges": [(0, 32767), (0, 32767)], + "names": ["Latency", "Jitter"], + "units": ["ms", "ms"] + }, + "corrupt": { + "params_count": 1, + "ranges": [(0, 100)], + "names": ["Chance"], + "units": ["%"] + }, + "bpf": { + "params_count": 1, + "is_text": True, + "names": ["Filters"] + } + } + + if filter_type not in VALIDATION_RULES: + raise FilterValidationError(f"Unknown filter type: {filter_type}") + + rules = VALIDATION_RULES[filter_type] + + # Check parameter count + if len(values) != rules["params_count"]: + raise FilterValidationError( + f"{filter_type} expects {rules['params_count']} parameter(s), got {len(values)}" + ) + + # Validate each parameter + for i, value in enumerate(values): + if rules.get("is_text"): + # Text validation (BPF) + if not isinstance(value, str): + raise FilterValidationError( + f"{filter_type} parameter {rules['names'][i]} must be a string" + ) + + # Validate BPF syntax using tshark (same method as gns3_copilot) + 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']}" + ) + else: + # Integer parameter validation + try: + if isinstance(value, str): + value = value.strip() + int_value = int(value) + else: + int_value = int(value) + except (ValueError, TypeError): + raise FilterValidationError( + f"{filter_type} parameter {rules['names'][i]} must be an integer, got: {value}" + ) + + # Range validation + min_val, max_val = rules["ranges"][i] + if int_value < min_val or int_value > max_val: + raise FilterValidationError( + f"{filter_type} parameter {rules['names'][i]} must be between " + f"{min_val} and {max_val} {rules['units'][i]}, got: {int_value}" + ) + + +def validate_all_filters(filters: Dict[str, List[Any]]) -> None: + """ + Validate all packet filters. + + Args: + filters: Dictionary mapping filter types to their values + + Raises: + FilterValidationError: If any filter is invalid + """ + + if not filters: + return + + for filter_type, values in filters.items(): + if not values or (isinstance(values, list) and len(values) == 0): + continue + + validate_filter_parameters(filter_type, values) \ No newline at end of file diff --git a/tests/utils/test_packet_filter_validation.py b/tests/utils/test_packet_filter_validation.py new file mode 100644 index 000000000..d248dd868 --- /dev/null +++ b/tests/utils/test_packet_filter_validation.py @@ -0,0 +1,157 @@ +""" +Unit tests for packet filter validation. +""" + +import pytest +from gns3server.utils.packet_filter_validation import ( + validate_filter_parameters, + validate_all_filters, + FilterValidationError +) + + +class TestPacketFilterValidation: + """Test packet filter parameter validation.""" + + def test_frequency_drop_valid(self): + """Test valid frequency drop parameters.""" + # Valid range: -1 to 32767 + validate_filter_parameters("frequency_drop", [-1]) + validate_filter_parameters("frequency_drop", [1]) + validate_filter_parameters("frequency_drop", [100]) + validate_filter_parameters("frequency_drop", [32767]) + + def test_frequency_drop_invalid(self): + """Test invalid frequency drop parameters.""" + # Too low + with pytest.raises(FilterValidationError, match="between -1 and 32767"): + validate_filter_parameters("frequency_drop", [-2]) + + # Too high + with pytest.raises(FilterValidationError, match="between -1 and 32767"): + validate_filter_parameters("frequency_drop", [32768]) + + # Wrong type + with pytest.raises(FilterValidationError, match="must be an integer"): + validate_filter_parameters("frequency_drop", ["invalid"]) + + def test_packet_loss_valid(self): + """Test valid packet loss parameters.""" + # Valid range: 0-100% + validate_filter_parameters("packet_loss", [0]) + validate_filter_parameters("packet_loss", [50]) + validate_filter_parameters("packet_loss", [100]) + + def test_packet_loss_invalid(self): + """Test invalid packet loss parameters.""" + # Negative + with pytest.raises(FilterValidationError, match="between 0 and 100"): + validate_filter_parameters("packet_loss", [-1]) + + # Over 100% + with pytest.raises(FilterValidationError, match="between 0 and 100"): + validate_filter_parameters("packet_loss", [101]) + + def test_delay_valid(self): + """Test valid delay parameters.""" + # Valid range: 0-32767ms + validate_filter_parameters("delay", [0, 0]) + validate_filter_parameters("delay", [100, 50]) + validate_filter_parameters("delay", [32767, 32767]) + + def test_delay_invalid(self): + """Test invalid delay parameters.""" + # Negative latency + with pytest.raises(FilterValidationError, match="between 0 and 32767"): + validate_filter_parameters("delay", [-1, 0]) + + # Over max + with pytest.raises(FilterValidationError, match="between 0 and 32767"): + validate_filter_parameters("delay", [32768, 0]) + + # Negative jitter + with pytest.raises(FilterValidationError, match="between 0 and 32767"): + validate_filter_parameters("delay", [100, -1]) + + def test_corrupt_valid(self): + """Test valid corrupt parameters.""" + # Valid range: 0-100% + validate_filter_parameters("corrupt", [0]) + validate_filter_parameters("corrupt", [50]) + validate_filter_parameters("corrupt", [100]) + + def test_corrupt_invalid(self): + """Test invalid corrupt parameters.""" + # Over 100% + with pytest.raises(FilterValidationError, match="between 0 and 100"): + validate_filter_parameters("corrupt", [101]) + + def test_bpf_valid(self): + """Test valid BPF parameters.""" + validate_filter_parameters("bpf", ["tcp port 80"]) + validate_filter_parameters("bpf", ["tcp and not port 22"]) + validate_filter_parameters("bpf", [""]) # Empty is valid + validate_filter_parameters("bpf", ["host 192.168.1.1 and port 443"]) + + 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: + 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") + + def test_parameter_count_mismatch(self): + """Test wrong number of parameters.""" + # frequency_drop expects 1 parameter + with pytest.raises(FilterValidationError, match="expects 1 parameter"): + validate_filter_parameters("frequency_drop", []) + + with pytest.raises(FilterValidationError, match="expects 1 parameter"): + validate_filter_parameters("frequency_drop", [1, 2]) + + # delay expects 2 parameters + with pytest.raises(FilterValidationError, match="expects 2 parameter"): + validate_filter_parameters("delay", [100]) + + def test_string_to_int_conversion(self): + """Test string to integer conversion.""" + # Should work with string numbers + validate_filter_parameters("frequency_drop", ["10"]) + validate_filter_parameters("packet_loss", ["50"]) + validate_filter_parameters("delay", ["100", "50"]) + + def test_validate_all_filters(self): + """Test validating multiple filters at once.""" + filters = { + "frequency_drop": [10], + "delay": [100, 50] + } + validate_all_filters(filters) # Should not raise + + def test_validate_all_filters_with_invalid(self): + """Test validate_all_filters with invalid filter.""" + filters = { + "frequency_drop": [10], + "packet_loss": [150] # Invalid: over 100% + } + with pytest.raises(FilterValidationError): + validate_all_filters(filters) + + def test_unknown_filter_type(self): + """Test unknown filter type.""" + with pytest.raises(FilterValidationError, match="Unknown filter type"): + validate_filter_parameters("unknown_filter", [1]) \ No newline at end of file