store file-based discovery cache in a private per-user directory

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.

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()

tests/test_discovery_cache.py

@@ -18,17 +18,50 @@
 """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
 except ImportError:
     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(