strat-gameplay-webapp/.claude/plans/001-websocket-authorization.md

Plan 001: WebSocket Authorization

Priority: CRITICAL Effort: 4-6 hours Status: NOT STARTED Risk Level: HIGH - Security vulnerability

---

Problem Statement

WebSocket handlers in backend/app/websocket/handlers.py have 11 TODO comments indicating missing access control. Any connected user can:

• Join any game
• Submit decisions for teams they don't own
• View game states they shouldn't access
• Manipulate games in progress

Impact

• Security: Unauthorized game access
• Data Integrity: Users can cheat by controlling opponent actions
• Trust: Players can't trust game outcomes

Files to Modify

File	Changes
backend/app/websocket/handlers.py	Add authorization checks to all handlers
backend/app/websocket/connection_manager.py	Track user-game associations
backend/app/models/db_models.py	May need game participant query
backend/app/database/operations.py	Add participant validation queries

Implementation Steps

Step 1: Create Authorization Utility (30 min)

Create backend/app/websocket/auth.py:

from uuid import UUID
from app.database.operations import db_ops

async def get_user_role_in_game(user_id: int, game_id: UUID) -> str | None:
    """
    Returns user's role in game: 'home', 'away', 'spectator', or None if not authorized.
    """
    game = await db_ops.get_game(game_id)
    if not game:
        return None

    if game.home_user_id == user_id:
        return "home"
    elif game.away_user_id == user_id:
        return "away"
    elif game.allow_spectators:
        return "spectator"
    return None

async def require_game_participant(sid: str, game_id: UUID, required_role: str | None = None) -> bool:
    """
    Validate user can access game. Emits error and returns False if unauthorized.
    """
    user_id = await manager.get_user_id(sid)
    role = await get_user_role_in_game(user_id, game_id)

    if role is None:
        await sio.emit("error", {"message": "Not authorized for this game"}, to=sid)
        return False

    if required_role and role != required_role:
        await sio.emit("error", {"message": f"Requires {required_role} role"}, to=sid)
        return False

    return True

async def require_team_control(sid: str, game_id: UUID, team_id: int) -> bool:
    """
    Validate user controls specified team.
    """
    user_id = await manager.get_user_id(sid)
    game = await db_ops.get_game(game_id)

    if team_id == game.home_team_id and game.home_user_id == user_id:
        return True
    elif team_id == game.away_team_id and game.away_user_id == user_id:
        return True

    await sio.emit("error", {"message": "Not authorized for this team"}, to=sid)
    return False

Step 2: Add User Tracking to ConnectionManager (30 min)

Update backend/app/websocket/connection_manager.py:

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, int] = {}  # sid -> user_id
        self.user_games: dict[int, set[UUID]] = {}    # user_id -> game_ids
        self.game_rooms: dict[UUID, set[str]] = {}    # game_id -> sids

    async def get_user_id(self, sid: str) -> int | None:
        return self.active_connections.get(sid)

    async def get_user_games(self, user_id: int) -> set[UUID]:
        return self.user_games.get(user_id, set())

Step 3: Update join_game Handler (30 min)

@sio.event
async def join_game(sid, data):
    game_id = UUID(data.get("game_id"))

    # Authorization check
    if not await require_game_participant(sid, game_id):
        return  # Error already emitted

    user_id = await manager.get_user_id(sid)
    role = await get_user_role_in_game(user_id, game_id)

    await manager.join_game(sid, game_id, role)
    # ... rest of handler

Step 4: Update Decision Handlers (1-2 hours)

Each decision handler needs team ownership validation:

@sio.event
async def submit_defensive_decision(sid, data):
    game_id = UUID(data.get("game_id"))
    team_id = data.get("team_id")

    # Authorization: must control this team
    if not await require_team_control(sid, game_id, team_id):
        return

    # ... rest of handler

Apply to:

• submit_defensive_decision
• submit_offensive_decision
• request_pinch_hitter
• request_defensive_replacement
• request_pitching_change
• roll_dice
• submit_manual_outcome

Step 5: Update Spectator-Only Handlers (30 min)

@sio.event
async def get_lineup(sid, data):
    game_id = UUID(data.get("game_id"))

    # Authorization: any participant (including spectators)
    if not await require_game_participant(sid, game_id):
        return

    # ... rest of handler

Apply to:

• get_lineup
• get_box_score

Step 6: Add Database Queries (30 min)

Add to backend/app/database/operations.py:

async def get_game_participants(self, game_id: UUID) -> dict:
    """Get home_user_id, away_user_id, allow_spectators for game."""
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(Game.home_user_id, Game.away_user_id, Game.allow_spectators)
            .where(Game.id == game_id)
        )
        row = result.first()
        if row:
            return {
                "home_user_id": row.home_user_id,
                "away_user_id": row.away_user_id,
                "allow_spectators": row.allow_spectators
            }
        return None

Step 7: Write Tests (1 hour)

Create backend/tests/unit/websocket/test_authorization.py:

import pytest
from app.websocket.auth import get_user_role_in_game, require_game_participant

class TestWebSocketAuthorization:
    """Tests for WebSocket authorization utilities."""

    async def test_home_user_gets_home_role(self):
        """Home team owner gets 'home' role."""

    async def test_away_user_gets_away_role(self):
        """Away team owner gets 'away' role."""

    async def test_spectator_allowed_when_enabled(self):
        """Non-participant gets 'spectator' when allowed."""

    async def test_unauthorized_user_rejected(self):
        """Non-participant rejected when spectators disabled."""

    async def test_require_team_control_validates_ownership(self):
        """User can only control their own team."""

Verification Checklist

• All 11 TODO comments addressed
• Home user can only control home team
• Away user can only control away team
• Spectators can view but not act
• Unauthorized users rejected with clear error
• Unit tests pass
• Manual test: try joining game as wrong user

Rollback Plan

If issues arise:

1. Revert handlers.py changes
2. Keep authorization utility for future use
3. Add rate limiting as temporary mitigation

Dependencies

• None (can be implemented independently)

Notes

• Consider caching user-game associations to reduce DB queries
• May want to add audit logging for authorization failures
• Future: Add game invite system for private games