Fix delay filter validation: ensure delay: [0, X] returns proper error message

This fix addresses the issue where delay: [0, X] configurations were being
silently dropped instead of returning validation errors.

Changes:
- Created new utility function filter_inactive_filters() in packet_filter_validation.py
- Implemented smart filtering logic for delay filter that checks both latency and jitter:
  * delay: [0, 0] → User wants to disable delay, filter out silently
  * delay: [0, X] where X > 0 → Invalid config, keep for validation error
  * delay: [X, X] where X > 0 → Normal configuration, validate normally
- Simplified link.py update_filters() method to use the new utility function
- Added comprehensive tests for the new filtering logic

Before this fix:
- delay: [0, 100] would be silently dropped with no error message
- Users wouldn't know their configuration was invalid

After this fix:
- delay: [0, 100] returns proper error: "delay parameter Latency must be between 1 and 32767 ms, got: 0"
- delay: [0, 0] is correctly handled as intentional disable
- Normal delay configurations continue to work as expected
This commit is contained in:
YueGuobin 2026-05-31 22:35:58 +08:00
parent 92a0fa6cd7
commit ced73574b4
No known key found for this signature in database
3 changed files with 171 additions and 14 deletions

View File

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

View File

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

View File

@ -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])
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 == {}