summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rwxr-xr-xsrc/conf_mode/container.py7
-rwxr-xr-xsrc/conf_mode/interfaces_wireguard.py2
-rw-r--r--src/conf_mode/load-balancing_haproxy.py37
-rwxr-xr-xsrc/conf_mode/nat66.py4
-rwxr-xr-xsrc/conf_mode/pki.py94
-rwxr-xr-xsrc/conf_mode/protocols_bgp.py14
-rwxr-xr-xsrc/init/vyos-router2
-rw-r--r--src/migration-scripts/vrf/1-to-25
-rw-r--r--src/migration-scripts/vrf/2-to-33
-rwxr-xr-xsrc/op_mode/image_installer.py26
-rwxr-xr-xsrc/op_mode/interfaces.py138
-rw-r--r--src/tests/test_template.py9
-rw-r--r--src/tests/test_utils_network.py11
13 files changed, 287 insertions, 65 deletions
diff --git a/src/conf_mode/container.py b/src/conf_mode/container.py
index 18d660a4e..94882fc14 100755
--- a/src/conf_mode/container.py
+++ b/src/conf_mode/container.py
@@ -324,6 +324,11 @@ def generate_run_arguments(name, container_config):
cap = cap.upper().replace('-', '_')
capabilities += f' --cap-add={cap}'
+ # Grant root capabilities to the container
+ privileged = ''
+ if 'privileged' in container_config:
+ privileged = '--privileged'
+
# Add a host device to the container /dev/x:/dev/x
device = ''
if 'device' in container_config:
@@ -402,7 +407,7 @@ def generate_run_arguments(name, container_config):
for ns in container_config['name_server']:
name_server += f'--dns {ns}'
- container_base_cmd = f'--detach --interactive --tty --replace {capabilities} --cpus {cpu_quota} {sysctl_opt} ' \
+ container_base_cmd = f'--detach --interactive --tty --replace {capabilities} {privileged} --cpus {cpu_quota} {sysctl_opt} ' \
f'--memory {memory}m --shm-size {shared_memory}m --memory-swap 0 --restart {restart} ' \
f'--name {name} {hostname} {device} {port} {name_server} {volume} {tmpfs} {env_opt} {label} {uid} {host_pid}'
diff --git a/src/conf_mode/interfaces_wireguard.py b/src/conf_mode/interfaces_wireguard.py
index 192937dba..3ca6ecdca 100755
--- a/src/conf_mode/interfaces_wireguard.py
+++ b/src/conf_mode/interfaces_wireguard.py
@@ -97,7 +97,7 @@ def verify(wireguard):
if 'port' in wireguard and 'port_changed' in wireguard:
listen_port = int(wireguard['port'])
- if check_port_availability('0.0.0.0', listen_port, 'udp') is not True:
+ if check_port_availability(None, listen_port, protocol='udp') is not True:
raise ConfigError(f'UDP port {listen_port} is busy or unavailable and '
'cannot be used for the interface!')
diff --git a/src/conf_mode/load-balancing_haproxy.py b/src/conf_mode/load-balancing_haproxy.py
index 5fd1beec9..504a90596 100644
--- a/src/conf_mode/load-balancing_haproxy.py
+++ b/src/conf_mode/load-balancing_haproxy.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2023-2024 VyOS maintainers and contributors
+# Copyright (C) 2023-2025 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
@@ -19,6 +19,7 @@ import os
from sys import exit
from shutil import rmtree
+from vyos.defaults import systemd_services
from vyos.config import Config
from vyos.configverify import verify_pki_certificate
from vyos.configverify import verify_pki_ca_certificate
@@ -39,7 +40,6 @@ airbag.enable()
load_balancing_dir = '/run/haproxy'
load_balancing_conf_file = f'{load_balancing_dir}/haproxy.cfg'
-systemd_service = 'haproxy.service'
systemd_override = '/run/systemd/system/haproxy.service.d/10-override.conf'
def get_config(config=None):
@@ -65,18 +65,18 @@ def verify(lb):
return None
if 'backend' not in lb or 'service' not in lb:
- raise ConfigError(f'"service" and "backend" must be configured!')
+ raise ConfigError('Both "service" and "backend" must be configured!')
for front, front_config in lb['service'].items():
if 'port' not in front_config:
raise ConfigError(f'"{front} service port" must be configured!')
# Check if bind address:port are used by another service
- tmp_address = front_config.get('address', '0.0.0.0')
+ tmp_address = front_config.get('address', None)
tmp_port = front_config['port']
if check_port_availability(tmp_address, int(tmp_port), 'tcp') is not True and \
not is_listen_port_bind_service(int(tmp_port), 'haproxy'):
- raise ConfigError(f'"TCP" port "{tmp_port}" is used by another service')
+ raise ConfigError(f'TCP port "{tmp_port}" is used by another service')
if 'http_compression' in front_config:
if front_config['mode'] != 'http':
@@ -85,16 +85,19 @@ def verify(lb):
raise ConfigError(f'service {front} must have at least one mime-type configured to use'
f'http_compression!')
+ for cert in dict_search('ssl.certificate', front_config) or []:
+ verify_pki_certificate(lb, cert)
+
for back, back_config in lb['backend'].items():
if 'http_check' in back_config:
http_check = back_config['http_check']
if 'expect' in http_check and 'status' in http_check['expect'] and 'string' in http_check['expect']:
- raise ConfigError(f'"expect status" and "expect string" can not be configured together!')
+ raise ConfigError('"expect status" and "expect string" can not be configured together!')
if 'health_check' in back_config:
if back_config['mode'] != 'tcp':
raise ConfigError(f'backend "{back}" can only be configured with {back_config["health_check"]} ' +
- f'health-check whilst in TCP mode!')
+ 'health-check whilst in TCP mode!')
if 'http_check' in back_config:
raise ConfigError(f'backend "{back}" cannot be configured with both http-check and health-check!')
@@ -112,20 +115,15 @@ def verify(lb):
if {'no_verify', 'ca_certificate'} <= set(back_config['ssl']):
raise ConfigError(f'backend {back} cannot have both ssl options no-verify and ca-certificate set!')
+ tmp = dict_search('ssl.ca_certificate', back_config)
+ if tmp: verify_pki_ca_certificate(lb, tmp)
+
# Check if http-response-headers are configured in any frontend/backend where mode != http
for group in ['service', 'backend']:
for config_name, config in lb[group].items():
if 'http_response_headers' in config and config['mode'] != 'http':
raise ConfigError(f'{group} {config_name} must be set to http mode to use http_response_headers!')
- for front, front_config in lb['service'].items():
- for cert in dict_search('ssl.certificate', front_config) or []:
- verify_pki_certificate(lb, cert)
-
- for back, back_config in lb['backend'].items():
- tmp = dict_search('ssl.ca_certificate', back_config)
- if tmp: verify_pki_ca_certificate(lb, tmp)
-
def generate(lb):
if not lb:
@@ -193,12 +191,11 @@ def generate(lb):
return None
def apply(lb):
+ action = 'stop'
+ if lb:
+ action = 'reload-or-restart'
call('systemctl daemon-reload')
- if not lb:
- call(f'systemctl stop {systemd_service}')
- else:
- call(f'systemctl reload-or-restart {systemd_service}')
-
+ call(f'systemctl {action} {systemd_services["haproxy"]}')
return None
diff --git a/src/conf_mode/nat66.py b/src/conf_mode/nat66.py
index 95dfae3a5..c65950c9e 100755
--- a/src/conf_mode/nat66.py
+++ b/src/conf_mode/nat66.py
@@ -92,6 +92,10 @@ def verify(nat):
if prefix != None:
if not is_ipv6(prefix):
raise ConfigError(f'{err_msg} source-prefix not specified')
+
+ if 'destination' in config and 'group' in config['destination']:
+ if len({'address_group', 'network_group', 'domain_group'} & set(config['destination']['group'])) > 1:
+ raise ConfigError('Only one address-group, network-group or domain-group can be specified')
if dict_search('destination.rule', nat):
for rule, config in dict_search('destination.rule', nat).items():
diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index 724f97555..869518dd9 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2021-2024 VyOS maintainers and contributors
+# Copyright (C) 2021-2025 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
@@ -19,6 +19,7 @@ import os
from sys import argv
from sys import exit
+from vyos.base import Message
from vyos.config import Config
from vyos.config import config_dict_merge
from vyos.configdep import set_dependents
@@ -27,6 +28,8 @@ from vyos.configdict import node_changed
from vyos.configdiff import Diff
from vyos.configdiff import get_config_diff
from vyos.defaults import directories
+from vyos.defaults import internal_ports
+from vyos.defaults import systemd_services
from vyos.pki import encode_certificate
from vyos.pki import is_ca_certificate
from vyos.pki import load_certificate
@@ -42,9 +45,11 @@ from vyos.utils.dict import dict_search
from vyos.utils.dict import dict_search_args
from vyos.utils.dict import dict_search_recursive
from vyos.utils.file import read_file
+from vyos.utils.network import check_port_availability
from vyos.utils.process import call
from vyos.utils.process import cmd
from vyos.utils.process import is_systemd_service_active
+from vyos.utils.process import is_systemd_service_running
from vyos import ConfigError
from vyos import airbag
airbag.enable()
@@ -128,8 +133,20 @@ def certbot_request(name: str, config: dict, dry_run: bool=True):
f'--standalone --agree-tos --no-eff-email --expand --server {config["url"]} '\
f'--email {config["email"]} --key-type rsa --rsa-key-size {config["rsa_key_size"]} '\
f'{domains}'
+
+ listen_address = None
if 'listen_address' in config:
- tmp += f' --http-01-address {config["listen_address"]}'
+ listen_address = config['listen_address']
+
+ # When ACME is used behind a reverse proxy, we always bind to localhost
+ # whatever the CLI listen-address is configured for.
+ if ('haproxy' in dict_search('used_by', config) and
+ is_systemd_service_running(systemd_services['haproxy']) and
+ not check_port_availability(listen_address, 80)):
+ tmp += f' --http-01-address 127.0.0.1 --http-01-port {internal_ports["certbot_haproxy"]}'
+ elif listen_address:
+ tmp += f' --http-01-address {listen_address}'
+
# verify() does not need to actually request a cert but only test for plausability
if dry_run:
tmp += ' --dry-run'
@@ -150,14 +167,18 @@ def get_config(config=None):
if len(argv) > 1 and argv[1] == 'certbot_renew':
pki['certbot_renew'] = {}
- changed_keys = ['ca', 'certificate', 'dh', 'key-pair', 'openssh', 'openvpn']
+ # Walk through the list of sync_translate mapping and build a list
+ # which is later used to check if the node was changed in the CLI config
+ changed_keys = []
+ for value in sync_translate.values():
+ if value not in changed_keys:
+ changed_keys.append(value)
+ # Check for changes to said given keys in the CLI config
for key in changed_keys:
tmp = node_changed(conf, base + [key], recursive=True, expand_nodes=Diff.DELETE | Diff.ADD)
-
if 'changed' not in pki:
pki.update({'changed':{}})
-
pki['changed'].update({key.replace('-', '_') : tmp})
# We only merge on the defaults of there is a configuration at all
@@ -219,8 +240,8 @@ def get_config(config=None):
continue
path = search['path']
- path_str = ' '.join(path + found_path)
- #print(f'PKI: Updating config: {path_str} {item_name}')
+ path_str = ' '.join(path + found_path).replace('_','-')
+ Message(f'Updating configuration: "{path_str} {item_name}"')
if path[0] == 'interfaces':
ifname = found_path[0]
@@ -230,6 +251,24 @@ def get_config(config=None):
if not D.node_changed_presence(path):
set_dependents(path[1], conf)
+ # Check PKI certificates if they are auto-generated by ACME. If they are,
+ # traverse the current configuration and determine the service where the
+ # certificate is used by.
+ # Required to check if we might need to run certbot behing a reverse proxy.
+ if 'certificate' in pki:
+ for name, cert_config in pki['certificate'].items():
+ if 'acme' not in cert_config:
+ continue
+ if not dict_search('system.load_balancing.haproxy', pki):
+ continue
+ used_by = []
+ for cert_list, _ in dict_search_recursive(
+ pki['system']['load_balancing']['haproxy'], 'certificate'):
+ if name in cert_list:
+ used_by.append('haproxy')
+ if used_by:
+ pki['certificate'][name]['acme'].update({'used_by': used_by})
+
return pki
def is_valid_certificate(raw_data):
@@ -321,6 +360,15 @@ def verify(pki):
raise ConfigError(f'An email address is required to request '\
f'certificate for "{name}" via ACME!')
+ listen_address = None
+ if 'listen_address' in cert_conf['acme']:
+ listen_address = cert_conf['acme']['listen_address']
+
+ if 'used_by' not in cert_conf['acme']:
+ if not check_port_availability(listen_address, 80):
+ raise ConfigError('Port 80 is already in use and not available '\
+ f'to provide ACME challenge for "{name}"!')
+
if 'certbot_renew' not in pki:
# Only run the ACME command if something on this entity changed,
# as this is time intensive
@@ -374,27 +422,35 @@ def verify(pki):
for search in sync_search:
for key in search['keys']:
changed_key = sync_translate[key]
-
if changed_key not in pki['changed']:
continue
-
for item_name in pki['changed'][changed_key]:
node_present = False
if changed_key == 'openvpn':
node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
else:
node_present = dict_search_args(pki, changed_key, item_name)
+ # If the node is still present, we can skip the check
+ # as we are not deleting it
+ if node_present:
+ continue
- if not node_present:
- search_dict = dict_search_args(pki['system'], *search['path'])
-
- if not search_dict:
- continue
+ search_dict = dict_search_args(pki['system'], *search['path'])
+ if not search_dict:
+ continue
- for found_name, found_path in dict_search_recursive(search_dict, key):
- if found_name == item_name:
- path_str = " ".join(search['path'] + found_path)
- raise ConfigError(f'PKI object "{item_name}" still in use by "{path_str}"')
+ for found_name, found_path in dict_search_recursive(search_dict, key):
+ # Check if the name matches either by string compare, or beeing
+ # part of a list
+ if ((isinstance(found_name, str) and found_name == item_name) or
+ (isinstance(found_name, list) and item_name in found_name)):
+ # We do not support _ in CLI paths - this is only a convenience
+ # as we mangle all - to _, now it's time to reverse this!
+ path_str = ' '.join(search['path'] + found_path).replace('_','-')
+ object = changed_key.replace('_','-')
+ tmp = f'Embedded PKI {object} with name "{item_name}" is still '\
+ f'in use by CLI path "{path_str}"'
+ raise ConfigError(tmp)
return None
@@ -490,7 +546,7 @@ def generate(pki):
if not ca_cert_present:
tmp = dict_search_args(pki, 'ca', f'{autochain_prefix}{cert}', 'certificate')
if not bool(tmp) or tmp != cert_chain_base64:
- print(f'Adding/replacing automatically imported CA certificate for "{cert}" ...')
+ Message(f'Add/replace automatically imported CA certificate for "{cert}"...')
add_cli_node(['pki', 'ca', f'{autochain_prefix}{cert}', 'certificate'], value=cert_chain_base64)
return None
diff --git a/src/conf_mode/protocols_bgp.py b/src/conf_mode/protocols_bgp.py
index 53e83c3b4..99d8eb9d1 100755
--- a/src/conf_mode/protocols_bgp.py
+++ b/src/conf_mode/protocols_bgp.py
@@ -413,15 +413,19 @@ def verify(config_dict):
verify_route_map(afi_config['route_map'][tmp], bgp)
if 'route_reflector_client' in afi_config:
- peer_group_as = peer_config.get('remote_as')
+ peer_as = peer_config.get('remote_as')
- if peer_group_as is None or (peer_group_as != 'internal' and peer_group_as != bgp['system_as']):
+ if peer_as is not None and (peer_as != 'internal' and peer_as != bgp['system_as']):
raise ConfigError('route-reflector-client only supported for iBGP peers')
else:
+ # Check into the peer group for the remote as, if we are in a peer group, check in peer itself
if 'peer_group' in peer_config:
peer_group_as = dict_search(f'peer_group.{peer_group}.remote_as', bgp)
- if peer_group_as is None or (peer_group_as != 'internal' and peer_group_as != bgp['system_as']):
- raise ConfigError('route-reflector-client only supported for iBGP peers')
+ elif neighbor == 'peer_group':
+ peer_group_as = peer_config.get('remote_as')
+
+ if peer_group_as is None or (peer_group_as != 'internal' and peer_group_as != bgp['system_as']):
+ raise ConfigError('route-reflector-client only supported for iBGP peers')
# T5833 not all AFIs are supported for VRF
if 'vrf' in bgp and 'address_family' in peer_config:
@@ -527,7 +531,7 @@ def verify(config_dict):
or dict_search('import.vrf', afi_config) is not None):
# FRR error: please unconfigure vpn to vrf commands before
# using import vrf commands
- if ('vpn' in afi_config['import']
+ if (dict_search('import.vpn', afi_config) is not None
or dict_search('export.vpn', afi_config) is not None):
raise ConfigError('Please unconfigure VPN to VRF commands before '\
'using "import vrf" commands!')
diff --git a/src/init/vyos-router b/src/init/vyos-router
index 8584234b3..6f1d386d6 100755
--- a/src/init/vyos-router
+++ b/src/init/vyos-router
@@ -417,7 +417,6 @@ gen_duid ()
start ()
{
- echo -e "Initializing VyOS router\033[0m"
# reset and clean config files
security_reset || log_failure_msg "security reset failed"
@@ -526,6 +525,7 @@ start ()
cleanup_post_commit_hooks
+ log_daemon_msg "Starting VyOS router"
disabled migrate || migrate_bootfile
restore_if_missing_preconfig_script
diff --git a/src/migration-scripts/vrf/1-to-2 b/src/migration-scripts/vrf/1-to-2
index 557a9ec58..89b0f708a 100644
--- a/src/migration-scripts/vrf/1-to-2
+++ b/src/migration-scripts/vrf/1-to-2
@@ -37,7 +37,10 @@ def migrate(config: ConfigTree) -> None:
new_static_base = vrf_base + [vrf, 'protocols']
config.set(new_static_base)
config.copy(static_base, new_static_base + ['static'])
- config.set_tag(new_static_base + ['static', 'route'])
+ if config.exists(new_static_base + ['static', 'route']):
+ config.set_tag(new_static_base + ['static', 'route'])
+ if config.exists(new_static_base + ['static', 'route6']):
+ config.set_tag(new_static_base + ['static', 'route6'])
# Now delete the old configuration
config.delete(base)
diff --git a/src/migration-scripts/vrf/2-to-3 b/src/migration-scripts/vrf/2-to-3
index acacffb41..5f396e7ed 100644
--- a/src/migration-scripts/vrf/2-to-3
+++ b/src/migration-scripts/vrf/2-to-3
@@ -76,7 +76,8 @@ def migrate(config: ConfigTree) -> None:
# Get a list of all currently used VRFs and tables
vrfs_current = {}
for vrf in config.list_nodes(base):
- vrfs_current[vrf] = int(config.return_value(base + [vrf, 'table']))
+ if config.exists(base + [vrf, 'table']):
+ vrfs_current[vrf] = int(config.return_value(base + [vrf, 'table']))
# Check VRF names and table numbers
name_regex = re.compile(r'^\d.*$')
diff --git a/src/op_mode/image_installer.py b/src/op_mode/image_installer.py
index 3af2232bb..ac5a84419 100755
--- a/src/op_mode/image_installer.py
+++ b/src/op_mode/image_installer.py
@@ -565,21 +565,18 @@ def validate_signature(file_path: str, sign_type: str) -> None:
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
+ # Server credentials are implicitly passed in environment variables
+ # that are set by add_image
if vrf is None:
download(local_file, remote_path, progressbar=progressbar,
check_space=check_space, raise_error=True)
else:
- remote_auth = f'REMOTE_USERNAME={username} REMOTE_PASSWORD={password}'
vrf_cmd = f'ip vrf exec {vrf} {external_download_script} \
--local-file {local_file} --remote-path {remote_path}'
- cmd(vrf_cmd, auth=remote_auth)
+ cmd(vrf_cmd, env=environ)
def image_fetch(image_path: str, vrf: str = None,
- username: str = '', password: str = '',
no_prompt: bool = False) -> Path:
"""Fetch an ISO image
@@ -598,9 +595,8 @@ def image_fetch(image_path: str, vrf: str = None,
if image_path == 'latest':
command = external_latest_image_url_script
if vrf:
- command = f'REMOTE_USERNAME={username} REMOTE_PASSWORD={password} \
- ip vrf exec {vrf} ' + command
- code, output = rc_cmd(command)
+ command = f'ip vrf exec {vrf} {command}'
+ code, output = rc_cmd(command, env=environ)
if code:
print(output)
exit(MSG_INFO_INSTALL_EXIT)
@@ -612,7 +608,6 @@ def image_fetch(image_path: str, vrf: str = None,
# Download the image file
ISO_DOWNLOAD_PATH = os.path.join(os.path.expanduser("~"), '{0}.iso'.format(uuid4()))
download_file(ISO_DOWNLOAD_PATH, image_path, vrf,
- username, password,
progressbar=True, check_space=True)
# Download the image signature
@@ -623,8 +618,7 @@ def image_fetch(image_path: str, vrf: str = None,
for sign_type in ['minisig']:
try:
download_file(f'{ISO_DOWNLOAD_PATH}.{sign_type}',
- f'{image_path}.{sign_type}', vrf,
- username, password)
+ f'{image_path}.{sign_type}', vrf)
sign_file = (True, sign_type)
break
except Exception:
@@ -928,8 +922,7 @@ def install_image() -> None:
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})')
+ f'{DIR_DST_ROOT}/boot/efi')
disk.partition_umount(disk_target.partition['efi'])
else:
print('Installing GRUB to the drive')
@@ -981,8 +974,11 @@ def add_image(image_path: str, vrf: str = None, username: str = '',
if image.is_live_boot():
exit(MSG_ERR_LIVE)
+ environ['REMOTE_USERNAME'] = username
+ environ['REMOTE_PASSWORD'] = password
+
# fetch an image
- iso_path: Path = image_fetch(image_path, vrf, username, password, no_prompt)
+ iso_path: Path = image_fetch(image_path, vrf, no_prompt)
try:
# mount an ISO
Path(DIR_ISO_MOUNT).mkdir(mode=0o755, parents=True)
diff --git a/src/op_mode/interfaces.py b/src/op_mode/interfaces.py
index e7afc4caa..c97f3b129 100755
--- a/src/op_mode/interfaces.py
+++ b/src/op_mode/interfaces.py
@@ -29,6 +29,7 @@ from vyos.ifconfig import Section
from vyos.ifconfig import Interface
from vyos.ifconfig import VRRP
from vyos.utils.process import cmd
+from vyos.utils.network import interface_exists
from vyos.utils.process import rc_cmd
from vyos.utils.process import call
@@ -84,6 +85,14 @@ def filtered_interfaces(ifnames: typing.Union[str, list],
yield interface
+def detailed_output(dataset, headers):
+ for data in dataset:
+ adjusted_rule = data + [""] * (len(headers) - len(data)) # account for different header length, like default-action
+ transformed_rule = [[header, adjusted_rule[i]] for i, header in enumerate(headers) if i < len(adjusted_rule)] # create key-pair list from headers and rules lists; wrap at 100 char
+
+ print(tabulate(transformed_rule, tablefmt="presto"))
+ print()
+
def _split_text(text, used=0):
"""
take a string and attempt to split it to fit with the width of the screen
@@ -296,6 +305,114 @@ def _get_counter_data(ifname: typing.Optional[str],
return ret
+def _get_kernel_data(raw, ifname = None, detail = False):
+ if ifname:
+ # Check if the interface exists
+ if not interface_exists(ifname):
+ raise vyos.opmode.IncorrectValue(f"{ifname} does not exist!")
+ int_name = f'dev {ifname}'
+ else:
+ int_name = ''
+
+ kernel_interface = json.loads(cmd(f'ip -j -d -s address show {int_name}'))
+
+ # Return early if raw
+ if raw:
+ return kernel_interface, None
+
+ # Format the kernel data
+ kernel_interface_out = _format_kernel_data(kernel_interface, detail)
+
+ return kernel_interface, kernel_interface_out
+
+def _format_kernel_data(data, detail):
+ output_list = []
+ tmpInfo = {}
+
+ # Sort interfaces by name
+ for interface in sorted(data, key=lambda x: x.get('ifname', '')):
+ if interface.get('linkinfo', {}).get('info_kind') == 'vrf':
+ continue
+
+ # Get the device model; ex. Intel Corporation Ethernet Controller I225-V
+ dev_model = interface.get('parentdev', '')
+ if 'parentdev' in interface:
+ parentdev = interface['parentdev']
+ if re.match(r'^[0-9a-fA-F]{4}:', parentdev):
+ dev_model = cmd(f'lspci -nn -s {parentdev}').split(']:')[1].strip()
+
+ # Get the IP addresses on interface
+ ip_list = []
+ has_global = False
+
+ for ip in interface['addr_info']:
+ if ip.get('scope') in ('global', 'host'):
+ has_global = True
+ local = ip.get('local', '-')
+ prefixlen = ip.get('prefixlen', '')
+ ip_list.append(f"{local}/{prefixlen}")
+
+
+ # If no global IP address, add '-'; indicates no IP address on interface
+ if not has_global:
+ ip_list.append('-')
+
+ sl_status = ('A' if not 'UP' in interface['flags'] else 'u') + '/' + ('D' if interface['operstate'] == 'DOWN' else 'u')
+
+ # Generate temporary dict to hold data
+ tmpInfo['ifname'] = interface.get('ifname', '')
+ tmpInfo['ip'] = ip_list
+ tmpInfo['mac'] = interface.get('address', '')
+ tmpInfo['mtu'] = interface.get('mtu', '')
+ tmpInfo['vrf'] = interface.get('master', 'default')
+ tmpInfo['status'] = sl_status
+ tmpInfo['description'] = interface.get('ifalias', '')
+ tmpInfo['device'] = dev_model
+ tmpInfo['alternate_names'] = interface.get('altnames', '')
+ tmpInfo['minimum_mtu'] = interface.get('min_mtu', '')
+ tmpInfo['maximum_mtu'] = interface.get('max_mtu', '')
+ rx_stats = interface.get('stats64', {}).get('rx')
+ tx_stats = interface.get('stats64', {}).get('tx')
+ tmpInfo['rx_packets'] = rx_stats.get('packets', "")
+ tmpInfo['rx_bytes'] = rx_stats.get('bytes', "")
+ tmpInfo['rx_errors'] = rx_stats.get('errors', "")
+ tmpInfo['rx_dropped'] = rx_stats.get('dropped', "")
+ tmpInfo['rx_over_errors'] = rx_stats.get('over_errors', '')
+ tmpInfo['multicast'] = rx_stats.get('multicast', "")
+ tmpInfo['tx_packets'] = tx_stats.get('packets', "")
+ tmpInfo['tx_bytes'] = tx_stats.get('bytes', "")
+ tmpInfo['tx_errors'] = tx_stats.get('errors', "")
+ tmpInfo['tx_dropped'] = tx_stats.get('dropped', "")
+ tmpInfo['tx_carrier_errors'] = tx_stats.get('carrier_errors', "")
+ tmpInfo['tx_collisions'] = tx_stats.get('collisions', "")
+
+ # Generate output list; detail adds more fields
+ output_list.append([tmpInfo['ifname'],
+ '\n'.join(tmpInfo['ip']),
+ tmpInfo['mac'],
+ tmpInfo['vrf'],
+ tmpInfo['mtu'],
+ tmpInfo['status'],
+ tmpInfo['description'],
+ *([tmpInfo['device']] if detail else []),
+ *(['\n'.join(tmpInfo['alternate_names'])] if detail else []),
+ *([tmpInfo['minimum_mtu']] if detail else []),
+ *([tmpInfo['maximum_mtu']] if detail else []),
+ *([tmpInfo['rx_packets']] if detail else []),
+ *([tmpInfo['rx_bytes']] if detail else []),
+ *([tmpInfo['rx_errors']] if detail else []),
+ *([tmpInfo['rx_dropped']] if detail else []),
+ *([tmpInfo['rx_over_errors']] if detail else []),
+ *([tmpInfo['multicast']] if detail else []),
+ *([tmpInfo['tx_packets']] if detail else []),
+ *([tmpInfo['tx_bytes']] if detail else []),
+ *([tmpInfo['tx_errors']] if detail else []),
+ *([tmpInfo['tx_dropped']] if detail else []),
+ *([tmpInfo['tx_carrier_errors']] if detail else []),
+ *([tmpInfo['tx_collisions']] if detail else [])])
+
+ return output_list
+
@catch_broken_pipe
def _format_show_data(data: list):
unhandled = []
@@ -445,6 +562,27 @@ def _format_show_counters(data: list):
print (output)
return output
+def show_kernel(raw: bool, intf_name: typing.Optional[str], detail: bool):
+ raw_data, data = _get_kernel_data(raw, intf_name, detail)
+
+ # Return early if raw
+ if raw:
+ return raw_data
+
+ # Normal headers; show interfaces kernel
+ headers = ['Interface', 'IP Address', 'MAC', 'VRF', 'MTU', 'S/L', 'Description']
+
+ # Detail headers; show interfaces kernel detail
+ detail_header = ['Interface', 'IP Address', 'MAC', 'VRF', 'MTU', 'S/L', 'Description',
+ 'Device', 'Alternate Names','Minimum MTU', 'Maximum MTU', 'RX_Packets',
+ 'RX_Bytes', 'RX_Errors', 'RX_Dropped', 'Receive Overrun Errors', 'Received Multicast',
+ 'TX_Packets', 'TX_Bytes', 'TX_Errors', 'TX_Dropped', 'Transmit Carrier Errors',
+ 'Transmit Collisions']
+
+ if detail:
+ detailed_output(data, detail_header)
+ else:
+ print(tabulate(data, headers))
def _show_raw(data: list, intf_name: str):
if intf_name is not None and len(data) <= 1:
diff --git a/src/tests/test_template.py b/src/tests/test_template.py
index 6377f6da5..7cae867a0 100644
--- a/src/tests/test_template.py
+++ b/src/tests/test_template.py
@@ -190,3 +190,12 @@ class TestVyOSTemplate(TestCase):
for group_name, group_config in data['ike_group'].items():
ciphers = vyos.template.get_esp_ike_cipher(group_config)
self.assertIn(IKEv2_DEFAULT, ','.join(ciphers))
+
+ def test_get_default_port(self):
+ from vyos.defaults import internal_ports
+
+ with self.assertRaises(RuntimeError):
+ vyos.template.get_default_port('UNKNOWN')
+
+ self.assertEqual(vyos.template.get_default_port('certbot_haproxy'),
+ internal_ports['certbot_haproxy'])
diff --git a/src/tests/test_utils_network.py b/src/tests/test_utils_network.py
index d68dec16f..92fde447d 100644
--- a/src/tests/test_utils_network.py
+++ b/src/tests/test_utils_network.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2020-2024 VyOS maintainers and contributors
+# Copyright (C) 2020-2025 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
@@ -43,3 +43,12 @@ class TestVyOSUtilsNetwork(TestCase):
self.assertFalse(vyos.utils.network.is_loopback_addr('::2'))
self.assertFalse(vyos.utils.network.is_loopback_addr('192.0.2.1'))
+
+ def test_check_port_availability(self):
+ self.assertTrue(vyos.utils.network.check_port_availability('::1', 8080))
+ self.assertTrue(vyos.utils.network.check_port_availability('127.0.0.1', 8080))
+ self.assertTrue(vyos.utils.network.check_port_availability(None, 8080, protocol='udp'))
+ # We do not have 192.0.2.1 configured on this system
+ self.assertFalse(vyos.utils.network.check_port_availability('192.0.2.1', 443))
+ # We do not have 2001:db8::1 configured on this system
+ self.assertFalse(vyos.utils.network.check_port_availability('2001:db8::1', 80, protocol='udp'))
generated by cgit v1.2.3 (git 2.39.1) at 2025-05-10 11:09:55 +0000