Implement AFF4-L container hashes

requirements/base.txt

@@ -1,5 +1,6 @@
+setuptools<82
 PySide6==6.10.*
 PyInstaller==6.16.*
-pyaff4 @ git+https://github.com/fservida/pyaff4@21e7eb8720907d5af2cf687dec768326eee2f210  # Use custom version of pyaff4 to support large files in zipfile as well as fixing build process.
+pyaff4 @ git+https://github.com/fservida/pyaff4@b28f5a2ccd3504145e10a0d4901cd2daf2211fb4  # Use custom version of pyaff4 to support large files in zipfile as well as fixing build process.
 puremagic==1.30
 pillow==12.0.*

src/main/python/gemino/threads/copy/logical/copy.py

@@ -482,6 +482,7 @@ def copy_aff4(self, src: str, destinations: list, hashes: list):
 
         files_hashes = {}  # {filepath: {hash_name:hash_value, ...}, ...}
         files_metadata = {}  # {filepath: metadata, ...}
+        container_hashes = {}
 
         start_time = self.initialize_log_files(destinations, base_path, src)
 
@@ -760,15 +761,44 @@ def copy_aff4(self, src: str, destinations: list, hashes: list):
             try:
                 report_file_path = path.join(dst, f"{base_path}_copy_report.txt")
                 with open(report_file_path, "a", encoding="utf-8") as report_file:
-                    report_file.write("\n")
-                    report_file.write(
-                        f"################## Verification Report ######################\n"
-                    )
                     with container.Container.openURNtoContainer(
                         rdfvalue.URN.FromFileName(container_path)
                     ) as volume:
                         resolver = volume.resolver
                         verification_listener = LinearVerificationListener(volume.urn)
+                        container_hashes = resolver.read_metadata_hashes(
+                            volume.zip_file
+                        )
+                        report_file.write("\n")
+                        report_file.write(
+                            f"################## Container Metadata Hashes ######################\n"
+                        )
+                        for algo, hash_value in container_hashes.items():
+                            report_file.write(f"- {algo.upper()}: {hash_value}\n")
+                        report_file.write("\n")
+                        report_file.write(
+                            f"################## Verification Report ######################\n"
+                        )
+                        metadata_verified: bool
+                        metadata_hashes: list[dict[str, str | bool]]
+                        metadata_verified, metadata_hashes = (
+                            resolver.verify_container_metadata_integrity(
+                                volume.zip_file
+                            )
+                        )
+                        if not metadata_verified:
+                            report_file.write(
+                                "Container Metadata Verification Failed:\n"
+                            )
+                            for hash_value in metadata_hashes:
+                                report_file.write(
+                                    f"- {hash_value['hash_type'].upper()} - {'VERIFIED' if hash_value['verified'] else 'FAILED'} | Stored: {hash_value['stored_hash']} - Calculated {hash_value['calculated_hash']}\n"
+                                )
+                        else:
+                            report_file.write(
+                                "Container Metadata Verification Successful\n"
+                            )
+
                         hasher = linear_hasher.LinearHasher2(
                             resolver, verification_listener
                         )
@@ -804,13 +834,14 @@ def copy_aff4(self, src: str, destinations: list, hashes: list):
                             hasher.hash(image, progress=progress_listener)
                             hashed_size += filesize
 
-                        if verification_listener.failed:
+                        if verification_listener.failed or not metadata_verified:
                             failed_files = len(verification_listener.failed)
                             progress[dst] = {
                                 "status": "error_hash",
                                 "processed_bytes": hashed_size,
                                 "processed_files": filecount,
                                 "current_file": "",
+                                "container_hashes": metadata_hashes,
                             }
                             for file in verification_listener.failed:
                                 report_file.write(
@@ -828,13 +859,14 @@ def copy_aff4(self, src: str, destinations: list, hashes: list):
                             )
                             self.copy_progress.emit(ProgressData(1, copy(progress)))
 
