Revert "Revert "Update all other code to remove IPList class""

This reverts commit e2603aa.

Author: bitterpanda63

aikido_zen/background_process/commands/sync_data_test.py

@@ -4,8 +4,8 @@
 from unittest.mock import MagicMock
 from .sync_data import process_sync_data
 from aikido_zen.background_process.routes import Routes
-from aikido_zen.helpers.iplist import IPList
 from ..packages import PackagesStore
+from ...helpers.ip_matcher import IPMatcher
 from ...storage.hostnames import Hostnames
 from ...storage.statistics import Statistics
 
@@ -18,8 +18,7 @@ def setup_connection_manager():
     connection_manager.hostnames = Hostnames()
     connection_manager.conf.endpoints = ["endpoint1", "endpoint2"]
 
-    connection_manager.conf.bypassed_ips = IPList()
-    connection_manager.conf.bypassed_ips.add("192.168.1.1")
+    connection_manager.conf.bypassed_ips = IPMatcher(["192.168.1.1"])
     connection_manager.conf.blocked_uids = ["user1", "user2"]
     connection_manager.conf.last_updated_at = 200
     connection_manager.statistics = Statistics()

aikido_zen/background_process/service_config.py

@@ -5,8 +5,9 @@
 from typing import Pattern
 
 import regex as re
+
+from aikido_zen.helpers.ip_matcher import IPMatcher
 from aikido_zen.helpers.match_endpoints import match_endpoints
-from aikido_zen.helpers.iplist import IPList
 
 
 # noinspection PyAttributeOutsideInit
@@ -50,7 +51,7 @@ def set_endpoints(self, endpoints):
             endpoint for endpoint in endpoints if not endpoint.get("graphql")
         ]
 
-        # Create a IPList instance for each endpoint
+        # Create an IPMatcher instance for each endpoint
         for endpoint in self.endpoints:
             if not "allowedIPAddresses" in endpoint:
                 #  This feature is not supported by the current aikido server version
@@ -62,7 +63,7 @@ def set_endpoints(self, endpoints):
                 #  Skip empty allowlist
                 continue
 
