Repository: pypa/pip
File: src/pip/_internal/network/cache.py (Python)

Code before:
    def _get_cache_path(self, name: str) -> str:
        # From cachecontrol.caches.file_cache.FileCache._fn, brought into our
        # class for backwards-compatibility and to avoid using a non-public
        # method.
        hashed = SeparateBodyFileCache.encode(name)
        parts = list(hashed[:5]) + [hashed]
        return os.path.join(self.directory, *parts)

    def get(self, key: str) -> Optional[bytes]:
        # The cache entry is only valid if both metadata and body exist.
        metadata_path = self._get_cache_path(key)
        body_path = metadata_path + ".body"
        if not (os.path.exists(metadata_path) and os.path.exists(body_path)):
            return None
        with suppressed_cache_errors():
            with open(metadata_path, "rb") as f:
                return f.read()

    def _write(self, path: str, data: bytes) -> None:
        with suppressed_cache_errors():
            ensure_dir(os.path.dirname(path))

            with adjacent_tmp_file(path) as f:
                f.write(data)

            # `& 0o666` selects the read/write permissions of the directory.
            # `| 0o600` sets owner read/write permissions.
            os.chmod(f.name, os.stat(self.directory).st_mode & 0o666 | 0o600)
            replace(f.name, path)

    def set(
        self, key: str, value: bytes, expires: Union[int, datetime, None] = None
    ) -> None:
        path = self._get_cache_path(key)
        self._write(path, value)

    def delete(self, key: str) -> None:
        path = self._get_cache_path(key)
        with suppressed_cache_errors():
            os.remove(path)
        with suppressed_cache_errors():
            os.remove(path + ".body")

    def get_body(self, key: str) -> Optional[BinaryIO]:
        # The cache entry is only valid if both metadata and body exist.
        metadata_path = self._get_cache_path(key)
        body_path = metadata_path + ".body"
        if not (os.path.exists(metadata_path) and os.path.exists(body_path)):
            return None
        with suppressed_cache_errors():
            return open(body_path, "rb")

Review comment (security):
```suggestion
            # `& 0o666` selects the read/write permissions of the cache directory.
```

Code after:
    def _get_cache_path(self, name: str) -> str:
        # From cachecontrol.caches.file_cache.FileCache._fn, brought into our
        # class for backwards-compatibility and to avoid using a non-public
        # method.
        hashed = SeparateBodyFileCache.encode(name)
        parts = list(hashed[:5]) + [hashed]
        return os.path.join(self.directory, *parts)

    def get(self, key: str) -> Optional[bytes]:
        # The cache entry is only valid if both metadata and body exist.
        metadata_path = self._get_cache_path(key)
        body_path = metadata_path + ".body"
        if not (os.path.exists(metadata_path) and os.path.exists(body_path)):
            return None
        with suppressed_cache_errors():
            with open(metadata_path, "rb") as f:
                return f.read()

    def _write(self, path: str, data: bytes) -> None:
        with suppressed_cache_errors():
            ensure_dir(os.path.dirname(path))

            with adjacent_tmp_file(path) as f:
                f.write(data)
                # Inherit the read/write permissions of the cache directory
                # to enable multi-user cache use-cases.
                mode = (
                    os.stat(self.directory).st_mode
                    & 0o666  # select read/write permissions of cache directory
                    | 0o600  # set owner read/write permissions
                )
                # Change permissions only if there is no risk of following a symlink.
                if os.chmod in os.supports_fd:
                    os.chmod(f.fileno(), mode)
                elif os.chmod in os.supports_follow_symlinks:
                    os.chmod(f.name, mode, follow_symlinks=False)

            replace(f.name, path)

    def set(
        self, key: str, value: bytes, expires: Union[int, datetime, None] = None
    ) -> None:
        path = self._get_cache_path(key)
        self._write(path, value)

    def delete(self, key: str) -> None:
        path = self._get_cache_path(key)
        with suppressed_cache_errors():
            os.remove(path)
        with suppressed_cache_errors():
            os.remove(path + ".body")