-                        if not verification_listener.failed:
+                        else:
                             # Signal the end with no errors of the hash verification for the current volume
                             progress[dst] = {
                                 "status": "done",
                                 "processed_bytes": hashed_size,
                                 "processed_files": filecount,
                                 "current_file": "",
+                                "container_hashes": metadata_hashes,
                             }
                             report_file.write(
                                 f"Verification successful for {filecount} files\n"
@@ -929,10 +961,42 @@ def verify_aff4(self, src: str):
         with container.Container.openURNtoContainer(
             rdfvalue.URN.FromFileName(src)
         ) as volume:
+            log: str = ""
             resolver = volume.resolver
+
+            # Read container metadata
+            container_hashes = resolver.read_metadata_hashes(volume.zip_file)
+            log += f"Container Metadata Hashes:\n"
+            if container_hashes:
+                for algo, hash_value in container_hashes.items():
+                    log += f"- {algo.upper()}: {hash_value}\n"
+            else:
+                log += f"- Container does not contain metadata hashes\n"
+            log += "\n"
+
+            # Verify container
+            log += f"################## Verification Report ######################\n"
+            # Verify container metadata
+            metadata_verified: bool
+            metadata_hashes: list[dict[str, str | bool]]
+            metadata_verified, metadata_hashes = (
+                resolver.verify_container_metadata_integrity(volume.zip_file)
+            )
+
+            if metadata_hashes:
+                if not metadata_verified:
+                    log += "Container Metadata Verification Failed:\n"
+                    for hash_value in metadata_hashes:
+                        log += f"- {hash_value['hash_type'].upper()} - {'VERIFIED' if hash_value['verified'] else 'FAILED'} | Stored: {hash_value['stored_hash']} - Calculated {hash_value['calculated_hash']}\n"
+                else:
+                    log += "Container Metadata Verification Successful\n"
+
+            # Verify container files
             verification_listener = LinearVerificationListener(volume.urn)
             hasher = linear_hasher.LinearHasher2(resolver, verification_listener)
 
+            hasher = linear_hasher.LinearHasher2(resolver, verification_listener)
+
             for image in volume.images():
                 # Each image is a file in the container.
                 # Update Byte Progress
@@ -964,9 +1028,10 @@ def verify_aff4(self, src: str):
                 hasher.hash(image, progress=progress_listener)
                 hashed_size += filesize
 
-            if verification_listener.failed:
+            if verification_listener.failed or not metadata_verified:
                 failed_files = len(verification_listener.failed)
-                log = f"Verification Failed for {failed_files} files:\n\n"
+                if failed_files:
+                    log += f"Verification Failed for {failed_files} files:\n\n"
                 for file in verification_listener.failed:
                     log += f"Verification failed for file: {trimVolume(volume.urn, file)}\n"
                     for hash_failed in verification_listener.failed[file]:
@@ -975,20 +1040,21 @@ def verify_aff4(self, src: str):
                     "status": "error_hash",
                     "processed_bytes": hashed_size,
                     "processed_files": filecount,
-                    "current_file": f"Verification Failed for {failed_files} files",
+                    "current_file": f"Verification Failed for {failed_files} files or container metadata (if existing).",
                     "log": log,
                 }
 
                 self.verify_progress.emit(ProgressData(4, copy(progress)))
 
-            if not verification_listener.failed:
+            else:
                 # Signal the end with no errors of the hash verification for the current volume
+                log += f"Verification successful for {filecount} files and container metadata (if existing).\n"
                 progress[src] = {
                     "status": "done",
                     "processed_bytes": hashed_size,
                     "processed_files": filecount,
                     "current_file": "",
-                    "log": f"Verification successful for {filecount} files\n",
+                    "log": log,
                 }
                 self.verify_progress.emit(ProgressData(4, copy(progress)))
 
@@ -1006,8 +1072,5 @@ def initialise_log_text(self):
         log += f"Container: {self.src}\n"
         log += f"Total Files: {self.total_files}\n"
         log += f"Size: {self.total_bytes} Bytes (~ {self.total_bytes / 10 ** 9} GB)\n"
-        log += f"\n"
-        log += f"\n"
-        log += f"################## Verification Report ######################\n"
 
         return log

src/main/python/gemino/vars.py

@@ -1 +1 @@
-VERSION = "2.10.0"
+VERSION = "2.11.0"

src/main/python/gemino/widgets/common/progress_window.py

@@ -194,6 +194,9 @@ def update_progress(self, progress: ProgressData):
                     volume_progress.processed_files = progress_status["processed_files"]
                     volume_progress.current_file = progress_status["current_file"]
                     volume_progress.status = progress_status["status"]
+                    volume_progress.container_hashes = progress_status.get(
+                        "container_hashes", []
+                    )
                     log = progress_status.get(
                         "log", None
                     )  # Support for ephemeral AFF4 verification log

src/main/python/gemino/widgets/common/volume_progress.py

@@ -29,6 +29,8 @@ def __init__(
         super().__init__()
 
         # Init Data
+        self.__container_hashes: list[dict[str, str | bool]] = []
+        self.__container_hashes_formatted: str = ""
         self.__status = "idle"
         self.__verified = 0
         self.__current_file = "-"
@@ -82,6 +84,9 @@ def __setup_ui(self):
         self.__show_log_button = QtWidgets.QPushButton("Show Log")
         self.__show_log_button.clicked.connect(self.__show_log)
 
+        self.__show_container_hashes_button = QtWidgets.QPushButton("Show Hashes")
+        self.__show_container_hashes_button.clicked.connect(self.__show_hashes)
+
         self.__layout_container = QtWidgets.QVBoxLayout()
         self.__layout_first_row = QtWidgets.QHBoxLayout()
         self.__layout_third_row = QtWidgets.QHBoxLayout()
@@ -100,6 +105,7 @@ def __setup_ui(self):
             self.__progress_bar
         )  # "Second Row" Dedicated to Progress Bar
         self.__layout_container.addWidget(self.__show_log_button)
