Add minimal version of in repository mandatory feature flags.

This should allow us to make sure older borg versions can be cleanly
prevented from doing operations that are no longer safe because of
repository format evolution. This allows more fine grained control than
just incrementing the manifest version. So for example a change that
still allows new archives to be created but would corrupt the repository
when an old version tries to delete an archive or check the repository
would add the new feature to the check and delete set but leave it out
of the write set.

This is somewhat inspired by ext{2,3,4} which uses sets for
compat (everything except fsck), ro-compat (may only be accessed
read-only by older versions) and features (refuse all access).

src/borg/archive.py

@@ -1152,7 +1152,7 @@ class ArchiveChecker:
             self.manifest = self.rebuild_manifest()
         else:
             try:
-                self.manifest, _ = Manifest.load(repository, key=self.key)
+                self.manifest, _ = Manifest.load(repository, (Manifest.Operation.CHECK,), key=self.key)
             except IntegrityError as exc:
                 logger.error('Repository manifest is corrupted: %s', exc)
                 self.error_found = True

src/borg/archiver.py

@@ -87,8 +87,9 @@ def argument(args, str_or_bool):
     return str_or_bool
 
 
-def with_repository(fake=False, invert_fake=False, create=False, lock=True, exclusive=False, manifest=True, cache=False,
-                    secure=True):
+def with_repository(fake=False, invert_fake=False, create=False, lock=True,
+                    exclusive=False, manifest=True, cache=False, secure=True,
+                    compatibility=None):
     """
     Method decorator for subcommand-handling methods: do_XYZ(self, args, repository, …)
 
@@ -100,7 +101,20 @@ def with_repository(fake=False, invert_fake=False, create=False, lock=True, excl
     :param manifest: load manifest and key, pass them as keyword arguments
     :param cache: open cache, pass it as keyword argument (implies manifest)
     :param secure: do assert_secure after loading manifest
+    :param compatibility: mandatory if not create and (manifest or cache), specifies mandatory feature categories to check
     """
+
+    if not create and (manifest or cache):
+        if compatibility is None:
+            raise AssertionError("with_repository decorator used without compatibility argument")
+        if type(compatibility) is not tuple:
+            raise AssertionError("with_repository decorator compatibility argument must be of type tuple")
+    else:
+        if compatibility is not None:
+            raise AssertionError("with_repository called with compatibility argument but would not check" + repr(compatibility))
+        if create:
+            compatibility = Manifest.NO_OPERATION_CHECK
+
     def decorator(method):
         @functools.wraps(method)
         def wrapper(self, args, **kwargs):
@@ -117,7 +131,7 @@ def with_repository(fake=False, invert_fake=False, create=False, lock=True, excl
                                         append_only=append_only)
             with repository:
                 if manifest or cache:
-                    kwargs['manifest'], kwargs['key'] = Manifest.load(repository)
+                    kwargs['manifest'], kwargs['key'] = Manifest.load(repository, compatibility)
                     if 'compression' in args:
                         kwargs['key'].compressor = args.compression.compressor
                     if secure:
@@ -276,7 +290,7 @@ class Archiver:
             return EXIT_WARNING
         return EXIT_SUCCESS
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.CHECK,))
     def do_change_passphrase(self, args, repository, manifest, key):
         """Change repository key file passphrase"""
         if not hasattr(key, 'change_passphrase'):
@@ -413,7 +427,7 @@ class Archiver:
             print(fmt % ('U', msg, total_size_MB / dt_update, count, file_size_formatted, content, dt_update))
             print(fmt % ('D', msg, total_size_MB / dt_delete, count, file_size_formatted, content, dt_delete))
 
-    @with_repository(fake='dry_run', exclusive=True)
+    @with_repository(fake='dry_run', exclusive=True, compatibility=(Manifest.Operation.WRITE,))
     def do_create(self, args, repository, manifest=None, key=None):
         """Create new archive"""
         matcher = PatternMatcher(fallback=True)
@@ -632,7 +646,7 @@ class Archiver:
                 return matched
         return item_filter
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.READ,))
     @with_archive
     def do_extract(self, args, repository, manifest, key, archive):
         """Extract archive contents"""
@@ -714,7 +728,7 @@ class Archiver:
             pi.finish()
         return self.exit_code
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.READ,))
     @with_archive
     def do_export_tar(self, args, repository, manifest, key, archive):
         """Export archive contents as a tarball"""
