Patched code sample

import os
from flask import Flask, request, jsonify

# --- The Fix ---
# A predefined, secure base directory where all legitimate pipelines are stored.
# os.path.abspath ensures we have a canonical base path to compare against.
SAFE_PIPELINE_DIRECTORY = os.path.abspath('./allowed_pipelines')
# --- End of Fix ---

# Create the directory if it doesn't exist for the example to run
if not os.path.exists(SAFE_PIPELINE_DIRECTORY):
    os.makedirs(SAFE_PIPELINE_DIRECTORY)
    # Create a dummy pipeline file for a valid test case
    with open(os.path.join(SAFE_PIPELINE_DIRECTORY, 'legit_pipeline.py'), 'w') as f:
        f.write('print("Executing the legitimate pipeline.")\n')

app = Flask(__name__)

@app.route('/api/runs', methods=['POST'])
def run_pipeline():
    data = request.get_json()
    if not data or 'pipeline_path' not in data:
        return jsonify({"error": "pipeline_path is required"}), 400

    pipeline_path = data['pipeline_path']

    # --- The Fix ---
    # 1. Safely join the trusted base directory with the user-provided path.
    #    This prevents the user path from being treated as an absolute path.
    requested_path = os.path.join(SAFE_PIPELINE_DIRECTORY, pipeline_path)

    # 2. Resolve the real, absolute path (e.g., handles '..', symlinks).
    absolute_path = os.path.abspath(requested_path)

    # 3. CRITICAL CHECK: Ensure the resolved absolute path is still inside
    #    the safe base directory. If not, it's a path traversal attempt.
    if not absolute_path.startswith(SAFE_PIPELINE_DIRECTORY):
        return jsonify({"error": "Path traversal attempt detected"}), 403
    # --- End of Fix ---

    if not os.path.isfile(absolute_path):
        return jsonify({"error": "Pipeline file not found"}), 404

    try:
        # In a real application, this would use a secure method to execute
        # the pipeline. For this example, we simulate execution.
        print(f"--- SIMULATING EXECUTION OF: {absolute_path} ---")
        with open(absolute_path, 'r') as f:
            code = f.read()
            # In a real scenario, this might be exec(), importlib, or a subprocess
            # call. The security lies in the path validation above.
            exec(code)
        print("--- SIMULATION COMPLETE ---")
        
        return jsonify({"status": "success", "executed_pipeline": absolute_path}), 200
    except Exception as e:
        return jsonify({"error": f"Failed to execute pipeline: {str(e)}"}), 500

if __name__ == '__main__':
    print(f"Allowed pipelines must be in: {SAFE_PIPELINE_DIRECTORY}")
    print("Example valid request:")
    print("curl -X POST -H \"Content-Type: application/json\" -d '{\"pipeline_path\": \"legit_pipeline.py\"}' http://127.0.0.1:5000/api/runs")
    print("Example invalid (path traversal) request:")
    print("curl -X POST -H \"Content-Type: application/json\" -d '{\"pipeline_path\": \"../../../../etc/passwd\"}' http://127.0.0.1:5000/api/runs")
    app.run(port=5000)