Patched code sample

import sys
from dataclasses import dataclass
from typing import Dict, Any

# This code represents a hypothetical server fixing a session validation vulnerability.
# The original vulnerability (described in the fictitious CVE-2026-52869)
# would route requests to a session using only the session_id, without
# checking if the currently authenticated user (the principal) is the one
# who created the session.

# --- Data models to represent server state and requests ---

@dataclass
class Session:
    """Represents a session created by a specific user."""
    session_id: str
    owner_principal: str  # The authenticated user who created the session, e.g., 'user_alice'
    data: Dict[str, Any]

@dataclass
class IncomingRequest:
    """Represents a request from a client."""
    session_id_from_header: str
    authenticated_principal: str  # The user authenticated for this request, e.g., 'user_bob'
    message_to_inject: str

# --- Mock server session store ---

# A mock database of active sessions. In a real application, this would be
# a more robust storage system (like Redis or a database).
SESSIONS: Dict[str, Session] = {
    "session-abc-123": Session(
        session_id="session-abc-123",
        owner_principal="user_alice",
        data={"messages": []}
    )
}

# --- The request handler function containing the fix ---

def handle_session_message_fixed(request: IncomingRequest, session_store: Dict[str, Session]):
    """
    Handles an incoming message for a session, with the vulnerability fixed.
    """
    # Step 1: Retrieve the session using the session ID from the request.
    # This step was also present in the vulnerable version.
    target_session = session_store.get(request.session_id_from_header)

    if not target_session:
        raise ValueError(f"Session '{request.session_id_from_header}' not found.")

    # --- START OF VULNERABILITY FIX ---
    # The vulnerable version would skip this check and proceed directly to
    # processing the message, allowing any authenticated user to access the session.
    #
    # The fix is to verify that the principal authenticated for the current
    # request matches the principal that originally created the session.

    if target_session.owner_principal != request.authenticated_principal:
        # If the principals do not match, authoritatively deny access.
        raise PermissionError(
            f"Access denied. Principal '{request.authenticated_principal}' "
            f"does not own session '{target_session.session_id}'."
        )

    # --- END OF VULNERABILITY FIX ---

    # If the check passes, it is now safe to process the message and modify the session.
    print(
        f"Success: Principal '{request.authenticated_principal}' is authorized. "
        f"Injecting message into session '{target_session.session_id}'.",
        file=sys.stderr
    )
    target_session.data["messages"].append(request.message_to_inject)


if __name__ == '__main__':
    print("--- Demonstrating the fix for CVE-2026-52869 ---\n")

    # Scenario 1: The legitimate owner ('user_alice') sends a message.
    # This should succeed.
    legitimate_request = IncomingRequest(
        session_id_from_header="session-abc-123",
        authenticated_principal="user_alice",
        message_to_inject="Hello from Alice!"
    )
    print("Attempt 1: Legitimate owner 'user_alice' accesses her own session.")
    try:
        handle_session_message_fixed(legitimate_request, SESSIONS)
        print("Result: OK. Message injected successfully.\n")
    except (ValueError, PermissionError) as e:
        print(f"Result: FAILED. Unexpected error: {e}\n")


    # Scenario 2: An attacker ('user_bob') knows the session ID and tries to
    # inject a message.
    # The vulnerable code would have allowed this. The fixed code must block it.
    malicious_request = IncomingRequest(
        session_id_from_header="session-abc-123",
        authenticated_principal="user_bob",  # Different principal!
        message_to_inject="Malicious message from Bob!"
    )
    print("Attempt 2: Attacker 'user_bob' tries to access 'user_alice' session.")
    try:
        handle_session_message_fixed(malicious_request, SESSIONS)
        print("Result: FAILED. The vulnerability is still present.\n")
    except PermissionError as e:
        print(f"Result: OK. The fix worked. The server raised an error: {e}\n")

    print(f"Final session state: {SESSIONS['session-abc-123']}")