@@ -927,7 +941,7 @@ class Archiver:
             self.print_warning("Include pattern '%s' never matched.", pattern)
         return self.exit_code
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.READ,))
     @with_archive
     def do_diff(self, args, repository, manifest, key, archive):
         """Diff contents of two archives"""
@@ -1140,7 +1154,7 @@ class Archiver:
 
         return self.exit_code
 
-    @with_repository(exclusive=True, cache=True)
+    @with_repository(exclusive=True, cache=True, compatibility=(Manifest.Operation.CHECK,))
     @with_archive
     def do_rename(self, args, repository, manifest, key, cache, archive):
         """Rename an existing archive"""
@@ -1161,7 +1175,7 @@ class Archiver:
 
     def _delete_archives(self, args, repository):
         """Delete archives"""
-        manifest, key = Manifest.load(repository)
+        manifest, key = Manifest.load(repository, (Manifest.Operation.DELETE,))
 
         if args.location.archive:
             archive_names = (args.location.archive,)
@@ -1219,7 +1233,7 @@ class Archiver:
         if not args.cache_only:
             msg = []
             try:
-                manifest, key = Manifest.load(repository)
+                manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             except NoManifestError:
                 msg.append("You requested to completely DELETE the repository *including* all archives it may "
                            "contain.")
@@ -1258,7 +1272,7 @@ class Archiver:
 
         return self._do_mount(args)
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.READ,))
     def _do_mount(self, args, repository, manifest, key):
         from .fuse import FuseOperations
 
@@ -1276,7 +1290,7 @@ class Archiver:
         """un-mount the FUSE filesystem"""
         return umount(args.mountpoint)
 
-    @with_repository()
+    @with_repository(compatibility=(Manifest.Operation.READ,))
     def do_list(self, args, repository, manifest, key):
         """List archive or repository contents"""
         if not hasattr(sys.stdout, 'buffer'):
@@ -1348,7 +1362,7 @@ class Archiver:
 
         return self.exit_code
 
-    @with_repository(cache=True)
+    @with_repository(cache=True, compatibility=(Manifest.Operation.READ,))
     def do_info(self, args, repository, manifest, key, cache):
         """Show archive details such as disk space used"""
         if any((args.location.archive, args.first, args.last, args.prefix)):
@@ -1438,7 +1452,7 @@ class Archiver:
             print(str(cache))
         return self.exit_code
 
