Add Key.assert_id function

src/borg/key.py

@@ -108,6 +108,12 @@ class KeyBase:
     def decrypt(self, id, data, decompress=True):
         pass
 
+    def assert_id(self, id, data):
+        if id:
+            id_computed = self.id_hash(data)
+            if not compare_digest(id_computed, id):
+                raise IntegrityError('Chunk id verification failed')
+
 
 class PlaintextKey(KeyBase):
     TYPE = 0x02
@@ -137,8 +143,7 @@ class PlaintextKey(KeyBase):
         if not decompress:
             return Chunk(payload)
         data = self.compressor.decompress(payload)
-        if id and sha256(data).digest() != id:
-            raise IntegrityError('Chunk id verification failed')
+        self.assert_id(id, data)
         return Chunk(data)
 
 
@@ -183,11 +188,7 @@ class AESKeyBase(KeyBase):
         if not decompress:
             return Chunk(payload)
         data = self.compressor.decompress(payload)
-        if id:
-            hmac_given = id
-            hmac_computed = hmac_sha256(self.id_key, data)
-            if not compare_digest(hmac_computed, hmac_given):
-                raise IntegrityError('Chunk id verification failed')
+        self.assert_id(id, data)
         return Chunk(data)
 
     def extract_nonce(self, payload):

src/borg/testsuite/key.py

@@ -169,6 +169,18 @@ class TestKey:
         assert key.decrypt(None, encrypted, decompress=False) != plaintext
         assert key.decrypt(None, encrypted) == plaintext
 
+    def test_assert_id(self, key):
+        plaintext = b'123456789'
+        id = key.id_hash(plaintext)
+        key.assert_id(id, plaintext)
+        id_changed = bytearray(id)
+        id_changed[0] += 1
+        with pytest.raises(IntegrityError):
+            key.assert_id(id_changed, plaintext)
+        plaintext_changed = plaintext + b'1'
+        with pytest.raises(IntegrityError):
+            key.assert_id(id, plaintext_changed)
+
 
 class TestPassphrase:
     def test_passphrase_new_verification(self, capsys, monkeypatch):