Harden media and gameday access control

This commit is contained in:
Codex
2026-04-24 08:09:31 -05:00
parent cc241d4ae7
commit 22d4f8c017
4 changed files with 333 additions and 70 deletions

View File

@@ -20,6 +20,14 @@ from ..schemas import (
router = APIRouter(prefix="/games", tags=["games"])
def resolve_session_player_id(session: UserSession, requested_player_id: str | None = None) -> str:
if not session.external_team_id or not session.external_player_id:
raise HTTPException(status_code=422, detail="Select a team and player before using gameday")
if requested_player_id and requested_player_id != session.external_player_id:
raise HTTPException(status_code=403, detail="This player does not match your selected session")
return session.external_player_id
def assignment_to_response(assignment: GameAssignment) -> GameAssignmentResponse:
normalized_url = f"/media/files/{assignment.clip.normalized_path}" if assignment.clip.normalized_path else None
return GameAssignmentResponse(
@@ -57,9 +65,7 @@ def list_pins(
session: UserSession = Depends(require_session),
db: Session = Depends(get_db),
) -> list[GameAssignmentResponse]:
player_id = external_player_id or session.external_player_id
if not player_id or not session.external_team_id:
raise HTTPException(status_code=422, detail="Provide a player to list pins")
player_id = resolve_session_player_id(session, external_player_id)
query = select(GameAssignment).join(GameAssignment.clip).where(
GameAssignment.external_team_id == session.external_team_id,
@@ -80,15 +86,15 @@ def list_assignments(
response: Response,
external_game_id: str,
external_player_id: str | None = Query(default=None),
_: UserSession = Depends(require_session),
session: UserSession = Depends(require_session),
db: Session = Depends(get_db),
) -> list[GameAssignmentResponse]:
resolve_session_player_id(session, external_player_id)
query = select(GameAssignment).join(GameAssignment.clip).where(
GameAssignment.external_team_id == session.external_team_id,
GameAssignment.external_game_id == external_game_id,
AudioClip.hidden.is_(False),
)
if external_player_id:
query = query.where(GameAssignment.external_player_id == external_player_id)
assignments = db.scalars(query.order_by(AudioClip.sort_order.asc(), GameAssignment.updated_at.desc())).all()
payload = [assignment_to_response(assignment) for assignment in assignments]
etag, not_modified = prepare_conditional_response(request, payload)
@@ -102,38 +108,42 @@ def list_assignments(
def create_assignment(
external_game_id: str,
payload: GameAssignmentCreate,
_: UserSession = Depends(require_session),
session: UserSession = Depends(require_session),
db: Session = Depends(get_db),
) -> GameAssignmentResponse:
player_id = resolve_session_player_id(session, payload.external_player_id)
if payload.external_team_id != session.external_team_id:
raise HTTPException(status_code=403, detail="This team does not match your selected session")
clip = db.get(AudioClip, payload.clip_id)
if clip is None or clip.normalization_status != "ready":
raise HTTPException(status_code=422, detail="Clip is not ready")
if clip.hidden:
raise HTTPException(status_code=404, detail="Clip not found")
if clip.asset.external_team_id != payload.external_team_id:
if clip.asset.external_team_id != session.external_team_id:
raise HTTPException(status_code=422, detail="Clip does not belong to this team")
if clip.asset.owner_external_player_id != payload.external_player_id:
if clip.asset.owner_external_player_id != player_id:
raise HTTPException(status_code=403, detail="You can only pin clips owned by that player")
assignment = db.scalar(
select(GameAssignment).where(
GameAssignment.external_game_id == external_game_id,
GameAssignment.clip_id == payload.clip_id,
GameAssignment.external_team_id == session.external_team_id,
)
)
if assignment is None:
assignment = GameAssignment(
external_team_id=payload.external_team_id,
external_team_id=session.external_team_id,
external_game_id=external_game_id,
external_player_id=payload.external_player_id,
external_player_id=player_id,
clip_id=payload.clip_id,
batting_slot=payload.batting_slot,
status=payload.status,
)
db.add(assignment)
else:
assignment.external_team_id = payload.external_team_id
assignment.external_player_id = payload.external_player_id
assignment.external_team_id = session.external_team_id
assignment.external_player_id = player_id
assignment.clip_id = payload.clip_id
assignment.batting_slot = payload.batting_slot
assignment.status = payload.status
@@ -147,14 +157,15 @@ def delete_assignment(
external_game_id: str,
assignment_id: int,
external_player_id: str | None = Query(default=None),
_: UserSession = Depends(require_session),
session: UserSession = Depends(require_session),
db: Session = Depends(get_db),
) -> None:
player_id = resolve_session_player_id(session, external_player_id)
assignment = db.get(GameAssignment, assignment_id)
if assignment is None or assignment.external_game_id != external_game_id:
raise HTTPException(status_code=404, detail="Pin not found")
if external_player_id is not None and assignment.external_player_id != external_player_id:
raise HTTPException(status_code=403, detail="Pin does not belong to that player")
if assignment.external_team_id != session.external_team_id or assignment.external_player_id != player_id:
raise HTTPException(status_code=403, detail="Pin does not belong to your selected session")
db.delete(assignment)
db.commit()
@@ -165,13 +176,18 @@ def prepare_game(
request: Request,
response: Response,
external_game_id: str,
_: UserSession = Depends(require_session),
session: UserSession = Depends(require_session),
db: Session = Depends(get_db),
) -> GamePrepResponse:
resolve_session_player_id(session, None)
assignments = db.scalars(
select(GameAssignment)
.join(GameAssignment.clip)
.where(GameAssignment.external_game_id == external_game_id, AudioClip.hidden.is_(False))
.where(
GameAssignment.external_team_id == session.external_team_id,
GameAssignment.external_game_id == external_game_id,
AudioClip.hidden.is_(False),
)
.order_by(AudioClip.sort_order.asc(), GameAssignment.updated_at.desc())
).all()
external_team_id = assignments[0].external_team_id if assignments else ""