Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/14884.bugfix.rst
Original file line number Diff line number Diff line change
@@ -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.
48 changes: 34 additions & 14 deletions src/_pytest/cacheprovider.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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.
Expand All @@ -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.
Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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
58 changes: 58 additions & 0 deletions testing/test_cacheprovider.py
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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
Expand Down
Loading