summaryrefslogtreecommitdiff
path: root/src/op_mode
diff options
context:
space:
mode:
Diffstat (limited to 'src/op_mode')
-rwxr-xr-xsrc/op_mode/conntrack.py23
-rwxr-xr-xsrc/op_mode/ipsec.py116
-rwxr-xr-xsrc/op_mode/nat.py136
-rwxr-xr-xsrc/op_mode/show_nat66_rules.py102
4 files changed, 260 insertions, 117 deletions
diff --git a/src/op_mode/conntrack.py b/src/op_mode/conntrack.py
index 036226418..b27aa6060 100755
--- a/src/op_mode/conntrack.py
+++ b/src/op_mode/conntrack.py
@@ -51,6 +51,21 @@ def _get_raw_data(family):
return _xml_to_dict(xml)
+def _get_raw_statistics():
+ entries = []
+ data = cmd('sudo conntrack -S')
+ data = data.replace(' \t', '').split('\n')
+ for entry in data:
+ entries.append(entry.split())
+ return entries
+
+
+def get_formatted_statistics(entries):
+ headers = ["CPU", "Found", "Invalid", "Insert", "Insert fail", "Drop", "Early drop", "Errors", "Search restart"]
+ output = tabulate(entries, headers, numalign="left")
+ return output
+
+
def get_formatted_output(dict_data):
"""
:param xml:
@@ -111,6 +126,14 @@ def show(raw: bool, family: str):
return get_formatted_output(conntrack_data)
+def show_statistics(raw: bool):
+ conntrack_statistics = _get_raw_statistics()
+ if raw:
+ return conntrack_statistics
+ else:
+ return get_formatted_statistics(conntrack_statistics)
+
+
if __name__ == '__main__':
try:
res = vyos.opmode.run(sys.modules[__name__])
diff --git a/src/op_mode/ipsec.py b/src/op_mode/ipsec.py
index 49c8e6142..a4d1b4cb1 100755
--- a/src/op_mode/ipsec.py
+++ b/src/op_mode/ipsec.py
@@ -16,13 +16,122 @@
import re
import sys
+
+from collections import OrderedDict
+from hurry import filesize
+from re import split as re_split
+from tabulate import tabulate
+
from vyos.util import call
+from vyos.util import convert_data
+from vyos.util import seconds_to_human
+
import vyos.opmode
SWANCTL_CONF = '/etc/swanctl/swanctl.conf'
+def _convert(text):
+ return int(text) if text.isdigit() else text.lower()
+
+
+def _alphanum_key(key):
+ return [_convert(c) for c in re_split('([0-9]+)', str(key))]
+
+
+def _get_vici_sas():
+ from vici import Session as vici_session
+
+ session = vici_session()
+ sas = list(session.list_sas())
+ return sas
+
+
+def _get_raw_data_sas():
+ get_sas = _get_vici_sas()
+ sas = convert_data(get_sas)
+ return sas
+
+
+def _get_formatted_output_sas(sas):
+ sa_data = []
+ for sa in sas:
+ for parent_sa in sa.values():
+ # create an item for each child-sa
+ for child_sa in parent_sa.get('child-sas', {}).values():
+ # prepare a list for output data
+ sa_out_name = sa_out_state = sa_out_uptime = sa_out_bytes = sa_out_packets = sa_out_remote_addr = sa_out_remote_id = sa_out_proposal = 'N/A'
+
+ # collect raw data
+ sa_name = child_sa.get('name')
+ sa_state = child_sa.get('state')
+ sa_uptime = child_sa.get('install-time')
+ sa_bytes_in = child_sa.get('bytes-in')
+ sa_bytes_out = child_sa.get('bytes-out')
+ sa_packets_in = child_sa.get('packets-in')
+ sa_packets_out = child_sa.get('packets-out')
+ sa_remote_addr = parent_sa.get('remote-host')
+ sa_remote_id = parent_sa.get('remote-id')
+ sa_proposal_encr_alg = child_sa.get('encr-alg')
+ sa_proposal_integ_alg = child_sa.get('integ-alg')
+ sa_proposal_encr_keysize = child_sa.get('encr-keysize')
+ sa_proposal_dh_group = child_sa.get('dh-group')
+
+ # format data to display
+ if sa_name:
+ sa_out_name = sa_name
+ if sa_state:
+ if sa_state == 'INSTALLED':
+ sa_out_state = 'up'
+ else:
+ sa_out_state = 'down'
+ if sa_uptime:
+ sa_out_uptime = seconds_to_human(sa_uptime)
+ if sa_bytes_in and sa_bytes_out:
+ bytes_in = filesize.size(int(sa_bytes_in))
+ bytes_out = filesize.size(int(sa_bytes_out))
+ sa_out_bytes = f'{bytes_in}/{bytes_out}'
+ if sa_packets_in and sa_packets_out:
+ packets_in = filesize.size(int(sa_packets_in),
+ system=filesize.si)
+ packets_out = filesize.size(int(sa_packets_out),
+ system=filesize.si)
+ packets_str = f'{packets_in}/{packets_out}'
+ sa_out_packets = re.sub(r'B', r'', packets_str)
+ if sa_remote_addr:
+ sa_out_remote_addr = sa_remote_addr
+ if sa_remote_id:
+ sa_out_remote_id = sa_remote_id
+ # format proposal
+ if sa_proposal_encr_alg:
+ sa_out_proposal = sa_proposal_encr_alg
+ if sa_proposal_encr_keysize:
+ sa_proposal_encr_keysize_str = sa_proposal_encr_keysize
+ sa_out_proposal = f'{sa_out_proposal}_{sa_proposal_encr_keysize_str}'
+ if sa_proposal_integ_alg:
+ sa_proposal_integ_alg_str = sa_proposal_integ_alg
+ sa_out_proposal = f'{sa_out_proposal}/{sa_proposal_integ_alg_str}'
+ if sa_proposal_dh_group:
+ sa_proposal_dh_group_str = sa_proposal_dh_group
+ sa_out_proposal = f'{sa_out_proposal}/{sa_proposal_dh_group_str}'
+
+ # add a new item to output data
+ sa_data.append([
+ sa_out_name, sa_out_state, sa_out_uptime, sa_out_bytes,
+ sa_out_packets, sa_out_remote_addr, sa_out_remote_id,
+ sa_out_proposal
+ ])
+
+ headers = [
+ "Connection", "State", "Uptime", "Bytes In/Out", "Packets In/Out",
+ "Remote address", "Remote ID", "Proposal"
+ ]
+ sa_data = sorted(sa_data, key=_alphanum_key)
+ output = tabulate(sa_data, headers)
+ return output
+
+
def get_peer_connections(peer, tunnel, return_all = False):
peer = peer.replace(':', '-')
search = rf'^[\s]*(peer_{peer}_(tunnel_[\d]+|vti)).*'
@@ -61,6 +170,13 @@ def reset_peer(peer: str, tunnel:str):
print('Peer reset result: ' + ('success' if result else 'failed'))
+def show_sa(raw: bool):
+ sa_data = _get_raw_data_sas()
+ if raw:
+ return sa_data
+ return _get_formatted_output_sas(sa_data)
+
+
if __name__ == '__main__':
try:
res = vyos.opmode.run(sys.modules[__name__])
diff --git a/src/op_mode/nat.py b/src/op_mode/nat.py
index 12fc4c782..1339d5b92 100755
--- a/src/op_mode/nat.py
+++ b/src/op_mode/nat.py
@@ -17,6 +17,7 @@
import jmespath
import json
import sys
+import xmltodict
from sys import exit
from tabulate import tabulate
@@ -27,7 +28,30 @@ from vyos.util import dict_search
import vyos.opmode
-def _get_json_data(direction):
+def _get_xml_translation(direction, family):
+ """
+ Get conntrack XML output --src-nat|--dst-nat
+ """
+ if direction == 'source':
+ opt = '--src-nat'
+ if direction == 'destination':
+ opt = '--dst-nat'
+ return cmd(f'sudo conntrack --dump --family {family} {opt} --output xml')
+
+
+def _xml_to_dict(xml):
+ """
+ Convert XML to dictionary
+ Return: dictionary
+ """
+ parse = xmltodict.parse(xml, attr_prefix='')
+ # If only one conntrack entry we must change dict
+ if 'meta' in parse['conntrack']['flow']:
+ return dict(conntrack={'flow': [parse['conntrack']['flow']]})
+ return parse
+
+
+def _get_json_data(direction, family):
"""
Get NAT format JSON
"""
@@ -35,14 +59,15 @@ def _get_json_data(direction):
chain = 'POSTROUTING'
if direction == 'destination':
chain = 'PREROUTING'
- return cmd(f'sudo nft --json list chain ip nat {chain}')
+ family = 'ip6' if family == 'inet6' else 'ip'
+ return cmd(f'sudo nft --json list chain {family} nat {chain}')
-def _get_raw_data_rules(direction):
+def _get_raw_data_rules(direction, family):
"""Get interested rules
:returns dict
"""
- data = _get_json_data(direction)
+ data = _get_json_data(direction, family)
data_dict = json.loads(data)
rules = []
for rule in data_dict['nftables']:
@@ -51,10 +76,28 @@ def _get_raw_data_rules(direction):
return rules
-def _get_formatted_output_rules(data, direction):
+def _get_raw_translation(direction, family):
+ """
+ Return: dictionary
+ """
+ xml = _get_xml_translation(direction, family)
+ if len(xml) == 0:
+ output = {'conntrack':
+ {
+ 'error': True,
+ 'reason': 'entries not found'
+ }
+ }
+ return output
+ return _xml_to_dict(xml)
+
+
+def _get_formatted_output_rules(data, direction, family):
# Add default values before loop
sport, dport, proto = 'any', 'any', 'any'
- saddr, daddr = '0.0.0.0/0', '0.0.0.0/0'
+ saddr = '::/0' if family == 'inet6' else '0.0.0.0/0'
+ daddr = '::/0' if family == 'inet6' else '0.0.0.0/0'
+
data_entries = []
for rule in data:
if 'comment' in rule['rule']:
@@ -69,11 +112,13 @@ def _get_formatted_output_rules(data, direction):
if 'prefix' in match['right'] or 'set' in match['right']:
# Merge dict src/dst l3_l4 parameters
my_dict = {**match['left']['payload'], **match['right']}
+ my_dict['op'] = match['op']
+ op = '!' if my_dict.get('op') == '!=' else ''
proto = my_dict.get('protocol').upper()
if my_dict['field'] == 'saddr':
- saddr = f'{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
+ saddr = f'{op}{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
elif my_dict['field'] == 'daddr':
- daddr = f'{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
+ daddr = f'{op}{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
elif my_dict['field'] == 'sport':
# Port range or single port
if jmespath.search('set[*].range', my_dict):
@@ -96,8 +141,8 @@ def _get_formatted_output_rules(data, direction):
if jmespath.search('left.payload.field', match) == 'daddr':
daddr = match.get('right')
else:
- saddr = '0.0.0.0/0'
- daddr = '0.0.0.0/0'
+ saddr = '::/0' if family == 'inet6' else '0.0.0.0/0'
+ daddr = '::/0' if family == 'inet6' else '0.0.0.0/0'
sport = 'any'
dport = 'any'
proto = 'any'
@@ -175,22 +220,83 @@ def _get_formatted_output_statistics(data, direction):
return output
-def show_rules(raw: bool, direction: str):
- nat_rules = _get_raw_data_rules(direction)
+def _get_formatted_translation(dict_data, nat_direction, family):
+ data_entries = []
+ if 'error' in dict_data['conntrack']:
+ return 'Entries not found'
+ for entry in dict_data['conntrack']['flow']:
+ orig_src, orig_dst, orig_sport, orig_dport = {}, {}, {}, {}
+ reply_src, reply_dst, reply_sport, reply_dport = {}, {}, {}, {}
+ proto = {}
+ for meta in entry['meta']:
+ direction = meta['direction']
+ if direction in ['original']:
+ if 'layer3' in meta:
+ orig_src = meta['layer3']['src']
+ orig_dst = meta['layer3']['dst']
+ if 'layer4' in meta:
+ if meta.get('layer4').get('sport'):
+ orig_sport = meta['layer4']['sport']
+ if meta.get('layer4').get('dport'):
+ orig_dport = meta['layer4']['dport']
+ proto = meta['layer4']['protoname']
+ if direction in ['reply']:
+ if 'layer3' in meta:
+ reply_src = meta['layer3']['src']
+ reply_dst = meta['layer3']['dst']
+ if 'layer4' in meta:
+ if meta.get('layer4').get('sport'):
+ reply_sport = meta['layer4']['sport']
+ if meta.get('layer4').get('dport'):
+ reply_dport = meta['layer4']['dport']
+ proto = meta['layer4']['protoname']
+ if direction == 'independent':
+ conn_id = meta['id']
+ timeout = meta['timeout']
+ orig_src = f'{orig_src}:{orig_sport}' if orig_sport else orig_src
+ orig_dst = f'{orig_dst}:{orig_dport}' if orig_dport else orig_dst
+ reply_src = f'{reply_src}:{reply_sport}' if reply_sport else reply_src
+ reply_dst = f'{reply_dst}:{reply_dport}' if reply_dport else reply_dst
+ state = meta['state'] if 'state' in meta else ''
+ mark = meta['mark']
+ zone = meta['zone'] if 'zone' in meta else ''
+ if nat_direction == 'source':
+ data_entries.append(
+ [orig_src, reply_dst, proto, timeout, mark, zone])
+ elif nat_direction == 'destination':
+ data_entries.append(
+ [orig_dst, reply_src, proto, timeout, mark, zone])
+
+ headers = ["Pre-NAT", "Post-NAT", "Proto", "Timeout", "Mark", "Zone"]
+ output = tabulate(data_entries, headers, numalign="left")
+ return output
+
+
+def show_rules(raw: bool, direction: str, family: str):
+ nat_rules = _get_raw_data_rules(direction, family)
if raw:
return nat_rules
else:
- return _get_formatted_output_rules(nat_rules, direction)
+ return _get_formatted_output_rules(nat_rules, direction, family)
-def show_statistics(raw: bool, direction: str):
- nat_statistics = _get_raw_data_rules(direction)
+def show_statistics(raw: bool, direction: str, family: str):
+ nat_statistics = _get_raw_data_rules(direction, family)
if raw:
return nat_statistics
else:
return _get_formatted_output_statistics(nat_statistics, direction)
+def show_translations(raw: bool, direction: str, family: str):
+ family = 'ipv6' if family == 'inet6' else 'ipv4'
+ nat_translation = _get_raw_translation(direction, family)
+ if raw:
+ return nat_translation
+ else:
+ return _get_formatted_translation(nat_translation, direction, family)
+
+
if __name__ == '__main__':
try:
res = vyos.opmode.run(sys.modules[__name__])
diff --git a/src/op_mode/show_nat66_rules.py b/src/op_mode/show_nat66_rules.py
deleted file mode 100755
index 967ec9d37..000000000
--- a/src/op_mode/show_nat66_rules.py
+++ /dev/null
@@ -1,102 +0,0 @@
-#!/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 jmespath
-import json
-
-from argparse import ArgumentParser
-from jinja2 import Template
-from sys import exit
-from vyos.util import cmd
-from vyos.util import dict_search
-
-parser = ArgumentParser()
-group = parser.add_mutually_exclusive_group()
-group.add_argument("--source", help="Show statistics for configured source NAT rules", action="store_true")
-group.add_argument("--destination", help="Show statistics for configured destination NAT rules", action="store_true")
-args = parser.parse_args()
-
-if args.source or args.destination:
- tmp = cmd('sudo nft -j list table ip6 nat')
- tmp = json.loads(tmp)
-
- format_nat66_rule = '{0: <10} {1: <50} {2: <50} {3: <10}'
- print(format_nat66_rule.format("Rule", "Source" if args.source else "Destination", "Translation", "Outbound Interface" if args.source else "Inbound Interface"))
- print(format_nat66_rule.format("----", "------" if args.source else "-----------", "-----------", "------------------" if args.source else "-----------------"))
-
- data_json = jmespath.search('nftables[?rule].rule[?chain]', tmp)
- for idx in range(0, len(data_json)):
- data = data_json[idx]
-
- # The following key values must exist
- # When the rule JSON does not have some keys, this is not a rule we can work with
- continue_rule = False
- for key in ['comment', 'chain', 'expr']:
- if key not in data:
- continue_rule = True
- continue
- if continue_rule:
- continue
-
- comment = data['comment']
-
- # Check the annotation to see if the annotation format is created by VYOS
- continue_rule = True
- for comment_prefix in ['SRC-NAT66-', 'DST-NAT66-']:
- if comment_prefix in comment:
- continue_rule = False
- if continue_rule:
- continue
-
- # When log is detected from the second index of expr, then this rule should be ignored
- if 'log' in data['expr'][2]:
- continue
-
- rule = comment.replace('SRC-NAT66-','')
- rule = rule.replace('DST-NAT66-','')
- chain = data['chain']
- if not ((args.source and chain == 'POSTROUTING') or (not args.source and chain == 'PREROUTING')):
- continue
- interface = dict_search('match.right', data['expr'][0])
- srcdest = dict_search('match.right.prefix.addr', data['expr'][2])
- if srcdest:
- addr_tmp = dict_search('match.right.prefix.len', data['expr'][2])
- if addr_tmp:
- srcdest = srcdest + '/' + str(addr_tmp)
- else:
- srcdest = dict_search('match.right', data['expr'][2])
-
- tran_addr_json = dict_search('snat.addr' if args.source else 'dnat.addr', data['expr'][3])
- if tran_addr_json:
- if isinstance(srcdest_json,str):
- tran_addr = tran_addr_json
-
- if 'prefix' in tran_addr_json:
- addr_tmp = dict_search('snat.addr.prefix.addr' if args.source else 'dnat.addr.prefix.addr', data['expr'][3])
- len_tmp = dict_search('snat.addr.prefix.len' if args.source else 'dnat.addr.prefix.len', data['expr'][3])
- if addr_tmp:
- tran_addr = addr_tmp + '/' + str(len_tmp)
- else:
- if 'masquerade' in data['expr'][3]:
- tran_addr = 'masquerade'
-
- print(format_nat66_rule.format(rule, srcdest, tran_addr, interface))
-
- exit(0)
-else:
- parser.print_help()
- exit(1)
-
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-30 09:17:47 +0000