-            endpoint["allowedIPAddresses"] = IPList(endpoint["allowedIPAddresses"])
+            endpoint["allowedIPAddresses"] = IPMatcher(endpoint["allowedIPAddresses"])
 
     def get_endpoints(self, route_metadata):
         """
@@ -72,14 +73,14 @@ def get_endpoints(self, route_metadata):
         return match_endpoints(route_metadata, self.endpoints)
 
     def set_bypassed_ips(self, bypassed_ips):
-        """Creates an IPList from the given bypassed ip set"""
-        self.bypassed_ips = IPList()
+        """Creates an IPMatcher from the given bypassed ip set"""
+        self.bypassed_ips = IPMatcher()
         for ip in bypassed_ips:
             self.bypassed_ips.add(ip)
 
     def is_bypassed_ip(self, ip):
         """Checks if the IP is on the bypass list"""
-        return self.bypassed_ips.matches(ip)
+        return self.bypassed_ips.has(ip)
 
     def set_blocked_ips(self, blocked_ip_entries):
         self.blocked_ips = list(map(parse_ip_entry, blocked_ip_entries))
@@ -89,7 +90,7 @@ def set_allowed_ips(self, allowed_ip_entries):
 
     def is_blocked_ip(self, ip):
         for entry in self.blocked_ips:
-            if entry["iplist"].matches(ip):
+            if entry["iplist"].has(ip):
                 return entry["description"]
         return False
 
@@ -111,12 +112,8 @@ def parse_ip_entry(entry):
     """
     Converts ip entry: {"source": "example", "description": "Example description", "ips": []}
     """
-    iplist = IPList()
-    for ip in entry["ips"]:
-        iplist.add(ip)
-
     return {
         "source": entry["source"],
         "description": entry["description"],
-        "iplist": iplist,
+        "iplist": IPMatcher(entry["ips"]),
     }

aikido_zen/background_process/service_config_test.py

@@ -1,6 +1,6 @@
 import pytest
 from .service_config import ServiceConfig
-from aikido_zen.helpers.iplist import IPList
+from aikido_zen.helpers.ip_matcher import IPMatcher
 
 
 def test_service_config_initialization():
@@ -66,10 +66,10 @@ def test_service_config_initialization():
     assert service_config.endpoints[1]["route"] == "/v3"
     assert service_config.endpoints[2]["route"] == "/admin"
     assert service_config.last_updated_at == last_updated_at
-    assert isinstance(service_config.bypassed_ips, IPList)
-    assert service_config.bypassed_ips.matches("127.0.0.1")
-    assert service_config.bypassed_ips.matches("123.1.2.2")
-    assert not service_config.bypassed_ips.matches("1.1.1.1")
+    assert isinstance(service_config.bypassed_ips, IPMatcher)
+    assert service_config.bypassed_ips.has("127.0.0.1")
+    assert service_config.bypassed_ips.has("123.1.2.2")
+    assert not service_config.bypassed_ips.has("1.1.1.1")
     assert service_config.blocked_uids == set(["1", "0", "5"])
 
     v1_endpoint = service_config.get_endpoints(
@@ -90,10 +90,10 @@ def test_service_config_initialization():
         }
     )[0]
     assert admin_endpoint["route"] == "/admin"
-    assert isinstance(admin_endpoint["allowedIPAddresses"], IPList)
-    assert admin_endpoint["allowedIPAddresses"].matches("192.168.2.1")
-    assert admin_endpoint["allowedIPAddresses"].matches("1.2.3.4")
-    assert not admin_endpoint["allowedIPAddresses"].matches("192.168.0.1")
+    assert isinstance(admin_endpoint["allowedIPAddresses"], IPMatcher)
+    assert admin_endpoint["allowedIPAddresses"].has("192.168.2.1")
+    assert admin_endpoint["allowedIPAddresses"].has("1.2.3.4")
+    assert not admin_endpoint["allowedIPAddresses"].has("192.168.0.1")
 
 
 # Sample data for testing
@@ -118,7 +118,7 @@ def service_config():
 def test_initialization(service_config):
     assert len(service_config.endpoints) == 2  # Only non-graphql endpoints
     assert service_config.last_updated_at == "2023-10-01T00:00:00Z"
-    assert isinstance(service_config.bypassed_ips, IPList)
+    assert isinstance(service_config.bypassed_ips, IPMatcher)
     assert service_config.blocked_uids == {"user1", "user2"}
 
 
@@ -194,10 +194,10 @@ def test_service_config_with_empty_allowlist():
     assert len(service_config.endpoints) == 1
     assert service_config.endpoints[0]["route"] == "/admin"
     assert service_config.last_updated_at == last_updated_at
-    assert isinstance(service_config.bypassed_ips, IPList)
-    assert service_config.bypassed_ips.matches("127.0.0.1")
-    assert service_config.bypassed_ips.matches("123.1.2.2")
-    assert not service_config.bypassed_ips.matches("1.1.1.1")
+    assert isinstance(service_config.bypassed_ips, IPMatcher)
+    assert service_config.bypassed_ips.has("127.0.0.1")
+    assert service_config.bypassed_ips.has("123.1.2.2")
+    assert not service_config.bypassed_ips.has("1.1.1.1")
     assert service_config.blocked_uids == set(["1", "0", "5"])
 
     admin_endpoint = service_config.get_endpoints(

aikido_zen/helpers/iplist.py

@@ -13,7 +13,7 @@ def is_ip_allowed_by_allowlist(service_config, ip):
         return True
 
     for entry in service_config.allowed_ips:
-        if entry["iplist"].matches(ip):
+        if entry["iplist"].has(ip):
             # If the IP matches one of the lists the IP is allowed :
             return True
     return False