services/firewall
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix issue where you couldn't update ports
Build and Push Container / build-and-push (push) Successful in 26s
Details

Browse Source
This commit is contained in:
ducoterra
2026-08-01 19:56:39 -04:00
parent e360546de9
commit 11389edf72
4 changed files with 587 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_main.py
+286 -1
View File
@@ -9,6 +9,7 @@ from main import (
RuleConfig,
build_zone_map,
load_config,
policy_matches_config,
process_rule,
send_ntfy_notification,
)
@@ -128,7 +129,7 @@ class TestProcessRule:
assert change["rule_name"] == "Test Rule"
assert change["error"] is None
def test_skips_rule_with_matching_ip(self) -> None:
def test_skips_rule_with_matching_config(self) -> None:
mock_session = MagicMock()
existing_policy = {
"id": "pol-1",
@@ -143,6 +144,14 @@ class TestProcessRule:
},
},
},
"destination": {"zoneId": "lan-id", "trafficFilter": None},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {"ipVersion": "IPV4", "protocolFilter": None},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
mock_session.get.return_value = MagicMock(json=MagicMock(return_value={"data": [existing_policy]}), raise_for_status=MagicMock())
@@ -224,3 +233,279 @@ class TestProcessRule:
assert change["action"] == "failed"
assert "Failed to list policies" in (change["error"] or "")
def test_updates_rule_with_different_ports_same_ip(self) -> None:
mock_session = MagicMock()
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {
"zoneId": "lan-id",
"trafficFilter": {
"type": "PORT",
"portFilter": {
"type": "PORTS",
"items": [
{"type": "PORT_NUMBER", "value": 80},
{"type": "PORT_NUMBER", "value": 443},
],
},
},
},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {"ipVersion": "IPV4", "protocolFilter": None},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
mock_session.get.return_value = MagicMock(json=MagicMock(return_value={"data": [existing_policy]}), raise_for_status=MagicMock())
mock_session.put.return_value = MagicMock(json=MagicMock(return_value={"id": "pol-1"}), raise_for_status=MagicMock())
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "ALLOW",
"dest_ports": [80, 443, 8080],
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
change = process_rule(mock_session, "https://unifi.local", "site-1", rule, zone_map, "1.2.3.4")
assert change["action"] == "updated"
mock_session.put.assert_called_once()
class TestPolicyMatchesConfig:
def test_returns_true_when_all_match(self) -> None:
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {
"zoneId": "lan-id",
"trafficFilter": {
"type": "PORT",
"portFilter": {
"type": "PORTS",
"items": [
{"type": "PORT_NUMBER", "value": 80},
{"type": "PORT_NUMBER", "value": 443},
],
},
},
},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {
"ipVersion": "IPV4",
"protocolFilter": {
"type": "NAMED_PROTOCOL",
"protocol": {"name": "TCP"},
},
},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": "Test description",
}
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "ALLOW",
"allow_return_traffic": True,
"protocol": "tcp",
"dest_ports": [443, 80],
"logging_enabled": False,
"enabled": True,
"description": "Test description",
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
result = policy_matches_config(existing_policy, rule, zone_map, "1.2.3.4") # type: ignore[arg-type]
assert result is True
def test_returns_false_different_ports(self) -> None:
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {
"zoneId": "lan-id",
"trafficFilter": {
"type": "PORT",
"portFilter": {
"type": "PORTS",
"items": [
{"type": "PORT_NUMBER", "value": 80},
{"type": "PORT_NUMBER", "value": 443},
],
},
},
},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {"ipVersion": "IPV4", "protocolFilter": None},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "ALLOW",
"dest_ports": [80, 443, 8080],
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
result = policy_matches_config(existing_policy, rule, zone_map, "1.2.3.4") # type: ignore[arg-type]
assert result is False
def test_returns_false_different_protocol(self) -> None:
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {"zoneId": "lan-id", "trafficFilter": None},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {
"ipVersion": "IPV4",
"protocolFilter": {
"type": "NAMED_PROTOCOL",
"protocol": {"name": "TCP"},
},
},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "ALLOW",
"protocol": "UDP",
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
result = policy_matches_config(existing_policy, rule, zone_map, "1.2.3.4") # type: ignore[arg-type]
assert result is False
def test_returns_false_different_action(self) -> None:
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {"zoneId": "lan-id", "trafficFilter": None},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {"ipVersion": "IPV4", "protocolFilter": None},
"enabled": True,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "BLOCK",
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
result = policy_matches_config(existing_policy, rule, zone_map, "1.2.3.4") # type: ignore[arg-type]
assert result is False
def test_returns_false_different_enabled(self) -> None:
existing_policy = {
"id": "pol-1",
"name": "Test Rule",
"source": {
"zoneId": "wan-id",
"trafficFilter": {
"type": "IP_ADDRESS",
"ipAddressFilter": {
"type": "IP_ADDRESSES",
"items": [{"type": "IP_ADDRESS", "value": "1.2.3.4"}],
},
},
},
"destination": {"zoneId": "lan-id", "trafficFilter": None},
"action": {"type": "ALLOW", "allowReturnTraffic": True},
"ipProtocolScope": {"ipVersion": "IPV4", "protocolFilter": None},
"enabled": False,
"loggingEnabled": False,
"index": 0,
"metadata": {},
"description": None,
}
rule: RuleConfig = {
"name": "Test Rule",
"source_zone": "WAN",
"dest_zone": "LAN",
"ip_version": "IPV4",
"action": "ALLOW",
"enabled": True,
}
zone_map = {"WAN": "wan-id", "LAN": "lan-id"}
result = policy_matches_config(existing_policy, rule, zone_map, "1.2.3.4") # type: ignore[arg-type]
assert result is False