Vulnerable code sample

import os

# Base class for all blocks in the system
class Block:
    """Represents a single unit of execution in a graph."""
    def __init__(self, disabled=False):
        self.disabled = disabled

    def execute(self, inputs: dict) -> dict:
        """Executes the block's logic."""
        raise NotImplementedError

# A normal, safe block for demonstration purposes
class SafeEchoBlock(Block):
    """A simple block that echoes its input."""
    def execute(self, inputs: dict) -> dict:
        message = inputs.get("message", "No message provided.")
        print(f"[SafeEchoBlock] Executing with message: {message}")
        return {"output": message}

# The dangerous development block that allows arbitrary code execution
class BlockInstallationBlock(Block):
    """
    A development tool capable of writing and importing Python code.
    This block is marked as disabled to prevent its use in production.
    """
    def __init__(self):
        # The block is explicitly marked as disabled.
        super().__init__(disabled=True)

    def execute(self, inputs: dict) -> dict:
        """
        Executes arbitrary Python code provided in the 'code' input.
        This is the source of the RCE.
        """
        code_to_run = inputs.get("code")
        if not code_to_run:
            return {"status": "error", "message": "No code provided."}

        print(f"[!!!] BlockInstallationBlock: Executing arbitrary code on the server...")
        try:
            # WARNING: exec() is used here to simulate the RCE vulnerability.
            exec(code_to_run, {"os": os})
            print(f"[!!!] BlockInstallationBlock: Code execution finished successfully.")
            return {"status": "success"}
        except Exception as e:
            print(f"[!!!] BlockInstallationBlock: Code execution failed: {e}")
            return {"status": "error", "message": str(e)}

# A central registry mapping block IDs to their corresponding class instances
BLOCK_REGISTRY = {
    "safe_echo_block": SafeEchoBlock(),
    "dev_install_block": BlockInstallationBlock()
}

# Simulates the API endpoint for executing a single block directly.
# This endpoint correctly enforces the 'disabled' flag.
def execute_single_block(block_id: str, inputs: dict):
    """
    Handles direct execution of a block.
    This function correctly checks if a block is disabled.
    """
    print(f"\n>>> Attempting direct execution of block: '{block_id}'")
    block = BLOCK_REGISTRY.get(block_id)

    if not block:
        print(f"Error: Block '{block_id}' not found.")
        return

    # This is the correct security check.
    if block.disabled:
        print(f"Access Denied: Block '{block_id}' is disabled and cannot be run directly.")
        return

    return block.execute(inputs)

# Simulates the graph execution engine.
# This is the vulnerable part of the system.
def execute_graph(graph: dict):
    """
    Processes and executes a graph of connected blocks.
    VULNERABILITY: This function fails to check the 'disabled' flag for blocks
    within the graph, allowing a disabled block to be executed.
    """
    print(f"\n>>> Executing graph...")
    results = {}
    nodes = graph.get("nodes", [])

    for i, node in enumerate(nodes):
        block_id = node.get("block_id")
        inputs = node.get("inputs", {})
        block = BLOCK_REGISTRY.get(block_id)

        if not block:
            print(f"  - Node {i}: Skipping, block '{block_id}' not found.")
            continue

        # --- VULNERABILITY ---
        # The code does NOT check `if block.disabled:` before executing.
        # This allows an attacker to include the 'dev_install_block' in a graph
        # and have it execute on the server, bypassing the restriction.
        print(f"  - Node {i}: Executing block '{block_id}'...")
        result = block.execute(inputs)
        results[f"node_{i}"] = result

    print(">>> Graph execution finished.")
    return results