-    @with_repository(exclusive=True)
+    @with_repository(exclusive=True, compatibility=(Manifest.Operation.DELETE,))
     def do_prune(self, args, repository, manifest, key):
         """Prune repository archives according to specified rules"""
         if not any((args.secondly, args.minutely, args.hourly, args.daily,
@@ -1517,7 +1531,7 @@ class Archiver:
     def do_upgrade(self, args, repository, manifest=None, key=None):
         """upgrade a repository from a previous version"""
         if args.tam:
-            manifest, key = Manifest.load(repository, force_tam_not_required=args.force)
+            manifest, key = Manifest.load(repository, (Manifest.Operation.CHECK,), force_tam_not_required=args.force)
 
             if not hasattr(key, 'change_passphrase'):
                 print('This repository is not encrypted, cannot enable TAM.')
@@ -1542,7 +1556,7 @@ class Archiver:
                 open(tam_file, 'w').close()
                 print('Updated security database')
         elif args.disable_tam:
-            manifest, key = Manifest.load(repository, force_tam_not_required=True)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK, force_tam_not_required=True)
             if tam_required(repository):
                 os.unlink(tam_required_file(repository))
             if key.tam_required:
@@ -1570,7 +1584,7 @@ class Archiver:
                 print("warning: %s" % e)
         return self.exit_code
 
-    @with_repository(cache=True, exclusive=True)
+    @with_repository(cache=True, exclusive=True, compatibility=(Manifest.Operation.CHECK,))
     def do_recreate(self, args, repository, manifest, key, cache):
         """Re-create archives"""
         msg = ("recreate is an experimental feature.\n"
@@ -1647,7 +1661,7 @@ class Archiver:
         print('Process ID:', get_process_id())
         return EXIT_SUCCESS
 
-    @with_repository()
+    @with_repository(compatibility=Manifest.NO_OPERATION_CHECK)
     def do_debug_dump_archive_items(self, args, repository, manifest, key):
         """dump (decrypted, decompressed) archive items metadata (not: data)"""
         archive = Archive(repository, key, manifest, args.location.archive,
@@ -1661,7 +1675,7 @@ class Archiver:
         print('Done.')
         return EXIT_SUCCESS
 
-    @with_repository()
+    @with_repository(compatibility=Manifest.NO_OPERATION_CHECK)
     def do_debug_dump_archive(self, args, repository, manifest, key):
         """dump decoded archive metadata (not: data)"""
 
@@ -1714,7 +1728,7 @@ class Archiver:
                 output(fd)
         return EXIT_SUCCESS
 
-    @with_repository()
+    @with_repository(compatibility=Manifest.NO_OPERATION_CHECK)
     def do_debug_dump_manifest(self, args, repository, manifest, key):
         """dump decoded repository manifest"""
 
@@ -1729,7 +1743,7 @@ class Archiver:
                 json.dump(meta, fd, indent=4)
         return EXIT_SUCCESS
 
-    @with_repository()
+    @with_repository(compatibility=Manifest.NO_OPERATION_CHECK)
     def do_debug_dump_repo_objs(self, args, repository, manifest, key):
         """dump (decrypted, decompressed) repo objects"""
         marker = None
@@ -1803,7 +1817,7 @@ class Archiver:
         print('Done.')
         return EXIT_SUCCESS
 
-    @with_repository(manifest=False, exclusive=True, cache=True)
+    @with_repository(manifest=False, exclusive=True, cache=True, compatibility=Manifest.NO_OPERATION_CHECK)
     def do_debug_refcount_obj(self, args, repository, manifest, key, cache):
         """display refcounts for the objects with the given IDs"""
         for hex_id in args.ids:

src/borg/helpers.py

@@ -1,6 +1,7 @@
 import argparse
 import contextlib
 import collections
+import enum
 import grp
 import hashlib
 import logging
@@ -123,6 +124,10 @@ def check_python():
         raise PythonLibcTooOld
 
 
+class MandatoryFeatureUnsupported(Error):
+    """Unsupported repository feature(s) {}. A newer version of borg is required to access this repository."""
+
+
 def check_extension_modules():
     from . import platform, compress, item
     if hashindex.API_VERSION != '1.1_01':
@@ -222,6 +227,34 @@ class Archives(abc.MutableMapping):
 
 class Manifest:
 
+    @enum.unique
+    class Operation(enum.Enum):
+        # The comments here only roughly describe the scope of each feature. In the end, additions need to be
+        # based on potential problems older clients could produce when accessing newer repositories and the
+        # tradeofs of locking version out or still allowing access. As all older versions and their exact
+        # behaviours are known when introducing new features sometimes this might not match the general descriptions
+        # below.
+
+        # The READ operation describes which features are needed to safely list and extract the archives in the
+        # repository.
+        READ = 'read'
+        # The CHECK operation is for all operations that need either to understand every detail
+        # of the repository (for consistency checks and repairs) or are seldom used functions that just
+        # should use the most restrictive feature set because more fine grained compatibility tracking is
+        # not needed.
+        CHECK = 'check'
+        # The WRITE operation is for adding archives. Features here ensure that older clients don't add archives
+        # in an old format, or is used to lock out clients that for other reasons can no longer safely add new
+        # archives.
+        WRITE = 'write'
+        # The DELETE operation is for all operations (like archive deletion) that need a 100% correct reference
+        # count and the need to be able to find all (directly and indirectly) referenced chunks of a given archive.
+        DELETE = 'delete'
+
+    NO_OPERATION_CHECK = tuple()
+
+    SUPPORTED_REPO_FEATURES = frozenset([])
+
     MANIFEST_ID = b'\0' * 32
 
     def __init__(self, key, repository, item_keys=None):
@@ -242,7 +275,7 @@ class Manifest:
         return datetime.strptime(self.timestamp, "%Y-%m-%dT%H:%M:%S.%f")
 
     @classmethod
-    def load(cls, repository, key=None, force_tam_not_required=False):
+    def load(cls, repository, operations, key=None, force_tam_not_required=False):
         from .item import ManifestItem
         from .crypto.key import key_factory, tam_required_file, tam_required
         from .repository import Repository
@@ -275,8 +308,23 @@ class Manifest:
             if not manifest_required and security_required:
                 logger.debug('Manifest is TAM verified and says TAM is *not* required, updating security database...')
                 os.unlink(tam_required_file(repository))
+        manifest.check_repository_compatibility(operations)
         return manifest, key
 
+    def check_repository_compatibility(self, operations):
+        for operation in operations:
+            assert isinstance(operation, self.Operation)
+            feature_flags = self.config.get(b'feature_flags', None)
+            if feature_flags is None:
+                return
+            if operation.value.encode() not in feature_flags:
+                continue
+            requirements = feature_flags[operation.value.encode()]
+            if b'mandatory' in requirements:
+                unsupported = set(requirements[b'mandatory']) - self.SUPPORTED_REPO_FEATURES
+                if unsupported:
+                    raise MandatoryFeatureUnsupported([f.decode() for f in unsupported])
+
     def write(self):
         from .item import ManifestItem
         if self.key.tam_required:

src/borg/testsuite/archiver.py

@@ -289,7 +289,7 @@ class ArchiverTestCaseBase(BaseTestCase):
     def open_archive(self, name):
         repository = Repository(self.repository_path, exclusive=True)
         with repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             archive = Archive(repository, key, manifest, name)
         return archive, repository
 
@@ -1248,7 +1248,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('extract', '--dry-run', self.repository_location + '::test.4')
         # Make sure both archives have been renamed
         with Repository(self.repository_path) as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
         self.assert_equal(len(manifest.archives), 2)
         self.assert_in('test.3', manifest.archives)
         self.assert_in('test.4', manifest.archives)
@@ -1349,7 +1349,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('init', '--encryption=none', self.repository_location)
         self.create_src_archive('test')
         with Repository(self.repository_path, exclusive=True) as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             archive = Archive(repository, key, manifest, 'test')
             for item in archive.iter_items():
                 if 'chunks' in item:
@@ -1367,7 +1367,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('init', '--encryption=none', self.repository_location)
         self.create_src_archive('test')
         with Repository(self.repository_path, exclusive=True) as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             archive = Archive(repository, key, manifest, 'test')
             id = archive.metadata.items[0]
             repository.put(id, b'corrupted items metadata stream chunk')
@@ -1417,7 +1417,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('create', '--dry-run', self.repository_location + '::test', 'input')
         # Make sure no archive has been created
         with Repository(self.repository_path) as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
         self.assert_equal(len(manifest.archives), 0)
 
     def test_progress_on(self):
@@ -2091,7 +2091,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('check', self.repository_location)
         # Then check that the cache on disk matches exactly what's in the repo.
         with self.open_repository() as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             with Cache(repository, key, manifest, sync=False) as cache:
                 original_chunks = cache.chunks
             cache.destroy(repository)
@@ -2112,7 +2112,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.cmd('init', '--encryption=repokey', self.repository_location)
         self.cmd('create', self.repository_location + '::test', 'input')
         with self.open_repository() as repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             with Cache(repository, key, manifest, sync=False) as cache:
                 cache.begin_txn()
                 cache.chunks.incref(list(cache.chunks.iteritems())[0][0])
@@ -2750,7 +2750,7 @@ class ArchiverCheckTestCase(ArchiverTestCaseBase):
 
         archive, repository = self.open_archive('archive1')
         with repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             with Cache(repository, key, manifest) as cache:
                 archive = Archive(repository, key, manifest, '0.13', cache=cache, create=True)
                 archive.items_buffer.add(Attic013Item)
@@ -2762,7 +2762,7 @@ class ArchiverCheckTestCase(ArchiverTestCaseBase):
 class ManifestAuthenticationTest(ArchiverTestCaseBase):
     def spoof_manifest(self, repository):
         with repository:
-            _, key = Manifest.load(repository)
+            _, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             repository.put(Manifest.MANIFEST_ID, key.encrypt(msgpack.packb({
                 'version': 1,
                 'archives': {},
@@ -2775,7 +2775,7 @@ class ManifestAuthenticationTest(ArchiverTestCaseBase):
         self.cmd('init', '--encryption=repokey', self.repository_location)
         repository = Repository(self.repository_path, exclusive=True)
         with repository:
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             repository.put(Manifest.MANIFEST_ID, key.encrypt(msgpack.packb({
                 'version': 1,
                 'archives': {},
@@ -2792,7 +2792,7 @@ class ManifestAuthenticationTest(ArchiverTestCaseBase):
         repository = Repository(self.repository_path, exclusive=True)
         with repository:
             shutil.rmtree(get_security_dir(bin_to_hex(repository.id)))
-            _, key = Manifest.load(repository)
+            _, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             key.tam_required = False
             key.change_passphrase(key._passphrase)