Manually resolves links relatively to root_dir, and prevent escape

This patch is a followup of #311.

It appeared that we were not resolving paths when reading from files.
This means that a symbolic link present under `root_dir` could be
blindly followed _outside_ of `root_dir`, possibly leading to host own
materials.

src/distro/distro.py

@@ -29,6 +29,7 @@
 """
 
 import argparse
+import errno
 import json
 import logging
 import os
@@ -768,8 +769,10 @@ def __init__(
 
             # NOTE: The idea is to respect order **and** have it set
             #       at all times for API backwards compatibility.
-            if os.path.isfile(etc_dir_os_release_file) or not os.path.isfile(
-                usr_lib_os_release_file
+            if (
+                self.root_dir is not None
+                or os.path.isfile(etc_dir_os_release_file)
+                or not os.path.isfile(usr_lib_os_release_file)
             ):
                 self.os_release_file = etc_dir_os_release_file
             else:
@@ -1083,6 +1086,59 @@ def uname_attr(self, attribute: str) -> str:
         """
         return self._uname_info.get(attribute, "")
 
+    @staticmethod
+    def __abs_path_join(root_path: str, abs_path: str) -> str:
+        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 os.path.join(root_path, rel_path)
+
+    def __resolve_chroot_symlink_as_needed(self, link_location: str) -> str:
+        """
+        Resolves a potential symlink in ``link_location``
+        against ``self.root_dir`` if inside the chroot,
+        else just return the original path.
+        We're doing this check at a central place, to making
+        the calling code more readable and to de-duplicate.
+        """
+        if self.root_dir is None:
+            return link_location
+
+        if os.path.commonprefix([self.root_dir, link_location]) != self.root_dir:
+            raise FileNotFoundError
+
+        seen_paths = []
+        while True:
+            try:
+                resolved = os.readlink(link_location)
+            except OSError:  # includes case "not a symlink"
+                if os.path.commonprefix(
+                    [
+                        os.path.realpath(self.root_dir),
+                        os.path.realpath(link_location),
+                    ]
+                ) != os.path.realpath(self.root_dir):
+                    # `link_location` resolves outside of `self.root_dir`.
+                    raise FileNotFoundError from None
+
+                return link_location
+
+            if os.path.isabs(resolved):
+                # i.e. absolute path (regarding to the chroot), that we need to
+                # "move" back inside the chroot.
+                resolved = self.__abs_path_join(self.root_dir, resolved)
+            else:
+                # i.e. relative path that we make absolute
+                resolved = os.path.join(os.path.dirname(link_location), resolved)
+
+            # prevent symlinks infinite loop
+            if resolved in seen_paths:
+                raise OSError(errno.ELOOP, os.strerror(errno.ELOOP), resolved)
+
+            seen_paths.append(link_location)
+            link_location = resolved
+
     @cached_property
     def _os_release_info(self) -> Dict[str, str]:
         """
@@ -1091,10 +1147,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_chroot_symlink_as_needed(self.os_release_file),
+                encoding="utf-8",
+            ) as release_file:
                 return self._parse_os_release_content(release_file)
-        return {}
+        except OSError:
+            return {}
 
     @staticmethod
     def _parse_os_release_content(lines: TextIO) -> Dict[str, str]:
@@ -1215,7 +1275,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_chroot_symlink_as_needed(
+                    os.path.join(self.etc_dir, "debian_version")
+                ),
+                encoding="ascii",
             ) as fp:
                 return fp.readline().rstrip()
         except FileNotFoundError:
@@ -1269,7 +1332,6 @@ def _distro_release_info(self) -> Dict[str, str]:
                     basename
                     for basename in os.listdir(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
@@ -1285,12 +1347,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 {}
@@ -1316,7 +1379,9 @@ 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_chroot_symlink_as_needed(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/absolutesymlinks/etc/os-release

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -0,0 +1 @@
+os-release-symlink

tests/resources/testdistros/distro/symlinksloop/etc/os-release-symlink

@@ -0,0 +1 @@
+os-release

tests/test_distro.py

@@ -731,6 +731,34 @@ def test_empty_release(self) -> None:
         desired_outcome = {"id": "empty"}
         self._test_outcome(desired_outcome)
 
+    def test_absolutesymlinks(self) -> None:
+        self.distro = distro.LinuxDistribution(
+            root_dir=os.path.join(TESTDISTROS, "distro", "absolutesymlinks")
+        )
+        desired_outcome = {"id": "absolutesymlinks"}
+        self._test_outcome(desired_outcome)
+
+    def test_rootdirescape(self) -> None:
+        self.distro = distro.LinuxDistribution(
+            root_dir=os.path.join(TESTDISTROS, "distro", "rootdirescape")
+        )
+        # loading existing file outside of root_dir has been prevented (empty data)
+        self._test_outcome({})
+
+    def test_rootdirnonescape(self) -> None:
+        self.distro = distro.LinuxDistribution(
+            root_dir=os.path.join(TESTDISTROS, "distro", "rootdirnonescape")
+        )
+        desired_outcome = {"id": "rootdirnonescape"}
+        self._test_outcome(desired_outcome)
+
+    def test_symlinksloop(self) -> None:
+        self.distro = distro.LinuxDistribution(
+            root_dir=os.path.join(TESTDISTROS, "distro", "symlinksloop")
+        )
+        # due to symbolic links loop, loading of file has been prevented (empty data)
+        self._test_outcome({})
+
     def test_dontincludeuname(self) -> None:
         self._setup_for_distro(os.path.join(TESTDISTROS, "distro", "dontincludeuname"))