CVE-2024-3571: Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal') (CWE-22)
value: langchain-ai/langchain is vulnerable to path traversal due to improper limitation of a pathname to a restricted directory ('Path Traversal') in its LocalFileStore functionality. An attacker can leverage this vulnerability to read or write files anywhere on the filesystem, potentially leading to information disclosure or remote code execution. The issue lies in the handling of file paths in the mset and mget methods, where user-supplied input is not adequately sanitized, allowing directory traversal sequences to reach unintended directories.
CWE description: The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
Published: 2024-04-16T00:15Z
Repository: https://github.com/langchain-ai/langchain
Language: Python
Files changed: libs/langchain/langchain/storage/file_system.py, libs/langchain/tests/unit_tests/storage/test_filesystem.py
Commit date: 2023-12-22 11:20:17 -0500
Commit message: langchain(patch): Restrict paths in LocalFileStore cache (#15065)

This PR restricts the paths that can be resolve using the local file system cache so that all paths must be contained within the root path.
Diff: diff --git a/libs/langchain/langchain/storage/file_system.py b/libs/langchain/langchain/storage/file_system.py
index 720acf085a..e737309a18 100644
--- a/libs/langchain/langchain/storage/file_system.py
+++ b/libs/langchain/langchain/storage/file_system.py
@@ -1,5 +1,6 @@
+import os
 import re
 from pathlib import Path
 from typing import Iterator, List, Optional, Sequence, Tuple, Union
 
 from langchain_core.stores import ByteStore
@@ -40,11 +41,11 @@ class LocalFileStore(ByteStore):
 
         Args:
             root_path (Union[str, Path]): The root path of the file store. All keys are
                 interpreted as paths relative to this root.
         """
-        self.root_path = Path(root_path)
+        self.root_path = Path(root_path).absolute()
 
     def _get_full_path(self, key: str) -> Path:
         """Get the full path for a given key relative to the root path.
 
         Args:
@@ -53,11 +54,19 @@ class LocalFileStore(ByteStore):
         Returns:
             Path: The full path for the given key.
         """
         if not re.match(r"^[a-zA-Z0-9_.\-/]+$", key):
             raise InvalidKeyException(f"Invalid characters in key: {key}")
-        return self.root_path / key
+        full_path = os.path.abspath(self.root_path / key)
+        common_path = os.path.commonpath([str(self.root_path), full_path])
+        if common_path != str(self.root_path):
+            raise InvalidKeyException(
+                f"Invalid key: {key}. Key should be relative to the full path."
+                f"{self.root_path} vs. {common_path} and full path of {full_path}"
+            )
+
+        return Path(full_path)
 
     def mget(self, keys: Sequence[str]) -> List[Optional[bytes]]:
         """Get the values associated with the given keys.
 
         Args:
diff --git a/libs/langchain/tests/unit_tests/storage/test_filesystem.py b/libs/langchain/tests/unit_tests/storage/test_filesystem.py
index 76991eb294..4213006bf2 100644
--- a/libs/langchain/tests/unit_tests/storage/test_filesystem.py
+++ b/libs/langchain/tests/unit_tests/storage/test_filesystem.py
@@ -75,5 +75,28 @@ def test_yield_keys(file_store: LocalFileStore) -> None:
     keys = list(file_store.yield_keys())
 
     # Assert that the yielded keys match the expected keys
     expected_keys = ["key1", os.path.join("subdir", "key2")]
     assert keys == expected_keys
+
+
+def test_catches_forbidden_keys(file_store: LocalFileStore) -> None:
+    """Make sure we raise exception on keys that are not allowed; e.g., absolute path"""
+    with pytest.raises(InvalidKeyException):
+        file_store.mset([("/etc", b"value1")])
+    with pytest.raises(InvalidKeyException):
+        list(file_store.yield_keys("/etc/passwd"))
+    with pytest.raises(InvalidKeyException):
+        file_store.mget(["/etc/passwd"])
+
+    # check relative paths
+    with pytest.raises(InvalidKeyException):
+        list(file_store.yield_keys(".."))
+
+    with pytest.raises(InvalidKeyException):
+        file_store.mget(["../etc/passwd"])
+
+    with pytest.raises(InvalidKeyException):
+        file_store.mset([("../etc", b"value1")])
+
+    with pytest.raises(InvalidKeyException):
+        list(file_store.yield_keys("../etc/passwd"))

Vulnerable code:         self.root_path = Path(root_path)
        return self.root_path / key
Fixed code: import os
        self.root_path = Path(root_path).absolute()
        full_path = os.path.abspath(self.root_path / key)
        common_path = os.path.commonpath([str(self.root_path), full_path])
        if common_path != str(self.root_path):
            raise InvalidKeyException(
                f"Invalid key: {key}. Key should be relative to the full path."
                f"{self.root_path} vs. {common_path} and full path of {full_path}"
            )

        return Path(full_path)


def test_catches_forbidden_keys(file_store: LocalFileStore) -> None:
    """Make sure we raise exception on keys that are not allowed; e.g., absolute path"""
    with pytest.raises(InvalidKeyException):
        file_store.mset([("/etc", b"value1")])
    with pytest.raises(InvalidKeyException):
        list(file_store.yield_keys("/etc/passwd"))
    with pytest.raises(InvalidKeyException):
        file_store.mget(["/etc/passwd"])

    # check relative paths
    with pytest.raises(InvalidKeyException):
        list(file_store.yield_keys(".."))

    with pytest.raises(InvalidKeyException):
        file_store.mget(["../etc/passwd"])

    with pytest.raises(InvalidKeyException):
        file_store.mset([("../etc", b"value1")])

    with pytest.raises(InvalidKeyException):
        list(file_store.yield_keys("../etc/passwd"))