Patched code sample

import os

# This example represents a hypothetical fix for the described vulnerability.
# The core principle of the fix is to stop executing code sent in the request payload
# for public flows and instead use the code already stored on the server for that flow.

# --- DATA STRUCTURES SIMULATING THE APPLICATION'S STATE ---

# Simulate a database table of trusted, pre-approved public flows.
# In a real application, this would be fetched from a database (e.g., PostgreSQL, SQLite).
# The key is the flow's public ID, and the value is the trusted flow data.
TRUSTED_PUBLIC_FLOWS_DB = {
    "public_flow_id_12345": {
        "name": "Greeting Flow",
        "nodes": [
            {
                "id": "node_A",
                "data": {
                    "node": {
                        "template": {
                            "code": {
                                # This is the trusted, server-side code. It cannot be
                                # overridden by the API request.
                                "value": "print('Executing trusted code from Node A: Hello, World!')"
                            }
                        }
                    }
                }
            }
        ]
    }
}


# --- VULNERABLE FUNCTION (FOR ILLUSTRATION) ---

def build_public_flow_vulnerable(payload: dict):
    """
    Represents the vulnerable behavior where code from the payload is executed.
    An attacker could provide a malicious payload to achieve RCE.
    """
    print("--- RUNNING VULNERABLE VERSION ---")
    try:
        # The vulnerability: The code is taken directly from the user's JSON payload.
        code_from_payload = payload['data']['nodes'][0]['data']['node']['template']['code']['value']
        print(f"Executing code received from payload: '{code_from_payload}'")
        exec(code_from_payload)
    except Exception as e:
        print(f"An error occurred: {e}")
    print("-" * 20 + "\n")


# --- FIXED FUNCTION ---

def build_public_flow_fixed(flow_id: str, payload: dict):
    """
    Represents the fixed behavior. It ignores the code in the user-provided
    payload and only executes the code stored on the server for the given flow_id.
    """
    print("--- RUNNING FIXED VERSION ---")
    # 1. Fetch the trusted flow definition from the server's storage using the ID.
    trusted_flow_data = TRUSTED_PUBLIC_FLOWS_DB.get(flow_id)

    if not trusted_flow_data:
        print(f"Error: Public flow with ID '{flow_id}' not found.")
        return

    # 2. THE FIX: Iterate through the nodes from the TRUSTED, server-side data.
    #    The `payload` from the user is ignored for sourcing executable code.
    for node_data in trusted_flow_data.get("nodes", []):
        try:
            # 3. Get the code from the trusted source, NOT the user's payload.
            trusted_code = node_data['data']['node']['template']['code']['value']
            print(f"Executing trusted code for node '{node_data['id']}': '{trusted_code}'")
            
            # 4. Safely execute the pre-approved code.
            exec(trusted_code)

        except Exception as e:
            print(f"An error occurred while executing trusted code: {e}")
    print("-" * 20 + "\n")


# --- DEMONSTRATION ---

if __name__ == "__main__":
    # A malicious payload trying to exploit the vulnerability.
    # The attacker wants to list directory contents on the server.
    malicious_payload = {
        "data": {
            "nodes": [
                {
                    "data": {
                        "node": {
                            "template": {
                                "code": {
                                    # This is the injected malicious code.
                                    "value": "print('Malicious code executed!'); import os; print(os.listdir('.'))"
                                }
                            }
                        }
                    }
                }
            ]
        }
    }
    
    # A legitimate, but user-submitted payload. The 'code' field should be ignored.
    legitimate_payload = {
        "data": {
            "nodes": [
                {
                    "data": {
                        "node": {
                            "template": {
                                "code": {
                                    "value": "print('This code should be ignored.')"
                                }
                            }
                        }
                    }
                }
            ]
        }
    }
    
    # Simulate the API call to the vulnerable endpoint.
    # The malicious code will be executed.
    build_public_flow_vulnerable(malicious_payload)
    
    # Simulate the API call to the fixed endpoint.
    # The `code` from the payload is ignored, and only the trusted,
    # stored code for 'public_flow_id_12345' is executed.
    build_public_flow_fixed(flow_id="public_flow_id_12345", payload=legitimate_payload)