diff --git a/interface-definitions/service_dhcp-server.xml.in b/interface-definitions/service_dhcp-server.xml.in
index 2afa05a8a..cb5f9a804 100644
--- a/interface-definitions/service_dhcp-server.xml.in
+++ b/interface-definitions/service_dhcp-server.xml.in
@@ -1,229 +1,250 @@
<?xml version="1.0"?>
<!-- DHCP server configuration -->
<interfaceDefinition>
<node name="service">
<children>
<node name="dhcp-server" owner="${vyos_conf_scripts_dir}/service_dhcp-server.py">
<properties>
<help>Dynamic Host Configuration Protocol (DHCP) for DHCP server</help>
<priority>911</priority>
</properties>
<children>
#include <include/generic-disable-node.xml.i>
<leafNode name="dynamic-dns-update">
<properties>
<help>Dynamically update Domain Name System (RFC4702)</help>
<valueless/>
</properties>
</leafNode>
<node name="high-availability">
<properties>
<help>DHCP high availability configuration</help>
</properties>
<children>
#include <include/source-address-ipv4.xml.i>
+ <leafNode name="mode">
+ <properties>
+ <help>Configure high availability mode</help>
+ <completionHelp>
+ <list>active-active active-passive</list>
+ </completionHelp>
+ <valueHelp>
+ <format>active-active</format>
+ <description>Both server attend DHCP requests</description>
+ </valueHelp>
+ <valueHelp>
+ <format>active-passive</format>
+ <description>Only primary server attends DHCP requests</description>
+ </valueHelp>
+ <constraint>
+ <regex>(active-active|active-passive)</regex>
+ </constraint>
+ <constraintErrorMessage>Invalid DHCP high availability mode</constraintErrorMessage>
+ </properties>
+ <defaultValue>active-active</defaultValue>
+ </leafNode>
<leafNode name="remote">
<properties>
<help>IPv4 remote address used for connection</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 address of high availability peer</description>
</valueHelp>
<constraint>
<validator name="ipv4-address"/>
</constraint>
</properties>
</leafNode>
<leafNode name="name">
<properties>
<help>Peer name used to identify connection</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
<constraintErrorMessage>Invalid failover peer name. May only contain letters, numbers and .-_</constraintErrorMessage>
</properties>
</leafNode>
<leafNode name="status">
<properties>
<help>High availability hierarchy</help>
<completionHelp>
<list>primary secondary</list>
</completionHelp>
<valueHelp>
<format>primary</format>
<description>Configure this server to be the primary node</description>
</valueHelp>
<valueHelp>
<format>secondary</format>
<description>Configure this server to be the secondary node</description>
</valueHelp>
<constraint>
<regex>(primary|secondary)</regex>
</constraint>
<constraintErrorMessage>Invalid DHCP high availability peer status</constraintErrorMessage>
</properties>
</leafNode>
#include <include/pki/ca-certificate.xml.i>
#include <include/pki/certificate.xml.i>
</children>
</node>
<leafNode name="hostfile-update">
<properties>
<help>Updating /etc/hosts file (per client lease)</help>
<valueless/>
</properties>
</leafNode>
#include <include/listen-address-ipv4.xml.i>
#include <include/listen-interface-multi-broadcast.xml.i>
<tagNode name="shared-network-name">
<properties>
<help>Name of DHCP shared network</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
<constraintErrorMessage>Invalid shared network name. May only contain letters, numbers and .-_</constraintErrorMessage>
</properties>
<children>
<leafNode name="authoritative">
<properties>
<help>Option to make DHCP server authoritative for this physical network</help>
<valueless/>
</properties>
</leafNode>
#include <include/dhcp/option-v4.xml.i>
#include <include/generic-description.xml.i>
#include <include/generic-disable-node.xml.i>
<tagNode name="subnet">
<properties>
<help>DHCP subnet for shared network</help>
<valueHelp>
<format>ipv4net</format>
<description>IPv4 address and prefix length</description>
</valueHelp>
<constraint>
<validator name="ipv4-prefix"/>
</constraint>
<constraintErrorMessage>Invalid IPv4 subnet definition</constraintErrorMessage>
</properties>
<children>
#include <include/dhcp/option-v4.xml.i>
#include <include/generic-description.xml.i>
#include <include/generic-disable-node.xml.i>
<leafNode name="exclude">
<properties>
<help>IP address to exclude from DHCP lease range</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 address to exclude from lease range</description>
</valueHelp>
<constraint>
<validator name="ipv4-address"/>
</constraint>
<multi/>
</properties>
</leafNode>
<leafNode name="ignore-client-id">
<properties>
<help>Ignore client identifier for lease lookups</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="lease">
<properties>
<help>Lease timeout in seconds</help>
<valueHelp>
<format>u32</format>
<description>DHCP lease time in seconds</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-4294967295"/>
</constraint>
<constraintErrorMessage>DHCP lease time must be between 0 and 4294967295 (49 days)</constraintErrorMessage>
</properties>
<defaultValue>86400</defaultValue>
</leafNode>
<tagNode name="range">
<properties>
<help>DHCP lease range</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
<constraintErrorMessage>Invalid range name, may only be alphanumeric, dot and hyphen</constraintErrorMessage>
</properties>
<children>
#include <include/dhcp/option-v4.xml.i>
<leafNode name="start">
<properties>
<help>First IP address for DHCP lease range</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 start address of pool</description>
</valueHelp>
<constraint>
<validator name="ipv4-address"/>
</constraint>
</properties>
</leafNode>
<leafNode name="stop">
<properties>
<help>Last IP address for DHCP lease range</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 end address of pool</description>
</valueHelp>
<constraint>
<validator name="ipv4-address"/>
</constraint>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="static-mapping">
<properties>
<help>Hostname for static mapping reservation</help>
<constraint>
<validator name="fqdn"/>
</constraint>
<constraintErrorMessage>Invalid static mapping hostname</constraintErrorMessage>
</properties>
<children>
#include <include/dhcp/option-v4.xml.i>
#include <include/generic-description.xml.i>
#include <include/generic-disable-node.xml.i>
<leafNode name="ip-address">
<properties>
<help>Fixed IP address of static mapping</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 address used in static mapping</description>
</valueHelp>
<constraint>
<validator name="ipv4-address"/>
</constraint>
</properties>
</leafNode>
#include <include/interface/mac.xml.i>
#include <include/interface/duid.xml.i>
</children>
</tagNode>
<leafNode name="subnet-id">
<properties>
<help>Unique ID mapped to leases in the lease file</help>
<valueHelp>
<format>u32</format>
<description>Unique subnet ID</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-4294967295"/>
</constraint>
</properties>
</leafNode>
</children>
</tagNode>
</children>
</tagNode>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/python/vyos/template.py b/python/vyos/template.py
index bde8e3554..ab06c7752 100644
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -1,962 +1,971 @@
# Copyright 2019-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 functools
import os
from jinja2 import Environment
from jinja2 import FileSystemLoader
from jinja2 import ChainableUndefined
from vyos.defaults import directories
from vyos.utils.dict import dict_search_args
from vyos.utils.file import makedir
from vyos.utils.permission import chmod
from vyos.utils.permission import chown
# Holds template filters registered via register_filter()
_FILTERS = {}
_TESTS = {}
# reuse Environments with identical settings to improve performance
@functools.lru_cache(maxsize=2)
def _get_environment(location=None):
if location is None:
loc_loader=FileSystemLoader(directories["templates"])
else:
loc_loader=FileSystemLoader(location)
env = Environment(
# Don't check if template files were modified upon re-rendering
auto_reload=False,
# Cache up to this number of templates for quick re-rendering
cache_size=100,
loader=loc_loader,
trim_blocks=True,
undefined=ChainableUndefined,
extensions=['jinja2.ext.loopcontrols']
)
env.filters.update(_FILTERS)
env.tests.update(_TESTS)
return env
def register_filter(name, func=None):
"""Register a function to be available as filter in templates under given name.
It can also be used as a decorator, see below in this module for examples.
:raise RuntimeError:
when trying to register a filter after a template has been rendered already
:raise ValueError: when trying to register a name which was taken already
"""
if func is None:
return functools.partial(register_filter, name)
if _get_environment.cache_info().currsize:
raise RuntimeError(
"Filters can only be registered before rendering the first template"
)
if name in _FILTERS:
raise ValueError(f"A filter with name {name!r} was registered already")
_FILTERS[name] = func
return func
def register_test(name, func=None):
"""Register a function to be available as test in templates under given name.
It can also be used as a decorator, see below in this module for examples.
:raise RuntimeError:
when trying to register a test after a template has been rendered already
:raise ValueError: when trying to register a name which was taken already
"""
if func is None:
return functools.partial(register_test, name)
if _get_environment.cache_info().currsize:
raise RuntimeError(
"Tests can only be registered before rendering the first template"
)
if name in _TESTS:
raise ValueError(f"A test with name {name!r} was registered already")
_TESTS[name] = func
return func
def render_to_string(template, content, formater=None, location=None):
"""Render a template from the template directory, raise on any errors.
:param template: the path to the template relative to the template folder
:param content: the dictionary of variables to put into rendering context
:param formater:
if given, it has to be a callable the rendered string is passed through
The parsed template files are cached, so rendering the same file multiple times
does not cause as too much overhead.
If used everywhere, it could be changed to load the template from Python
environment variables from an importable Python module generated when the Debian
package is build (recovering the load time and overhead caused by having the
file out of the code).
"""
template = _get_environment(location).get_template(template)
rendered = template.render(content)
if formater is not None:
rendered = formater(rendered)
return rendered
def render(
destination,
template,
content,
formater=None,
permission=None,
user=None,
group=None,
location=None,
):
"""Render a template from the template directory to a file, raise on any errors.
:param destination: path to the file to save the rendered template in
:param permission: permission bitmask to set for the output file
:param user: user to own the output file
:param group: group to own the output file
All other parameters are as for :func:`render_to_string`.
"""
# Create the directory if it does not exist
folder = os.path.dirname(destination)
makedir(folder, user, group)
# As we are opening the file with 'w', we are performing the rendering before
# calling open() to not accidentally erase the file if rendering fails
rendered = render_to_string(template, content, formater, location)
# Write to file
with open(destination, "w") as file:
chmod(file.fileno(), permission)
chown(file.fileno(), user, group)
file.write(rendered)
##################################
# Custom template filters follow #
##################################
@register_filter('force_to_list')
def force_to_list(value):
""" Convert scalars to single-item lists and leave lists untouched """
if isinstance(value, list):
return value
else:
return [value]
@register_filter('seconds_to_human')
def seconds_to_human(seconds, separator=""):
""" Convert seconds to human-readable values like 1d6h15m23s """
from vyos.utils.convert import seconds_to_human
return seconds_to_human(seconds, separator=separator)
@register_filter('bytes_to_human')
def bytes_to_human(bytes, initial_exponent=0, precision=2):
""" Convert bytes to human-readable values like 1.44M """
from vyos.utils.convert import bytes_to_human
return bytes_to_human(bytes, initial_exponent=initial_exponent, precision=precision)
@register_filter('human_to_bytes')
def human_to_bytes(value):
""" Convert a data amount with a unit suffix to bytes, like 2K to 2048 """
from vyos.utils.convert import human_to_bytes
return human_to_bytes(value)
@register_filter('ip_from_cidr')
def ip_from_cidr(prefix):
""" Take an IPv4/IPv6 CIDR host and strip cidr mask.
Example:
192.0.2.1/24 -> 192.0.2.1, 2001:db8::1/64 -> 2001:db8::1
"""
from ipaddress import ip_interface
return str(ip_interface(prefix).ip)
@register_filter('address_from_cidr')
def address_from_cidr(prefix):
""" Take an IPv4/IPv6 CIDR prefix and convert the network to an "address".
Example:
192.0.2.0/24 -> 192.0.2.0, 2001:db8::/48 -> 2001:db8::
"""
from ipaddress import ip_network
return str(ip_network(prefix).network_address)
@register_filter('bracketize_ipv6')
def bracketize_ipv6(address):
""" Place a passed IPv6 address into [] brackets, do nothing for IPv4 """
if is_ipv6(address):
return f'[{address}]'
return address
@register_filter('dot_colon_to_dash')
def dot_colon_to_dash(text):
""" Replace dot and colon to dash for string
Example:
192.0.2.1 => 192-0-2-1, 2001:db8::1 => 2001-db8--1
"""
text = text.replace(":", "-")
text = text.replace(".", "-")
return text
@register_filter('generate_uuid4')
def generate_uuid4(text):
""" Generate random unique ID
Example:
% uuid4()
UUID('958ddf6a-ef14-4e81-8cfb-afb12456d1c5')
"""
from uuid import uuid4
return uuid4()
@register_filter('netmask_from_cidr')
def netmask_from_cidr(prefix):
""" Take CIDR prefix and convert the prefix length to a "subnet mask".
Example:
- 192.0.2.0/24 -> 255.255.255.0
- 2001:db8::/48 -> ffff:ffff:ffff::
"""
from ipaddress import ip_network
return str(ip_network(prefix).netmask)
@register_filter('netmask_from_ipv4')
def netmask_from_ipv4(address):
""" Take IP address and search all attached interface IP addresses for the
given one. After address has been found, return the associated netmask.
Example:
- 172.18.201.10 -> 255.255.255.128
"""
from netifaces import interfaces
from netifaces import ifaddresses
from netifaces import AF_INET
for interface in interfaces():
tmp = ifaddresses(interface)
if AF_INET in tmp:
for af_addr in tmp[AF_INET]:
if 'addr' in af_addr:
if af_addr['addr'] == address:
return af_addr['netmask']
raise ValueError
@register_filter('is_ip_network')
def is_ip_network(addr):
""" Take IP(v4/v6) address and validate if the passed argument is a network
or a host address.
Example:
- 192.0.2.0 -> False
- 192.0.2.10/24 -> False
- 192.0.2.0/24 -> True
- 2001:db8:: -> False
- 2001:db8::100 -> False
- 2001:db8::/48 -> True
- 2001:db8:1000::/64 -> True
"""
try:
from ipaddress import ip_network
# input variables must contain a / to indicate its CIDR notation
if len(addr.split('/')) != 2:
raise ValueError()
ip_network(addr)
return True
except:
return False
@register_filter('network_from_ipv4')
def network_from_ipv4(address):
""" Take IP address and search all attached interface IP addresses for the
given one. After address has been found, return the associated network
address.
Example:
- 172.18.201.10 has mask 255.255.255.128 -> network is 172.18.201.0
"""
netmask = netmask_from_ipv4(address)
from ipaddress import ip_interface
cidr_prefix = ip_interface(f'{address}/{netmask}').network
return address_from_cidr(cidr_prefix)
@register_filter('is_interface')
def is_interface(interface):
""" Check if parameter is a valid local interface name """
return os.path.exists(f'/sys/class/net/{interface}')
@register_filter('is_ip')
def is_ip(addr):
""" Check addr if it is an IPv4 or IPv6 address """
return is_ipv4(addr) or is_ipv6(addr)
@register_filter('is_ipv4')
def is_ipv4(text):
""" Filter IP address, return True on IPv4 address, False otherwise """
from ipaddress import ip_interface
try: return ip_interface(text).version == 4
except: return False
@register_filter('is_ipv6')
def is_ipv6(text):
""" Filter IP address, return True on IPv6 address, False otherwise """
from ipaddress import ip_interface
try: return ip_interface(text).version == 6
except: return False
@register_filter('first_host_address')
def first_host_address(prefix):
""" Return first usable (host) IP address from given prefix.
Example:
- 10.0.0.0/24 -> 10.0.0.1
- 2001:db8::/64 -> 2001:db8::
"""
from ipaddress import ip_interface
tmp = ip_interface(prefix).network
return str(tmp.network_address +1)
@register_filter('last_host_address')
def last_host_address(text):
""" Return first usable IP address from given prefix.
Example:
- 10.0.0.0/24 -> 10.0.0.254
- 2001:db8::/64 -> 2001:db8::ffff:ffff:ffff:ffff
"""
from ipaddress import ip_interface
from ipaddress import IPv4Network
from ipaddress import IPv6Network
addr = ip_interface(text)
if addr.version == 4:
return str(IPv4Network(addr).broadcast_address - 1)
return str(IPv6Network(addr).broadcast_address)
@register_filter('inc_ip')
def inc_ip(address, increment):
""" Increment given IP address by 'increment'
Example (inc by 2):
- 10.0.0.0/24 -> 10.0.0.2
- 2001:db8::/64 -> 2001:db8::2
"""
from ipaddress import ip_interface
return str(ip_interface(address).ip + int(increment))
@register_filter('dec_ip')
def dec_ip(address, decrement):
""" Decrement given IP address by 'decrement'
Example (inc by 2):
- 10.0.0.0/24 -> 10.0.0.2
- 2001:db8::/64 -> 2001:db8::2
"""
from ipaddress import ip_interface
return str(ip_interface(address).ip - int(decrement))
@register_filter('compare_netmask')
def compare_netmask(netmask1, netmask2):
"""
Compare two IP netmask if they have the exact same size.
compare_netmask('10.0.0.0/8', '20.0.0.0/8') -> True
compare_netmask('10.0.0.0/8', '20.0.0.0/16') -> False
"""
from ipaddress import ip_network
try:
return ip_network(netmask1).netmask == ip_network(netmask2).netmask
except:
return False
@register_filter('isc_static_route')
def isc_static_route(subnet, router):
# https://ercpe.de/blog/pushing-static-routes-with-isc-dhcp-server
# Option format is:
# <netmask>, <network-byte1>, <network-byte2>, <network-byte3>, <router-byte1>, <router-byte2>, <router-byte3>
# where bytes with the value 0 are omitted.
from ipaddress import ip_network
net = ip_network(subnet)
# add netmask
string = str(net.prefixlen) + ','
# add network bytes
if net.prefixlen:
width = net.prefixlen // 8
if net.prefixlen % 8:
width += 1
string += ','.join(map(str,tuple(net.network_address.packed)[:width])) + ','
# add router bytes
string += ','.join(router.split('.'))
return string
@register_filter('is_file')
def is_file(filename):
if os.path.exists(filename):
return os.path.isfile(filename)
return False
@register_filter('get_dhcp_router')
def get_dhcp_router(interface):
""" Static routes can point to a router received by a DHCP reply. This
helper is used to get the current default router from the DHCP reply.
Returns False of no router is found, returns the IP address as string if
a router is found.
"""
lease_file = directories['isc_dhclient_dir'] + f'/dhclient_{interface}.leases'
if not os.path.exists(lease_file):
return None
from vyos.utils.file import read_file
for line in read_file(lease_file).splitlines():
if 'option routers' in line:
(_, _, address) = line.split()
return address.rstrip(';')
@register_filter('natural_sort')
def natural_sort(iterable):
import re
from jinja2.runtime import Undefined
if isinstance(iterable, Undefined) or iterable is None:
return list()
def convert(text):
return int(text) if text.isdigit() else text.lower()
def alphanum_key(key):
return [convert(c) for c in re.split('([0-9]+)', str(key))]
return sorted(iterable, key=alphanum_key)
@register_filter('get_ipv4')
def get_ipv4(interface):
""" Get interface IPv4 addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr_v4()
@register_filter('get_ipv6')
def get_ipv6(interface):
""" Get interface IPv6 addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr_v6()
@register_filter('get_ip')
def get_ip(interface):
""" Get interface IP addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr()
def get_first_ike_dh_group(ike_group):
if ike_group and 'proposal' in ike_group:
for priority, proposal in ike_group['proposal'].items():
if 'dh_group' in proposal:
return 'dh-group' + proposal['dh_group']
return 'dh-group2' # Fallback on dh-group2
@register_filter('get_esp_ike_cipher')
def get_esp_ike_cipher(group_config, ike_group=None):
pfs_lut = {
'dh-group1' : 'modp768',
'dh-group2' : 'modp1024',
'dh-group5' : 'modp1536',
'dh-group14' : 'modp2048',
'dh-group15' : 'modp3072',
'dh-group16' : 'modp4096',
'dh-group17' : 'modp6144',
'dh-group18' : 'modp8192',
'dh-group19' : 'ecp256',
'dh-group20' : 'ecp384',
'dh-group21' : 'ecp521',
'dh-group22' : 'modp1024s160',
'dh-group23' : 'modp2048s224',
'dh-group24' : 'modp2048s256',
'dh-group25' : 'ecp192',
'dh-group26' : 'ecp224',
'dh-group27' : 'ecp224bp',
'dh-group28' : 'ecp256bp',
'dh-group29' : 'ecp384bp',
'dh-group30' : 'ecp512bp',
'dh-group31' : 'curve25519',
'dh-group32' : 'curve448'
}
ciphers = []
if 'proposal' in group_config:
for priority, proposal in group_config['proposal'].items():
# both encryption and hash need to be specified for a proposal
if not {'encryption', 'hash'} <= set(proposal):
continue
tmp = '{encryption}-{hash}'.format(**proposal)
if 'prf' in proposal:
tmp += '-' + proposal['prf']
if 'dh_group' in proposal:
tmp += '-' + pfs_lut[ 'dh-group' + proposal['dh_group'] ]
elif 'pfs' in group_config and group_config['pfs'] != 'disable':
group = group_config['pfs']
if group_config['pfs'] == 'enable':
group = get_first_ike_dh_group(ike_group)
tmp += '-' + pfs_lut[group]
ciphers.append(tmp)
return ciphers
@register_filter('get_uuid')
def get_uuid(interface):
""" Get interface IP addresses"""
from uuid import uuid1
return uuid1()
openvpn_translate = {
'des': 'des-cbc',
'3des': 'des-ede3-cbc',
'bf128': 'bf-cbc',
'bf256': 'bf-cbc',
'aes128gcm': 'aes-128-gcm',
'aes128': 'aes-128-cbc',
'aes192gcm': 'aes-192-gcm',
'aes192': 'aes-192-cbc',
'aes256gcm': 'aes-256-gcm',
'aes256': 'aes-256-cbc'
}
@register_filter('openvpn_cipher')
def get_openvpn_cipher(cipher):
if cipher in openvpn_translate:
return openvpn_translate[cipher].upper()
return cipher.upper()
@register_filter('openvpn_ncp_ciphers')
def get_openvpn_ncp_ciphers(ciphers):
out = []
for cipher in ciphers:
if cipher in openvpn_translate:
out.append(openvpn_translate[cipher])
else:
out.append(cipher)
return ':'.join(out).upper()
@register_filter('snmp_auth_oid')
def snmp_auth_oid(type):
if type not in ['md5', 'sha', 'aes', 'des', 'none']:
raise ValueError()
OIDs = {
'md5' : '.1.3.6.1.6.3.10.1.1.2',
'sha' : '.1.3.6.1.6.3.10.1.1.3',
'aes' : '.1.3.6.1.6.3.10.1.2.4',
'des' : '.1.3.6.1.6.3.10.1.2.2',
'none': '.1.3.6.1.6.3.10.1.2.1'
}
return OIDs[type]
@register_filter('nft_action')
def nft_action(vyos_action):
if vyos_action == 'accept':
return 'return'
return vyos_action
@register_filter('nft_rule')
def nft_rule(rule_conf, fw_hook, fw_name, rule_id, ip_name='ip'):
from vyos.firewall import parse_rule
return parse_rule(rule_conf, fw_hook, fw_name, rule_id, ip_name)
@register_filter('nft_default_rule')
def nft_default_rule(fw_conf, fw_name, family):
output = ['counter']
default_action = fw_conf['default_action']
#family = 'ipv6' if ipv6 else 'ipv4'
if 'default_log' in fw_conf:
action_suffix = default_action[:1].upper()
output.append(f'log prefix "[{family}-{fw_name[:19]}-default-{action_suffix}]"')
#output.append(nft_action(default_action))
output.append(f'{default_action}')
if 'default_jump_target' in fw_conf:
target = fw_conf['default_jump_target']
def_suffix = '6' if family == 'ipv6' else ''
output.append(f'NAME{def_suffix}_{target}')
output.append(f'comment "{fw_name} default-action {default_action}"')
return " ".join(output)
@register_filter('nft_state_policy')
def nft_state_policy(conf, state):
out = [f'ct state {state}']
if 'log' in conf:
log_state = state[:3].upper()
log_action = (conf['action'] if 'action' in conf else 'accept')[:1].upper()
out.append(f'log prefix "[STATE-POLICY-{log_state}-{log_action}]"')
if 'log_level' in conf:
log_level = conf['log_level']
out.append(f'level {log_level}')
out.append('counter')
if 'action' in conf:
out.append(conf['action'])
return " ".join(out)
@register_filter('nft_intra_zone_action')
def nft_intra_zone_action(zone_conf, ipv6=False):
if 'intra_zone_filtering' in zone_conf:
intra_zone = zone_conf['intra_zone_filtering']
fw_name = 'ipv6_name' if ipv6 else 'name'
name_prefix = 'NAME6_' if ipv6 else 'NAME_'
if 'action' in intra_zone:
if intra_zone['action'] == 'accept':
return 'return'
return intra_zone['action']
elif dict_search_args(intra_zone, 'firewall', fw_name):
name = dict_search_args(intra_zone, 'firewall', fw_name)
return f'jump {name_prefix}{name}'
return 'return'
@register_filter('nft_nested_group')
def nft_nested_group(out_list, includes, groups, key):
if not vyos_defined(out_list):
out_list = []
def add_includes(name):
if key in groups[name]:
for item in groups[name][key]:
if item in out_list:
continue
out_list.append(item)
if 'include' in groups[name]:
for name_inc in groups[name]['include']:
add_includes(name_inc)
for name in includes:
add_includes(name)
return out_list
@register_filter('nat_rule')
def nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
from vyos.nat import parse_nat_rule
return parse_nat_rule(rule_conf, rule_id, nat_type, ipv6)
@register_filter('nat_static_rule')
def nat_static_rule(rule_conf, rule_id, nat_type):
from vyos.nat import parse_nat_static_rule
return parse_nat_static_rule(rule_conf, rule_id, nat_type)
@register_filter('conntrack_rule')
def conntrack_rule(rule_conf, rule_id, action, ipv6=False):
ip_prefix = 'ip6' if ipv6 else 'ip'
def_suffix = '6' if ipv6 else ''
output = []
if 'inbound_interface' in rule_conf:
ifname = rule_conf['inbound_interface']
if ifname != 'any':
output.append(f'iifname {ifname}')
if 'protocol' in rule_conf:
if action != 'timeout':
proto = rule_conf['protocol']
else:
for protocol, protocol_config in rule_conf['protocol'].items():
proto = protocol
output.append(f'meta l4proto {proto}')
tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags')
if tcp_flags and action != 'timeout':
from vyos.firewall import parse_tcp_flags
output.append(parse_tcp_flags(tcp_flags))
for side in ['source', 'destination']:
if side in rule_conf:
side_conf = rule_conf[side]
prefix = side[0]
if 'address' in side_conf:
address = side_conf['address']
operator = ''
if address[0] == '!':
operator = '!='
address = address[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} {address}')
if 'port' in side_conf:
port = side_conf['port']
operator = ''
if port[0] == '!':
operator = '!='
port = port[1:]
output.append(f'th {prefix}port {operator} {port}')
if 'group' in side_conf:
group = side_conf['group']
if 'address_group' in group:
group_name = group['address_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @A{def_suffix}_{group_name}')
# Generate firewall group domain-group
elif 'domain_group' in group:
group_name = group['domain_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @D_{group_name}')
elif 'network_group' in group:
group_name = group['network_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @N{def_suffix}_{group_name}')
if 'port_group' in group:
group_name = group['port_group']
if proto == 'tcp_udp':
proto = 'th'
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{proto} {prefix}port {operator} @P_{group_name}')
if action == 'ignore':
output.append('counter notrack')
output.append(f'comment "ignore-{rule_id}"')
else:
output.append(f'counter ct timeout set ct-timeout-{rule_id}')
output.append(f'comment "timeout-{rule_id}"')
return " ".join(output)
@register_filter('conntrack_ct_policy')
def conntrack_ct_policy(protocol_conf):
output = []
for item in protocol_conf:
item_value = protocol_conf[item]
output.append(f'{item}: {item_value}')
return ", ".join(output)
@register_filter('range_to_regex')
def range_to_regex(num_range):
"""Convert range of numbers or list of ranges
to regex
% range_to_regex('11-12')
'(1[1-2])'
% range_to_regex(['11-12', '14-15'])
'(1[1-2]|1[4-5])'
"""
from vyos.range_regex import range_to_regex
if isinstance(num_range, list):
data = []
for entry in num_range:
if '-' not in entry:
data.append(entry)
else:
data.append(range_to_regex(entry))
return f'({"|".join(data)})'
if '-' not in num_range:
return num_range
regex = range_to_regex(num_range)
return f'({regex})'
@register_filter('kea_address_json')
def kea_address_json(addresses):
from json import dumps
from vyos.utils.network import is_addr_assigned
out = []
for address in addresses:
ifname = is_addr_assigned(address, return_ifname=True)
if not ifname:
continue
out.append(f'{ifname}/{address}')
return dumps(out)
@register_filter('kea_high_availability_json')
def kea_high_availability_json(config):
from json import dumps
source_addr = config['source_address']
remote_addr = config['remote']
+ ha_mode = 'hot-standby' if config['mode'] == 'active-passive' else 'load-balancing'
+ ha_role = config['status']
+
+ if ha_role == 'primary':
+ peer1_role = 'primary'
+ peer2_role = 'standby' if ha_mode == 'hot-standby' else 'secondary'
+ else:
+ peer1_role = 'standby' if ha_mode == 'hot-standby' else 'secondary'
+ peer2_role = 'primary'
data = {
'this-server-name': os.uname()[1],
- 'mode': 'hot-standby',
+ 'mode': ha_mode,
'heartbeat-delay': 10000,
'max-response-delay': 10000,
'max-ack-delay': 5000,
'max-unacked-clients': 0,
'peers': [
{
'name': os.uname()[1],
'url': f'http://{source_addr}:647/',
- 'role': 'standby' if config['status'] == 'secondary' else 'primary',
+ 'role': peer1_role,
'auto-failover': True
},
{
'name': config['name'],
'url': f'http://{remote_addr}:647/',
- 'role': 'primary' if config['status'] == 'secondary' else 'standby',
+ 'role': peer2_role,
'auto-failover': True
}]
}
if 'ca_cert_file' in config:
data['trust-anchor'] = config['ca_cert_file']
if 'cert_file' in config:
data['cert-file'] = config['cert_file']
if 'cert_key_file' in config:
data['key-file'] = config['cert_key_file']
return dumps(data)
@register_filter('kea_shared_network_json')
def kea_shared_network_json(shared_networks):
from vyos.kea import kea_parse_options
from vyos.kea import kea_parse_subnet
from json import dumps
out = []
for name, config in shared_networks.items():
if 'disable' in config:
continue
network = {
'name': name,
'authoritative': ('authoritative' in config),
'subnet4': []
}
if 'option' in config:
network['option-data'] = kea_parse_options(config['option'])
if 'bootfile_name' in config['option']:
network['boot-file-name'] = config['option']['bootfile_name']
if 'bootfile_server' in config['option']:
network['next-server'] = config['option']['bootfile_server']
if 'subnet' in config:
for subnet, subnet_config in config['subnet'].items():
if 'disable' in subnet_config:
continue
network['subnet4'].append(kea_parse_subnet(subnet, subnet_config))
out.append(network)
return dumps(out, indent=4)
@register_filter('kea6_shared_network_json')
def kea6_shared_network_json(shared_networks):
from vyos.kea import kea6_parse_options
from vyos.kea import kea6_parse_subnet
from json import dumps
out = []
for name, config in shared_networks.items():
if 'disable' in config:
continue
network = {
'name': name,
'subnet6': []
}
if 'common_options' in config:
network['option-data'] = kea6_parse_options(config['common_options'])
if 'interface' in config:
network['interface'] = config['interface']
if 'subnet' in config:
for subnet, subnet_config in config['subnet'].items():
network['subnet6'].append(kea6_parse_subnet(subnet, subnet_config))
out.append(network)
return dumps(out, indent=4)
@register_test('vyos_defined')
def vyos_defined(value, test_value=None, var_type=None):
"""
Jinja2 plugin to test if a variable is defined and not none - vyos_defined
will test value if defined and is not none and return true or false.
If test_value is supplied, the value must also pass == test_value to return true.
If var_type is supplied, the value must also be of the specified class/type
Examples:
1. Test if var is defined and not none:
{% if foo is vyos_defined %}
...
{% endif %}
2. Test if variable is defined, not none and has value "something"
{% if bar is vyos_defined("something") %}
...
{% endif %}
Parameters
----------
value : any
Value to test from ansible
test_value : any, optional
Value to test in addition of defined and not none, by default None
var_type : ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'], optional
Type or Class to test for
Returns
-------
boolean
True if variable matches criteria, False in other cases.
Implementation inspired and re-used from https://github.com/aristanetworks/ansible-avd/
"""
from jinja2 import Undefined
if isinstance(value, Undefined) or value is None:
# Invalid value - return false
return False
elif test_value is not None and value != test_value:
# Valid value but not matching the optional argument
return False
elif str(var_type).lower() in ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'] and str(var_type).lower() != type(value).__name__:
# Invalid class - return false
return False
else:
# Valid value and is matching optional argument if provided - return true
return True
diff --git a/smoketest/scripts/cli/test_service_dhcp-server.py b/smoketest/scripts/cli/test_service_dhcp-server.py
index 24bd14af2..69260e412 100755
--- a/smoketest/scripts/cli/test_service_dhcp-server.py
+++ b/smoketest/scripts/cli/test_service_dhcp-server.py
@@ -1,742 +1,805 @@
#!/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 os
import unittest
from json import loads
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 inc_ip
from vyos.template import dec_ip
PROCESS_NAME = 'kea-dhcp4'
CTRL_PROCESS_NAME = 'kea-ctrl-agent'
KEA4_CONF = '/run/kea/kea-dhcp4.conf'
KEA4_CTRL = '/run/kea/dhcp4-ctrl-socket'
base_path = ['service', 'dhcp-server']
interface = 'dum8765'
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()
# Clear out current configuration to allow running this test on a live system
cls.cli_delete(cls, base_path)
cidr_mask = subnet.split('/')[-1]
cls.cli_set(cls, ['interfaces', 'dummy', interface, 'address', f'{router}/{cidr_mask}'])
@classmethod
def tearDownClass(cls):
cls.cli_delete(cls, ['interfaces', 'dummy', interface])
super(TestServiceDHCPServer, cls).tearDownClass()
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
def walk_path(self, obj, path):
current = obj
for i, key in enumerate(path):
if isinstance(key, str):
self.assertTrue(isinstance(current, dict), msg=f'Failed path: {path}')
self.assertTrue(key in current, msg=f'Failed path: {path}')
elif isinstance(key, int):
self.assertTrue(isinstance(current, list), msg=f'Failed path: {path}')
self.assertTrue(0 <= key < len(current), msg=f'Failed path: {path}')
else:
assert False, "Invalid type"
current = current[key]
return current
def verify_config_object(self, obj, path, value):
base_obj = self.walk_path(obj, path)
self.assertTrue(isinstance(base_obj, list))
self.assertTrue(any(True for v in base_obj if v == value))
def verify_config_value(self, obj, path, key, value):
base_obj = self.walk_path(obj, path)
if isinstance(base_obj, list):
self.assertTrue(any(True for v in base_obj if key in v and v[key] == value))
elif isinstance(base_obj, dict):
self.assertTrue(key in base_obj)
self.assertEqual(base_obj[key], value)
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 + ['listen-interface', interface])
pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
self.cli_set(pool + ['subnet-id', '1'])
self.cli_set(pool + ['ignore-client-id'])
# we use the first subnet IP address as default gateway
self.cli_set(pool + ['option', 'default-router', router])
self.cli_set(pool + ['option', 'name-server', dns_1])
self.cli_set(pool + ['option', 'name-server', dns_2])
self.cli_set(pool + ['option', 'domain-name', domain_name])
# 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(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'interfaces-config'], 'interfaces', [interface])
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'id', 1)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'match-client-id', False)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'valid-lifetime', 86400)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'max-valid-lifetime', 86400)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name', 'data': domain_name})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': f'{dns_1}, {dns_2}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
# Verify pools
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_1_start} - {range_1_stop}'})
# 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]
self.cli_set(pool + ['subnet-id', '1'])
# we use the first subnet IP address as default gateway
self.cli_set(pool + ['option', 'default-router', router])
self.cli_set(pool + ['option', 'name-server', dns_1])
self.cli_set(pool + ['option', 'name-server', dns_2])
self.cli_set(pool + ['option', 'domain-name', domain_name])
self.cli_set(pool + ['option', 'ip-forwarding'])
self.cli_set(pool + ['option', 'smtp-server', smtp_server])
self.cli_set(pool + ['option', 'pop-server', smtp_server])
self.cli_set(pool + ['option', 'time-server', time_server])
self.cli_set(pool + ['option', 'tftp-server-name', tftp_server])
for search in search_domains:
self.cli_set(pool + ['option', 'domain-search', search])
self.cli_set(pool + ['option', 'bootfile-name', bootfile_name])
self.cli_set(pool + ['option', 'bootfile-server', bootfile_server])
self.cli_set(pool + ['option', 'wpad-url', wpad])
self.cli_set(pool + ['option', 'server-identifier', server_identifier])
self.cli_set(pool + ['option', 'static-route', '10.0.0.0/24', 'next-hop', '192.0.2.1'])
self.cli_set(pool + ['option', 'ipv6-only-preferred', ipv6_only_preferred])
self.cli_set(pool + ['option', 'time-zone', 'Europe/London'])
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(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'boot-file-name', bootfile_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'next-server', bootfile_server)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'valid-lifetime', 86400)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'max-valid-lifetime', 86400)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name', 'data': domain_name})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': f'{dns_1}, {dns_2}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-search', 'data': ', '.join(search_domains)})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'pop-server', 'data': smtp_server})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'smtp-server', 'data': smtp_server})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'time-servers', 'data': time_server})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'dhcp-server-identifier', 'data': server_identifier})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'tftp-server-name', 'data': tftp_server})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'wpad-url', 'data': wpad})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'rfc3442-static-route', 'data': '24,10,0,0,192,0,2,1, 0,192,0,2,1'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'windows-static-route', 'data': '24,10,0,0,192,0,2,1'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'v6-only-preferred', 'data': ipv6_only_preferred})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'ip-forwarding', 'data': "true"})
# Time zone
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'pcode', 'data': 'GMT0BST,M3.5.0/1,M10.5.0'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'tcode', 'data': 'Europe/London'})
# Verify pools
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
# Check for running process
self.assertTrue(process_named_running(PROCESS_NAME))
def test_dhcp_single_pool_options_scoped(self):
shared_net_name = 'SMOKE-2'
range_0_start = inc_ip(subnet, 10)
range_0_stop = inc_ip(subnet, 20)
range_router = inc_ip(subnet, 5)
range_dns_1 = inc_ip(subnet, 6)
range_dns_2 = inc_ip(subnet, 7)
shared_network = base_path + ['shared-network-name', shared_net_name]
pool = shared_network + ['subnet', subnet]
self.cli_set(pool + ['subnet-id', '1'])
# we use the first subnet IP address as default gateway
self.cli_set(shared_network + ['option', 'default-router', router])
self.cli_set(shared_network + ['option', 'name-server', dns_1])
self.cli_set(shared_network + ['option', 'name-server', dns_2])
self.cli_set(shared_network + ['option', 'domain-name', domain_name])
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', '0', 'option', 'default-router', range_router])
self.cli_set(pool + ['range', '0', 'option', 'name-server', range_dns_1])
self.cli_set(pool + ['range', '0', 'option', 'name-server', range_dns_2])
# commit changes
self.cli_commit()
config = read_file(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'valid-lifetime', 86400)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'max-valid-lifetime', 86400)
# Verify shared-network options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'option-data'],
{'name': 'domain-name', 'data': domain_name})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': f'{dns_1}, {dns_2}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'option-data'],
{'name': 'routers', 'data': router})
# Verify range options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': f'{range_dns_1}, {range_dns_2}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools', 0, 'option-data'],
{'name': 'routers', 'data': range_router})
# Verify pool
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'], 'pool', f'{range_0_start} - {range_0_stop}')
# 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]
self.cli_set(pool + ['subnet-id', '1'])
# we use the first subnet IP address as default gateway
self.cli_set(pool + ['option', 'default-router', router])
self.cli_set(pool + ['option', 'name-server', dns_1])
self.cli_set(pool + ['option', 'name-server', dns_2])
self.cli_set(pool + ['option', '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', mac])
self.cli_set(pool + ['static-mapping', client, 'ip-address', inc_ip(subnet, client_base)])
client_base += 1
# cannot have both mac-address and duid set
with self.assertRaises(ConfigSessionError):
self.cli_set(pool + ['static-mapping', 'client1', 'duid', '00:01:00:01:12:34:56:78:aa:bb:cc:dd:ee:11'])
self.cli_commit()
self.cli_delete(pool + ['static-mapping', 'client1', 'duid'])
# cannot have mappings with duplicate IP addresses
self.cli_set(pool + ['static-mapping', 'dupe1', 'mac', '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()
# Should allow disabled duplicate
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', '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'])
# cannot have mappings with duplicate MAC addresses
self.cli_set(pool + ['static-mapping', 'dupe3', 'duid', '00:01:02:03:04:05:06:07:aa:aa:aa:aa:aa:01'])
self.cli_set(pool + ['static-mapping', 'dupe3', 'ip-address', inc_ip(subnet, 121)])
self.cli_set(pool + ['static-mapping', 'dupe4', 'duid', '00:01:02:03:04:05:06:07:aa:aa:aa:aa:aa:01'])
self.cli_set(pool + ['static-mapping', 'dupe4', 'ip-address', inc_ip(subnet, 121)])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(pool + ['static-mapping', 'dupe3'])
self.cli_delete(pool + ['static-mapping', 'dupe4'])
# commit changes
self.cli_commit()
config = read_file(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'id', 1)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'valid-lifetime', 86400)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'max-valid-lifetime', 86400)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name', 'data': domain_name})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': f'{dns_1}, {dns_2}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
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.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'reservations'],
{'hostname': client, 'hw-address': mac, 'ip-address': ip})
client_base += 1
# 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]
self.cli_set(pool + ['subnet-id', str(int(network) + 1)])
# we use the first subnet IP address as default gateway
self.cli_set(pool + ['option', 'default-router', router])
self.cli_set(pool + ['option', 'name-server', dns_1])
self.cli_set(pool + ['option', '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', 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(KEA4_CONF)
obj = loads(config)
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)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', int(network), 'subnet4'], 'subnet', subnet)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', int(network), 'subnet4'], 'id', int(network) + 1)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', int(network), 'subnet4'], 'valid-lifetime', int(lease_time))
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', int(network), 'subnet4'], 'max-valid-lifetime', int(lease_time))
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'option-data'],
{'name': 'domain-name', 'data': domain_name})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'option-data'],
{'name': 'domain-name-servers', 'data': dns_1})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'pools'],
{'pool': f'{range_1_start} - {range_1_stop}'})
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.verify_config_object(
obj,
['Dhcp4', 'shared-networks', int(network), 'subnet4', 0, 'reservations'],
{'hostname': client, 'hw-address': mac, 'ip-address': ip})
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 + ['subnet-id', '1'])
self.cli_set(pool + ['option', '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()
config = read_file(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', 'EXCLUDE-TEST')
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
# Verify pools
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
# 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 + ['subnet-id', '1'])
self.cli_set(pool + ['option', '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()
config = read_file(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', 'EXCLUDE-TEST-2')
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop_excl}'})
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start_excl} - {range_0_stop}'})
# 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 + ['subnet-id', '1'])
self.cli_set(pool + ['option', '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(KEA4_CONF)
obj = loads(config)
self.verify_config_value(obj, ['Dhcp4', 'interfaces-config'], 'interfaces', [f'{interface}/{router}'])
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', 'RELAY')
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', relay_subnet)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': relay_router})
# Verify pools
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
# Check for running process
self.assertTrue(process_named_running(PROCESS_NAME))
def test_dhcp_high_availability(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]
self.cli_set(pool + ['subnet-id', '1'])
# we use the first subnet IP address as default gateway
self.cli_set(pool + ['option', '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 + ['high-availability', 'source-address', failover_local])
self.cli_set(base_path + ['high-availability', 'name', failover_name])
self.cli_set(base_path + ['high-availability', 'remote', failover_remote])
self.cli_set(base_path + ['high-availability', 'status', 'primary'])
+ ## No mode defined -> its active-active mode by default
# commit changes
self.cli_commit()
config = read_file(KEA4_CONF)
obj = loads(config)
# Verify failover
self.verify_config_value(obj, ['Dhcp4', 'control-socket'], 'socket-name', KEA4_CTRL)
self.verify_config_object(
obj,
['Dhcp4', 'hooks-libraries', 0, 'parameters', 'high-availability', 0, 'peers'],
{'name': os.uname()[1], 'url': f'http://{failover_local}:647/', 'role': 'primary', 'auto-failover': True})
self.verify_config_object(
obj,
['Dhcp4', 'hooks-libraries', 0, 'parameters', 'high-availability', 0, 'peers'],
- {'name': failover_name, 'url': f'http://{failover_remote}:647/', 'role': 'standby', 'auto-failover': True})
+ {'name': failover_name, 'url': f'http://{failover_remote}:647/', 'role': 'secondary', 'auto-failover': True})
+
+ self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
+ self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
+
+ # Verify options
+ self.verify_config_object(
+ obj,
+ ['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
+ {'name': 'routers', 'data': router})
+
+ # Verify pools
+ self.verify_config_object(
+ obj,
+ ['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
+ {'pool': f'{range_0_start} - {range_0_stop}'})
+
+ # Check for running process
+ self.assertTrue(process_named_running(PROCESS_NAME))
+ self.assertTrue(process_named_running(CTRL_PROCESS_NAME))
+
+ def test_dhcp_high_availability_standby(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]
+ self.cli_set(pool + ['subnet-id', '1'])
+ # we use the first subnet IP address as default gateway
+ self.cli_set(pool + ['option', 'default-router', router])
+ 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 + ['high-availability', 'source-address', failover_local])
+ self.cli_set(base_path + ['high-availability', 'name', failover_name])
+ self.cli_set(base_path + ['high-availability', 'remote', failover_remote])
+ self.cli_set(base_path + ['high-availability', 'status', 'secondary'])
+ self.cli_set(base_path + ['high-availability', 'mode', 'active-passive'])
+
+ # commit changes
+ self.cli_commit()
+
+ config = read_file(KEA4_CONF)
+ obj = loads(config)
+
+ # Verify failover
+ self.verify_config_value(obj, ['Dhcp4', 'control-socket'], 'socket-name', KEA4_CTRL)
+
+ self.verify_config_object(
+ obj,
+ ['Dhcp4', 'hooks-libraries', 0, 'parameters', 'high-availability', 0, 'peers'],
+ {'name': os.uname()[1], 'url': f'http://{failover_local}:647/', 'role': 'standby', 'auto-failover': True})
+
+ self.verify_config_object(
+ obj,
+ ['Dhcp4', 'hooks-libraries', 0, 'parameters', 'high-availability', 0, 'peers'],
+ {'name': failover_name, 'url': f'http://{failover_remote}:647/', 'role': 'primary', 'auto-failover': True})
self.verify_config_value(obj, ['Dhcp4', 'shared-networks'], 'name', shared_net_name)
self.verify_config_value(obj, ['Dhcp4', 'shared-networks', 0, 'subnet4'], 'subnet', subnet)
# Verify options
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'option-data'],
{'name': 'routers', 'data': router})
# Verify pools
self.verify_config_object(
obj,
['Dhcp4', 'shared-networks', 0, 'subnet4', 0, 'pools'],
{'pool': f'{range_0_start} - {range_0_stop}'})
# Check for running process
self.assertTrue(process_named_running(PROCESS_NAME))
self.assertTrue(process_named_running(CTRL_PROCESS_NAME))
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 ba3d69b07..518625ffc 100755
--- a/src/conf_mode/service_dhcp-server.py
+++ b/src/conf_mode/service_dhcp-server.py
@@ -1,420 +1,424 @@
#!/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 IPRange
from sys import exit
from vyos.config import Config
from vyos.pki import wrap_certificate
from vyos.pki import wrap_private_key
from vyos.template import render
from vyos.utils.dict import dict_search
from vyos.utils.dict import dict_search_args
from vyos.utils.file import chmod_775
from vyos.utils.file import makedir
from vyos.utils.file import write_file
from vyos.utils.process import call
from vyos.utils.network import interface_exists
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()
ctrl_config_file = '/run/kea/kea-ctrl-agent.conf'
ctrl_socket = '/run/kea/dhcp4-ctrl-socket'
config_file = '/run/kea/kea-dhcp4.conf'
lease_file = '/config/dhcp/dhcp4-leases.csv'
systemd_override = r'/run/systemd/system/kea-ctrl-agent.service.d/10-override.conf'
user_group = '_kea'
ca_cert_file = '/run/kea/kea-failover-ca.pem'
cert_file = '/run/kea/kea-failover.pem'
cert_key_file = '/run/kea/kea-failover-key.pem'
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})
- if dict_search('high_availability.certificate', dhcp):
- dhcp['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True)
+ if len(dhcp['high_availability']) == 1:
+ ## only default value for mode is set, need to remove ha node
+ del dhcp['high_availability']
+ else:
+ if dict_search('high_availability.certificate', dhcp):
+ dhcp['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True)
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
subnet_ids = []
# A shared-network requires a subnet definition
for network, network_config in dhcp['shared_network_name'].items():
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_id' not in subnet_config:
raise ConfigError(f'Unique subnet ID not specified for subnet "{subnet}"')
if subnet_config['subnet_id'] in subnet_ids:
raise ConfigError(f'Subnet ID for subnet "{subnet}" is not unique')
subnet_ids.append(subnet_config['subnet_id'])
# 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!')
# 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 = []
used_duid = []
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' not in mapping_config and 'duid' not in mapping_config) or \
('mac' in mapping_config and 'duid' in mapping_config):
raise ConfigError(f'Either MAC address or Client identifier (DUID) is required for '
f'static mapping "{mapping}" 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 'disable' not in mapping_config:
if 'mac' in mapping_config:
if mapping_config['mac'] 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'])
if 'duid' in mapping_config:
if mapping_config['duid'] in used_duid:
raise ConfigError(f'Configured DUID for static mapping "{mapping}" already exists on another static mapping')
used_duid.append(mapping_config['duid'])
# 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 'high_availability' in dhcp:
for key in ['name', 'remote', 'source_address', 'status']:
if key not in dhcp['high_availability']:
tmp = key.replace('_', '-')
raise ConfigError(f'DHCP high-availability requires "{tmp}" to be specified!')
if len({'certificate', 'ca_certificate'} & set(dhcp['high_availability'])) == 1:
raise ConfigError(f'DHCP secured high-availability requires both certificate and CA certificate')
if 'certificate' in dhcp['high_availability']:
cert_name = dhcp['high_availability']['certificate']
if cert_name not in dhcp['pki']['certificate']:
raise ConfigError(f'Invalid certificate specified for DHCP high-availability')
if not dict_search_args(dhcp['pki']['certificate'], cert_name, 'certificate'):
raise ConfigError(f'Invalid certificate specified for DHCP high-availability')
if not dict_search_args(dhcp['pki']['certificate'], cert_name, 'private', 'key'):
raise ConfigError(f'Missing private key on certificate specified for DHCP high-availability')
if 'ca_certificate' in dhcp['high_availability']:
ca_cert_name = dhcp['high_availability']['ca_certificate']
if ca_cert_name not in dhcp['pki']['ca']:
raise ConfigError(f'Invalid CA certificate specified for DHCP high-availability')
if not dict_search_args(dhcp['pki']['ca'], ca_cert_name, 'certificate'):
raise ConfigError(f'Invalid CA certificate specified for DHCP high-availability')
for address in (dict_search('listen_address', dhcp) or []):
if is_addr_assigned(address):
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!')
if 'listen_address' in dhcp and 'listen_interface' in dhcp:
raise ConfigError(f'Cannot define listen-address and listen-interface at the same time')
for interface in (dict_search('listen_interface', dhcp) or []):
if not interface_exists(interface):
raise ConfigError(f'listen-interface "{interface}" does not exist')
return None
def generate(dhcp):
# bail out early - looks like removal from running config
if not dhcp or 'disable' in dhcp:
return None
dhcp['lease_file'] = lease_file
dhcp['machine'] = os.uname().machine
# Create directory for lease file if necessary
lease_dir = os.path.dirname(lease_file)
if not os.path.isdir(lease_dir):
makedir(lease_dir, group='vyattacfg')
chmod_775(lease_dir)
# Create lease file if necessary and let kea own it - 'kea-lfc' expects it that way
if not os.path.exists(lease_file):
write_file(lease_file, '', user=user_group, group=user_group, mode=0o644)
for f in [cert_file, cert_key_file, ca_cert_file]:
if os.path.exists(f):
os.unlink(f)
if 'high_availability' in dhcp:
if 'certificate' in dhcp['high_availability']:
cert_name = dhcp['high_availability']['certificate']
cert_data = dhcp['pki']['certificate'][cert_name]['certificate']
key_data = dhcp['pki']['certificate'][cert_name]['private']['key']
write_file(cert_file, wrap_certificate(cert_data), user=user_group, mode=0o600)
write_file(cert_key_file, wrap_private_key(key_data), user=user_group, mode=0o600)
dhcp['high_availability']['cert_file'] = cert_file
dhcp['high_availability']['cert_key_file'] = cert_key_file
if 'ca_certificate' in dhcp['high_availability']:
ca_cert_name = dhcp['high_availability']['ca_certificate']
ca_cert_data = dhcp['pki']['ca'][ca_cert_name]['certificate']
write_file(ca_cert_file, wrap_certificate(ca_cert_data), user=user_group, mode=0o600)
dhcp['high_availability']['ca_cert_file'] = ca_cert_file
render(systemd_override, 'dhcp-server/10-override.conf.j2', dhcp)
render(ctrl_config_file, 'dhcp-server/kea-ctrl-agent.conf.j2', dhcp, user=user_group, group=user_group)
render(config_file, 'dhcp-server/kea-dhcp4.conf.j2', dhcp, user=user_group, group=user_group)
return None
def apply(dhcp):
services = ['kea-ctrl-agent', 'kea-dhcp4-server', 'kea-dhcp-ddns-server']
if not dhcp or 'disable' in dhcp:
for service in services:
call(f'systemctl stop {service}.service')
if os.path.exists(config_file):
os.unlink(config_file)
return None
for service in services:
action = 'restart'
if service == 'kea-dhcp-ddns-server' and 'dynamic_dns_update' not in dhcp:
action = 'stop'
if service == 'kea-ctrl-agent' and 'high_availability' not in dhcp:
action = 'stop'
call(f'systemctl {action} {service}.service')
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)