fix: validate module, class, and filename during MediaUpload deserialization

googleapiclient/http.py

@@ -410,19 +410,36 @@ def new_from_json(cls, s):
         representation produced by to_json().
 
         Args:
-          s: string, JSON from to_json().
+          s: string, JSON string to parse.
 
         Returns:
-          An instance of the subclass of MediaUpload that was serialized with
-          to_json().
+          An instance of the MediaUpload subclass specified in the JSON.
+
+        Raises:
+          ValueError: If the serialized data is not a dictionary, or specifies an
+            untrusted module or unsupported class name.
+          TypeError: If `s` is not a string.
+          OSError: If an underlying file cannot be opened (for file-based uploads).
         """
         data = json.loads(s)
-        # Find and call the right classmethod from_json() to restore the object.
-        module = data["_module"]
-        m = __import__(module, fromlist=module.split(".")[:-1])
-        kls = getattr(m, data["_class"])
-        from_json = getattr(kls, "from_json")
-        return from_json(s)
+        if not isinstance(data, dict):
+            raise ValueError("Serialized MediaUpload data must be a JSON object.")
+
+        module = data.get("_module")
+        class_name = data.get("_class")
+
+        # Security check (CWE-502): Reject any module outside of googleapiclient.http
+        # and any class not explicitly allowlisted in _ALLOWED_MEDIA_UPLOAD_CLASSES.
+        if (
+            module != "googleapiclient.http"
+            or class_name not in _ALLOWED_MEDIA_UPLOAD_CLASSES
+        ):
+            raise ValueError(
+                f"Refusing to deserialize untrusted class: {module}.{class_name}"
+            )
+
+        kls = _ALLOWED_MEDIA_UPLOAD_CLASSES[class_name]
+        return kls.from_json(s)
 
 
 class MediaIoBaseUpload(MediaUpload):
@@ -615,17 +632,80 @@ def to_json(self):
         """
         return self._to_json(strip=["_fd"])
 
-    @staticmethod
-    def from_json(s):
+    @classmethod
+    def from_json(cls, s):
+        """Reconstructs a MediaFileUpload instance from a JSON string.
+
+        Args:
+          s: str, JSON-encoded string produced by MediaFileUpload.to_json().
+
+        Returns:
+          A MediaFileUpload instance (or instance of a subclass).
+
+        Raises:
+          ValueError: If the JSON payload is invalid, is not a dictionary, or
+            contains missing, malformed, or invalid fields.
+          TypeError: If `s` is not a string, or if parameters passed to the
+            constructor have invalid types.
+          OSError: If the file specified by `_filename` cannot be opened or read
+            (e.g., FileNotFoundError, PermissionError).
+        """
         d = json.loads(s)
