nik/FreedomBox
1
0
Fork 0
mirror of https://github.com/freedombox/FreedomBox.git synced 2026-08-19 12:36:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

dynamicdns: Allow updating both IPv4 and IPv6 records together

Browse Source 
- Change the IPv6 checkbox into an option to select from IPv4 only, IPv6 only,
or IPv4 and IPv6. Update functional tests.

- Add type hints everywhere.

- Separate out the generic URL update method to its own module. Add unit tests.

- Migrate old configuration to new configuration on first read.

- Store multiple IP addresses in the status. Support showing old status format
of single IP address as well. Show multiple IP addresses in the status table.

- Handle errors during update using exceptions for clearer code.

- For generic URL based updates, capture return code, stdout and stdin in error
message.

Tests:

- Adding a new GnuDIP domain leads to configuration setting ip_type = ipv4

- When adding a new dynamic domain, when generic option is selected IP Address
  Type is shown. When GnuDIP is selected, the field is hidden.

- IP Address Type field is as expected. Options and description are as expected.

- When a new domain is added for generic domain, the value of ip_type is set as
  expected. Changing the value to other values works. Configuration is updated
  as expected. Option values and description are as expected.

- The column for IP Address is now 'IP Addresses'. When multiple IP addresses
  are present in the status, they are shown in multiple lines in the tables as
  <pre>. When no IP addresses are present, a '-' is shown in the column.

- When generic domain update fails, subprocess failure code, stderr and stdout
  are shown in the error message.

- Unit tests work.

- Functional tests work.

- After 5 minute period, attempts are made to update the domain

- For GnuDIP and generic method:

  - IP address lookup works for IPv4 and IPv6.

  - IP address updates works for IPv4 (tested) and IPv6 (untested).

  - In 'both' configuration, when IPv6 update fails, IPv4 still succeeds.

  - Updates are skipped if the record is up-to-date.

  - Successful updates are performed otherwise.

  - Wrong password leads to proper error being shown.

- When the old status is stored, it is updated during read and proper status is
  shown.

- When configuration has old use_ipv6 key, then it is migrated when newer
  version of the service is started. True value is converted as 'ipv6', False
  value is converted as 'ipv4'. 'null' key is properly removed.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa 2026-06-26 07:39:06 -07:00 committed by James Valleroy
parent ca1b54f3cf
commit 62deda0ec7
No known key found for this signature in database
GPG Key ID: 77C0C75E7B650808
9 changed files with 321 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
plinth/modules/dynamicdns/__init__.py
View File

@ -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('<Ip>', 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('<Domain>', quote(domain['domain']))
if domain['username']:
update_url = update_url.replace('<User>', quote(domain['username']))
if domain['password']:
update_url = update_url.replace('<Pass>', 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',

16
plinth/modules/dynamicdns/forms.py
View File

@ -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

45
plinth/modules/dynamicdns/generic.py Normal file
View File

@ -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('<Ip>', quote(external_address))
if domain['domain']:
update_url = update_url.replace('<Domain>', quote(domain['domain']))
if domain['username']:
update_url = update_url.replace('<User>', quote(domain['username']))
if domain['password']:
update_url = update_url.replace('<Pass>', 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

13
plinth/modules/dynamicdns/gnudip.py
View File

@ -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')

4
plinth/modules/dynamicdns/static/dynamicdns.js
View File

@ -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';
}

6
plinth/modules/dynamicdns/templates/dynamicdns.html
View File

@ -26,7 +26,7 @@
<th>{% trans "Domain" %}</th>
<th>{% trans "Last update" %}</th>
<th>{% trans "Result" %}</th>
<th>{% trans "IP Address" %}</th>
<th>{% trans "IP Addresses" %}</th>
<th>{% trans "Actions" %}</th>
</tr>
</thead>
@ -66,7 +66,9 @@
({{ domain.error_code }})
{% endif %}
</td>
<td>{{ domain.ip_address|default_if_none:'-' }}</td>
<td><pre>{% for ip_address in domain.ip_addresses %}{{ ip_address }}
{% empty %}-
{% endfor %}</pre></td>
<td>
<a href="{% url 'dynamicdns:domain-edit' domain.domain %}"
class="btn btn-default btn-sm domain-edit" role="button"

8
plinth/modules/dynamicdns/tests/test_functional.py
View File

@ -35,7 +35,7 @@ _configs = {
'domain': 'freedombox3.example.com',
'username': 'tester3',
'password': 'testingtesting3',
'use_ipv6': True,
'ip_type': 'ipv6',
},
'freedns.afraid.org': {
'service_type': 'freedns.afraid.org',
@ -45,7 +45,7 @@ _configs = {
'domain': 'freedombox5.example.com',
'username': '',
'password': '',
'use_ipv6': False,
'ip_type': 'ipv4',
},
'other': {
'service_type': 'other',
@ -55,7 +55,7 @@ _configs = {
'domain': 'freedombox6.example.com',
'username': 'tester6',
'password': 'testingtesting6',
'use_ipv6': False,
'ip_type': 'both',
},
}
@ -115,7 +115,7 @@ def _configure(browser, config):
'/freedombox/sys/dynamicdns/domain/add/')
for key, value in config.items():
field_id = f'id_domain-{key}'
if key == 'service_type':
if key in ('service_type', 'ip_type'):
browser.find_by_id(field_id).select(value)
elif isinstance(value, bool):
if value:

112
plinth/modules/dynamicdns/tests/test_generic.py Normal file
View File

@ -0,0 +1,112 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Tests for generic URL based Dynamic DNS updates.
"""
import subprocess
from unittest.mock import call, patch
import pytest
from .. import generic
@pytest.fixture(name='domain')
def fixture_domain():
"""Return a domain configuration."""
return {
'domain': 'example.org',
'update_url': ('https://<User>:<Pass>@example.com'
'/update?hostname=<Domain>&ip=<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=<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']

24
plinth/modules/dynamicdns/tests/test_gnudip.py
View File

@ -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')