diff --git a/googleapiclient/discovery_cache/__init__.py b/googleapiclient/discovery_cache/__init__.py index 6051191e73..ce4afa8a77 100644 --- a/googleapiclient/discovery_cache/__init__.py +++ b/googleapiclient/discovery_cache/__init__.py @@ -16,8 +16,11 @@ from __future__ import absolute_import +import getpass import logging import os +import stat +import tempfile LOGGER = logging.getLogger(__name__) @@ -26,6 +29,57 @@ os.path.dirname(os.path.realpath(__file__)), "documents" ) +_CACHE_SUBDIR_PREFIX = "google-api-python-client-discovery-cache-" + + +def _get_cache_directory(): + """Return a private, per-user directory for file based discovery caches. + + The file based discovery cache was historically written to a fixed, + world-predictable file name directly in the shared system temporary + directory (for example ``/tmp`` on Linux, which is writable by every + local user). Any local user could pre-create, replace, read, or symlink + that file and poison the cached discovery document. Because a discovery + document controls ``rootUrl``, a poisoned entry redirects every later + authenticated request, including the caller's OAuth credentials, to an + attacker chosen host. + + Keep the cache inside a per-user directory created with ``0o700`` so + other local users cannot access it, and refuse to use the directory when + it is not a regular directory owned by the current user with no access + for group or others. + """ + if hasattr(os, "getuid"): + suffix = str(os.getuid()) + else: + try: + raw = getpass.getuser() + except Exception: + raw = "" + suffix = "".join(c for c in raw if c.isalnum() or c in "._-") or "default" + + cache_dir = os.path.join(tempfile.gettempdir(), _CACHE_SUBDIR_PREFIX + suffix) + os.makedirs(cache_dir, exist_ok=True) + + if hasattr(os, "getuid"): + try: + os.chmod(cache_dir, 0o700) + except OSError: + pass + info = os.lstat(cache_dir) + if ( + not stat.S_ISDIR(info.st_mode) + or info.st_uid != os.getuid() + or info.st_mode & 0o077 + ): + raise OSError( + "Refusing to use discovery cache directory %r: it must be a " + "directory owned by the current user with no group or other " + "access" % cache_dir + ) + + return cache_dir + def autodetect(): """Detects an appropriate cache module and returns it. diff --git a/googleapiclient/discovery_cache/file_cache.py b/googleapiclient/discovery_cache/file_cache.py index 28a7219847..236aa69509 100644 --- a/googleapiclient/discovery_cache/file_cache.py +++ b/googleapiclient/discovery_cache/file_cache.py @@ -26,7 +26,6 @@ import json import logging import os -import tempfile try: from oauth2client.contrib.locked_file import LockedFile @@ -41,7 +40,7 @@ ) from . import base -from ..discovery_cache import DISCOVERY_DOC_MAX_AGE +from ..discovery_cache import DISCOVERY_DOC_MAX_AGE, _get_cache_directory LOGGER = logging.getLogger(__name__) @@ -85,7 +84,7 @@ def __init__(self, max_age): max_age: Cache expiration in seconds. """ self._max_age = max_age - self._file = os.path.join(tempfile.gettempdir(), FILENAME) + self._file = os.path.join(_get_cache_directory(), FILENAME) f = LockedFile(self._file, "a+", "r") try: f.open_and_lock() diff --git a/tests/test_discovery_cache.py b/tests/test_discovery_cache.py index 678350505d..70d406fb61 100644 --- a/tests/test_discovery_cache.py +++ b/tests/test_discovery_cache.py @@ -18,10 +18,17 @@ """Discovery document cache tests.""" import datetime +import os +import stat +import tempfile import unittest from unittest import mock -from googleapiclient.discovery_cache import DISCOVERY_DOC_MAX_AGE +from googleapiclient.discovery_cache import ( + _CACHE_SUBDIR_PREFIX, + DISCOVERY_DOC_MAX_AGE, + _get_cache_directory, +) try: from googleapiclient.discovery_cache.file_cache import Cache as FileCache @@ -29,6 +36,32 @@ FileCache = None +class CacheDirectoryTest(unittest.TestCase): + def test_cache_directory_is_private_and_contained(self): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch("tempfile.gettempdir", return_value=tmp): + cache_dir = _get_cache_directory() + # The cache lives in a dedicated subdirectory, not directly under + # the shared temporary directory with a predictable file name. + self.assertEqual(os.path.dirname(cache_dir), tmp) + self.assertTrue(os.path.isdir(cache_dir)) + if hasattr(os, "getuid"): + info = os.stat(cache_dir) + self.assertEqual(stat.S_IMODE(info.st_mode), 0o700) + self.assertEqual(info.st_uid, os.getuid()) + + @unittest.skipUnless(hasattr(os, "getuid"), "POSIX permission check") + def test_rejects_symlinked_cache_directory(self): + with tempfile.TemporaryDirectory() as tmp: + target = os.path.join(tmp, "attacker") + os.mkdir(target) + link = os.path.join(tmp, _CACHE_SUBDIR_PREFIX + str(os.getuid())) + os.symlink(target, link) + with mock.patch("tempfile.gettempdir", return_value=tmp): + with self.assertRaises(OSError): + _get_cache_directory() + + @unittest.skipIf(FileCache is None, "FileCache unavailable.") class FileCacheTest(unittest.TestCase): @mock.patch(