From 4b09d91f93c7e7222bb5d6e22244cbbcbb4c542e Mon Sep 17 00:00:00 2001 From: Sunil Mohan Adapa Date: Tue, 5 Mar 2024 08:58:25 -0800 Subject: [PATCH] *: Add type hints for diagnose method Helps: #2410. - Ensure that diagnostics methods and parameters are type checked so that we can catch any potential issues. - Move plinth/modules/diagnostics/check.py to plinth/diagnostic_check.py to avoid many circular dependencies created. This is due to plinth.modules.diagnostics automatically imported when plinth.modules.diagnostics.check is imported. Also app.py is already (type) dependent on diagnostic_check due to diagnose() method. To make the Check classes independent of diagnostic module is okay. Tests: - Run make check-type. - Run full diagnostics with following apps installed: torproxy, tor. - Test to netcat to 9051 in tor works. - Test 'port available for internal/external networks' in firewall works. - Test 'Package is latest' works. - Test 'Access url with proxy' in privoxy works. Signed-off-by: Sunil Mohan Adapa Reviewed-by: James Valleroy [jvalleroy: Also move tests for diagnostic_check] Signed-off-by: James Valleroy --- doc/dev/reference/components/index.rst | 4 +- plinth/action_utils.py | 6 +-- plinth/app.py | 14 ++--- plinth/config.py | 8 +-- plinth/daemon.py | 52 +++++++++++-------- .../check.py => diagnostic_check.py} | 5 +- plinth/modules/apache/components.py | 26 +++++++--- .../modules/apache/tests/test_components.py | 2 +- plinth/modules/datetime/__init__.py | 6 +-- plinth/modules/diagnostics/__init__.py | 5 +- plinth/modules/diagnostics/views.py | 2 +- plinth/modules/firewall/__init__.py | 13 +++-- plinth/modules/firewall/components.py | 16 ++++-- plinth/modules/firewall/privileged.py | 7 ++- .../modules/firewall/tests/test_components.py | 2 +- plinth/modules/letsencrypt/__init__.py | 4 +- plinth/modules/networks/__init__.py | 3 +- plinth/modules/privoxy/__init__.py | 9 ++-- plinth/modules/tor/__init__.py | 10 ++-- plinth/modules/torproxy/__init__.py | 8 +-- plinth/modules/users/__init__.py | 18 ++++--- plinth/modules/users/privileged.py | 2 +- plinth/package.py | 21 ++++---- plinth/tests/test_app.py | 4 +- plinth/tests/test_config.py | 2 +- plinth/tests/test_daemon.py | 22 ++++---- .../test_diagnostic_check.py} | 5 +- plinth/tests/test_package.py | 2 +- 28 files changed, 161 insertions(+), 117 deletions(-) rename plinth/{modules/diagnostics/check.py => diagnostic_check.py} (91%) rename plinth/{modules/diagnostics/tests/test_check.py => tests/test_diagnostic_check.py} (91%) diff --git a/doc/dev/reference/components/index.rst b/doc/dev/reference/components/index.rst index ad3fd9d97..0e523cba7 100644 --- a/doc/dev/reference/components/index.rst +++ b/doc/dev/reference/components/index.rst @@ -37,8 +37,8 @@ Base Classes Other Classes ^^^^^^^^^^^^^ -.. autoclass:: plinth.modules.diagnostics.check.DiagnosticCheck +.. autoclass:: plinth.diagnostic_check.DiagnosticCheck :members: -.. autoclass:: plinth.modules.diagnostics.check.Result +.. autoclass:: plinth.diagnostic_check.Result :members: diff --git a/plinth/action_utils.py b/plinth/action_utils.py index c3d6945f9..3e6bffaca 100644 --- a/plinth/action_utils.py +++ b/plinth/action_utils.py @@ -275,7 +275,7 @@ def uwsgi_disable(config_name): service_start('uwsgi') -def get_addresses(): +def get_addresses() -> list[dict[str, str | bool]]: """Return a list of IP addresses and hostnames.""" addresses = get_ip_addresses() @@ -309,14 +309,14 @@ def get_addresses(): return addresses -def get_ip_addresses(): +def get_ip_addresses() -> list[dict[str, str | bool]]: """Return a list of IP addresses assigned to the system.""" addresses = [] output = subprocess.check_output(['ip', '-o', 'addr']) for line in output.decode().splitlines(): parts = line.split() - address = { + address: dict[str, str | bool] = { 'kind': '4' if parts[2] == 'inet' else '6', 'address': parts[3].split('/')[0], 'url_address': parts[3].split('/')[0], diff --git a/plinth/app.py b/plinth/app.py index c7e74548a..6969720f1 100644 --- a/plinth/app.py +++ b/plinth/app.py @@ -7,9 +7,10 @@ import collections import enum import inspect import logging -from typing import ClassVar +from typing import ClassVar, TypeAlias from plinth import cfg +from plinth.diagnostic_check import DiagnosticCheck from plinth.signals import post_app_loading from . import clients as clients_module @@ -17,6 +18,8 @@ from . import db logger = logging.getLogger(__name__) +_list_type: TypeAlias = list + class App: """Implement common functionality for an app. @@ -208,11 +211,11 @@ class App: if not component.is_leader: component.set_enabled(enabled) - def diagnose(self): + def diagnose(self) -> _list_type[DiagnosticCheck]: """Run diagnostics and return results. Return value must be a list of results. Each result is a - :class:`~plinth.modules.diagnostics.check.DiagnosticCheck` with a + :class:`~plinth.diagnostic_check.DiagnosticCheck` with a unique check_id, a user visible description of the test, and the result. The test result is a string enumeration from 'failed', 'passed', 'error', 'warning' and 'not_done'. @@ -300,12 +303,11 @@ class Component: def disable(self): """Run operations to disable the component.""" - @staticmethod - def diagnose(): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return results. Return value must be a list of results. Each result is a - :class:`~plinth.modules.diagnostics.check.DiagnosticCheck` with a + :class:`~plinth.diagnostic_check.DiagnosticCheck` with a unique check_id, a user visible description of the test, and the result. The test result is a string enumeration from 'failed', 'passed', 'error', 'warning' and 'not_done'. diff --git a/plinth/config.py b/plinth/config.py index 1ab32cf0d..d274afd08 100644 --- a/plinth/config.py +++ b/plinth/config.py @@ -5,6 +5,8 @@ import pathlib from django.utils.translation import gettext_noop +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) from plinth.privileged import config as privileged from . import app as app_module @@ -99,10 +101,8 @@ class DropinConfigs(app_module.FollowerComponent): for path in self.etc_paths: privileged.dropin_unlink(self.app_id, path, missing_ok=True) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Check all links/copies and return generate diagnostic results.""" - from plinth.modules.diagnostics.check import DiagnosticCheck, Result - results = [] for path in self.etc_paths: etc_path = self._get_etc_path(path) @@ -118,7 +118,7 @@ class DropinConfigs(app_module.FollowerComponent): result_string = Result.PASSED if result else Result.FAILED description = gettext_noop( 'Static configuration {etc_path} is setup properly') - parameters = {'etc_path': str(etc_path)} + parameters: DiagnosticCheckParameters = {'etc_path': str(etc_path)} results.append( DiagnosticCheck(check_id, description, result_string, parameters)) diff --git a/plinth/daemon.py b/plinth/daemon.py index 24844f1be..dad74d8d0 100644 --- a/plinth/daemon.py +++ b/plinth/daemon.py @@ -8,13 +8,17 @@ import psutil from django.utils.translation import gettext_noop from plinth import action_utils, app +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) class Daemon(app.LeaderComponent): """Component to manage a background daemon or any systemd unit.""" - def __init__(self, component_id, unit, strict_check=False, - listen_ports=None, alias=None): + def __init__(self, component_id: str, unit: str, + strict_check: bool = False, + listen_ports: list[tuple[int, str]] | None = None, + alias: str | None = None): """Initialize a new daemon component. 'component_id' must be a unique string across all apps and components @@ -82,7 +86,7 @@ class Daemon(app.LeaderComponent): """Return whether the daemon/unit is running.""" return action_utils.service_is_running(self.unit) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Check if the daemon is running and listening on expected ports. See :py:meth:`plinth.app.Component.diagnose`. @@ -95,15 +99,15 @@ class Daemon(app.LeaderComponent): return results - def _diagnose_unit_is_running(self): + def _diagnose_unit_is_running(self) -> DiagnosticCheck: """Check if a daemon is running.""" - from plinth.modules.diagnostics.check import DiagnosticCheck, Result - check_id = f'daemon-running-{self.unit}' result = Result.PASSED if self.is_running() else Result.FAILED description = gettext_noop('Service {service_name} is running') - parameters = {'service_name': self.unit} + parameters: DiagnosticCheckParameters = { + 'service_name': str(self.unit) + } return DiagnosticCheck(check_id, description, result, parameters) @@ -179,7 +183,9 @@ def app_is_running(app_): return True -def diagnose_port_listening(port, kind='tcp', listen_address=None): +def diagnose_port_listening( + port: int, kind: str = 'tcp', + listen_address: str | None = None) -> DiagnosticCheck: """Run a diagnostic on whether a port is being listened on. Kind must be one of inet, inet4, inet6, tcp, tcp4, tcp6, udp, @@ -187,11 +193,9 @@ def diagnose_port_listening(port, kind='tcp', listen_address=None): information. """ - from plinth.modules.diagnostics.check import DiagnosticCheck, Result - result = _check_port(port, kind, listen_address) - parameters = {'kind': kind, 'port': port} + parameters: DiagnosticCheckParameters = {'kind': kind, 'port': port} if listen_address: parameters['listen_address'] = listen_address check_id = f'daemon-listening-address-{kind}-{port}-{listen_address}' @@ -206,7 +210,8 @@ def diagnose_port_listening(port, kind='tcp', listen_address=None): parameters) -def _check_port(port, kind='tcp', listen_address=None): +def _check_port(port: int, kind: str = 'tcp', + listen_address: str | None = None) -> bool: """Return whether a port is being listened on.""" run_kind = kind @@ -228,11 +233,12 @@ def _check_port(port, kind='tcp', listen_address=None): continue # Port should match - if connection.laddr[1] != port: + if connection.laddr[1] != port: # type: ignore[misc] continue # Listen address if requested should match - if listen_address and connection.laddr[0] != listen_address: + if listen_address and connection.laddr[ + 0] != listen_address: # type: ignore[misc] continue # Special additional checks only for IPv4 @@ -244,22 +250,21 @@ def _check_port(port, kind='tcp', listen_address=None): return True # Full IPv6 address range includes mapped IPv4 address also - if connection.laddr[0] == '::': + if connection.laddr[0] == '::': # type: ignore[misc] return True return False -def diagnose_netcat(host, port, input='', negate=False): +def diagnose_netcat(host: str, port: int, remote_input: str = '', + negate: bool = False) -> DiagnosticCheck: """Run a diagnostic using netcat.""" - from plinth.modules.diagnostics.check import DiagnosticCheck, Result - try: process = subprocess.Popen(['nc', host, str(port)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - process.communicate(input=input.encode()) + process.communicate(input=remote_input.encode()) if process.returncode != 0: result = Result.FAILED if not negate else Result.PASSED else: @@ -269,10 +274,13 @@ def diagnose_netcat(host, port, input='', negate=False): check_id = f'daemon-netcat-{host}-{port}' description = gettext_noop('Connect to {host}:{port}') - parameters = {'host': host, 'port': port, 'negate': negate} + parameters: DiagnosticCheckParameters = { + 'host': host, + 'port': port, + 'negate': negate + } if negate: check_id = f'daemon-netcat-negate-{host}-{port}' description = gettext_noop('Cannot connect to {host}:{port}') - return DiagnosticCheck(check_id, description.format(host=host, port=port), - result, parameters) + return DiagnosticCheck(check_id, description, result, parameters) diff --git a/plinth/modules/diagnostics/check.py b/plinth/diagnostic_check.py similarity index 91% rename from plinth/modules/diagnostics/check.py rename to plinth/diagnostic_check.py index 70afc75fb..f67661421 100644 --- a/plinth/modules/diagnostics/check.py +++ b/plinth/diagnostic_check.py @@ -5,11 +5,14 @@ import dataclasses import json from dataclasses import dataclass, field from enum import StrEnum +from typing import TypeAlias from django.utils.translation import gettext from plinth.utils import SafeFormatter +DiagnosticCheckParameters: TypeAlias = dict[str, str | int | bool | None] + class Result(StrEnum): """The result of a diagnostic check.""" @@ -26,7 +29,7 @@ class DiagnosticCheck: check_id: str description: str result: Result = Result.NOT_DONE - parameters: dict = field(default_factory=dict) + parameters: DiagnosticCheckParameters = field(default_factory=dict) @property def translated_description(self): diff --git a/plinth/modules/apache/components.py b/plinth/modules/apache/components.py index 4fdce303e..2a9ed9e1d 100644 --- a/plinth/modules/apache/components.py +++ b/plinth/modules/apache/components.py @@ -7,7 +7,8 @@ import subprocess from django.utils.translation import gettext_noop from plinth import action_utils, app -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) from plinth.privileged import service as service_privileged from . import privileged @@ -58,7 +59,7 @@ class Webserver(app.LeaderComponent): """Disable the Apache configuration.""" privileged.disable(self.web_name, self.kind) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Check if the web path is accessible by clients. See :py:meth:`plinth.app.Component.diagnose`. @@ -135,8 +136,12 @@ class Uwsgi(app.LeaderComponent): and action_utils.service_is_running('uwsgi') -def diagnose_url(url, kind=None, env=None, check_certificate=True, - extra_options=None, wrapper=None, expected_output=None): +def diagnose_url(url: str, kind: str | None = None, + env: dict[str, str] | None = None, + check_certificate: bool = True, + extra_options: list[str] | None = None, + wrapper: str | None = None, + expected_output: str | None = None) -> DiagnosticCheck: """Run a diagnostic on whether a URL is accessible. Kind can be '4' for IPv4 or '6' for IPv6. @@ -148,7 +153,7 @@ def diagnose_url(url, kind=None, env=None, check_certificate=True, except FileNotFoundError: result = Result.ERROR - parameters = {'url': url, 'kind': kind} + parameters: DiagnosticCheckParameters = {'url': url, 'kind': kind} if kind: check_id = f'apache-url-kind-{url}-{kind}' description = gettext_noop('Access URL {url} on tcp{kind}') @@ -159,7 +164,8 @@ def diagnose_url(url, kind=None, env=None, check_certificate=True, return DiagnosticCheck(check_id, description, result, parameters) -def diagnose_url_on_all(url, expect_redirects=False, **kwargs): +def diagnose_url_on_all(url: str, expect_redirects: bool = False, + **kwargs) -> list[DiagnosticCheck]: """Run a diagnostic on whether a URL is accessible.""" results = [] for address in action_utils.get_addresses(): @@ -173,8 +179,12 @@ def diagnose_url_on_all(url, expect_redirects=False, **kwargs): return results -def check_url(url, kind=None, env=None, check_certificate=True, - extra_options=None, wrapper=None, expected_output=None): +def check_url(url: str, kind: str | None = None, + env: dict[str, str] | None = None, + check_certificate: bool = True, + extra_options: list[str] | None = None, + wrapper: str | None = None, + expected_output: str | None = None) -> bool: """Check whether a URL is accessible.""" command = ['curl', '--location', '-f', '-w', '%{response_code}'] diff --git a/plinth/modules/apache/tests/test_components.py b/plinth/modules/apache/tests/test_components.py index 6cbd1c574..39d6e1229 100644 --- a/plinth/modules/apache/tests/test_components.py +++ b/plinth/modules/apache/tests/test_components.py @@ -9,10 +9,10 @@ from unittest.mock import call, patch import pytest from plinth import app +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules.apache.components import (Uwsgi, Webserver, check_url, diagnose_url, diagnose_url_on_all) -from plinth.modules.diagnostics.check import DiagnosticCheck, Result def test_webserver_init(): diff --git a/plinth/modules/datetime/__init__.py b/plinth/modules/datetime/__init__.py index 743385c70..052349009 100644 --- a/plinth/modules/datetime/__init__.py +++ b/plinth/modules/datetime/__init__.py @@ -11,8 +11,8 @@ from django.utils.translation import gettext_noop from plinth import app as app_module from plinth import menu from plinth.daemon import Daemon, RelatedDaemon +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules.backups.components import BackupRestore -from plinth.modules.diagnostics.check import DiagnosticCheck, Result from plinth.package import Packages from . import manifest @@ -89,7 +89,7 @@ class DateTimeApp(app_module.App): **manifest.backup) self.add(backup_restore) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() if self._is_time_managed(): @@ -107,7 +107,7 @@ class DateTimeApp(app_module.App): self.enable() -def _diagnose_time_synchronized(): +def _diagnose_time_synchronized() -> DiagnosticCheck: """Check whether time is synchronized to NTP server.""" result = Result.FAILED try: diff --git a/plinth/modules/diagnostics/__init__.py b/plinth/modules/diagnostics/__init__.py index 64992bf05..9e9a56bc0 100644 --- a/plinth/modules/diagnostics/__init__.py +++ b/plinth/modules/diagnostics/__init__.py @@ -17,11 +17,12 @@ from django.utils.translation import gettext_noop from plinth import app as app_module from plinth import daemon, glib, kvstore, menu from plinth import operation as operation_module +from plinth.diagnostic_check import (CheckJSONDecoder, CheckJSONEncoder, + DiagnosticCheck, Result) from plinth.modules.apache.components import diagnose_url_on_all from plinth.modules.backups.components import BackupRestore from . import manifest -from .check import CheckJSONDecoder, CheckJSONEncoder, Result _description = [ _('The system diagnostic test will run a number of checks on your ' @@ -75,7 +76,7 @@ class DiagnosticsApp(app_module.App): super().setup(old_version) self.enable() - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() results.append(daemon.diagnose_port_listening(8000, 'tcp4')) diff --git a/plinth/modules/diagnostics/views.py b/plinth/modules/diagnostics/views.py index 6a9437631..6102db56e 100644 --- a/plinth/modules/diagnostics/views.py +++ b/plinth/modules/diagnostics/views.py @@ -14,10 +14,10 @@ from django.views.generic import TemplateView from plinth import operation from plinth.app import App +from plinth.diagnostic_check import Result from plinth.modules import diagnostics from plinth.views import AppView -from .check import Result from .forms import ConfigureForm logger = logging.getLogger(__name__) diff --git a/plinth/modules/firewall/__init__.py b/plinth/modules/firewall/__init__.py index 9de569029..2eb1c87ac 100644 --- a/plinth/modules/firewall/__init__.py +++ b/plinth/modules/firewall/__init__.py @@ -10,8 +10,8 @@ from django.utils.translation import gettext_noop from plinth import app as app_module from plinth import cfg, menu from plinth.daemon import Daemon +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules.backups.components import BackupRestore -from plinth.modules.diagnostics.check import DiagnosticCheck, Result from plinth.package import Packages, install from plinth.utils import Version, format_lazy, import_from_gi @@ -96,7 +96,7 @@ class FirewallApp(app_module.App): _run_setup() return True - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() config = privileged.get_config() @@ -265,7 +265,8 @@ def remove_passthrough(ipv, *args): config_direct.removePassthrough('(sas)', ipv, args) -def _diagnose_default_zone(config): +def _diagnose_default_zone( + config: privileged.FirewallConfig) -> DiagnosticCheck: """Diagnose whether the default zone is external.""" check_id = 'firewall-default-zone' description = gettext_noop('Default zone is external') @@ -274,7 +275,8 @@ def _diagnose_default_zone(config): return DiagnosticCheck(check_id, description, result) -def _diagnose_firewall_backend(config): +def _diagnose_firewall_backend( + config: privileged.FirewallConfig) -> DiagnosticCheck: """Diagnose whether the firewall backend is nftables.""" check_id = 'firewall-backend' description = gettext_noop('Firewall backend is nftables') @@ -283,7 +285,8 @@ def _diagnose_firewall_backend(config): return DiagnosticCheck(check_id, description, result) -def _diagnose_direct_passthroughs(config): +def _diagnose_direct_passthroughs( + config: privileged.FirewallConfig) -> DiagnosticCheck: """Diagnose direct passthroughs for local service protection. Currently, we just check that the number of passthroughs is at least 12, diff --git a/plinth/modules/firewall/components.py b/plinth/modules/firewall/components.py index 666292c18..e3c381b26 100644 --- a/plinth/modules/firewall/components.py +++ b/plinth/modules/firewall/components.py @@ -5,16 +5,19 @@ App component for other apps to use firewall functionality. import logging import re -from typing import ClassVar +from typing import ClassVar, TypeAlias from django.utils.translation import gettext_noop from plinth import app +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) from plinth.modules import firewall -from plinth.modules.diagnostics.check import DiagnosticCheck, Result logger = logging.getLogger(__name__) +_list_type: TypeAlias = list + class Firewall(app.FollowerComponent): """Component to open/close firewall ports for an app.""" @@ -114,7 +117,7 @@ class Firewall(app.FollowerComponent): if not re.fullmatch(r'tun\d+', interface) ] - def diagnose(self): + def diagnose(self) -> _list_type[DiagnosticCheck]: """Check if the firewall ports are open and only as expected. See :py:meth:`plinth.app.Component.diagnose`. @@ -124,7 +127,7 @@ class Firewall(app.FollowerComponent): internal_ports = firewall.get_enabled_services(zone='internal') external_ports = firewall.get_enabled_services(zone='external') for port_detail in self.ports_details: - port = port_detail['name'] + port = str(port_detail['name']) details = ', '.join( (f'{port_number}/{protocol}' for port_number, protocol in port_detail['details'])) @@ -134,7 +137,10 @@ class Firewall(app.FollowerComponent): result = Result.PASSED if port in internal_ports else Result.FAILED description = gettext_noop( 'Port {name} ({details}) available for internal networks') - parameters = {'name': port, 'details': details} + parameters: DiagnosticCheckParameters = { + 'name': port, + 'details': details + } results.append( DiagnosticCheck(check_id, description, result, parameters)) diff --git a/plinth/modules/firewall/privileged.py b/plinth/modules/firewall/privileged.py index f673b50dc..4463b528e 100644 --- a/plinth/modules/firewall/privileged.py +++ b/plinth/modules/firewall/privileged.py @@ -2,12 +2,15 @@ """Configuration helper for FreedomBox firewall interface.""" import subprocess +from typing import TypeAlias import augeas from plinth import action_utils from plinth.actions import privileged +FirewallConfig: TypeAlias = dict[str, str | list[str]] + def _flush_iptables_rules(): """Flush firewalld iptables rules before restarting it. @@ -132,9 +135,9 @@ def setup(): @privileged -def get_config(): +def get_config() -> FirewallConfig: """Return firewalld configuration for diagnostics.""" - config = {} + config: FirewallConfig = {} # Get the default zone. output = subprocess.check_output(['firewall-cmd', '--get-default-zone']) diff --git a/plinth/modules/firewall/tests/test_components.py b/plinth/modules/firewall/tests/test_components.py index e4a53f303..a6d943db9 100644 --- a/plinth/modules/firewall/tests/test_components.py +++ b/plinth/modules/firewall/tests/test_components.py @@ -8,7 +8,7 @@ from unittest.mock import call, patch import pytest from plinth.app import App -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules.firewall.components import (Firewall, FirewallLocalProtection) diff --git a/plinth/modules/letsencrypt/__init__.py b/plinth/modules/letsencrypt/__init__.py index ab42b18f6..823bd4045 100644 --- a/plinth/modules/letsencrypt/__init__.py +++ b/plinth/modules/letsencrypt/__init__.py @@ -11,10 +11,10 @@ from django.utils.translation import gettext_noop from plinth import app as app_module from plinth import cfg, menu from plinth.config import DropinConfigs +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules import names from plinth.modules.apache.components import diagnose_url from plinth.modules.backups.components import BackupRestore -from plinth.modules.diagnostics.check import DiagnosticCheck, Result from plinth.modules.names.components import DomainType from plinth.package import Packages from plinth.signals import domain_added, domain_removed, post_app_loading @@ -89,7 +89,7 @@ class LetsEncryptApp(app_module.App): post_app_loading.connect(_certificate_handle_modified) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() diff --git a/plinth/modules/networks/__init__.py b/plinth/modules/networks/__init__.py index e793dddb0..df9e168d2 100644 --- a/plinth/modules/networks/__init__.py +++ b/plinth/modules/networks/__init__.py @@ -9,6 +9,7 @@ from django.utils.translation import gettext_lazy as _ from plinth import app as app_module from plinth import daemon, kvstore, menu, network from plinth.config import DropinConfigs +from plinth.diagnostic_check import DiagnosticCheck from plinth.package import Packages from . import privileged @@ -72,7 +73,7 @@ class NetworksApp(app_module.App): ]) self.add(dropin_configs) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() diff --git a/plinth/modules/privoxy/__init__.py b/plinth/modules/privoxy/__init__.py index 72f83ea22..0e4a01e28 100644 --- a/plinth/modules/privoxy/__init__.py +++ b/plinth/modules/privoxy/__init__.py @@ -11,6 +11,7 @@ from plinth import action_utils from plinth import app as app_module from plinth import cfg, frontpage, menu from plinth.daemon import Daemon +from plinth.diagnostic_check import DiagnosticCheck from plinth.modules.apache.components import diagnose_url from plinth.modules.backups.components import BackupRestore from plinth.modules.firewall.components import Firewall @@ -86,7 +87,7 @@ class PrivoxyApp(app_module.App): **manifest.backup) self.add(backup_restore) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() results.append(diagnose_url('https://www.debian.org')) @@ -102,16 +103,16 @@ class PrivoxyApp(app_module.App): self.enable() -def diagnose_url_with_proxy(): +def diagnose_url_with_proxy() -> list[DiagnosticCheck]: """Run a diagnostic on a URL with a proxy.""" url = 'https://debian.org/' # Gives a simple redirect to www. results = [] for address in action_utils.get_addresses(): - proxy = 'http://{host}:8118/'.format(host=address['url_address']) + proxy = f'http://{address["url_address"]}:8118/' env = {'https_proxy': proxy} - result = diagnose_url(url, kind=address['kind'], env=env) + result = diagnose_url(url, kind=str(address['kind']), env=env) result.check_id = f'privoxy-url-proxy-kind-{url}-{address["kind"]}' result.description = gettext_noop( 'Access {url} with proxy {proxy} on tcp{kind}') diff --git a/plinth/modules/tor/__init__.py b/plinth/modules/tor/__init__.py index d72781b7b..d603aab2b 100644 --- a/plinth/modules/tor/__init__.py +++ b/plinth/modules/tor/__init__.py @@ -13,10 +13,10 @@ from plinth import cfg, kvstore, menu from plinth import setup as setup_module_ # Not setup_module, for pytest from plinth.daemon import (Daemon, app_is_running, diagnose_netcat, diagnose_port_listening) +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.modules import torproxy from plinth.modules.apache.components import Webserver from plinth.modules.backups.components import BackupRestore -from plinth.modules.diagnostics.check import DiagnosticCheck, Result from plinth.modules.firewall.components import Firewall from plinth.modules.names.components import DomainType from plinth.modules.torproxy.utils import is_apt_transport_tor_enabled @@ -123,7 +123,7 @@ class TorApp(app_module.App): super().disable() update_hidden_service_domain() - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() @@ -235,7 +235,7 @@ def update_hidden_service_domain(status=None): name=status['hs_hostname'], services=services) -def _diagnose_control_port(): +def _diagnose_control_port() -> list[DiagnosticCheck]: """Diagnose whether Tor control port is open on 127.0.0.1 only.""" results = [] @@ -249,7 +249,7 @@ def _diagnose_control_port(): negate = False results.append( - diagnose_netcat(address['address'], 9051, input='QUIT\n', - negate=negate)) + diagnose_netcat(str(address['address']), 9051, + remote_input='QUIT\n', negate=negate)) return results diff --git a/plinth/modules/torproxy/__init__.py b/plinth/modules/torproxy/__init__.py index 289be5fb6..cd38b8c17 100644 --- a/plinth/modules/torproxy/__init__.py +++ b/plinth/modules/torproxy/__init__.py @@ -11,6 +11,7 @@ from django.utils.translation import gettext_noop from plinth import app as app_module from plinth import cfg, frontpage, kvstore, menu from plinth.daemon import Daemon +from plinth.diagnostic_check import DiagnosticCheck from plinth.modules.apache.components import diagnose_url from plinth.modules.backups.components import BackupRestore from plinth.modules.firewall.components import Firewall @@ -100,7 +101,7 @@ class TorProxyApp(app_module.App): privileged.configure(apt_transport_tor=False) super().disable() - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() results.append(_diagnose_url_via_tor('http://www.debian.org', '4')) @@ -133,7 +134,8 @@ class TorProxyApp(app_module.App): privileged.uninstall() -def _diagnose_url_via_tor(url, kind=None): +def _diagnose_url_via_tor(url: str, + kind: str | None = None) -> DiagnosticCheck: """Diagnose whether a URL is reachable via Tor.""" result = diagnose_url(url, kind=kind, wrapper='torsocks') result.check_id = 'torproxy-url' @@ -142,7 +144,7 @@ def _diagnose_url_via_tor(url, kind=None): return result -def _diagnose_tor_use(url, kind=None): +def _diagnose_tor_use(url: str, kind: str | None = None) -> DiagnosticCheck: """Diagnose whether webpage at URL reports that we are using Tor.""" expected_output = 'Congratulations. This browser is configured to use Tor.' result = diagnose_url(url, kind=kind, wrapper='torsocks', diff --git a/plinth/modules/users/__init__.py b/plinth/modules/users/__init__.py index c9774b405..61b398c6a 100644 --- a/plinth/modules/users/__init__.py +++ b/plinth/modules/users/__init__.py @@ -13,7 +13,8 @@ from plinth import app as app_module from plinth import cfg, menu from plinth.config import DropinConfigs from plinth.daemon import Daemon -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) from plinth.package import Packages from plinth.privileged import service as service_privileged @@ -85,7 +86,7 @@ class UsersApp(app_module.App): groups=groups) self.add(users_and_groups) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return the results.""" results = super().diagnose() @@ -112,7 +113,7 @@ class UsersApp(app_module.App): privileged.create_group('freedombox-share') -def _diagnose_ldap_entry(search_item): +def _diagnose_ldap_entry(search_item: str) -> DiagnosticCheck: """Diagnose that an LDAP entry exists.""" check_id = f'users-ldap-entry-{search_item}' result = Result.FAILED @@ -125,12 +126,13 @@ def _diagnose_ldap_entry(search_item): pass description = gettext_noop('Check LDAP entry "{search_item}"') - parameters = {'search_item': search_item} + parameters: DiagnosticCheckParameters = {'search_item': search_item} return DiagnosticCheck(check_id, description, result, parameters) -def _diagnose_nslcd_config(config, key, value): +def _diagnose_nslcd_config(config: dict[str, str], key: str, + value: str) -> DiagnosticCheck: """Diagnose that nslcd has a configuration.""" check_id = f'users-nslcd-config-{key}' try: @@ -139,12 +141,12 @@ def _diagnose_nslcd_config(config, key, value): result = Result.FAILED description = gettext_noop('Check nslcd config "{key} {value}"') - parameters = {'key': key, 'value': value} + parameters: DiagnosticCheckParameters = {'key': key, 'value': value} return DiagnosticCheck(check_id, description, result, parameters) -def _diagnose_nsswitch_config(): +def _diagnose_nsswitch_config() -> list[DiagnosticCheck]: """Diagnose that Name Service Switch is configured to use LDAP.""" nsswitch_conf = '/etc/nsswitch.conf' aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD + @@ -169,7 +171,7 @@ def _diagnose_nsswitch_config(): break description = gettext_noop('Check nsswitch config "{database}"') - parameters = {'database': database} + parameters: DiagnosticCheckParameters = {'database': database} results.append( DiagnosticCheck(check_id, description, result, parameters)) diff --git a/plinth/modules/users/privileged.py b/plinth/modules/users/privileged.py index a67284e84..b1189691b 100644 --- a/plinth/modules/users/privileged.py +++ b/plinth/modules/users/privileged.py @@ -178,7 +178,7 @@ def _configure_ldapscripts(): @privileged -def get_nslcd_config(): +def get_nslcd_config() -> dict[str, str]: """Get nslcd configuration for diagnostics.""" nslcd_conf = '/etc/nslcd.conf' aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD + diff --git a/plinth/package.py b/plinth/package.py index 5de556c60..94ff19ee8 100644 --- a/plinth/package.py +++ b/plinth/package.py @@ -12,6 +12,8 @@ from django.utils.translation import gettext_lazy, gettext_noop import plinth.privileged.packages as privileged from plinth import app as app_module +from plinth.diagnostic_check import (DiagnosticCheck, + DiagnosticCheckParameters, Result) from plinth.errors import MissingPackageError from plinth.utils import format_lazy @@ -200,10 +202,8 @@ class Packages(app_module.FollowerComponent): uninstall([package for package in packages if package in packages_set], purge=True) - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Run diagnostics and return results.""" - from plinth.modules.diagnostics.check import DiagnosticCheck, Result - results = super().diagnose() cache = apt.Cache() for package_expression in self.package_expressions: @@ -213,7 +213,9 @@ class Packages(app_module.FollowerComponent): check_id = f'package-available-{package_expression}' description = gettext_noop('Package {package_expression} is ' 'not available for install') - parameters = {'package_expression': str(package_expression)} + parameters: DiagnosticCheckParameters = { + 'package_expression': str(package_expression) + } results.append( DiagnosticCheck(check_id, description, Result.FAILED, parameters)) @@ -223,16 +225,17 @@ class Packages(app_module.FollowerComponent): latest_version = '?' if package_name in cache: package = cache[package_name] - latest_version = package.candidate.version - if package.candidate.is_installed: - result = Result.PASSED + if package.candidate: + latest_version = package.candidate.version + if package.candidate.is_installed: + result = Result.PASSED check_id = f'package-latest-{package_name}' description = gettext_noop('Package {package_name} is the latest ' 'version ({latest_version})') parameters = { - 'package_name': package_name, - 'latest_version': latest_version, + 'package_name': str(package_name), + 'latest_version': str(latest_version) } results.append( DiagnosticCheck(check_id, description, result, parameters)) diff --git a/plinth/tests/test_app.py b/plinth/tests/test_app.py index bb161ba9f..86bcfec65 100644 --- a/plinth/tests/test_app.py +++ b/plinth/tests/test_app.py @@ -10,7 +10,7 @@ import pytest from plinth.app import (App, Component, EnableState, FollowerComponent, Info, LeaderComponent, apps_init) -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import DiagnosticCheck, Result # pylint: disable=protected-access @@ -34,7 +34,7 @@ class LeaderTest(FollowerComponent): """Test class for using LeaderComponent in tests.""" is_leader = True - def diagnose(self): + def diagnose(self) -> list[DiagnosticCheck]: """Return diagnostic results.""" return [ DiagnosticCheck('test-result-' + self.component_id, diff --git a/plinth/tests/test_config.py b/plinth/tests/test_config.py index 08376d358..6f82b8172 100644 --- a/plinth/tests/test_config.py +++ b/plinth/tests/test_config.py @@ -9,7 +9,7 @@ import pytest from plinth.app import App from plinth.config import DropinConfigs -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import DiagnosticCheck, Result pytestmark = pytest.mark.usefixtures('mock_privileged') privileged_modules_to_mock = ['plinth.privileged.config'] diff --git a/plinth/tests/test_daemon.py b/plinth/tests/test_daemon.py index 9dabdf092..d6cfcff69 100644 --- a/plinth/tests/test_daemon.py +++ b/plinth/tests/test_daemon.py @@ -12,7 +12,7 @@ import pytest from plinth.app import App, FollowerComponent, Info from plinth.daemon import (Daemon, RelatedDaemon, SharedDaemon, app_is_running, diagnose_netcat, diagnose_port_listening) -from plinth.modules.diagnostics.check import DiagnosticCheck, Result +from plinth.diagnostic_check import DiagnosticCheck, Result privileged_modules_to_mock = ['plinth.privileged.service'] @@ -295,32 +295,32 @@ def test_diagnose_port_listening(connections): def test_diagnose_netcat(popen): """Test running diagnostic test using netcat.""" popen().returncode = 0 - result = diagnose_netcat('test-host', 3300, input='test-input') + result = diagnose_netcat('test-host', 3300, remote_input='test-input') parameters = {'host': 'test-host', 'port': 3300, 'negate': False} assert result == DiagnosticCheck('daemon-netcat-test-host-3300', - 'Connect to test-host:3300', - Result.PASSED, parameters) + 'Connect to {host}:{port}', Result.PASSED, + parameters) assert popen.mock_calls[1][1] == (['nc', 'test-host', '3300'], ) assert popen.mock_calls[2] == call().communicate(input=b'test-input') - result = diagnose_netcat('test-host', 3300, input='test-input', + result = diagnose_netcat('test-host', 3300, remote_input='test-input', negate=True) parameters2 = parameters.copy() parameters2['negate'] = True assert result == DiagnosticCheck('daemon-netcat-negate-test-host-3300', - 'Cannot connect to test-host:3300', + 'Cannot connect to {host}:{port}', Result.FAILED, parameters2) popen().returncode = 1 - result = diagnose_netcat('test-host', 3300, input='test-input') + result = diagnose_netcat('test-host', 3300, remote_input='test-input') assert result == DiagnosticCheck('daemon-netcat-test-host-3300', - 'Connect to test-host:3300', - Result.FAILED, parameters) + 'Connect to {host}:{port}', Result.FAILED, + parameters) - result = diagnose_netcat('test-host', 3300, input='test-input', + result = diagnose_netcat('test-host', 3300, remote_input='test-input', negate=True) assert result == DiagnosticCheck('daemon-netcat-negate-test-host-3300', - 'Cannot connect to test-host:3300', + 'Cannot connect to {host}:{port}', Result.PASSED, parameters2) diff --git a/plinth/modules/diagnostics/tests/test_check.py b/plinth/tests/test_diagnostic_check.py similarity index 91% rename from plinth/modules/diagnostics/tests/test_check.py rename to plinth/tests/test_diagnostic_check.py index 734200ca8..caf6f1199 100644 --- a/plinth/modules/diagnostics/tests/test_check.py +++ b/plinth/tests/test_diagnostic_check.py @@ -5,9 +5,8 @@ import json import pytest -from plinth.modules.diagnostics.check import (CheckJSONDecoder, - CheckJSONEncoder, - DiagnosticCheck, Result) +from plinth.diagnostic_check import (CheckJSONDecoder, CheckJSONEncoder, + DiagnosticCheck, Result) def test_result(): diff --git a/plinth/tests/test_package.py b/plinth/tests/test_package.py index 76ec16f44..993484d08 100644 --- a/plinth/tests/test_package.py +++ b/plinth/tests/test_package.py @@ -9,8 +9,8 @@ from unittest.mock import Mock, call, patch import pytest from plinth.app import App +from plinth.diagnostic_check import DiagnosticCheck, Result from plinth.errors import MissingPackageError -from plinth.modules.diagnostics.check import DiagnosticCheck, Result from plinth.package import Package, Packages, packages_installed