fix(audit): fermer les frontieres restantes

app/routes/tryouts.py

@@ -10,6 +10,7 @@ from flask import Blueprint, abort, flash, redirect, render_template, request, u
 from flask_babel import gettext as _
 from flask_login import current_user, login_required
 from marshmallow import ValidationError
+from sqlalchemy import select
 
 from app.extensions import db
 from app.forms import flash_validation_errors, form_payload
@@ -32,7 +33,14 @@ from app.models import (
     TryoutRegistration,
     User,
 )
-from app.validators import PlayerSelectionSchema, TryoutSchema
+from app.validators import (
+    PlayerSelectionSchema,
+    TryoutRegistrationStatusSchema,
+    TryoutSchema,
+    TryoutStatusSchema,
+    TryoutTeamMemberSchema,
+    TryoutTeamSchema,
+)
 
 tryouts_bp = Blueprint('tryouts', __name__, url_prefix='/tryouts')
 
@@ -79,6 +87,25 @@ def _users_by_id(user_ids):
     return {user.id: user for user in User.query.filter(User.id.in_(wanted)).all()}
 
 
+def registration_lock_statement(tryout_id):
+    """The PostgreSQL row lock used by both registration entry points."""
+    return select(Tryout).where(Tryout.id == tryout_id).with_for_update()
+
+
+def locked_tryout_or_404(tryout_id):
+    """Load and row-lock a tryout while a registration slot is decided.
+
+    PostgreSQL serializes concurrent registration attempts on this row. The
+    duplicate check, capacity count and insert that follow therefore form
+    one decision instead of three independently racing statements. SQLite
+    ignores ``FOR UPDATE`` in tests, but production does not.
+    """
+    tryout = db.session.execute(registration_lock_statement(tryout_id)).scalar_one_or_none()
+    if tryout is None:
+        abort(404)
+    return tryout
+
+
 @tryouts_bp.route('')
 @login_required
 def list_tryouts():
@@ -406,10 +433,10 @@ def view_tryout(tryout_id):
 @login_required
 def register_for_tryout(tryout_id):
     """Register a player for a tryout. Only Players can self-register."""
-    tryout = Tryout.query.get_or_404(tryout_id)
     if not isinstance(current_user, Player):
         flash(_('Only players can register for tryouts.'), 'danger')
         return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
+    tryout = locked_tryout_or_404(tryout_id)
 
     if tryout.status not in ['upcoming', 'in_progress']:
         flash(_('This tryout is not accepting registrations.'), 'danger')
@@ -443,11 +470,15 @@ def update_status(tryout_id):
     if not current_user.can_manage_this_tryout(tryout):
         flash(_('Permission denied.'), 'danger')
         return redirect(url_for('tryouts.list_tryouts'))
-    new_status = request.form.get('status')
-    if new_status in ['upcoming', 'in_progress', 'completed']:
-        tryout.status = new_status
-        db.session.commit()
-        flash(_('Tryout status updated to %(new_status)s.', new_status=new_status), 'success')
+    try:
+        data = TryoutStatusSchema().load(form_payload(list_fields=()))
+    except ValidationError as err:
+        flash_validation_errors(err)
+        return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
+
+    tryout.status = data['status']
+    db.session.commit()
+    flash(_('Tryout status updated to %(new_status)s.', new_status=data['status']), 'success')
     return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
 
 
@@ -463,11 +494,15 @@ def update_registration_status(tryout_id, player_id):
     registration = TryoutRegistration.query.filter_by(
         tryout_id=tryout_id, player_id=player_id
     ).first_or_404()
-    new_status = request.form.get('status')
-    if new_status in ['registered', 'attended', 'no_show']:
-        registration.status = new_status
-        db.session.commit()
-        flash(_('Registration status updated.'), 'success')
+    try:
+        data = TryoutRegistrationStatusSchema().load(form_payload(list_fields=()))
+    except ValidationError as err:
+        flash_validation_errors(err)
+        return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
+
+    registration.status = data['status']
+    db.session.commit()
+    flash(_('Registration status updated.'), 'success')
     return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
 
 
@@ -475,7 +510,7 @@ def update_registration_status(tryout_id, player_id):
 @login_required
 def register_player(tryout_id):
     """Manually register a player for a tryout (by managers/coaches)."""
-    tryout = Tryout.query.get_or_404(tryout_id)
+    tryout = locked_tryout_or_404(tryout_id)
     if not current_user.can_manage_this_tryout(tryout):
         flash(_('Permission denied.'), 'danger')
         return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
@@ -563,12 +598,16 @@ def create_team(tryout_id):
         flash(_('Permission denied.'), 'danger')
         return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
 
-    team_name = request.form.get('team_name')
-    if team_name:
-        team = Team(tryout_id=tryout_id, name=team_name, created_by=current_user.id)
-        db.session.add(team)
-        db.session.commit()
-        flash(_('Team "%(team_name)s" created!', team_name=team_name), 'success')
+    try:
+        data = TryoutTeamSchema().load(form_payload(list_fields=()))
+    except ValidationError as err:
+        flash_validation_errors(err)
+        return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
+
+    team = Team(tryout_id=tryout_id, name=data['team_name'], created_by=current_user.id)
+    db.session.add(team)
+    db.session.commit()
+    flash(_('Team "%(team_name)s" created!', team_name=data['team_name']), 'success')
     return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
 
 
@@ -588,10 +627,12 @@ def add_to_team(tryout_id, team_id):
     if team.tryout_id != tryout_id:
         abort(404)
 
-    player_id = request.form.get('player_id', type=int)
-    if not player_id:
-        flash(_('Please select a player.'), 'danger')
+    try:
+        data = TryoutTeamMemberSchema().load(form_payload(list_fields=()))
+    except ValidationError as err:
+        flash_validation_errors(err)
         return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
+    player_id = data['player_id']
 
     # Only players registered for this tryout may be placed on its teams.
     is_registered = (
@@ -602,12 +643,11 @@ def add_to_team(tryout_id, team_id):
         flash(_('That player is not registered for this tryout.'), 'danger')
         return redirect(url_for('tryouts.view_tryout', tryout_id=tryout_id))
 
-    position = request.form.get('position', '')
     existing = TeamMember.query.filter_by(team_id=team_id, player_id=player_id).first()
     if existing:
         flash(_('Player is already on this team.'), 'info')
     else:
-        member = TeamMember(team_id=team_id, player_id=player_id, position=position)
+        member = TeamMember(team_id=team_id, player_id=player_id, position=data['position'])
         db.session.add(member)
         db.session.commit()
         flash(_('Player added to team!'), 'success')