+        self.__layout_container.addWidget(self.__show_container_hashes_button)
         self.__layout_container.addLayout(self.__layout_third_row)
         self.__layout_container.addLayout(self.__layout_fourth_row)
 
@@ -132,6 +138,11 @@ def __update_ui(self):
         else:
             self.__show_log_button.setHidden(True)
 
+        if self.__container_hashes_formatted:
+            self.__show_container_hashes_button.setHidden(False)
+        else:
+            self.__show_container_hashes_button.setHidden(True)
+
         try:
             current_percent = self.__processed_bytes / self.__total_bytes * 100
         except ZeroDivisionError:
@@ -247,6 +258,37 @@ def volume(self, volume):
             "The Volume associated to a widget cannot be changed after the widget creation!"
         )
 
+    @property
+    def container_hashes(self):
+        return self.__container_hashes
+
+    @container_hashes.setter
+    def container_hashes(self, container_hashes: list[dict[str, str | bool]]):
+        self.__container_hashes = container_hashes
+        if self.__container_hashes:
+            self.__container_hashes_formatted = "Container Hashes:\n"
+            for hash_value in self.__container_hashes:
+                if hash_value["verified"]:
+                    self.__container_hashes_formatted += f"- {hash_value['hash_type'].upper()} - VERIFIED\n\t- {hash_value['stored_hash']} (stored, calculated)\n"
+                else:
+                    self.__container_hashes_formatted += f"- {hash_value['hash_type'].upper()} - FAILED\n\t- {hash_value['stored_hash']} (stored)\n\t- {hash_value['calculated_hash']} (calculated)\n"
+        else:
+            self.__container_hashes_formatted = ""
+        self.__update_ui()
+
+    def __show_hashes(self):
+        self.hash_dialog = LogDialog(
+            self,
+            f"Container Hashes for {self.volume}",
+            self.__container_hashes_formatted,
+        )
+        self.hash_dialog.setWindowFlags(
+            QtCore.Qt.CustomizeWindowHint | QtCore.Qt.Dialog
+        )
+        self.hash_dialog.setModal(True)
+        self.hash_dialog.setWindowModality(QtCore.Qt.ApplicationModal)
+        self.hash_dialog.open()
+
     def __open_folder(self):
         if self.__aff4_verify:
             webbrowser.open(f"file://{os.path.dirname(self.__volume_name)}")

src/main/python/gemino/widgets/viewer/advanced.py

@@ -355,12 +355,22 @@ def show_case_metadata(self):
             self.volume.urn,
             rdfvalue.URN("http://aff4.org/Schema#endTime"),
         )[0]
+        container_hashes = self.volume.resolver.read_metadata_hashes(
+            self.volume.zip_file
+        )
 
         case_metadata = "<table>"
         case_metadata += f"<tr><td><b>Case Name:</b></td><td>{case_name}</td></tr>"
         case_metadata += f"<tr><td><b>Examiner:</b></td><td>{case_examiner}</td></tr>"
         case_metadata += f"<tr><td><b>Image Start:</b></td><td>{image_start}</td></tr>"
         case_metadata += f"<tr><td><b>Image End:</b></td><td>{image_end}</td></tr>"
+        if container_hashes:
+            case_metadata += f"<tr><td><b>Container Hashes:</b></td><td></td></tr>"
+            for algo, hash_value in container_hashes.items():
+                case_metadata += (
+                    f"<tr><td><b>- {algo.upper()}</b></td><td>{hash_value}</td></tr>"
+                )
+            case_metadata += f'<tr><td></td><td><i>Unverified, Verify using "Tools -> Verify AFF4-L Container"</i></td></tr>'
         case_metadata += f"<tr><td><b>Case Description:</b></td><td>{'<br/>'.join(str(case_description).splitlines())}</td></tr>"
         case_metadata += "</table>"
 
@@ -380,7 +390,7 @@ def show_case_metadata(self):
         )
         self.case_metadata_dialog.setModal(True)
         self.case_metadata_dialog.setWindowModality(QtCore.Qt.ApplicationModal)
-        self.case_metadata_dialog.setMinimumSize(500, 400)
+        self.case_metadata_dialog.setMinimumSize(700, 400)
         self.case_metadata_dialog.open()
 
     def load_metadata(self, urn, folder):

src/main/python/main.py

@@ -7,7 +7,6 @@
 from gemino.widgets.common import error_box
 from gemino.vars import VERSION
 
-
 if __name__ == "__main__":
     exit_code = -1
     app = QApplication(sys.argv)