deluge: Don't use code execution for editing configuration

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: Veiko Aasa <veiko17@disroot.org>

actions/deluge

@@ -6,22 +6,17 @@ Configuration helper for BitTorrent web client.
 
 import argparse
 import json
-import os
-import shutil
+import pathlib
 import subprocess
 import time
 
 import augeas
-from plinth import action_utils
 
-try:
-    from deluge import config
-except ImportError:
-    # deluge is not installed or is python2 version
-    config = None
+from plinth import action_utils
+from plinth.modules.deluge.utils import Config
 
 DELUGED_DEFAULT_FILE = '/etc/default/deluged'
-DELUGE_CONF_DIR = '/var/lib/deluged/.config/deluge/'
+DELUGE_CONF_DIR = pathlib.Path('/var/lib/deluged/.config/deluge/')
 
 DELUGE_WEB_SYSTEMD_SERVICE_PATH = '/etc/systemd/system/deluge-web.service'
 DELUGE_WEB_SYSTEMD_SERVICE = '''
@@ -88,18 +83,8 @@ def _set_configuration(filename, parameter, value):
     if deluge_web_is_running:
         action_utils.service_stop('deluge-web')
 
-    filepath = os.path.join(DELUGE_CONF_DIR, filename)
-    if config is None:
-        script = 'from deluge import config;\
-            conf = config.Config(filename="{0}");\
-            conf["{1}"] = "{2}";\
-            conf.save()'.format(filepath, parameter, value)
-        subprocess.check_call(['python2', '-c', script])
-    else:
-        conf = config.Config(filename=filepath)
-        conf[parameter] = value
-        conf.save()
-    shutil.chown(filepath, 'debian-deluged', 'debian-deluged')
+    with Config(DELUGE_CONF_DIR / filename) as config:
+        config.content[parameter] = value
 
     if deluged_is_running:
         action_utils.service_start('deluged')
@@ -109,23 +94,18 @@ def _set_configuration(filename, parameter, value):
 
 def _get_host_id():
     """Get default host id."""
-    if config is None:
-        hosts_conf_file = os.path.join(DELUGE_CONF_DIR, 'hostlist.conf.1.2')
-        script = 'from deluge import config;\
-            conf = config.Config(filename="{0}");\
-            print(conf["hosts"][0][0])'.format(hosts_conf_file)
-        output = subprocess.check_output(['python2', '-c', script]).decode()
-        return output.strip()
-    else:
-        hosts_conf_file = os.path.join(DELUGE_CONF_DIR, 'hostlist.conf')
-        conf = config.Config(filename=hosts_conf_file)
-        return conf["hosts"][0][0]
+    try:
+        with Config(DELUGE_CONF_DIR / 'hostlist.conf') as config:
+            return config.content['hosts'][0][0]
+    except FileNotFoundError:
+        with Config(DELUGE_CONF_DIR / 'hostlist.conf.1.2') as config:
+            return config.content['hosts'][0][0]
 
 
 def _set_deluged_daemon_options():
     """Set deluged daemon options."""
-    aug = augeas.Augeas(
-        flags=augeas.Augeas.NO_LOAD + augeas.Augeas.NO_MODL_AUTOLOAD)
+    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
+                        augeas.Augeas.NO_MODL_AUTOLOAD)
     aug.set('/augeas/load/Shellvars/lens', 'Shellvars.lns')
     aug.set('/augeas/load/Shellvars/incl[last() + 1]', DELUGED_DEFAULT_FILE)
     aug.load()
@@ -138,16 +118,8 @@ def _set_deluged_daemon_options():
 
 def subcommand_get_configuration(_):
     """Return the current deluged configuration in JSON format."""
-    deluged_conf_file = os.path.join(DELUGE_CONF_DIR, 'core.conf')
-    if config is None:
-        script = 'from deluge import config;\
-            conf = config.Config(filename="{0}");\
-            print(conf["download_location"])'.format(deluged_conf_file)
-        output = subprocess.check_output(['python2', '-c', script]).decode()
-        download_location = output.strip()
-    else:
-        conf = config.Config(filename=deluged_conf_file)
-        download_location = conf["download_location"]
+    with Config(DELUGE_CONF_DIR / 'core.conf') as config:
+        download_location = config.content['download_location']
 
     print(json.dumps({'download_location': download_location}))
 
@@ -156,6 +128,7 @@ def subcommand_set_configuration(arguments):
     """Set the deluged configuration."""
     if arguments.parameter != 'download_location':
         return
+
     _set_configuration('core.conf', arguments.parameter, arguments.value)
 
 

plinth/modules/deluge/tests/test_utils.py

@@ -0,0 +1,69 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+Tests for utilities that edit Deluge configuration files.
+"""
+
+import pytest
+
+from plinth.modules.deluge.utils import Config
+
+test_content = '''{
+    "file": 3,
+    "format": 1
+}{
+    "hosts": [
+        [
+            "c582deb3aeac48e5ba6f629ec363ea68",
+            "127.0.0.1",
+            58846,
+            "localclient",
+            "aa1d33728187a2c2516e7363d6e8fd9178abb6aa"
+        ]
+    ]
+}'''
+
+
+@pytest.fixture(name='deluge_config')
+def fixture_deluge_config(tmp_path):
+    """Fixture to provide a test deluge configuration file."""
+    path = tmp_path / 'deluge_config'
+    path.write_text(test_content)
+    yield path
+
+
+def test_initialization(tmp_path):
+    """Test object initialization."""
+    test_file = tmp_path / 'test_file'
+    config = Config(str(test_file))
+    assert config.file_name == str(test_file)
+    assert config.file == test_file
+    assert config._version is None
+    assert config.content is None
+    assert config._original_content is None
+
+
+def test_load(deluge_config):
+    """Test loading the configuration file."""
+    with Config(str(deluge_config)) as config:
+        assert config._version['file'] == 3
+        assert config._version['format'] == 1
+        assert config.content['hosts'][0][1] == '127.0.0.1'
+
+
+def test_save(deluge_config):
+    """Test save the configuration file."""
+    with Config(str(deluge_config)) as config:
+        pass
+
+    assert deluge_config.read_text() == test_content
+
+    with Config(str(deluge_config)) as config:
+        config.content['hosts'][0][1] = '0.0.0.0'
+
+    assert deluge_config.read_text() == test_content.replace(
+        '127.0.0.1', '0.0.0.0')
+
+    with Config(str(deluge_config)) as config:
+        assert config.content['hosts'][0][1] == '0.0.0.0'
+
+    assert deluge_config.stat().st_mode & 0o777 == 0o600

plinth/modules/deluge/utils.py

@@ -0,0 +1,74 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+Utilities for editing configuration files of Deluge.
+"""
+
+import copy
+import json
+import os
+import pathlib
+import re
+import shutil
+import tempfile
+
+_JSON_FORMAT = {'indent': 4, 'sort_keys': True, 'ensure_ascii': False}
+
+
+class Config:
+    """Read or edit a Deluge configuration file."""
+    def __init__(self, file_name):
+        """Initialize the configuration object."""
+        self.file_name = file_name
+        self.file = pathlib.Path(self.file_name)
+        self._version = None
+        self.content = None
+        self._original_content = None
+
+    def load(self):
+        """Parse the configuration file into memory."""
+        text = self.file.read_text()
+        matches = re.match(r'^({[^}]*})(.*)$', text, re.DOTALL)
+        if not matches:
+            raise Exception('Unexpected file format.')
+
+        try:
+            self._version = json.loads(matches.group(1))
+            self.content = json.loads(matches.group(2))
+        except json.decoder.JSONDecoderError:
+            raise Exception('Unable to parse JSON in file.')
+
+        if self._version['format'] != 1:
+            raise Exception('Version of the config file not understood')
+
+        self._original_content = copy.deepcopy(self.content)
+
+    def save(self):
+        """Atomically save the modified configuration to file."""
+        if self.content == self._original_content:
+            return
+
+        with tempfile.NamedTemporaryFile(dir=self.file.parent,
+                                         delete=False) as new_file:
+            new_file.write(json.dumps(self._version, **_JSON_FORMAT).encode())
+            new_file.write(json.dumps(self.content, **_JSON_FORMAT).encode())
+            new_file.flush()
+            os.fsync(new_file.fileno())
+
+        new_file_path = pathlib.Path(new_file.name)
+        new_file_path.chmod(0o600)
+        try:
+            shutil.chown(str(new_file_path), 'debian-deluged',
+                         'debian-deluged')
+        except (PermissionError, LookupError):
+            pass  # Not running as root, or deluge is not installed
+
+        new_file_path.rename(self.file)
+
+    def __enter__(self):
+        """Enter the context."""
+        self.load()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Exit the context."""
+        self.save()