diff --git a/changelogs/fragments/firewall_rule_cli_change_version_1.4.yaml b/changelogs/fragments/firewall_rule_cli_change_version_1.4.yaml new file mode 100644 index 0000000..652c39d --- /dev/null +++ b/changelogs/fragments/firewall_rule_cli_change_version_1.4.yaml @@ -0,0 +1,3 @@ +--- +minor_changes: + - firewall_rules - icmpv6 type - add support for vyos sw >= 1.4. diff --git a/plugins/module_utils/network/vyos/config/firewall_rules/firewall_rules.py b/plugins/module_utils/network/vyos/config/firewall_rules/firewall_rules.py index 2920878..ca47652 100644 --- a/plugins/module_utils/network/vyos/config/firewall_rules/firewall_rules.py +++ b/plugins/module_utils/network/vyos/config/firewall_rules/firewall_rules.py @@ -1,1042 +1,1068 @@ # # -*- coding: utf-8 -*- # Copyright 2019 Red Hat # GNU General Public License v3.0+ # (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) """ The vyos_firewall_rules class It is in this file where the current configuration (as dict) is compared to the provided configuration (as dict) and the command set necessary to bring the current configuration to it's desired end-state is created """ from __future__ import absolute_import, division, print_function __metaclass__ = type from copy import deepcopy from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base import ( ConfigBase, ) from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.utils import ( to_list, remove_empties, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.facts import ( Facts, ) from ansible.module_utils.six import iteritems from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.utils.utils import ( list_diff_want_only, ) +import re class Firewall_rules(ConfigBase): """ The vyos_firewall_rules class """ gather_subset = [ "!all", "!min", ] gather_network_resources = [ "firewall_rules", ] def __init__(self, module): super(Firewall_rules, self).__init__(module) def get_firewall_rules_facts(self, data=None): """Get the 'facts' (the current configuration) :rtype: A dictionary :returns: The current configuration as a dictionary """ facts, _warnings = Facts(self._module).get_facts( self.gather_subset, self.gather_network_resources, data=data ) firewall_rules_facts = facts["ansible_network_resources"].get( "firewall_rules" ) if not firewall_rules_facts: return [] return firewall_rules_facts def execute_module(self): """Execute the module :rtype: A dictionary :returns: The result from module execution """ result = {"changed": False} warnings = list() commands = list() if self.state in self.ACTION_STATES: existing_firewall_rules_facts = self.get_firewall_rules_facts() else: existing_firewall_rules_facts = [] if self.state in self.ACTION_STATES or self.state == "rendered": commands.extend(self.set_config(existing_firewall_rules_facts)) if commands and self.state in self.ACTION_STATES: if not self._module.check_mode: self._connection.edit_config(commands) result["changed"] = True if self.state in self.ACTION_STATES: result["commands"] = commands if self.state in self.ACTION_STATES or self.state == "gathered": changed_firewall_rules_facts = self.get_firewall_rules_facts() elif self.state == "rendered": result["rendered"] = commands elif self.state == "parsed": running_config = self._module.params["running_config"] if not running_config: self._module.fail_json( msg="value of running_config parameter must not be empty for state parsed" ) result["parsed"] = self.get_firewall_rules_facts( data=running_config ) else: changed_firewall_rules_facts = [] if self.state in self.ACTION_STATES: result["before"] = existing_firewall_rules_facts if result["changed"]: result["after"] = changed_firewall_rules_facts elif self.state == "gathered": result["gathered"] = changed_firewall_rules_facts result["warnings"] = warnings return result def set_config(self, existing_firewall_rules_facts): """Collect the configuration from the args passed to the module, collect the current configuration (as a dict from facts) :rtype: A list :returns: the commands necessary to migrate the current configuration to the desired configuration """ want = self._module.params["config"] have = existing_firewall_rules_facts resp = self.set_state(want, have) return to_list(resp) def set_state(self, w, h): """Select the appropriate function based on the state provided :param want: the desired configuration as a dictionary :param have: the current configuration as a dictionary :rtype: A list :returns: the commands necessary to migrate the current configuration to the desired configuration """ commands = [] if ( self.state in ("merged", "replaced", "overridden", "rendered") and not w ): self._module.fail_json( msg="value of config parameter must not be empty for state {0}".format( self.state ) ) if self.state == "overridden": commands.extend(self._state_overridden(w, h)) elif self.state == "deleted": commands.extend(self._state_deleted(w, h)) elif w: if self.state == "merged" or self.state == "rendered": commands.extend(self._state_merged(w, h)) elif self.state == "replaced": commands.extend(self._state_replaced(w, h)) return commands def _state_replaced(self, want, have): """The command generator when state is replaced :rtype: A list :returns: the commands necessary to migrate the current configuration to the desired configuration """ commands = [] if have: # Iterate over the afi rule sets we already have. for h in have: r_sets = self._get_r_sets(h) # Iterate over each rule set we already have. for rs in r_sets: # In the desired configuration, search for the rule set we # already have (to be replaced by our desired # configuration's rule set). wanted_rule_set = self.search_r_sets_in_have( want, rs["name"], "r_list", h["afi"] ) if wanted_rule_set is not None: # Remove the rules that we already have if the wanted # rules exist under the same name. commands.extend( self._add_r_sets( h["afi"], want=rs, have=wanted_rule_set, opr=False, ) ) # Merge the desired configuration into what we already have. commands.extend(self._state_merged(want, have)) return commands def _state_overridden(self, want, have): """The command generator when state is overridden :rtype: A list :returns: the commands necessary to migrate the current configuration to the desired configuration """ commands = [] if have: for h in have: r_sets = self._get_r_sets(h) for rs in r_sets: w = self.search_r_sets_in_have( want, rs["name"], "r_list", h["afi"] ) if not w: commands.append( self._compute_command( h["afi"], rs["name"], remove=True ) ) else: commands.extend( self._add_r_sets(h["afi"], rs, w, opr=False) ) commands.extend(self._state_merged(want, have)) return commands def _state_merged(self, want, have): """The command generator when state is merged :rtype: A list :returns: the commands necessary to merge the provided into the current configuration """ commands = [] for w in want: r_sets = self._get_r_sets(w) for rs in r_sets: h = self.search_r_sets_in_have( have, rs["name"], "r_list", w["afi"] ) commands.extend(self._add_r_sets(w["afi"], rs, h)) return commands def _state_deleted(self, want, have): """The command generator when state is deleted :rtype: A list :returns: the commands necessary to remove the current configuration of the provided objects """ commands = [] if want: for w in want: r_sets = self._get_r_sets(w) if r_sets: for rs in r_sets: h = self.search_r_sets_in_have( have, rs["name"], "r_list", w["afi"] ) if h: commands.append( self._compute_command( w["afi"], h["name"], remove=True ) ) elif have: for h in have: if h["afi"] == w["afi"]: commands.append( self._compute_command(w["afi"], remove=True) ) elif have: for h in have: r_sets = self._get_r_sets(h) if r_sets: commands.append( self._compute_command(afi=h["afi"], remove=True) ) return commands def _add_r_sets(self, afi, want, have, opr=True): """ This function forms the set/delete commands based on the 'opr' type for rule-sets attributes. :param afi: address type. :param want: desired config. :param have: target config. :param opr: True/False. :return: generated commands list. """ commands = [] l_set = ("description", "default_action", "enable_default_log") h_rs = {} h_rules = {} w_rs = deepcopy(remove_empties(want)) w_rules = w_rs.pop("rules", None) if have: h_rs = deepcopy(remove_empties(have)) h_rules = h_rs.pop("rules", None) if w_rs: for key, val in iteritems(w_rs): if ( opr and key in l_set and not (h_rs and self._is_w_same(w_rs, h_rs, key)) ): if key == "enable_default_log": if val and ( not h_rs or key not in h_rs or not h_rs[key] ): commands.append( self._add_rs_base_attrib( afi, want["name"], key, w_rs ) ) else: commands.append( self._add_rs_base_attrib( afi, want["name"], key, w_rs ) ) elif not opr and key in l_set: if ( key == "enable_default_log" and val and h_rs and (key not in h_rs or not h_rs[key]) ): commands.append( self._add_rs_base_attrib( afi, want["name"], key, w_rs, opr ) ) elif not (h_rs and self._in_target(h_rs, key)): commands.append( self._add_rs_base_attrib( afi, want["name"], key, w_rs, opr ) ) commands.extend( self._add_rules(afi, want["name"], w_rules, h_rules, opr) ) if h_rules: have["rules"] = h_rules if w_rules: want["rules"] = w_rules return commands def _add_rules(self, afi, name, w_rules, h_rules, opr=True): """ This function forms the set/delete commands based on the 'opr' type for rules attributes. :param want: desired config. :param have: target config. :return: generated commands list. """ commands = [] l_set = ( "ipsec", "action", "number", "protocol", "fragment", "disable", "description", "log", ) if w_rules: for w in w_rules: cmd = self._compute_command(afi, name, w["number"], opr=opr) h = self.search_r_sets_in_have( h_rules, w["number"], type="rules" ) for key, val in iteritems(w): if val: if ( opr and key in l_set and not (h and self._is_w_same(w, h, key)) ): if key == "disable": if not ( not val and (not h or key not in h or not h[key]) ): commands.append( self._add_r_base_attrib( afi, name, key, w ) ) else: commands.append( self._add_r_base_attrib(afi, name, key, w) ) elif not opr: if key == "number" and self._is_del(l_set, h): commands.append( self._add_r_base_attrib( afi, name, key, w, opr=opr ) ) continue if ( key == "disable" and val and h and (key not in h or not h[key]) ): commands.append( self._add_r_base_attrib( afi, name, key, w, opr=opr ) ) elif ( key in l_set and not (h and self._in_target(h, key)) and not self._is_del(l_set, h) ): commands.append( self._add_r_base_attrib( afi, name, key, w, opr=opr ) ) elif key == "p2p": commands.extend(self._add_p2p(key, w, h, cmd, opr)) elif key == "tcp": commands.extend(self._add_tcp(key, w, h, cmd, opr)) elif key == "time": commands.extend( self._add_time(key, w, h, cmd, opr) ) elif key == "icmp": commands.extend( self._add_icmp(key, w, h, cmd, opr) ) elif key == "state": commands.extend( self._add_state(key, w, h, cmd, opr) ) elif key == "limit": commands.extend( self._add_limit(key, w, h, cmd, opr) ) elif key == "recent": commands.extend( self._add_recent(key, w, h, cmd, opr) ) elif key == "destination" or key == "source": commands.extend( self._add_src_or_dest(key, w, h, cmd, opr) ) return commands def _add_p2p(self, attr, w, h, cmd, opr): """ This function forms the set/delete commands based on the 'opr' type for p2p applications attributes. :param want: desired config. :param have: target config. :return: generated commands list. """ commands = [] have = [] if w: want = w.get(attr) or [] if h: have = h.get(attr) or [] if want: if opr: applications = list_diff_want_only(want, have) for app in applications: commands.append( cmd + (" " + attr + " " + app["application"]) ) elif not opr and have: applications = list_diff_want_only(want, have) for app in applications: commands.append( cmd + (" " + attr + " " + app["application"]) ) return commands def _add_state(self, attr, w, h, cmd, opr): """ This function forms the command for 'state' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ h_state = {} commands = [] l_set = ("new", "invalid", "related", "established") if w[attr]: if h and attr in h.keys(): h_state = h.get(attr) or {} for item, val in iteritems(w[attr]): if ( opr and item in l_set and not ( h_state and self._is_w_same(w[attr], h_state, item) ) ): commands.append( cmd + ( " " + attr + " " + item + " " + self._bool_to_str(val) ) ) elif ( not opr and item in l_set and not (h_state and self._in_target(h_state, item)) ): commands.append(cmd + (" " + attr + " " + item)) return commands def _add_recent(self, attr, w, h, cmd, opr): """ This function forms the command for 'recent' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ commands = [] h_recent = {} l_set = ("count", "time") if w[attr]: if h and attr in h.keys(): h_recent = h.get(attr) or {} for item, val in iteritems(w[attr]): if ( opr and item in l_set and not ( h_recent and self._is_w_same(w[attr], h_recent, item) ) ): commands.append( cmd + (" " + attr + " " + item + " " + str(val)) ) elif ( not opr and item in l_set and not (h_recent and self._in_target(h_recent, item)) ): commands.append(cmd + (" " + attr + " " + item)) return commands def _add_icmp(self, attr, w, h, cmd, opr): """ This function forms the commands for 'icmp' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ commands = [] h_icmp = {} l_set = ("code", "type", "type_name") if w[attr]: if h and attr in h.keys(): h_icmp = h.get(attr) or {} for item, val in iteritems(w[attr]): if ( opr and item in l_set and not (h_icmp and self._is_w_same(w[attr], h_icmp, item)) ): if item == "type_name": + os_version = self._get_os_version() + ver = re.search( + "vyos ([\\d\\.]+)-?.*", # noqa: W605 + os_version, + re.IGNORECASE, + ) + if ver.group(1) >= "1.4": + param_name = "type-name" + else: + param_name = "type" if "ipv6-name" in cmd: commands.append( cmd - + (" " + "icmpv6" + " " + "type" + " " + val) + + ( + " " + + "icmpv6" + + " " + + param_name + + " " + + val + ) ) else: commands.append( cmd + ( " " + attr + " " + item.replace("_", "-") + " " + val ) ) else: commands.append( cmd + (" " + attr + " " + item + " " + str(val)) ) elif ( not opr and item in l_set and not (h_icmp and self._in_target(h_icmp, item)) ): commands.append(cmd + (" " + attr + " " + item)) return commands def _add_time(self, attr, w, h, cmd, opr): """ This function forms the commands for 'time' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ commands = [] h_time = {} l_set = ( "utc", "stopdate", "stoptime", "weekdays", "monthdays", "startdate", "starttime", ) if w[attr]: if h and attr in h.keys(): h_time = h.get(attr) or {} for item, val in iteritems(w[attr]): if ( opr and item in l_set and not (h_time and self._is_w_same(w[attr], h_time, item)) ): if item == "utc": if not ( not val and (not h_time or item not in h_time) ): commands.append(cmd + (" " + attr + " " + item)) else: commands.append( cmd + (" " + attr + " " + item + " " + val) ) elif ( not opr and item in l_set and not (h_time and self._is_w_same(w[attr], h_time, item)) ): commands.append(cmd + (" " + attr + " " + item)) return commands def _add_tcp(self, attr, w, h, cmd, opr): """ This function forms the commands for 'tcp' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ h_tcp = {} commands = [] if w[attr]: key = "flags" flags = w[attr].get(key) or {} if flags: if h and key in h[attr].keys(): h_tcp = h[attr].get(key) or {} if flags: if opr and not ( h_tcp and self._is_w_same(w[attr], h[attr], key) ): commands.append( cmd + (" " + attr + " " + key + " " + flags) ) if not opr and not ( h_tcp and self._is_w_same(w[attr], h[attr], key) ): commands.append( cmd + (" " + attr + " " + key + " " + flags) ) return commands def _add_limit(self, attr, w, h, cmd, opr): """ This function forms the commands for 'limit' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ h_limit = {} commands = [] if w[attr]: key = "burst" if ( opr and key in w[attr].keys() and not ( h and attr in h.keys() and self._is_w_same(w[attr], h[attr], key) ) ): commands.append( cmd + (" " + attr + " " + key + " " + str(w[attr].get(key))) ) elif ( not opr and key in w[attr].keys() and not ( h and attr in h.keys() and self._in_target(h[attr], key) ) ): commands.append( cmd + (" " + attr + " " + key + " " + str(w[attr].get(key))) ) key = "rate" rate = w[attr].get(key) or {} if rate: if h and key in h[attr].keys(): h_limit = h[attr].get(key) or {} if "unit" in rate and "number" in rate: if opr and not ( h_limit and self._is_w_same(rate, h_limit, "unit") and self.is_w_same(rate, h_limit, "number") ): commands.append( cmd + ( " " + attr + " " + key + " " + str(rate["number"]) + "/" + rate["unit"] ) ) if not opr and not ( h_limit and self._is_w_same(rate, h_limit, "unit") and self._is_w_same(rate, h_limit, "number") ): commands.append(cmd + (" " + attr + " " + key)) return commands def _add_src_or_dest(self, attr, w, h, cmd, opr=True): """ This function forms the commands for 'src/dest' attributes based on the 'opr'. :param attr: attribute name. :param w: base config. :param h: target config. :param cmd: commands to be prepend. :return: generated list of commands. """ commands = [] h_group = {} g_set = ("port_group", "address_group", "network_group") if w[attr]: keys = ("address", "mac_address", "port") for key in keys: if ( opr and key in w[attr].keys() and not ( h and attr in h.keys() and self._is_w_same(w[attr], h[attr], key) ) ): commands.append( cmd + ( " " + attr + " " + key.replace("_", "-") + " " + w[attr].get(key) ) ) elif ( not opr and key in w[attr].keys() and not ( h and attr in h.keys() and self._in_target(h[attr], key) ) ): commands.append(cmd + (" " + attr + " " + key)) key = "group" group = w[attr].get(key) or {} if group: h_group = {} if h and h.get(attr) and key in h[attr].keys(): h_group = h[attr].get(key) for item, val in iteritems(group): if val: if ( opr and item in g_set and not ( h_group and self._is_w_same(group, h_group, item) ) ): commands.append( cmd + ( " " + attr + " " + key + " " + item.replace("_", "-") + " " + val ) ) elif ( not opr and item in g_set and not ( h_group and self._in_target(h_group, item) ) ): commands.append( cmd + ( " " + attr + " " + key + " " + item.replace("_", "-") ) ) return commands def search_r_sets_in_have(self, have, w_name, type="rule_sets", afi=None): """ This function returns the rule-set/rule if it is present in target config. :param have: target config. :param w_name: rule-set name. :param type: rule_sets/rule/r_list. :param afi: address family (when type is r_list). :return: rule-set/rule. """ if have: key = "name" if type == "rules": key = "number" for r in have: if r[key] == w_name: return r elif type == "r_list": for h in have: if h["afi"] == afi: r_sets = self._get_r_sets(h) for rs in r_sets: if rs[key] == w_name: return rs else: for rs in have: if rs[key] == w_name: return rs return None def _get_r_sets(self, item, type="rule_sets"): """ This function returns the list of rule-sets/rules. :param item: config dictionary. :param type: rule_sets/rule/r_list. :return: list of rule-sets/rules. """ rs_list = [] r_sets = item[type] if r_sets: for rs in r_sets: rs_list.append(rs) return rs_list def _compute_command( self, afi, name=None, number=None, attrib=None, value=None, remove=False, opr=True, ): """ This function construct the add/delete command based on passed attributes. :param afi: address type. :param name: rule-set name. :param number: rule-number. :param attrib: attribute name. :param value: value. :param remove: True if delete command needed to be construct. :param opr: opeeration flag. :return: generated command. """ if remove or not opr: cmd = "delete firewall " + self._get_fw_type(afi) else: cmd = "set firewall " + self._get_fw_type(afi) if name: cmd += " " + name if number: cmd += " rule " + str(number) if attrib: cmd += " " + attrib.replace("_", "-") if ( value and opr and attrib != "enable_default_log" and attrib != "disable" ): cmd += " '" + str(value) + "'" return cmd def _add_r_base_attrib(self, afi, name, attr, rule, opr=True): """ This function forms the command for 'rules' attributes which doesn't have further sub attributes. :param afi: address type. :param name: rule-set name :param attrib: attribute name :param rule: rule config dictionary. :param opr: True/False. :return: generated command. """ if attr == "number": command = self._compute_command( afi=afi, name=name, number=rule["number"], opr=opr ) else: command = self._compute_command( afi=afi, name=name, number=rule["number"], attrib=attr, value=rule[attr], opr=opr, ) return command def _add_rs_base_attrib(self, afi, name, attrib, rule, opr=True): """ This function forms the command for 'rule-sets' attributes which doesn't have further sub attributes. :param afi: address type. :param name: rule-set name :param attrib: attribute name :param rule: rule config dictionary. :param opr: True/False. :return: generated command. """ command = self._compute_command( afi=afi, name=name, attrib=attrib, value=rule[attrib], opr=opr ) return command def _bool_to_str(self, val): """ This function converts the bool value into string. :param val: bool value. :return: enable/disable. """ return "enable" if val else "disable" def _get_fw_type(self, afi): """ This function returns the firewall rule-set type based on IP address. :param afi: address type :return: rule-set type. """ return "ipv6-name" if afi == "ipv6" else "name" def _is_del(self, l_set, h, key="number"): """ This function checks whether rule needs to be deleted based on the rule number. :param l_set: attribute set. :param h: target config. :param key: number. :return: True/False. """ return key in l_set and not (h and self._in_target(h, key)) def _is_w_same(self, w, h, key): """ This function checks whether the key value is same in base and target config dictionary. :param w: base config. :param h: target config. :param key:attribute name. :return: True/False. """ return True if h and key in h and h[key] == w[key] else False def _in_target(self, h, key): """ This function checks whether the target nexist and key present in target config. :param h: target config. :param key: attribute name. :return: True/False. """ return True if h and key in h else False def _is_base_attrib(self, key): """ This function checks whether key is present in predefined based attribute set. :param key: :return: True/False. """ r_set = ( "p2p", "ipsec", "log", "action", "fragment", "protocol", "disable", "description", "mac_address", "default_action", "enable_default_log", ) return True if key in r_set else False + + def _get_os_version(self): + os_version = "1.2" + if self._connection: + os_version = self._connection.get_device_info()[ + "network_os_version" + ] + return os_version diff --git a/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py b/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py index 4e1f344..80d0db5 100644 --- a/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py +++ b/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py @@ -1,1146 +1,1198 @@ # (c) 2016 Red Hat Inc. # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.plugins.modules import vyos_firewall_rules from ansible_collections.vyos.vyos.tests.unit.modules.utils import ( set_module_args, ) from .vyos_module import TestVyosModule, load_fixture class TestVyosFirewallRulesModule(TestVyosModule): module = vyos_firewall_rules def setUp(self): super(TestVyosFirewallRulesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = ( self.mock_get_resource_connection_config.start() ) self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = ( self.mock_get_resource_connection_facts.start() ) self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.static_routes.static_routes.Static_routesFacts.get_device_data" ) self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.firewall_rules.firewall_rules.Firewall_rulesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() + self.mock_get_os_version = patch( + "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.config.firewall_rules.firewall_rules.Firewall_rules._get_os_version" + ) + self.get_os_version = self.mock_get_os_version.start() + self.get_os_version.return_value = "Vyos 1.2" + def tearDown(self): super(TestVyosFirewallRulesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() + self.mock_get_os_version.stop() def load_fixtures(self, commands=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_firewall_rules_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_firewall_rule_set_01_merged(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="V6-INBOUND", description="This is IPv6 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V6-OUTBOUND", description="This is IPv6 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), dict( afi="ipv4", rule_sets=[ dict( name="V4-INBOUND", description="This is IPv4 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V4-OUTBOUND", description="This is IPv4 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), ], state="merged", ) ) commands = [ "set firewall ipv6-name V6-INBOUND default-action 'reject'", "set firewall ipv6-name V6-INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name V6-INBOUND enable-default-log", "set firewall ipv6-name V6-OUTBOUND default-action 'accept'", "set firewall ipv6-name V6-OUTBOUND description 'This is IPv6 OUTBOUND rule set'", "set firewall name V4-INBOUND default-action 'reject'", "set firewall name V4-INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name V4-INBOUND enable-default-log", "set firewall name V4-OUTBOUND default-action 'accept'", "set firewall name V4-OUTBOUND description 'This is IPv4 OUTBOUND rule set'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_02_merged(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="V6-INBOUND", description="This is IPv6 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V6-OUTBOUND", description="This is IPv6 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), dict( afi="ipv4", rule_sets=[ dict( name="V4-INBOUND", description="This is IPv4 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V4-OUTBOUND", description="This is IPv4 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), ], state="merged", ) ) commands = [ "set firewall ipv6-name V6-INBOUND default-action 'reject'", "set firewall ipv6-name V6-INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name V6-INBOUND enable-default-log", "set firewall ipv6-name V6-OUTBOUND default-action 'accept'", "set firewall ipv6-name V6-OUTBOUND description 'This is IPv6 OUTBOUND rule set'", "set firewall name V4-INBOUND default-action 'reject'", "set firewall name V4-INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name V4-INBOUND enable-default-log", "set firewall name V4-OUTBOUND default-action 'accept'", "set firewall name V4-OUTBOUND description 'This is IPv4 OUTBOUND rule set'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", description="This is IPv4 INBOUND rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", log="disable", protocol="icmp", fragment="match-frag", disable=True, ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND default-action 'accept'", "set firewall name INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name INBOUND enable-default-log", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", "set firewall name INBOUND rule 101 fragment 'match-frag'", "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 disable", "set firewall name INBOUND rule 101 action 'accept'", "set firewall name INBOUND rule 101 ipsec 'match-ipsec'", "set firewall name INBOUND rule 101 log 'disable'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="tcp", source=dict( address="192.0.2.0", mac_address="38:00:25:19:76:0c", port=2127, ), destination=dict( address="192.0.1.0", port=2124 ), limit=dict( burst=10, rate=dict( number=20, unit="second" ), ), recent=dict(count=10, time=20), state=dict( established=True, related=True, invalid=True, new=True, ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 protocol 'tcp'", "set firewall name INBOUND rule 101 destination address 192.0.1.0", "set firewall name INBOUND rule 101 destination port 2124", "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 source address 192.0.2.0", "set firewall name INBOUND rule 101 source mac-address 38:00:25:19:76:0c", "set firewall name INBOUND rule 101 source port 2127", "set firewall name INBOUND rule 101 state new enable", "set firewall name INBOUND rule 101 state invalid enable", "set firewall name INBOUND rule 101 state related enable", "set firewall name INBOUND rule 101 state established enable", "set firewall name INBOUND rule 101 limit burst 10", "set firewall name INBOUND rule 101 limit rate 20/second", "set firewall name INBOUND rule 101 recent count 10", "set firewall name INBOUND rule 101 recent time 20", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_03(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", destination=dict( group=dict( address_group="OUT-ADDR-GROUP", network_group="OUT-NET-GROUP", port_group="OUT-PORT-GROUP", ) ), source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 source group address-group IN-ADDR-GROUP", "set firewall name INBOUND rule 101 source group network-group IN-NET-GROUP", "set firewall name INBOUND rule 101 source group port-group IN-PORT-GROUP", "set firewall name INBOUND rule 101 destination group address-group OUT-ADDR-GROUP", "set firewall name INBOUND rule 101 destination group network-group OUT-NET-GROUP", "set firewall name INBOUND rule 101 destination group port-group OUT-PORT-GROUP", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_04(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", time=dict( monthdays="2", startdate="2020-01-24", starttime="13:20:00", stopdate="2020-01-28", stoptime="13:30:00", weekdays="!Sat,Sun", utc=True, ), tcp=dict(flags="ALL"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 tcp flags ALL", "set firewall name INBOUND rule 101 time utc", "set firewall name INBOUND rule 101 time monthdays 2", "set firewall name INBOUND rule 101 time startdate 2020-01-24", "set firewall name INBOUND rule 101 time stopdate 2020-01-28", "set firewall name INBOUND rule 101 time weekdays !Sat,Sun", "set firewall name INBOUND rule 101 time stoptime 13:30:00", "set firewall name INBOUND rule 101 time starttime 13:20:00", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_01(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", description="This is IPv6 INBOUND rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", disabled=True, icmp=dict(type_name="echo-request"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND default-action 'accept'", "set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name INBOUND enable-default-log", "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", "set firewall ipv6-name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 disable", "set firewall ipv6-name INBOUND rule 101 action 'accept'", "set firewall ipv6-name INBOUND rule 101 ipsec 'match-ipsec'", "set firewall ipv6-name INBOUND rule 101 icmpv6 type echo-request", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_02(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="tcp", source=dict( address="2001:db8::12", mac_address="38:00:25:19:76:0c", port=2127, ), destination=dict( address="2001:db8::11", port=2124 ), limit=dict( burst=10, rate=dict( number=20, unit="second" ), ), recent=dict(count=10, time=20), state=dict( established=True, related=True, invalid=True, new=True, ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 protocol 'tcp'", "set firewall ipv6-name INBOUND rule 101 destination address 2001:db8::11", "set firewall ipv6-name INBOUND rule 101 destination port 2124", "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 source address 2001:db8::12", "set firewall ipv6-name INBOUND rule 101 source mac-address 38:00:25:19:76:0c", "set firewall ipv6-name INBOUND rule 101 source port 2127", "set firewall ipv6-name INBOUND rule 101 state new enable", "set firewall ipv6-name INBOUND rule 101 state invalid enable", "set firewall ipv6-name INBOUND rule 101 state related enable", "set firewall ipv6-name INBOUND rule 101 state established enable", "set firewall ipv6-name INBOUND rule 101 limit burst 10", "set firewall ipv6-name INBOUND rule 101 recent count 10", "set firewall ipv6-name INBOUND rule 101 recent time 20", "set firewall ipv6-name INBOUND rule 101 limit rate 20/second", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_03(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", destination=dict( group=dict( address_group="OUT-ADDR-GROUP", network_group="OUT-NET-GROUP", port_group="OUT-PORT-GROUP", ) ), source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 source group address-group IN-ADDR-GROUP", "set firewall ipv6-name INBOUND rule 101 source group network-group IN-NET-GROUP", "set firewall ipv6-name INBOUND rule 101 source group port-group IN-PORT-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group address-group OUT-ADDR-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group network-group OUT-NET-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group port-group OUT-PORT-GROUP", "set firewall ipv6-name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_04(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", time=dict( monthdays="2", startdate="2020-01-24", starttime="13:20:00", stopdate="2020-01-28", stoptime="13:30:00", weekdays="!Sat,Sun", utc=True, ), tcp=dict(flags="ALL"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 tcp flags ALL", "set firewall ipv6-name INBOUND rule 101 time utc", "set firewall ipv6-name INBOUND rule 101 time monthdays 2", "set firewall ipv6-name INBOUND rule 101 time startdate 2020-01-24", "set firewall ipv6-name INBOUND rule 101 time stopdate 2020-01-28", "set firewall ipv6-name INBOUND rule 101 time weekdays !Sat,Sun", "set firewall ipv6-name INBOUND rule 101 time stoptime 13:30:00", "set firewall ipv6-name INBOUND rule 101 time starttime 13:20:00", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_icmp_01(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict( type_name="port-unreachable" ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 icmpv6 type port-unreachable", "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", "set firewall ipv6-name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_icmp_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict(type=1, code=1), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 icmp type 1", "set firewall name INBOUND rule 101 icmp code 1", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_icmp_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict(type_name="echo-request"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 icmp type-name echo-request", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_del_01(self): set_module_args( dict( config=[dict(afi="ipv4", rule_sets=[dict(name="V4-INGRESS")])], state="deleted", ) ) commands = ["delete firewall name V4-INGRESS"] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_02(self): set_module_args( dict( config=[ dict(afi="ipv4", rule_sets=[dict(name="V4-INGRESS")]), dict(afi="ipv6", rule_sets=[dict(name="V6-INGRESS")]), ], state="deleted", ) ) commands = [ "delete firewall name V4-INGRESS", "delete firewall ipv6-name V6-INGRESS", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_03(self): set_module_args(dict(config=[], state="deleted")) commands = ["delete firewall name", "delete firewall ipv6-name"] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_04(self): set_module_args( dict( config=[ dict(afi="ipv4", rule_sets=[dict(name="V4-ING")]), dict(afi="ipv6", rule_sets=[dict(name="V6-ING")]), ], state="deleted", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_rep_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="reject", description="Rule 101 is configured by Ansible RM", ipsec="match-ipsec", protocol="tcp", fragment="match-frag", disabled=False, ), dict( number="102", action="accept", description="Rule 102 is configured by Ansible RM", protocol="icmp", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", description="This rule-set is configured by Ansible RM", ), dict( name="EGRESS", default_action="reject", description="This rule-set is configured by Ansible RM", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) commands = [ "delete firewall name V4-INGRESS rule 101 disable", "set firewall name V4-INGRESS description 'This is IPv4 INGRESS rule set'", "set firewall name V4-INGRESS rule 101 protocol 'tcp'", "set firewall name V4-INGRESS rule 101 description 'Rule 101 is configured by Ansible RM'", "set firewall name V4-INGRESS rule 101 action 'reject'", "set firewall name V4-INGRESS rule 102 disable", "set firewall name V4-INGRESS rule 102 action 'accept'", "set firewall name V4-INGRESS rule 102 protocol 'icmp'", "set firewall name V4-INGRESS rule 102 description 'Rule 102 is configured by Ansible RM'", "set firewall name V4-INGRESS rule 102", "set firewall ipv6-name V6-INGRESS description 'This rule-set is configured by Ansible RM'", "set firewall ipv6-name EGRESS description 'This rule-set is configured by Ansible RM'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_rep_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=False, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) commands = [ "delete firewall name V4-INGRESS enable-default-log", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ), ], ), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_mer_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_ovr_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-IN", description="This is IPv4 INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="1", action="reject", description="Rule 1 is configured by Ansible RM", ipsec="match-ipsec", log="enable", protocol="tcp", fragment="match-frag", disabled=False, source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ), dict( number="2", action="accept", description="Rule 102 is configured by Ansible RM", protocol="icmp", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-IN", default_action="accept", description="This rule-set is configured by Ansible RM", ), dict( name="V6-EG", default_action="reject", description="This rule-set is configured by Ansible RM", ), ], ), ], state="overridden", ) ) commands = [ "delete firewall ipv6-name V6-INGRESS", "delete firewall ipv6-name EGRESS", "delete firewall name V4-INGRESS", "delete firewall name EGRESS", "set firewall name V4-IN default-action 'accept'", "set firewall name V4-IN description 'This is IPv4 INGRESS rule set'", "set firewall name V4-IN enable-default-log", "set firewall name V4-IN rule 1 protocol 'tcp'", "set firewall name V4-IN rule 1 log 'enable'", "set firewall name V4-IN rule 1 description 'Rule 1 is configured by Ansible RM'", "set firewall name V4-IN rule 1 fragment 'match-frag'", "set firewall name V4-IN rule 1 source group address-group IN-ADDR-GROUP", "set firewall name V4-IN rule 1 source group network-group IN-NET-GROUP", "set firewall name V4-IN rule 1 source group port-group IN-PORT-GROUP", "set firewall name V4-IN rule 1", "set firewall name V4-IN rule 1 action 'reject'", "set firewall name V4-IN rule 1 ipsec 'match-ipsec'", "set firewall name V4-IN rule 2 disable", "set firewall name V4-IN rule 2 action 'accept'", "set firewall name V4-IN rule 2 protocol 'icmp'", "set firewall name V4-IN rule 2 description 'Rule 102 is configured by Ansible RM'", "set firewall name V4-IN rule 2", "set firewall ipv6-name V6-IN default-action 'accept'", "set firewall ipv6-name V6-IN description 'This rule-set is configured by Ansible RM'", "set firewall ipv6-name V6-EG default-action 'reject'", "set firewall ipv6-name V6-EG description 'This rule-set is configured by Ansible RM'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_ovr_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) + + def test_vyos_firewall_v6_rule_sets_rule_merged_01_version(self): + self.get_os_version.return_value = "VyOS 1.4-rolling-202007010117" + set_module_args( + dict( + config=[ + dict( + afi="ipv6", + rule_sets=[ + dict( + name="INBOUND", + description="This is IPv6 INBOUND rule set", + default_action="accept", + enable_default_log=True, + rules=[ + dict( + number="101", + action="accept", + description="Rule 101 is configured by Ansible", + ipsec="match-ipsec", + protocol="icmp", + disabled=True, + icmp=dict(type_name="echo-request"), + ) + ], + ), + ], + ) + ], + state="merged", + ) + ) + commands = [ + "set firewall ipv6-name INBOUND default-action 'accept'", + "set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set'", + "set firewall ipv6-name INBOUND enable-default-log", + "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", + "set firewall ipv6-name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", + "set firewall ipv6-name INBOUND rule 101", + "set firewall ipv6-name INBOUND rule 101 disable", + "set firewall ipv6-name INBOUND rule 101 action 'accept'", + "set firewall ipv6-name INBOUND rule 101 ipsec 'match-ipsec'", + "set firewall ipv6-name INBOUND rule 101 icmpv6 type-name echo-request", + ] + self.execute_module(changed=True, commands=commands)