cache: Warn user before accessing relocated repositories

This also closes #225

CHANGES

@@ -7,6 +7,7 @@ Version 0.15
 ------------
 
 (feature release, released on X)
+- Require approval before accessing relocated/moved repository (#271)
 - Require approval before accessing previously unknown unencrypted repositories (#271)
 - Fix issue with hash index files larger than 2GB.
 - Fix Python 3.2 compatibility issue with noatime open() (#164)

attic/cache.py

@@ -21,6 +21,8 @@ class Cache(object):
     class CacheInitAbortedError(Error):
         """Cache initialization aborted"""
 
+    class RepositoryAccessAborted(Error):
+        """Repository access aborted"""
 
     class EncryptionMethodMismatch(Error):
         """Repository encryption method changed since last acccess, refusing to continue
@@ -34,15 +36,20 @@ class Cache(object):
         self.key = key
         self.manifest = manifest
         self.path = path or os.path.join(get_cache_dir(), hexlify(repository.id).decode('ascii'))
+        # Warn user before sending data to a never seen before unencrypted repository
         if not os.path.exists(self.path):
             if warn_if_unencrypted and isinstance(key, PlaintextKey):
-                if 'ATTIC_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK' not in os.environ:
-                    print("""Warning: Attempting to access a previously unknown unencrypted repository\n""", file=sys.stderr)
-                    answer = input('Do you want to continue? [yN] ')
-                    if not (answer and answer in 'Yy'):
-                        raise self.CacheInitAbortedError()
+                if not self._confirm('Warning: Attempting to access a previously unknown unencrypted repository',
+                                     'ATTIC_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK'):
+                    raise self.CacheInitAbortedError()
             self.create()
         self.open()
+        # Warn user before sending data to a relocated repository
+        if self.previous_location and self.previous_location != repository._location.canonical_path():
+            msg = 'Warning: The repository at location {} was previously located at {}'.format(repository._location.canonical_path(), self.previous_location)
+            if not self._confirm(msg, 'ATTIC_RELOCATED_REPO_ACCESS_IS_OK'):
+                raise self.RepositoryAccessAborted()
+
         if sync and self.manifest.id != self.manifest_id:
             # If repository is older than the cache something fishy is going on
             if self.timestamp and self.timestamp > manifest.timestamp:
@@ -56,6 +63,16 @@ class Cache(object):
     def __del__(self):
         self.close()
 
+    def _confirm(self, message, env_var_override=None):
+        print(message, file=sys.stderr)
+        if env_var_override and os.environ.get(env_var_override):
+            print("Yes (From {})".format(env_var_override))
+            return True
+        if sys.stdin.isatty():
+            return False
+        answer = input('Do you want to continue? [yN] ')
+        return answer and answer in 'Yy'
+
     def create(self):
         """Create a new empty cache at `path`
         """
@@ -86,12 +103,14 @@ class Cache(object):
         self.manifest_id = unhexlify(self.config.get('cache', 'manifest'))
         self.timestamp = self.config.get('cache', 'timestamp', fallback=None)
         self.key_type = self.config.get('cache', 'key_type', fallback=None)
+        self.previous_location = self.config.get('cache', 'previous_location', fallback=None)
         self.chunks = ChunkIndex.read(os.path.join(self.path, 'chunks').encode('utf-8'))
         self.files = None
 
     def close(self):
         if self.lock:
             self.lock.release()
+            self.lock = None
 
     def _read_files(self):
         self.files = {}
@@ -134,6 +153,7 @@ class Cache(object):
         self.config.set('cache', 'manifest', hexlify(self.manifest.id).decode('ascii'))
         self.config.set('cache', 'timestamp', self.manifest.timestamp)
         self.config.set('cache', 'key_type', str(self.key.TYPE))
+        self.config.set('cache', 'previous_location', self.repository._location.canonical_path())
         with open(os.path.join(self.path, 'config'), 'w') as fd:
             self.config.write(fd)
         self.chunks.write(os.path.join(self.path, 'chunks').encode('utf-8'))

attic/helpers.py

@@ -457,6 +457,21 @@ class Location:
     def __repr__(self):
         return "Location(%s)" % self
 
+    def canonical_path(self):
+        if self.proto == 'file':
+            return self.path
+        else:
+            if self.path and self.path.startswith('~'):
+                path = '/' + self.path
+            elif self.path and not self.path.startswith('/'):
+                path = '/~/' + self.path
+            else:
+                path = self.path
+            return 'ssh://{}{}{}{}'.format('{}@'.format(self.user) if self.user else '',
+                                                        self.host,
+                                                        ':{}'.format(self.port) if self.port else '',
+                                                        path)
+
 
 def location_validator(archive=None):
     def validator(text):

attic/testsuite/archiver.py

@@ -209,6 +209,16 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         self.assert_equal(repository_id, self._extract_repository_id(self.repository_path))
         self.assert_raises(Cache.EncryptionMethodMismatch, lambda :self.attic('create', self.repository_location + '::test.2', 'input'))
 
+    def test_repository_swap_detection2(self):
+        self.create_test_files()
+        self.attic('init', '--encryption=none', self.repository_location + '_unencrypted')
+        os.environ['ATTIC_PASSPHRASE'] = 'passphrase'
+        self.attic('init', '--encryption=passphrase', self.repository_location + '_encrypted')
+        self.attic('create', self.repository_location + '_encrypted::test', 'input')
+        shutil.rmtree(self.repository_path + '_encrypted')
+        os.rename(self.repository_path + '_unencrypted', self.repository_path + '_encrypted')
+        self.assert_raises(Cache.RepositoryAccessAborted, lambda :self.attic('create', self.repository_location + '_encrypted::test.2', 'input'))
+
     def test_strip_components(self):
         self.attic('init', self.repository_location)
         self.create_regular_file('dir/file')

attic/testsuite/helpers.py

@@ -51,6 +51,14 @@ class LocationTestCase(AtticTestCase):
         )
         self.assert_raises(ValueError, lambda: Location('ssh://localhost:22/path:archive'))
 
+    def test_canonical_path(self):
+        locations = ['some/path::archive', 'file://some/path::archive', 'host:some/path::archive',
+                     'host:~user/some/path::archive', 'ssh://host/some/path::archive',
+                     'ssh://user@host:1234/some/path::archive']
+        for location in locations:
+            self.assert_equal(Location(location).canonical_path(),
+                              Location(Location(location).canonical_path()).canonical_path())
+
 
 class FormatTimedeltaTestCase(AtticTestCase):