borg config: add some validation, fixes #3566

only a few basic checks, it does not prevent all stupidities.

src/borg/archiver.py

@@ -1523,6 +1523,43 @@ class Archiver:
     @with_repository(exclusive=True, cache=True, compatibility=(Manifest.Operation.WRITE,))
     def do_config(self, args, repository, manifest, key, cache):
         """get, set, and delete values in a repository or cache config file"""
+
+        def repo_validate(section, name, value=None, check_value=True):
+            if section not in ['repository', ]:
+                raise ValueError('Invalid section')
+            if name in ['segments_per_dir', 'max_segment_size', 'storage_quota', ]:
+                if check_value:
+                    try:
+                        int(value)
+                    except ValueError:
+                        raise ValueError('Invalid value') from None
+            elif name in ['additional_free_space', ]:
+                if check_value:
+                    try:
+                        parse_file_size(value)
+                    except ValueError:
+                        raise ValueError('Invalid value') from None
+            elif name in ['append_only', ]:
+                if check_value and value not in ['0', '1']:
+                    raise ValueError('Invalid value')
+            elif name in ['id', ]:
+                if check_value:
+                    try:
+                        bin_id = unhexlify(value)
+                    except:
+                        raise ValueError('Invalid value, must be 64 hex digits') from None
+                    if len(bin_id) != 32:
+                        raise ValueError('Invalid value, must be 64 hex digits')
+            else:
+                raise ValueError('Invalid name')
+
+        def cache_validate(section, name, value=None, check_value=True):
+            if section not in ['cache', ]:
+                raise ValueError('Invalid section')
+            # I looked at the cache config and did not see anything a user would want to edit,
+            # so, for now, raise for any key name
+            raise ValueError('Invalid name')
+
         try:
             section, name = args.name.split('.')
         except ValueError:
@@ -1533,16 +1570,20 @@ class Archiver:
             cache.cache_config.load()
             config = cache.cache_config._config
             save = cache.cache_config.save
+            validate = cache_validate
         else:
             config = repository.config
             save = lambda: repository.save_config(repository.path, repository.config)
+            validate = repo_validate
 
         if args.delete:
+            validate(section, name, check_value=False)
             config.remove_option(section, name)
             if len(config.options(section)) == 0:
                 config.remove_section(section)
             save()
         elif args.value:
+            validate(section, name, args.value)
             if section not in config.sections():
                 config.add_section(section)
             config.set(section, name, args.value)

src/borg/testsuite/archiver.py

@@ -2782,14 +2782,17 @@ id: 2 / e29442 3506da 4e1ea7 / 25f62a 5a3d41 - 02
         self.create_test_files()
         os.unlink('input/flagfile')
         self.cmd('init', '--encryption=repokey', self.repository_location)
-        for flags in [[], ['--cache']]:
-            for cfg_key in {'testkey', 'testsection.testkey'}:
-                self.cmd('config', self.repository_location, *flags, cfg_key, exit_code=1)
-                self.cmd('config', self.repository_location, *flags, cfg_key, 'testcontents')
-                output = self.cmd('config', self.repository_location, *flags, cfg_key)
-                assert output == 'testcontents\n'
-                self.cmd('config', self.repository_location, *flags, '--delete', cfg_key)
-                self.cmd('config', self.repository_location, *flags, cfg_key, exit_code=1)
+        for cfg_key, cfg_value in [
+            ('additional_free_space', '2G'),
+            ('repository.append_only', '1'),
+        ]:
+            output = self.cmd('config', self.repository_location, cfg_key)
+            assert output == '0' + '\n'
+            self.cmd('config', self.repository_location, cfg_key, cfg_value)
+            output = self.cmd('config', self.repository_location, cfg_key)
+            assert output == cfg_value + '\n'
+            self.cmd('config', self.repository_location, '--delete', cfg_key)
+            self.cmd('config', self.repository_location, cfg_key, exit_code=1)
 
     requires_gnutar = pytest.mark.skipif(not have_gnutar(), reason='GNU tar must be installed for this test.')
     requires_gzip = pytest.mark.skipif(not shutil.which('gzip'), reason='gzip must be installed for this test.')