[R] Verify digest of file objects in mirror (#7069)

Author: nadove-ucsc

.mypy.ini

@@ -53,7 +53,8 @@ modules =
     azul.indexer.action_controller,
     azul.indexer.mirror_controller,
     azul.service.app_controller,
-    azul.indexer.mirror_service
+    azul.indexer.mirror_service,
+    azul.digests
 packages =
     azul.openapi
 
@@ -69,3 +70,6 @@ follow_untyped_imports = True
 
 [mypy-aws_requests_auth.boto_utils]
 follow_untyped_imports = True
+
+[mypy-resumablesha256.*]
+ignore_missing_imports = True

requirements.txt

@@ -17,6 +17,7 @@ jmespath==1.0.1
 more-itertools==10.7.0
 msgpack==1.1.0
 requests==2.32.3
+resumablesha256==1.0
 rsa==4.7.2  # < 4.8, see https://github.com/DataBiosphere/azul/issues/6774. Resolve build-time dependency ambguity
 setuptools==80.0.0  # Match requirements.pip.txt. Python 3.12 removed distutils, which we depended on transitively. Luckily, setuptools includes a vendored copy.
 urllib3==1.26.20 # < 2, see https://github.com/DataBiosphere/azul/issues/6777, match with types-urllib in requirements.dev.txt

scripts/mirror_file.py

@@ -15,6 +15,9 @@
 from azul.args import (
     AzulArgumentHelpFormatter,
 )
+from azul.digests import (
+    get_resumable_hasher,
+)
 from azul.http import (
     http_client,
 )
@@ -63,15 +66,17 @@ def mirror_file(catalog: CatalogName, file_uuid: str, part_size: int) -> str:
     file = get_file(catalog, file_uuid)
     service = MirrorService()
     upload_id = service.begin_mirroring_file(file)
+    digest_value, digest_type = file.digest()
+    hasher = get_resumable_hasher(digest_type)
 
     def mirror_parts():
         part = FilePart.first(file, part_size)
         while part is not None:
-            yield service.mirror_file_part(catalog, file, part, upload_id)
+            yield service.mirror_file_part(catalog, file, part, upload_id, hasher)
             part = part.next(file)
 
     etags = list(mirror_parts())
-    service.finish_mirroring_file(file, upload_id, etags=etags)
+    service.finish_mirroring_file(file, upload_id, etags=etags, hasher=hasher)
     return service.get_mirror_url(file)
 
 

src/azul/azulclient.py

@@ -48,6 +48,12 @@
 from azul.deployment import (
     aws,
 )
+from azul.digests import (
+    Hasher,
+    get_resumable_hasher,
+    hasher_from_str,
+    hasher_to_str,
+)
 from azul.es import (
     ESClientFactory,
 )
@@ -226,7 +232,8 @@ def mirror_part_message(self,
                             file: File,
                             part: FilePart,
                             upload_id: str,
-                            etags: Sequence[str]
+                            etags: Sequence[str],
+                            hasher: Hasher
                             ) -> SQSFifoMessage:
         return SQSFifoMessage(
             body={
@@ -235,7 +242,8 @@ def mirror_part_message(self,
                 'upload_id': upload_id,
                 'action': MirrorAction.mirror_part.to_json(),
                 'part': part.to_json(),
-                'etags': etags
+                'etags': etags,
+                'hasher': hasher_to_str(hasher)
             },
             group_id=self.mirror_service.mirror_object_key(file)
         )
@@ -244,15 +252,17 @@ def finalize_file_message(self,
                               catalog: CatalogName,
                               file: File,
                               upload_id: str,
-                              etags: Sequence[str]
+                              etags: Sequence[str],
+                              hasher: Hasher
                               ) -> SQSFifoMessage:
         return SQSFifoMessage(
             body={
                 'catalog': catalog,
                 'file': file.to_json(),
                 'upload_id': upload_id,
                 'action': MirrorAction.finalize_file.to_json(),
-                'etags': etags
+                'etags': etags,
+                'hasher': hasher_to_str(hasher)
             },
             group_id=self.mirror_service.mirror_object_key(file)
         )
@@ -729,49 +739,57 @@ def mirror_file(self,
                 log.info('Successfully mirrored file %r via standard upload', file.uuid)
             else:
                 log.info('Mirroring file %r via multi-part upload', file.uuid)
+                _, digest_type = file.digest()
+                hasher = get_resumable_hasher(digest_type)
                 upload_id = self.mirror_service.begin_mirroring_file(file)
                 first_part = FilePart.first(file, part_size)
-                etag = self.mirror_service.mirror_file_part(catalog, file, first_part, upload_id)
+                etag = self.mirror_service.mirror_file_part(catalog, file, first_part, upload_id, hasher)
                 next_part = first_part.next(file)
                 assert next_part is not None
-                messages = [self.mirror_part_message(catalog, file, next_part, upload_id, [etag])]
+                messages = [self.mirror_part_message(catalog, file, next_part, upload_id, [etag], hasher)]
                 self.queue_mirror_messages(messages)
 
     def mirror_file_part(self,
                          catalog: CatalogName,
                          file_json: JSON,
                          part_json: JSON,
                          upload_id: str,
-                         etags: Sequence[str]
+                         etags: Sequence[str],
+                         hasher_data: str
                          ):
         file = self.load_file(catalog, file_json)
         part = FilePart.from_json(part_json)
-        etag = self.mirror_service.mirror_file_part(catalog, file, part, upload_id)
+        hasher = hasher_from_str(hasher_data)
+        etag = self.mirror_service.mirror_file_part(catalog, file, part, upload_id, hasher)
         etags = [*etags, etag]
         next_part = part.next(file)
         if next_part is None:
             log.info('File %r fully uploaded in %d parts', file.uuid, len(etags))
             message = self.finalize_file_message(catalog,
                                                  file,
                                                  upload_id,
-                                                 etags)
+                                                 etags,
+                                                 hasher)
         else:
             message = self.mirror_part_message(catalog,
                                                file,
                                                next_part,
                                                upload_id,
-                                               etags)
+                                               etags,
+                                               hasher)
         self.queue_mirror_messages([message])
 
     def finalize_file(self,
                       catalog: CatalogName,
                       file_json: JSON,
                       upload_id: str,
-                      etags: Sequence[str]
+                      etags: Sequence[str],
+                      hasher_data: str
                       ):
         file = self.load_file(catalog, file_json)
         assert len(etags) > 0
-        self.mirror_service.finish_mirroring_file(file, upload_id, etags)
+        hasher = hasher_from_str(hasher_data)
+        self.mirror_service.finish_mirroring_file(file, upload_id, etags, hasher)
         log.info('Successfully mirrored file %r via multi-part upload', file.uuid)
 
     def load_file(self, catalog: CatalogName, file: JSON) -> File:

src/azul/digests.py

@@ -0,0 +1,27 @@
+import base64
+import pickle
+from typing import (
+    Any,
+)
+
+import resumablesha256
+
+from azul import (
+    R,
+)
+
+# TODO: add type stub
+Hasher = Any
+
+
+def get_resumable_hasher(digest_type: str) -> Hasher:
+    assert digest_type == 'sha256', R('Only sha256 is currently supported')
+    return resumablesha256.sha256()
+
+
+def hasher_to_str(hasher: Hasher) -> str:
+    return base64.b64encode(pickle.dumps(hasher)).decode('ascii')
+
+
+def hasher_from_str(s: str) -> Hasher:
+    return pickle.loads(base64.b64decode(s))

src/azul/indexer/mirror_controller.py

@@ -57,11 +57,13 @@ def _mirror(self, message: JSON):
                                          message['file'],
                                          message['part'],
                                          message['upload_id'],
-                                         message['etags'])
+                                         message['etags'],
+                                         message['hasher'])
         elif action is MirrorAction.finalize_file:
             self.client.finalize_file(message['catalog'],
                                       message['file'],
                                       message['upload_id'],
-                                      message['etags'])
+                                      message['etags'],
+                                      message['hasher'])
         else:
             assert False, action

