names: Sort domains by priority of their domain types

- First of the list is the most important one and may be used as "primary"
domain in apps.

- Change the return type of DomainName.list() from set to list so that order can
be preserved. Update all users of the API accordingly. Add type hints to all the
methods using this API to catch any errors.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa 2025-01-17 22:33:02 -08:00 committed by James Valleroy
parent 045b336a9b
commit aac12f4391
No known key found for this signature in database
GPG Key ID: 77C0C75E7B650808
18 changed files with 73 additions and 54 deletions

View File

@ -7,14 +7,14 @@ import copy
import json
from django.core.serializers.json import DjangoJSONEncoder
from django.http import HttpResponse
from django.http import HttpRequest, HttpResponse
from django.templatetags.static import static
from plinth import frontpage
from plinth.modules import names
def access_info(request, **kwargs):
def access_info(request: HttpRequest, **kwargs) -> HttpResponse:
"""API view to return a list of domains and types."""
domains = [{
'domain': domain.name,

View File

@ -3,6 +3,7 @@
import logging
import pathlib
from typing import Iterator
from django.urls import reverse_lazy
from django.utils.translation import gettext_lazy as _
@ -120,7 +121,7 @@ class CoturnApp(app_module.App):
privileged.uninstall()
def get_available_domains():
def get_available_domains() -> Iterator[str]:
"""Return an iterator with all domains able to have a certificate."""
return (domain.name for domain in names.components.DomainName.list()
if domain.domain_type.can_have_certificate)

View File

@ -10,6 +10,7 @@ from django.utils.translation import gettext_lazy as _
from plinth import cfg
from plinth.modules import ejabberd
from plinth.modules.coturn.forms import turn_uris_validator
from plinth.modules.names.components import DomainName
from plinth.utils import format_lazy
@ -51,15 +52,16 @@ class EjabberdForm(forms.Form):
help_text=_('Shared secret used to compute passwords for the '
'TURN server.'))
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
# Start with any existing domains from ejabberd configuration.
domains = set(ejabberd.get_domains())
domains = ejabberd.get_domains()
# Add other domains that can be configured.
from plinth.modules.names.components import DomainName
domains |= DomainName.list_names()
for domain in DomainName.list_names():
if domain not in domains:
domains.append(domain)
self.fields['domain_names'].choices = zip(domains, domains)

View File

@ -1,9 +1,8 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Forms for the email app.
"""
"""Forms for the email app."""
import re
from typing import Iterator
from django import forms
from django.core.exceptions import ValidationError
@ -14,7 +13,7 @@ from plinth.modules.names.components import DomainName
from . import aliases as aliases_module
def _get_domain_choices():
def _get_domain_choices() -> Iterator[tuple[str, str]]:
"""Double domain entries for inclusion in the choice field."""
return ((domain.name, domain.name) for domain in DomainName.list())

View File

@ -12,7 +12,6 @@ import re
from plinth.actions import privileged
from plinth.app import App
from plinth.modules import names
from plinth.modules.email import postfix
from plinth.modules.names.components import DomainName
@ -28,13 +27,13 @@ def get_domains():
return {'primary_domain': conf['mydomain'], 'all_domains': domains}
def set_all_domains(primary_domain=None):
def set_all_domains(primary_domain: str | None = None) -> None:
"""Set the primary domain and all the domains for postfix."""
all_domains = DomainName.list_names()
if not primary_domain:
primary_domain = get_domains()['primary_domain']
if primary_domain not in all_domains:
primary_domain = names.get_domain_name() or list(all_domains)[0]
primary_domain = all_domains[0]
# Update configuration and don't restart daemons
set_domains(primary_domain, list(all_domains))

View File

@ -211,7 +211,7 @@ def on_domain_removed(sender, domain_type, name='', **kwargs):
return False
def get_status():
def get_status() -> dict[str, dict]:
"""Get the current settings."""
status = privileged.get_status()

View File

@ -14,6 +14,8 @@ from . import privileged
logger = logging.getLogger(__name__)
_list_type = list
class LetsEncrypt(app.FollowerComponent):
"""Component to receive Let's Encrypt renewal hooks.
@ -129,7 +131,7 @@ class LetsEncrypt(app.FollowerComponent):
self._all[component_id] = self
@property
def domains(self):
def domains(self) -> list[str] | str:
"""Return a list of domains this component's app is interested in."""
if callable(self._domains):
return self._domains()
@ -141,7 +143,8 @@ class LetsEncrypt(app.FollowerComponent):
"""Return a list of all Let's Encrypt components."""
return cls._all.values()
def setup_certificates(self, app_domains=None):
def setup_certificates(
self, app_domains: str | _list_type[str] | None = None) -> None:
"""Setup app certificates for all interested domains.
For every domain, a certificate is copied. If a valid certificate is
@ -151,7 +154,6 @@ class LetsEncrypt(app.FollowerComponent):
app_domains is the list of domains for which certificates must be
copied. If it is not provided, the component's list of domains (which
may be acquired by a callable) is used.
"""
if not app_domains:
app_domains = self.domains

View File

@ -117,7 +117,7 @@ def setup(old_version: int):
@privileged
def get_status() -> dict[str, Any]:
def get_status() -> dict[str, dict]:
"""Return a dictionary of currently configured domains."""
domain_status = _get_status()
return {'domains': domain_status}

View File

@ -31,7 +31,7 @@ class SetupView(FormView):
matrixsynapse.setup_domain(form.cleaned_data['domain_name'])
return super().form_valid(form)
def get_context_data(self, *args, **kwargs):
def get_context_data(self, *args, **kwargs) -> dict[str, object]:
"""Provide context data to the template."""
context = super().get_context_data(**kwargs)
app = app_module.App.get('matrixsynapse')

View File

@ -24,7 +24,7 @@ class MinetestAppView(AppView): # pylint: disable=too-many-ancestors
initial.update(get_configuration())
return initial
def get_context_data(self, *args, **kwargs):
def get_context_data(self, *args, **kwargs) -> dict[str, object]:
"""Add service to the context data."""
context = super().get_context_data(*args, **kwargs)
context['domains'] = names.components.DomainName.list_names(

View File

@ -1,9 +1,8 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
FreedomBox app to configure Mumble server.
"""
"""FreedomBox app to configure Mumble server."""
import pathlib
from typing import Iterator
from django.urls import reverse_lazy
from django.utils.translation import gettext_lazy as _
@ -123,7 +122,7 @@ class MumbleApp(app_module.App):
return results
def get_available_domains():
def get_available_domains() -> Iterator[str]:
"""Return an iterator with all domains able to have a certificate."""
return (domain.name for domain in names.components.DomainName.list()
if domain.domain_type.can_have_certificate)

View File

@ -6,6 +6,7 @@ FreedomBox app to configure name services.
import logging
import pathlib
import subprocess
from typing import Iterator
from django.utils.translation import gettext_lazy as _
from django.utils.translation import gettext_noop
@ -200,8 +201,9 @@ def diagnose_resolution(domain: str) -> DiagnosticCheck:
return DiagnosticCheck('names-resolve', description, result, parameters)
def on_domain_added(sender, domain_type, name='', description='',
services=None, **kwargs):
def on_domain_added(sender: str, domain_type: str, name: str = '',
description: str = '',
services: str | list[str] | None = None, **kwargs):
"""Add domain to global list."""
if not domain_type:
return
@ -217,7 +219,7 @@ def on_domain_added(sender, domain_type, name='', description='',
domain_type, str(services))
def on_domain_removed(sender, domain_type, name='', **kwargs):
def on_domain_removed(sender: str, domain_type: str, name: str = '', **kwargs):
"""Remove domain from global list."""
if name:
component_id = 'domain-' + sender + '-' + name
@ -264,7 +266,7 @@ def set_hostname(hostname):
new_hostname=hostname)
def get_available_tls_domains():
def get_available_tls_domains() -> Iterator[str]:
"""Return an iterator with all domains able to have a certificate."""
return (domain.name for domain in components.DomainName.list()
if domain.domain_type.can_have_certificate)

View File

@ -28,6 +28,8 @@ _SERVICES = {
},
}
list_type = list
class DomainType(app.FollowerComponent):
"""Component to create a new type of domain.
@ -98,7 +100,10 @@ class DomainType(app.FollowerComponent):
@classmethod
def list(cls) -> dict[str, 'DomainType']:
"""Return a list of all domain types."""
return dict(cls._all)
sorted_items = sorted(cls._all.items(),
key=lambda item: item[1].priority, reverse=True)
domain_types = {key: value for key, value in sorted_items}
return domain_types
class DomainName(app.FollowerComponent):
@ -115,7 +120,8 @@ class DomainName(app.FollowerComponent):
"""
_all: ClassVar[dict[str, 'DomainName']] = {}
def __init__(self, component_id, name, domain_type, services):
def __init__(self, component_id: str, name: str, domain_type: str,
services: list[str] | str):
"""Initialize a domain name.
component_id should be a unique ID across all components of an app and
@ -172,7 +178,6 @@ class DomainName(app.FollowerComponent):
"""Return the service ID for a given port number.
XXX: Eliminate this and use a generalized approach eventually.
"""
if isinstance(service, str):
return service
@ -186,7 +191,7 @@ class DomainName(app.FollowerComponent):
return str(service)
def get_readable_services(self):
def get_readable_services(self) -> set[str]:
"""Return list of unique service strings that can be shown to user."""
services = self.services
if self.services == '__all__':
@ -197,16 +202,15 @@ class DomainName(app.FollowerComponent):
for service in services
}
def has_service(self, service):
def has_service(self, service: str | None) -> bool:
"""Return whether a service is available for this domain name."""
return (service is None or self.services == '__all__'
or service in self.services)
def remove(self):
def remove(self) -> None:
"""Remove the domain name from global list of domains.
It is acceptable to call remove() multiple times.
"""
try:
del self._all[self.component_id]
@ -219,24 +223,35 @@ class DomainName(app.FollowerComponent):
return cls._all[component_id]
@classmethod
def list(cls, filter_for_service=None):
"""Return list of domains."""
return [
def list(cls,
filter_for_service: str | None = None) -> list_type['DomainName']:
"""Return list of domains sorted by importance.
Domains are first sorted by priority with higher values showing up
first and then by their domain name.
"""
domains = [
domain for domain in cls._all.values()
if domain.has_service(filter_for_service)
]
return sorted(
domains, key=lambda domain:
(-domain.domain_type.priority, domain.name))
@classmethod
def list_names(cls, filter_for_service=None):
def list_names(cls,
filter_for_service: str | None = None) -> list_type[str]:
"""Return a set of unique domain names.
Domains are first sorted by priority with higher values showing up
first and then by their domain name.
Multiple different components may provide the same domain name. This
method could be used to retrieve a list of all domain names without
duplication.
"""
return {
domain.name
for domain in cls._all.values()
if domain.has_service(filter_for_service)
}
domain_names: dict[str, bool] = {}
for domain in cls.list(filter_for_service):
domain_names[domain.name] = True
return list(domain_names.keys())

View File

@ -22,7 +22,7 @@
</tr>
</thead>
<tbody>
{% for domain in status.domains|dictsort:"domain_type.display_name" %}
{% for domain in status.domains %}
<tr>
<td>{{ domain.domain_type.display_name }}</td>
<td class="names-domain-column">{{ domain.name }}</td>

View File

@ -141,10 +141,10 @@ def test_domain_name_list_names(domain_name):
DomainName('test-domain-name3', 'test3.example.com', 'test-domain-type',
'__all__')
domains = DomainName.list_names()
assert domains == {'test.example.com', 'test3.example.com'}
assert domains == ['test.example.com', 'test3.example.com']
domains = DomainName.list_names('http')
assert domains == {'test.example.com', 'test3.example.com'}
assert domains == ['test.example.com', 'test3.example.com']
domains = DomainName.list_names('unknown')
assert domains == {'test3.example.com'}
assert domains == ['test3.example.com']

View File

@ -144,7 +144,7 @@ class DomainDeleteView(TemplateView):
return redirect('names:index')
def get_status():
def get_status() -> dict[str, object]:
"""Get configured services per name."""
domains = components.DomainName.list()
used_domain_types = {domain.domain_type for domain in domains}

View File

@ -241,11 +241,11 @@ def _on_domain_removed(sender, domain_type, name='', **kwargs):
_set_trusted_domains()
def _set_trusted_domains():
def _set_trusted_domains() -> None:
"""Set the list of trusted domains."""
all_domains = DomainName.list_names()
with _ensure_nextcloud_running():
privileged.set_trusted_domains(list(all_domains))
privileged.set_trusted_domains(all_domains)
@contextlib.contextmanager

View File

@ -65,7 +65,7 @@ class ShowClientView(SuccessMessageMixin, TemplateView):
"""View to show a client's details."""
template_name = 'wireguard_show_client.html'
def get_context_data(self, **kwargs):
def get_context_data(self, **kwargs) -> dict[str, object]:
"""Return additional context data for rendering the template."""
context = super().get_context_data(**kwargs)
context['title'] = _('Allowed Client')