pagekite: Use privileged decorator for actions

Tests:

- Functional tests work
- Initial setup succeeds
- Configuration can be set and new configuration is properly reflected in app
  page and configuration files.
- A new service can be added and reflects in configuration files.
- Service can be deleted and reflects in configuration files.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>

plinth/modules/pagekite/__init__.py

@@ -1,11 +1,8 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
-"""
+"""FreedomBox app to configure PageKite."""
-FreedomBox app to configure PageKite.
-"""
 
 from django.utils.translation import gettext_lazy as _
 
-from plinth import actions
 from plinth import app as app_module
 from plinth import cfg, menu
 from plinth.daemon import Daemon

plinth/modules/pagekite/forms.py

@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
+"""Forms for configuring Pagekite."""
 
 import copy
-import json
 
 from django import forms
 from django.contrib import messages
@@ -9,16 +9,14 @@ from django.core import validators
 from django.utils.translation import gettext as _
 from django.utils.translation import gettext_lazy
 
-from plinth.errors import ActionError
+from . import privileged, utils
-
-from . import utils
 
 
 class TrimmedCharField(forms.CharField):
-    """Trim the contents of a CharField"""
+    """Trim the contents of a CharField."""
 
     def clean(self, value):
-        """Clean and validate the field value"""
+        """Clean and validate the field value."""
         if value:
             value = value.strip()
 
@@ -26,7 +24,7 @@ class TrimmedCharField(forms.CharField):
 
 
 class ConfigurationForm(forms.Form):
-    """Configure PageKite credentials and frontend"""
+    """Configure PageKite credentials and frontend."""
 
     server_domain = forms.CharField(
         label=gettext_lazy('Server domain'), required=False,
@@ -72,9 +70,7 @@ class ConfigurationForm(forms.Form):
 
         if old != new:
             frontend = f"{new['server_domain']}:{new['server_port']}"
-            utils.run([
+            privileged.set_config(frontend, kite_name, new['kite_secret'])
-                'set-config', '--kite-name', kite_name, '--frontend', frontend
-            ], input=new['kite_secret'].encode())
             messages.success(request, _('Configuration updated'))
 
             # Update kite name registered with Name Services module.
@@ -82,7 +78,8 @@ class ConfigurationForm(forms.Form):
 
 
 class BaseCustomServiceForm(forms.Form):
-    """Basic form functionality to handle a custom service"""
+    """Basic form functionality to handle a custom service."""
+
     choices = [('http', 'http'), ('https', 'https'), ('raw', 'raw')]
     protocol = forms.ChoiceField(choices=choices,
                                  label=gettext_lazy('protocol'))
@@ -96,7 +93,7 @@ class BaseCustomServiceForm(forms.Form):
                                     required=False)
 
     def convert_formdata_to_service(self, formdata):
-        """Add information to make a service out of the form data"""
+        """Add information to make a service out of the form data."""
         # convert integers to str (to compare values with DEFAULT_SERVICES)
         for field in ('frontend_port', 'backend_port'):
             formdata[field] = str(formdata[field])
@@ -126,15 +123,15 @@ class DeleteCustomServiceForm(BaseCustomServiceForm):
 
     def delete(self, request):
         service = self.convert_formdata_to_service(self.cleaned_data)
-        utils.run(['remove-service', '--service', json.dumps(service)])
+        privileged.remove_service(service)
         messages.success(request, _('Deleted custom service'))
 
 
 class AddCustomServiceForm(BaseCustomServiceForm):
-    """Adds the save() method and validation to not add predefined services"""
+    """Adds the save() method and validation to not add predefined services."""
 
     def matches_predefined_service(self, formdata):
-        """Returns whether the user input matches a predefined service"""
+        """Return whether the user input matches a predefined service."""
         service = self.convert_formdata_to_service(formdata)
         match_found = False
         for predefined_service_obj in utils.PREDEFINED_SERVICES.values():
@@ -168,9 +165,9 @@ class AddCustomServiceForm(BaseCustomServiceForm):
     def save(self, request):
         service = self.convert_formdata_to_service(self.cleaned_data)
         try:
-            utils.run(['add-service', '--service', json.dumps(service)])
+            privileged.add_service(service)
             messages.success(request, _('Added custom service'))
-        except ActionError as exception:
+        except Exception as exception:
             if "already exists" in str(exception):
                 messages.error(request, _('This service already exists'))
             else:

plinth/modules/pagekite/privileged.py

@@ -1,21 +1,15 @@
-#!/usr/bin/python3
 # SPDX-License-Identifier: AGPL-3.0-or-later
-"""
+"""Configure PageKite."""
-Configuration helper for PageKite interface.
-"""
 
-import argparse
-import json
 import os
-import sys
+from typing import Union
 
 import augeas
 
 from plinth import action_utils
+from plinth.actions import privileged
 from plinth.modules.pagekite import utils
 
