diff options
Diffstat (limited to 'src/op_mode')
| -rwxr-xr-x | src/op_mode/conntrack_sync.py | 25 | ||||
| -rwxr-xr-x | src/op_mode/ikev2_profile_generator.py | 57 | ||||
| -rwxr-xr-x | src/op_mode/lldp.py | 23 | ||||
| -rw-r--r-- | src/op_mode/tcpdump.py | 165 |
4 files changed, 238 insertions, 32 deletions
diff --git a/src/op_mode/conntrack_sync.py b/src/op_mode/conntrack_sync.py index 6c86ff492..f3b09b452 100755 --- a/src/op_mode/conntrack_sync.py +++ b/src/op_mode/conntrack_sync.py @@ -19,6 +19,8 @@ import sys import syslog import xmltodict +from tabulate import tabulate + import vyos.opmode from vyos.configquery import CliShellApiConfigQuery @@ -27,7 +29,6 @@ from vyos.utils.commit import commit_in_progress from vyos.utils.process import call from vyos.utils.process import cmd from vyos.utils.process import run -from vyos.template import render_to_string conntrackd_bin = '/usr/sbin/conntrackd' conntrackd_config = '/run/conntrackd/conntrackd.conf' @@ -59,6 +60,26 @@ def flush_cache(direction): if tmp > 0: raise vyos.opmode.Error('Failed to clear {direction} cache') +def get_formatted_output(data): + data_entries = [] + for parsed in data: + for meta in parsed.get('flow', {}).get('meta', []): + direction = meta['@direction'] + if direction == 'original': + src = meta['layer3']['src'] + dst = meta['layer3']['dst'] + sport = meta['layer4'].get('sport') + dport = meta['layer4'].get('dport') + protocol = meta['layer4'].get('@protoname') + orig_src = f'{src}:{sport}' if sport else src + orig_dst = f'{dst}:{dport}' if dport else dst + + data_entries.append([orig_src, orig_dst, protocol]) + + headers = ["Source", "Destination", "Protocol"] + output = tabulate(data_entries, headers, tablefmt="simple") + return output + def from_xml(raw, xml): out = [] for line in xml.splitlines(): @@ -70,7 +91,7 @@ def from_xml(raw, xml): if raw: return out else: - return render_to_string('conntrackd/conntrackd.op-mode.j2', {'data' : out}) + return get_formatted_output(out) def restart(): is_configured() diff --git a/src/op_mode/ikev2_profile_generator.py b/src/op_mode/ikev2_profile_generator.py index 169a15840..b193d8109 100755 --- a/src/op_mode/ikev2_profile_generator.py +++ b/src/op_mode/ikev2_profile_generator.py @@ -21,6 +21,7 @@ from socket import getfqdn from cryptography.x509.oid import NameOID from vyos.configquery import ConfigTreeQuery +from vyos.config import config_dict_mangle_acme from vyos.pki import CERT_BEGIN from vyos.pki import CERT_END from vyos.pki import find_chain @@ -123,6 +124,8 @@ pki_base = ['pki'] conf = ConfigTreeQuery() if not conf.exists(config_base): exit('IPsec remote-access is not configured!') +if not conf.exists(pki_base): + exit('PKI is not configured!') profile_name = 'VyOS IKEv2 Profile' if args.profile: @@ -147,30 +150,36 @@ tmp = getfqdn().split('.') tmp = reversed(tmp) data['rfqdn'] = '.'.join(tmp) -pki = conf.get_config_dict(pki_base, get_first_key=True) -cert_name = data['authentication']['x509']['certificate'] - -cert_data = load_certificate(pki['certificate'][cert_name]['certificate']) -data['cert_common_name'] = cert_data.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value -data['ca_common_name'] = cert_data.issuer.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value -data['ca_certificates'] = [] - -loaded_ca_certs = {load_certificate(c['certificate']) - for c in pki['ca'].values()} if 'ca' in pki else {} - -for ca_name in data['authentication']['x509']['ca_certificate']: - loaded_ca_cert = load_certificate(pki['ca'][ca_name]['certificate']) - ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs) - for ca in ca_full_chain: - tmp = { - 'ca_name' : ca.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value, - 'ca_chain' : encode_certificate(ca).replace(CERT_BEGIN, '').replace(CERT_END, '').replace('\n', ''), - } - data['ca_certificates'].append(tmp) - -# Remove duplicate list entries for CA certificates, as they are added by their common name -# https://stackoverflow.com/a/9427216 -data['ca_certificates'] = [dict(t) for t in {tuple(d.items()) for d in data['ca_certificates']}] +if args.os == 'ios': + pki = conf.get_config_dict(pki_base, get_first_key=True) + if 'certificate' in pki: + for certificate in pki['certificate']: + pki['certificate'][certificate] = config_dict_mangle_acme(certificate, pki['certificate'][certificate]) + + cert_name = data['authentication']['x509']['certificate'] + + + cert_data = load_certificate(pki['certificate'][cert_name]['certificate']) + data['cert_common_name'] = cert_data.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value + data['ca_common_name'] = cert_data.issuer.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value + data['ca_certificates'] = [] + + loaded_ca_certs = {load_certificate(c['certificate']) + for c in pki['ca'].values()} if 'ca' in pki else {} + + for ca_name in data['authentication']['x509']['ca_certificate']: + loaded_ca_cert = load_certificate(pki['ca'][ca_name]['certificate']) + ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs) + for ca in ca_full_chain: + tmp = { + 'ca_name' : ca.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value, + 'ca_chain' : encode_certificate(ca).replace(CERT_BEGIN, '').replace(CERT_END, '').replace('\n', ''), + } + data['ca_certificates'].append(tmp) + + # Remove duplicate list entries for CA certificates, as they are added by their common name + # https://stackoverflow.com/a/9427216 + data['ca_certificates'] = [dict(t) for t in {tuple(d.items()) for d in data['ca_certificates']}] esp_proposals = conf.get_config_dict(ipsec_base + ['esp-group', data['esp_group'], 'proposal'], key_mangling=('-', '_'), get_first_key=True) diff --git a/src/op_mode/lldp.py b/src/op_mode/lldp.py index 58cfce443..fac622b81 100755 --- a/src/op_mode/lldp.py +++ b/src/op_mode/lldp.py @@ -120,7 +120,12 @@ def _get_formatted_output(raw_data): tmp.append('') # Remote interface - interface = jmespath.search('port.descr', values) + interface = None + if jmespath.search('port.id.type', values) == 'ifname': + # Remote peer has explicitly returned the interface name as the PortID + interface = jmespath.search('port.id.value', values) + if not interface: + interface = jmespath.search('port.descr', values) if not interface: interface = jmespath.search('port.id.value', values) if not interface: @@ -136,11 +141,17 @@ def _get_formatted_output(raw_data): @_verify def show_neighbors(raw: bool, interface: typing.Optional[str], detail: typing.Optional[bool]): - lldp_data = _get_raw_data(interface=interface, detail=detail) - if raw: - return lldp_data - else: - return _get_formatted_output(lldp_data) + if raw or not detail: + lldp_data = _get_raw_data(interface=interface, detail=detail) + if raw: + return lldp_data + else: + return _get_formatted_output(lldp_data) + else: # non-raw, detail + tmp = 'lldpcli -f text show neighbors details' + if interface: + tmp += f' ports {interface}' + return cmd(tmp) if __name__ == "__main__": try: diff --git a/src/op_mode/tcpdump.py b/src/op_mode/tcpdump.py new file mode 100644 index 000000000..607b59603 --- /dev/null +++ b/src/op_mode/tcpdump.py @@ -0,0 +1,165 @@ +#! /usr/bin/env python3 + +# Copyright (C) 2024 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 sys + +from vyos.utils.process import call + +options = { + 'dump': { + 'cmd': '{command} -A', + 'type': 'noarg', + 'help': 'Print each packet (minus its link level header) in ASCII.' + }, + 'hexdump': { + 'cmd': '{command} -X', + 'type': 'noarg', + 'help': 'Print each packet (minus its link level header) in both hex and ASCII.' + }, + 'filter': { + 'cmd': '{command} \'{value}\'', + 'type': '<pcap-filter>', + 'help': 'Match traffic for capture and display with a pcap-filter expression.' + }, + 'numeric': { + 'cmd': '{command} -nn', + 'type': 'noarg', + 'help': 'Do not attempt to resolve addresses, protocols or services to names.' + }, + 'save': { + 'cmd': '{command} -w {value}', + 'type': '<file>', + 'help': 'Write captured raw packets to <file> rather than parsing or printing them out.' + }, + 'verbose': { + 'cmd': '{command} -vvv -ne', + 'type': 'noarg', + 'help': 'Parse packets with increased detail output, including link-level headers and extended decoding protocol sanity checks.' + }, +} + +tcpdump = 'sudo /usr/bin/tcpdump' + +class List(list): + def first(self): + return self.pop(0) if self else '' + + def last(self): + return self.pop() if self else '' + + def prepend(self, value): + self.insert(0, value) + + +def completion_failure(option: str) -> None: + """ + Shows failure message after TAB when option is wrong + :param option: failure option + :type str: + """ + sys.stderr.write('\n\n Invalid option: {}\n\n'.format(option)) + sys.stdout.write('<nocomps>') + sys.exit(1) + + +def expansion_failure(option, completions): + reason = 'Ambiguous' if completions else 'Invalid' + sys.stderr.write( + '\n\n {} command: {} [{}]\n\n'.format(reason, ' '.join(sys.argv), + option)) + if completions: + sys.stderr.write(' Possible completions:\n ') + sys.stderr.write('\n '.join(completions)) + sys.stderr.write('\n') + sys.stdout.write('<nocomps>') + sys.exit(1) + + +def complete(prefix): + return [o for o in options if o.startswith(prefix)] + + +def convert(command, args): + while args: + shortname = args.first() + longnames = complete(shortname) + if len(longnames) != 1: + expansion_failure(shortname, longnames) + longname = longnames[0] + if options[longname]['type'] == 'noarg': + command = options[longname]['cmd'].format( + command=command, value='') + elif not args: + sys.exit(f'monitor traffic: missing argument for {longname} option') + else: + command = options[longname]['cmd'].format( + command=command, value=args.first()) + return command + + +if __name__ == '__main__': + args = List(sys.argv[1:]) + ifname = args.first() + + # Slightly simplified & tweaked version of the code from mtr.py - it may be + # worthwhile to combine and centralise this in a common module. + if ifname == '--get-options-nested': + args.first() # pop monitor + args.first() # pop traffic + args.first() # pop interface + args.first() # pop <ifname> + usedoptionslist = [] + while args: + option = args.first() # pop option + matched = complete(option) # get option parameters + usedoptionslist.append(option) # list of used options + # Select options + if not args: + # remove from Possible completions used options + for o in usedoptionslist: + if o in matched: + matched.remove(o) + if not matched: + sys.stdout.write('<nocomps>') + else: + sys.stdout.write(' '.join(matched)) + sys.exit(0) + + if len(matched) > 1: + sys.stdout.write(' '.join(matched)) + sys.exit(0) + # If option doesn't have value + if matched: + if options[matched[0]]['type'] == 'noarg': + continue + else: + # Unexpected option + completion_failure(option) + + value = args.first() # pop option's value + if not args: + matched = complete(option) + helplines = options[matched[0]]['type'] + # Run helpfunction to get list of possible values + if 'helpfunction' in options[matched[0]]: + result = options[matched[0]]['helpfunction']() + if result: + helplines = '\n' + ' '.join(result) + sys.stdout.write(helplines) + sys.exit(0) + + command = convert(tcpdump, args) + call(f'{command} -i {ifname}') |