Add Podman and OCI container support

dissect/target/filesystems/overlay.py

@@ -8,19 +8,21 @@
 from dissect.target.filesystems.dir import DirectoryFilesystem
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from pathlib import Path
 
 log = logging.getLogger(__name__)
 
 
 class Overlay2Filesystem(LayerFilesystem):
-    """Overlay 2 filesystem implementation.
+    """Docker Overlay 2 filesystem implementation.
 
     Deleted files will be present on the reconstructed filesystem.
     Volumes and bind mounts will be added to their respective mount locations.
     Does not support tmpfs mounts.
 
     References:
+        - https://docs.docker.com/engine/storage/drivers/overlayfs-driver/
         - https://docs.docker.com/storage/storagedriver/
         - https://docs.docker.com/storage/volumes/
         - https://www.didactic-security.com/resources/docker-forensics.pdf
@@ -108,8 +110,48 @@ def __init__(self, path: Path, *args, **kwargs):
             else:
                 layer_fs = DirectoryFilesystem(layer)
 
-            log.info("Adding layer %s to destination %s", layer, dest)
+            log.debug("Adding layer %s to destination %s", layer, dest)
             self.append_layer().mount("/" if layer.is_file() else dest, layer_fs)
 
     def __repr__(self) -> str:
         return f"<{self.__class__.__name__} {self.base_path}>"
+
+
+class OverlayFilesystem(LayerFilesystem):
+    """Podman overlay filesystem implementation.
+
+    Currently does not support mounting of (anonymous) volumes, named volumes and bind mounts.
+    Also does not map mount point files, hosts, hostname and resolv.conf files.
+
+    Resources:
+        - https://github.com/containers/podman
+        - https://docs.podman.io/en/latest/
+    """
+
+    __type__ = "overlay"
+
+    def __init__(self, path: Path, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.base_path = path
+
+        for dest, layer in oci_layers(path):
+            if not layer.exists():
+                log.warning(
+                    "Can not mount layer %s for container %s as it does not exist on the host", layer, path.name
+                )
+                continue
+
+            layer_fs = DirectoryFilesystem(layer)
+            log.debug("Adding layer %s to destination %s", layer, dest)
+            self.append_layer().mount("/" if layer.is_file() else dest, layer_fs)
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} {self.base_path}>"
+
+
+def oci_layers(path: Path) -> Iterator[tuple[str, Path]]:
+    """Yield the layers of an OCI container provided the ``mount_path``."""
+    yield ("/", path.joinpath("diff"))
+    for symlink in path.joinpath("lower").read_text().split(":"):
+        if symlink:
+            yield ("/", path.parent.joinpath(symlink).resolve())

dissect/target/loader.py

@@ -245,7 +245,8 @@ def open(item: str | Path, *args, **kwargs) -> Loader | None:
 register("log", "LogLoader")
 # Disabling ResLoader because of DIS-536
 # register("res", "ResLoader")
-register("overlay", "Overlay2Loader")
+register("overlay2", "Overlay2Loader")
+register("overlay", "OverlayLoader")
 register("phobos", "PhobosLoader")
 register("velociraptor", "VelociraptorLoader")
 register("uac", "UacLoader")

dissect/target/loaders/containerimage.py

@@ -10,6 +10,7 @@
 
 if TYPE_CHECKING:
     import tarfile
+    from pathlib import Path
 
     from dissect.target.target import Target
 