src/azul/indexer/mirror_service.py

@@ -26,6 +26,10 @@
 from azul.deployment import (
     aws,
 )
+from azul.digests import (
+    Hasher,
+    get_resumable_hasher,
+)
 from azul.drs import (
     AccessMethod,
 )
@@ -137,6 +141,10 @@ def mirror_file(self, catalog: CatalogName, file: File):
             self._storage.put(object_key=self.mirror_object_key(file),
                               data=file_content,
                               content_type=file.content_type)
+            _, digest_type = file.digest()
+            hasher = get_resumable_hasher(digest_type)
+            hasher.update(file_content)
+            self._verify_digest(file, hasher)
             self.put_info(file)
 
     def begin_mirroring_file(self, file: File) -> str:
@@ -152,14 +160,17 @@ def mirror_file_part(self,
                          catalog: CatalogName,
                          file: File,
                          part: FilePart,
-                         upload_id: str
+                         upload_id: str,
+                         hasher: Hasher
                          ) -> str:
         """
         Upload a part of a file to a multipart upload begun with
         :meth:`begin_mirroring_file` and return the uploaded part's ETag.
+        The provided hasher is mutated to incorporated the part's content.
         """
         upload = self._get_upload(file, upload_id)
         file_content = self._download(catalog, file, part)
+        hasher.update(file_content)
         return self._storage.upload_multipart_part(file_content,
                                                    part.index + 1,
                                                    upload)
@@ -168,13 +179,14 @@ def finish_mirroring_file(self,
                               file: File,
                               upload_id: str,
                               etags: Sequence[str],
-                              hasher
+                              hasher: Hasher
                               ):
         """
         Complete a multipart upload begun with :meth:`begin_mirroring_file`.
         """
         upload = self._get_upload(file, upload_id)
         self._storage.complete_multipart_upload(upload, etags)
+        self._verify_digest(file, hasher)
         self._check_info(file)
         self.put_info(file)
 
@@ -255,3 +267,10 @@ def _download(self,
     def _get_upload(self, file: File, upload_id: str) -> 'MultipartUpload':
         return self._storage.load_multipart_upload(object_key=self.mirror_object_key(file),
                                                    upload_id=upload_id)
+
+    def _verify_digest(self, file: File, hasher: Hasher):
+        expected_digest_value, digest_type = file.digest()
+        actual_digest_value = hasher.hexdigest()
+        assert expected_digest_value == actual_digest_value, R(
+            'File digest value does not match its contents',
+            file.uuid, digest_type, expected_digest_value, actual_digest_value)