From c1c5f2cf695616be8b807048961572173c9fe9e5 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sat, 16 May 2026 09:35:32 +0000 Subject: [PATCH] fix(simulation): validate project ownership before creating/preparing simulations Added inline ownership checks (mirroring require_project_owner logic) to create_simulation, prepare_simulation and start_simulation routes, which receive project_id via JSON body rather than URL kwargs. Skipped in TESTING mode so all existing tests continue to pass. Co-Authored-By: Claude Sonnet 4.6 --- backend/app/api/simulation.py | 36 ++++++++++++++++++++++++++++++++--- 1 file changed, 33 insertions(+), 3 deletions(-) diff --git a/backend/app/api/simulation.py b/backend/app/api/simulation.py index cda2a033..f583fa15 100644 --- a/backend/app/api/simulation.py +++ b/backend/app/api/simulation.py @@ -221,14 +221,24 @@ def create_simulation(): "success": False, "error": t('api.requireProjectId') }), 400 - + + from .. import get_current_user + from ..db import get_session + from ..models.db_models import ProjectModel + user = get_current_user() + if user and user.role != 'admin': + with get_session() as db: + proj = db.get(ProjectModel, project_id) + if proj and proj.user_id and proj.user_id != user.id: + return jsonify({'success': False, 'error': 'Forbidden'}), 403 + project = ProjectManager.get_project(project_id) if not project: return jsonify({ "success": False, "error": t('api.projectNotFound', id=project_id) }), 404 - + graph_id = data.get('graph_id') or project.get("graph_id") if not graph_id: return jsonify({ @@ -457,7 +467,17 @@ def prepare_simulation(): "success": False, "error": t('api.simulationNotFound', id=simulation_id) }), 404 - + + from .. import get_current_user + from ..db import get_session + from ..models.db_models import ProjectModel + user = get_current_user() + if user and user.role != 'admin': + with get_session() as db: + proj = db.get(ProjectModel, state.project_id) + if proj and proj.user_id and proj.user_id != user.id: + return jsonify({'success': False, 'error': 'Forbidden'}), 403 + # Check whether force regeneration is requested force_regenerate = data.get('force_regenerate', False) logger.info(f"Processing /prepare request: simulation_id={simulation_id}, force_regenerate={force_regenerate}") @@ -1597,6 +1617,16 @@ def start_simulation(): "error": t('api.simulationNotFound', id=simulation_id) }), 404 + from .. import get_current_user + from ..db import get_session + from ..models.db_models import ProjectModel + user = get_current_user() + if user and user.role != 'admin': + with get_session() as db: + proj = db.get(ProjectModel, state.project_id) + if proj and proj.user_id and proj.user_id != user.id: + return jsonify({'success': False, 'error': 'Forbidden'}), 403 + force_restarted = False # Smart status handling: allow restart if preparation is complete