@@ -21,8 +22,6 @@
 }
 
 OCI_IMAGE = {
-    "manifest.json",
-    "repositories",
     "blobs",
     "oci-layout",
     "index.json",
@@ -34,6 +33,8 @@ class ContainerImageTarSubLoader(TarSubLoader):
 
     Supports both the Docker and OCI image specifications.
 
+    Tested with output from ``docker image save`` and ``podman image save``.
+
     References:
         - https://snyk.io/blog/container-image-formats/
         - https://github.com/moby/docker-image-spec/
@@ -43,7 +44,9 @@ class ContainerImageTarSubLoader(TarSubLoader):
     def __init__(self, tar: tarfile.TarFile, *args, **kwargs):
         super().__init__(tar, *args, **kwargs)
 
-        self.tarfs = None
+        self.tarfs: TarFilesystem = None
+        self.layers: list[Path] = []
+
         self.manifest = None
         self.name = None
         self.config = None
@@ -52,16 +55,34 @@ def __init__(self, tar: tarfile.TarFile, *args, **kwargs):
             self.tarfs = TarFilesystem(None, tarfile=tar)
         except Exception as e:
             raise ValueError(f"Unable to open {tar} as TarFilesystem: {e}") from e
-        try:
-            self.manifest = json.loads(self.tarfs.path("/manifest.json").read_text())[0]
-            self.name = self.manifest.get("RepoTags", [None])[0]
-        except Exception as e:
-            raise ValueError(f"Unable to read manifest.json inside docker image filesystem: {e}") from e
 
-        try:
-            self.config = json.loads(self.tarfs.path(self.manifest.get("Config")).read_text())
-        except Exception as e:
-            raise ValueError(f"Unable to read config inside docker image filesystem: {e}") from e
+        # Moby/Docker spec uses manifest.json
+        if self.tarfs.path("/manifest.json").exists():
+            try:
+                self.manifest = json.loads(self.tarfs.path("/manifest.json").read_text())[0]
+                self.name = self.manifest.get("RepoTags", [None])[0]
+                self.layers = [self.tarfs.path(p) for p in self.manifest.get("Layers", [])]
+            except Exception as e:
+                raise ValueError(f"Unable to read manifest.json inside docker image filesystem: {e}") from e
+
+            try:
+                self.config = json.loads(self.tarfs.path(self.manifest.get("Config")).read_text())
+            except Exception as e:
+                raise ValueError(f"Unable to read config inside docker image filesystem: {e}") from e
+
+        # OCI spec only has index.json
+        elif self.tarfs.path("/index.json").exists():
+            try:
+                index = json.loads(self.tarfs.path("/index.json").read_text())
+                self.config = json.loads(
+                    self.tarfs.path("/blobs").joinpath(index["manifests"][0]["digest"].replace(":", "/")).read_text()
+                )
+                self.layers = [
+                    self.tarfs.path("/blobs").joinpath(layer["digest"].replace(":", "/"))
+                    for layer in self.config.get("layers", [])
+                ]
+            except Exception as e:
+                raise ValueError(f"Unable to load OCI container: {e}") from e
 
     @staticmethod
     def detect(tar: tarfile.TarFile) -> bool:
@@ -71,7 +92,7 @@ def detect(tar: tarfile.TarFile) -> bool:
     def map(self, target: Target) -> None:
         fs = LayerFilesystem()
 
-        for layer in [self.tarfs.path(p) for p in self.manifest.get("Layers", [])]:
+        for layer in self.layers:
             if not layer.exists():
                 log.warning("Layer %s does not exist in container image", layer)
                 continue

dissect/target/loaders/overlay.py

@@ -2,30 +2,30 @@
 
 from typing import TYPE_CHECKING
 
-from dissect.target.filesystems.overlay import Overlay2Filesystem
+from dissect.target.filesystems.overlay import OverlayFilesystem
 from dissect.target.loader import Loader
 
 if TYPE_CHECKING:
     from dissect.target.helpers.fsutil import TargetPath
     from dissect.target.target import Target
 
 
-class Overlay2Loader(Loader):
-    """Load overlay2 filesystems."""
+class OverlayLoader(Loader):
+    """Load Podman OCI overlay filesystems."""
 
     @staticmethod
     def detect(path: TargetPath) -> bool:
         # path should be a folder
         if not path.is_dir():
             return False
 
-        # with the following three files
-        for required_file in ["init-id", "parent", "mount-id"]:
-            if not path.joinpath(required_file).exists():
+        # with the following files
+        for file in ["diff", "link", "lower", "work"]:  # "merged" is optional
+            if not path.joinpath(file).exists():
                 return False
 
         # and should have the following parent folders
-        return not "image/overlay2/layerdb/mounts/" not in path.as_posix()
+        return "containers/storage/overlay/" in path.as_posix()
 
     def map(self, target: Target) -> None:
-        target.filesystems.add(Overlay2Filesystem(self.absolute_path))
+        target.filesystems.add(OverlayFilesystem(self.absolute_path))

dissect/target/loaders/overlay2.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dissect.target.filesystems.overlay import Overlay2Filesystem
+from dissect.target.loader import Loader
+
+if TYPE_CHECKING:
+    from dissect.target.helpers.fsutil import TargetPath
+    from dissect.target.target import Target
+
+
+class Overlay2Loader(Loader):
+    """Load Docker overlay2 filesystems."""
+
+    @staticmethod
+    def detect(path: TargetPath) -> bool:
+        # path should be a folder
+        if not path.is_dir():
+            return False
+
+        # with the following three files
+        for required_file in ["init-id", "parent", "mount-id"]:
+            if not path.joinpath(required_file).exists():
+                return False
+
+        # and should have the following parent folders
+        return "image/overlay2/layerdb/mounts/" in path.as_posix()
+
+    def map(self, target: Target) -> None:
+        target.filesystems.add(Overlay2Filesystem(self.absolute_path))

dissect/target/plugins/apps/container/container.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from dissect.target.plugin import NamespacePlugin
+
+COMMON_IMAGE_FIELDS = [
+    ("string", "name"),
+    ("string", "tag"),
+    ("string", "image_id"),
+    ("string", "hash"),
+    ("datetime", "created"),
+    ("path", "source"),
+]
+
+COMMON_CONTAINER_FIELDS = [
+    ("string", "container_id"),
+    ("string", "image"),
+    ("string", "image_id"),
+    ("string", "command"),
+    ("datetime", "created"),
+    ("boolean", "running"),
+    ("varint", "pid"),
+    ("datetime", "started"),
+    ("datetime", "finished"),
+    ("string[]", "ports"),
+    ("string", "names"),
+    ("string[]", "volumes"),
+    ("string[]", "environment"),
+    ("path", "mount_path"),
+    ("path", "config_path"),
+    ("path", "image_path"),
+    ("path", "source"),
+]
+
+COMMON_LOG_FIELDS = [
+    ("datetime", "ts"),
+    ("string", "container"),
+    ("string", "stream"),
+    ("string", "message"),
+    ("path", "source"),
+]
+
+
+class ContainerPlugin(NamespacePlugin):
+    __namespace__ = "container"