mirror of
https://github.com/evroon/bracket.git
synced 2026-08-02 19:27:25 -04:00
Fix SQLAlchemy and operator misuse causing broken query predicates and cross-tournament access (#1721)
Python's `and` operator is unsafe with SQLAlchemy clause elements: in
SQLAlchemy 2.0 it raises `TypeError` (routes return 500); in older
versions it short-circuits to the right-hand operand, silently dropping
the `id = X` filter and enabling cross-tournament entity access.
## Changes
**`routes/util.py`**
- `team_dependency`: replace `and` with `&`
- `round_dependency` / `match_dependency`: `rounds` and `matches` have
no direct `tournament_id`; replace the broken `and` chain with proper
JOINs through `stage_items → stages` to enforce tournament scoping
**`routes/courts.py`**
- `create_court` post-insert fetch: replace `and` with `&`
**`tests/.../teams_test.py`**
- Add `test_cross_tournament_team_access_denied`: asserts that a `PUT`
on tournament A's URL using a `team_id` belonging to tournament B
returns 404
```python
# Before (BROKEN — evaluates to just the right-hand side)
teams.select().where(teams.c.id == team_id and teams.c.tournament_id == tournament_id)
# After (CORRECT)
teams.select().where((teams.c.id == team_id) & (teams.c.tournament_id == tournament_id))
# round_dependency — no tournament_id on rounds table, use JOIN
rounds.select()
.join(stage_items, rounds.c.stage_item_id == stage_items.c.id)
.join(stages, stage_items.c.stage_id == stages.c.id)
.where((rounds.c.id == round_id) & (stages.c.tournament_id == tournament_id))
```
---------
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -108,7 +108,7 @@ async def create_court(
|
||||
database,
|
||||
Court,
|
||||
courts.select().where(
|
||||
courts.c.id == last_record_id and courts.c.tournament_id == tournament_id
|
||||
(courts.c.id == last_record_id) & (courts.c.tournament_id == tournament_id)
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -40,17 +40,17 @@ async def round_with_matches_dependency(
|
||||
|
||||
|
||||
async def stage_dependency(tournament_id: TournamentId, stage_id: StageId) -> StageWithStageItems:
|
||||
stages = await get_full_tournament_details(
|
||||
stages_result = await get_full_tournament_details(
|
||||
tournament_id, no_draft_rounds=False, stage_id=stage_id
|
||||
)
|
||||
|
||||
if len(stages) < 1:
|
||||
if len(stages_result) < 1:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Could not find stage with id {stage_id}",
|
||||
)
|
||||
|
||||
return stages[0]
|
||||
return stages_result[0]
|
||||
|
||||
|
||||
async def stage_item_dependency(
|
||||
@@ -81,7 +81,7 @@ async def team_dependency(tournament_id: TournamentId, team_id: TeamId) -> Team:
|
||||
team = await fetch_one_parsed(
|
||||
database,
|
||||
Team,
|
||||
teams.select().where(teams.c.id == team_id and teams.c.tournament_id == tournament_id),
|
||||
teams.select().where((teams.c.id == team_id) & (teams.c.tournament_id == tournament_id)),
|
||||
)
|
||||
|
||||
if team is None:
|
||||
|
||||
@@ -6,11 +6,19 @@ from bracket.database import database
|
||||
from bracket.models.db.team import Team
|
||||
from bracket.schema import players, teams
|
||||
from bracket.utils.db import fetch_one_parsed_certain
|
||||
from bracket.utils.dummy_records import DUMMY_MOCK_TIME, DUMMY_TEAM1
|
||||
from bracket.utils.dummy_records import DUMMY_MOCK_TIME, DUMMY_TEAM1, DUMMY_TOURNAMENT
|
||||
from bracket.utils.http import HTTPMethod
|
||||
from tests.integration_tests.api.shared import SUCCESS_RESPONSE, send_tournament_request
|
||||
from tests.integration_tests.api.shared import (
|
||||
SUCCESS_RESPONSE,
|
||||
send_auth_request,
|
||||
send_tournament_request,
|
||||
)
|
||||
from tests.integration_tests.models import AuthContext
|
||||
from tests.integration_tests.sql import assert_row_count_and_clear, inserted_team
|
||||
from tests.integration_tests.sql import (
|
||||
assert_row_count_and_clear,
|
||||
inserted_team,
|
||||
inserted_tournament,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
@@ -153,3 +161,25 @@ async def test_team_upload_and_remove_logo(
|
||||
assert not await aiofiles.os.path.exists(
|
||||
f"static/team-logos/{response['data']['logo_path']}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_cross_tournament_team_access_denied(
|
||||
startup_and_shutdown_uvicorn_server: None, auth_context: AuthContext
|
||||
) -> None:
|
||||
"""Regression test: a team from tournament B cannot be accessed via tournament A's URL."""
|
||||
async with inserted_tournament(
|
||||
DUMMY_TOURNAMENT.model_copy(
|
||||
update={"club_id": auth_context.club.id, "dashboard_endpoint": None}
|
||||
)
|
||||
) as other_tournament:
|
||||
async with inserted_team(
|
||||
DUMMY_TEAM1.model_copy(update={"tournament_id": other_tournament.id})
|
||||
) as other_team:
|
||||
response = await send_auth_request(
|
||||
HTTPMethod.PUT,
|
||||
f"tournaments/{auth_context.tournament.id}/teams/{other_team.id}",
|
||||
auth_context,
|
||||
json={"name": "Hacked", "active": True, "player_ids": []},
|
||||
)
|
||||
assert response.get("detail") == f"Could not find team with id {other_team.id}"
|
||||
|
||||
Reference in New Issue
Block a user