diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index 76011b9f2..556f5ba5f 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -23,7 +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 +from gns3server.utils.packet_filter_validation import validate_all_filters, filter_inactive_filters, FilterValidationError import logging @@ -148,19 +148,11 @@ class Link: """ Modify the filters list. - Filter with value 0 will be dropped because not active + Filters with value 0 will be filtered out as inactive, with special + handling for delay filter to distinguish between "disabled" and "invalid config". """ - new_filters = {} - for (filter, values) in filters.items(): - new_values = [] - for value in values: - if isinstance(value, str): - new_values.append(value.strip("\n ")) - else: - new_values.append(int(value)) - values = new_values - if len(values) != 0 and values[0] != 0 and values[0] != "": - new_filters[filter] = values + # Filter out inactive filters using the utility function + new_filters = filter_inactive_filters(filters) # Validate filter parameters before applying try: diff --git a/gns3server/utils/packet_filter_validation.py b/gns3server/utils/packet_filter_validation.py index 0bd36fc5e..2ba67f6f5 100644 --- a/gns3server/utils/packet_filter_validation.py +++ b/gns3server/utils/packet_filter_validation.py @@ -170,6 +170,69 @@ def validate_filter_parameters(filter_type: str, values: List[Any]) -> None: ) +def filter_inactive_filters(filters: Dict[str, List[Any]]) -> Dict[str, List[Any]]: + """ + Filter out inactive packet filters before validation. + + This function implements smart filtering logic: + - For most filters: value 0 means "disabled" and will be filtered out + - For delay filter: check both latency and jitter to determine intent + * delay: [0, 0] → User wants to disable delay completely, filter it out + * delay: [0, X] where X > 0 → Invalid config (latency must be >= 1), keep for validation error + * delay: [X, X] where X > 0 → Normal configuration, keep for validation + + Args: + filters: Dictionary mapping filter types to their values + + Returns: + Filtered dictionary with only active filters for validation + """ + + if not filters: + return {} + + active_filters = {} + for filter_type, values in filters.items(): + if not values or (isinstance(values, list) and len(values) == 0): + continue + + # Normalize values (strip strings, convert to int) + normalized_values = [] + for value in values: + if isinstance(value, str): + normalized_values.append(value.strip("\n ")) + else: + normalized_values.append(int(value)) + values = normalized_values + + # Skip empty filters after normalization + if len(values) == 0: + continue + + # Special handling for delay filter - check both latency and jitter + if filter_type == "delay": + if len(values) >= 1 and values[0] == 0: # latency = 0 + if len(values) >= 2 and values[1] == 0: # jitter = 0 too + # User intentionally disabling delay completely: [0, 0] + log.debug(f"Filter {filter_type} with values {values} skipped (disabled)") + continue # Skip this filter silently + else: + # Invalid config: latency=0 but jitter>0, keep for validation error + log.debug(f"Filter {filter_type} with values {values} kept for validation (invalid config)") + active_filters[filter_type] = values + else: + # latency>0, normal configuration + active_filters[filter_type] = values + # For other filters, skip if first value is 0 or empty string (means "disabled") + elif values[0] != 0 and values[0] != "": + active_filters[filter_type] = values + else: + # Filters like packet_loss=0, corrupt=0, frequency_drop=0 are intentionally disabled + log.debug(f"Filter {filter_type} with values {values} skipped (disabled)") + + return active_filters + + def validate_all_filters(filters: Dict[str, List[Any]]) -> None: """ Validate all packet filters. diff --git a/tests/utils/test_packet_filter_validation.py b/tests/utils/test_packet_filter_validation.py index 3a9de41bd..d15ae1fc3 100644 --- a/tests/utils/test_packet_filter_validation.py +++ b/tests/utils/test_packet_filter_validation.py @@ -6,6 +6,7 @@ import pytest from gns3server.utils.packet_filter_validation import ( validate_filter_parameters, validate_all_filters, + filter_inactive_filters, FilterValidationError ) @@ -160,4 +161,105 @@ class TestPacketFilterValidation: 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 + validate_filter_parameters("unknown_filter", [1]) + + +class TestFilterInactiveFilters: + """Test filter_inactive_filters function for smart filter filtering logic.""" + + def test_filter_inactive_delay_disabled(self): + """Test delay [0, 0] is filtered out (user wants to disable delay).""" + filters = {"delay": [0, 0]} + result = filter_inactive_filters(filters) + assert result == {} # Should be filtered out + + def test_filter_inactive_delay_invalid_config(self): + """Test delay [0, 100] is kept for validation (invalid config).""" + filters = {"delay": [0, 100]} + result = filter_inactive_filters(filters) + assert result == {"delay": [0, 100]} # Should be kept for validation error + + def test_filter_inactive_delay_normal_config(self): + """Test delay [100, 20] is kept (normal configuration).""" + filters = {"delay": [100, 20]} + result = filter_inactive_filters(filters) + assert result == {"delay": [100, 20]} # Should be kept + + def test_filter_inactive_delay_zero_jitter(self): + """Test delay [100, 0] is kept (normal config with zero jitter).""" + filters = {"delay": [100, 0]} + result = filter_inactive_filters(filters) + assert result == {"delay": [100, 0]} # Should be kept + + def test_filter_inactive_packet_loss_zero(self): + """Test packet_loss [0] is filtered out (disabled).""" + filters = {"packet_loss": [0]} + result = filter_inactive_filters(filters) + assert result == {} # Should be filtered out + + def test_filter_inactive_packet_loss_active(self): + """Test packet_loss [5] is kept (active).""" + filters = {"packet_loss": [5]} + result = filter_inactive_filters(filters) + assert result == {"packet_loss": [5]} # Should be kept + + def test_filter_inactive_corrupt_zero(self): + """Test corrupt [0] is filtered out (disabled).""" + filters = {"corrupt": [0]} + result = filter_inactive_filters(filters) + assert result == {} # Should be filtered out + + def test_filter_inactive_corrupt_active(self): + """Test corrupt [2] is kept (active).""" + filters = {"corrupt": [2]} + result = filter_inactive_filters(filters) + assert result == {"corrupt": [2]} # Should be kept + + def test_filter_inactive_frequency_drop_zero(self): + """Test frequency_drop [0] is filtered out (disabled).""" + filters = {"frequency_drop": [0]} + result = filter_inactive_filters(filters) + assert result == {} # Should be filtered out + + def test_filter_inactive_frequency_drop_active(self): + """Test frequency_drop [10] is kept (active).""" + filters = {"frequency_drop": [10]} + result = filter_inactive_filters(filters) + assert result == {"frequency_drop": [10]} # Should be kept + + def test_filter_inactive_bpf_empty(self): + """Test BPF empty string is filtered out.""" + filters = {"bpf": [""]} + result = filter_inactive_filters(filters) + assert result == {} # Should be filtered out + + def test_filter_inactive_bpf_active(self): + """Test BPF with expression is kept.""" + filters = {"bpf": ["tcp port 80"]} + result = filter_inactive_filters(filters) + assert result == {"bpf": ["tcp port 80"]} # Should be kept + + def test_filter_inactive_multiple_filters_mixed(self): + """Test multiple filters with mixed active/inactive states.""" + filters = { + "delay": [0, 0], # Disabled: [0, 0] + "packet_loss": [0], # Disabled: 0% + "corrupt": [2], # Active: 2% + "frequency_drop": [10] # Active: every 10th packet + } + result = filter_inactive_filters(filters) + assert result == { + "corrupt": [2], + "frequency_drop": [10] + } + + def test_filter_inactive_empty_filters(self): + """Test empty filters dictionary.""" + filters = {} + result = filter_inactive_filters(filters) + assert result == {} + + def test_filter_inactive_none_filters(self): + """Test None filters.""" + result = filter_inactive_filters(None) + assert result == {} \ No newline at end of file