fix: enforce run ownership on all mutation endpoints

Add require_run_owner helper in auth.py that enforces ownership on
mutation endpoints. Unowned (legacy) runs are now read-only.

Applied ownership checks to:
- All 4 encounter mutation endpoints
- Both boss result mutation endpoints
- Run update/delete endpoints
- All 5 genlocke mutation endpoints (via first leg's run owner)

Also sets owner_id on run creation in genlockes.py (create_genlocke,
advance_leg) and adds 22 comprehensive ownership enforcement tests.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-03-21 13:28:32 +01:00
parent a12958ae32
commit eeb1609452
8 changed files with 660 additions and 39 deletions

View File

@@ -6,7 +6,7 @@ from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload
from app.core.auth import AuthUser, get_current_user, require_auth
from app.core.auth import AuthUser, get_current_user, require_auth, require_run_owner
from app.core.database import get_session
from app.models.boss_result import BossResult
from app.models.encounter import Encounter
@@ -181,31 +181,17 @@ def _build_run_response(run: NuzlockeRun) -> RunResponse:
)
def _check_run_access(
run: NuzlockeRun, user: AuthUser | None, require_owner: bool = False
) -> None:
def _check_run_read_access(run: NuzlockeRun, user: AuthUser | None) -> None:
"""
Check if user can access the run.
Check if user can read the run.
Raises 403 for private runs if user is not owner.
If require_owner=True, always requires ownership (for mutations).
Unowned runs are readable by everyone (legacy).
"""
if run.owner_id is None:
# Unowned runs are accessible by everyone (legacy)
if require_owner:
raise HTTPException(
status_code=403, detail="Only the run owner can perform this action"
)
return
user_id = UUID(user.id) if user else None
if require_owner:
if user_id != run.owner_id:
raise HTTPException(
status_code=403, detail="Only the run owner can perform this action"
)
return
if run.visibility == RunVisibility.PRIVATE and user_id != run.owner_id:
raise HTTPException(status_code=403, detail="This run is private")
@@ -301,7 +287,7 @@ async def get_run(
raise HTTPException(status_code=404, detail="Run not found")
# Check visibility access
_check_run_access(run, user)
_check_run_read_access(run, user)
# Check if this run belongs to a genlocke
genlocke_context = None
@@ -375,8 +361,7 @@ async def update_run(
if run is None:
raise HTTPException(status_code=404, detail="Run not found")
# Check ownership for mutations (unowned runs allow anyone for backwards compat)
_check_run_access(run, user, require_owner=run.owner_id is not None)
require_run_owner(run, user)
update_data = data.model_dump(exclude_unset=True)
@@ -484,8 +469,7 @@ async def delete_run(
if run is None:
raise HTTPException(status_code=404, detail="Run not found")
# Check ownership for deletion (unowned runs allow anyone for backwards compat)
_check_run_access(run, user, require_owner=run.owner_id is not None)
require_run_owner(run, user)
# Block deletion if run is linked to a genlocke leg
leg_result = await session.execute(