agent firewall scenario

tests_e2e/orchestrator/runbook.yml

@@ -51,7 +51,7 @@ variable:
   #
   # The test suites to execute
   - name: test_suites
-    value: "agent_bvt, no_outbound_connections, extensions_disabled, agent_not_provisioned, fips, agent_ext_workflow, agent_update, agent_status, multi_config_ext, agent_cgroups, ext_cgroups"
+    value: "agent_bvt, no_outbound_connections, extensions_disabled, agent_not_provisioned, fips, agent_ext_workflow, agent_update, agent_status, multi_config_ext, agent_cgroups, ext_cgroups, agent_firewall"
   - name: cloud
     value: "AzureCloud"
     is_case_visible: true

tests_e2e/test_suites/agent_firewall.yml

@@ -0,0 +1,15 @@
+#
+# This test verifies that the agent firewall rules are set correctly. The expected firewall rules are:
+#        0        0 ACCEPT     tcp  --  *      *       0.0.0.0/0            168.63.129.16        tcp dpt:53
+#       0        0 ACCEPT     tcp  --  *      *       0.0.0.0/0            168.63.129.16        owner UID match 0
+#       0        0 DROP       tcp  --  *      *       0.0.0.0/0            168.63.129.16        ctstate INVALID,NEW
+# The first rule allows tcp traffic to port 53 for non root user. The second rule allows traffic to wireserver for root user.
+# The third rule drops all other traffic to wireserver.
+#
+name: "AgentFirewall"
+tests:
+  - "agent_firewall/agent_firewall.py"
+images:
+  - "endorsed"
+  - "endorsed-arm64"
+owns_vm: true  # This vm cannot be shared with other tests because it modifies the firewall rules and agent status.

tests_e2e/tests/agent_firewall/agent_firewall.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+
+# Microsoft Azure Linux Agent
+#
+# Copyright 2018 Microsoft Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from tests_e2e.tests.lib.agent_test import AgentTest
+from tests_e2e.tests.lib.agent_test_context import AgentTestContext
+from tests_e2e.tests.lib.logging import log
+
+
+class AgentFirewall(AgentTest):
+    """
+    This test verifies the agent firewall rules are added properly. It checks each firewall rule is present and working as expected.
+    """
+
+    def __init__(self, context: AgentTestContext):
+        super().__init__(context)
+        self._ssh_client = self._context.create_ssh_client()
+
+    def run(self):
+        log.info("Checking iptable rules added by the agent")
+        self._run_remote_test(f"agent_firewall-verify_all_firewall_rules.py --user {self._context.username}", use_sudo=True)
+        log.info("Successfully verified all rules present and working as expected.")
+
+
+if __name__ == "__main__":
+    AgentFirewall.run_from_command_line()
+
+

tests_e2e/tests/lib/add_network_security_group.py

@@ -16,10 +16,10 @@
 #
 
 import json
-import http.client
 
 from typing import Any, Dict, List
 
+from azurelinuxagent.common.utils import shellutil
 from tests_e2e.tests.lib.logging import log
 from tests_e2e.tests.lib.retry import retry
 from tests_e2e.tests.lib.update_arm_template import UpdateArmTemplate
@@ -140,10 +140,10 @@ def _my_ip_address(self) -> str:
         """
         if self.__my_ip_address is None:
             def get_my_address():
-                connection = http.client.HTTPSConnection("ifconfig.io")
-                connection.request("GET", "/ip")
-                response = connection.getresponse()
-                return response.read().decode().strip()
+                # Forcing -4 option to fetch the ipv4 address
+                cmd = ["curl", "-4", "ifconfig.io/ip"]
+                stdout = shellutil.run_command(cmd)
+                return stdout.strip()
             self.__my_ip_address = retry(get_my_address, attempts=3, delay=10)
         return self.__my_ip_address
 

tests_e2e/tests/lib/retry.py

@@ -72,9 +72,9 @@ def retry_if_false(operation: Callable[[], bool], attempts: int = 5, delay: int
             log.warning("Error in operation: %s", e)
             if attempts == 0:
                 raise
-        if not success:
+        if not success and attempts != 0:
             log.info("Current operation failed, retrying in %s secs.", delay)
-        time.sleep(delay)
+            time.sleep(delay)
     return success