diff --git a/changelog/14884.bugfix.rst b/changelog/14884.bugfix.rst new file mode 100644 index 00000000000..462d78dcb2b --- /dev/null +++ b/changelog/14884.bugfix.rst @@ -0,0 +1 @@ +``Cache.get()``, ``Cache.set()`` and ``Cache.mkdir()`` now reject keys that resolve outside the cache directory, and ``--cache-show`` ignores glob matches outside it. diff --git a/src/_pytest/cacheprovider.py b/src/_pytest/cacheprovider.py index 6bcac1ad97a..db3839198d9 100644 --- a/src/_pytest/cacheprovider.py +++ b/src/_pytest/cacheprovider.py @@ -160,6 +160,20 @@ def _mkdir(self, path: Path) -> None: self._ensure_cache_dir_and_supporting_files() path.mkdir(exist_ok=True, parents=True) + @staticmethod + def _join_within(base: Path, name: str) -> Path: + """Join ``name`` onto ``base``, keeping the result inside ``base``. + + ``joinpath()`` lets an absolute or drive-qualified ``name`` replace + ``base`` outright, and keeps ``..`` segments verbatim. Normalizing + lexically and re-checking containment rejects both, while still + allowing a ``..`` that cancels out within ``base``. + """ + path = Path(os.path.normpath(base.joinpath(name))) + if not path.is_relative_to(base): + raise ValueError(f"{name!r} is not allowed to escape the cache directory") + return path + def mkdir(self, name: str) -> Path: """Return a directory path object with the given name. @@ -174,15 +188,14 @@ def mkdir(self, name: str) -> Path: Make sure the name contains your plugin or application identifiers to prevent clashes with other cache users. """ - path = Path(name) - if len(path.parts) > 1: + if len(Path(name).parts) > 1: raise ValueError("name is not allowed to contain path separators") - res = self._cachedir.joinpath(self._CACHE_PREFIX_DIRS, path) + res = self._join_within(self._cachedir / self._CACHE_PREFIX_DIRS, name) self._mkdir(res) return res def _getvaluepath(self, key: str) -> Path: - return self._cachedir.joinpath(self._CACHE_PREFIX_VALUES, Path(key)) + return self._join_within(self._cachedir / self._CACHE_PREFIX_VALUES, key) def get(self, key: str, default): """Return the cached value for the given key. @@ -191,7 +204,8 @@ def get(self, key: str, default): default is returned. :param key: - Must be a ``/`` separated value. Usually the first + Must be a ``/`` separated value that does not resolve outside + the cache directory. Usually the first name is the name of your plugin or your application. :param default: The value to return in case of a cache-miss or invalid cache value. @@ -207,7 +221,8 @@ def set(self, key: str, value: object) -> None: """Save value for the given key. :param key: - Must be a ``/`` separated value. Usually the first + Must be a ``/`` separated value that does not resolve outside + the cache directory. Usually the first name is the name of your plugin or your application. :param value: Must be of any combination of basic python types, @@ -613,11 +628,20 @@ def cacheshow(config: Config, session: Session) -> int: if glob is None: glob = "*" + def globfiles(base: Path) -> Iterable[Path]: + """Glob for files under `base`, discarding matches that escape it. + + A glob may contain `..` segments, which `rglob()` happily follows. + """ + for x in base.rglob(glob): + if x.is_file() and Path(os.path.normpath(x)).is_relative_to(base): + yield x + dummy = object() basedir = config.cache._cachedir vdir = basedir / Cache._CACHE_PREFIX_VALUES tw.sep("-", f"cache values for {glob!r}") - for valpath in sorted(x for x in vdir.rglob(glob) if x.is_file()): + for valpath in sorted(globfiles(vdir)): key = str(valpath.relative_to(vdir)) val = config.cache.get(key, dummy) if val is dummy: @@ -629,12 +653,8 @@ def cacheshow(config: Config, session: Session) -> int: ddir = basedir / Cache._CACHE_PREFIX_DIRS if ddir.is_dir(): - contents = sorted(ddir.rglob(glob)) tw.sep("-", f"cache directories for {glob!r}") - for p in contents: - # if p.is_dir(): - # print("%s/" % p.relative_to(basedir)) - if p.is_file(): - key = str(p.relative_to(basedir)) - tw.line(f"{key} is a file of length {p.stat().st_size}") + for p in sorted(globfiles(ddir)): + key = str(p.relative_to(basedir)) + tw.line(f"{key} is a file of length {p.stat().st_size}") return 0 diff --git a/testing/test_cacheprovider.py b/testing/test_cacheprovider.py index 7ac3f38ab64..f84e03491fe 100644 --- a/testing/test_cacheprovider.py +++ b/testing/test_cacheprovider.py @@ -31,6 +31,41 @@ def test_config_cache_mkdir(self, pytester: Pytester) -> None: p = config.cache.mkdir("name") assert p.is_dir() + def test_config_cache_mkdir_escape(self, pytester: Pytester) -> None: + """`..` is a single path part, so it passes the separator check.""" + pytester.makeini("[pytest]") + config = pytester.parseconfigure() + assert config.cache is not None + with pytest.raises(ValueError): + config.cache.mkdir("..") + + @pytest.mark.parametrize( + "key", + [ + "../escaped", + "plugin/../../escaped", + "/absolute/escaped", + "//absolute/escaped", + ], + ) + def test_cache_key_escape(self, pytester: Pytester, key: str) -> None: + """Keys must not resolve outside the cache's values directory.""" + pytester.makeini("[pytest]") + config = pytester.parseconfigure() + assert config.cache is not None + with pytest.raises(ValueError): + config.cache.set(key, 1) + with pytest.raises(ValueError): + config.cache.get(key, None) + + def test_cache_key_normalized(self, pytester: Pytester) -> None: + """A `..` that cancels out within the values directory is fine.""" + pytester.makeini("[pytest]") + config = pytester.parseconfigure() + assert config.cache is not None + config.cache.set("plugin/sub/../value", 42) + assert config.cache.get("plugin/value", None) == 42 + def test_cache_dir_permissions(self, pytester: Pytester) -> None: """The .pytest_cache directory should have world-readable permissions (depending on umask). @@ -281,6 +316,29 @@ def pytest_configure(config): assert result.ret == 0 +def test_cache_show_escaping_glob(pytester: Pytester) -> None: + """A glob with `..` must not reach outside the cache directory.""" + pytester.makeconftest( + """ + def pytest_configure(config): + config.cache.set("my/name", [1, 2, 3]) + config.cache.mkdir("mydb").joinpath("hello").touch() + """ + ) + assert pytester.runpytest().ret == 5 # no tests executed + pytester.path.joinpath("secret.json").write_text( + '{"token": "s3cr3t"}', encoding="utf-8" + ) + + result = pytester.runpytest("--cache-show", "../../../secret.json") + assert result.ret == 0 + stdout = result.stdout.str() + # the glob itself is echoed in the section headers, the contents are not. + assert "s3cr3t" not in stdout + assert "contains" not in stdout + assert "is a file of length" not in stdout + + class TestLastFailed: def test_lastfailed_usecase( self, pytester: Pytester, monkeypatch: MonkeyPatch