new_module proxmox_firewall

meta/runtime.yml

@@ -19,6 +19,8 @@ action_groups:
     - proxmox_cluster_join_info
     - proxmox_disk
     - proxmox_domain_info
+    - proxmox_firewall
+    - proxmox_firewall_info
     - proxmox_group
     - proxmox_group_info
     - proxmox_kvm

plugins/module_utils/proxmox.py

@@ -73,6 +73,53 @@ def ansible_to_proxmox_bool(value):
     return 1 if value else 0
 
 
+def compare_list_of_dicts(existing_list, new_list, uid, params_to_ignore=None):
+    """ Compare 2 list of dicts
+    Use case - for firewall rules we will be getting a list of rules from user.
+    We want to filter out which rules needs to be updated and which rules are completely new and needs to be created
+
+    :param existing_list: Existing values example - list of existing rules
+    :param new_list: New values example - list of rules passed to module
+    :param uid: unique identifier in dict. It should always be present in both lists - in case of firewall rules it's pos
+    :param params_to_ignore:  list of params we want to ignore which are present in existing_list's dict.
+                            In case of firewall rules we want to ignore ['digest', 'ipversion']
+
+    :return: returns 2 list items 1st is the list of items which are completely new and needs to be created
+            2nd is a list of items which needs to be updated
+    """
+    if params_to_ignore is None:
+        params_to_ignore = list()
+    items_to_update = []
+    new_list = [{k: v for k, v in item.items() if v is not None and k not in params_to_ignore} for item in new_list]
+
+    if existing_list is None:
+        items_to_create = new_list
+        items_to_update = list()
+        return items_to_create, items_to_update
+
+    existing_list = {x[uid]: x for x in existing_list}
+    new_list = {x[uid]: x for x in new_list}
+
+    common_uids = set(existing_list.keys()).intersection(set(new_list.keys()))
+    missing_uids = set(new_list.keys()) - set(existing_list.keys())
+    items_to_create = [new_list[uid] for uid in missing_uids]
+
+    for uid in common_uids:
+        # If new rule has a parameter that is not present in existing rule we need to update
+        if set(new_list[uid].keys()) - set(existing_list[uid].keys()) != set():
+            items_to_update.append(new_list[uid])
+            continue
+
+        # If existing rule param value doesn't match new rule param OR
+        # If existing rule has a param that is not present in new rule except for params in params_to_ignore
+        for existing_rule_param, existing_parm_value in existing_list[uid].items():
+            if (existing_rule_param not in params_to_ignore and
+                    new_list[uid].get(existing_rule_param) != existing_parm_value):
+                items_to_update.append(new_list[uid])
+
+    return items_to_create, items_to_update
+
+
 class ProxmoxAnsible(object):
     """Base class for Proxmox modules"""
     TASK_TIMED_OUT = 'timeout expired'

plugins/module_utils/proxmox_sdn.py

@@ -99,3 +99,47 @@ def get_zones(self, zone_type: str = None) -> List[Dict]:
             self.module.fail_json(
                 msg=f'Failed to retrieve zone information from cluster: {e}'
             )
+
+    def get_aliases(self, firewall_obj):
+        """Get aliases for IP/CIDR at given firewall endpoint level
+
+        :param firewall_obj: Firewall endpoint as a ProxmoxResource e.g. self.proxmox_api.cluster().firewall
+                            If it is None it'll return an empty list
+        :return: List of aliases and corresponding IP/CIDR
+        """
+        if firewall_obj is None:
+            return list()
+        try:
+            return firewall_obj().aliases().get()
+        except Exception as e:
+            self.module.fail_json(
+                msg=f'Failed to retrieve aliases - {e}'
+            )
+
+    def get_fw_rules(self, rules_obj, pos=None):
+        """Get firewall rules at given rules endpoint level
+
+        :param rules_obj: Firewall Rules endpoint as a ProxmoxResource e.g. self.proxmox_api.cluster().firewall().rules
+        :param pos: Rule position if it is None it'll return all rules
+        :return: Firewall rules as a list of dict
+        """
+        if pos is not None:
+            pos = str(pos)
+        try:
+            return rules_obj(pos).get()
+        except Exception as e:
+            self.module.fail_json(
+                msg=f'Failed to retrieve firewall rules: {e}'
+            )
+
+    def get_groups(self):
+        """Get firewall security groups
+
+        :return: list of groups
+        """
+        try:
+            return [x['group'] for x in self.proxmox_api.cluster().firewall().groups().get()]
+        except Exception as e:
+            self.module.fail_json(
+                msg=f'Failed to retrieve firewall security groups: {e}'
+            )