Linter: enable attribute-defined-outside-init rule

.pylintrc

@@ -29,5 +29,4 @@ disable =
     no-value-for-parameter,
     unexpected-keyword-arg,
     inconsistent-return-statements,
-    duplicate-code,
-    attribute-defined-outside-init,
+    duplicate-code

aikido_zen/background_process/cloud_connection_manager/__init__.py

@@ -35,13 +35,7 @@ def __init__(self, block, api, token, serverless):
         self.token = token  # Should be instance of the Token class!
         self.routes = Routes(200)
         self.hostnames = Hostnames(200)
-        self.conf = ServiceConfig(
-            endpoints=[],
-            last_updated_at=-1,  # Has not been updated yet
-            blocked_uids=[],
-            bypassed_ips=[],
-            received_any_stats=True,
-        )
+        self.conf = ServiceConfig()
         self.firewall_lists = FirewallLists()
         self.rate_limiter = RateLimiter(
             max_items=5000, time_to_live_in_ms=120 * 60 * 1000  # 120 minutes

aikido_zen/background_process/cloud_connection_manager/update_service_config.py

@@ -15,13 +15,14 @@ def update_service_config(connection_manager, res):
         logger.debug("Updating blocking, setting blocking to : %s", res["block"])
         connection_manager.block = bool(res["block"])
 
-    connection_manager.conf.update(
-        endpoints=res.get("endpoints", []),
-        last_updated_at=res.get("configUpdatedAt", get_unixtime_ms()),
-        blocked_uids=res.get("blockedUserIds", []),
-        bypassed_ips=res.get("allowedIPAddresses", []),
-        received_any_stats=res.get("receivedAnyStats", True),
+    connection_manager.conf.set_endpoints(res.get("endpoints", []))
+    connection_manager.conf.set_last_updated_at(
+        res.get("configUpdatedAt", get_unixtime_ms())
     )
+    connection_manager.conf.set_blocked_user_ids(res.get("blockedUserIds", []))
+    connection_manager.conf.set_bypassed_ips(res.get("allowedIPAddresses", []))
+    if res.get("receivedAnyStats", True):
+        connection_manager.conf.enable_received_any_stats()
 
     # Handle outbound request blocking configuration
     if "blockNewOutgoingRequests" in res:

aikido_zen/background_process/cloud_connection_manager/update_service_config_test.py

@@ -11,13 +11,7 @@ def test_update_service_config_outbound_blocking():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Test response with blockNewOutgoingRequests
@@ -45,13 +39,7 @@ def test_update_service_config_outbound_blocking_false():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = True
 
     # Test response with blockNewOutgoingRequests=False
@@ -69,13 +57,7 @@ def test_update_service_config_outbound_blocking_missing():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Test response without outbound blocking fields
@@ -97,13 +79,7 @@ def test_update_service_config_failure():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Set initial values
@@ -127,13 +103,7 @@ def test_update_service_config_complete():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Test complete response
@@ -174,13 +144,7 @@ def test_update_service_config_domains_only():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Test response with only domains
@@ -207,13 +171,7 @@ def test_update_service_config_block_new_outgoing_requests_only():
 
     # Create a mock connection manager with a real ServiceConfig
     connection_manager = MagicMock()
-    connection_manager.conf = ServiceConfig(
-        endpoints=[],
-        last_updated_at=0,
-        blocked_uids=set(),
-        bypassed_ips=[],
-        received_any_stats=False,
-    )
+    connection_manager.conf = ServiceConfig()
     connection_manager.block = False
 
     # Set initial domains

aikido_zen/background_process/service_config.py

@@ -6,42 +6,18 @@
 from aikido_zen.helpers.match_endpoints import match_endpoints
 
 
-# noinspection PyAttributeOutsideInit
 class ServiceConfig:
-    """Class holding the config of the connection_manager"""
-
-    def __init__(
-        self,
-        endpoints,
-        last_updated_at: int,
-        blocked_uids,
-        bypassed_ips,
-        received_any_stats: bool,
-    ):
-        # Init the class using update function :
-        self.update(
-            endpoints, last_updated_at, blocked_uids, bypassed_ips, received_any_stats
-        )
+
+    def __init__(self):
+        self.endpoints = []
+        self.bypassed_ips = IPMatcher()
+        self.blocked_uids = set()
+        self.last_updated_at = -1
+        self.received_any_stats = False
         self.block_new_outgoing_requests = False
         self.outbound_domains = {}
 
-    def update(
-        self,
-        endpoints,
-        last_updated_at: int,
-        blocked_uids,
-        bypassed_ips,
-        received_any_stats: bool,
-    ):
-        self.last_updated_at = last_updated_at
-        self.received_any_stats = bool(received_any_stats)
-        self.blocked_uids = set(blocked_uids)
-        self.set_endpoints(endpoints)
-        self.set_bypassed_ips(bypassed_ips)
-
     def set_endpoints(self, endpoints):
-        """Sets non-graphql endpoints"""
-
         self.endpoints = [
             endpoint for endpoint in endpoints if not endpoint.get("graphql")
         ]
@@ -68,7 +44,6 @@ def get_endpoints(self, route_metadata):
         return match_endpoints(route_metadata, self.endpoints)
 
     def set_bypassed_ips(self, bypassed_ips):
-        """Creates an IPMatcher from the given bypassed ip set"""
         self.bypassed_ips = IPMatcher()
         for ip in bypassed_ips:
             self.bypassed_ips.add(ip)
@@ -77,6 +52,15 @@ def is_bypassed_ip(self, ip):
         """Checks if the IP is on the bypass list"""
         return self.bypassed_ips.has(ip)
 
+    def set_blocked_user_ids(self, blocked_user_ids):
+        self.blocked_uids = set(blocked_user_ids)
+
+    def enable_received_any_stats(self):
+        self.received_any_stats = True
+
+    def set_last_updated_at(self, last_updated_at: int):
+        self.last_updated_at = last_updated_at
+
     def update_outbound_domains(self, domains):
         self.outbound_domains = {
             domain["hostname"]: domain["mode"] for domain in domains