No OneTemporary
Actions

Size

58 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py
	index b58e02d91..63c9e263d 100644
	--- a/python/vyos/utils/network.py
	+++ b/python/vyos/utils/network.py
	@@ -1,545 +1,545 @@
	# 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/>.

	def _are_same_ip(one, two):
	from socket import AF_INET
	from socket import AF_INET6
	from socket import inet_pton
	from vyos.template import is_ipv4
	# compare the binary representation of the IP
	f_one = AF_INET if is_ipv4(one) else AF_INET6
	s_two = AF_INET if is_ipv4(two) else AF_INET6
	return inet_pton(f_one, one) == inet_pton(f_one, two)

	def get_protocol_by_name(protocol_name):
	"""Get protocol number by protocol name

	% get_protocol_by_name('tcp')
	% 6
	"""
	import socket
	try:
	protocol_number = socket.getprotobyname(protocol_name)
	return protocol_number
	except socket.error:
	return protocol_name

	def interface_exists(interface) -> bool:
	import os
	return os.path.exists(f'/sys/class/net/{interface}')

	def interface_exists_in_netns(interface_name, netns):
	from vyos.utils.process import rc_cmd
	rc, out = rc_cmd(f'ip netns exec {netns} ip link show dev {interface_name}')
	if rc == 0:
	return True
	return False

	def get_vrf_members(vrf: str) -> list:
	"""
	Get list of interface VRF members
	:param vrf: str
	:return: list
	"""
	import json
	from vyos.utils.process import cmd
	interfaces = []
	try:
	if not interface_exists(vrf):
	raise ValueError(f'VRF "{vrf}" does not exist!')
	output = cmd(f'ip --json --brief link show vrf {vrf}')
	answer = json.loads(output)
	for data in answer:
	if 'ifname' in data:
	interfaces.append(data.get('ifname'))
	except:
	pass
	return interfaces

	def get_interface_vrf(interface):
	""" Returns VRF of given interface """
	from vyos.utils.dict import dict_search
	from vyos.utils.network import get_interface_config
	tmp = get_interface_config(interface)
	if dict_search('linkinfo.info_slave_kind', tmp) == 'vrf':
	return tmp['master']
	return 'default'

	def get_interface_config(interface):
	""" Returns the used encapsulation protocol for given interface.
	If interface does not exist, None is returned.
	"""
	import os
	if not os.path.exists(f'/sys/class/net/{interface}'):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'ip --detail --json link show dev {interface}'))[0]
	return tmp

	def get_interface_address(interface):
	""" Returns the used encapsulation protocol for given interface.
	If interface does not exist, None is returned.
	"""
	import os
	if not os.path.exists(f'/sys/class/net/{interface}'):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'ip --detail --json addr show dev {interface}'))[0]
	return tmp

	def get_interface_namespace(iface):
	"""
	Returns wich netns the interface belongs to
	"""
	from json import loads
	from vyos.utils.process import cmd
	# Check if netns exist
	tmp = loads(cmd(f'ip --json netns ls'))
	if len(tmp) == 0:
	return None

	for ns in tmp:
	netns = f'{ns["name"]}'
	# Search interface in each netns
	data = loads(cmd(f'ip netns exec {netns} ip --json link show'))
	for tmp in data:
	if iface == tmp["ifname"]:
	return netns

	def is_ipv6_tentative(iface: str, ipv6_address: str) -> bool:
	"""Check if IPv6 address is in tentative state.

	This function checks if an IPv6 address on a specific network interface is
	in the tentative state. IPv6 tentative addresses are not fully configured
	and are undergoing Duplicate Address Detection (DAD) to ensure they are
	unique on the network.

	Args:
	iface (str): The name of the network interface.
	ipv6_address (str): The IPv6 address to check.

	Returns:
	bool: True if the IPv6 address is tentative, False otherwise.
	"""
	import json
	from vyos.utils.process import rc_cmd

	rc, out = rc_cmd(f'ip -6 --json address show dev {iface}')
	if rc:
	return False

	data = json.loads(out)
	for addr_info in data[0]['addr_info']:
	if (
	addr_info.get('local') == ipv6_address and
	addr_info.get('tentative', False)
	):
	return True
	return False

	def is_wwan_connected(interface):
	""" Determine if a given WWAN interface, e.g. wwan0 is connected to the
	carrier network or not """
	import json
	from vyos.utils.dict import dict_search
	from vyos.utils.process import cmd
	from vyos.utils.process import is_systemd_service_active

	if not interface.startswith('wwan'):
	raise ValueError(f'Specified interface "{interface}" is not a WWAN interface')

	# ModemManager is required for connection(s) - if service is not running,
	# there won't be any connection at all!
	if not is_systemd_service_active('ModemManager.service'):
	return False

	modem = interface.lstrip('wwan')

	tmp = cmd(f'mmcli --modem {modem} --output-json')
	tmp = json.loads(tmp)

	# return True/False if interface is in connected state
	return dict_search('modem.generic.state', tmp) == 'connected'

	def get_bridge_fdb(interface):
	""" Returns the forwarding database entries for a given interface """
	import os
	if not os.path.exists(f'/sys/class/net/{interface}'):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'bridge -j fdb show dev {interface}'))
	return tmp

	def get_all_vrfs():
	""" Return a dictionary of all system wide known VRF instances """
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd('ip --json vrf list'))
	# Result is of type [{"name":"red","table":1000},{"name":"blue","table":2000}]
	# so we will re-arrange it to a more nicer representation:
	# {'red': {'table': 1000}, 'blue': {'table': 2000}}
	data = {}
	for entry in tmp:
	name = entry.pop('name')
	data[name] = entry
	return data

	def interface_list() -> list:
	from vyos.ifconfig import Section
	"""
	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())

	def mac2eui64(mac, prefix=None):
	"""
	Convert a MAC address to a EUI64 address or, with prefix provided, a full
	IPv6 address.
	Thankfully copied from https://gist.github.com/wido/f5e32576bb57b5cc6f934e177a37a0d3
	"""
	import re
	from ipaddress import ip_network
	# http://tools.ietf.org/html/rfc4291#section-2.5.1
	eui64 = re.sub(r'[.:-]', '', mac).lower()
	eui64 = eui64[0:6] + 'fffe' + eui64[6:]
	eui64 = hex(int(eui64[0:2], 16) ^ 2)[2:].zfill(2) + eui64[2:]

	if prefix is None:
	return ':'.join(re.findall(r'.{4}', eui64))
	else:
	try:
	net = ip_network(prefix, strict=False)
	euil = int('0x{0}'.format(eui64), 16)
	return str(net[euil])
	except: # pylint: disable=bare-except
	return

	def check_port_availability(ipaddress, port, protocol):
	"""
	Check if port is available and not used by any service
	Return False if a port is busy or IP address does not exists
	Should be used carefully for services that can start listening
	dynamically, because IP address may be dynamic too
	"""
	from socketserver import TCPServer, UDPServer
	from ipaddress import ip_address

	# verify arguments
	try:
	ipaddress = ip_address(ipaddress).compressed
	except:
	raise ValueError(f'The {ipaddress} is not a valid IPv4 or IPv6 address')
	if port not in range(1, 65536):
	raise ValueError(f'The port number {port} is not in the 1-65535 range')
	if protocol not in ['tcp', 'udp']:
	raise ValueError(f'The protocol {protocol} is not supported. Only tcp and udp are allowed')

	# check port availability
	try:
	if protocol == 'tcp':
	server = TCPServer((ipaddress, port), None, bind_and_activate=True)
	if protocol == 'udp':
	server = UDPServer((ipaddress, port), None, bind_and_activate=True)
	server.server_close()
	except Exception as e:
	# errno.h:
	#define EADDRINUSE 98 /* Address already in use */
	if e.errno == 98:
	return False

	return True

	def is_listen_port_bind_service(port: int, service: str) -> bool:
	"""Check if listen port bound to expected program name
	:param port: Bind port
	:param service: Program name
	:return: bool

	Example:
	% is_listen_port_bind_service(443, 'nginx')
	True
	% is_listen_port_bind_service(443, 'ocserv-main')
	False
	"""
	from psutil import net_connections as connections
	from psutil import Process as process
	for connection in connections():
	addr = connection.laddr
	pid = connection.pid
	pid_name = process(pid).name()
	pid_port = addr.port
	if service == pid_name and port == pid_port:
	return True
	return False

	def is_ipv6_link_local(addr):
	""" Check if addrsss is an IPv6 link-local address. Returns True/False """
	from ipaddress import ip_interface
	from vyos.template import is_ipv6
	addr = addr.split('%')[0]
	if is_ipv6(addr):
	if ip_interface(addr).is_link_local:
	return True

	return False

	-def is_addr_assigned(ip_address, vrf=None) -> bool:
	+def is_addr_assigned(ip_address, vrf=None, include_vrf=False) -> bool:
	""" Verify if the given IPv4/IPv6 address is assigned to any interface """
	from netifaces import interfaces
	from vyos.utils.network import get_interface_config
	from vyos.utils.dict import dict_search

	for interface in interfaces():
	# Check if interface belongs to the requested VRF, if this is not the
	# case there is no need to proceed with this data set - continue loop
	# with next element
	tmp = get_interface_config(interface)
	- if dict_search('master', tmp) != vrf:
	+ if dict_search('master', tmp) != vrf and not include_vrf:
	continue

	if is_intf_addr_assigned(interface, ip_address):
	return True

	return False

	def is_intf_addr_assigned(intf, address) -> bool:
	"""
	Verify if the given IPv4/IPv6 address is assigned to specific interface.
	It can check both a single IP address (e.g. 192.0.2.1 or a assigned CIDR
	address 192.0.2.1/24.
	"""
	from vyos.template import is_ipv4

	from netifaces import ifaddresses
	from netifaces import AF_INET
	from netifaces import AF_INET6

	# check if the requested address type is configured at all
	# {
	# 17: [{'addr': '08:00:27:d9:5b:04', 'broadcast': 'ff:ff:ff:ff:ff:ff'}],
	# 2: [{'addr': '10.0.2.15', 'netmask': '255.255.255.0', 'broadcast': '10.0.2.255'}],
	# 10: [{'addr': 'fe80::a00:27ff:fed9:5b04%eth0', 'netmask': 'ffff:ffff:ffff:ffff::'}]
	# }
	try:
	addresses = ifaddresses(intf)
	except ValueError as e:
	print(e)
	return False

	# determine IP version (AF_INET or AF_INET6) depending on passed address
	addr_type = AF_INET if is_ipv4(address) else AF_INET6

	# Check every IP address on this interface for a match
	netmask = None
	if '/' in address:
	address, netmask = address.split('/')
	for ip in addresses.get(addr_type, []):
	# ip can have the interface name in the 'addr' field, we need to remove it
	# {'addr': 'fe80::a00:27ff:fec5:f821%eth2', 'netmask': 'ffff:ffff:ffff:ffff::'}
	ip_addr = ip['addr'].split('%')[0]

	if not _are_same_ip(address, ip_addr):
	continue

	# we do not have a netmask to compare against, they are the same
	if not netmask:
	return True

	prefixlen = ''
	if is_ipv4(ip_addr):
	prefixlen = sum([bin(int(_)).count('1') for _ in ip['netmask'].split('.')])
	else:
	prefixlen = sum([bin(int(_,16)).count('1') for _ in ip['netmask'].split('/')[0].split(':') if _])

	if str(prefixlen) == netmask:
	return True

	return False

	def is_loopback_addr(addr):
	""" Check if supplied IPv4/IPv6 address is a loopback address """
	from ipaddress import ip_address
	return ip_address(addr).is_loopback

	def is_wireguard_key_pair(private_key: str, public_key:str) -> bool:
	"""
	Checks if public/private keys are keypair
	:param private_key: Wireguard private key
	:type private_key: str
	:param public_key: Wireguard public key
	:type public_key: str
	:return: If public/private keys are keypair returns True else False
	:rtype: bool
	"""
	from vyos.utils.process import cmd
	gen_public_key = cmd('wg pubkey', input=private_key)
	if gen_public_key == public_key:
	return True
	else:
	return False

	def is_subnet_connected(subnet, primary=False):
	"""
	Verify is the given IPv4/IPv6 subnet is connected to any interface on this
	system.

	primary check if the subnet is reachable via the primary IP address of this
	interface, or in other words has a broadcast address configured. ISC DHCP
	for instance will complain if it should listen on non broadcast interfaces.

	Return True/False
	"""
	from ipaddress import ip_address
	from ipaddress import ip_network

	from netifaces import ifaddresses
	from netifaces import interfaces
	from netifaces import AF_INET
	from netifaces import AF_INET6

	from vyos.template import is_ipv6

	# determine IP version (AF_INET or AF_INET6) depending on passed address
	addr_type = AF_INET
	if is_ipv6(subnet):
	addr_type = AF_INET6

	for interface in interfaces():
	# check if the requested address type is configured at all
	if addr_type not in ifaddresses(interface).keys():
	continue

	# An interface can have multiple addresses, but some software components
	# only support the primary address :(
	if primary:
	ip = ifaddresses(interface)[addr_type][0]['addr']
	if ip_address(ip) in ip_network(subnet):
	return True
	else:
	# Check every assigned IP address if it is connected to the subnet
	# in question
	for ip in ifaddresses(interface)[addr_type]:
	# remove interface extension (e.g. %eth0) that gets thrown on the end of _some_ addrs
	addr = ip['addr'].split('%')[0]
	if ip_address(addr) in ip_network(subnet):
	return True

	return False

	def is_afi_configured(interface: str, afi):
	""" Check if given address family is configured, or in other words - an IP
	address is assigned to the interface. """
	from netifaces import ifaddresses
	from netifaces import AF_INET
	from netifaces import AF_INET6

	if afi not in [AF_INET, AF_INET6]:
	raise ValueError('Address family must be in [AF_INET, AF_INET6]')

	try:
	addresses = ifaddresses(interface)
	except ValueError as e:
	print(e)
	return False

	return afi in addresses

	def get_vxlan_vlan_tunnels(interface: str) -> list:
	""" Return a list of strings with VLAN IDs configured in the Kernel """
	from json import loads
	from vyos.utils.process import cmd

	if not interface.startswith('vxlan'):
	raise ValueError('Only applicable for VXLAN interfaces!')

	# Determine current OS Kernel configured VLANs
	#
	# $ bridge -j -p vlan tunnelshow dev vxlan0
	# [ {
	# "ifname": "vxlan0",
	# "tunnels": [ {
	# "vlan": 10,
	# "vlanEnd": 11,
	# "tunid": 10010,
	# "tunidEnd": 10011
	# },{
	# "vlan": 20,
	# "tunid": 10020
	# } ]
	# } ]
	#
	os_configured_vlan_ids = []
	tmp = loads(cmd(f'bridge --json vlan tunnelshow dev {interface}'))
	if tmp:
	for tunnel in tmp[0].get('tunnels', {}):
	vlanStart = tunnel['vlan']
	if 'vlanEnd' in tunnel:
	vlanEnd = tunnel['vlanEnd']
	# Build a real list for user VLAN IDs
	vlan_list = list(range(vlanStart, vlanEnd +1))
	# Convert list of integers to list or strings
	os_configured_vlan_ids.extend(map(str, vlan_list))
	# Proceed with next tunnel - this one is complete
	continue

	# Add single tunel id - not part of a range
	os_configured_vlan_ids.append(str(vlanStart))

	return os_configured_vlan_ids

	def get_vxlan_vni_filter(interface: str) -> list:
	""" Return a list of strings with VNIs configured in the Kernel"""
	from json import loads
	from vyos.utils.process import cmd

	if not interface.startswith('vxlan'):
	raise ValueError('Only applicable for VXLAN interfaces!')

	# Determine current OS Kernel configured VNI filters in VXLAN interface
	#
	# $ bridge -j vni show dev vxlan1
	# [{"ifname":"vxlan1","vnis":[{"vni":100},{"vni":200},{"vni":300,"vniEnd":399}]}]
	#
	# Example output: ['10010', '10020', '10021', '10022']
	os_configured_vnis = []
	tmp = loads(cmd(f'bridge --json vni show dev {interface}'))
	if tmp:
	for tunnel in tmp[0].get('vnis', {}):
	vniStart = tunnel['vni']
	if 'vniEnd' in tunnel:
	vniEnd = tunnel['vniEnd']
	# Build a real list for user VNIs
	vni_list = list(range(vniStart, vniEnd +1))
	# Convert list of integers to list or strings
	os_configured_vnis.extend(map(str, vni_list))
	# Proceed with next tunnel - this one is complete
	continue

	# Add single tunel id - not part of a range
	os_configured_vnis.append(str(vniStart))

	return os_configured_vnis
	diff --git a/smoketest/scripts/cli/test_service_dhcp-server.py b/smoketest/scripts/cli/test_service_dhcp-server.py
	index 24ae2f4e2..3d1f47ece 100755
	--- a/smoketest/scripts/cli/test_service_dhcp-server.py
	+++ b/smoketest/scripts/cli/test_service_dhcp-server.py
	@@ -1,505 +1,527 @@
	#!/usr/bin/env python3
	#
	# Copyright (C) 2020-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 unittest

	from base_vyostest_shim import VyOSUnitTestSHIM

	from vyos.configsession import ConfigSessionError
	from vyos.utils.process import process_named_running
	from vyos.utils.file import read_file
	from vyos.template import address_from_cidr
	from vyos.template import inc_ip
	from vyos.template import dec_ip
	from vyos.template import netmask_from_cidr

	PROCESS_NAME = 'dhcpd'
	DHCPD_CONF = '/run/dhcp-server/dhcpd.conf'
	base_path = ['service', 'dhcp-server']
	subnet = '192.0.2.0/25'
	router = inc_ip(subnet, 1)
	dns_1 = inc_ip(subnet, 2)
	dns_2 = inc_ip(subnet, 3)
	domain_name = 'vyos.net'

	class TestServiceDHCPServer(VyOSUnitTestSHIM.TestCase):
	@classmethod
	def setUpClass(cls):
	super(TestServiceDHCPServer, cls).setUpClass()

	cidr_mask = subnet.split('/')[-1]
	cls.cli_set(cls, ['interfaces', 'dummy', 'dum8765', 'address', f'{router}/{cidr_mask}'])

	@classmethod
	def tearDownClass(cls):
	cls.cli_delete(cls, ['interfaces', 'dummy', 'dum8765'])
	super(TestServiceDHCPServer, cls).tearDownClass()

	def tearDown(self):
	self.cli_delete(base_path)
	self.cli_commit()

	def test_dhcp_single_pool_range(self):
	shared_net_name = 'SMOKE-1'

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)
	range_1_start = inc_ip(subnet, 40)
	range_1_stop = inc_ip(subnet, 50)

	self.cli_set(base_path + ['dynamic-dns-update'])

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['name-server', dns_1])
	self.cli_set(pool + ['name-server', dns_2])
	self.cli_set(pool + ['domain-name', domain_name])
	self.cli_set(pool + ['ping-check'])

	# check validate() - No DHCP address range or active static-mapping set
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])
	self.cli_set(pool + ['range', '1', 'start', range_1_start])
	self.cli_set(pool + ['range', '1', 'stop', range_1_stop])

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)
	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)
	self.assertIn(f'ddns-update-style interim;', config)
	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option domain-name-servers {dns_1}, {dns_2};', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'option domain-name "{domain_name}";', config)
	self.assertIn(f'default-lease-time 86400;', config)
	self.assertIn(f'max-lease-time 86400;', config)
	self.assertIn(f'ping-check true;', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)
	self.assertIn(f'range {range_1_start} {range_1_stop};', config)
	self.assertIn(f'set shared-networkname = "{shared_net_name}";', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_single_pool_options(self):
	shared_net_name = 'SMOKE-0815'

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)
	smtp_server = '1.2.3.4'
	time_server = '4.3.2.1'
	tftp_server = 'tftp.vyos.io'
	search_domains = ['foo.vyos.net', 'bar.vyos.net']
	bootfile_name = 'vyos'
	bootfile_server = '192.0.2.1'
	wpad = 'http://wpad.vyos.io/foo/bar'
	server_identifier = bootfile_server
	ipv6_only_preferred = '300'

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['name-server', dns_1])
	self.cli_set(pool + ['name-server', dns_2])
	self.cli_set(pool + ['domain-name', domain_name])
	self.cli_set(pool + ['ip-forwarding'])
	self.cli_set(pool + ['smtp-server', smtp_server])
	self.cli_set(pool + ['pop-server', smtp_server])
	self.cli_set(pool + ['time-server', time_server])
	self.cli_set(pool + ['tftp-server-name', tftp_server])
	for search in search_domains:
	self.cli_set(pool + ['domain-search', search])
	self.cli_set(pool + ['bootfile-name', bootfile_name])
	self.cli_set(pool + ['bootfile-server', bootfile_server])
	self.cli_set(pool + ['wpad-url', wpad])
	self.cli_set(pool + ['server-identifier', server_identifier])

	self.cli_set(pool + ['static-route', '10.0.0.0/24', 'next-hop', '192.0.2.1'])
	self.cli_set(pool + ['ipv6-only-preferred', ipv6_only_preferred])

	# check validate() - No DHCP address range or active static-mapping set
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)

	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)
	self.assertIn(f'ddns-update-style none;', config)
	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option domain-name-servers {dns_1}, {dns_2};', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'option domain-name "{domain_name}";', config)

	search = '"' + ('", "').join(search_domains) + '"'
	self.assertIn(f'option domain-search {search};', config)

	self.assertIn(f'option ip-forwarding true;', config)
	self.assertIn(f'option smtp-server {smtp_server};', config)
	self.assertIn(f'option pop-server {smtp_server};', config)
	self.assertIn(f'option time-servers {time_server};', config)
	self.assertIn(f'option wpad-url "{wpad}";', config)
	self.assertIn(f'option dhcp-server-identifier {server_identifier};', config)
	self.assertIn(f'option tftp-server-name "{tftp_server}";', config)
	self.assertIn(f'option bootfile-name "{bootfile_name}";', config)
	self.assertIn(f'filename "{bootfile_name}";', config)
	self.assertIn(f'next-server {bootfile_server};', config)
	self.assertIn(f'default-lease-time 86400;', config)
	self.assertIn(f'max-lease-time 86400;', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)
	self.assertIn(f'set shared-networkname = "{shared_net_name}";', config)
	self.assertIn(f'option rfc8925-ipv6-only-preferred {ipv6_only_preferred};', config)

	# weird syntax for those static routes
	self.assertIn(f'option rfc3442-static-route 24,10,0,0,192,0,2,1, 0,192,0,2,1;', config)
	self.assertIn(f'option windows-static-route 24,10,0,0,192,0,2,1;', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_single_pool_static_mapping(self):
	shared_net_name = 'SMOKE-2'
	domain_name = 'private'

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['name-server', dns_1])
	self.cli_set(pool + ['name-server', dns_2])
	self.cli_set(pool + ['domain-name', domain_name])

	# check validate() - No DHCP address range or active static-mapping set
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	client_base = 10
	for client in ['client1', 'client2', 'client3']:
	mac = '00:50:00:00:00:{}'.format(client_base)
	self.cli_set(pool + ['static-mapping', client, 'mac-address', mac])
	self.cli_set(pool + ['static-mapping', client, 'ip-address', inc_ip(subnet, client_base)])
	client_base += 1

	# cannot have mappings with duplicate IP addresses
	self.cli_set(pool + ['static-mapping', 'dupe1', 'mac-address', '00:50:00:00:fe:ff'])
	self.cli_set(pool + ['static-mapping', 'dupe1', 'ip-address', inc_ip(subnet, 10)])
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_set(pool + ['static-mapping', 'dupe1', 'disable'])
	self.cli_commit()
	self.cli_delete(pool + ['static-mapping', 'dupe1'])

	# cannot have mappings with duplicate MAC addresses
	self.cli_set(pool + ['static-mapping', 'dupe2', 'mac-address', '00:50:00:00:00:10'])
	self.cli_set(pool + ['static-mapping', 'dupe2', 'ip-address', inc_ip(subnet, 120)])
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_delete(pool + ['static-mapping', 'dupe2'])

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)
	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)
	self.assertIn(f'ddns-update-style none;', config)
	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option domain-name-servers {dns_1}, {dns_2};', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'option domain-name "{domain_name}";', config)
	self.assertIn(f'default-lease-time 86400;', config)
	self.assertIn(f'max-lease-time 86400;', config)

	client_base = 10
	for client in ['client1', 'client2', 'client3']:
	mac = '00:50:00:00:00:{}'.format(client_base)
	ip = inc_ip(subnet, client_base)
	self.assertIn(f'host {shared_net_name}_{client}' + ' {', config)
	self.assertIn(f'fixed-address {ip};', config)
	self.assertIn(f'hardware ethernet {mac};', config)
	client_base += 1

	self.assertIn(f'set shared-networkname = "{shared_net_name}";', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_multiple_pools(self):
	lease_time = '14400'

	for network in ['0', '1', '2', '3']:
	shared_net_name = f'VyOS-SMOKETEST-{network}'
	subnet = f'192.0.{network}.0/24'
	router = inc_ip(subnet, 1)
	dns_1 = inc_ip(subnet, 2)

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)
	range_1_start = inc_ip(subnet, 30)
	range_1_stop = inc_ip(subnet, 40)

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['name-server', dns_1])
	self.cli_set(pool + ['domain-name', domain_name])
	self.cli_set(pool + ['lease', lease_time])

	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])
	self.cli_set(pool + ['range', '1', 'start', range_1_start])
	self.cli_set(pool + ['range', '1', 'stop', range_1_stop])

	client_base = 60
	for client in ['client1', 'client2', 'client3', 'client4']:
	mac = '02:50:00:00:00:{}'.format(client_base)
	self.cli_set(pool + ['static-mapping', client, 'mac-address', mac])
	self.cli_set(pool + ['static-mapping', client, 'ip-address', inc_ip(subnet, client_base)])
	client_base += 1

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)
	for network in ['0', '1', '2', '3']:
	shared_net_name = f'VyOS-SMOKETEST-{network}'
	subnet = f'192.0.{network}.0/24'
	router = inc_ip(subnet, 1)
	dns_1 = inc_ip(subnet, 2)

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)
	range_1_start = inc_ip(subnet, 30)
	range_1_stop = inc_ip(subnet, 40)

	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)

	self.assertIn(f'ddns-update-style none;', config)
	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option domain-name-servers {dns_1};', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'option domain-name "{domain_name}";', config)
	self.assertIn(f'default-lease-time {lease_time};', config)
	self.assertIn(f'max-lease-time {lease_time};', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)
	self.assertIn(f'range {range_1_start} {range_1_stop};', config)
	self.assertIn(f'set shared-networkname = "{shared_net_name}";', config)

	client_base = 60
	for client in ['client1', 'client2', 'client3', 'client4']:
	mac = '02:50:00:00:00:{}'.format(client_base)
	ip = inc_ip(subnet, client_base)
	self.assertIn(f'host {shared_net_name}_{client}' + ' {', config)
	self.assertIn(f'fixed-address {ip};', config)
	self.assertIn(f'hardware ethernet {mac};', config)
	client_base += 1

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_exclude_not_in_range(self):
	# T3180: verify else path when slicing DHCP ranges and exclude address
	# is not part of the DHCP range
	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)

	pool = base_path + ['shared-network-name', 'EXCLUDE-TEST', 'subnet', subnet]
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['exclude', router])
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	# commit changes
	self.cli_commit()

	# VErify
	config = read_file(DHCPD_CONF)
	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)

	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_exclude_in_range(self):
	# T3180: verify else path when slicing DHCP ranges and exclude address
	# is not part of the DHCP range
	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 100)

	# the DHCP exclude addresse is blanked out of the range which is done
	# by slicing one range into two ranges
	exclude_addr = inc_ip(range_0_start, 20)
	range_0_stop_excl = dec_ip(exclude_addr, 1)
	range_0_start_excl = inc_ip(exclude_addr, 1)

	pool = base_path + ['shared-network-name', 'EXCLUDE-TEST-2', 'subnet', subnet]
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['exclude', exclude_addr])
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	# commit changes
	self.cli_commit()

	# Verify
	config = read_file(DHCPD_CONF)
	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)

	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'range {range_0_start} {range_0_stop_excl};', config)
	self.assertIn(f'range {range_0_start_excl} {range_0_stop};', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_relay_server(self):
	# Listen on specific address and return DHCP leases from a non
	# directly connected pool
	self.cli_set(base_path + ['listen-address', router])

	relay_subnet = '10.0.0.0/16'
	relay_router = inc_ip(relay_subnet, 1)

	range_0_start = '10.0.1.0'
	range_0_stop = '10.0.250.255'

	pool = base_path + ['shared-network-name', 'RELAY', 'subnet', relay_subnet]
	self.cli_set(pool + ['default-router', relay_router])
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)
	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)
	# Check the relay network
	self.assertIn(f'subnet {network} netmask {netmask}' + r' { }', config)

	relay_network = address_from_cidr(relay_subnet)
	relay_netmask = netmask_from_cidr(relay_subnet)
	self.assertIn(f'subnet {relay_network} netmask {relay_netmask}' + r' {', config)
	self.assertIn(f'option routers {relay_router};', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_invalid_raw_options(self):
	shared_net_name = 'SMOKE-5'

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	self.cli_set(base_path + ['global-parameters', 'this-is-crap'])
	# check generate() - dhcpd should not acceot this garbage config
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_delete(base_path + ['global-parameters'])

	# commit changes
	self.cli_commit()

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	def test_dhcp_failover(self):
	shared_net_name = 'FAILOVER'
	failover_name = 'VyOS-Failover'

	range_0_start = inc_ip(subnet, 10)
	range_0_stop = inc_ip(subnet, 20)

	pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
	# we use the first subnet IP address as default gateway
	self.cli_set(pool + ['default-router', router])

	# check validate() - No DHCP address range or active static-mapping set
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_set(pool + ['range', '0', 'start', range_0_start])
	self.cli_set(pool + ['range', '0', 'stop', range_0_stop])

	# failover
	failover_local = router
	failover_remote = inc_ip(router, 1)

	self.cli_set(base_path + ['failover', 'source-address', failover_local])
	self.cli_set(base_path + ['failover', 'name', failover_name])
	self.cli_set(base_path + ['failover', 'remote', failover_remote])
	self.cli_set(base_path + ['failover', 'status', 'primary'])

	# check validate() - failover needs to be enabled for at least one subnet
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_set(pool + ['enable-failover'])

	# commit changes
	self.cli_commit()

	config = read_file(DHCPD_CONF)

	self.assertIn(f'failover peer "{failover_name}"' + r' {', config)
	self.assertIn(f'primary;', config)
	self.assertIn(f'mclt 1800;', config)
	self.assertIn(f'mclt 1800;', config)
	self.assertIn(f'split 128;', config)
	self.assertIn(f'port 647;', config)
	self.assertIn(f'peer port 647;', config)
	self.assertIn(f'max-response-delay 30;', config)
	self.assertIn(f'max-unacked-updates 10;', config)
	self.assertIn(f'load balance max seconds 3;', config)
	self.assertIn(f'address {failover_local};', config)
	self.assertIn(f'peer address {failover_remote};', config)

	network = address_from_cidr(subnet)
	netmask = netmask_from_cidr(subnet)
	self.assertIn(f'ddns-update-style none;', config)
	self.assertIn(f'subnet {network} netmask {netmask}' + r' {', config)
	self.assertIn(f'option routers {router};', config)
	self.assertIn(f'range {range_0_start} {range_0_stop};', config)
	self.assertIn(f'set shared-networkname = "{shared_net_name}";', config)
	self.assertIn(f'failover peer "{failover_name}";', config)
	self.assertIn(f'deny dynamic bootp clients;', config)

	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))

	+ def test_dhcp_on_interface_with_vrf(self):
	+ self.cli_set(['interfaces', 'ethernet', 'eth1', 'address', '10.1.1.1/30'])
	+ self.cli_set(['interfaces', 'ethernet', 'eth1', 'vrf', 'SMOKE-DHCP'])
	+ self.cli_set(['protocols', 'static', 'route', '10.1.10.0/24', 'interface', 'eth1', 'vrf', 'SMOKE-DHCP'])
	+ self.cli_set(['vrf', 'name', 'SMOKE-DHCP', 'protocols', 'static', 'route', '10.1.10.0/24', 'next-hop', '10.1.1.2'])
	+ self.cli_set(['vrf', 'name', 'SMOKE-DHCP', 'table', '1000'])
	+ self.cli_set(base_path + ['shared-network-name', 'SMOKE-DHCP-NETWORK', 'subnet', '10.1.10.0/24', 'default-router', '10.1.10.1'])
	+ self.cli_set(base_path + ['shared-network-name', 'SMOKE-DHCP-NETWORK', 'subnet', '10.1.10.0/24', 'name-server', '1.1.1.1'])
	+ self.cli_set(base_path + ['shared-network-name', 'SMOKE-DHCP-NETWORK', 'subnet', '10.1.10.0/24', 'range', '1', 'start', '10.1.10.10'])
	+ self.cli_set(base_path + ['shared-network-name', 'SMOKE-DHCP-NETWORK', 'subnet', '10.1.10.0/24', 'range', '1', 'stop', '10.1.10.20'])
	+ self.cli_set(base_path + ['listen-address', '10.1.1.1'])
	+ self.cli_commit()
	+
	+ config = read_file(DHCPD_CONF)
	+ self.assertIn('subnet 10.1.1.0 netmask 255.255.255.252', config)
	+
	+ self.cli_delete(['interfaces', 'ethernet', 'eth1', 'vrf', 'SMOKE-DHCP'])
	+ self.cli_delete(['protocols', 'static', 'route', '10.1.10.0/24', 'interface', 'eth1', 'vrf'])
	+ self.cli_delete(['vrf', 'name', 'SMOKE-DHCP'])
	+ self.cli_commit()
	+
	+
	if __name__ == '__main__':
	unittest.main(verbosity=2)
	diff --git a/src/conf_mode/service_dhcp-server.py b/src/conf_mode/service_dhcp-server.py
	index 90056887a..e910ecdf7 100755
	--- a/src/conf_mode/service_dhcp-server.py
	+++ b/src/conf_mode/service_dhcp-server.py
	@@ -1,350 +1,349 @@
	#!/usr/bin/env python3
	#
	# Copyright (C) 2018-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 os

	from ipaddress import ip_address
	from ipaddress import ip_network
	from netaddr import IPAddress
	from netaddr import IPRange
	from sys import exit

	from vyos.base import DeprecationWarning
	from vyos.config import Config
	from vyos.template import render
	from vyos.utils.dict import dict_search
	from vyos.utils.process import call
	from vyos.utils.process import run
	from vyos.utils.network import is_subnet_connected
	from vyos.utils.network import is_addr_assigned
	from vyos import ConfigError
	from vyos import airbag
	airbag.enable()

	config_file = '/run/dhcp-server/dhcpd.conf'
	systemd_override = r'/run/systemd/system/isc-dhcp-server.service.d/10-override.conf'

	def dhcp_slice_range(exclude_list, range_dict):
	"""
	This function is intended to slice a DHCP range. What does it mean?

	Lets assume we have a DHCP range from '192.0.2.1' to '192.0.2.100'
	but want to exclude address '192.0.2.74' and '192.0.2.75'. We will
	pass an input 'range_dict' in the format:
	{'start' : '192.0.2.1', 'stop' : '192.0.2.100' }
	and we will receive an output list of:
	[{'start' : '192.0.2.1' , 'stop' : '192.0.2.73' },
	{'start' : '192.0.2.76', 'stop' : '192.0.2.100' }]
	The resulting list can then be used in turn to build the proper dhcpd
	configuration file.
	"""
	output = []
	# exclude list must be sorted for this to work
	exclude_list = sorted(exclude_list)
	range_start = range_dict['start']
	range_stop = range_dict['stop']
	range_last_exclude = ''

	for e in exclude_list:
	if (ip_address(e) >= ip_address(range_start)) and \
	(ip_address(e) <= ip_address(range_stop)):
	range_last_exclude = e

	for e in exclude_list:
	if (ip_address(e) >= ip_address(range_start)) and \
	(ip_address(e) <= ip_address(range_stop)):

	# Build new address range ending one address before exclude address
	r = {
	'start' : range_start,
	'stop' : str(ip_address(e) -1)
	}
	# On the next run our address range will start one address after
	# the exclude address
	range_start = str(ip_address(e) + 1)

	# on subsequent exclude addresses we can not
	# append them to our output
	if not (ip_address(r['start']) > ip_address(r['stop'])):
	# Everything is fine, add range to result
	output.append(r)

	# Take care of last IP address range spanning from the last exclude
	# address (+1) to the end of the initial configured range
	if ip_address(e) == ip_address(range_last_exclude):
	r = {
	'start': str(ip_address(e) + 1),
	'stop': str(range_stop)
	}
	if not (ip_address(r['start']) > ip_address(r['stop'])):
	output.append(r)
	else:
	# if the excluded address was not part of the range, we simply return
	# the entire ranga again
	if not range_last_exclude:
	if range_dict not in output:
	output.append(range_dict)

	return output

	def get_config(config=None):
	if config:
	conf = config
	else:
	conf = Config()
	base = ['service', 'dhcp-server']
	if not conf.exists(base):
	return None

	dhcp = conf.get_config_dict(base, key_mangling=('-', '_'),
	no_tag_node_value_mangle=True,
	get_first_key=True,
	with_recursive_defaults=True)

	if 'shared_network_name' in dhcp:
	for network, network_config in dhcp['shared_network_name'].items():
	if 'subnet' in network_config:
	for subnet, subnet_config in network_config['subnet'].items():
	# If exclude IP addresses are defined we need to slice them out of
	# the defined ranges
	if {'exclude', 'range'} <= set(subnet_config):
	new_range_id = 0
	new_range_dict = {}
	for r, r_config in subnet_config['range'].items():
	for slice in dhcp_slice_range(subnet_config['exclude'], r_config):
	new_range_dict.update({new_range_id : slice})
	new_range_id +=1

	dhcp['shared_network_name'][network]['subnet'][subnet].update(
	{'range' : new_range_dict})

	return dhcp

	def verify(dhcp):
	# bail out early - looks like removal from running config
	if not dhcp or 'disable' in dhcp:
	return None

	# If DHCP is enabled we need one share-network
	if 'shared_network_name' not in dhcp:
	raise ConfigError('No DHCP shared networks configured.\n' \
	'At least one DHCP shared network must be configured.')

	# Inspect shared-network/subnet
	listen_ok = False
	subnets = []
	failover_ok = False
	shared_networks = len(dhcp['shared_network_name'])
	disabled_shared_networks = 0

	common_deprecation_msg = 'are subject of removal in VyOS 1.5! Please raise a feature request for proper CLI nodes!'
	if 'global_parameters' in dhcp:
	DeprecationWarning(f'Additional global parameters {common_deprecation_msg}')

	# A shared-network requires a subnet definition
	for network, network_config in dhcp['shared_network_name'].items():
	if 'shared_network_parameters' in network_config:
	DeprecationWarning(f'Additional shared network parameters in "{network}" {common_deprecation_msg}')

	if 'disable' in network_config:
	disabled_shared_networks += 1

	if 'subnet' not in network_config:
	raise ConfigError(f'No subnets defined for {network}. At least one\n' \
	'lease subnet must be configured.')

	for subnet, subnet_config in network_config['subnet'].items():
	if 'subnet_parameters' in subnet_config:
	DeprecationWarning(f'Additional subnet parameters in "{subnet}" {common_deprecation_msg}')

	# All delivered static routes require a next-hop to be set
	if 'static_route' in subnet_config:
	for route, route_option in subnet_config['static_route'].items():
	if 'next_hop' not in route_option:
	raise ConfigError(f'DHCP static-route "{route}" requires router to be defined!')

	# DHCP failover needs at least one subnet that uses it
	if 'enable_failover' in subnet_config:
	if 'failover' not in dhcp:
	raise ConfigError(f'Can not enable failover for "{subnet}" in "{network}".\n' \
	'Failover is not configured globally!')
	failover_ok = True

	# Check if DHCP address range is inside configured subnet declaration
	if 'range' in subnet_config:
	networks = []
	for range, range_config in subnet_config['range'].items():
	if not {'start', 'stop'} <= set(range_config):
	raise ConfigError(f'DHCP range "{range}" start and stop address must be defined!')

	# Start/Stop address must be inside network
	for key in ['start', 'stop']:
	if ip_address(range_config[key]) not in ip_network(subnet):
	raise ConfigError(f'DHCP range "{range}" {key} address not within shared-network "{network}, {subnet}"!')

	# Stop address must be greater or equal to start address
	if ip_address(range_config['stop']) < ip_address(range_config['start']):
	raise ConfigError(f'DHCP range "{range}" stop address must be greater or equal\n' \
	'to the ranges start address!')

	for network in networks:
	start = range_config['start']
	stop = range_config['stop']
	if start in network:
	raise ConfigError(f'Range "{range}" start address "{start}" already part of another range!')
	if stop in network:
	raise ConfigError(f'Range "{range}" stop address "{stop}" already part of another range!')

	tmp = IPRange(range_config['start'], range_config['stop'])
	networks.append(tmp)

	# Exclude addresses must be in bound
	if 'exclude' in subnet_config:
	for exclude in subnet_config['exclude']:
	if ip_address(exclude) not in ip_network(subnet):
	raise ConfigError(f'Excluded IP address "{exclude}" not within shared-network "{network}, {subnet}"!')

	# At least one DHCP address range or static-mapping required
	if 'range' not in subnet_config and 'static_mapping' not in subnet_config:
	raise ConfigError(f'No DHCP address range or active static-mapping configured\n' \
	f'within shared-network "{network}, {subnet}"!')

	if 'static_mapping' in subnet_config:
	# Static mappings require just a MAC address (will use an IP from the dynamic pool if IP is not set)
	used_ips = []
	used_mac = []
	for mapping, mapping_config in subnet_config['static_mapping'].items():
	if 'ip_address' in mapping_config:
	if ip_address(mapping_config['ip_address']) not in ip_network(subnet):
	raise ConfigError(f'Configured static lease address for mapping "{mapping}" is\n' \
	f'not within shared-network "{network}, {subnet}"!')

	if 'mac_address' not in mapping_config:
	raise ConfigError(f'MAC address required for static mapping "{mapping}"\n' \
	f'within shared-network "{network}, {subnet}"!')

	if 'disable' not in mapping_config:
	if mapping_config['ip_address'] in used_ips:
	raise ConfigError(f'Configured IP address for static mapping "{mapping}" already exists on another static mapping')
	used_ips.append(mapping_config['ip_address'])

	if 'mac_address' in mapping_config and 'disable' not in mapping_config:
	if mapping_config['mac_address'] in used_mac:
	raise ConfigError(f'Configured MAC address for static mapping "{mapping}" already exists on another static mapping')
	used_mac.append(mapping_config['mac_address'])

	# There must be one subnet connected to a listen interface.
	# This only counts if the network itself is not disabled!
	if 'disable' not in network_config:
	if is_subnet_connected(subnet, primary=False):
	listen_ok = True

	# Subnets must be non overlapping
	if subnet in subnets:
	raise ConfigError(f'Configured subnets must be unique! Subnet "{subnet}"\n'
	'defined multiple times!')
	subnets.append(subnet)

	# Check for overlapping subnets
	net = ip_network(subnet)
	for n in subnets:
	net2 = ip_network(n)
	if (net != net2):
	if net.overlaps(net2):
	raise ConfigError(f'Conflicting subnet ranges: "{net}" overlaps "{net2}"!')

	# Prevent 'disable' for shared-network if only one network is configured
	if (shared_networks - disabled_shared_networks) < 1:
	raise ConfigError(f'At least one shared network must be active!')

	if 'failover' in dhcp:
	if not failover_ok:
	raise ConfigError('DHCP failover must be enabled for at least one subnet!')

	for key in ['name', 'remote', 'source_address', 'status']:
	if key not in dhcp['failover']:
	tmp = key.replace('_', '-')
	raise ConfigError(f'DHCP failover requires "{tmp}" to be specified!')

	for address in (dict_search('listen_address', dhcp) or []):
	- if is_addr_assigned(address):
	+ if is_addr_assigned(address, include_vrf=True):
	listen_ok = True
	# no need to probe further networks, we have one that is valid
	continue
	else:
	raise ConfigError(f'listen-address "{address}" not configured on any interface')

	-
	if not listen_ok:
	raise ConfigError('None of the configured subnets have an appropriate primary IP address on any\n'
	'broadcast interface configured, nor was there an explicit listen-address\n'
	'configured for serving DHCP relay packets!')

	return None

	def generate(dhcp):
	# bail out early - looks like removal from running config
	if not dhcp or 'disable' in dhcp:
	return None

	# Please see: https://vyos.dev/T1129 for quoting of the raw
	# parameters we can pass to ISC DHCPd
	tmp_file = '/tmp/dhcpd.conf'
	render(tmp_file, 'dhcp-server/dhcpd.conf.j2', dhcp,
	formater=lambda _: _.replace(""", '"'))
	# XXX: as we have the ability for a user to pass in "raw" options via VyOS
	# CLI (see T3544) we now ask ISC dhcpd to test the newly rendered
	# configuration
	tmp = run(f'/usr/sbin/dhcpd -4 -q -t -cf {tmp_file}')
	if tmp > 0:
	if os.path.exists(tmp_file):
	os.unlink(tmp_file)
	raise ConfigError('Configuration file errors encountered - check your options!')

	# Now that we know that the newly rendered configuration is "good" we can
	# render the "real" configuration
	render(config_file, 'dhcp-server/dhcpd.conf.j2', dhcp,
	formater=lambda _: _.replace(""", '"'))
	render(systemd_override, 'dhcp-server/10-override.conf.j2', dhcp)

	# Clean up configuration test file
	if os.path.exists(tmp_file):
	os.unlink(tmp_file)

	return None

	def apply(dhcp):
	call('systemctl daemon-reload')
	# bail out early - looks like removal from running config
	if not dhcp or 'disable' in dhcp:
	call('systemctl stop isc-dhcp-server.service')
	if os.path.exists(config_file):
	os.unlink(config_file)

	return None

	call('systemctl restart isc-dhcp-server.service')
	return None

	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)

File Metadata

Mime Type: text/x-diff
Expires: Fri, Nov 6, 8:59 AM (1 d, 4 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4641823
Default Alt Text: (58 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions