summaryrefslogtreecommitdiff
path: root/src/op_mode
diff options
context:
space:
mode:
Diffstat (limited to 'src/op_mode')
-rwxr-xr-xsrc/op_mode/firewall.py361
-rwxr-xr-xsrc/op_mode/monitor_bandwidth_test.sh3
-rwxr-xr-xsrc/op_mode/policy_route.py189
-rwxr-xr-xsrc/op_mode/show_virtual_server.py33
-rwxr-xr-xsrc/op_mode/vrrp.py13
-rwxr-xr-xsrc/op_mode/zone_policy.py81
6 files changed, 679 insertions, 1 deletions
diff --git a/src/op_mode/firewall.py b/src/op_mode/firewall.py
new file mode 100755
index 000000000..3146fc357
--- /dev/null
+++ b/src/op_mode/firewall.py
@@ -0,0 +1,361 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2021 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+import ipaddress
+import json
+import re
+import tabulate
+
+from vyos.config import Config
+from vyos.util import cmd
+from vyos.util import dict_search_args
+
+def get_firewall_interfaces(conf, firewall, name=None, ipv6=False):
+ interfaces = conf.get_config_dict(['interfaces'], key_mangling=('-', '_'),
+ get_first_key=True, no_tag_node_value_mangle=True)
+
+ directions = ['in', 'out', 'local']
+
+ def parse_if(ifname, if_conf):
+ if 'firewall' in if_conf:
+ for direction in directions:
+ if direction in if_conf['firewall']:
+ fw_conf = if_conf['firewall'][direction]
+ name_str = f'({ifname},{direction})'
+
+ if 'name' in fw_conf:
+ fw_name = fw_conf['name']
+
+ if not name:
+ firewall['name'][fw_name]['interface'].append(name_str)
+ elif not ipv6 and name == fw_name:
+ firewall['interface'].append(name_str)
+
+ if 'ipv6_name' in fw_conf:
+ fw_name = fw_conf['ipv6_name']
+
+ if not name:
+ firewall['ipv6_name'][fw_name]['interface'].append(name_str)
+ elif ipv6 and name == fw_name:
+ firewall['interface'].append(name_str)
+
+ for iftype in ['vif', 'vif_s', 'vif_c']:
+ if iftype in if_conf:
+ for vifname, vif_conf in if_conf[iftype].items():
+ parse_if(f'{ifname}.{vifname}', vif_conf)
+
+ for iftype, iftype_conf in interfaces.items():
+ for ifname, if_conf in iftype_conf.items():
+ parse_if(ifname, if_conf)
+
+ return firewall
+
+def get_config_firewall(conf, name=None, ipv6=False, interfaces=True):
+ config_path = ['firewall']
+ if name:
+ config_path += ['ipv6-name' if ipv6 else 'name', name]
+
+ firewall = conf.get_config_dict(config_path, key_mangling=('-', '_'),
+ get_first_key=True, no_tag_node_value_mangle=True)
+ if firewall and interfaces:
+ if name:
+ firewall['interface'] = []
+ else:
+ if 'name' in firewall:
+ for fw_name, name_conf in firewall['name'].items():
+ name_conf['interface'] = []
+
+ if 'ipv6_name' in firewall:
+ for fw_name, name_conf in firewall['ipv6_name'].items():
+ name_conf['interface'] = []
+
+ get_firewall_interfaces(conf, firewall, name, ipv6)
+ return firewall
+
+def get_nftables_details(name, ipv6=False):
+ suffix = '6' if ipv6 else ''
+ name_prefix = 'NAME6_' if ipv6 else 'NAME_'
+ command = f'sudo nft list chain ip{suffix} filter {name_prefix}{name}'
+ try:
+ results = cmd(command)
+ except:
+ return {}
+
+ out = {}
+ for line in results.split('\n'):
+ comment_search = re.search(rf'{name}[\- ](\d+|default-action)', line)
+ if not comment_search:
+ continue
+
+ rule = {}
+ rule_id = comment_search[1]
+ counter_search = re.search(r'counter packets (\d+) bytes (\d+)', line)
+ if counter_search:
+ rule['packets'] = counter_search[1]
+ rule['bytes'] = counter_search[2]
+
+ rule['conditions'] = re.sub(r'(\b(counter packets \d+ bytes \d+|drop|reject|return|log)\b|comment "[\w\-]+")', '', line).strip()
+ out[rule_id] = rule
+ return out
+
+def output_firewall_name(name, name_conf, ipv6=False, single_rule_id=None):
+ ip_str = 'IPv6' if ipv6 else 'IPv4'
+ print(f'\n---------------------------------\n{ip_str} Firewall "{name}"\n')
+
+ if name_conf['interface']:
+ print('Active on: {0}\n'.format(" ".join(name_conf['interface'])))
+
+ details = get_nftables_details(name, ipv6)
+ rows = []
+
+ if 'rule' in name_conf:
+ for rule_id, rule_conf in name_conf['rule'].items():
+ if single_rule_id and rule_id != single_rule_id:
+ continue
+
+ if 'disable' in rule_conf:
+ continue
+
+ row = [rule_id, rule_conf['action'], rule_conf['protocol'] if 'protocol' in rule_conf else 'all']
+ if rule_id in details:
+ rule_details = details[rule_id]
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ row.append(rule_details['conditions'])
+ rows.append(row)
+
+ if 'default_action' in name_conf and not single_rule_id:
+ row = ['default', name_conf['default_action'], 'all']
+ if 'default-action' in details:
+ rule_details = details['default-action']
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ rows.append(row)
+
+ if rows:
+ header = ['Rule', 'Action', 'Protocol', 'Packets', 'Bytes', 'Conditions']
+ print(tabulate.tabulate(rows, header) + '\n')
+
+def output_firewall_name_statistics(name, name_conf, ipv6=False, single_rule_id=None):
+ ip_str = 'IPv6' if ipv6 else 'IPv4'
+ print(f'\n---------------------------------\n{ip_str} Firewall "{name}"\n')
+
+ if name_conf['interface']:
+ print('Active on: {0}\n'.format(" ".join(name_conf['interface'])))
+
+ details = get_nftables_details(name, ipv6)
+ rows = []
+
+ if 'rule' in name_conf:
+ for rule_id, rule_conf in name_conf['rule'].items():
+ if single_rule_id and rule_id != single_rule_id:
+ continue
+
+ if 'disable' in rule_conf:
+ continue
+
+ source_addr = dict_search_args(rule_conf, 'source', 'address') or '0.0.0.0/0'
+ dest_addr = dict_search_args(rule_conf, 'destination', 'address') or '0.0.0.0/0'
+
+ row = [rule_id]
+ if rule_id in details:
+ rule_details = details[rule_id]
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ else:
+ row.append('0')
+ row.append('0')
+ row.append(rule_conf['action'])
+ row.append(source_addr)
+ row.append(dest_addr)
+ rows.append(row)
+
+ if 'default_action' in name_conf and not single_rule_id:
+ row = ['default']
+ if 'default-action' in details:
+ rule_details = details['default-action']
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ else:
+ row.append('0')
+ row.append('0')
+ row.append(name_conf['default_action'])
+ row.append('0.0.0.0/0') # Source
+ row.append('0.0.0.0/0') # Dest
+ rows.append(row)
+
+ if rows:
+ header = ['Rule', 'Packets', 'Bytes', 'Action', 'Source', 'Destination']
+ print(tabulate.tabulate(rows, header) + '\n')
+
+def show_firewall():
+ print('Rulesets Information')
+
+ conf = Config()
+ firewall = get_config_firewall(conf)
+
+ if not firewall:
+ return
+
+ if 'name' in firewall:
+ for name, name_conf in firewall['name'].items():
+ output_firewall_name(name, name_conf, ipv6=False)
+
+ if 'ipv6_name' in firewall:
+ for name, name_conf in firewall['ipv6_name'].items():
+ output_firewall_name(name, name_conf, ipv6=True)
+
+def show_firewall_name(name, ipv6=False):
+ print('Ruleset Information')
+
+ conf = Config()
+ firewall = get_config_firewall(conf, name, ipv6)
+ if firewall:
+ output_firewall_name(name, firewall, ipv6)
+
+def show_firewall_rule(name, rule_id, ipv6=False):
+ print('Rule Information')
+
+ conf = Config()
+ firewall = get_config_firewall(conf, name, ipv6)
+ if firewall:
+ output_firewall_name(name, firewall, ipv6, rule_id)
+
+def show_firewall_group(name=None):
+ conf = Config()
+ firewall = get_config_firewall(conf, interfaces=False)
+
+ if 'group' not in firewall:
+ return
+
+ def find_references(group_type, group_name):
+ out = []
+ for name_type in ['name', 'ipv6_name']:
+ if name_type not in firewall:
+ continue
+ for name, name_conf in firewall[name_type].items():
+ if 'rule' not in name_conf:
+ continue
+ for rule_id, rule_conf in name_conf['rule'].items():
+ source_group = dict_search_args(rule_conf, 'source', 'group', group_type)
+ dest_group = dict_search_args(rule_conf, 'destination', 'group', group_type)
+ if source_group and group_name == source_group:
+ out.append(f'{name}-{rule_id}')
+ elif dest_group and group_name == dest_group:
+ out.append(f'{name}-{rule_id}')
+ return out
+
+ header = ['Name', 'Type', 'References', 'Members']
+ rows = []
+
+ for group_type, group_type_conf in firewall['group'].items():
+ for group_name, group_conf in group_type_conf.items():
+ if name and name != group_name:
+ continue
+
+ references = find_references(group_type, group_name)
+ row = [group_name, group_type, '\n'.join(references) or 'N/A']
+ if 'address' in group_conf:
+ row.append("\n".join(sorted(group_conf['address'], key=ipaddress.ip_address)))
+ elif 'network' in group_conf:
+ row.append("\n".join(sorted(group_conf['network'], key=ipaddress.ip_network)))
+ elif 'mac_address' in group_conf:
+ row.append("\n".join(sorted(group_conf['mac_address'])))
+ elif 'port' in group_conf:
+ row.append("\n".join(sorted(group_conf['port'])))
+ else:
+ row.append('N/A')
+ rows.append(row)
+
+ if rows:
+ print('Firewall Groups\n')
+ print(tabulate.tabulate(rows, header))
+
+def show_summary():
+ print('Ruleset Summary')
+
+ conf = Config()
+ firewall = get_config_firewall(conf)
+
+ if not firewall:
+ return
+
+ header = ['Ruleset Name', 'Description', 'References']
+ v4_out = []
+ v6_out = []
+
+ if 'name' in firewall:
+ for name, name_conf in firewall['name'].items():
+ description = name_conf.get('description', '')
+ interfaces = ", ".join(name_conf['interface'])
+ v4_out.append([name, description, interfaces])
+
+ if 'ipv6_name' in firewall:
+ for name, name_conf in firewall['ipv6_name'].items():
+ description = name_conf.get('description', '')
+ interfaces = ", ".join(name_conf['interface'])
+ v6_out.append([name, description, interfaces or 'N/A'])
+
+ if v6_out:
+ print('\nIPv6 name:\n')
+ print(tabulate.tabulate(v6_out, header) + '\n')
+
+ if v4_out:
+ print('\nIPv4 name:\n')
+ print(tabulate.tabulate(v4_out, header) + '\n')
+
+ show_firewall_group()
+
+def show_statistics():
+ print('Rulesets Statistics')
+
+ conf = Config()
+ firewall = get_config_firewall(conf)
+
+ if not firewall:
+ return
+
+ if 'name' in firewall:
+ for name, name_conf in firewall['name'].items():
+ output_firewall_name_statistics(name, name_conf, ipv6=False)
+
+ if 'ipv6_name' in firewall:
+ for name, name_conf in firewall['ipv6_name'].items():
+ output_firewall_name_statistics(name, name_conf, ipv6=True)
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--action', help='Action', required=False)
+ parser.add_argument('--name', help='Firewall name', required=False, action='store', nargs='?', default='')
+ parser.add_argument('--rule', help='Firewall Rule ID', required=False)
+ parser.add_argument('--ipv6', help='IPv6 toggle', action='store_true')
+
+ args = parser.parse_args()
+
+ if args.action == 'show':
+ if not args.rule:
+ show_firewall_name(args.name, args.ipv6)
+ else:
+ show_firewall_rule(args.name, args.rule, args.ipv6)
+ elif args.action == 'show_all':
+ show_firewall()
+ elif args.action == 'show_group':
+ show_firewall_group(args.name)
+ elif args.action == 'show_statistics':
+ show_statistics()
+ elif args.action == 'show_summary':
+ show_summary()
diff --git a/src/op_mode/monitor_bandwidth_test.sh b/src/op_mode/monitor_bandwidth_test.sh
index 900223bca..a6ad0b42c 100755
--- a/src/op_mode/monitor_bandwidth_test.sh
+++ b/src/op_mode/monitor_bandwidth_test.sh
@@ -24,6 +24,9 @@ elif [[ $(dig $1 AAAA +short | grep -v '\.$' | wc -l) -gt 0 ]]; then
# Set address family to IPv6 when FQDN has at least one AAAA record
OPT="-V"
+else
+ # It's not IPv6, no option needed
+ OPT=""
fi
/usr/bin/iperf $OPT -c $1 $2
diff --git a/src/op_mode/policy_route.py b/src/op_mode/policy_route.py
new file mode 100755
index 000000000..5be40082f
--- /dev/null
+++ b/src/op_mode/policy_route.py
@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2021 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+import re
+import tabulate
+
+from vyos.config import Config
+from vyos.util import cmd
+from vyos.util import dict_search_args
+
+def get_policy_interfaces(conf, policy, name=None, ipv6=False):
+ interfaces = conf.get_config_dict(['interfaces'], key_mangling=('-', '_'),
+ get_first_key=True, no_tag_node_value_mangle=True)
+
+ routes = ['route', 'route6']
+
+ def parse_if(ifname, if_conf):
+ if 'policy' in if_conf:
+ for route in routes:
+ if route in if_conf['policy']:
+ route_name = if_conf['policy'][route]
+ name_str = f'({ifname},{route})'
+
+ if not name:
+ policy[route][route_name]['interface'].append(name_str)
+ elif not ipv6 and name == route_name:
+ policy['interface'].append(name_str)
+
+ for iftype in ['vif', 'vif_s', 'vif_c']:
+ if iftype in if_conf:
+ for vifname, vif_conf in if_conf[iftype].items():
+ parse_if(f'{ifname}.{vifname}', vif_conf)
+
+ for iftype, iftype_conf in interfaces.items():
+ for ifname, if_conf in iftype_conf.items():
+ parse_if(ifname, if_conf)
+
+def get_config_policy(conf, name=None, ipv6=False, interfaces=True):
+ config_path = ['policy']
+ if name:
+ config_path += ['route6' if ipv6 else 'route', name]
+
+ policy = conf.get_config_dict(config_path, key_mangling=('-', '_'),
+ get_first_key=True, no_tag_node_value_mangle=True)
+ if policy and interfaces:
+ if name:
+ policy['interface'] = []
+ else:
+ if 'route' in policy:
+ for route_name, route_conf in policy['route'].items():
+ route_conf['interface'] = []
+
+ if 'route6' in policy:
+ for route_name, route_conf in policy['route6'].items():
+ route_conf['interface'] = []
+
+ get_policy_interfaces(conf, policy, name, ipv6)
+
+ return policy
+
+def get_nftables_details(name, ipv6=False):
+ suffix = '6' if ipv6 else ''
+ command = f'sudo nft list chain ip{suffix} mangle VYOS_PBR{suffix}_{name}'
+ try:
+ results = cmd(command)
+ except:
+ return {}
+
+ out = {}
+ for line in results.split('\n'):
+ comment_search = re.search(rf'{name}[\- ](\d+|default-action)', line)
+ if not comment_search:
+ continue
+
+ rule = {}
+ rule_id = comment_search[1]
+ counter_search = re.search(r'counter packets (\d+) bytes (\d+)', line)
+ if counter_search:
+ rule['packets'] = counter_search[1]
+ rule['bytes'] = counter_search[2]
+
+ rule['conditions'] = re.sub(r'(\b(counter packets \d+ bytes \d+|drop|reject|return|log)\b|comment "[\w\-]+")', '', line).strip()
+ out[rule_id] = rule
+ return out
+
+def output_policy_route(name, route_conf, ipv6=False, single_rule_id=None):
+ ip_str = 'IPv6' if ipv6 else 'IPv4'
+ print(f'\n---------------------------------\n{ip_str} Policy Route "{name}"\n')
+
+ if route_conf['interface']:
+ print('Active on: {0}\n'.format(" ".join(route_conf['interface'])))
+
+ details = get_nftables_details(name, ipv6)
+ rows = []
+
+ if 'rule' in route_conf:
+ for rule_id, rule_conf in route_conf['rule'].items():
+ if single_rule_id and rule_id != single_rule_id:
+ continue
+
+ if 'disable' in rule_conf:
+ continue
+
+ action = rule_conf['action'] if 'action' in rule_conf else 'set'
+ protocol = rule_conf['protocol'] if 'protocol' in rule_conf else 'all'
+
+ row = [rule_id, action, protocol]
+ if rule_id in details:
+ rule_details = details[rule_id]
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ row.append(rule_details['conditions'])
+ rows.append(row)
+
+ if 'default_action' in route_conf and not single_rule_id:
+ row = ['default', route_conf['default_action'], 'all']
+ if 'default-action' in details:
+ rule_details = details['default-action']
+ row.append(rule_details.get('packets', 0))
+ row.append(rule_details.get('bytes', 0))
+ rows.append(row)
+
+ if rows:
+ header = ['Rule', 'Action', 'Protocol', 'Packets', 'Bytes', 'Conditions']
+ print(tabulate.tabulate(rows, header) + '\n')
+
+def show_policy(ipv6=False):
+ print('Ruleset Information')
+
+ conf = Config()
+ policy = get_config_policy(conf)
+
+ if not policy:
+ return
+
+ if not ipv6 and 'route' in policy:
+ for route, route_conf in policy['route'].items():
+ output_policy_route(route, route_conf, ipv6=False)
+
+ if ipv6 and 'route6' in policy:
+ for route, route_conf in policy['route6'].items():
+ output_policy_route(route, route_conf, ipv6=True)
+
+def show_policy_name(name, ipv6=False):
+ print('Ruleset Information')
+
+ conf = Config()
+ policy = get_config_policy(conf, name, ipv6)
+ if policy:
+ output_policy_route(name, policy, ipv6)
+
+def show_policy_rule(name, rule_id, ipv6=False):
+ print('Rule Information')
+
+ conf = Config()
+ policy = get_config_policy(conf, name, ipv6)
+ if policy:
+ output_policy_route(name, policy, ipv6, rule_id)
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--action', help='Action', required=False)
+ parser.add_argument('--name', help='Policy name', required=False, action='store', nargs='?', default='')
+ parser.add_argument('--rule', help='Policy Rule ID', required=False)
+ parser.add_argument('--ipv6', help='IPv6 toggle', action='store_true')
+
+ args = parser.parse_args()
+
+ if args.action == 'show':
+ if not args.rule:
+ show_policy_name(args.name, args.ipv6)
+ else:
+ show_policy_rule(args.name, args.rule, args.ipv6)
+ elif args.action == 'show_all':
+ show_policy(args.ipv6)
diff --git a/src/op_mode/show_virtual_server.py b/src/op_mode/show_virtual_server.py
new file mode 100755
index 000000000..377180dec
--- /dev/null
+++ b/src/op_mode/show_virtual_server.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2022 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
+
+from vyos.configquery import ConfigTreeQuery
+from vyos.util import call
+
+def is_configured():
+ """ Check if high-availability virtual-server is configured """
+ config = ConfigTreeQuery()
+ if not config.exists(['high-availability', 'virtual-server']):
+ return False
+ return True
+
+if __name__ == '__main__':
+
+ if is_configured() == False:
+ print('Virtual server not configured!')
+ exit(0)
+
+ call('sudo ipvsadm --list --numeric')
diff --git a/src/op_mode/vrrp.py b/src/op_mode/vrrp.py
index 2c1db20bf..dab146d28 100755
--- a/src/op_mode/vrrp.py
+++ b/src/op_mode/vrrp.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2018 VyOS maintainers and contributors
+# Copyright (C) 2018-2022 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
@@ -23,6 +23,7 @@ import tabulate
import vyos.util
+from vyos.configquery import ConfigTreeQuery
from vyos.ifconfig.vrrp import VRRP
from vyos.ifconfig.vrrp import VRRPError, VRRPNoData
@@ -35,7 +36,17 @@ group.add_argument("-d", "--data", action="store_true", help="Print detailed VRR
args = parser.parse_args()
+def is_configured():
+ """ Check if VRRP is configured """
+ config = ConfigTreeQuery()
+ if not config.exists(['high-availability', 'vrrp', 'group']):
+ return False
+ return True
+
# Exit early if VRRP is dead or not configured
+if is_configured() == False:
+ print('VRRP not configured!')
+ exit(0)
if not VRRP.is_running():
print('VRRP is not running')
sys.exit(0)
diff --git a/src/op_mode/zone_policy.py b/src/op_mode/zone_policy.py
new file mode 100755
index 000000000..7b43018c2
--- /dev/null
+++ b/src/op_mode/zone_policy.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2021 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+import tabulate
+
+from vyos.config import Config
+from vyos.util import dict_search_args
+
+def get_config_zone(conf, name=None):
+ config_path = ['zone-policy']
+ if name:
+ config_path += ['zone', name]
+
+ zone_policy = conf.get_config_dict(config_path, key_mangling=('-', '_'),
+ get_first_key=True, no_tag_node_value_mangle=True)
+ return zone_policy
+
+def output_zone_name(zone, zone_conf):
+ print(f'\n---------------------------------\nZone: "{zone}"\n')
+
+ interfaces = ', '.join(zone_conf['interface']) if 'interface' in zone_conf else ''
+ if 'local_zone' in zone_conf:
+ interfaces = 'LOCAL'
+
+ print(f'Interfaces: {interfaces}\n')
+
+ header = ['From Zone', 'Firewall']
+ rows = []
+
+ if 'from' in zone_conf:
+ for from_name, from_conf in zone_conf['from'].items():
+ row = [from_name]
+ v4_name = dict_search_args(from_conf, 'firewall', 'name')
+ v6_name = dict_search_args(from_conf, 'firewall', 'ipv6_name')
+
+ if v4_name:
+ rows.append(row + [v4_name])
+
+ if v6_name:
+ rows.append(row + [f'{v6_name} [IPv6]'])
+
+ if rows:
+ print('From Zones:\n')
+ print(tabulate.tabulate(rows, header))
+
+def show_zone_policy(zone):
+ conf = Config()
+ zone_policy = get_config_zone(conf, zone)
+
+ if not zone_policy:
+ return
+
+ if 'zone' in zone_policy:
+ for zone, zone_conf in zone_policy['zone'].items():
+ output_zone_name(zone, zone_conf)
+ elif zone:
+ output_zone_name(zone, zone_policy)
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--action', help='Action', required=False)
+ parser.add_argument('--name', help='Zone name', required=False, action='store', nargs='?', default='')
+
+ args = parser.parse_args()
+
+ if args.action == 'show':
+ show_zone_policy(args.name)