Manually resolve paths relatively to root_dir to prevent escape

This patch is a followup of #311 (2a89f76).

It appeared that we were not resolving paths when reading from files.
This means that symbolic links present under `root_dir` could be
blindly followed, eventually leading _outside_ of `root_dir` (i.e host
own files).

Note : this patch **only** changes LinuxDistribution's `root_dir`
parameter behavior.

Author: HorlogeSkynet

src/distro/distro.py

@@ -37,6 +37,7 @@
 import subprocess
 import sys
 import warnings
+from pathlib import Path
 from typing import (
     Any,
     Callable,
@@ -1091,6 +1092,80 @@ def uname_attr(self, attribute: str) -> str:
         """
         return self._uname_info.get(attribute, "")
 
+    @staticmethod
+    def __abs_path_join(root_path: Path, abs_path: Path) -> Path:
+        rel_path = os.path.splitdrive(abs_path)[1].lstrip(os.sep)
+        if os.altsep is not None:
+            rel_path = rel_path.lstrip(os.altsep)
+
+        return root_path / Path(rel_path)
+
+    def __resolve_path_relatively_to_chroot(self, path: str) -> Path:
+        """
+        Resolves potential symlinks encountered in ``path`` against
+        ``self.root_dir`` if inside the chroot, else just return the original
+        path.
+        We're doing this check at a central place, to make calling code more
+        readable and to de-duplicate.
+        """
+        path_to_resolve = Path(path)
+
+        if self.root_dir is None:
+            return path_to_resolve
+
+        # resolve `self.root_dir`, once and for all.
+        chroot_path = Path(self.root_dir).resolve()
+
+        # consider non-absolute `path_to_resolve` relative to chroot
+        if not path_to_resolve.is_absolute():
+            path_to_resolve = chroot_path / path_to_resolve
+
+        seen_paths = set()
+        while True:
+            # although `path_to_resolve` _should_ be relative to chroot (either
+            # passed from trusted code or already resolved by previous loop
+            # iteration), we enforce this check as some input files are
+            # made available through API
+            try:
+                relative_parts = path_to_resolve.relative_to(chroot_path).parts
+            except ValueError:
+                raise FileNotFoundError
+
+            for i, part in enumerate(relative_parts, start=1):
+                if part == os.pardir:
+                    # normalize path parts up to this segment (relatively to chroot)
+                    path_to_resolve = self.__abs_path_join(
+                        chroot_path,
+                        Path(os.path.normpath("/" / Path(*relative_parts[:i]))),
+                    ) / Path(*relative_parts[i:])
+                    break  # restart path resolution
+
+                # attempt symbolic link resolution
+                symlink_candidate = chroot_path / Path(*relative_parts[:i])
+                try:
+                    symlink_resolved = Path(os.readlink(symlink_candidate))
+                except OSError:  # not a symlink, go to next path part
+                    continue
+
+                # "bend" absolute resolved path inside the chroot
+                # consider non-absolute resolved path relatively to chroot
+                if symlink_resolved.is_absolute():
+                    path_to_resolve = self.__abs_path_join(
+                        chroot_path, symlink_resolved
+                    )
+                else:
+                    path_to_resolve = symlink_candidate.parent / symlink_resolved
+                path_to_resolve /= Path(*relative_parts[i:])
+                break  # restart path resolution
+            else:
+                # return final path as it can be considered resolved
+                return path_to_resolve
+
+            # prevent symlinks infinite loop by tracking successive resolutions
+            if path_to_resolve in seen_paths:
+                raise FileNotFoundError
+            seen_paths.add(path_to_resolve)
+
     @cached_property
     def _os_release_info(self) -> Dict[str, str]:
         """
@@ -1099,10 +1174,14 @@ def _os_release_info(self) -> Dict[str, str]:
         Returns:
             A dictionary containing all information items.
         """
-        if os.path.isfile(self.os_release_file):
-            with open(self.os_release_file, encoding="utf-8") as release_file:
+        try:
+            with open(
+                self.__resolve_path_relatively_to_chroot(self.os_release_file),
+                encoding="utf-8",
+            ) as release_file:
                 return self._parse_os_release_content(release_file)
-        return {}
+        except FileNotFoundError:
+            return {}
 
     @staticmethod
     def _parse_os_release_content(lines: TextIO) -> Dict[str, str]:
@@ -1225,7 +1304,10 @@ def _oslevel_info(self) -> str:
     def _debian_version(self) -> str:
         try:
             with open(
-                os.path.join(self.etc_dir, "debian_version"), encoding="ascii"
+                self.__resolve_path_relatively_to_chroot(
+                    os.path.join(self.etc_dir, "debian_version")
+                ),
+                encoding="ascii",
             ) as fp:
                 return fp.readline().rstrip()
         except FileNotFoundError:
@@ -1235,7 +1317,10 @@ def _debian_version(self) -> str:
     def _armbian_version(self) -> str:
         try:
             with open(
-                os.path.join(self.etc_dir, "armbian-release"), encoding="ascii"
+                self.__resolve_path_relatively_to_chroot(
+                    os.path.join(self.etc_dir, "armbian-release")
+                ),
+                encoding="ascii",
             ) as fp:
                 return self._parse_os_release_content(fp).get("version", "")
         except FileNotFoundError:
@@ -1287,9 +1372,10 @@ def _distro_release_info(self) -> Dict[str, str]:
             try:
                 basenames = [
                     basename
-                    for basename in os.listdir(self.etc_dir)
+                    for basename in os.listdir(
+                        self.__resolve_path_relatively_to_chroot(self.etc_dir)
+                    )
                     if basename not in _DISTRO_RELEASE_IGNORE_BASENAMES
-                    and os.path.isfile(os.path.join(self.etc_dir, basename))
                 ]
                 # We sort for repeatability in cases where there are multiple
                 # distro specific files; e.g. CentOS, Oracle, Enterprise all
@@ -1305,12 +1391,13 @@ def _distro_release_info(self) -> Dict[str, str]:
                 match = _DISTRO_RELEASE_BASENAME_PATTERN.match(basename)
                 if match is None:
                     continue
-                filepath = os.path.join(self.etc_dir, basename)
-                distro_info = self._parse_distro_release_file(filepath)
+                # NOTE: _parse_distro_release_file below will be resolving for us
+                unresolved_filepath = os.path.join(self.etc_dir, basename)
+                distro_info = self._parse_distro_release_file(unresolved_filepath)
                 # The name is always present if the pattern matches.
                 if "name" not in distro_info:
                     continue
-                self.distro_release_file = filepath
+                self.distro_release_file = unresolved_filepath
                 break
             else:  # the loop didn't "break": no candidate.
                 return {}
@@ -1344,7 +1431,10 @@ def _parse_distro_release_file(self, filepath: str) -> Dict[str, str]:
             A dictionary containing all information items.
         """
         try:
-            with open(filepath, encoding="utf-8") as fp:
+            with open(
+                self.__resolve_path_relatively_to_chroot(filepath),
+                encoding="utf-8",
+            ) as fp:
                 # Only parse the first line. For instance, on SLES there
                 # are multiple lines. We don't want them...
                 return self._parse_distro_release_content(fp.readline())

tests/resources/testdistros/distro/root_dir_absolute_symlinks/etc/os-release

@@ -0,0 +1 @@
+/tmp/another-link

tests/resources/testdistros/distro/root_dir_absolute_symlinks/tmp/another-link

@@ -0,0 +1 @@
+/usr/lib/os-release

tests/resources/testdistros/distro/root_dir_absolute_symlinks/usr/lib/os-release

@@ -0,0 +1 @@
+ID=absolute_symlinks

tests/resources/testdistros/distro/root_dir_escape/etc/os-release

@@ -0,0 +1 @@
+../../../../distros/debian8/etc/os-release

tests/resources/testdistros/distro/root_dir_escape_abs/etc/os-release

@@ -0,0 +1 @@
+/../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../../usr/lib/os-release

tests/resources/testdistros/distro/root_dir_non_escape/etc/os-release

@@ -0,0 +1 @@
+/tmp/another-link/../../../../../../usr/lib/os-release

tests/resources/testdistros/distro/root_dir_non_escape/tmp/another-link

@@ -0,0 +1 @@
+nested/nested

tests/resources/testdistros/distro/root_dir_non_escape/tmp/nested/nested/.gitkeep

@@ -0,0 +1 @@
+ID=root_dir_non_escape