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 <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-05-16 09:35:32 +00:00
parent 0d04dee670
commit c1c5f2cf69
1 changed files with 33 additions and 3 deletions

View File

@ -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