fix(cpio): Fix dupplicated CPIO entries by overwriting them.

Author: nyuware

tests/integration/archive/cpio/cpio_binary/__input__/duplicate_entries.cpio

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:680e5a70389cde3c851563f81cbfc02be2d2460ac20765246d472f1d4b3bf24a
+size 512

tests/integration/archive/cpio/cpio_binary/__output__/duplicate_entries.cpio_extract/colors/blue.txt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a0bee6616b5e5eae6799cb4525a884a82e7161614f11122bbdf4383b2ac05998
+size 5

tests/integration/archive/cpio/cpio_binary/__output__/duplicate_entries.cpio_extract/colors/purple

@@ -0,0 +1 @@
+purple.txt

tests/integration/archive/cpio/cpio_binary/__output__/duplicate_entries.cpio_extract/colors/purple.txt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:596157e5858a0c647af6a4f55945b9c09ed70dc3f5d0e3db248ea5d5136e8ad3
+size 7

tests/integration/archive/cpio/cpio_binary/__output__/duplicate_entries.cpio_extract/colors/red.txt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6ace33171ce0acb6891e3cc311d75a97aa429d77c05cba600d49ed9652ed49de
+size 4

unblob/file_utils.py

@@ -628,7 +628,7 @@ def open(  # noqa: A003
 
     def unlink(self, path):
         """Delete file within extraction path."""
-        logger.debug("unlink file", file_path=path)
+        logger.debug("unlink file", file_path=path, _verbosity=3)
         safe_path = self._get_extraction_path(path, "unlink")
 
         safe_path.unlink(missing_ok=True)

unblob/handlers/archive/cpio.py

@@ -210,6 +210,11 @@ def dump_entries(self, fs: FileSystem):
             if entry.path.name in ("", "."):
                 continue
 
+            # There are cases where CPIO archives have duplicated entries
+            # We then unlink the files to overwrite them and avoid an error.
+            if not stat.S_ISDIR(entry.mode):
+                fs.unlink(entry.path)
+
             if stat.S_ISREG(entry.mode):
                 fs.carve(entry.path, self.file, entry.start_offset, entry.size)
             elif stat.S_ISDIR(entry.mode):