Vulnerable code sample

import json
from fastapi import FastAPI, WebSocket

# This code is a simplified, functional representation of the vulnerability
# described in CVE-2026-54695, as it might have existed in
# `pipecat/runner/utils.py` before version 1.4.0. It is not the actual
# source code but is designed to demonstrate the flaw for educational purposes.

# --- Mock Objects to simulate the application's environment ---

class MockAuthenticatedApiClient:
    """
    Simulates a client (for Twilio, Plivo, or Telnyx) that is authenticated
    with the server operator's credentials.
    """
    def __init__(self, api_key, api_secret):
        # In a real app, these are the server operator's credentials.
        self._api_key = api_key
        self._api_secret = api_secret
        print(f"Authenticated client initialized for API Key: {self._api_key}")

    def hangup_call(self, call_sid: str):
        """
        Simulates making an authenticated API request to terminate a call.
        This is the sensitive action an attacker can trigger.
        """
        print(
            f"[VULNERABLE ACTION] Issuing an authenticated hang-up request "
            f"for call SID: {call_sid}"
        )
        print(f"Call {call_sid} would be terminated now.")


# --- Vulnerable Application Code ---

# The server's globally available, authenticated API client.
# It is initialized with the server operator's private credentials.
authenticated_client = MockAuthenticatedApiClient(
    api_key="SERVER_OPERATOR_API_KEY",
    api_secret="SERVER_OPERATOR_API_SECRET"
)

app = FastAPI()

@app.websocket("/ws")
async def vulnerable_telephony_websocket(websocket: WebSocket):
    """
    This is the vulnerable WebSocket endpoint. It is exposed by the development
    runner and crucially, it does not perform any authentication or authorization
    checks on incoming connections.
    """
    await websocket.accept()
    try:
        # The endpoint expects a handshake message, like one from Twilio Streams.
        data = await websocket.receive_text()
        message = json.loads(data)

        # The vulnerability lies in trusting the initial handshake message.
        if message.get("event") == "start":
            start_data = message.get("start", {})
            # An attacker can supply any `callSid` in this message.
            call_sid = start_data.get("callSid")

            if call_sid:
                # The unvalidated `callSid` from the attacker is passed directly
                # to a function that performs a privileged action (hanging up a call)
                # using the server's pre-configured, authenticated client.
                authenticated_client.hangup_call(call_sid)

    except Exception:
        # Ignore exceptions for this demonstration.
        pass
    finally:
        await websocket.close()