storage: Add Directory.readlink() method

3 files changed, 24 insertions, 0 deletions

diff --git a/src/buildstream/storage/_casbaseddirectory.py b/src/buildstream/storage/_casbaseddirectory.py
index 10d7a27d2..eafdbd283 100644
--- a/src/buildstream/storage/_casbaseddirectory.py
+++ b/src/buildstream/storage/_casbaseddirectory.py
@@ -884,6 +884,13 @@ class CasBasedDirectory(Directory):
 
         return entry.digest.hash
 
+    def readlink(self, *path):
+        entry = self._entry_from_path(*path)
+        if entry.type != _FileType.SYMLINK:
+            raise VirtualDirectoryError("Unsupported file type for readlink: {}".format(entry.type))
+
+        return entry.target
+
     def __iter__(self):
         yield from self.index.keys()
 
diff --git a/src/buildstream/storage/_filebaseddirectory.py b/src/buildstream/storage/_filebaseddirectory.py
index f8b9f95a0..f15dccedf 100644
--- a/src/buildstream/storage/_filebaseddirectory.py
+++ b/src/buildstream/storage/_filebaseddirectory.py
@@ -265,6 +265,15 @@ class FileBasedDirectory(Directory):
         newpath = os.path.join(subdir.external_directory, path[-1])
         return utils.sha256sum(newpath)
 
+    def readlink(self, *path):
+        # Use descend() to avoid following symlinks (potentially escaping the sandbox)
+        subdir = self.descend(*path[:-1])
+        if subdir.exists(path[-1]) and not subdir.islink(path[-1]):
+            raise VirtualDirectoryError("Unsupported file type for readlink")
+
+        newpath = os.path.join(subdir.external_directory, path[-1])
+        return os.readlink(newpath)
+
     def open_file(self, *path: str, mode: str = "r"):
         # Use descend() to avoid following symlinks (potentially escaping the sandbox)
         subdir = self.descend(*path[:-1])
diff --git a/src/buildstream/storage/directory.py b/src/buildstream/storage/directory.py
index 2dab9bcad..5b27c1a5a 100644
--- a/src/buildstream/storage/directory.py
+++ b/src/buildstream/storage/directory.py
@@ -289,6 +289,14 @@ class Directory:
         """
         raise NotImplementedError()
 
+    def readlink(self, *paths: str) -> str:
+        """ Return a string representing the path to which the symbolic link points.
+
+        Args:
+          *paths: A list of strings which are all path components.
+        """
+        raise NotImplementedError()
+
     def _create_empty_file(self, *paths):
         with self.open_file(*paths, mode="w"):
             pass