Page MenuHomeVyOS Platform
Files F119063716

No OneTemporary
Actions

Size
94 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/changelogs/fragments/fix-icmp-parse-attr.yml b/changelogs/fragments/fix-icmp-parse-attr.yml
new file mode 100644
index 00000000..5cf0e199
--- /dev/null
+++ b/changelogs/fragments/fix-icmp-parse-attr.yml
@@ -0,0 +1,5 @@
+---
+bugfixes:
+ - firewall_rules - fix parse_icmp_attr() using wrong split delimiter ('.' instead of '/')
+ and referencing undefined variable type_no in the numeric-only branch, which caused
+ ValueError or UnboundLocalError when gathering firewall rules with ICMP type conditions.
diff --git a/plugins/module_utils/network/vyos/facts/firewall_rules/firewall_rules.py b/plugins/module_utils/network/vyos/facts/firewall_rules/firewall_rules.py
index 270b4a62..f8f02e7a 100644
--- a/plugins/module_utils/network/vyos/facts/firewall_rules/firewall_rules.py
+++ b/plugins/module_utils/network/vyos/facts/firewall_rules/firewall_rules.py
@@ -1,544 +1,544 @@
#
# -*- 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 fact class
It is in this file the configuration is collected from the device
for a given resource, parsed, and the facts tree is populated
based on the configuration.
"""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from copy import deepcopy
from re import M, findall, search
from ansible_collections.ansible.netcommon.plugins.module_utils.network.common import utils
from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.argspec.firewall_rules.firewall_rules import (
Firewall_rulesArgs,
)
class Firewall_rulesFacts(object):
"""The vyos firewall_rules fact class"""
def __init__(self, module, subspec="config", options="options"):
self._module = module
self.argument_spec = Firewall_rulesArgs.argument_spec
spec = deepcopy(self.argument_spec)
if subspec:
if options:
facts_argument_spec = spec[subspec][options]
else:
facts_argument_spec = spec[subspec]
else:
facts_argument_spec = spec
self.generated_spec = utils.generate_dict(facts_argument_spec)
def get_device_data(self, connection):
return connection.get_config()
def populate_facts(self, connection, ansible_facts, data=None):
"""Populate the facts for firewall_rules
:param connection: the device connection
:param ansible_facts: Facts dictionary
:param data: previously collected conf
:rtype: dictionary
:returns: facts
"""
if not data:
# typically data is populated from the current device configuration
# data = connection.get('show running-config | section ^interface')
# using mock data instead
data = self.get_device_data(connection)
# split the config into instances of the resource
objs = []
# check 1.4+ first
new_rules = True
v6_rules = findall(
r"^set firewall ipv6 (name|forward|input|output) (?:\'*)(\S+)(?:\'*)",
data,
M,
)
if not v6_rules:
v6_rules = findall(r"^set firewall ipv6-name (?:\'*)(\S+)(?:\'*)", data, M)
if v6_rules:
new_rules = False
v4_rules = findall(
r"^set firewall ipv4 (name|forward|input|output) (?:\'*)(\S+)(?:\'*)",
data,
M,
)
if not v4_rules:
v4_rules = findall(r"^set firewall name (?:\'*)(\S+)(?:\'*)", data, M)
if v4_rules:
new_rules = False
if v6_rules:
if new_rules:
config = self.get_rules_post_1_4(data, v6_rules, type="ipv6")
else:
config = self.get_rules(data, v6_rules, type="ipv6")
if config:
config = utils.remove_empties(config)
objs.append(config)
if v4_rules:
if new_rules:
config = self.get_rules_post_1_4(data, v4_rules, type="ipv4")
else:
config = self.get_rules(data, v4_rules, type="ipv4")
if config:
config = utils.remove_empties(config)
objs.append(config)
ansible_facts["ansible_network_resources"].pop("firewall_rules", None)
facts = {}
if objs:
facts["firewall_rules"] = []
params = utils.validate_config(self.argument_spec, {"config": objs})
for cfg in params["config"]:
facts["firewall_rules"].append(utils.remove_empties(cfg))
ansible_facts["ansible_network_resources"].update(facts)
return ansible_facts
def get_rules(self, data, rules, type):
"""
This function performs following:
- Form regex to fetch 'rule-sets' specific config from data.
- Form the rule-set list based on ip address.
:param data: configuration.
:param rules: list of rule-sets.
:param type: ip address type.
:return: generated rule-sets configuration.
"""
r_v4 = []
r_v6 = []
for r in set(rules):
name_key = "ipv6-name" if type == "ipv6" else "name"
rule_regex = r" %s %s .+$" % (name_key, r.strip("'"))
cfg = findall(rule_regex, data, M)
fr = self.render_config(cfg, r.strip("'"))
fr["name"] = r.strip("'")
if type == "ipv6":
r_v6.append(fr)
else:
r_v4.append(fr)
if r_v4:
config = {"afi": "ipv4", "rule_sets": r_v4}
if r_v6:
config = {"afi": "ipv6", "rule_sets": r_v6}
return config
def get_rules_post_1_4(self, data, rules, type):
"""
This function performs following:
- Form regex to fetch 'rule-sets' specific config from data.
- Form the rule-set list based on ip address.
Specifically for v1.4+ version.
:param data: configuration.
:param rules: list of rule-sets.
:param type: ip address type.
:return: generated rule-sets configuration.
"""
r_v4 = []
r_v6 = []
for kind, name in set(rules):
rule_regex = r" %s %s %s .+$" % (type, kind, name.strip("'"))
cfg = findall(rule_regex, data, M)
fr = self.render_config(cfg, name.strip("'"))
if kind == "name":
fr["name"] = name.strip("'")
elif kind in ("forward", "input", "output"):
fr["filter"] = kind
else:
raise ValueError("Unknown rule kind: %s %s" % kind, name)
if type == "ipv6":
r_v6.append(fr)
else:
r_v4.append(fr)
if r_v4:
config = {"afi": "ipv4", "rule_sets": r_v4}
if r_v6:
config = {"afi": "ipv6", "rule_sets": r_v6}
return config
def render_config(self, conf, match):
"""
Render config as dictionary structure and delete keys
from spec for null values
:param spec: The facts tree, generated from the argspec
:param conf: The configuration
:rtype: dictionary
:returns: The generated config
"""
conf = "\n".join(filter(lambda x: x, conf))
a_lst = [
"description",
"default_action",
"default_jump_target",
"enable_default_log",
"default_log",
]
config = self.parse_attr(conf, a_lst, match)
if not config:
config = {}
if "default_log" in config:
config["enable_default_log"] = config.pop("default_log")
config["rules"] = self.parse_rules_lst(conf)
return config
def parse_rules_lst(self, conf):
"""
This function forms the regex to fetch the 'rules' with in
'rule-sets'
:param conf: configuration data.
:return: generated rule list configuration.
"""
r_lst = []
rules = findall(r"rule (?:\'*)(\d+)(?:\'*)", conf, M)
if rules:
rules_lst = []
for r in set(rules):
r_regex = r" %s .+$" % r
cfg = "\n".join(findall(r_regex, conf, M))
obj = self.parse_rules(cfg)
obj["number"] = int(r)
if obj:
rules_lst.append(obj)
r_lst = sorted(rules_lst, key=lambda i: i["number"])
return r_lst
def parse_rules(self, conf):
"""
This function triggers the parsing of 'rule' attributes.
a_lst is a list having rule attributes which doesn't
have further sub attributes.
:param conf: configuration
:return: generated rule configuration dictionary.
"""
a_lst = [
"ipsec",
"log",
"action",
"protocol",
"fragment",
"disable",
"description",
"icmp",
"jump_target",
"offload_target",
"queue",
"queue_options",
]
rule = self.parse_attr(conf, a_lst)
r_sub = {
"tcp": self.parse_tcp(conf),
"icmp": self.parse_icmp(conf, "icmp"),
"time": self.parse_time(conf, "time"),
"limit": self.parse_limit(conf, "limit"),
"state": self.parse_state(conf, "state"),
"recent": self.parse_recent(conf, "recent"),
"source": self.parse_src_or_dest(conf, "source"),
"destination": self.parse_src_or_dest(conf, "destination"),
"inbound_interface": self.parse_interface(conf, "inbound-interface"),
"outbound_interface": self.parse_interface(conf, "outbound-interface"),
"packet_length": self.parse_packet_length(conf, "packet-length"),
"packet_length_exclude": self.parse_packet_length(conf, "packet-length-exclude"),
}
rule.update(r_sub)
return rule
def parse_interface(self, conf, attrib):
"""
This function triggers the parsing of 'interface' attributes.
:param conf: configuration.
:param attrib: 'interface'.
:return: generated config dictionary.
"""
a_lst = ["name", "group"]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
return cfg_dict
def parse_packet_length(self, conf, attrib=None):
"""
This function triggers the parsing of 'packet-length' attributes.
:param conf: configuration.
:param attrib: 'packet-length'.
:return: generated config dictionary.
"""
lengths = []
rule_regex = r"%s (.+)$" % attrib
found_lengths = findall(rule_regex, conf, M)
if found_lengths:
lengths = []
for pplen in set(found_lengths):
obj = {"length": pplen.strip("'")}
lengths.append(obj)
return lengths
def parse_src_or_dest(self, conf, attrib=None):
"""
This function triggers the parsing of 'source or
destination' attributes.
:param conf: configuration.
:param attrib:'source/destination'.
:return:generated source/destination configuration dictionary.
"""
a_lst = ["port", "address", "mac_address"]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
cfg_dict["group"] = self.parse_group(conf, attrib + " group")
return cfg_dict
def parse_recent(self, conf, attrib=None):
"""
This function triggers the parsing of 'recent' attributes
:param conf: configuration.
:param attrib: 'recent'.
:return: generated config dictionary.
"""
a_lst = ["time", "count"]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
return cfg_dict
def parse_tcp(self, conf):
"""
This function triggers the parsing of 'tcp' attributes.
:param conf: configuration.
:param attrib: 'tcp'.
:return: generated config dictionary.
"""
f_lst = []
flags = findall(r"tcp flags (not )?(?:\'*)([\w!,]+)(?:\'*)", conf, M)
# for pre 1.4, this is a string including possible commas
# and ! as an inverter. For 1.4+ this is a single flag per
# command and 'not' as the inverter
if flags:
flag_lst = []
for n, f in set(flags):
f = f.strip("'").lower()
if "," in f:
# pre 1.4 version with multiple flags
fs = f.split(",")
for f in fs:
if "!" in f:
obj = {"flag": f.strip("'!"), "invert": True}
else:
obj = {"flag": f.strip("'")}
flag_lst.append(obj)
elif "!" in f:
obj = {"flag": f.strip("'!"), "invert": True}
flag_lst.append(obj)
else:
obj = {"flag": f.strip("'")}
if n:
obj["invert"] = True
flag_lst.append(obj)
f_lst = sorted(flag_lst, key=lambda i: i["flag"])
return {"flags": f_lst}
def parse_time(self, conf, attrib=None):
"""
This function triggers the parsing of 'time' attributes.
:param conf: configuration.
:param attrib: 'time'.
:return: generated config dictionary.
"""
a_lst = [
"stopdate",
"stoptime",
"weekdays",
"monthdays",
"startdate",
"starttime",
]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
return cfg_dict
def parse_state(self, conf, attrib=None):
"""
This function triggers the parsing of 'state' attributes.
:param conf: configuration
:param attrib: 'state'.
:return: generated config dictionary.
"""
a_lst = ["new", "invalid", "related", "established"]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
return cfg_dict
def parse_group(self, conf, attrib=None):
"""
This function triggers the parsing of 'group' attributes.
:param conf: configuration.
:param attrib: 'group'.
:return: generated config dictionary.
"""
a_lst = ["port_group", "address_group", "network_group"]
cfg_dict = self.parse_attr(conf, a_lst, match=attrib)
return cfg_dict
def parse_icmp_attr(self, conf, match):
"""
This function peforms the following:
- parse ICMP arguemnts for firewall rules
- consider that older versions may need numbers or letters
in type, newer ones are more specific
:param conf: configuration.
:param match: parent node/attribute name.
:return: generated config dictionary.
"""
config = {}
if not conf:
return config
for attrib in ("code", "type", "type-name"):
regex = self.map_regex(attrib)
if match:
regex = match + " " + regex
out = search(r"^.*" + regex + " (.+)", conf, M)
if out:
val = out.group(1).strip("'")
if attrib == "type-name":
config["type_name"] = val
if attrib == "code":
config["code"] = int(val)
if attrib == "type":
# <1.3 could be # (type), #/# (type/code) or 'type' (type_name)
# recent this is only for strings
if "/" in val: # type/code
- (type_no, code) = val.split(".")
- config["type"] = type_no
- config["code"] = code
+ (type_no, code) = val.split("/")
+ config["type"] = int(type_no)
+ config["code"] = int(code)
elif val.isnumeric():
- config["type"] = type_no
+ config["type"] = int(val)
else:
config["type_name"] = val
return config
def parse_icmp(self, conf, attrib=None):
"""
This function triggers the parsing of 'icmp' attributes.
:param conf: configuration to be parsed.
:param attrib: 'icmp'.
:return: generated config dictionary.
"""
cfg_dict = self.parse_icmp_attr(conf, "icmp")
if len(cfg_dict) == 0:
cfg_dict = self.parse_icmp_attr(conf, "icmpv6")
return cfg_dict
def parse_limit(self, conf, attrib=None):
"""
This function triggers the parsing of 'limit' attributes.
:param conf: configuration to be parsed.
:param attrib: 'limit'
:return: generated config dictionary.
"""
cfg_dict = self.parse_attr(conf, ["burst"], match=attrib)
cfg_dict["rate"] = self.parse_rate(conf, "rate")
return cfg_dict
def parse_attr(self, conf, attr_list, match=None):
"""
This function peforms the following:
- Form the regex to fetch the required attribute config.
- Type cast the output in desired format.
:param conf: configuration.
:param attr_list: list of attributes.
:param match: parent node/attribute name.
:return: generated config dictionary.
"""
config = {}
for attrib in attr_list:
regex = self.map_regex(attrib)
if match:
regex = match + " " + regex
if conf:
if self.is_bool(attrib):
out = conf.find(attrib.replace("_", "-"))
dis = conf.find(attrib.replace("_", "-") + " 'disable'")
if out >= 1:
if dis >= 1:
config[attrib] = False
else:
config[attrib] = True
else:
out = search(r"^.*" + regex + " (.+)", conf, M)
if not out:
if attrib == "disable":
out = search(r"^.*\d+" + " (disable$)", conf, M)
if attrib == "log":
out = search(r"^.*\d+" + " (log$)", conf, M)
if out:
val = out.group(1).strip("'")
if self.is_num(attrib):
val = int(val)
if attrib == "disable":
val = True
if attrib == "log":
val = "enable"
config[attrib] = val
return config
def map_regex(self, attrib):
"""
- This function construct the regex string.
- replace the underscore with hyphen.
:param attrib: attribute
:return: regex string
"""
regex = attrib.replace("_", "-")
if attrib == "disabled":
regex = "disable"
return regex
def is_bool(self, attrib):
"""
This function looks for the attribute in predefined bool type set.
:param attrib: attribute.
:return: True/False
"""
bool_set = (
"new",
"invalid",
"related",
"disabled",
"established",
"enable_default_log",
"default_log",
)
return True if attrib in bool_set else False
def is_num(self, attrib):
"""
This function looks for the attribute in predefined integer type set.
:param attrib: attribute.
:return: True/false.
"""
num_set = ("time", "code", "type", "count", "burst", "number")
return True if attrib in num_set else False
def parse_rate(self, conf, match):
"""
This function triggers the parsing of 'rate' attributes.
:param conf: configuration.
:param attrib: 'rate'
:return: generated config dictionary.
"""
config = {}
out = search(r"^.*" + match + " (.+)", conf, M)
if out:
val = out.group(1).strip("'")
if "/" in val: # number/unit
(number, unit) = val.split("/")
config["number"] = number
config["unit"] = unit
return config
diff --git a/tests/unit/modules/network/vyos/test_vyos_firewall_rules13.py b/tests/unit/modules/network/vyos/test_vyos_firewall_rules13.py
index 1af61752..edce123b 100644
--- a/tests/unit/modules/network/vyos/test_vyos_firewall_rules13.py
+++ b/tests/unit/modules/network/vyos/test_vyos_firewall_rules13.py
@@ -1,1578 +1,1592 @@
# (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 <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from unittest.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 TestVyosFirewallRulesModule13(TestVyosModule):
module = vyos_firewall_rules
def setUp(self):
super(TestVyosFirewallRulesModule13, 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.get_os_version",
)
self.get_os_version = self.mock_get_os_version.start()
self.get_os_version.return_value = "1.2"
def tearDown(self):
super(TestVyosFirewallRulesModule13, 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, filename=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_v4_rule_sets_rule_merged_01(self):
"""Test if plugin correctly adds new rules set and a rule with variant attributes"""
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,
),
dict(
number="105",
action="offload",
offload_target="test",
description="Rule 105 is configured by Ansible",
protocol="udp",
),
],
),
],
),
],
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'",
"set firewall name INBOUND rule 105",
"set firewall name INBOUND rule 105 action 'offload'",
"set firewall name INBOUND rule 105 offload-target 'test'",
"set firewall name INBOUND rule 105 description 'Rule 105 is configured by Ansible'",
"set firewall name INBOUND rule 105 protocol 'udp'",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_v4_rule_sets_rule_merged_02(self):
"""Test if plugin correctly adds new rules with variant attributes
within existing rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing rule set
"""
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=[
dict(flag="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):
"""Test if plugin correctly adds new ipv6 rules set and a rule with variant attributes"""
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",
disable=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):
"""Test if plugin correctly adds new rules with variant attributes
within existing ipv6 rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing ipv6 rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing ipv6 rule set
"""
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=[
dict(flag="all"),
],
),
),
dict(
number="102",
tcp=dict(
flags=[
dict(flag="ack"),
dict(flag="syn"),
dict(flag="fin", invert=True),
],
),
),
],
),
],
),
],
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",
"set firewall ipv6-name INBOUND rule 102",
"set firewall ipv6-name INBOUND rule 102 tcp flags ACK,SYN,!FIN",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_v6_rule_sets_rule_merged_icmp_01(self):
"""Test if plugin correctly adds new rules with variant attributes
within existing ipv6 rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing rule set
"""
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):
"""Test if plugin correctly adds new rules with variant attributes
within existing rule set
"""
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):
"""Test if plugin correctly removes existing rule set"""
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):
"""Test if plugin correctly removes existing rule sets, both ipv4 and ipv6"""
set_module_args(
dict(
config=[
dict(afi="ipv4", rule_sets=[dict(name="V4-INGRESS")]),
dict(afi="ipv6", rule_sets=[dict(name="V6-INGRESS")]),
dict(afi="ipv4", rule_sets=[dict(name="V4-OFFLOAD")]),
],
state="deleted",
),
)
commands = [
"delete firewall name V4-OFFLOAD",
"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):
"""Test if plugin correctly removes existing AFIs, both ipv4 and ipv6"""
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):
"""Test if plugin has no effect on non-existent rule sets"""
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):
"""Test if plugin correctly replaces a particular rule set(s)
without affecting the others
"""
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",
disable=False,
),
dict(
number="102",
action="accept",
description="Rule 102 is configured by Ansible RM",
protocol="icmp",
disable=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",
"set firewall name V4-INGRESS rule 101",
"set firewall name V4-INGRESS description 'This is IPv4 INGRESS rule set'",
"set firewall name V4-INGRESS rule 101 fragment 'match-frag'",
"set firewall name V4-INGRESS rule 101 ipsec 'match-ipsec'",
"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):
"""Test if plugin correctly replaces a particular rule(s) and rule set attribute(s)
without affecting the others
"""
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",
disable=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 rule 101",
"delete firewall name V4-INGRESS enable-default-log",
"set firewall name V4-INGRESS rule 101",
"set firewall name V4-INGRESS rule 101 action 'accept'",
"set firewall name V4-INGRESS rule 101 description 'Rule 101 is configured by Ansible'",
"set firewall name V4-INGRESS rule 101 disable",
"set firewall name V4-INGRESS rule 101 fragment 'match-frag'",
"set firewall name V4-INGRESS rule 101 ipsec 'match-ipsec'",
"set firewall name V4-INGRESS rule 101 protocol 'icmp'",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_01(self):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
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",
disable=True,
log="enable",
),
],
),
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):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
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",
disable=True,
log="enable",
),
],
),
],
),
],
state="replaced",
),
)
self.execute_module(changed=False, commands=[])
def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_03(self):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
set_module_args(
dict(
config=[
dict(
afi="ipv4",
rule_sets=[
dict(
name="V4-OFFLOAD",
rules=[
dict(
number="1",
action="offload",
offload_target="test",
),
],
),
],
),
],
state="replaced",
),
)
self.execute_module(changed=False, commands=[])
def test_vyos_firewall_v4v6_rule_sets_rule_mer_idem_01(self):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
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",
disable=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_mer_idem_02(self):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
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",
disable=True,
),
],
),
dict(
name="EGRESS",
default_action="reject",
),
dict(
name="V4-OFFLOAD",
rules=[
dict(
number="1",
action="offload",
offload_target="test",
),
],
),
],
),
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):
"""Test if plugin correctly resets the entire rule set if there is a change in the configuration"""
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",
disable=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",
disable=True,
),
],
),
dict(
name="MULTIPLE-RULE",
default_action="drop",
rules=[
dict(
number="1",
action="accept",
protocol="all",
),
],
),
],
),
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",
"delete firewall name MULTIPLE-RULE",
"delete firewall name V4-OFFLOAD",
"set firewall name MULTIPLE-RULE default-action 'drop'",
"set firewall name MULTIPLE-RULE rule 1",
"set firewall name MULTIPLE-RULE rule 1 action 'accept'",
"set firewall name MULTIPLE-RULE rule 1 protocol 'all'",
"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_02(self):
"""Test if plugin correctly resets the entire rule set
while removing the absent ones if there is a change in the configuration
"""
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="accept",
protocol="udp",
),
],
),
],
),
dict(
afi="ipv6",
rule_sets=[
dict(
name="EGRESS",
default_action="reject",
description="This rule-set is configured by Ansible RM",
rules=[
dict(
number="20",
action="accept",
protocol="udp",
),
],
),
],
),
],
state="overridden",
),
)
commands = [
"delete firewall ipv6-name V6-INGRESS",
"delete firewall ipv6-name EGRESS",
"delete firewall name V4-INGRESS",
"delete firewall name EGRESS",
"delete firewall name MULTIPLE-RULE",
"delete firewall name V4-OFFLOAD",
"set firewall name V4-INGRESS rule 101",
"set firewall name V4-INGRESS description 'This is IPv4 INGRESS rule set'",
"set firewall name V4-INGRESS default-action 'accept'",
"set firewall name V4-INGRESS enable-default-log",
"set firewall name V4-INGRESS rule 101 protocol 'udp'",
"set firewall name V4-INGRESS rule 101 action 'accept'",
"set firewall ipv6-name EGRESS description 'This rule-set is configured by Ansible RM'",
"set firewall ipv6-name EGRESS default-action 'reject'",
"set firewall ipv6-name EGRESS rule 20",
"set firewall ipv6-name EGRESS rule 20 protocol 'udp'",
"set firewall ipv6-name EGRESS rule 20 action 'accept'",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_v4v6_rule_sets_rule_ovr_idem_01(self):
"""Test if plugin correctly has no effect if there is no change in the configuration"""
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",
disable=True,
log="enable",
),
],
),
dict(
name="EGRESS",
default_action="reject",
),
dict(
name="MULTIPLE-RULE",
default_action="drop",
rules=[
dict(
number="1",
action="accept",
protocol="all",
),
dict(
number="2",
action="drop",
protocol="all",
),
],
),
dict(
name="V4-OFFLOAD",
rules=[
dict(
number="1",
action="offload",
offload_target="test",
),
],
),
],
),
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):
"""Test if plugin correctly adds ipv6 rule set with rules"""
self.get_os_version.return_value = "1.3"
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",
disable=True,
icmp=dict(type_name="echo-request"),
log="enable",
),
dict(
number="102",
action="reject",
description="Rule 102 is configured by Ansible",
protocol="ipv6-icmp",
icmp=dict(type=7),
),
],
),
],
),
],
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",
"set firewall ipv6-name INBOUND rule 101 log 'enable'",
"set firewall ipv6-name INBOUND rule 102",
"set firewall ipv6-name INBOUND rule 102 action 'reject'",
"set firewall ipv6-name INBOUND rule 102 description 'Rule 102 is configured by Ansible'",
"set firewall ipv6-name INBOUND rule 102 protocol 'ipv6-icmp'",
"set firewall ipv6-name INBOUND rule 102 icmpv6 type 7",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_jump_rules_merged_01(self):
"""Test if plugin correctly adds rule set with a jump action"""
set_module_args(
dict(
config=[
dict(
afi="ipv6",
rule_sets=[
dict(
name="INBOUND",
description="This is IPv6 INBOUND rule set with a jump action",
default_action="accept",
enable_default_log=True,
rules=[
dict(
number="101",
action="jump",
description="Rule 101 is configured by Ansible",
ipsec="match-ipsec",
protocol="icmp",
icmp=dict(type_name="echo-request"),
jump_target="PROTECT-RE",
packet_length_exclude=[dict(length=100), dict(length=200)],
),
dict(
number="102",
action="reject",
description="Rule 102 is configured by Ansible",
protocol="ipv6-icmp",
icmp=dict(type=7),
),
],
),
],
),
],
state="merged",
),
)
commands = [
"set firewall ipv6-name INBOUND default-action 'accept'",
"set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set with a jump action'",
"set firewall ipv6-name INBOUND enable-default-log",
"set firewall ipv6-name INBOUND rule 101 protocol 'icmp'",
"set firewall ipv6-name INBOUND rule 101 packet-length-exclude 100",
"set firewall ipv6-name INBOUND rule 101 packet-length-exclude 200",
"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 ipsec 'match-ipsec'",
"set firewall ipv6-name INBOUND rule 101 icmpv6 type echo-request",
"set firewall ipv6-name INBOUND rule 101 action 'jump'",
"set firewall ipv6-name INBOUND rule 101 jump-target 'PROTECT-RE'",
"set firewall ipv6-name INBOUND rule 102",
"set firewall ipv6-name INBOUND rule 102 action 'reject'",
"set firewall ipv6-name INBOUND rule 102 description 'Rule 102 is configured by Ansible'",
"set firewall ipv6-name INBOUND rule 102 protocol 'ipv6-icmp'",
"set firewall ipv6-name INBOUND rule 102 icmpv6 type 7",
]
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_log_merged_01(self):
"""Test if new stanza log is correctly applied"""
set_module_args(
dict(
config=[
dict(
afi="ipv6",
rule_sets=[
dict(
name="INBOUND",
description="This is IPv6 INBOUND rule set with a log",
default_action="accept",
enable_default_log=True,
rules=[
dict(
number="101",
action="accept",
description="Rule 101 is configured by Ansible",
log="enable",
),
],
),
],
),
],
state="merged",
),
)
commands = [
"set firewall ipv6-name INBOUND default-action 'accept'",
"set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set with a log'",
"set firewall ipv6-name INBOUND enable-default-log",
"set firewall ipv6-name INBOUND rule 101 log 'enable'",
"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 action 'accept'",
]
self.maxDiff = None
self.execute_module(changed=True, commands=commands)
def test_vyos_firewall_log_replace_01(self):
"""Test that stanza is correctly replaced
without touching the other stanzas
"""
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",
packet_length_exclude=[dict(length=100), dict(length=200)],
packet_length=[dict(length=22)],
log="enable",
),
],
),
],
),
],
state="replaced",
),
)
commands = [
"delete firewall name V4-INGRESS rule 101",
"set firewall name V4-INGRESS rule 101",
"set firewall name V4-INGRESS rule 101 action 'accept'",
"set firewall name V4-INGRESS rule 101 description 'Rule 101 is configured by Ansible'",
"set firewall name V4-INGRESS rule 101 packet-length-exclude 100",
"set firewall name V4-INGRESS rule 101 packet-length-exclude 200",
"set firewall name V4-INGRESS rule 101 packet-length 22",
"set firewall name V4-INGRESS rule 101 log 'enable'",
]
self.maxDiff = None
self.execute_module(changed=True, commands=commands)
+
+ def test_vyos_firewall_rules_parsed_icmp_type_code(self):
+ """parse_icmp_attr: legacy 'type/code' token parses into integer type and code."""
+ raw = (
+ "set firewall name TEST rule 1 action 'accept'\n"
+ "set firewall name TEST rule 1 protocol 'icmp'\n"
+ "set firewall name TEST rule 1 icmp type '3/4'\n"
+ )
+ set_module_args(dict(running_config=raw, state="parsed"))
+ result = self.execute_module(changed=False)
+ parsed = result["parsed"]
+ rule = parsed[0]["rule_sets"][0]["rules"][0]
+ self.assertEqual(rule["icmp"]["type"], 3)
+ self.assertEqual(rule["icmp"]["code"], 4)

File Metadata

Mime Type
text/x-diff
Expires
Fri, Nov 6, 9:02 AM (23 h, 19 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4628061
Default Alt Text
(94 KB)

Event Timeline