diff --git a/plinth/modules/dynamicdns/__init__.py b/plinth/modules/dynamicdns/__init__.py index 5dfaa13bf..a99a67665 100644 --- a/plinth/modules/dynamicdns/__init__.py +++ b/plinth/modules/dynamicdns/__init__.py @@ -7,7 +7,7 @@ import json import logging import subprocess import time -import urllib +from typing import Any, Literal, Tuple from django.utils.translation import gettext_lazy as _ @@ -20,7 +20,7 @@ from plinth.modules.users.components import UsersAndGroups from plinth.signals import domain_added, domain_removed from plinth.utils import format_lazy -from . import gnudip, manifest +from . import generic, gnudip, manifest logger = logging.getLogger(__name__) @@ -125,99 +125,99 @@ class DynamicDNSApp(app_module.App): self.enable() -def _lookup_public_address(domain): +def _lookup_public_address(ip_type: Literal['ipv4', 'ipv6']): """Return the IP address by querying an external server.""" try: - ip_type = 'ipv6' if domain['use_ipv6'] else 'ipv4' return lookup_public_address(ip_type) except Exception: return None -def _query_dns_address(domain): +def _query_dns_address(domain: str, ip_type: Literal['ipv4', + 'ipv6']) -> str | None: """Return the IP address in the DNS records.""" - ip_option = 'AAAA' if domain['use_ipv6'] else 'A' + ip_option = 'AAAA' if ip_type == 'ipv6' else 'A' try: - output = subprocess.check_output( - ['host', '-t', ip_option, domain['domain']]) + output = subprocess.check_output(['host', '-t', ip_option, domain]) return output.decode().split(' ')[-1].strip().lower() except subprocess.CalledProcessError as exception: - logger.warning('Unable to lookup DNS for host %s: %s', - domain['domain'], exception) + logger.warning('Unable to lookup DNS for host %s: %s', domain, + exception) return None -def _update_using_url(domain, external_address): - """Update DNS entry using an update URL.""" - update_url = domain['update_url'] - quote = urllib.parse.quote - if external_address: - update_url = update_url.replace('', quote(external_address)) +def _check_uptodate_for_ip_type( + domain: dict[str, Any], + ip_type: Literal['ipv4', 'ipv6']) -> Tuple[bool, str | None, str]: + """Return whether a domain is up-to-date for a given IP address type.""" + uptodate = False + dns_address = _query_dns_address(domain['domain'], ip_type) + external_address = _lookup_public_address(ip_type) + if dns_address == external_address and dns_address is not None: + logger.info('Dynamic domain %s is up-to-date: %s (%s)', + domain['domain'], dns_address, ip_type) + uptodate = True - if domain['domain']: - update_url = update_url.replace('', quote(domain['domain'])) - - if domain['username']: - update_url = update_url.replace('', quote(domain['username'])) - - if domain['password']: - update_url = update_url.replace('', quote(domain['password'])) - - options = ['-o', '/dev/null', '-t', '3', '-T', '3'] - if domain['use_http_basic_auth']: - options += [ - '--user', domain['username'], '--password', domain['password'] - ] - - if domain['disable_ssl_cert_check']: - options += ['--no-check-certificate'] - - if domain['use_ipv6']: - options += ['-6'] - else: - options += ['-4'] - - command = ['wget', '-O', '/dev/null'] + options + [update_url] - process = subprocess.run(command, check=False) - return process.returncode == 0, external_address + return uptodate, dns_address, external_address -def _update_dns_for_domain(domain): +def _update_dns_for_domain(domain: dict[str, Any], ip_type: Literal['ipv4', + 'ipv6'], + dns_address: str | None, + external_address: str) -> str | None: """Update DNS records for a single domain.""" - result = False - ip_address = None - error = None + logger.info( + 'Updating dynamic domain %s, DNS address %s, looked up ' + 'external address %s', domain['domain'], dns_address, external_address) + if domain['service_type'] == 'gnudip': + return gnudip.update(domain['server'], ip_type, domain['domain'], + domain['username'], domain['password']) + else: + return generic.update(domain, ip_type, external_address) + + +def _check_and_update_dns_for_domain(domain: dict[str, Any]): + """Update DNS records for a single domain only when it is out-of-date.""" + result = True + ip_addresses: list[str] = [] + error_code: str | None = None + error_message: str | None = None + + ip_types: list[Literal['ipv4', 'ipv6']] + if domain['ip_type'] == 'both': + ip_types = ['ipv4', 'ipv6'] + else: + ip_types = [domain['ip_type']] + + for ip_type in ip_types: + try: + uptodate, dns_address, external_address = \ + _check_uptodate_for_ip_type(domain, ip_type) - try: - dns_address = _query_dns_address(domain) - external_address = _lookup_public_address(domain) - if dns_address == external_address and dns_address is not None: - logger.info('Dynamic domain %s is up-to-date: %s', - domain['domain'], dns_address) - result = True ip_address = dns_address - error = ValueError('up-to-date') - else: - logger.info( - 'Updating dynamic domain %s, DNS address %s, looked up ' - 'external address %s', domain['domain'], dns_address, - external_address) - if domain['service_type'] == 'gnudip': - result, ip_address = gnudip.update(domain['server'], - domain['domain'], - domain['username'], - domain['password']) - else: - result, ip_address = _update_using_url(domain, - external_address) - except Exception as exception: - logger.exception('Failed to be update Dynamic DNS - %s', exception) - error = exception + if not uptodate: + ip_address = _update_dns_for_domain(domain, ip_type, + dns_address, + external_address) - set_status(domain, result, ip_address, error) + ip_addresses.append(ip_address) # type: ignore + except subprocess.CalledProcessError as exception: + logger.exception('Failed to be update Dynamic DNS - %s', exception) + result = False + error_code = str(exception.__class__.__name__) + error_message = f'Command failed: code={exception.returncode}, ' \ + f'stderr={exception.stderr.decode()}, ' \ + f'stdout={exception.stdout.decode()}' + except Exception as exception: + logger.exception('Failed to be update Dynamic DNS - %s', exception) + result = False + error_code = str(exception.__class__.__name__) + error_message = str(exception.args[0]) + + set_status(domain, result, ip_addresses, error_code, error_message) -def update_dns(_data): +def update_dns(_data) -> None: """For all configured domains, check and up to date DNS records.""" config = get_config() app = app_module.App.get('dynamicdns') @@ -226,10 +226,10 @@ def update_dns(_data): # Update for each domain for domain in config['domains'].values(): - _update_dns_for_domain(domain) + _check_and_update_dns_for_domain(domain) -def get_status(): +def get_status() -> dict[str, Any]: """Return the status of recent update for each domain.""" status = kvstore.get_default('dynamicdns_status', '{}') status = json.loads(status) @@ -242,16 +242,25 @@ def get_status(): status['domains'][domain] = { 'domain': domain, 'result': False, - 'ip_address': None, + 'ip_addresses': [], 'error_code': None, 'error_message': None, 'timestamp': 0, } + for domain_name, domain in status['domains'].items(): + domain.setdefault('ip_addresses', []) + if 'ip_address' in domain: + if domain['ip_address']: + domain['ip_addresses'].append(domain['ip_address']) + + del domain['ip_address'] + return status -def set_status(domain, result, ip_address, error=None): +def set_status(domain: dict[str, Any], result: bool, ip_addresses: list[str], + error_code: str | None, error_message: str | None): """Set the status of most recent update.""" status = kvstore.get_default('dynamicdns_status', '{}') status = json.loads(status) @@ -259,38 +268,50 @@ def set_status(domain, result, ip_address, error=None): domains[domain['domain']] = { 'domain': domain['domain'], 'result': result, - 'ip_address': ip_address, - 'error_code': str(error.__class__.__name__) if error else None, - 'error_message': str(error.args[0]) if error and error.args else None, + 'ip_addresses': ip_addresses, + 'error_code': error_code, + 'error_message': error_message, 'timestamp': int(time.time()), } kvstore.set('dynamicdns_status', json.dumps(status)) -def get_config(): +def get_config() -> dict[str, Any]: """Return the current configuration.""" - default_config = {'domains': {}} + default_config: dict[str, Any] = {'domains': {}} config = kvstore.get_default('dynamicdns_config', '{}') config = json.loads(config) or default_config - return _fix_corrupt_config(config) + return _migrate_old_config(config) -def _fix_corrupt_config(config): - """Fix malformed configuration result of bug in older version.""" - if 'null' not in config['domains']: - return config +def _migrate_old_config(config: dict[str, Any]) -> dict[str, Any]: + """Upgrade the old configuration to newer format.""" + updated = False + + # Fix malformed configuration result of bug in older version. + if 'null' in config['domains']: + del config['domains']['null'] + updated = True + + # Upgrade from 'use_ipv6' to 'ip_type' key in domain configuration. + for domain_name, domain in config['domains'].items(): + if 'use_ipv6' in domain: + domain['ip_type'] = 'ipv6' if domain['use_ipv6'] else 'ipv4' + del domain['use_ipv6'] + updated = True + + if updated: + set_config(config) - del config['domains']['null'] - set_config(config) return config -def set_config(config): +def set_config(config: dict[str, Any]): """Set a new configuration.""" kvstore.set('dynamicdns_config', json.dumps(config)) -def notify_domain_added(domain_name): +def notify_domain_added(domain_name: str): """Send a signal that domain has been added.""" if app_module.App.get('dynamicdns').is_enabled(): domain_added.send_robust(sender='dynamicdns', @@ -298,7 +319,7 @@ def notify_domain_added(domain_name): name=domain_name, services='__all__') -def notify_domain_removed(domain_name): +def notify_domain_removed(domain_name: str): """Send a signal that domain has been removed.""" if app_module.App.get('dynamicdns').is_enabled(): domain_removed.send_robust(sender='dynamicdns', diff --git a/plinth/modules/dynamicdns/forms.py b/plinth/modules/dynamicdns/forms.py index a870ed28a..0d9f7a7a5 100644 --- a/plinth/modules/dynamicdns/forms.py +++ b/plinth/modules/dynamicdns/forms.py @@ -45,11 +45,19 @@ class DomainForm(forms.Form): gettext_lazy('The username that was used when the account was ' 'created.') + help_ip_type = \ + gettext_lazy('Select based on what your ISP provides: a public IPv4 ' + 'address, a public IPv6 address, or both.') + provider_choices = (('gnudip', gettext_lazy('GnuDIP')), ('noip.com', 'noip.com'), ('freedns.afraid.org', 'freedns.afraid.org'), ('other', gettext_lazy('Other update URL'))) + ip_type_choices = (('ipv4', gettext_lazy('IPv4 address only')), + ('ipv6', gettext_lazy('IPv6 address only')), + ('both', gettext_lazy('Both IPv4 and IPv6 addresses'))) + service_type = forms.ChoiceField(label=gettext_lazy('Service Type'), help_text=help_service_type, choices=provider_choices) @@ -89,8 +97,9 @@ class DomainForm(forms.Form): show_password = forms.BooleanField(label=gettext_lazy('Show password'), required=False) - use_ipv6 = forms.BooleanField( - label=gettext_lazy('Use IPv6 instead of IPv4'), required=False) + ip_type = forms.ChoiceField(label=gettext_lazy('IP Address Type'), + help_text=help_ip_type, + choices=ip_type_choices) def clean(self): """Further validate and transform field data.""" @@ -125,6 +134,9 @@ class DomainForm(forms.Form): if not cleaned_data.get(field_name): self.add_error(field_name, message) + if service_type == 'gnudip': + cleaned_data['ip_type'] = 'ipv4' + del cleaned_data['show_password'] return cleaned_data diff --git a/plinth/modules/dynamicdns/generic.py b/plinth/modules/dynamicdns/generic.py new file mode 100644 index 000000000..da0952984 --- /dev/null +++ b/plinth/modules/dynamicdns/generic.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +""" +Generic HTTP request client for updating Dynamic DNS records. +""" + +import subprocess +import urllib.parse +from typing import Any, Literal + + +def update(domain: dict[str, Any], ip_type: Literal['ipv4', 'ipv6'], + external_address: str | None) -> str | None: + """Update DNS entry using an update URL.""" + update_url = domain['update_url'] + quote = urllib.parse.quote + if external_address: + update_url = update_url.replace('', quote(external_address)) + + if domain['domain']: + update_url = update_url.replace('', quote(domain['domain'])) + + if domain['username']: + update_url = update_url.replace('', quote(domain['username'])) + + if domain['password']: + update_url = update_url.replace('', quote(domain['password'])) + + options = ['-t', '3', '-T', '3'] + if domain['use_http_basic_auth']: + options += [ + '--user', domain['username'], '--password', domain['password'] + ] + + if domain['disable_ssl_cert_check']: + options += ['--no-check-certificate'] + + if ip_type == 'ipv6': + options += ['-6'] + else: + options += ['-4'] + + command = ['wget', '-O', '-'] + options + [update_url] + subprocess.run(command, check=True, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + return external_address diff --git a/plinth/modules/dynamicdns/gnudip.py b/plinth/modules/dynamicdns/gnudip.py index 6f6b93b6a..99397c48e 100644 --- a/plinth/modules/dynamicdns/gnudip.py +++ b/plinth/modules/dynamicdns/gnudip.py @@ -7,6 +7,7 @@ import hashlib import logging import socket from html.parser import HTMLParser +from typing import Literal import requests @@ -67,8 +68,8 @@ def _request_get_ipv4(*args, **kwargs): socket.getaddrinfo = original -def update(server: str, domain: str, username: str, - password: str) -> tuple[bool, str | None]: +def update(server: str, ip_type: Literal['ipv4', 'ipv6'], domain: str, + username: str, password: str) -> str | None: """Update Dynamic DNS record using GnuDIP protocol. Protocol documentation: @@ -78,6 +79,9 @@ def update(server: str, domain: str, username: str, support IPv6 (it does have any code to update AAAA records). So, make a request only using IPv4 stack. """ + if ip_type == 'ipv6': + raise NotImplementedError + domain = domain.removeprefix(username + '.') password_digest = hashlib.md5(password.encode()).hexdigest() @@ -105,4 +109,7 @@ def update(server: str, domain: str, username: str, update_result = _extract_content_from_meta_tags(update_response.text) _check_required_keys(update_result, ['retc']) result = (int(update_result['retc']) == 0) - return result, update_result.get('addr') + if not result: + raise Exception('Server responded with an error') + + return update_result.get('addr') diff --git a/plinth/modules/dynamicdns/static/dynamicdns.js b/plinth/modules/dynamicdns/static/dynamicdns.js index acc18d840..b467b8aaa 100644 --- a/plinth/modules/dynamicdns/static/dynamicdns.js +++ b/plinth/modules/dynamicdns/static/dynamicdns.js @@ -65,7 +65,7 @@ document.addEventListener('DOMContentLoaded', () => { document.getElementById('id_domain-update_url').closest('.form-group').style.display = 'none'; document.getElementById('id_domain-disable_ssl_cert_check').closest('.form-group').style.display = 'none'; document.getElementById('id_domain-use_http_basic_auth').closest('.form-group').style.display = 'none'; - document.getElementById('id_domain-use_ipv6').closest('.form-group').style.display = 'none'; + document.getElementById('id_domain-ip_type').closest('.form-group').style.display = 'none'; document.getElementById('id_domain-server').closest('.form-group').style.display = 'block'; } @@ -73,7 +73,7 @@ document.addEventListener('DOMContentLoaded', () => { document.getElementById('id_domain-update_url').closest('.form-group').style.display = 'block'; document.getElementById('id_domain-disable_ssl_cert_check').closest('.form-group').style.display = 'block'; document.getElementById('id_domain-use_http_basic_auth').closest('.form-group').style.display = 'block'; - document.getElementById('id_domain-use_ipv6').closest('.form-group').style.display = 'block'; + document.getElementById('id_domain-ip_type').closest('.form-group').style.display = 'block'; document.getElementById('id_domain-server').closest('.form-group').style.display = 'none'; } diff --git a/plinth/modules/dynamicdns/templates/dynamicdns.html b/plinth/modules/dynamicdns/templates/dynamicdns.html index 7eaf9c74e..c8bf8ace0 100644 --- a/plinth/modules/dynamicdns/templates/dynamicdns.html +++ b/plinth/modules/dynamicdns/templates/dynamicdns.html @@ -26,7 +26,7 @@ {% trans "Domain" %} {% trans "Last update" %} {% trans "Result" %} - {% trans "IP Address" %} + {% trans "IP Addresses" %} {% trans "Actions" %} @@ -66,7 +66,9 @@ ({{ domain.error_code }}) {% endif %} - {{ domain.ip_address|default_if_none:'-' }} + 
{% for ip_address in domain.ip_addresses %}{{ ip_address }}
+{% empty %}-
+{% endfor %}
:@example.com' + '/update?hostname=&ip='), + 'username': 'tester', + 'password': 'testingtesting', + 'use_http_basic_auth': False, + 'disable_ssl_cert_check': True, + } + + +@patch('subprocess.run') +def test_update_ip_types(run, domain): + """Test that various IP types are handled as expected.""" + # IPv4 + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '-4' in run.mock_calls[0].args[0] + + # IPv6 + run.reset_mock() + generic.update(domain, 'ipv6', external_address='1.1.1.1') + assert '-6' in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_ssl_cert_check(run, domain): + """Test that SSL certificate check can be disabled.""" + # Check certificate + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--no-check-certificate' in run.mock_calls[0].args[0] + + # Don't check certificate + run.reset_mock() + domain['disable_ssl_cert_check'] = False + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--no-check-certificate' not in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_basic_auth(run, domain): + """Test that using basic authentication works.""" + # Check certificate + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--username' not in run.mock_calls[0].args[0] + assert '--password' not in run.mock_calls[0].args[0] + + # Don't check certificate + run.reset_mock() + domain['use_http_basic_auth'] = True + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert ['--user', 'tester', '--password', + 'testingtesting'] == run.mock_calls[0].args[0][7:11] + + +@patch('subprocess.run') +def test_update_url_parameters(run, domain): + """Test that replacing parameters in the URL works as expected.""" + generic.update(domain, 'ipv4', external_address='1.1.1.1') + expected_url = ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=1.1.1.1') + assert expected_url in run.mock_calls[0].args[0] + + # No substitution for unavailable values + run.reset_mock() + generic.update(domain, 'ipv4', external_address=None) + expected_url = ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=') + assert expected_url in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_call(run, domain): + """Test that calling the wget command is as expected.""" + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert run.mock_calls == [ + call([ + 'wget', '-O', '-', '-t', '3', '-T', '3', '--no-check-certificate', + '-4', + ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=1.1.1.1') + ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ] + + +@patch('subprocess.run') +def test_update_return_value(run, domain): + """Test that return value or raising exception works.""" + assert '1.1.1.1' == generic.update(domain, 'ipv4', + external_address='1.1.1.1') + + run.side_effect = subprocess.CalledProcessError(1, ['foo']) + with pytest.raises(subprocess.CalledProcessError) as exception_info: + generic.update(domain, 'ipv4', external_address='1.1.1.1') + + assert exception_info.value.returncode == 1 + assert exception_info.value.cmd == ['foo'] diff --git a/plinth/modules/dynamicdns/tests/test_gnudip.py b/plinth/modules/dynamicdns/tests/test_gnudip.py index c1ff29f20..1546096e1 100644 --- a/plinth/modules/dynamicdns/tests/test_gnudip.py +++ b/plinth/modules/dynamicdns/tests/test_gnudip.py @@ -1,3 +1,8 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +""" +Tests for GnuDIP based Dynamic DNS updates. +""" + from unittest.mock import Mock, patch import pytest @@ -66,11 +71,18 @@ def test_update_success(): update_resp = Mock() update_resp.text = response_to_update_request - with patch("plinth.modules.dynamicdns.gnudip.requests.get", + with patch('plinth.modules.dynamicdns.gnudip.requests.get', side_effect=[salt_resp, update_resp]) as mock_get: - result, addr = gnudip.update(server="http://www.2mbit.com:80", - domain="gnudip.dyn.mpis.net", - username="gnudip", password="password") - assert result - assert addr == "24.81.172.128" + addr = gnudip.update(server='http://www.2mbit.com:80', ip_type='ipv4', + domain='gnudip.dyn.mpis.net', username='gnudip', + password='password') + assert addr == '24.81.172.128' assert mock_get.call_count == 2 + + +def test_update_ipv6(): + """Test that updating IPv6 raises and exception.""" + with pytest.raises(NotImplementedError): + gnudip.update(server='http://www.2mbit.com:80', ip_type='ipv6', + domain='gnudip.dyn.mpis.net', username='gnudip', + password='password')