diff options
Diffstat (limited to 'src/op_mode')
31 files changed, 2803 insertions, 554 deletions
diff --git a/src/op_mode/bridge.py b/src/op_mode/bridge.py index 185db4f20..412a4eba8 100755 --- a/src/op_mode/bridge.py +++ b/src/op_mode/bridge.py @@ -56,6 +56,13 @@ def _get_raw_data_vlan(tunnel:bool=False): data_dict = json.loads(json_data) return data_dict +def _get_raw_data_vni() -> dict: + """ + :returns dict + """ + json_data = cmd(f'bridge --json vni show') + data_dict = json.loads(json_data) + return data_dict def _get_raw_data_fdb(bridge): """Get MAC-address for the bridge brX @@ -165,6 +172,22 @@ def _get_formatted_output_vlan_tunnel(data): output = tabulate(data_entries, headers) return output +def _get_formatted_output_vni(data): + data_entries = [] + for entry in data: + interface = entry.get('ifname') + vlans = entry.get('vnis') + for vlan_entry in vlans: + vlan = vlan_entry.get('vni') + if vlan_entry.get('vniEnd'): + vlan_end = vlan_entry.get('vniEnd') + vlan = f'{vlan}-{vlan_end}' + data_entries.append([interface, vlan]) + + headers = ["Interface", "VNI"] + output = tabulate(data_entries, headers) + return output + def _get_formatted_output_fdb(data): data_entries = [] for entry in data: @@ -228,6 +251,12 @@ def show_vlan(raw: bool, tunnel: typing.Optional[bool]): else: return _get_formatted_output_vlan(bridge_vlan) +def show_vni(raw: bool): + bridge_vni = _get_raw_data_vni() + if raw: + return bridge_vni + else: + return _get_formatted_output_vni(bridge_vni) def show_fdb(raw: bool, interface: str): fdb_data = _get_raw_data_fdb(interface) diff --git a/src/op_mode/clear_dhcp_lease.py b/src/op_mode/clear_dhcp_lease.py index f372d3af0..7d4b47104 100755 --- a/src/op_mode/clear_dhcp_lease.py +++ b/src/op_mode/clear_dhcp_lease.py @@ -1,20 +1,34 @@ #!/usr/bin/env python3 +# +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library 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 +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. import argparse import re -from isc_dhcp_leases import Lease -from isc_dhcp_leases import IscDhcpLeases - from vyos.configquery import ConfigTreeQuery +from vyos.kea import kea_parse_leases from vyos.utils.io import ask_yes_no from vyos.utils.process import call from vyos.utils.commit import commit_in_progress +# TODO: Update to use Kea control socket command "lease4-del" config = ConfigTreeQuery() base = ['service', 'dhcp-server'] -lease_file = '/config/dhcpd.leases' +lease_file = '/config/dhcp/dhcp4-leases.csv' def del_lease_ip(address): @@ -25,8 +39,7 @@ def del_lease_ip(address): """ with open(lease_file, encoding='utf-8') as f: data = f.read().rstrip() - lease_config_ip = '{(?P<config>[\s\S]+?)\n}' - pattern = rf"lease {address} {lease_config_ip}" + pattern = rf"^{address},[^\n]+\n" # Delete lease for ip block data = re.sub(pattern, '', data) @@ -38,15 +51,12 @@ def is_ip_in_leases(address): """ Return True if address found in the lease file """ - leases = IscDhcpLeases(lease_file) - lease_ips = [] - for lease in leases.get(): - lease_ips.append(lease.ip) - if address not in lease_ips: - print(f'Address "{address}" not found in "{lease_file}"') - return False - return True - + leases = kea_parse_leases(lease_file) + for lease in leases: + if address == lease['address']: + return True + print(f'Address "{address}" not found in "{lease_file}"') + return False if not config.exists(base): print('DHCP-server not configured!') @@ -75,4 +85,4 @@ if __name__ == '__main__': exit(1) else: del_lease_ip(address) - call('systemctl restart isc-dhcp-server.service') + call('systemctl restart kea-dhcp4-server.service') diff --git a/src/op_mode/connect_disconnect.py b/src/op_mode/connect_disconnect.py index 89f929be7..10034e499 100755 --- a/src/op_mode/connect_disconnect.py +++ b/src/op_mode/connect_disconnect.py @@ -55,7 +55,7 @@ def connect(interface): if is_wwan_connected(interface): print(f'Interface {interface}: already connected!') else: - call(f'VYOS_TAGNODE_VALUE={interface} /usr/libexec/vyos/conf_mode/interfaces-wwan.py') + call(f'VYOS_TAGNODE_VALUE={interface} /usr/libexec/vyos/conf_mode/interfaces_wwan.py') else: print(f'Unknown interface {interface}, can not connect. Aborting!') diff --git a/src/op_mode/dhcp.py b/src/op_mode/dhcp.py index 77f38992b..a64acec31 100755 --- a/src/op_mode/dhcp.py +++ b/src/op_mode/dhcp.py @@ -21,7 +21,6 @@ import typing from datetime import datetime from glob import glob from ipaddress import ip_address -from isc_dhcp_leases import IscDhcpLeases from tabulate import tabulate import vyos.opmode @@ -29,9 +28,9 @@ import vyos.opmode from vyos.base import Warning from vyos.configquery import ConfigTreeQuery -from vyos.utils.dict import dict_search -from vyos.utils.file import read_file -from vyos.utils.process import cmd +from vyos.kea import kea_get_active_config +from vyos.kea import kea_get_pool_from_subnet_id +from vyos.kea import kea_parse_leases from vyos.utils.process import is_systemd_service_running time_string = "%a %b %d %H:%M:%S %Z %Y" @@ -43,6 +42,7 @@ sort_valid_inet6 = ['end', 'iaid_duid', 'ip', 'last_communication', 'pool', 'rem ArgFamily = typing.Literal['inet', 'inet6'] ArgState = typing.Literal['all', 'active', 'free', 'expired', 'released', 'abandoned', 'reset', 'backup'] +ArgOrigin = typing.Literal['local', 'remote'] def _utc_to_local(utc_dt): return datetime.fromtimestamp((datetime.fromtimestamp(utc_dt) - datetime(1970, 1, 1)).total_seconds()) @@ -71,42 +71,47 @@ def _find_list_of_dict_index(lst, key='ip', value='') -> int: return idx -def _get_raw_server_leases(family='inet', pool=None, sorted=None, state=[]) -> list: +def _get_raw_server_leases(family='inet', pool=None, sorted=None, state=[], origin=None) -> list: """ Get DHCP server leases :return list """ - lease_file = '/config/dhcpdv6.leases' if family == 'inet6' else '/config/dhcpd.leases' - data = [] - leases = IscDhcpLeases(lease_file).get() + inet_suffix = '6' if family == 'inet6' else '4' + lease_file = f'/config/dhcp/dhcp{inet_suffix}-leases.csv' + leases = kea_parse_leases(lease_file) if pool is None: pool = _get_dhcp_pools(family=family) else: pool = [pool] + active_config = kea_get_active_config(inet_suffix) + + data = [] for lease in leases: data_lease = {} - data_lease['ip'] = lease.ip - data_lease['state'] = lease.binding_state - data_lease['pool'] = lease.sets.get('shared-networkname', '') - data_lease['end'] = lease.end.timestamp() if lease.end else None + data_lease['ip'] = lease['address'] + lease_state_long = {'0': 'active', '1': 'rejected', '2': 'expired'} + data_lease['state'] = lease_state_long[lease['state']] + data_lease['pool'] = kea_get_pool_from_subnet_id(active_config, inet_suffix, lease['subnet_id']) if active_config else '-' + data_lease['end'] = lease['expire_timestamp'].timestamp() if lease['expire_timestamp'] else None + data_lease['origin'] = 'local' # TODO: Determine remote in HA if family == 'inet': - data_lease['mac'] = lease.ethernet - data_lease['start'] = lease.start.timestamp() - data_lease['hostname'] = lease.hostname + data_lease['mac'] = lease['hwaddr'] + data_lease['start'] = lease['start_timestamp'].timestamp() + data_lease['hostname'] = lease['hostname'] if family == 'inet6': - data_lease['last_communication'] = lease.last_communication.timestamp() - data_lease['iaid_duid'] = _format_hex_string(lease.host_identifier_string) - lease_types_long = {'na': 'non-temporary', 'ta': 'temporary', 'pd': 'prefix delegation'} - data_lease['type'] = lease_types_long[lease.type] + data_lease['last_communication'] = lease['start_timestamp'].timestamp() + data_lease['iaid_duid'] = _format_hex_string(lease['duid']) + lease_types_long = {'0': 'non-temporary', '1': 'temporary', '2': 'prefix delegation'} + data_lease['type'] = lease_types_long[lease['lease_type']] data_lease['remaining'] = '-' - if lease.end: - data_lease['remaining'] = lease.end - datetime.utcnow() + if lease['expire']: + data_lease['remaining'] = lease['expire_timestamp'] - datetime.utcnow() if data_lease['remaining'].days >= 0: # substraction gives us a timedelta object which can't be formatted with strftime @@ -115,7 +120,7 @@ def _get_raw_server_leases(family='inet', pool=None, sorted=None, state=[]) -> l # Do not add old leases if data_lease['remaining'] != '' and data_lease['pool'] in pool and data_lease['state'] != 'free': - if not state or data_lease['state'] in state: + if not state or state == 'all' or data_lease['state'] in state: data.append(data_lease) # deduplicate @@ -150,10 +155,11 @@ def _get_formatted_server_leases(raw_data, family='inet'): remain = lease.get('remaining') pool = lease.get('pool') hostname = lease.get('hostname') - data_entries.append([ipaddr, hw_addr, state, start, end, remain, pool, hostname]) + origin = lease.get('origin') + data_entries.append([ipaddr, hw_addr, state, start, end, remain, pool, hostname, origin]) headers = ['IP Address', 'MAC address', 'State', 'Lease start', 'Lease expiration', 'Remaining', 'Pool', - 'Hostname'] + 'Hostname', 'Origin'] if family == 'inet6': for lease in raw_data: @@ -188,14 +194,11 @@ def _get_pool_size(pool, family='inet'): size = 0 subnets = config.list_nodes(f'{base} subnet') for subnet in subnets: - if family == 'inet6': - ranges = config.list_nodes(f'{base} subnet {subnet} address-range start') - else: - ranges = config.list_nodes(f'{base} subnet {subnet} range') + ranges = config.list_nodes(f'{base} subnet {subnet} range') for range in ranges: if family == 'inet6': - start = config.list_nodes(f'{base} subnet {subnet} address-range start')[0] - stop = config.value(f'{base} subnet {subnet} address-range start {start} stop') + start = config.value(f'{base} subnet {subnet} range {range} start') + stop = config.value(f'{base} subnet {subnet} range {range} stop') else: start = config.value(f'{base} subnet {subnet} range {range} start') stop = config.value(f'{base} subnet {subnet} range {range} stop') @@ -267,12 +270,12 @@ def show_pool_statistics(raw: bool, family: ArgFamily, pool: typing.Optional[str @_verify def show_server_leases(raw: bool, family: ArgFamily, pool: typing.Optional[str], - sorted: typing.Optional[str], state: typing.Optional[ArgState]): + sorted: typing.Optional[str], state: typing.Optional[ArgState], + origin: typing.Optional[ArgOrigin] ): # if dhcp server is down, inactive leases may still be shown as active, so warn the user. - v = '6' if family == 'inet6' else '' - service_name = 'DHCPv6' if family == 'inet6' else 'DHCP' - if not is_systemd_service_running(f'isc-dhcp-server{v}.service'): - Warning(f'{service_name} server is configured but not started. Data may be stale.') + v = '6' if family == 'inet6' else '4' + if not is_systemd_service_running(f'kea-dhcp{v}-server.service'): + Warning('DHCP server is configured but not started. Data may be stale.') v = 'v6' if family == 'inet6' else '' if pool and pool not in _get_dhcp_pools(family=family): @@ -285,7 +288,7 @@ def show_server_leases(raw: bool, family: ArgFamily, pool: typing.Optional[str], if sorted and sorted not in sort_valid: raise vyos.opmode.IncorrectValue(f'DHCP{v} sort "{sorted}" is invalid!') - lease_data = _get_raw_server_leases(family=family, pool=pool, sorted=sorted, state=state) + lease_data = _get_raw_server_leases(family=family, pool=pool, sorted=sorted, state=state, origin=origin) if raw: return lease_data else: diff --git a/src/op_mode/dns.py b/src/op_mode/dns.py index 2168aef89..16c462f23 100755 --- a/src/op_mode/dns.py +++ b/src/op_mode/dns.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2022 VyOS maintainers and contributors +# Copyright (C) 2022-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 @@ -15,17 +15,35 @@ # along with this program. If not, see <http://www.gnu.org/licenses/>. +import os import sys +import time +import typing +import vyos.opmode from tabulate import tabulate - from vyos.configquery import ConfigTreeQuery -from vyos.utils.process import cmd - -import vyos.opmode - - -def _data_to_dict(data, sep="\t") -> dict: +from vyos.utils.process import cmd, rc_cmd +from vyos.template import is_ipv4, is_ipv6 + +_dynamic_cache_file = r'/run/ddclient/ddclient.cache' + +_dynamic_status_columns = { + 'host': 'Hostname', + 'ipv4': 'IPv4 address', + 'status-ipv4': 'IPv4 status', + 'ipv6': 'IPv6 address', + 'status-ipv6': 'IPv6 status', + 'mtime': 'Last update', +} + +_forwarding_statistics_columns = { + 'cache-entries': 'Cache entries', + 'max-cache-entries': 'Max cache entries', + 'cache-size': 'Cache size', +} + +def _forwarding_data_to_dict(data, sep="\t") -> dict: """ Return dictionary from plain text separated by tab @@ -51,37 +69,135 @@ def _data_to_dict(data, sep="\t") -> dict: dictionary[key] = value return dictionary +def _get_dynamic_host_records_raw() -> dict: + + data = [] + + if os.path.isfile(_dynamic_cache_file): # A ddclient status file might not always exist + with open(_dynamic_cache_file, 'r') as f: + for line in f: + if line.startswith('#'): + continue + + props = {} + # ddclient cache rows have properties in 'key=value' format separated by comma + # we pick up the ones we are interested in + for kvraw in line.split(' ')[0].split(','): + k, v = kvraw.split('=') + if k in list(_dynamic_status_columns.keys()) + ['ip', 'status']: # ip and status are legacy keys + props[k] = v + + # Extract IPv4 and IPv6 address and status from legacy keys + # Dual-stack isn't supported in legacy format, 'ip' and 'status' are for one of IPv4 or IPv6 + if 'ip' in props: + if is_ipv4(props['ip']): + props['ipv4'] = props['ip'] + props['status-ipv4'] = props['status'] + elif is_ipv6(props['ip']): + props['ipv6'] = props['ip'] + props['status-ipv6'] = props['status'] + del props['ip'] + + # Convert mtime to human readable format + if 'mtime' in props: + props['mtime'] = time.strftime( + "%Y-%m-%d %H:%M:%S", time.localtime(int(props['mtime'], base=10))) + + data.append(props) -def _get_raw_forwarding_statistics() -> dict: - command = cmd('rec_control --socket-dir=/run/powerdns get-all') - data = _data_to_dict(command) - data['cache-size'] = "{0:.2f}".format( int( - cmd('rec_control --socket-dir=/run/powerdns get cache-bytes')) / 1024 ) return data - -def _get_formatted_forwarding_statistics(data): - cache_entries = data.get('cache-entries') - max_cache_entries = data.get('max-cache-entries') - cache_size = data.get('cache-size') - data_entries = [[cache_entries, max_cache_entries, f'{cache_size} kbytes']] - headers = ["Cache entries", "Max cache entries" , "Cache size"] - output = tabulate(data_entries, headers, numalign="left") +def _get_dynamic_host_records_formatted(data): + data_entries = [] + for entry in data: + data_entries.append([entry.get(key) for key in _dynamic_status_columns.keys()]) + header = _dynamic_status_columns.values() + output = tabulate(data_entries, header, numalign='left') return output +def _get_forwarding_statistics_raw() -> dict: + command = cmd('rec_control get-all') + data = _forwarding_data_to_dict(command) + data['cache-size'] = "{0:.2f} kbytes".format( int( + cmd('rec_control get cache-bytes')) / 1024 ) + return data -def show_forwarding_statistics(raw: bool): +def _get_forwarding_statistics_formatted(data): + data_entries = [] + data_entries.append([data.get(key) for key in _forwarding_statistics_columns.keys()]) + header = _forwarding_statistics_columns.values() + output = tabulate(data_entries, header, numalign='left') + return output - config = ConfigTreeQuery() - if not config.exists('service dns forwarding'): - raise vyos.opmode.UnconfiguredSubsystem('DNS forwarding is not configured') +def _verify(target): + """Decorator checks if config for DNS related service exists""" + from functools import wraps + + if target not in ['dynamic', 'forwarding']: + raise ValueError('Invalid target') + + def _verify_target(func): + @wraps(func) + def _wrapper(*args, **kwargs): + config = ConfigTreeQuery() + if not config.exists(f'service dns {target}'): + _prefix = f'Dynamic DNS' if target == 'dynamic' else 'DNS Forwarding' + raise vyos.opmode.UnconfiguredSubsystem(f'{_prefix} is not configured') + return func(*args, **kwargs) + return _wrapper + return _verify_target + +@_verify('dynamic') +def show_dynamic_status(raw: bool): + host_data = _get_dynamic_host_records_raw() + if raw: + return host_data + else: + return _get_dynamic_host_records_formatted(host_data) - dns_data = _get_raw_forwarding_statistics() +@_verify('dynamic') +def reset_dynamic(): + """ + Reset Dynamic DNS cache + """ + if os.path.exists(_dynamic_cache_file): + os.remove(_dynamic_cache_file) + rc, output = rc_cmd('systemctl restart ddclient.service') + if rc != 0: + print(output) + return None + print(f'Dynamic DNS state reset!') + +@_verify('forwarding') +def show_forwarding_statistics(raw: bool): + dns_data = _get_forwarding_statistics_raw() if raw: return dns_data else: - return _get_formatted_forwarding_statistics(dns_data) + return _get_forwarding_statistics_formatted(dns_data) + +@_verify('forwarding') +def reset_forwarding(all: bool, domain: typing.Optional[str]): + """ + Reset DNS Forwarding cache + :param all (bool): reset cache all domains + :param domain (str): reset cache for specified domain + """ + if all: + rc, output = rc_cmd('rec_control wipe-cache ".$"') + if rc != 0: + print(output) + return None + print('DNS Forwarding cache reset for all domains!') + return output + elif domain: + rc, output = rc_cmd(f'rec_control wipe-cache "{domain}$"') + if rc != 0: + print(output) + return None + print(f'DNS Forwarding cache reset for domain "{domain}"!') + return output if __name__ == '__main__': try: diff --git a/src/op_mode/dns_dynamic.py b/src/op_mode/dns_dynamic.py deleted file mode 100755 index 12aa5494a..000000000 --- a/src/op_mode/dns_dynamic.py +++ /dev/null @@ -1,113 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2018-2023 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 os -import argparse -import sys -import time -from tabulate import tabulate - -from vyos.config import Config -from vyos.template import is_ipv4, is_ipv6 -from vyos.utils.process import call - -cache_file = r'/run/ddclient/ddclient.cache' - -columns = { - 'host': 'Hostname', - 'ipv4': 'IPv4 address', - 'status-ipv4': 'IPv4 status', - 'ipv6': 'IPv6 address', - 'status-ipv6': 'IPv6 status', - 'mtime': 'Last update', -} - - -def _get_formatted_host_records(host_data): - data_entries = [] - for entry in host_data: - data_entries.append([entry.get(key) for key in columns.keys()]) - - header = columns.values() - output = tabulate(data_entries, header, numalign='left') - return output - - -def show_status(): - # A ddclient status file might not always exist - if not os.path.exists(cache_file): - sys.exit(0) - - data = [] - - with open(cache_file, 'r') as f: - for line in f: - if line.startswith('#'): - continue - - props = {} - # ddclient cache rows have properties in 'key=value' format separated by comma - # we pick up the ones we are interested in - for kvraw in line.split(' ')[0].split(','): - k, v = kvraw.split('=') - if k in list(columns.keys()) + ['ip', 'status']: # ip and status are legacy keys - props[k] = v - - # Extract IPv4 and IPv6 address and status from legacy keys - # Dual-stack isn't supported in legacy format, 'ip' and 'status' are for one of IPv4 or IPv6 - if 'ip' in props: - if is_ipv4(props['ip']): - props['ipv4'] = props['ip'] - props['status-ipv4'] = props['status'] - elif is_ipv6(props['ip']): - props['ipv6'] = props['ip'] - props['status-ipv6'] = props['status'] - del props['ip'] - - # Convert mtime to human readable format - if 'mtime' in props: - props['mtime'] = time.strftime( - "%Y-%m-%d %H:%M:%S", time.localtime(int(props['mtime'], base=10))) - - data.append(props) - - print(_get_formatted_host_records(data)) - - -def update_ddns(): - call('systemctl stop ddclient.service') - if os.path.exists(cache_file): - os.remove(cache_file) - call('systemctl start ddclient.service') - - -if __name__ == '__main__': - parser = argparse.ArgumentParser() - group = parser.add_mutually_exclusive_group() - group.add_argument("--status", help="Show DDNS status", action="store_true") - group.add_argument("--update", help="Update DDNS on a given interface", action="store_true") - args = parser.parse_args() - - # Do nothing if service is not configured - c = Config() - if not c.exists_effective('service dns dynamic'): - print("Dynamic DNS not configured") - sys.exit(1) - - if args.status: - show_status() - elif args.update: - update_ddns() diff --git a/src/op_mode/dns_forwarding_reset.py b/src/op_mode/dns_forwarding_reset.py deleted file mode 100755 index 55e20918f..000000000 --- a/src/op_mode/dns_forwarding_reset.py +++ /dev/null @@ -1,54 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2018 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/>. -# -# File: vyos-show-version -# Purpose: -# Displays image version and system information. -# Used by the "run show version" command. - - -import os -import argparse - -from sys import exit -from vyos.config import Config -from vyos.utils.process import call - -PDNS_CMD='/usr/bin/rec_control --socket-dir=/run/powerdns' - -parser = argparse.ArgumentParser() -parser.add_argument("-a", "--all", action="store_true", help="Reset all cache") -parser.add_argument("domain", type=str, nargs="?", help="Domain to reset cache entries for") - -if __name__ == '__main__': - args = parser.parse_args() - - # Do nothing if service is not configured - c = Config() - if not c.exists_effective(['service', 'dns', 'forwarding']): - print("DNS forwarding is not configured") - exit(0) - - if args.all: - call(f"{PDNS_CMD} wipe-cache \'.$\'") - exit(0) - - elif args.domain: - call(f"{PDNS_CMD} wipe-cache \'{0}$\'".format(args.domain)) - - else: - parser.print_help() - exit(1) diff --git a/src/op_mode/dns_forwarding_restart.sh b/src/op_mode/dns_forwarding_restart.sh deleted file mode 100755 index 64cc92115..000000000 --- a/src/op_mode/dns_forwarding_restart.sh +++ /dev/null @@ -1,8 +0,0 @@ -#!/bin/sh - -if cli-shell-api existsEffective service dns forwarding; then - echo "Restarting the DNS forwarding service" - systemctl restart pdns-recursor.service -else - echo "DNS forwarding is not configured" -fi diff --git a/src/op_mode/dns_forwarding_statistics.py b/src/op_mode/dns_forwarding_statistics.py deleted file mode 100755 index 32b5c76a7..000000000 --- a/src/op_mode/dns_forwarding_statistics.py +++ /dev/null @@ -1,32 +0,0 @@ -#!/usr/bin/env python3 - -import jinja2 -from sys import exit - -from vyos.config import Config -from vyos.utils.process import cmd - -PDNS_CMD='/usr/bin/rec_control --socket-dir=/run/powerdns' - -OUT_TMPL_SRC = """ -DNS forwarding statistics: - -Cache entries: {{ cache_entries }} -Cache size: {{ cache_size }} kbytes - -""" - -if __name__ == '__main__': - # Do nothing if service is not configured - c = Config() - if not c.exists_effective('service dns forwarding'): - print("DNS forwarding is not configured") - exit(0) - - data = {} - - data['cache_entries'] = cmd(f'{PDNS_CMD} get cache-entries') - data['cache_size'] = "{0:.2f}".format( int(cmd(f'{PDNS_CMD} get cache-bytes')) / 1024 ) - - tmpl = jinja2.Template(OUT_TMPL_SRC) - print(tmpl.render(data)) diff --git a/src/op_mode/firewall.py b/src/op_mode/firewall.py index 3434707ec..4dcffc412 100755 --- a/src/op_mode/firewall.py +++ b/src/op_mode/firewall.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021 VyOS maintainers and contributors +# Copyright (C) 2023 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 @@ -24,19 +24,28 @@ from vyos.config import Config from vyos.utils.process import cmd from vyos.utils.dict import dict_search_args -def get_config_firewall(conf, family=None, hook=None, priority=None): - config_path = ['firewall'] - if family: - config_path += [family] - if hook: - config_path += [hook] - if priority: - config_path += [priority] +def get_config_node(conf, node=None, family=None, hook=None, priority=None): + if node == 'nat': + if family == 'ipv6': + config_path = ['nat66'] + else: + config_path = ['nat'] - firewall = conf.get_config_dict(config_path, key_mangling=('-', '_'), + elif node == 'policy': + config_path = ['policy'] + else: + config_path = ['firewall'] + if family: + config_path += [family] + if hook: + config_path += [hook] + if priority: + config_path += [priority] + + node_config = conf.get_config_dict(config_path, key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True) - return firewall + return node_config def get_nftables_details(family, hook, priority): if family == 'ipv6': @@ -102,13 +111,16 @@ def output_firewall_name(family, hook, priority, firewall_conf, single_rule_id=N row.append(rule_details['conditions']) rows.append(row) - if 'default_action' in firewall_conf and not single_rule_id: - row = ['default', firewall_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 hook in ['input', 'forward', 'output']: + def_action = firewall_conf['default_action'] if 'default_action' in firewall_conf else 'accept' + else: + def_action = firewall_conf['default_action'] if 'default_action' in firewall_conf else 'drop' + row = ['default', def_action, 'all'] + 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'] @@ -167,16 +179,16 @@ def output_firewall_name_statistics(family, hook, prior, prior_conf, single_rule dest_addr = 'any' # Get inbound interface - iiface = dict_search_args(rule_conf, 'inbound_interface', 'interface_name') + iiface = dict_search_args(rule_conf, 'inbound_interface', 'name') if not iiface: - iiface = dict_search_args(rule_conf, 'inbound_interface', 'interface_group') + iiface = dict_search_args(rule_conf, 'inbound_interface', 'group') if not iiface: iiface = 'any' # Get outbound interface - oiface = dict_search_args(rule_conf, 'outbound_interface', 'interface_name') + oiface = dict_search_args(rule_conf, 'outbound_interface', 'name') if not oiface: - oiface = dict_search_args(rule_conf, 'outbound_interface', 'interface_group') + oiface = dict_search_args(rule_conf, 'outbound_interface', 'group') if not oiface: oiface = 'any' @@ -198,8 +210,9 @@ def output_firewall_name_statistics(family, hook, prior, prior_conf, single_rule if hook in ['input', 'forward', 'output']: row = ['default'] - row.append('N/A') - row.append('N/A') + rule_details = details['default-action'] + row.append(rule_details.get('packets', 0)) + row.append(rule_details.get('bytes', 0)) if 'default_action' in prior_conf: row.append(prior_conf['default_action']) else: @@ -234,7 +247,7 @@ def show_firewall(): print('Rulesets Information') conf = Config() - firewall = get_config_firewall(conf) + firewall = get_config_node(conf) if not firewall: return @@ -249,7 +262,7 @@ def show_firewall_family(family): print(f'Rulesets {family} Information') conf = Config() - firewall = get_config_firewall(conf) + firewall = get_config_node(conf) if not firewall or family not in firewall: return @@ -262,7 +275,7 @@ def show_firewall_name(family, hook, priority): print('Ruleset Information') conf = Config() - firewall = get_config_firewall(conf, family, hook, priority) + firewall = get_config_node(conf, 'firewall', family, hook, priority) if firewall: output_firewall_name(family, hook, priority, firewall) @@ -270,17 +283,20 @@ def show_firewall_rule(family, hook, priority, rule_id): print('Rule Information') conf = Config() - firewall = get_config_firewall(conf, family, hook, priority) + firewall = get_config_node(conf, 'firewall', family, hook, priority) if firewall: output_firewall_name(family, hook, priority, firewall, rule_id) def show_firewall_group(name=None): conf = Config() - firewall = get_config_firewall(conf) + firewall = get_config_node(conf, node='firewall') if 'group' not in firewall: return + nat = get_config_node(conf, node='nat') + policy = get_config_node(conf, node='policy') + def find_references(group_type, group_name): out = [] family = [] @@ -293,9 +309,10 @@ def show_firewall_group(name=None): family = ['ipv6'] group_type = 'network_group' else: - family = ['ipv4', 'ipv6'] + family = ['ipv4', 'ipv6', 'bridge'] for item in family: + # Look references in firewall for name_type in ['name', 'ipv6_name', 'forward', 'input', 'output']: if item in firewall: if name_type not in firewall[item]: @@ -308,8 +325,10 @@ def show_firewall_group(name=None): for rule_id, rule_conf in priority_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) - in_interface = dict_search_args(rule_conf, 'inbound_interface', 'interface_group') - out_interface = dict_search_args(rule_conf, 'outbound_interface', 'interface_group') + in_interface = dict_search_args(rule_conf, 'inbound_interface', 'group') + out_interface = dict_search_args(rule_conf, 'outbound_interface', 'group') + dyn_group_source = dict_search_args(rule_conf, 'add_address_to_group', 'source_address', group_type) + dyn_group_dst = dict_search_args(rule_conf, 'add_address_to_group', 'destination_address', group_type) if source_group: if source_group[0] == "!": source_group = source_group[1:] @@ -330,31 +349,121 @@ def show_firewall_group(name=None): out_interface = out_interface[1:] if group_name == out_interface: out.append(f'{item}-{name_type}-{priority}-{rule_id}') + + if dyn_group_source: + if group_name == dyn_group_source: + out.append(f'{item}-{name_type}-{priority}-{rule_id}') + if dyn_group_dst: + if group_name == dyn_group_dst: + out.append(f'{item}-{name_type}-{priority}-{rule_id}') + + + # Look references in route | route6 + for name_type in ['route', 'route6']: + if name_type not in policy: + continue + if name_type == 'route' and item == 'ipv6': + continue + elif name_type == 'route6' and item == 'ipv4': + continue + else: + for policy_name, policy_conf in policy[name_type].items(): + if 'rule' not in policy_conf: + continue + for rule_id, rule_conf in policy_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) + in_interface = dict_search_args(rule_conf, 'inbound_interface', 'group') + out_interface = dict_search_args(rule_conf, 'outbound_interface', 'group') + if source_group: + if source_group[0] == "!": + source_group = source_group[1:] + if group_name == source_group: + out.append(f'{name_type}-{policy_name}-{rule_id}') + if dest_group: + if dest_group[0] == "!": + dest_group = dest_group[1:] + if group_name == dest_group: + out.append(f'{name_type}-{policy_name}-{rule_id}') + if in_interface: + if in_interface[0] == "!": + in_interface = in_interface[1:] + if group_name == in_interface: + out.append(f'{name_type}-{policy_name}-{rule_id}') + if out_interface: + if out_interface[0] == "!": + out_interface = out_interface[1:] + if group_name == out_interface: + out.append(f'{name_type}-{policy_name}-{rule_id}') + + ## Look references in nat table + for direction in ['source', 'destination']: + if direction in nat: + if 'rule' not in nat[direction]: + continue + for rule_id, rule_conf in nat[direction]['rule'].items(): + source_group = dict_search_args(rule_conf, 'source', 'group', group_type) + dest_group = dict_search_args(rule_conf, 'destination', 'group', group_type) + in_interface = dict_search_args(rule_conf, 'inbound_interface', 'group') + out_interface = dict_search_args(rule_conf, 'outbound_interface', 'group') + if source_group: + if source_group[0] == "!": + source_group = source_group[1:] + if group_name == source_group: + out.append(f'nat-{direction}-{rule_id}') + if dest_group: + if dest_group[0] == "!": + dest_group = dest_group[1:] + if group_name == dest_group: + out.append(f'nat-{direction}-{rule_id}') + if in_interface: + if in_interface[0] == "!": + in_interface = in_interface[1:] + if group_name == in_interface: + out.append(f'nat-{direction}-{rule_id}') + if out_interface: + if out_interface[0] == "!": + out_interface = out_interface[1:] + if group_name == out_interface: + out.append(f'nat-{direction}-{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 + ## + if group_type != 'dynamic_group': + + 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/D'] + if 'address' in group_conf: + row.append("\n".join(sorted(group_conf['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']))) + elif 'interface' in group_conf: + row.append("\n".join(sorted(group_conf['interface']))) + else: + row.append('N/D') + rows.append(row) - references = find_references(group_type, group_name) - row = [group_name, group_type, '\n'.join(references) or 'N/D'] - if 'address' in group_conf: - row.append("\n".join(sorted(group_conf['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']))) - elif 'interface' in group_conf: - row.append("\n".join(sorted(group_conf['interface']))) - else: - row.append('N/D') - rows.append(row) + else: + for dynamic_type in ['address_group', 'ipv6_address_group']: + if dynamic_type in firewall['group']['dynamic_group']: + for dynamic_name, dynamic_conf in firewall['group']['dynamic_group'][dynamic_type].items(): + references = find_references(dynamic_type, dynamic_name) + row = [dynamic_name, dynamic_type + '(dynamic)', '\n'.join(references) or 'N/D'] + row.append('N/D') + rows.append(row) if rows: print('Firewall Groups\n') @@ -364,7 +473,7 @@ def show_summary(): print('Ruleset Summary') conf = Config() - firewall = get_config_firewall(conf) + firewall = get_config_node(conf) if not firewall: return @@ -410,7 +519,7 @@ def show_statistics(): print('Rulesets Statistics') conf = Config() - firewall = get_config_firewall(conf) + firewall = get_config_node(conf) if not firewall: return diff --git a/src/op_mode/generate_firewall_rule-resequence.py b/src/op_mode/generate_firewall_rule-resequence.py index eb82a1a0a..21441f689 100755 --- a/src/op_mode/generate_firewall_rule-resequence.py +++ b/src/op_mode/generate_firewall_rule-resequence.py @@ -41,6 +41,10 @@ def convert_to_set_commands(config_dict, parent_key=''): commands.extend( convert_to_set_commands(value, f"{current_key} ")) + elif isinstance(value, list): + for item in value: + commands.append(f"set {current_key} '{item}'") + elif isinstance(value, str): commands.append(f"set {current_key} '{value}'") diff --git a/src/op_mode/generate_ipsec_debug_archive.py b/src/op_mode/generate_ipsec_debug_archive.py index 60195d48b..ca2eeb511 100755 --- a/src/op_mode/generate_ipsec_debug_archive.py +++ b/src/op_mode/generate_ipsec_debug_archive.py @@ -24,7 +24,6 @@ from vyos.utils.process import rc_cmd # define a list of commands that needs to be executed CMD_LIST: list[str] = [ - 'ipsec status', 'swanctl -L', 'swanctl -l', 'swanctl -P', @@ -36,7 +35,7 @@ CMD_LIST: list[str] = [ 'ip route | head -100', 'ip route show table 220' ] -JOURNALCTL_CMD: str = 'journalctl -b -n 10000 /usr/lib/ipsec/charon' +JOURNALCTL_CMD: str = 'journalctl --no-hostname --boot --unit strongswan.service' # execute a command and save the output to a file def save_stdout(command: str, file: Path) -> None: diff --git a/src/op_mode/generate_tech-support_archive.py b/src/op_mode/generate_tech-support_archive.py new file mode 100755 index 000000000..c490b0137 --- /dev/null +++ b/src/op_mode/generate_tech-support_archive.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2023 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 os +import argparse +import glob +from datetime import datetime +from pathlib import Path +from shutil import rmtree + +from socket import gethostname +from sys import exit +from tarfile import open as tar_open +from vyos.utils.process import rc_cmd +from vyos.remote import upload + +def op(cmd: str) -> str: + """Returns a command with the VyOS operational mode wrapper.""" + return f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {cmd}' + +def save_stdout(command: str, file: Path) -> None: + rc, stdout = rc_cmd(command) + body: str = f'''### {command} ### +Command: {command} +Exit code: {rc} +Stdout: +{stdout} + +''' + with file.open(mode='a') as f: + f.write(body) +def __rotate_logs(path: str, log_pattern:str): + files_list = glob.glob(f'{path}/{log_pattern}') + if len(files_list) > 5: + oldest_file = min(files_list, key=os.path.getctime) + os.remove(oldest_file) + + +def __generate_archived_files(location_path: str) -> None: + """ + Generate arhives of main directories + :param location_path: path to temporary directory + :type location_path: str + """ + # Dictionary arhive_name:directory_to_arhive + archive_dict = { + 'etc': '/etc', + 'home': '/home', + 'var-log': '/var/log', + 'root': '/root', + 'tmp': '/tmp', + 'core-dump': '/var/core', + 'config': '/opt/vyatta/etc/config' + } + # Dictionary arhive_name:excluding pattern + archive_excludes = { + # Old location of archives + 'config': 'tech-support-archive', + # New locations of arhives + 'tmp': 'tech-support-archive' + } + for archive_name, path in archive_dict.items(): + archive_file: str = f'{location_path}/{archive_name}.tar.gz' + with tar_open(name=archive_file, mode='x:gz') as tar_file: + if archive_name in archive_excludes: + tar_file.add(path, filter=lambda x: None if str(archive_excludes[archive_name]) in str(x.name) else x) + else: + tar_file.add(path) + + +def __generate_main_archive_file(archive_file: str, tmp_dir_path: str) -> None: + """ + Generate main arhive file + :param archive_file: name of arhive file + :type archive_file: str + :param tmp_dir_path: path to arhive memeber + :type tmp_dir_path: str + """ + with tar_open(name=archive_file, mode='x:gz') as tar_file: + tar_file.add(tmp_dir_path, arcname=os.path.basename(tmp_dir_path)) + + +if __name__ == '__main__': + defualt_tmp_dir = '/tmp' + parser = argparse.ArgumentParser() + parser.add_argument("path", nargs='?', default=defualt_tmp_dir) + args = parser.parse_args() + location_path = args.path[:-1] if args.path[-1] == '/' else args.path + + hostname: str = gethostname() + time_now: str = datetime.now().isoformat(timespec='seconds').replace(":", "-") + + remote = False + tmp_path = '' + tmp_dir_path = '' + if 'ftp://' in args.path or 'scp://' in args.path: + remote = True + tmp_path = defualt_tmp_dir + else: + tmp_path = location_path + archive_pattern = f'_tech-support-archive_' + archive_file_name = f'{hostname}{archive_pattern}{time_now}.tar.gz' + + # Log rotation in tmp directory + if tmp_path == defualt_tmp_dir: + __rotate_logs(tmp_path, f'*{archive_pattern}*') + + # Temporary directory creation + tmp_dir_path = f'{tmp_path}/drops-debug_{time_now}' + tmp_dir: Path = Path(tmp_dir_path) + tmp_dir.mkdir() + + report_file: Path = Path(f'{tmp_dir_path}/show_tech-support_report.txt') + report_file.touch() + try: + + save_stdout(op('show tech-support report'), report_file) + # Generate included archives + __generate_archived_files(tmp_dir_path) + + # Generate main archive + __generate_main_archive_file(f'{tmp_path}/{archive_file_name}', tmp_dir_path) + # Delete temporary directory + rmtree(tmp_dir) + # Upload to remote site if it is scpecified + if remote: + upload(f'{tmp_path}/{archive_file_name}', args.path) + print(f'Debug file is generated and located in {location_path}/{archive_file_name}') + except Exception as err: + print(f'Error during generating a debug file: {err}') + # cleanup + if tmp_dir.exists(): + rmtree(tmp_dir) + finally: + # cleanup + exit() diff --git a/src/op_mode/image_info.py b/src/op_mode/image_info.py new file mode 100755 index 000000000..791001e00 --- /dev/null +++ b/src/op_mode/image_info.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +# +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This file is part of VyOS. +# +# VyOS 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. +# +# VyOS 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 +# VyOS. If not, see <https://www.gnu.org/licenses/>. + +import sys +from typing import List, Union + +from tabulate import tabulate + +from vyos import opmode +from vyos.system import disk, grub, image +from vyos.utils.convert import bytes_to_human + + +def _format_show_images_summary(images_summary: image.BootDetails) -> str: + headers: list[str] = ['Name', 'Default boot', 'Running'] + table_data: list[list[str]] = list() + for image_item in images_summary.get('images_available', []): + name: str = image_item + if images_summary.get('image_default') == name: + default: str = 'Yes' + else: + default: str = '' + + if images_summary.get('image_running') == name: + running: str = 'Yes' + else: + running: str = '' + + table_data.append([name, default, running]) + tabulated: str = tabulate(table_data, headers) + + return tabulated + + +def _format_show_images_details( + images_details: list[image.ImageDetails]) -> str: + headers: list[str] = [ + 'Name', 'Version', 'Storage Read-Only', 'Storage Read-Write', + 'Storage Total' + ] + table_data: list[list[Union[str, int]]] = list() + for image_item in images_details: + name: str = image_item.get('name') + version: str = image_item.get('version') + disk_ro: str = bytes_to_human(image_item.get('disk_ro'), + precision=1, int_below_exponent=30) + disk_rw: str = bytes_to_human(image_item.get('disk_rw'), + precision=1, int_below_exponent=30) + disk_total: str = bytes_to_human(image_item.get('disk_total'), + precision=1, int_below_exponent=30) + table_data.append([name, version, disk_ro, disk_rw, disk_total]) + tabulated: str = tabulate(table_data, headers, + colalign=('left', 'left', 'right', 'right', 'right')) + + return tabulated + + +def show_images_summary(raw: bool) -> Union[image.BootDetails, str]: + images_available: list[str] = grub.version_list() + root_dir: str = disk.find_persistence() + boot_vars: dict = grub.vars_read(f'{root_dir}/{image.CFG_VYOS_VARS}') + + images_summary: image.BootDetails = dict() + + images_summary['image_default'] = image.get_default_image() + images_summary['image_running'] = image.get_running_image() + images_summary['images_available'] = images_available + images_summary['console_type'] = boot_vars.get('console_type') + images_summary['console_num'] = boot_vars.get('console_num') + + if raw: + return images_summary + else: + return _format_show_images_summary(images_summary) + + +def show_images_details(raw: bool) -> Union[list[image.ImageDetails], str]: + images_details = image.get_images_details() + + if raw: + return images_details + else: + return _format_show_images_details(images_details) + + +if __name__ == '__main__': + try: + res = opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, opmode.Error) as e: + print(e) + sys.exit(1) diff --git a/src/op_mode/image_installer.py b/src/op_mode/image_installer.py new file mode 100755 index 000000000..d677c2cf8 --- /dev/null +++ b/src/op_mode/image_installer.py @@ -0,0 +1,929 @@ +#!/usr/bin/env python3 +# +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This file is part of VyOS. +# +# VyOS 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. +# +# VyOS 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 +# VyOS. If not, see <https://www.gnu.org/licenses/>. + +from argparse import ArgumentParser, Namespace +from pathlib import Path +from shutil import copy, chown, rmtree, copytree +from glob import glob +from sys import exit +from os import environ +from time import sleep +from typing import Union +from urllib.parse import urlparse +from passlib.hosts import linux_context + +from psutil import disk_partitions + +from vyos.configtree import ConfigTree +from vyos.configquery import ConfigTreeQuery +from vyos.remote import download +from vyos.system import disk, grub, image, compat, raid, SYSTEM_CFG_VER +from vyos.template import render +from vyos.utils.io import ask_input, ask_yes_no, select_entry +from vyos.utils.file import chmod_2775 +from vyos.utils.process import cmd, run +from vyos.version import get_remote_version + +# define text messages +MSG_ERR_NOT_LIVE: str = 'The system is already installed. Please use "add system image" instead.' +MSG_ERR_LIVE: str = 'The system is in live-boot mode. Please use "install image" instead.' +MSG_ERR_NO_DISK: str = 'No suitable disk was found. There must be at least one disk of 2GB or greater size.' +MSG_ERR_IMPROPER_IMAGE: str = 'Missing sha256sum.txt.\nEither this image is corrupted, or of era 1.2.x (md5sum) and would downgrade image tools;\ndisallowed in either case.' +MSG_INFO_INSTALL_WELCOME: str = 'Welcome to VyOS installation!\nThis command will install VyOS to your permanent storage.' +MSG_INFO_INSTALL_EXIT: str = 'Exiting from VyOS installation' +MSG_INFO_INSTALL_SUCCESS: str = 'The image installed successfully; please reboot now.' +MSG_INFO_INSTALL_DISKS_LIST: str = 'The following disks were found:' +MSG_INFO_INSTALL_DISK_SELECT: str = 'Which one should be used for installation?' +MSG_INFO_INSTALL_RAID_CONFIGURE: str = 'Would you like to configure RAID-1 mirroring?' +MSG_INFO_INSTALL_RAID_FOUND_DISKS: str = 'Would you like to configure RAID-1 mirroring on them?' +MSG_INFO_INSTALL_RAID_CHOOSE_DISKS: str = 'Would you like to choose two disks for RAID-1 mirroring?' +MSG_INFO_INSTALL_DISK_CONFIRM: str = 'Installation will delete all data on the drive. Continue?' +MSG_INFO_INSTALL_RAID_CONFIRM: str = 'Installation will delete all data on both drives. Continue?' +MSG_INFO_INSTALL_PARTITONING: str = 'Creating partition table...' +MSG_INPUT_CONFIG_FOUND: str = 'An active configuration was found. Would you like to copy it to the new image?' +MSG_INPUT_IMAGE_NAME: str = 'What would you like to name this image?' +MSG_INPUT_IMAGE_DEFAULT: str = 'Would you like to set the new image as the default one for boot?' +MSG_INPUT_PASSWORD: str = 'Please enter a password for the "vyos" user' +MSG_INPUT_ROOT_SIZE_ALL: str = 'Would you like to use all the free space on the drive?' +MSG_INPUT_ROOT_SIZE_SET: str = 'Please specify the size (in GB) of the root partition (min is 1.5 GB)?' +MSG_INPUT_CONSOLE_TYPE: str = 'What console should be used by default? (K: KVM, S: Serial, U: USB-Serial)?' +MSG_INPUT_COPY_DATA: str = 'Would you like to copy data to the new image?' +MSG_INPUT_CHOOSE_COPY_DATA: str = 'From which image would you like to save config information?' +MSG_WARN_ISO_SIGN_INVALID: str = 'Signature is not valid. Do you want to continue with installation?' +MSG_WARN_ISO_SIGN_UNAVAL: str = 'Signature is not available. Do you want to continue with installation?' +MSG_WARN_ROOT_SIZE_TOOBIG: str = 'The size is too big. Try again.' +MSG_WARN_ROOT_SIZE_TOOSMALL: str = 'The size is too small. Try again' +MSG_WARN_IMAGE_NAME_WRONG: str = 'The suggested name is unsupported!\n'\ +'It must be between 1 and 64 characters long and contains only the next characters: .+-_ a-z A-Z 0-9' +CONST_MIN_DISK_SIZE: int = 2147483648 # 2 GB +CONST_MIN_ROOT_SIZE: int = 1610612736 # 1.5 GB +# a reserved space: 2MB for header, 1 MB for BIOS partition, 256 MB for EFI +CONST_RESERVED_SPACE: int = (2 + 1 + 256) * 1024**2 + +# define directories and paths +DIR_INSTALLATION: str = '/mnt/installation' +DIR_ROOTFS_SRC: str = f'{DIR_INSTALLATION}/root_src' +DIR_ROOTFS_DST: str = f'{DIR_INSTALLATION}/root_dst' +DIR_ISO_MOUNT: str = f'{DIR_INSTALLATION}/iso_src' +DIR_DST_ROOT: str = f'{DIR_INSTALLATION}/disk_dst' +DIR_KERNEL_SRC: str = '/boot/' +FILE_ROOTFS_SRC: str = '/usr/lib/live/mount/medium/live/filesystem.squashfs' +ISO_DOWNLOAD_PATH: str = '/tmp/vyos_installation.iso' + +external_download_script = '/usr/libexec/vyos/simple-download.py' + +# default boot variables +DEFAULT_BOOT_VARS: dict[str, str] = { + 'timeout': '5', + 'console_type': 'tty', + 'console_num': '0', + 'console_speed': '115200', + 'bootmode': 'normal' +} + + +def bytes_to_gb(size: int) -> float: + """Convert Bytes to GBytes, rounded to 1 decimal number + + Args: + size (int): input size in bytes + + Returns: + float: size in GB + """ + return round(size / 1024**3, 1) + + +def gb_to_bytes(size: float) -> int: + """Convert GBytes to Bytes + + Args: + size (float): input size in GBytes + + Returns: + int: size in bytes + """ + return int(size * 1024**3) + + +def find_disks() -> dict[str, int]: + """Find a target disk for installation + + Returns: + dict[str, int]: a list of available disks by name and size + """ + # check for available disks + print('Probing disks') + disks_available: dict[str, int] = disk.disks_size() + for disk_name, disk_size in disks_available.copy().items(): + if disk_size < CONST_MIN_DISK_SIZE: + del disks_available[disk_name] + if not disks_available: + print(MSG_ERR_NO_DISK) + exit(MSG_INFO_INSTALL_EXIT) + + num_disks: int = len(disks_available) + print(f'{num_disks} disk(s) found') + + return disks_available + + +def ask_root_size(available_space: int) -> int: + """Define a size of root partition + + Args: + available_space (int): available space in bytes for a root partition + + Returns: + int: defined size + """ + if ask_yes_no(MSG_INPUT_ROOT_SIZE_ALL, default=True): + return available_space + + while True: + root_size_gb: str = ask_input(MSG_INPUT_ROOT_SIZE_SET) + root_size_kbytes: int = (gb_to_bytes(float(root_size_gb))) // 1024 + + if root_size_kbytes > available_space: + print(MSG_WARN_ROOT_SIZE_TOOBIG) + continue + if root_size_kbytes < CONST_MIN_ROOT_SIZE / 1024: + print(MSG_WARN_ROOT_SIZE_TOOSMALL) + continue + + return root_size_kbytes + +def create_partitions(target_disk: str, target_size: int, + prompt: bool = True) -> None: + """Create partitions on a target disk + + Args: + target_disk (str): a target disk + target_size (int): size of disk in bytes + """ + # define target rootfs size in KB (smallest unit acceptable by sgdisk) + available_size: int = (target_size - CONST_RESERVED_SPACE) // 1024 + if prompt: + rootfs_size: int = ask_root_size(available_size) + else: + rootfs_size: int = available_size + + print(MSG_INFO_INSTALL_PARTITONING) + raid.clear() + disk.disk_cleanup(target_disk) + disk_details: disk.DiskDetails = disk.parttable_create(target_disk, + rootfs_size) + + return disk_details + + +def search_format_selection(image: tuple[str, str]) -> str: + """Format a string for selection of image + + Args: + image (tuple[str, str]): a tuple of image name and drive + + Returns: + str: formatted string + """ + return f'{image[0]} on {image[1]}' + + +def search_previous_installation(disks: list[str]) -> None: + """Search disks for previous installation config and SSH keys + + Args: + disks (list[str]): a list of available disks + """ + mnt_config = '/mnt/config' + mnt_ssh = '/mnt/ssh' + mnt_tmp = '/mnt/tmp' + rmtree(Path(mnt_config), ignore_errors=True) + rmtree(Path(mnt_ssh), ignore_errors=True) + Path(mnt_tmp).mkdir(exist_ok=True) + + print('Searching for data from previous installations') + image_data = [] + for disk_name in disks: + for partition in disk.partition_list(disk_name): + if disk.partition_mount(partition, mnt_tmp): + if Path(mnt_tmp + '/boot').exists(): + for path in Path(mnt_tmp + '/boot').iterdir(): + if path.joinpath('rw/config/.vyatta_config').exists(): + image_data.append((path.name, partition)) + + disk.partition_umount(partition) + + if len(image_data) == 1: + image_name, image_drive = image_data[0] + print('Found data from previous installation:') + print(f'\t{image_name} on {image_drive}') + if not ask_yes_no(MSG_INPUT_COPY_DATA, default=True): + return + + elif len(image_data) > 1: + print('Found data from previous installations') + if not ask_yes_no(MSG_INPUT_COPY_DATA, default=True): + return + + image_name, image_drive = select_entry(image_data, + 'Available versions:', + MSG_INPUT_CHOOSE_COPY_DATA, + search_format_selection) + else: + print('No previous installation found') + return + + disk.partition_mount(image_drive, mnt_tmp) + + copytree(f'{mnt_tmp}/boot/{image_name}/rw/config', mnt_config) + Path(mnt_ssh).mkdir() + host_keys: list[str] = glob(f'{mnt_tmp}/boot/{image_name}/rw/etc/ssh/ssh_host*') + for host_key in host_keys: + copy(host_key, mnt_ssh) + + disk.partition_umount(image_drive) + +def copy_preserve_owner(src: str, dst: str, *, follow_symlinks=True): + if not Path(src).is_file(): + return + if Path(dst).is_dir(): + dst = Path(dst).joinpath(Path(src).name) + st = Path(src).stat() + copy(src, dst, follow_symlinks=follow_symlinks) + chown(dst, user=st.st_uid) + + +def copy_previous_installation_data(target_dir: str) -> None: + if Path('/mnt/config').exists(): + copytree('/mnt/config', f'{target_dir}/opt/vyatta/etc/config', + dirs_exist_ok=True) + if Path('/mnt/ssh').exists(): + copytree('/mnt/ssh', f'{target_dir}/etc/ssh', + dirs_exist_ok=True) + + +def ask_single_disk(disks_available: dict[str, int]) -> str: + """Ask user to select a disk for installation + + Args: + disks_available (dict[str, int]): a list of available disks + """ + print(MSG_INFO_INSTALL_DISKS_LIST) + default_disk: str = list(disks_available)[0] + for disk_name, disk_size in disks_available.items(): + disk_size_human: str = bytes_to_gb(disk_size) + print(f'Drive: {disk_name} ({disk_size_human} GB)') + disk_selected: str = ask_input(MSG_INFO_INSTALL_DISK_SELECT, + default=default_disk, + valid_responses=list(disks_available)) + + # create partitions + if not ask_yes_no(MSG_INFO_INSTALL_DISK_CONFIRM): + print(MSG_INFO_INSTALL_EXIT) + exit() + + search_previous_installation(list(disks_available)) + + disk_details: disk.DiskDetails = create_partitions(disk_selected, + disks_available[disk_selected]) + + disk.filesystem_create(disk_details.partition['efi'], 'efi') + disk.filesystem_create(disk_details.partition['root'], 'ext4') + + return disk_details + + +def check_raid_install(disks_available: dict[str, int]) -> Union[str, None]: + """Ask user to select disks for RAID installation + + Args: + disks_available (dict[str, int]): a list of available disks + """ + if len(disks_available) < 2: + return None + + if not ask_yes_no(MSG_INFO_INSTALL_RAID_CONFIGURE, default=True): + return None + + def format_selection(disk_name: str) -> str: + return f'{disk_name}\t({bytes_to_gb(disks_available[disk_name])} GB)' + + disk0, disk1 = list(disks_available)[0], list(disks_available)[1] + disks_selected: dict[str, int] = { disk0: disks_available[disk0], + disk1: disks_available[disk1] } + + target_size: int = min(disks_selected[disk0], disks_selected[disk1]) + + print(MSG_INFO_INSTALL_DISKS_LIST) + for disk_name, disk_size in disks_selected.items(): + disk_size_human: str = bytes_to_gb(disk_size) + print(f'\t{disk_name} ({disk_size_human} GB)') + if not ask_yes_no(MSG_INFO_INSTALL_RAID_FOUND_DISKS, default=True): + if not ask_yes_no(MSG_INFO_INSTALL_RAID_CHOOSE_DISKS, default=True): + return None + else: + disks_selected = {} + disk0 = select_entry(list(disks_available), 'Disks available:', + 'Select first disk:', format_selection) + + disks_selected[disk0] = disks_available[disk0] + del disks_available[disk0] + disk1 = select_entry(list(disks_available), 'Remaining disks:', + 'Select second disk:', format_selection) + disks_selected[disk1] = disks_available[disk1] + + target_size: int = min(disks_selected[disk0], + disks_selected[disk1]) + + # create partitions + if not ask_yes_no(MSG_INFO_INSTALL_RAID_CONFIRM): + print(MSG_INFO_INSTALL_EXIT) + exit() + + search_previous_installation(list(disks_available)) + + disks: list[disk.DiskDetails] = [] + for disk_selected in list(disks_selected): + print(f'Creating partitions on {disk_selected}') + disk_details = create_partitions(disk_selected, target_size, + prompt=False) + disk.filesystem_create(disk_details.partition['efi'], 'efi') + + disks.append(disk_details) + + print('Creating RAID array') + members = [disk.partition['root'] for disk in disks] + raid_details: raid.RaidDetails = raid.raid_create(members) + # raid init stuff + print('Updating initramfs') + raid.update_initramfs() + # end init + print('Creating filesystem on RAID array') + disk.filesystem_create(raid_details.name, 'ext4') + + return raid_details + + +def prepare_tmp_disr() -> None: + """Create temporary directories for installation + """ + print('Creating temporary directories') + for dir in [DIR_ROOTFS_SRC, DIR_ROOTFS_DST, DIR_DST_ROOT]: + dirpath = Path(dir) + dirpath.mkdir(mode=0o755, parents=True) + + +def setup_grub(root_dir: str) -> None: + """Install GRUB configurations + + Args: + root_dir (str): a path to the root of target filesystem + """ + print('Installing GRUB configuration files') + grub_cfg_main = f'{root_dir}/{grub.GRUB_DIR_MAIN}/grub.cfg' + grub_cfg_vars = f'{root_dir}/{grub.CFG_VYOS_VARS}' + grub_cfg_modules = f'{root_dir}/{grub.CFG_VYOS_MODULES}' + grub_cfg_menu = f'{root_dir}/{grub.CFG_VYOS_MENU}' + grub_cfg_options = f'{root_dir}/{grub.CFG_VYOS_OPTIONS}' + + # create new files + render(grub_cfg_main, grub.TMPL_GRUB_MAIN, {}) + grub.common_write(root_dir) + grub.vars_write(grub_cfg_vars, DEFAULT_BOOT_VARS) + grub.modules_write(grub_cfg_modules, []) + grub.write_cfg_ver(1, root_dir) + render(grub_cfg_menu, grub.TMPL_GRUB_MENU, {}) + render(grub_cfg_options, grub.TMPL_GRUB_OPTS, {}) + + +def configure_authentication(config_file: str, password: str) -> None: + """Write encrypted password to config file + + Args: + config_file (str): path of target config file + password (str): plaintext password + + N.B. this can not be deferred by simply setting the plaintext password + and relying on the config mode script to process at boot, as the config + will not automatically be saved in that case, thus leaving the + plaintext exposed + """ + encrypted_password = linux_context.hash(password) + + with open(config_file) as f: + config_string = f.read() + + config = ConfigTree(config_string) + config.set([ + 'system', 'login', 'user', 'vyos', 'authentication', + 'encrypted-password' + ], + value=encrypted_password, + replace=True) + config.set_tag(['system', 'login', 'user']) + + with open(config_file, 'w') as f: + f.write(config.to_string()) + +def validate_signature(file_path: str, sign_type: str) -> None: + """Validate a file by signature and delete a signature file + + Args: + file_path (str): a path to file + sign_type (str): a signature type + """ + print('Validating signature') + signature_valid: bool = False + # validate with minisig + if sign_type == 'minisig': + pub_key_list = glob('/usr/share/vyos/keys/*.minisign.pub') + for pubkey in pub_key_list: + if run(f'minisign -V -q -p {pubkey} -m {file_path} -x {file_path}.minisig' + ) == 0: + signature_valid = True + break + Path(f'{file_path}.minisig').unlink() + # validate with GPG + if sign_type == 'asc': + if run(f'gpg --verify ${file_path}.asc ${file_path}') == 0: + signature_valid = True + Path(f'{file_path}.asc').unlink() + + # warn or pass + if not signature_valid: + if not ask_yes_no(MSG_WARN_ISO_SIGN_INVALID, default=False): + exit(MSG_INFO_INSTALL_EXIT) + else: + print('Signature is valid') + +def download_file(local_file: str, remote_path: str, vrf: str, + username: str, password: str, + progressbar: bool = False, check_space: bool = False): + environ['REMOTE_USERNAME'] = username + environ['REMOTE_PASSWORD'] = password + if vrf is None: + download(local_file, remote_path, progressbar=progressbar, + check_space=check_space, raise_error=True) + else: + vrf_cmd = f'REMOTE_USERNAME={username} REMOTE_PASSWORD={password} \ + ip vrf exec {vrf} {external_download_script} \ + --local-file {local_file} --remote-path {remote_path}' + cmd(vrf_cmd) + +def image_fetch(image_path: str, vrf: str = None, + username: str = '', password: str = '', + no_prompt: bool = False) -> Path: + """Fetch an ISO image + + Args: + image_path (str): a path, remote or local + + Returns: + Path: a path to a local file + """ + # Latest version gets url from configured "system update-check url" + if image_path == 'latest': + config = ConfigTreeQuery() + if config.exists('system update-check url'): + configured_url_version = config.value('system update-check url') + remote_url_list = get_remote_version(configured_url_version) + image_path = remote_url_list[0].get('url') + + try: + # check a type of path + if urlparse(image_path).scheme: + # download an image + download_file(ISO_DOWNLOAD_PATH, image_path, vrf, + username, password, + progressbar=True, check_space=True) + + # download a signature + sign_file = (False, '') + for sign_type in ['minisig', 'asc']: + try: + download_file(f'{ISO_DOWNLOAD_PATH}.{sign_type}', + f'{image_path}.{sign_type}', vrf, + username, password) + sign_file = (True, sign_type) + break + except Exception: + print(f'{sign_type} signature is not available') + # validate a signature if it is available + if sign_file[0]: + validate_signature(ISO_DOWNLOAD_PATH, sign_file[1]) + else: + if (not no_prompt and + not ask_yes_no(MSG_WARN_ISO_SIGN_UNAVAL, default=False)): + cleanup() + exit(MSG_INFO_INSTALL_EXIT) + + return Path(ISO_DOWNLOAD_PATH) + else: + local_path: Path = Path(image_path) + if local_path.is_file(): + return local_path + else: + raise FileNotFoundError + except Exception as e: + print(f'The image cannot be fetched from: {image_path} {e}') + exit(1) + + +def migrate_config() -> bool: + """Check for active config and ask user for migration + + Returns: + bool: user's decision + """ + active_config_path: Path = Path('/opt/vyatta/etc/config/config.boot') + if active_config_path.exists(): + if ask_yes_no(MSG_INPUT_CONFIG_FOUND, default=True): + return True + return False + + +def copy_ssh_host_keys() -> bool: + """Ask user to copy SSH host keys + + Returns: + bool: user's decision + """ + if ask_yes_no('Would you like to copy SSH host keys?', default=True): + return True + return False + + +def cleanup(mounts: list[str] = [], remove_items: list[str] = []) -> None: + """Clean up after installation + + Args: + mounts (list[str], optional): List of mounts to unmount. + Defaults to []. + remove_items (list[str], optional): List of files or directories + to remove. Defaults to []. + """ + print('Cleaning up') + # clean up installation directory by default + mounts_all = disk_partitions(all=True) + for mounted_device in mounts_all: + if mounted_device.mountpoint.startswith(DIR_INSTALLATION) and not ( + mounted_device.device in mounts or + mounted_device.mountpoint in mounts): + mounts.append(mounted_device.mountpoint) + # add installation dir to cleanup list + if DIR_INSTALLATION not in remove_items: + remove_items.append(DIR_INSTALLATION) + # also delete an ISO file + if Path(ISO_DOWNLOAD_PATH).exists( + ) and ISO_DOWNLOAD_PATH not in remove_items: + remove_items.append(ISO_DOWNLOAD_PATH) + + if mounts: + print('Unmounting target filesystems') + for mountpoint in mounts: + disk.partition_umount(mountpoint) + for mountpoint in mounts: + disk.wait_for_umount(mountpoint) + if remove_items: + print('Removing temporary files') + for remove_item in remove_items: + if Path(remove_item).exists(): + if Path(remove_item).is_file(): + Path(remove_item).unlink() + if Path(remove_item).is_dir(): + rmtree(remove_item, ignore_errors=True) + + +def cleanup_raid(details: raid.RaidDetails) -> None: + efiparts = [] + for raid_disk in details.disks: + efiparts.append(raid_disk.partition['efi']) + cleanup([details.name, *efiparts], + ['/mnt/installation']) + + +def is_raid_install(install_object: Union[disk.DiskDetails, raid.RaidDetails]) -> bool: + """Check if installation target is a RAID array + + Args: + install_object (Union[disk.DiskDetails, raid.RaidDetails]): a target disk + + Returns: + bool: True if it is a RAID array + """ + if isinstance(install_object, raid.RaidDetails): + return True + return False + + +def install_image() -> None: + """Install an image to a disk + """ + if not image.is_live_boot(): + exit(MSG_ERR_NOT_LIVE) + + print(MSG_INFO_INSTALL_WELCOME) + if not ask_yes_no('Would you like to continue?'): + print(MSG_INFO_INSTALL_EXIT) + exit() + + # configure image name + running_image_name: str = image.get_running_image() + while True: + image_name: str = ask_input(MSG_INPUT_IMAGE_NAME, + running_image_name) + if image.validate_name(image_name): + break + print(MSG_WARN_IMAGE_NAME_WRONG) + + # ask for password + user_password: str = ask_input(MSG_INPUT_PASSWORD, default='vyos', + no_echo=True) + + # ask for default console + console_type: str = ask_input(MSG_INPUT_CONSOLE_TYPE, + default='K', + valid_responses=['K', 'S', 'U']) + console_dict: dict[str, str] = {'K': 'tty', 'S': 'ttyS', 'U': 'ttyUSB'} + + disks: dict[str, int] = find_disks() + + install_target: Union[disk.DiskDetails, raid.RaidDetails, None] = None + try: + install_target = check_raid_install(disks) + if install_target is None: + install_target = ask_single_disk(disks) + + # create directories for installation media + prepare_tmp_disr() + + # mount target filesystem and create required dirs inside + print('Mounting new partitions') + if is_raid_install(install_target): + disk.partition_mount(install_target.name, DIR_DST_ROOT) + Path(f'{DIR_DST_ROOT}/boot/efi').mkdir(parents=True) + else: + disk.partition_mount(install_target.partition['root'], DIR_DST_ROOT) + Path(f'{DIR_DST_ROOT}/boot/efi').mkdir(parents=True) + disk.partition_mount(install_target.partition['efi'], f'{DIR_DST_ROOT}/boot/efi') + + # a config dir. It is the deepest one, so the comand will + # create all the rest in a single step + print('Creating a configuration file') + target_config_dir: str = f'{DIR_DST_ROOT}/boot/{image_name}/rw/opt/vyatta/etc/config/' + Path(target_config_dir).mkdir(parents=True) + chown(target_config_dir, group='vyattacfg') + chmod_2775(target_config_dir) + # copy config + copy('/opt/vyatta/etc/config/config.boot', target_config_dir) + configure_authentication(f'{target_config_dir}/config.boot', + user_password) + Path(f'{target_config_dir}/.vyatta_config').touch() + + # create a persistence.conf + Path(f'{DIR_DST_ROOT}/persistence.conf').write_text('/ union\n') + + # copy system image and kernel files + print('Copying system image files') + for file in Path(DIR_KERNEL_SRC).iterdir(): + if file.is_file(): + copy(file, f'{DIR_DST_ROOT}/boot/{image_name}/') + copy(FILE_ROOTFS_SRC, + f'{DIR_DST_ROOT}/boot/{image_name}/{image_name}.squashfs') + + # copy saved config data and SSH keys + # owner restored on copy of config data by chmod_2775, above + copy_previous_installation_data(f'{DIR_DST_ROOT}/boot/{image_name}/rw') + + if is_raid_install(install_target): + write_dir: str = f'{DIR_DST_ROOT}/boot/{image_name}/rw' + raid.update_default(write_dir) + + setup_grub(DIR_DST_ROOT) + # add information about version + grub.create_structure() + grub.version_add(image_name, DIR_DST_ROOT) + grub.set_default(image_name, DIR_DST_ROOT) + grub.set_console_type(console_dict[console_type], DIR_DST_ROOT) + + if is_raid_install(install_target): + # add RAID specific modules + grub.modules_write(f'{DIR_DST_ROOT}/{grub.CFG_VYOS_MODULES}', + ['part_msdos', 'part_gpt', 'diskfilter', + 'ext2','mdraid1x']) + # install GRUB + if is_raid_install(install_target): + print('Installing GRUB to the drives') + l = install_target.disks + for disk_target in l: + disk.partition_mount(disk_target.partition['efi'], f'{DIR_DST_ROOT}/boot/efi') + grub.install(disk_target.name, f'{DIR_DST_ROOT}/boot/', + f'{DIR_DST_ROOT}/boot/efi', + id=f'VyOS (RAID disk {l.index(disk_target) + 1})') + disk.partition_umount(disk_target.partition['efi']) + else: + print('Installing GRUB to the drive') + grub.install(install_target.name, f'{DIR_DST_ROOT}/boot/', + f'{DIR_DST_ROOT}/boot/efi') + + # umount filesystems and remove temporary files + if is_raid_install(install_target): + cleanup([install_target.name], + ['/mnt/installation']) + else: + cleanup([install_target.partition['efi'], + install_target.partition['root']], + ['/mnt/installation']) + + # we are done + print(MSG_INFO_INSTALL_SUCCESS) + exit() + + except Exception as err: + print(f'Unable to install VyOS: {err}') + # unmount filesystems and clenup + try: + if install_target is not None: + if is_raid_install(install_target): + cleanup_raid(install_target) + else: + cleanup([install_target.partition['efi'], + install_target.partition['root']], + ['/mnt/installation']) + except Exception as err: + print(f'Cleanup failed: {err}') + + exit(1) + + +@compat.grub_cfg_update +def add_image(image_path: str, vrf: str = None, username: str = '', + password: str = '', no_prompt: bool = False) -> None: + """Add a new image + + Args: + image_path (str): a path to an ISO image + """ + if image.is_live_boot(): + exit(MSG_ERR_LIVE) + + # fetch an image + iso_path: Path = image_fetch(image_path, vrf, username, password, no_prompt) + try: + # mount an ISO + Path(DIR_ISO_MOUNT).mkdir(mode=0o755, parents=True) + disk.partition_mount(iso_path, DIR_ISO_MOUNT, 'iso9660') + + # check sums + print('Validating image checksums') + if not Path(DIR_ISO_MOUNT).joinpath('sha256sum.txt').exists(): + cleanup() + exit(MSG_ERR_IMPROPER_IMAGE) + if run(f'cd {DIR_ISO_MOUNT} && sha256sum --status -c sha256sum.txt'): + cleanup() + exit('Image checksum verification failed.') + + # mount rootfs (to get a system version) + Path(DIR_ROOTFS_SRC).mkdir(mode=0o755, parents=True) + disk.partition_mount(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs', + DIR_ROOTFS_SRC, 'squashfs') + + cfg_ver: str = image.get_image_tools_version(DIR_ROOTFS_SRC) + version_name: str = image.get_image_version(DIR_ROOTFS_SRC) + + disk.partition_umount(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs') + + if cfg_ver < SYSTEM_CFG_VER: + raise compat.DowngradingImageTools( + f'Adding image would downgrade image tools to v.{cfg_ver}; disallowed') + + if not no_prompt: + while True: + image_name: str = ask_input(MSG_INPUT_IMAGE_NAME, version_name) + if image.validate_name(image_name): + break + print(MSG_WARN_IMAGE_NAME_WRONG) + set_as_default: bool = ask_yes_no(MSG_INPUT_IMAGE_DEFAULT, default=True) + else: + image_name: str = version_name + set_as_default: bool = True + + # find target directory + root_dir: str = disk.find_persistence() + + # a config dir. It is the deepest one, so the comand will + # create all the rest in a single step + target_config_dir: str = f'{root_dir}/boot/{image_name}/rw/opt/vyatta/etc/config/' + # copy config + if no_prompt or migrate_config(): + print('Copying configuration directory') + # copytree preserves perms but not ownership: + Path(target_config_dir).mkdir(parents=True) + chown(target_config_dir, group='vyattacfg') + chmod_2775(target_config_dir) + copytree('/opt/vyatta/etc/config/', target_config_dir, + copy_function=copy_preserve_owner, dirs_exist_ok=True) + else: + Path(target_config_dir).mkdir(parents=True) + chown(target_config_dir, group='vyattacfg') + chmod_2775(target_config_dir) + Path(f'{target_config_dir}/.vyatta_config').touch() + + target_ssh_dir: str = f'{root_dir}/boot/{image_name}/rw/etc/ssh/' + if no_prompt or copy_ssh_host_keys(): + print('Copying SSH host keys') + Path(target_ssh_dir).mkdir(parents=True) + host_keys: list[str] = glob('/etc/ssh/ssh_host*') + for host_key in host_keys: + copy(host_key, target_ssh_dir) + + # copy system image and kernel files + print('Copying system image files') + for file in Path(f'{DIR_ISO_MOUNT}/live').iterdir(): + if file.is_file() and (file.match('initrd*') or + file.match('vmlinuz*')): + copy(file, f'{root_dir}/boot/{image_name}/') + copy(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs', + f'{root_dir}/boot/{image_name}/{image_name}.squashfs') + + # unmount an ISO and cleanup + cleanup([str(iso_path)]) + + # add information about version + grub.version_add(image_name, root_dir) + if set_as_default: + grub.set_default(image_name, root_dir) + + except Exception as err: + # unmount an ISO and cleanup + cleanup([str(iso_path)]) + exit(f'Error: {err}') + + +def parse_arguments() -> Namespace: + """Parse arguments + + Returns: + Namespace: a namespace with parsed arguments + """ + parser: ArgumentParser = ArgumentParser( + description='Install new system images') + parser.add_argument('--action', + choices=['install', 'add'], + required=True, + help='action to perform with an image') + parser.add_argument('--vrf', + help='vrf name for image download') + parser.add_argument('--no-prompt', action='store_true', + help='perform action non-interactively') + parser.add_argument('--username', default='', + help='username for image download') + parser.add_argument('--password', default='', + help='password for image download') + parser.add_argument('--image-path', + help='a path (HTTP or local file) to an image that needs to be installed' + ) + # parser.add_argument('--image_new_name', help='a new name for image') + args: Namespace = parser.parse_args() + # Validate arguments + if args.action == 'add' and not args.image_path: + exit('A path to image is required for add action') + + return args + + +if __name__ == '__main__': + try: + args: Namespace = parse_arguments() + if args.action == 'install': + install_image() + if args.action == 'add': + add_image(args.image_path, args.vrf, + args.username, args.password, args.no_prompt) + + exit() + + except KeyboardInterrupt: + print('Stopped by Ctrl+C') + cleanup() + exit() + + except Exception as err: + exit(f'{err}') diff --git a/src/op_mode/image_manager.py b/src/op_mode/image_manager.py new file mode 100755 index 000000000..e64a85b95 --- /dev/null +++ b/src/op_mode/image_manager.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +# +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This file is part of VyOS. +# +# VyOS 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. +# +# VyOS 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 +# VyOS. If not, see <https://www.gnu.org/licenses/>. + +from argparse import ArgumentParser, Namespace +from pathlib import Path +from shutil import rmtree +from sys import exit +from typing import Optional + +from vyos.system import disk, grub, image, compat +from vyos.utils.io import ask_yes_no, select_entry + +SET_IMAGE_LIST_MSG: str = 'The following images are available:' +SET_IMAGE_PROMPT_MSG: str = 'Select an image to set as default:' +DELETE_IMAGE_LIST_MSG: str = 'The following images are installed:' +DELETE_IMAGE_PROMPT_MSG: str = 'Select an image to delete:' +MSG_DELETE_IMAGE_RUNNING: str = 'Currently running image cannot be deleted; reboot into another image first' +MSG_DELETE_IMAGE_DEFAULT: str = 'Default image cannot be deleted; set another image as default first' + +def annotated_list(images_list: list[str]) -> list[str]: + """Annotate list of images with additional info + + Args: + images_list (list[str]): a list of image names + + Returns: + list[str]: a list of image names with additional info + """ + index_running: int = None + index_default: int = None + try: + index_running = images_list.index(image.get_running_image()) + index_default = images_list.index(image.get_default_image()) + except ValueError: + pass + if index_running is not None: + images_list[index_running] += ' (running)' + if index_default is not None: + images_list[index_default] += ' (default boot)' + return images_list + +@compat.grub_cfg_update +def delete_image(image_name: Optional[str] = None, + no_prompt: bool = False) -> None: + """Remove installed image files and boot entry + + Args: + image_name (str): a name of image to delete + """ + available_images: list[str] = annotated_list(grub.version_list()) + if image_name is None: + if no_prompt: + exit('An image name is required for delete action') + else: + image_name = select_entry(available_images, + DELETE_IMAGE_LIST_MSG, + DELETE_IMAGE_PROMPT_MSG) + if image_name == image.get_running_image(): + exit(MSG_DELETE_IMAGE_RUNNING) + if image_name == image.get_default_image(): + exit(MSG_DELETE_IMAGE_DEFAULT) + if image_name not in available_images: + exit(f'The image "{image_name}" cannot be found') + persistence_storage: str = disk.find_persistence() + if not persistence_storage: + exit('Persistence storage cannot be found') + + if (not no_prompt and + not ask_yes_no(f'Do you really want to delete the image {image_name}?', + default=False)): + exit() + + # remove files and menu entry + version_path: Path = Path(f'{persistence_storage}/boot/{image_name}') + try: + rmtree(version_path) + grub.version_del(image_name, persistence_storage) + print(f'The image "{image_name}" was successfully deleted') + except Exception as err: + exit(f'Unable to remove the image "{image_name}": {err}') + + +@compat.grub_cfg_update +def set_image(image_name: Optional[str] = None, + prompt: bool = True) -> None: + """Set default boot image + + Args: + image_name (str): an image name + """ + available_images: list[str] = annotated_list(grub.version_list()) + if image_name is None: + if not prompt: + exit('An image name is required for set action') + else: + image_name = select_entry(available_images, + SET_IMAGE_LIST_MSG, + SET_IMAGE_PROMPT_MSG) + if image_name == image.get_default_image(): + exit(f'The image "{image_name}" already configured as default') + if image_name not in available_images: + exit(f'The image "{image_name}" cannot be found') + persistence_storage: str = disk.find_persistence() + if not persistence_storage: + exit('Persistence storage cannot be found') + + # set default boot image + try: + grub.set_default(image_name, persistence_storage) + print(f'The image "{image_name}" is now default boot image') + except Exception as err: + exit(f'Unable to set default image "{image_name}": {err}') + + +@compat.grub_cfg_update +def rename_image(name_old: str, name_new: str) -> None: + """Rename installed image + + Args: + name_old (str): old name + name_new (str): new name + """ + if name_old == image.get_running_image(): + exit('Currently running image cannot be renamed') + available_images: list[str] = grub.version_list() + if name_old not in available_images: + exit(f'The image "{name_old}" cannot be found') + if name_new in available_images: + exit(f'The image "{name_new}" already exists') + if not image.validate_name(name_new): + exit(f'The image name "{name_new}" is not allowed') + + persistence_storage: str = disk.find_persistence() + if not persistence_storage: + exit('Persistence storage cannot be found') + + if not ask_yes_no( + f'Do you really want to rename the image {name_old} ' + f'to the {name_new}?', + default=False): + exit() + + try: + # replace default boot item + if name_old == image.get_default_image(): + grub.set_default(name_new, persistence_storage) + + # rename files and dirs + old_path: Path = Path(f'{persistence_storage}/boot/{name_old}') + new_path: Path = Path(f'{persistence_storage}/boot/{name_new}') + old_path.rename(new_path) + + # replace boot item + grub.version_del(name_old, persistence_storage) + grub.version_add(name_new, persistence_storage) + + print(f'The image "{name_old}" was renamed to "{name_new}"') + except Exception as err: + exit(f'Unable to rename image "{name_old}" to "{name_new}": {err}') + + +def list_images() -> None: + """Print list of available images for CLI hints""" + images_list: list[str] = grub.version_list() + for image_name in images_list: + print(image_name) + + +def parse_arguments() -> Namespace: + """Parse arguments + + Returns: + Namespace: a namespace with parsed arguments + """ + parser: ArgumentParser = ArgumentParser(description='Manage system images') + parser.add_argument('--action', + choices=['delete', 'set', 'rename', 'list'], + required=True, + help='action to perform with an image') + parser.add_argument('--no-prompt', action='store_true', + help='perform action non-interactively') + parser.add_argument( + '--image-name', + help= + 'a name of an image to add, delete, install, rename, or set as default') + parser.add_argument('--image-new-name', help='a new name for image') + args: Namespace = parser.parse_args() + # Validate arguments + if args.action == 'rename' and (not args.image_name or + not args.image_new_name): + exit('Both old and new image names are required for rename action') + + return args + + +if __name__ == '__main__': + try: + args: Namespace = parse_arguments() + if args.action == 'delete': + delete_image(args.image_name, args.no_prompt) + if args.action == 'set': + set_image(args.image_name) + if args.action == 'rename': + rename_image(args.image_name, args.image_new_name) + if args.action == 'list': + list_images() + + exit() + + except KeyboardInterrupt: + print('Stopped by Ctrl+C') + exit() + + except Exception as err: + exit(f'{err}') diff --git a/src/op_mode/interfaces.py b/src/op_mode/interfaces.py index 782e178c6..14ffdca9f 100755 --- a/src/op_mode/interfaces.py +++ b/src/op_mode/interfaces.py @@ -235,6 +235,11 @@ def _get_summary_data(ifname: typing.Optional[str], if iftype is None: iftype = '' ret = [] + + def is_interface_has_mac(interface_name): + interface_no_mac = ('tun', 'wg') + return not any(interface_name.startswith(prefix) for prefix in interface_no_mac) + for interface in filtered_interfaces(ifname, iftype, vif, vrrp): res_intf = {} @@ -243,6 +248,9 @@ def _get_summary_data(ifname: typing.Optional[str], res_intf['admin_state'] = interface.get_admin_state() res_intf['addr'] = [_ for _ in interface.get_addr() if not _.startswith('fe80::')] res_intf['description'] = interface.get_alias() + res_intf['mtu'] = interface.get_mtu() + res_intf['mac'] = interface.get_mac() if is_interface_has_mac(interface.ifname) else 'n/a' + res_intf['vrf'] = interface.get_vrf() ret.append(res_intf) @@ -373,6 +381,51 @@ def _format_show_summary(data): return 0 @catch_broken_pipe +def _format_show_summary_extended(data): + headers = ["Interface", "IP Address", "MAC", "VRF", "MTU", "S/L", "Description"] + table_data = [] + + print('Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down') + + for intf in data: + if 'unhandled' in intf: + continue + + ifname = intf['ifname'] + oper_state = 'u' if intf['oper_state'] in ('up', 'unknown') else 'D' + admin_state = 'u' if intf['admin_state'] in ('up', 'unknown') else 'A' + addrs = intf['addr'] or ['-'] + description = '\n'.join(_split_text(intf['description'], 0)) + mac = intf['mac'] if intf['mac'] else 'n/a' + mtu = intf['mtu'] if intf['mtu'] else 'n/a' + vrf = intf['vrf'] if intf['vrf'] else 'default' + + ip_addresses = '\n'.join(ip for ip in addrs) + + # Create a row for the table + row = [ + ifname, + ip_addresses, + mac, + vrf, + mtu, + f"{admin_state}/{oper_state}", + description, + ] + + # Append the row to the table data + table_data.append(row) + + for intf in data: + if 'unhandled' in intf: + string = {'C': 'u/D', 'D': 'A/D'}[intf['state']] + table_data.append([intf['ifname'], '', '', '', '', string, '']) + + print(tabulate(table_data, headers)) + + return 0 + +@catch_broken_pipe def _format_show_counters(data: list): data_entries = [] for entry in data: @@ -408,6 +461,14 @@ def show_summary(raw: bool, intf_name: typing.Optional[str], return data return _format_show_summary(data) +def show_summary_extended(raw: bool, intf_name: typing.Optional[str], + intf_type: typing.Optional[str], + vif: bool, vrrp: bool): + data = _get_summary_data(intf_name, intf_type, vif, vrrp) + if raw: + return data + return _format_show_summary_extended(data) + def show_counters(raw: bool, intf_name: typing.Optional[str], intf_type: typing.Optional[str], vif: bool, vrrp: bool): diff --git a/src/op_mode/interfaces_wireless.py b/src/op_mode/interfaces_wireless.py new file mode 100755 index 000000000..259fd3900 --- /dev/null +++ b/src/op_mode/interfaces_wireless.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2023 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 re +import sys +import typing +import vyos.opmode + +from copy import deepcopy +from tabulate import tabulate +from vyos.utils.process import popen +from vyos.configquery import ConfigTreeQuery + +def _verify(func): + """Decorator checks if Wireless LAN config exists""" + from functools import wraps + + @wraps(func) + def _wrapper(*args, **kwargs): + config = ConfigTreeQuery() + if not config.exists(['interfaces', 'wireless']): + unconf_message = 'No Wireless interfaces configured' + raise vyos.opmode.UnconfiguredSubsystem(unconf_message) + return func(*args, **kwargs) + return _wrapper + +def _get_raw_info_data(): + output_data = [] + + config = ConfigTreeQuery() + raw = config.get_config_dict(['interfaces', 'wireless'], effective=True, + get_first_key=True, key_mangling=('-', '_')) + for interface, interface_config in raw.items(): + tmp = {'name' : interface} + + if 'type' in interface_config: + tmp.update({'type' : interface_config['type']}) + else: + tmp.update({'type' : '-'}) + + if 'ssid' in interface_config: + tmp.update({'ssid' : interface_config['ssid']}) + else: + tmp.update({'ssid' : '-'}) + + if 'channel' in interface_config: + tmp.update({'channel' : interface_config['channel']}) + else: + tmp.update({'channel' : '-'}) + + output_data.append(tmp) + + return output_data + +def _get_formatted_info_output(raw_data): + output=[] + for ssid in raw_data: + output.append([ssid['name'], ssid['type'], ssid['ssid'], ssid['channel']]) + + headers = ["Interface", "Type", "SSID", "Channel"] + print(tabulate(output, headers, numalign="left")) + +def _get_raw_scan_data(intf_name): + # XXX: This ignores errors + tmp, _ = popen(f'iw dev {intf_name} scan ap-force') + networks = [] + data = { + 'ssid': '', + 'mac': '', + 'channel': '', + 'signal': '' + } + re_mac = re.compile(r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})') + for line in tmp.splitlines(): + if line.startswith('BSS '): + ssid = deepcopy(data) + ssid['mac'] = re.search(re_mac, line).group() + + elif line.lstrip().startswith('SSID: '): + # SSID can be " SSID: WLAN-57 6405", thus strip all leading whitespaces + ssid['ssid'] = line.lstrip().split(':')[-1].lstrip() + + elif line.lstrip().startswith('signal: '): + # Siganl can be " signal: -67.00 dBm", thus strip all leading whitespaces + ssid['signal'] = line.lstrip().split(':')[-1].split()[0] + + elif line.lstrip().startswith('DS Parameter set: channel'): + # Channel can be " DS Parameter set: channel 6" , thus + # strip all leading whitespaces + ssid['channel'] = line.lstrip().split(':')[-1].split()[-1] + networks.append(ssid) + continue + + return networks + +def _format_scan_data(raw_data): + output=[] + for ssid in raw_data: + output.append([ssid['mac'], ssid['ssid'], ssid['channel'], ssid['signal']]) + headers = ["Address", "SSID", "Channel", "Signal (dbm)"] + return tabulate(output, headers, numalign="left") + +def _get_raw_station_data(intf_name): + # XXX: This ignores errors + tmp, _ = popen(f'iw dev {intf_name} station dump') + clients = [] + data = { + 'mac': '', + 'signal': '', + 'rx_bytes': '', + 'rx_packets': '', + 'tx_bytes': '', + 'tx_packets': '' + } + re_mac = re.compile(r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})') + for line in tmp.splitlines(): + if line.startswith('Station'): + client = deepcopy(data) + client['mac'] = re.search(re_mac, line).group() + + elif line.lstrip().startswith('signal avg:'): + client['signal'] = line.lstrip().split(':')[-1].lstrip().split()[0] + + elif line.lstrip().startswith('rx bytes:'): + client['rx_bytes'] = line.lstrip().split(':')[-1].lstrip() + + elif line.lstrip().startswith('rx packets:'): + client['rx_packets'] = line.lstrip().split(':')[-1].lstrip() + + elif line.lstrip().startswith('tx bytes:'): + client['tx_bytes'] = line.lstrip().split(':')[-1].lstrip() + + elif line.lstrip().startswith('tx packets:'): + client['tx_packets'] = line.lstrip().split(':')[-1].lstrip() + clients.append(client) + continue + + return clients + +def _format_station_data(raw_data): + output=[] + for ssid in raw_data: + output.append([ssid['mac'], ssid['signal'], ssid['rx_bytes'], ssid['rx_packets'], ssid['tx_bytes'], ssid['tx_packets']]) + headers = ["Station", "Signal", "RX bytes", "RX packets", "TX bytes", "TX packets"] + return tabulate(output, headers, numalign="left") + +@_verify +def show_info(raw: bool): + info_data = _get_raw_info_data() + if raw: + return info_data + return _get_formatted_info_output(info_data) + +def show_scan(raw: bool, intf_name: str): + data = _get_raw_scan_data(intf_name) + if raw: + return data + return _format_scan_data(data) + +@_verify +def show_stations(raw: bool, intf_name: str): + data = _get_raw_station_data(intf_name) + if raw: + return data + return _format_station_data(data) + +if __name__ == '__main__': + try: + res = vyos.opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, vyos.opmode.Error) as e: + print(e) + sys.exit(1) diff --git a/src/op_mode/lldp.py b/src/op_mode/lldp.py index c287b8fa6..58cfce443 100755 --- a/src/op_mode/lldp.py +++ b/src/op_mode/lldp.py @@ -114,7 +114,10 @@ def _get_formatted_output(raw_data): # Remote software platform platform = jmespath.search('chassis.[*][0][0].descr', values) - tmp.append(platform[:37]) + if platform: + tmp.append(platform[:37]) + else: + tmp.append('') # Remote interface interface = jmespath.search('port.descr', values) diff --git a/src/op_mode/mtr.py b/src/op_mode/mtr.py new file mode 100644 index 000000000..de139f2fa --- /dev/null +++ b/src/op_mode/mtr.py @@ -0,0 +1,306 @@ +#! /usr/bin/env python3 + +# Copyright (C) 2023 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 +import socket +import ipaddress + +from vyos.utils.network import interface_list +from vyos.utils.network import vrf_list +from vyos.utils.process import call + +options = { + 'report': { + 'mtr': '{command} --report', + 'type': 'noarg', + 'help': 'This option puts mtr into report mode. When in this mode, mtr will run for the number of cycles specified by the -c option, and then print statistics and exit.' + }, + 'report-wide': { + 'mtr': '{command} --report-wide', + 'type': 'noarg', + 'help': 'This option puts mtr into wide report mode. When in this mode, mtr will not cut hostnames in the report.' + }, + 'raw': { + 'mtr': '{command} --raw', + 'type': 'noarg', + 'help': 'Use the raw output format. This format is better suited for archival of the measurement results.' + }, + 'json': { + 'mtr': '{command} --json', + 'type': 'noarg', + 'help': 'Use this option to tell mtr to use the JSON output format.' + }, + 'split': { + 'mtr': '{command} --split', + 'type': 'noarg', + 'help': 'Use this option to set mtr to spit out a format that is suitable for a split-user interface.' + }, + 'no-dns': { + 'mtr': '{command} --no-dns', + 'type': 'noarg', + 'help': 'Use this option to force mtr to display numeric IP numbers and not try to resolve the host names.' + }, + 'show-ips': { + 'mtr': '{command} --show-ips {value}', + 'type': '<num>', + 'help': 'Use this option to tell mtr to display both the host names and numeric IP numbers.' + }, + 'ipinfo': { + 'mtr': '{command} --ipinfo {value}', + 'type': '<num>', + 'help': 'Displays information about each IP hop.' + }, + 'aslookup': { + 'mtr': '{command} --aslookup', + 'type': 'noarg', + 'help': 'Displays the Autonomous System (AS) number alongside each hop. Equivalent to --ipinfo 0.' + }, + 'interval': { + 'mtr': '{command} --interval {value}', + 'type': '<num>', + 'help': 'Use this option to specify the positive number of seconds between ICMP ECHO requests. The default value for this parameter is one second. The root user may choose values between zero and one.' + }, + 'report-cycles': { + 'mtr': '{command} --report-cycles {value}', + 'type': '<num>', + 'help': 'Use this option to set the number of pings sent to determine both the machines on the network and the reliability of those machines. Each cycle lasts one second.' + }, + 'psize': { + 'mtr': '{command} --psize {value}', + 'type': '<num>', + 'help': 'This option sets the packet size used for probing. It is in bytes, inclusive IP and ICMP headers. If set to a negative number, every iteration will use a different, random packet size up to that number.' + }, + 'bitpattern': { + 'mtr': '{command} --bitpattern {value}', + 'type': '<num>', + 'help': 'Specifies bit pattern to use in payload. Should be within range 0 - 255. If NUM is greater than 255, a random pattern is used.' + }, + 'gracetime': { + 'mtr': '{command} --gracetime {value}', + 'type': '<num>', + 'help': 'Use this option to specify the positive number of seconds to wait for responses after the final request. The default value is five seconds.' + }, + 'tos': { + 'mtr': '{command} --tos {value}', + 'type': '<tos>', + 'help': 'Specifies value for type of service field in IP header. Should be within range 0 - 255.' + }, + 'mpls': { + 'mtr': '{command} --mpls {value}', + 'type': 'noarg', + 'help': 'Use this option to tell mtr to display information from ICMP extensions for MPLS (RFC 4950) that are encoded in the response packets.' + }, + 'interface': { + 'mtr': '{command} --interface {value}', + 'type': '<interface>', + 'helpfunction': interface_list, + 'help': 'Use the network interface with a specific name for sending network probes. This can be useful when you have multiple network interfaces with routes to your destination, for example both wired Ethernet and WiFi, and wish to test a particular interface.' + }, + 'address': { + 'mtr': '{command} --address {value}', + 'type': '<x.x.x.x> <h:h:h:h:h:h:h:h>', + 'help': 'Use this option to bind the outgoing socket to ADDRESS, so that all packets will be sent with ADDRESS as source address.' + }, + 'first-ttl': { + 'mtr': '{command} --first-ttl {value}', + 'type': '<num>', + 'help': 'Specifies with what TTL to start. Defaults to 1.' + }, + 'max-ttl': { + 'mtr': '{command} --max-ttl {value}', + 'type': '<num>', + 'help': 'Specifies the maximum number of hops or max time-to-live value mtr will probe. Default is 30.' + }, + 'max-unknown': { + 'mtr': '{command} --max-unknown {value}', + 'type': '<num>', + 'help': 'Specifies the maximum unknown host. Default is 5.' + }, + 'udp': { + 'mtr': '{command} --udp', + 'type': 'noarg', + 'help': 'Use UDP datagrams instead of ICMP ECHO.' + }, + 'tcp': { + 'mtr': '{command} --tcp', + 'type': 'noarg', + 'help': ' Use TCP SYN packets instead of ICMP ECHO. PACKETSIZE is ignored, since SYN packets can not contain data.' + }, + 'sctp': { + 'mtr': '{command} --sctp', + 'type': 'noarg', + 'help': 'Use Stream Control Transmission Protocol packets instead of ICMP ECHO.' + }, + 'port': { + 'mtr': '{command} --port {value}', + 'type': '<port>', + 'help': 'The target port number for TCP/SCTP/UDP traces.' + }, + 'localport': { + 'mtr': '{command} --localport {value}', + 'type': '<port>', + 'help': 'The source port number for UDP traces.' + }, + 'timeout': { + 'mtr': '{command} --timeout {value}', + 'type': '<num>', + 'help': ' The number of seconds to keep probe sockets open before giving up on the connection.' + }, + 'mark': { + 'mtr': '{command} --mark {value}', + 'type': '<num>', + 'help': ' Set the mark for each packet sent through this socket similar to the netfilter MARK target but socket-based. MARK is 32 unsigned integer.' + }, + 'vrf': { + 'mtr': 'sudo ip vrf exec {value} {command}', + 'type': '<vrf>', + 'help': 'Use specified VRF table', + 'helpfunction': vrf_list, + 'dflt': 'default' + } + } + +mtr = { + 4: '/bin/mtr -4', + 6: '/bin/mtr -6', +} + +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 expension_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: + expension_failure(shortname, longnames) + longname = longnames[0] + if options[longname]['type'] == 'noarg': + command = options[longname]['mtr'].format( + command=command, value='') + elif not args: + sys.exit(f'mtr: missing argument for {longname} option') + else: + command = options[longname]['mtr'].format( + command=command, value=args.first()) + return command + + +if __name__ == '__main__': + args = List(sys.argv[1:]) + host = args.first() + + if not host: + sys.exit("mtr: Missing host") + + + if host == '--get-options' or host == '--get-options-nested': + if host == '--get-options-nested': + args.first() # pop monitor + args.first() # pop mtr | traceroute + args.first() # pop IP + 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) + 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) + + for name, option in options.items(): + if 'dflt' in option and name not in args: + args.append(name) + args.append(option['dflt']) + + try: + ip = socket.gethostbyname(host) + except UnicodeError: + sys.exit(f'mtr: Unknown host: {host}') + except socket.gaierror: + ip = host + + try: + version = ipaddress.ip_address(ip).version + except ValueError: + sys.exit(f'mtr: Unknown host: {host}') + + command = convert(mtr[version], args) + call(f'{command} --curses --displaymode 0 {host}') diff --git a/src/op_mode/multicast.py b/src/op_mode/multicast.py new file mode 100755 index 000000000..0666f8af3 --- /dev/null +++ b/src/op_mode/multicast.py @@ -0,0 +1,72 @@ +#!/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 json +import sys +import typing + +from tabulate import tabulate +from vyos.utils.process import cmd + +import vyos.opmode + +ArgFamily = typing.Literal['inet', 'inet6'] + +def _get_raw_data(family, interface=None): + tmp = 'ip -4' + if family == 'inet6': + tmp = 'ip -6' + tmp = f'{tmp} -j maddr show' + if interface: + tmp = f'{tmp} dev {interface}' + output = cmd(tmp) + data = json.loads(output) + if not data: + return [] + return data + +def _get_formatted_output(raw_data): + data_entries = [] + + # sort result by interface name + for interface in sorted(raw_data, key=lambda x: x['ifname']): + for address in interface['maddr']: + tmp = [] + tmp.append(interface['ifname']) + tmp.append(address['family']) + tmp.append(address['address']) + + data_entries.append(tmp) + + headers = ["Interface", "Family", "Address"] + output = tabulate(data_entries, headers, numalign="left") + return output + +def show_group(raw: bool, family: ArgFamily, interface: typing.Optional[str]): + multicast_data = _get_raw_data(family=family, interface=interface) + if raw: + return multicast_data + else: + return _get_formatted_output(multicast_data) + +if __name__ == "__main__": + try: + res = vyos.opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, vyos.opmode.Error) as e: + print(e) + sys.exit(1) diff --git a/src/op_mode/nat.py b/src/op_mode/nat.py index 71a40c0e1..2bc7e24fe 100755 --- a/src/op_mode/nat.py +++ b/src/op_mode/nat.py @@ -28,9 +28,6 @@ from vyos.configquery import ConfigTreeQuery from vyos.utils.process import cmd from vyos.utils.dict import dict_search -base = 'nat' -unconf_message = 'NAT is not configured' - ArgDirection = typing.Literal['source', 'destination'] ArgFamily = typing.Literal['inet', 'inet6'] @@ -293,8 +290,9 @@ def _verify(func): @wraps(func) def _wrapper(*args, **kwargs): config = ConfigTreeQuery() + base = 'nat66' if 'inet6' in sys.argv[1:] else 'nat' if not config.exists(base): - raise vyos.opmode.UnconfiguredSubsystem(unconf_message) + raise vyos.opmode.UnconfiguredSubsystem(f'{base.upper()} is not configured') return func(*args, **kwargs) return _wrapper diff --git a/src/op_mode/ping.py b/src/op_mode/ping.py index f1d87a118..583d8792c 100755 --- a/src/op_mode/ping.py +++ b/src/op_mode/ping.py @@ -1,6 +1,6 @@ #! /usr/bin/env python3 -# Copyright (C) 2020 VyOS maintainers and contributors +# Copyright (C) 2023 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 @@ -14,29 +14,13 @@ # 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 os import sys import socket import ipaddress -from vyos.utils.network import get_all_vrfs -from vyos.ifconfig import Section - - -def interface_list() -> list: - """ - Get list of interfaces in system - :rtype: list - """ - return Section.interfaces() - - -def vrf_list() -> list: - """ - Get list of VRFs in system - :rtype: list - """ - return list(get_all_vrfs().keys()) +from vyos.utils.network import interface_list +from vyos.utils.network import vrf_list +from vyos.utils.process import call options = { 'audible': { @@ -295,6 +279,4 @@ if __name__ == '__main__': sys.exit(f'ping: Unknown host: {host}') command = convert(ping[version], args) - - # print(f'{command} {host}') - os.system(f'{command} {host}') + call(f'{command} {host}') diff --git a/src/op_mode/pki.py b/src/op_mode/pki.py index 35c7ce0e2..ad2c1ada0 100755 --- a/src/op_mode/pki.py +++ b/src/op_mode/pki.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2023 VyOS maintainers and contributors +# Copyright (C) 2021-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 @@ -25,6 +25,7 @@ from cryptography import x509 from cryptography.x509.oid import ExtendedKeyUsageOID from vyos.config import Config +from vyos.config import config_dict_mangle_acme from vyos.pki import encode_certificate, encode_public_key, encode_private_key, encode_dh_parameters from vyos.pki import get_certificate_fingerprint from vyos.pki import create_certificate, create_certificate_request, create_certificate_revocation_list @@ -79,9 +80,14 @@ def get_config_certificate(name=None): if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']): return False - return conf.get_config_dict(base, key_mangling=('-', '_'), + pki = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True) + if pki: + for certificate in pki: + pki[certificate] = config_dict_mangle_acme(certificate, pki[certificate]) + + return pki def get_certificate_ca(cert, ca_certs): # Find CA certificate for given certificate @@ -896,11 +902,15 @@ def show_certificate(name=None, pem=False): cert_subject_cn = cert.subject.rfc4514_string().split(",")[0] cert_issuer_cn = cert.issuer.rfc4514_string().split(",")[0] cert_type = 'Unknown' - ext = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage) - if ext and ExtendedKeyUsageOID.SERVER_AUTH in ext.value: - cert_type = 'Server' - elif ext and ExtendedKeyUsageOID.CLIENT_AUTH in ext.value: - cert_type = 'Client' + + try: + ext = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage) + if ext and ExtendedKeyUsageOID.SERVER_AUTH in ext.value: + cert_type = 'Server' + elif ext and ExtendedKeyUsageOID.CLIENT_AUTH in ext.value: + cert_type = 'Client' + except: + pass revoked = 'Yes' if 'revoke' in cert_dict else 'No' have_private = 'Yes' if 'private' in cert_dict and 'key' in cert_dict['private'] else 'No' @@ -1069,7 +1079,9 @@ if __name__ == '__main__': show_crl(None if args.crl == 'all' else args.crl, args.pem) else: show_certificate_authority() + print('\n') show_certificate() + print('\n') show_crl() except KeyboardInterrupt: print("Aborted") diff --git a/src/op_mode/powerctrl.py b/src/op_mode/powerctrl.py index 3ac5991b4..c07d0c4bd 100755 --- a/src/op_mode/powerctrl.py +++ b/src/op_mode/powerctrl.py @@ -191,7 +191,7 @@ def main(): nargs="*", metavar="HH:MM") - action.add_argument("--reboot_in", "-i", + action.add_argument("--reboot-in", "-i", help="Reboot the system", nargs="*", metavar="Minutes") @@ -214,7 +214,7 @@ def main(): if args.reboot is not None: for r in args.reboot: if ':' not in r and '/' not in r and '.' not in r: - print("Incorrect format! Use HH:MM") + print("Incorrect format! Use HH:MM") exit(1) execute_shutdown(args.reboot, reboot=True, ask=args.yes) if args.reboot_in is not None: diff --git a/src/op_mode/restart_frr.py b/src/op_mode/restart_frr.py index 820a3846c..8841b0eca 100755 --- a/src/op_mode/restart_frr.py +++ b/src/op_mode/restart_frr.py @@ -139,9 +139,7 @@ def _reload_config(daemon): # define program arguments cmd_args_parser = argparse.ArgumentParser(description='restart frr daemons') cmd_args_parser.add_argument('--action', choices=['restart'], required=True, help='action to frr daemons') -# Full list of FRR 9.0/stable daemons for reference -#cmd_args_parser.add_argument('--daemon', choices=['zebra', 'staticd', 'bgpd', 'ospfd', 'ospf6d', 'ripd', 'ripngd', 'isisd', 'pim6d', 'ldpd', 'eigrpd', 'babeld', 'sharpd', 'bfdd', 'fabricd', 'pathd'], required=False, nargs='*', help='select single or multiple daemons') -cmd_args_parser.add_argument('--daemon', choices=['zebra', 'staticd', 'bgpd', 'ospfd', 'ospf6d', 'ripd', 'ripngd', 'isisd', 'pim6d', 'ldpd', 'babeld', 'bfdd'], required=False, nargs='*', help='select single or multiple daemons') +cmd_args_parser.add_argument('--daemon', choices=['zebra', 'staticd', 'bgpd', 'eigrpd', 'ospfd', 'ospf6d', 'ripd', 'ripngd', 'isisd', 'pimd', 'pim6d', 'ldpd', 'babeld', 'bfdd'], required=False, nargs='*', help='select single or multiple daemons') # parse arguments cmd_args = cmd_args_parser.parse_args() diff --git a/src/op_mode/show_openvpn.py b/src/op_mode/show_openvpn.py index e29e594a5..6abafc8b6 100755 --- a/src/op_mode/show_openvpn.py +++ b/src/op_mode/show_openvpn.py @@ -63,9 +63,11 @@ def get_vpn_tunnel_address(peer, interface): # filter out subnet entries lst = [l for l in lst[1:] if '/' not in l.split(',')[0]] - tunnel_ip = lst[0].split(',')[0] + if lst: + tunnel_ip = lst[0].split(',')[0] + return tunnel_ip - return tunnel_ip + return 'n/a' def get_status(mode, interface): status_file = '/var/run/openvpn/{}.status'.format(interface) diff --git a/src/op_mode/show_wireless.py b/src/op_mode/show_wireless.py deleted file mode 100755 index 340163057..000000000 --- a/src/op_mode/show_wireless.py +++ /dev/null @@ -1,149 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2019-2023 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 - -from sys import exit -from copy import deepcopy - -from vyos.config import Config -from vyos.utils.process import popen - -parser = argparse.ArgumentParser() -parser.add_argument("-s", "--scan", help="Scan for Wireless APs on given interface, e.g. 'wlan0'") -parser.add_argument("-b", "--brief", action="store_true", help="Show wireless configuration") -parser.add_argument("-c", "--stations", help="Show wireless clients connected on interface, e.g. 'wlan0'") - -def show_brief(): - config = Config() - if len(config.list_effective_nodes('interfaces wireless')) == 0: - print("No Wireless interfaces configured") - exit(0) - - interfaces = [] - for intf in config.list_effective_nodes('interfaces wireless'): - config.set_level(f'interfaces wireless {intf}') - data = { 'name': intf } - data['type'] = config.return_effective_value('type') or '-' - data['ssid'] = config.return_effective_value('ssid') or '-' - data['channel'] = config.return_effective_value('channel') or '-' - interfaces.append(data) - - return interfaces - -def ssid_scan(intf): - # XXX: This ignores errors - tmp, _ = popen(f'/sbin/iw dev {intf} scan ap-force') - networks = [] - data = { - 'ssid': '', - 'mac': '', - 'channel': '', - 'signal': '' - } - re_mac = re.compile(r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})') - for line in tmp.splitlines(): - if line.startswith('BSS '): - ssid = deepcopy(data) - ssid['mac'] = re.search(re_mac, line).group() - - elif line.lstrip().startswith('SSID: '): - # SSID can be " SSID: WLAN-57 6405", thus strip all leading whitespaces - ssid['ssid'] = line.lstrip().split(':')[-1].lstrip() - - elif line.lstrip().startswith('signal: '): - # Siganl can be " signal: -67.00 dBm", thus strip all leading whitespaces - ssid['signal'] = line.lstrip().split(':')[-1].split()[0] - - elif line.lstrip().startswith('DS Parameter set: channel'): - # Channel can be " DS Parameter set: channel 6" , thus - # strip all leading whitespaces - ssid['channel'] = line.lstrip().split(':')[-1].split()[-1] - networks.append(ssid) - continue - - return networks - -def show_clients(intf): - # XXX: This ignores errors - tmp, _ = popen(f'/sbin/iw dev {intf} station dump') - clients = [] - data = { - 'mac': '', - 'signal': '', - 'rx_bytes': '', - 'rx_packets': '', - 'tx_bytes': '', - 'tx_packets': '' - } - re_mac = re.compile(r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})') - for line in tmp.splitlines(): - if line.startswith('Station'): - client = deepcopy(data) - client['mac'] = re.search(re_mac, line).group() - - elif line.lstrip().startswith('signal avg:'): - client['signal'] = line.lstrip().split(':')[-1].lstrip().split()[0] - - elif line.lstrip().startswith('rx bytes:'): - client['rx_bytes'] = line.lstrip().split(':')[-1].lstrip() - - elif line.lstrip().startswith('rx packets:'): - client['rx_packets'] = line.lstrip().split(':')[-1].lstrip() - - elif line.lstrip().startswith('tx bytes:'): - client['tx_bytes'] = line.lstrip().split(':')[-1].lstrip() - - elif line.lstrip().startswith('tx packets:'): - client['tx_packets'] = line.lstrip().split(':')[-1].lstrip() - clients.append(client) - continue - - return clients - -if __name__ == '__main__': - args = parser.parse_args() - - if args.scan: - print("Address SSID Channel Signal (dbm)") - for network in ssid_scan(args.scan): - print("{:<17} {:<32} {:>3} {}".format(network['mac'], - network['ssid'], - network['channel'], - network['signal'])) - exit(0) - - elif args.brief: - print("Interface Type SSID Channel") - for intf in show_brief(): - print("{:<9} {:<12} {:<32} {:>3}".format(intf['name'], - intf['type'], - intf['ssid'], - intf['channel'])) - exit(0) - - elif args.stations: - print("Station Signal RX: bytes packets TX: bytes packets") - for client in show_clients(args.stations): - print("{:<17} {:>3} {:>15} {:>9} {:>15} {:>10} ".format(client['mac'], - client['signal'], client['rx_bytes'], client['rx_packets'], client['tx_bytes'], client['tx_packets'])) - - exit(0) - - else: - parser.print_help() - exit(1) diff --git a/src/op_mode/ssh.py b/src/op_mode/ssh.py new file mode 100755 index 000000000..102becc55 --- /dev/null +++ b/src/op_mode/ssh.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +# +# Copyright 2017-2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library 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 +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import json +import sys +import glob +import vyos.opmode +from vyos.utils.process import cmd +from vyos.configquery import ConfigTreeQuery +from tabulate import tabulate + +def show_fingerprints(raw: bool, ascii: bool): + config = ConfigTreeQuery() + if not config.exists("service ssh"): + raise vyos.opmode.UnconfiguredSubsystem("SSH server is not enabled.") + + publickeys = glob.glob("/etc/ssh/*.pub") + + if publickeys: + keys = [] + for keyfile in publickeys: + try: + if ascii: + keydata = cmd("ssh-keygen -l -v -E sha256 -f " + keyfile).splitlines() + else: + keydata = cmd("ssh-keygen -l -E sha256 -f " + keyfile).splitlines() + type = keydata[0].split(None)[-1].strip("()") + key_size = keydata[0].split(None)[0] + fingerprint = keydata[0].split(None)[1] + comment = keydata[0].split(None)[2:-1][0] + if ascii: + ascii_art = "\n".join(keydata[1:]) + keys.append({"type": type, "key_size": key_size, "fingerprint": fingerprint, "comment": comment, "ascii_art": ascii_art}) + else: + keys.append({"type": type, "key_size": key_size, "fingerprint": fingerprint, "comment": comment}) + except: + # Ignore invalid public keys + pass + if raw: + return keys + else: + headers = {"type": "Type", "key_size": "Key Size", "fingerprint": "Fingerprint", "comment": "Comment", "ascii_art": "ASCII Art"} + output = "SSH server public key fingerprints:\n\n" + tabulate(keys, headers=headers, tablefmt="simple") + return output + else: + if raw: + return [] + else: + return "No SSH server public keys are found." + +def show_dynamic_protection(raw: bool): + config = ConfigTreeQuery() + if not config.exists(['service', 'ssh', 'dynamic-protection']): + raise vyos.opmode.UnconfiguredSubsystem("SSH server dynamic-protection is not enabled.") + + attackers = [] + try: + # IPv4 + attackers = attackers + json.loads(cmd("nft -j list set ip sshguard attackers"))["nftables"][1]["set"]["elem"] + except: + pass + try: + # IPv6 + attackers = attackers + json.loads(cmd("nft -j list set ip6 sshguard attackers"))["nftables"][1]["set"]["elem"] + except: + pass + if attackers: + if raw: + return attackers + else: + output = "Blocked attackers:\n" + "\n".join(attackers) + return output + else: + if raw: + return [] + else: + return "No blocked attackers." + +if __name__ == '__main__': + try: + res = vyos.opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, vyos.opmode.Error) as e: + print(e) + sys.exit(1) diff --git a/src/op_mode/traceroute.py b/src/op_mode/traceroute.py index 2f0edf53a..d2bac3f7c 100755 --- a/src/op_mode/traceroute.py +++ b/src/op_mode/traceroute.py @@ -14,29 +14,13 @@ # 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 os import sys import socket import ipaddress -from vyos.utils.network import get_all_vrfs -from vyos.ifconfig import Section - - -def interface_list() -> list: - """ - Get list of interfaces in system - :rtype: list - """ - return Section.interfaces() - - -def vrf_list() -> list: - """ - Get list of VRFs in system - :rtype: list - """ - return list(get_all_vrfs().keys()) +from vyos.utils.network import interface_list +from vyos.utils.network import vrf_list +from vyos.utils.process import call options = { 'backward-hops': { @@ -251,6 +235,4 @@ if __name__ == '__main__': sys.exit(f'traceroute: Unknown host: {host}') command = convert(traceroute[version], args) - - # print(f'{command} {host}') - os.system(f'{command} {host}') + call(f'{command} {host}') diff --git a/src/op_mode/zone.py b/src/op_mode/zone.py new file mode 100644 index 000000000..d24b1065b --- /dev/null +++ b/src/op_mode/zone.py @@ -0,0 +1,215 @@ +#!/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 typing +import sys +import vyos.opmode + +import tabulate +from vyos.configquery import ConfigTreeQuery +from vyos.utils.dict import dict_search_args +from vyos.utils.dict import dict_search + + +def get_config_zone(conf, name=None): + config_path = ['firewall', 'zone'] + if name: + config_path += [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 _convert_one_zone_data(zone: str, zone_config: dict) -> dict: + """ + Convert config dictionary of one zone to API dictionary + :param zone: Zone name + :type zone: str + :param zone_config: config dictionary + :type zone_config: dict + :return: AP dictionary + :rtype: dict + """ + list_of_rules = [] + intrazone_dict = {} + if dict_search('from', zone_config): + for from_zone, from_zone_config in zone_config['from'].items(): + from_zone_dict = {'name': from_zone} + if dict_search('firewall.name', from_zone_config): + from_zone_dict['firewall'] = dict_search('firewall.name', + from_zone_config) + if dict_search('firewall.ipv6_name', from_zone_config): + from_zone_dict['firewall_v6'] = dict_search( + 'firewall.ipv6_name', from_zone_config) + list_of_rules.append(from_zone_dict) + + zone_dict = { + 'name': zone, + 'interface': dict_search('interface', zone_config), + 'type': 'LOCAL' if dict_search('local_zone', + zone_config) is not None else None, + } + if list_of_rules: + zone_dict['from'] = list_of_rules + if dict_search('intra_zone_filtering.firewall.name', zone_config): + intrazone_dict['firewall'] = dict_search( + 'intra_zone_filtering.firewall.name', zone_config) + if dict_search('intra_zone_filtering.firewall.ipv6_name', zone_config): + intrazone_dict['firewall_v6'] = dict_search( + 'intra_zone_filtering.firewall.ipv6_name', zone_config) + if intrazone_dict: + zone_dict['intrazone'] = intrazone_dict + return zone_dict + + +def _convert_zones_data(zone_policies: dict) -> list: + """ + Convert all config dictionary to API list of zone dictionaries + :param zone_policies: config dictionary + :type zone_policies: dict + :return: API list + :rtype: list + """ + zone_list = [] + for zone, zone_config in zone_policies.items(): + zone_list.append(_convert_one_zone_data(zone, zone_config)) + return zone_list + + +def _convert_config(zones_config: dict, zone: str = None) -> list: + """ + convert config to API list + :param zones_config: zones config + :type zones_config: + :param zone: zone name + :type zone: str + :return: API list + :rtype: list + """ + if zone: + if zones_config: + output = [_convert_one_zone_data(zone, zones_config)] + else: + raise vyos.opmode.DataUnavailable(f'Zone {zone} not found') + else: + if zones_config: + output = _convert_zones_data(zones_config) + else: + raise vyos.opmode.UnconfiguredSubsystem( + 'Zone entries are not configured') + return output + + +def output_zone_list(zone_conf: dict) -> list: + """ + Format one zone row + :param zone_conf: zone config + :type zone_conf: dict + :return: formatted list of zones + :rtype: list + """ + zone_info = [zone_conf['name']] + if zone_conf['type'] == 'LOCAL': + zone_info.append('LOCAL') + else: + zone_info.append("\n".join(zone_conf['interface'])) + + from_zone = [] + firewall = [] + firewall_v6 = [] + if 'intrazone' in zone_conf: + from_zone.append(zone_conf['name']) + + v4_name = dict_search_args(zone_conf['intrazone'], 'firewall') + v6_name = dict_search_args(zone_conf['intrazone'], 'firewall_v6') + if v4_name: + firewall.append(v4_name) + else: + firewall.append('') + if v6_name: + firewall_v6.append(v6_name) + else: + firewall_v6.append('') + + if 'from' in zone_conf: + for from_conf in zone_conf['from']: + from_zone.append(from_conf['name']) + + v4_name = dict_search_args(from_conf, 'firewall') + v6_name = dict_search_args(from_conf, 'firewall_v6') + if v4_name: + firewall.append(v4_name) + else: + firewall.append('') + if v6_name: + firewall_v6.append(v6_name) + else: + firewall_v6.append('') + + zone_info.append("\n".join(from_zone)) + zone_info.append("\n".join(firewall)) + zone_info.append("\n".join(firewall_v6)) + return zone_info + + +def get_formatted_output(zone_policy: list) -> str: + """ + Formatted output of all zones + :param zone_policy: list of zones + :type zone_policy: list + :return: formatted table with zones + :rtype: str + """ + headers = ["Zone", + "Interfaces", + "From Zone", + "Firewall IPv4", + "Firewall IPv6" + ] + formatted_list = [] + for zone_conf in zone_policy: + formatted_list.append(output_zone_list(zone_conf)) + tabulate.PRESERVE_WHITESPACE = True + output = tabulate.tabulate(formatted_list, headers, numalign="left") + return output + + +def show(raw: bool, zone: typing.Optional[str]): + """ + Show zone-policy command + :param raw: if API + :type raw: bool + :param zone: zone name + :type zone: str + """ + conf: ConfigTreeQuery = ConfigTreeQuery() + zones_config: dict = get_config_zone(conf, zone) + zone_policy_api: list = _convert_config(zones_config, zone) + if raw: + return zone_policy_api + else: + return get_formatted_output(zone_policy_api) + + +if __name__ == '__main__': + try: + res = vyos.opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, vyos.opmode.Error) as e: + print(e) + sys.exit(1)
\ No newline at end of file |