-        return MediaFileUpload(
-            d["_filename"],
-            mimetype=d["_mimetype"],
-            chunksize=d["_chunksize"],
-            resumable=d["_resumable"],
+        if not isinstance(d, dict):
+            raise ValueError("Serialized MediaFileUpload data must be a JSON object.")
+
+        filename = d.get("_filename")
+        # Check for null bytes to prevent null-byte injection / path truncation attacks
+        # when opening files on the local filesystem.
+        if not isinstance(filename, str) or not filename or "\x00" in filename:
+            raise ValueError(
+                "Invalid or missing '_filename' in serialized MediaFileUpload."
+            )
+
+        chunksize = d.get("_chunksize")
+        if chunksize is None:
+            chunksize = DEFAULT_CHUNK_SIZE
+        elif not isinstance(chunksize, int) or isinstance(chunksize, bool):
+            raise ValueError("'_chunksize' must be an integer.")
+
+        resumable = d.get("_resumable")
+        if resumable is None:
+            resumable = False
+        elif not isinstance(resumable, bool):
+            raise ValueError("'_resumable' must be a boolean.")
+
+        mimetype = d.get("_mimetype")
+        if mimetype is not None and not isinstance(mimetype, str):
+            raise ValueError("'_mimetype' must be a string.")
+
+        return cls(
+            filename,
+            mimetype=mimetype,
+            chunksize=chunksize,
+            resumable=resumable,
         )
 
 
+# Safe Deserialization Class Map (CWE-502 Mitigation)
+# Unsafe dynamic deserialization vulnerability:
+# In previous versions, MediaUpload.new_from_json() dynamically executed:
+#     m = __import__(module, fromlist=module.split(".")[:-1])
+#     kls = getattr(m, data["_class"])
+#     from_json = getattr(kls, "from_json")
+#     return from_json(s)
+#
+# Passing untrusted JSON to __import__() and getattr() allowed attackers who could
+# tamper with serialized state (e.g., in databases, task queues, or caches) to:
+#   1. Force arbitrary module loading from sys.path (leading to Remote Code Execution).
+#   2. Instantiate arbitrary classes within googleapiclient.http or other reachable modules.
+#
+# To eliminate reflection and prevent CWE-502, we strictly map allowed class names
+# directly to their factory/class references.
+_ALLOWED_MEDIA_UPLOAD_CLASSES = {
+    "MediaFileUpload": MediaFileUpload,
+}
+
+
 class MediaInMemoryUpload(MediaIoBaseUpload):
     """MediaUpload for a chunk of bytes.
 

tests/test_http.py

@@ -32,6 +32,7 @@
 import random
 import socket
 import ssl
+import tempfile
 import time
 import unittest
 from unittest import mock
@@ -42,6 +43,7 @@
 from googleapiclient.discovery import build
 from googleapiclient.errors import BatchError, HttpError, InvalidChunkSizeError
 from googleapiclient.http import (
+    DEFAULT_CHUNK_SIZE,
     MAX_URI_LENGTH,
     BatchHttpRequest,
     HttpMock,
@@ -1729,6 +1731,164 @@ def test_build_http_default_308_is_excluded_as_redirect(self):
         self.assertTrue(308 not in http.redirect_codes)
 
 
+class TestMediaUploadSerialization(unittest.TestCase):
+    """Tests input validation and safe reconstruction behavior for MediaUpload.
+
+    Covers mitigations for CWE-502 (Deserialization of Untrusted Data) and validates
+    that arbitrary reflection and file manipulation vectors are strictly blocked.
+    """
+
+    def test_deserialize_untrusted_class_raises_value_error(self):
+        """Verify that MediaUpload.new_from_json strictly rejects untrusted classes."""
+        cases = [
+            # Security test: Reject arbitrary standard library / built-in modules.
+            # Prevents untrusted JSON from importing modules like 'os' or looking up
+            # dangerous callables (e.g. 'os.system').
+            ("os", "system"),
+            # Security test: Reject non-upload classes in googleapiclient.http.
+            # Even if the module name is valid, non-MediaUpload classes like
+            # HttpRequest must be rejected to prevent unexpected dispatch.
+            ("googleapiclient.http", "HttpRequest"),
+            # Security test: Reject arbitrary external modules from sys.path.
+            # Prevents untrusted JSON from triggering dynamic __import__() on modules
+            # that might exist in /tmp, shared volumes, or writable site-packages (RCE).
+            ("nonexistent_module", "CustomClass"),
+        ]
+        for module, class_name in cases:
+            with self.subTest(module=module, class_name=class_name):
+                payload = json.dumps({"_module": module, "_class": class_name})
+                with self.assertRaisesRegex(
+                    ValueError, "Refusing to deserialize untrusted class"
+                ):
+                    MediaUpload.new_from_json(payload)
+
+    def test_deserialize_invalid_filename_raises_value_error(self):
+        """Verify that MediaFileUpload.from_json rejects malformed filenames.
+
+        Guards against type confusion and null-byte injection during filename parsing.
+        """
+        cases = [
+            None,
+            "",
+            123,
+            "/path/with/\x00/nullbyte",
+        ]
+        for invalid_filename in cases:
+            with self.subTest(invalid_filename=invalid_filename):
+                payload = json.dumps(
+                    {
+                        "_module": "googleapiclient.http",
+                        "_class": "MediaFileUpload",
+                        "_filename": invalid_filename,
+                        "_mimetype": "text/plain",
+                        "_chunksize": 1048576,
+                        "_resumable": True,
+                    }
+                )
+                with self.assertRaisesRegex(
+                    ValueError, "Invalid or missing '_filename'"
+                ):
+                    MediaUpload.new_from_json(payload)
+
+    def test_deserialize_valid_media_file_upload_roundtrip(self):
+        """Verify legitimate MediaFileUpload roundtrip serialization.
+
+        Ensures that valid MediaFileUpload instances continue to serialize and
+        reconstruct correctly without breaking backwards compatibility.
+        """
+        with tempfile.TemporaryDirectory() as tmpdir:
+            test_file = os.path.join(tmpdir, "test.txt")
+            with open(test_file, "wb") as f:
+                f.write(b"valid content")
+
+            upload = MediaFileUpload(test_file, mimetype="text/plain", resumable=True)
+            serialized = upload.to_json()
+
+            deserialized = MediaUpload.new_from_json(serialized)
+            self.assertIsInstance(deserialized, MediaFileUpload)
+            self.assertEqual(deserialized.getbytes(0, 13), b"valid content")
+
+    def test_deserialize_media_file_upload_default_fallbacks(self):
+        """Verify fallback handling when _chunksize or _resumable are missing/null."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            test_file = os.path.join(tmpdir, "test.txt")
+            with open(test_file, "wb") as f:
+                f.write(b"fallback test content")
+
+            # Payload omitting _chunksize and _resumable (or with explicit null values)
+            payload = json.dumps(
+                {
+                    "_module": "googleapiclient.http",
+                    "_class": "MediaFileUpload",
+                    "_filename": test_file,
+                    "_chunksize": None,
+                    "_resumable": None,
+                }
+            )
+
+            deserialized = MediaUpload.new_from_json(payload)
+            self.assertIsInstance(deserialized, MediaFileUpload)
+            self.assertEqual(deserialized.chunksize(), DEFAULT_CHUNK_SIZE)
+            self.assertFalse(deserialized.resumable())
+            self.assertEqual(deserialized.getbytes(0, 21), b"fallback test content")
+
+    def test_deserialize_invalid_field_types_raises_value_error(self):
+        """Verify that MediaFileUpload.from_json rejects invalid field types."""
+        cases = [
+            # Invalid _chunksize
+            ({"_chunksize": "not_an_int"}, "'_chunksize' must be an integer."),
+            ({"_chunksize": True}, "'_chunksize' must be an integer."),
+            ({"_chunksize": 1.5}, "'_chunksize' must be an integer."),
+            # Invalid _resumable
+            ({"_resumable": "true"}, "'_resumable' must be a boolean."),
+            ({"_resumable": 1}, "'_resumable' must be a boolean."),
+            # Invalid _mimetype
+            ({"_mimetype": 123}, "'_mimetype' must be a string."),
+            ({"_mimetype": False}, "'_mimetype' must be a string."),
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            test_file = os.path.join(tmpdir, "test.txt")
+            with open(test_file, "wb") as f:
+                f.write(b"content")
+
+            for override, error_msg in cases:
+                with self.subTest(override=override):
+                    data = {
+                        "_module": "googleapiclient.http",
+                        "_class": "MediaFileUpload",
+                        "_filename": test_file,
+                        "_chunksize": DEFAULT_CHUNK_SIZE,
+                        "_resumable": True,
+                        "_mimetype": "text/plain",
+                    }
+                    data.update(override)
+                    payload = json.dumps(data)
+                    with self.assertRaisesRegex(ValueError, error_msg):
+                        MediaUpload.new_from_json(payload)
+
+    def test_deserialize_non_dict_payload_raises_value_error(self):
+        """Verify that non-dictionary JSON payloads raise ValueError."""
+        cases = [
+            "[]",
+            '"string_payload"',
+            "123",
+            "true",
+            "null",
+        ]
+        for payload in cases:
+            with self.subTest(payload=payload):
+                with self.assertRaisesRegex(
+                    ValueError, "Serialized MediaUpload data must be a JSON object."
+                ):
+                    MediaUpload.new_from_json(payload)
+
+                with self.assertRaisesRegex(
+                    ValueError, "Serialized MediaFileUpload data must be a JSON object."
+                ):
+                    MediaFileUpload.from_json(payload)
+
+
 if __name__ == "__main__":
     logging.getLogger().setLevel(logging.ERROR)
     unittest.main()