Add in Zstd support (only present in >=python-3.14)

Author: ikelos

volatility3/framework/layers/resources.py

@@ -14,7 +14,7 @@
 import urllib.parse
 import urllib.request
 import zipfile
-from typing import Any, IO, List, Optional
+from typing import IO, Any, List, Optional
 from urllib import error
 
 from volatility3 import framework
@@ -27,6 +27,13 @@
 except ImportError:
     HAS_MAGIC = False
 
+try:
+    import zstd
+
+    ZSTD_SUPPORTED = True
+except ImportError:
+    ZSTD_SUPPORTED = False
+
 try:
     # Import so that the handler is found by the framework.class_subclasses callc
     from smb import SMBHandler as SMBHandler  # lgtm [py/unused-import]
@@ -232,21 +239,26 @@ def open(self, url: str, mode: str = "rb") -> Any:
                     # Only file's python has magic.detect_from_fobj
 
                 if detected:
+                    inside_compressed_file = False
                     if detected.mime_type == "application/x-xz":
                         curfile = cascadeCloseFile(
                             lzma.LZMAFile(curfile, mode), curfile
                         )
+                        inside_compressed_file = True
                     elif detected.mime_type == "application/x-bzip2":
                         curfile = cascadeCloseFile(bz2.BZ2File(curfile, mode), curfile)
+                        inside_compressed_file = True
                     elif detected.mime_type == "application/x-gzip":
                         curfile = cascadeCloseFile(
                             gzip.GzipFile(fileobj=curfile, mode=mode), curfile
                         )
-                    if detected.mime_type in [
-                        "application/x-xz",
-                        "application/x-bzip2",
-                        "application/x-gzip",
-                    ]:
+                        inside_compressed_file = True
+                    elif detected.mime_type == "application/zstd" and ZSTD_SUPPORTED:
+                        curfile = cascadeCloseFile(
+                            zstd.ZstdFile(fileobj=curfile, mode=mode), curfile
+                        )
+                        inside_compressed_file = True
+                    if inside_compressed_file:
                         # Read and rewind to ensure we're inside any compressed file layers
                         curfile.read(1)
                         curfile.seek(0)
@@ -272,6 +284,10 @@ def open(self, url: str, mode: str = "rb") -> Any:
                     curfile = cascadeCloseFile(
                         gzip.GzipFile(fileobj=curfile, mode=mode), curfile
                     )
+                elif extension == "zstd" and ZSTD_SUPPORTED:
+                    curfile = cascadeCloseFile(
+                        zstd.ZstdFile(fileobj=curfile, mode=mode), curfile
+                    )
                 else:
                     stop = True