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).

borg/archive.py

@@ -870,7 +870,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

borg/archiver.py

@@ -56,7 +56,8 @@ 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):
+def with_repository(fake=False, invert_fake=False, create=False, lock=True,
+                    exclusive=False, manifest=True, cache=False, compatibility=None):
     """
     Method decorator for subcommand-handling methods: do_XYZ(self, args, repository, …)
 
@@ -67,7 +68,20 @@ def with_repository(fake=False, invert_fake=False, create=False, lock=True, excl
     :param exclusive: (str or bool) lock repository exclusively (for writing)
     :param manifest: load manifest and key, pass them as keyword arguments
     :param cache: open cache, pass it as keyword argument (implies 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):
@@ -84,7 +98,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 cache:
                     with Cache(repository, kwargs['key'], kwargs['manifest'],
                                do_files=getattr(args, 'cache_files', False), lock_wait=self.lock_wait) as cache_:
@@ -176,7 +190,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'):
@@ -241,7 +255,7 @@ class Archiver:
         logger.info('Key updated')
         return EXIT_SUCCESS
 
-    @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)
@@ -424,7 +438,7 @@ class Archiver:
                 return matcher.match(item[b'path'])
         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"""
@@ -490,7 +504,7 @@ class Archiver:
                 self.print_warning("Include pattern '%s' never matched.", pattern)
         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"""
@@ -506,7 +520,7 @@ class Archiver:
         """Delete an existing repository or archive"""
         if args.location.archive:
             archive_name = args.location.archive
-            manifest, key = Manifest.load(repository)
+            manifest, key = Manifest.load(repository, (Manifest.Operation.DELETE,))
 
             if args.forced == 2:
                 try:
@@ -537,7 +551,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.")
                     msg.append("This repository seems to have no manifest, so we can't tell anything about its contents.")
@@ -573,7 +587,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
 
@@ -595,7 +609,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 args.location.archive:
@@ -681,7 +695,7 @@ class Archiver:
                     print(format_archive(archive_info))
         return self.exit_code
 
-    @with_repository(cache=True)
+    @with_repository(cache=True, compatibility=(Manifest.Operation.READ,))
     @with_archive
     def do_info(self, args, repository, manifest, key, archive, cache):
         """Show archive details such as disk space used"""
@@ -699,7 +713,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.hourly, args.daily,
@@ -759,7 +773,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.')
@@ -784,7 +798,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:
@@ -817,7 +831,7 @@ class Archiver:
         print(sysinfo())
         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)
@@ -830,7 +844,7 @@ class Archiver:
         print('Done.')
         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
@@ -904,7 +918,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:

borg/helpers.py

@@ -2,6 +2,7 @@ import argparse
 from binascii import hexlify
 from collections import namedtuple
 import contextlib
+import enum
 from functools import wraps
 import grp
 import os
@@ -110,6 +111,10 @@ class PlaceholderError(Error):
     """Formatting Error: "{}".format({}): {}({})"""
 
 
+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
     if hashindex.API_VERSION != '1.0_01':
@@ -126,6 +131,34 @@ def check_extension_modules():
 
 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):
@@ -139,7 +172,7 @@ class Manifest:
         self.timestamp = None
 
     @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 .key import key_factory, tam_required_file, tam_required
         from .repository import Repository
         from .archive import ITEM_KEYS
@@ -173,8 +206,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):
         if self.key.tam_required:
             self.config[b'tam_required'] = True

borg/testsuite/archiver.py

@@ -248,7 +248,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
 
@@ -815,7 +815,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)
@@ -853,7 +853,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:
@@ -870,7 +870,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[b'items'][0]
             repository.put(id, b'corrupted items metadata stream chunk')
@@ -915,7 +915,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(self):
@@ -1594,7 +1594,7 @@ class ArchiverCheckTestCase(ArchiverTestCaseBase):
         # This bug can still live on in Borg repositories (through borg upgrade).
         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({
@@ -1611,7 +1611,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': {},
@@ -1624,7 +1624,7 @@ class ManifestAuthenticationTest(ArchiverTestCaseBase):
         self.cmd('init', 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': {},
@@ -1641,7 +1641,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)