-aug = None
-
 PATHS = {
     'service_on':
         os.path.join(utils.CONF_PATH, '*', 'service_on', '*'),
@@ -32,35 +26,10 @@ PATHS = {
 }
 
 
-def parse_arguments():
+@privileged
-    """Return parsed command line arguments as dictionary"""
+def get_config() -> dict[str, object]:
-    parser = argparse.ArgumentParser()
+    """Return the current configuration as JSON dictionary."""
-    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
+    aug = _augeas_load()
-
-    # Configuration
-    subparsers.add_parser('get-config', help='Return current configuration')
-    set_config = subparsers.add_parser(
-        'set-config',
-        help='Configure kite name, its secret and frontend. Secret is read '
-        'from stdin.')
-    set_config.add_argument('--kite-name',
-                            help='Name of the kite (eg: mybox.pagekite.me)')
-    set_config.add_argument('--frontend', help='Frontend url')
-
-    # Add/remove pagekite services (service_on entries)
-    add_service = subparsers.add_parser('add-service',
-                                        help='Add a pagekite service')
-    add_service.add_argument('--service', help='json service dictionary')
-    remove_service = subparsers.add_parser('remove-service',
-                                           help='Remove a pagekite service')
-    remove_service.add_argument('--service', help='json service dictionary')
-
-    subparsers.required = True
-    return parser.parse_args()
-
-
-def subcommand_get_config(_):
-    """Print the current configuration as JSON dictionary."""
     if aug.match(PATHS['abort_not_configured']):
         aug.remove(PATHS['abort_not_configured'])
         aug.save()
@@ -70,9 +39,9 @@ def subcommand_get_config(_):
     else:
         frontend = aug.get(PATHS['frontend']) or ''
 
-    frontend = frontend.split(':')
+    frontend_parts = frontend.split(':')
-    server_domain = frontend[0]
+    server_domain = frontend_parts[0]
-    server_port = frontend[1] if len(frontend) >= 2 else '80'
+    server_port = frontend_parts[1] if len(frontend_parts) >= 2 else '80'
 
     status = {
         'kite_name': aug.get(PATHS['kitename']),
@@ -113,30 +82,32 @@ def subcommand_get_config(_):
 
             service['url'] = url
 
-    print(json.dumps(status))
+    return status
 
 
-def subcommand_set_config(arguments):
+@privileged
+def set_config(frontend: str, kite_name: str, kite_secret: str):
     """Set pagekite kite name, secret and frontend URL."""
+    aug = _augeas_load()
     aug.remove(PATHS['abort_not_configured'])
 
-    aug.set(PATHS['kitename'], arguments.kite_name)
+    aug.set(PATHS['kitename'], kite_name)
-    aug.set(PATHS['kitesecret'], sys.stdin.read())
+    aug.set(PATHS['kitesecret'], kite_secret)
 
-    frontend_domain = arguments.frontend.split(':')[0]
+    frontend_domain = frontend.split(':')[0]
     if frontend_domain in ('pagekite.net', 'defaults', 'default'):
         aug.set(PATHS['defaults'], '')
         aug.remove(PATHS['frontend'])
     else:
         aug.remove(PATHS['defaults'])
-        aug.set(PATHS['frontend'], arguments.frontend)
+        aug.set(PATHS['frontend'], frontend)
 
     aug.save()
 
     for service_name in utils.PREDEFINED_SERVICES.keys():
         service = utils.PREDEFINED_SERVICES[service_name]['params']
         try:
-            _add_service(service)
+            _add_service(aug, service)
         except RuntimeError:
             pass
 
@@ -146,10 +117,12 @@ def subcommand_set_config(arguments):
         action_utils.service_restart('pagekite')
 
 
-def subcommand_remove_service(arguments):
+@privileged
-    """Searches and removes the service(s) that match all given parameters"""
+def remove_service(service: dict[str, Union[str, bool]]):
-    service = utils.load_service(arguments.service)
+    """Search and remove the service(s) that match all given parameters."""
-    paths = _get_existing_service_paths(service)
+    aug = _augeas_load()
+    service = utils.load_service(service)
+    paths = _get_existing_service_paths(aug, service)
     # TODO: theoretically, everything to do here is:
     # [aug.remove(path) for path in paths]
     # but augeas won't let you save the changed files and doesn't say why
@@ -172,8 +145,8 @@ def subcommand_remove_service(arguments):
     action_utils.service_restart('pagekite')
 
 
-def _get_existing_service_paths(service, keys=None):
+def _get_existing_service_paths(aug, service, keys=None):
-    """Return paths of existing services that match the given service params"""
+    """Return paths of existing services that match the given service."""
     # construct an augeas query path with patterns like:
     #     */service_on/*[protocol='http']
     path = PATHS['service_on']
@@ -182,13 +155,13 @@ def _get_existing_service_paths(service, keys=None):
     return aug.match(path)
 
 
-def _add_service(service):
+def _add_service(aug, service):
     """Add a new service into configuration."""
-    if _get_existing_service_paths(service, ['protocol', 'kitename']):
+    if _get_existing_service_paths(aug, service, ['protocol', 'kitename']):
         msg = "Service with the parameters %s already exists"
         raise RuntimeError(msg % service)
 
-    root = _get_new_service_path(service['protocol'])
+    root = _get_new_service_path(aug, service['protocol'])
     # TODO: after adding a service, augeas fails writing the config;
     # so add the service_on entry manually instead
     path = _convert_augeas_path_to_filepath(root)
@@ -197,16 +170,18 @@ def _add_service(service):
         servicefile.write(line)
 
 
-def subcommand_add_service(arguments):
+@privileged
-    """Add one service"""
+def add_service(service: dict[str, Union[str, bool]]):
-    service = utils.load_service(arguments.service)
+    """Add one service."""
-    _add_service(service)
+    aug = _augeas_load()
+    service = utils.load_service(service)
+    _add_service(aug, service)
     action_utils.service_try_restart('pagekite')
 
 
 def _convert_augeas_path_to_filepath(augpath, prefix='/files',
                                      suffix='service_on'):
-    """Convert an augeas service_on path to the actual file path"""
+    """Convert an augeas service_on path to the actual file path."""
     if augpath.startswith(prefix):
         augpath = augpath.replace(prefix, "", 1)
 
@@ -216,35 +191,21 @@ def _convert_augeas_path_to_filepath(augpath, prefix='/files',
     return augpath.rstrip('/')
 
 
-def _get_new_service_path(protocol):
+def _get_new_service_path(aug, protocol):
-    """Get the augeas path of a new service for a protocol
+    """Get the augeas path of a new service for a protocol.
 
-    This takes care of existing services using a /service_on/*/ query"""
+    This takes care of existing services using a /service_on/*/ query
+    """
     root = utils.get_augeas_servicefile_path(protocol)
     new_index = len(aug.match(root + '/*')) + 1
     return os.path.join(root, str(new_index))
 
 
-def augeas_load():
+def _augeas_load():
     """Initialize Augeas."""
-    global aug
     aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                         augeas.Augeas.NO_MODL_AUTOLOAD)
     aug.set('/augeas/load/Pagekite/lens', 'Pagekite.lns')
     aug.set('/augeas/load/Pagekite/incl[last() + 1]', '/etc/pagekite.d/*.rc')
     aug.load()
-
+    return aug
-
-def main():
-    """Parse arguments and perform all duties"""
-    augeas_load()
-
-    arguments = parse_arguments()
-
-    subcommand = arguments.subcommand.replace('-', '_')
-    subcommand_method = globals()['subcommand_' + subcommand]
-    subcommand_method(arguments)
-
-
-if __name__ == "__main__":
-    main()

plinth/modules/pagekite/utils.py

@@ -1,12 +1,11 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
+"""Utilities for configuring Pagekite."""
 
-import json
 import logging
 import os
 
 from django.utils.translation import gettext_lazy as _
 
-from plinth import actions
 from plinth import app as app_module
 from plinth.signals import domain_added, domain_removed
 
@@ -80,21 +79,6 @@ PREDEFINED_SERVICES = {
 }
 
 
-def get_config():
-    """Return the current PageKite configuration."""
-    return json.loads(run(['get-config']))
-
-
-def run(arguments, superuser=True, input=None):
-    """Run a given command and raise exception if there was an error"""
-    command = 'pagekite'
-
-    if superuser:
-        return actions.superuser_run(command, arguments, input=input)
-    else:
-        return actions.run(command, arguments, input=input)
-
-
 def convert_service_to_string(service):
     """ Convert service dict into a ":"-separated parameter string
 
@@ -110,15 +94,14 @@ def convert_service_to_string(service):
     return service_string
 
 
-def load_service(json_service):
+def load_service(service):
-    """ create a service out of json command-line argument
+    """Create a service out of json command-line argument.
 
     1) parse json
     2) only use the parameters that we need (SERVICE_PARAMS)
     3) convert unicode to strings
     """
-    service = json.loads(json_service)
+    return {str(key): str(service[key]) for key in SERVICE_PARAMS}
-    return dict((str(key), str(service[key])) for key in SERVICE_PARAMS)
 
 
 def get_augeas_servicefile_path(protocol):
@@ -176,7 +159,8 @@ def update_names_module(is_enabled=None):
     if is_enabled is None and not app_module.App.get('pagekite').is_enabled():
         return
 
-    config = get_config()
+    from . import privileged
+    config = privileged.get_config()
     enabled_services = [
         service for service, value in config['predefined_services'].items()
         if value
@@ -186,8 +170,3 @@ def update_names_module(is_enabled=None):
                                  domain_type='domain-type-pagekite',
                                  name=config['kite_name'],
                                  services=enabled_services)
-
-
-if __name__ == "__main__":
-    import doctest
-    doctest.testmod()

plinth/modules/pagekite/views.py

@@ -8,7 +8,7 @@ from django.views.generic.edit import FormView
 
 from plinth.views import AppView
 
-from . import utils
+from . import privileged
 from .forms import (AddCustomServiceForm, ConfigurationForm,
                     DeleteCustomServiceForm)
 
@@ -50,7 +50,7 @@ class ConfigurationView(AppView):
     def __init__(self, *args, **kwargs):
         """Load and store the current configuration."""
         super().__init__(*args, **kwargs)
-        self.config = utils.get_config()
+        self.config = privileged.get_config()
         self.initial = self.config
 
     def get_context_data(self, *args, **kwargs):