diff --git a/smoketest/scripts/cli/test_service_dns_dynamic.py b/smoketest/scripts/cli/test_service_dns_dynamic.py
index 044d053b4..11d411cb4 100755
--- a/smoketest/scripts/cli/test_service_dns_dynamic.py
+++ b/smoketest/scripts/cli/test_service_dns_dynamic.py
@@ -1,175 +1,183 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2019-2020 VyOS maintainers and contributors
+# Copyright (C) 2019-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import re
import os
import unittest
import tempfile
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.util import cmd
from vyos.util import process_running
DDCLIENT_CONF = '/run/ddclient/ddclient.conf'
DDCLIENT_PID = '/run/ddclient/ddclient.pid'
base_path = ['service', 'dns', 'dynamic']
hostname = 'test.ddns.vyos.io'
+zone = 'vyos.io'
+password = 'paSS_@4ord'
interface = 'eth0'
+
def get_config_value(key):
tmp = cmd(f'sudo cat {DDCLIENT_CONF}')
- tmp = re.findall(r'\n?{}=+(.*)'.format(key), tmp)
- tmp = tmp[0].rstrip(', \\')
- return tmp
+ vals = re.findall(r'\n?{}=([.-@_A-Za-z0-9]+),? \\'.format(key), tmp)
+ return vals[0] if vals else ''
+
class TestServiceDDNS(VyOSUnitTestSHIM.TestCase):
def tearDown(self):
# Check for running process
self.assertTrue(process_running(DDCLIENT_PID))
# Delete DDNS configuration
self.cli_delete(base_path)
self.cli_commit()
# PID file must no londer exist after process exited
self.assertFalse(os.path.exists(DDCLIENT_PID))
- def test_dyndns_service(self):
+ # IPv4 standard DDNS service configuration
+ def test_dyndns_service_standard(self):
ddns = ['address', interface, 'service']
- services = {'cloudflare': {'protocol': 'cloudflare', 'zone': 'vyos.io'},
+ services = {'cloudflare': {'protocol': 'cloudflare'},
'freedns': {'protocol': 'freedns', 'username': 'vyos_user'},
'zoneedit': {'protocol': 'zoneedit1', 'username': 'vyos_user'}}
- password = 'vyos_pass'
- zone = 'vyos.io'
for svc, details in services.items():
self.cli_delete(base_path)
self.cli_set(base_path + ddns + [svc, 'host-name', hostname])
for opt, value in details.items():
self.cli_set(base_path + ddns + [svc, opt, value])
self.cli_set(base_path + ddns + [svc, 'password', password])
self.cli_set(base_path + ddns + [svc, 'zone', zone])
# commit changes
if details['protocol'] == 'cloudflare':
self.cli_commit()
else:
# zone option does not work on all protocols, an exception is
# raised for all others
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ddns + [svc, 'zone', zone])
# commit changes again - now it should work
self.cli_commit()
+ # Check the generating config parameters
+ self.assertEqual(get_config_value('use'), 'if')
+ self.assertEqual(get_config_value('if'), interface)
+ self.assertEqual(get_config_value('password'), password)
+
for opt in details.keys():
if opt == 'username':
- self.assertTrue(get_config_value('login') == details[opt])
+ self.assertEqual(get_config_value('login'), details[opt])
else:
- self.assertTrue(get_config_value(opt) == details[opt])
-
- self.assertTrue(get_config_value('use') == 'if')
- self.assertTrue(get_config_value('if') == interface)
+ self.assertEqual(get_config_value(opt), details[opt])
- def test_dyndns_rfc2136(self):
- # Check if DDNS service can be configured and runs
- ddns = ['address', interface, 'rfc2136', 'vyos']
- srv = 'ns1.vyos.io'
- zone = 'vyos.io'
- ttl = '300'
-
- with tempfile.NamedTemporaryFile(prefix='/config/auth/') as key_file:
- key_file.write(b'S3cretKey')
+ # IPv6 only DDNS service configuration
+ def test_dyndns_service_ipv6(self):
+ ddns = ['address', interface, 'service', 'dynv6']
+ proto = 'dyndns2'
+ user = 'none'
+ password = 'paSS_4ord'
+ srv = 'ddns.vyos.io'
+ ip_version = 'ipv6'
- self.cli_set(base_path + ddns + ['key', key_file.name])
- self.cli_set(base_path + ddns + ['host-name', hostname])
- self.cli_set(base_path + ddns + ['server', srv])
- self.cli_set(base_path + ddns + ['ttl', ttl])
- self.cli_set(base_path + ddns + ['zone', zone])
+ self.cli_set(base_path + ddns + ['ip-version', ip_version])
+ self.cli_set(base_path + ddns + ['protocol', proto])
+ self.cli_set(base_path + ddns + ['server', srv])
+ self.cli_set(base_path + ddns + ['username', user])
+ self.cli_set(base_path + ddns + ['password', password])
+ self.cli_set(base_path + ddns + ['host-name', hostname])
- # commit changes
- self.cli_commit()
+ # commit changes
+ self.cli_commit()
- # Check some generating config parameters
- self.assertEqual(get_config_value('protocol'), 'nsupdate')
- self.assertTrue(get_config_value('password') == key_file.name)
- self.assertTrue(get_config_value('server') == srv)
- self.assertTrue(get_config_value('zone') == zone)
- self.assertTrue(get_config_value('ttl') == ttl)
- self.assertEqual(get_config_value('use'), 'if')
- self.assertEqual(get_config_value('if'), interface)
+ # Check the generating config parameters
+ self.assertEqual(get_config_value('usev6'), 'ifv6')
+ self.assertEqual(get_config_value('ifv6'), interface)
+ self.assertEqual(get_config_value('protocol'), proto)
+ self.assertEqual(get_config_value('server'), srv)
+ self.assertEqual(get_config_value('login'), user)
+ self.assertEqual(get_config_value('password'), password)
- def test_dyndns_dual(self):
+ # IPv4+IPv6 dual DDNS service configuration
+ def test_dyndns_service_dual_stack(self):
ddns = ['address', interface, 'service']
services = {'cloudflare': {'protocol': 'cloudflare', 'zone': 'vyos.io'},
'freedns': {'protocol': 'freedns', 'username': 'vyos_user'}}
password = 'vyos_pass'
ip_version = 'both'
for svc, details in services.items():
self.cli_delete(base_path)
self.cli_set(base_path + ddns + [svc, 'host-name', hostname])
for opt, value in details.items():
self.cli_set(base_path + ddns + [svc, opt, value])
self.cli_set(base_path + ddns + [svc, 'password', password])
self.cli_set(base_path + ddns + [svc, 'ip-version', ip_version])
# commit changes
self.cli_commit()
- # Check some generating config parameters
+ # Check the generating config parameters
+ self.assertEqual(get_config_value('usev4'), 'ifv4')
+ self.assertEqual(get_config_value('usev6'), 'ifv6')
+ self.assertEqual(get_config_value('ifv4'), interface)
+ self.assertEqual(get_config_value('ifv6'), interface)
+ self.assertEqual(get_config_value('password'), password)
+
for opt in details.keys():
if opt == 'username':
- self.assertTrue(get_config_value('login') == details[opt])
+ self.assertEqual(get_config_value('login'), details[opt])
else:
- self.assertTrue(get_config_value(opt) == details[opt])
+ self.assertEqual(get_config_value(opt), details[opt])
- self.assertTrue(get_config_value('usev4') == 'ifv4')
- self.assertTrue(get_config_value('usev6') == 'ifv6')
- self.assertTrue(get_config_value('ifv4') == interface)
- self.assertTrue(get_config_value('ifv6') == interface)
+ def test_dyndns_rfc2136(self):
+ # Check if DDNS service can be configured and runs
+ ddns = ['address', interface, 'rfc2136', 'vyos']
+ srv = 'ns1.vyos.io'
+ zone = 'vyos.io'
+ ttl = '300'
- def test_dyndns_ipv6(self):
- ddns = ['address', interface, 'service', 'dynv6']
- proto = 'dyndns2'
- user = 'none'
- password = 'paSS_4ord'
- srv = 'ddns.vyos.io'
- ip_version = 'ipv6'
+ with tempfile.NamedTemporaryFile(prefix='/config/auth/') as key_file:
+ key_file.write(b'S3cretKey')
- self.cli_set(base_path + ddns + ['host-name', hostname])
- self.cli_set(base_path + ddns + ['username', user])
- self.cli_set(base_path + ddns + ['password', password])
- self.cli_set(base_path + ddns + ['protocol', proto])
- self.cli_set(base_path + ddns + ['server', srv])
- self.cli_set(base_path + ddns + ['ip-version', ip_version])
+ self.cli_set(base_path + ddns + ['server', srv])
+ self.cli_set(base_path + ddns + ['zone', zone])
+ self.cli_set(base_path + ddns + ['key', key_file.name])
+ self.cli_set(base_path + ddns + ['ttl', ttl])
+ self.cli_set(base_path + ddns + ['host-name', hostname])
- # commit changes
- self.cli_commit()
+ # commit changes
+ self.cli_commit()
+
+ # Check some generating config parameters
+ self.assertEqual(get_config_value('use'), 'if')
+ self.assertEqual(get_config_value('if'), interface)
+ self.assertEqual(get_config_value('protocol'), 'nsupdate')
+ self.assertEqual(get_config_value('server'), srv)
+ self.assertEqual(get_config_value('zone'), zone)
+ self.assertEqual(get_config_value('password'), key_file.name)
+ self.assertEqual(get_config_value('ttl'), ttl)
- # Check some generating config parameters
- self.assertEqual(get_config_value('protocol'), proto)
- self.assertEqual(get_config_value('login'), user)
- self.assertEqual(get_config_value('password'), password)
- self.assertEqual(get_config_value('server'), srv)
- self.assertEqual(get_config_value('usev6'), 'ifv6')
- self.assertEqual(get_config_value('ifv6'), interface)
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/dns_dynamic.py b/src/conf_mode/dns_dynamic.py
index f97225370..e070a3502 100755
--- a/src/conf_mode/dns_dynamic.py
+++ b/src/conf_mode/dns_dynamic.py
@@ -1,134 +1,138 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2018-2020 VyOS maintainers and contributors
+# Copyright (C) 2018-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from sys import exit
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.template import render
from vyos.util import call
from vyos.xml import defaults
from vyos import ConfigError
from vyos import airbag
airbag.enable()
config_file = r'/run/ddclient/ddclient.conf'
systemd_override = r'/run/systemd/system/ddclient.service.d/override.conf'
# Protocols that require zone
zone_allowed = ['cloudflare', 'godaddy', 'hetzner', 'gandi', 'nfsn']
# Protocols that do not require username
username_unnecessary = ['1984', 'cloudflare', 'cloudns', 'duckdns', 'freemyip', 'hetzner', 'keysystems', 'njalla']
# Protocols that support both IPv4 and IPv6
dualstack_supported = ['cloudflare', 'dyndns2', 'freedns', 'njalla']
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base_level = ['service', 'dns', 'dynamic']
if not conf.exists(base_level):
return None
dyndns = conf.get_config_dict(base_level, key_mangling=('-', '_'), get_first_key=True)
- for address in dyndns['address']:
- # Apply service specific defaults (stype = ['rfc2136', 'service'])
- for svc_type in dyndns['address'][address]:
- default_values = defaults(base_level + ['address', svc_type])
- for svc_cfg in dyndns['address'][address][svc_type]:
- dyndns['address'][address][svc_type][svc_cfg] = dict_merge(
- default_values, dyndns['address'][address][svc_type][svc_cfg])
+ if 'address' in dyndns:
+ for address in dyndns['address']:
+ # Apply service specific defaults (svc_type = ['rfc2136', 'service'])
+ for svc_type in dyndns['address'][address]:
+ default_values = defaults(base_level + ['address', svc_type])
+ for svc_cfg in dyndns['address'][address][svc_type]:
+ dyndns['address'][address][svc_type][svc_cfg] = dict_merge(
+ default_values, dyndns['address'][address][svc_type][svc_cfg])
dyndns['config_file'] = config_file
return dyndns
def verify(dyndns):
# bail out early - looks like removal from running config
- if not dyndns:
+ if not dyndns or 'address' not in dyndns:
return None
for address in dyndns['address']:
# RFC2136 - configuration validation
if 'rfc2136' in dyndns['address'][address]:
for config in dyndns['address'][address]['rfc2136'].values():
for field in ['host_name', 'zone', 'server', 'key']:
if field not in config:
raise ConfigError(f'"{field.replace("_", "-")}" is required for RFC2136 '
f'based Dynamic DNS service on "{address}"')
# Dynamic DNS service provider - configuration validation
if 'service' in dyndns['address'][address]:
for service, config in dyndns['address'][address]['service'].items():
error_msg = f'is required for Dynamic DNS service "{service}" on "{address}"'
for field in ['host_name', 'password', 'protocol']:
if field not in config:
raise ConfigError(f'"{field.replace("_", "-")}" {error_msg}')
if config['protocol'] in zone_allowed and 'zone' not in config:
raise ConfigError(f'"zone" {error_msg}')
if config['protocol'] not in zone_allowed and 'zone' in config:
raise ConfigError(f'"{config["protocol"]}" does not support "zone"')
if config['protocol'] not in username_unnecessary:
if 'username' not in config:
raise ConfigError(f'"username" {error_msg}')
if config['ip_version'] == 'both':
if config['protocol'] not in dualstack_supported:
- raise ConfigError(f'"{config["protocol"]}" does not support IPv4 and IPv6 at the same time')
+ raise ConfigError(f'"{config["protocol"]}" does not support '
+ f'both IPv4 and IPv6 at the same time')
# dyndns2 protocol in ddclient honors dual stack only for dyn.com (dyndns.org)
if config['protocol'] == 'dyndns2' and 'server' in config and config['server'] != 'members.dyndns.org':
- raise ConfigError(f'"{config["protocol"]}" for "{config["server"]}" does not support IPv4 and IPv6 at the same time')
+ raise ConfigError(f'"{config["protocol"]}" does not support '
+ f'both IPv4 and IPv6 at the same time for "{config["server"]}"')
return None
def generate(dyndns):
# bail out early - looks like removal from running config
- if not dyndns:
+ if not dyndns or 'address' not in dyndns:
return None
render(config_file, 'dns-dynamic/ddclient.conf.j2', dyndns)
render(systemd_override, 'dns-dynamic/override.conf.j2', dyndns)
return None
def apply(dyndns):
- if not dyndns:
+ # bail out early - looks like removal from running config
+ if not dyndns or 'address' not in dyndns:
call('systemctl stop ddclient.service')
if os.path.exists(config_file):
os.unlink(config_file)
else:
call('systemctl restart ddclient.service')
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)
diff --git a/src/op_mode/dns_dynamic.py b/src/op_mode/dns_dynamic.py
index d41a74db3..76ca5249b 100755
--- a/src/op_mode/dns_dynamic.py
+++ b/src/op_mode/dns_dynamic.py
@@ -1,113 +1,113 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2018-2020 VyOS maintainers and contributors
+# Copyright (C) 2018-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import argparse
import sys
import time
from tabulate import tabulate
from vyos.config import Config
from vyos.template import is_ipv4, is_ipv6
from vyos.util import call
cache_file = r'/run/ddclient/ddclient.cache'
columns = {
'host': 'Hostname',
'ipv4': 'IPv4 address',
'status-ipv4': 'IPv4 status',
'ipv6': 'IPv6 address',
'status-ipv6': 'IPv6 status',
'mtime': 'Last update',
}
def _get_formatted_host_records(host_data):
data_entries = []
for entry in host_data:
data_entries.append([entry.get(key) for key in columns.keys()])
header = columns.values()
output = tabulate(data_entries, header, numalign='left')
return output
def show_status():
# A ddclient status file might not always exist
if not os.path.exists(cache_file):
sys.exit(0)
data = []
with open(cache_file, 'r') as f:
for line in f:
if line.startswith('#'):
continue
props = {}
# ddclient cache rows have properties in 'key=value' format separated by comma
# we pick up the ones we are interested in
for kvraw in line.split(' ')[0].split(','):
k, v = kvraw.split('=')
if k in list(columns.keys()) + ['ip', 'status']: # ip and status are legacy keys
props[k] = v
# Extract IPv4 and IPv6 address and status from legacy keys
# Dual-stack isn't supported in legacy format, 'ip' and 'status' are for one of IPv4 or IPv6
if 'ip' in props:
if is_ipv4(props['ip']):
props['ipv4'] = props['ip']
props['status-ipv4'] = props['status']
elif is_ipv6(props['ip']):
props['ipv6'] = props['ip']
props['status-ipv6'] = props['status']
del props['ip']
# Convert mtime to human readable format
if 'mtime' in props:
props['mtime'] = time.strftime(
"%Y-%m-%d %H:%M:%S", time.localtime(int(props['mtime'], base=10)))
data.append(props)
print(_get_formatted_host_records(data))
def update_ddns():
call('systemctl stop ddclient.service')
if os.path.exists(cache_file):
os.remove(cache_file)
call('systemctl start ddclient.service')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
group = parser.add_mutually_exclusive_group()
group.add_argument("--status", help="Show DDNS status", action="store_true")
group.add_argument("--update", help="Update DDNS on a given interface", action="store_true")
args = parser.parse_args()
# Do nothing if service is not configured
c = Config()
if not c.exists_effective('service dns dynamic'):
print("Dynamic DNS not configured")
sys.exit(1)
if args.status:
show_status()
elif args.update:
update_ddns()