jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d5596afe10a67b8ea6d15500b2c81c2ede8da6a0
F-r-ldraledigplannerare/backend/tests/unit/test_authz.py
T
jacobh2 4fa79bad77 Merge branch 'worktree-check-security-audit' 
Security and frontend-correctness hardening: analytics IDOR fix, scraper
log scrubbing, lowercase email normalisation (migration 009), stale-response
guards in the plan hooks, busy flags cleared unconditionally across all
useLatestRequest consumers, Calendar no longer blanks on mutation, plus
dependabot/CI config and removal of the stray frontend/out.js.

Conflicts came from main's later "day assertions" restructure, which made
usePlannedDays read-only and replaced the calendar's per-day add/update with
useDayAssertions.assertDay. Both intents kept:

- usePlannedDays: branch's fetch machinery (isRefreshing, in-flight dedup,
  unguarded busy clear); mutators dropped as on main.
- usePlanSync: branch's useLatestRequest guard also applied to main's new
  syncWithWish.
- useDayAssertions / useLatestBalance (new on main): busy flag now cleared
  unconditionally, as the branch did for every other consumer.
- useDayAssertions.assertDay gains the branch's `skipRefresh` option and a
  `refresh`, so Calendar's multi-day save keeps the batch shape (N writes,
  one trailing bumpPlanVersion + one assertions refresh) on main's
  assertDay path. Branch's M7 tests ported from the removed add/update API
  to assertDay.
- PlanSyncProvider: branch's isAuthenticated gate kept; eviction-preview
  state stays removed as on main.
- usePlannedDayExclusions.ts stays deleted (superseded by useDayAssertions).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T4wjgbVL4HAC5Rdaw75VbV
2026-09-21 07:40:09 +02:00

727 lines
28 KiB
Python
Raw Blame History

"""Cross-cutting authorization tests (T1 authz audit).
Two things every route in the ten scoped routers must satisfy (see
`app/dependencies.py`'s `WorkspaceScope`/`WorkspaceIdDep` and the audit at
`graphify`-adjacent scratch notes for the full route table):
1. No `Authorization` header -> 401 (F1: `plan`, `leave_wishes`,
`day_overrides`, `vacation_days` had NO auth at all before this task).
2. A real id belonging to a DIFFERENT workspace -> 404, never 403 (F3: IDOR
on every workspace-/child-/pd-keyed route) — a 403 would confirm the id
exists, which is itself information disclosure.
This file exercises the real app + dependency graph (httpx ASGITransport),
following the pattern in `tests/unit/test_plan_router.py`, rather than only
the service layer — the bug class here is specifically about what the
ROUTER does (or fails to do) before calling the service.
"""
from datetime import date, datetime, timedelta
from uuid import uuid4
import aiosqlite
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from app.dependencies import get_connection, get_current_user
from app.entities.child import Child
from app.entities.day_override import DayOverride
from app.entities.leave_wish import LeaveWish
from app.entities.personal_data import PersonalData
from app.entities.planned_leave import PlannedLeaveDay
from app.entities.user import User
from app.entities.vacation_days import VacationDays
from app.entities.workspace import Workspace
from app.entities.workspace_invitation import WorkspaceInvitation
from app.main import app
from app.repositories_impl.sqlite.child_repository import SQLiteChildRepository
from app.repositories_impl.sqlite.day_override_repository import SQLiteDayOverrideRepository
from app.repositories_impl.sqlite.leave_wish_repository import SQLiteLeaveWishRepository
from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository
from app.repositories_impl.sqlite.planned_leave_repository import SQLitePlannedLeaveRepository
from app.repositories_impl.sqlite.user_repository import SQLiteUserRepository
from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository
from app.repositories_impl.sqlite.workspace_repository import (
SQLiteWorkspaceInvitationRepository,
SQLiteWorkspaceRepository,
)
# The caller's fixed identity across this file's authenticated tests: a
# member of "workspace A" (seeded per-test by `two_workspaces`).
_CALLER_PD_UUID = uuid4()
@pytest_asyncio.fixture
async def anon_client(connection: aiosqlite.Connection) -> AsyncClient:
"""No `get_current_user` override at all — a real, unauthenticated
request (no Authorization header) must be rejected by the real JWT
dependency, not by a test double standing in for it.
"""
app.dependency_overrides[get_connection] = lambda: connection
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
yield ac
app.dependency_overrides.pop(get_connection, None)
@pytest_asyncio.fixture
async def client(connection: aiosqlite.Connection) -> AsyncClient:
"""Authenticated as `_CALLER_PD_UUID` — a member of whatever workspace
the test seeds for them (see `two_workspaces`).
"""
app.dependency_overrides[get_connection] = lambda: connection
app.dependency_overrides[get_current_user] = lambda: User(
id=uuid4(),
personal_data_uuid=_CALLER_PD_UUID,
email="caller@example.com",
hashed_password="not-a-real-hash",
created_at=datetime.now(),
)
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
yield ac
app.dependency_overrides.pop(get_connection, None)
app.dependency_overrides.pop(get_current_user, None)
class _TwoWorkspaces:
def __init__(
self,
ws_a: Workspace,
ws_b: Workspace,
child_a: Child,
child_b: Child,
pd_b: PersonalData,
) -> None:
self.ws_a = ws_a
self.ws_b = ws_b
self.child_a = child_a
self.child_b = child_b
self.pd_b = pd_b
@pytest_asyncio.fixture
async def two_workspaces(connection: aiosqlite.Connection) -> _TwoWorkspaces:
"""Seed workspace A (the caller's own) and workspace B (a foreign
tenant), each with its own child and member.
"""
now = datetime.now()
ws_repo = SQLiteWorkspaceRepository(connection)
pd_repo = SQLitePersonalDataRepository(connection)
child_repo = SQLiteChildRepository(connection)
ws_a = await ws_repo.create(Workspace(id=uuid4(), name="Workspace A", created_at=now))
await pd_repo.create(
PersonalData(uuid=_CALLER_PD_UUID, workspace_id=ws_a.id, name="Caller", created_at=now)
)
child_a = await child_repo.create(
Child(id=uuid4(), workspace_id=ws_a.id, name="Child A", birth_date=date(2025, 1, 1))
)
ws_b = await ws_repo.create(Workspace(id=uuid4(), name="Workspace B", created_at=now))
pd_b = await pd_repo.create(
PersonalData(uuid=uuid4(), workspace_id=ws_b.id, name="Victim", created_at=now)
)
child_b = await child_repo.create(
Child(id=uuid4(), workspace_id=ws_b.id, name="Child B", birth_date=date(2025, 1, 1))
)
return _TwoWorkspaces(ws_a, ws_b, child_a, child_b, pd_b)
# ---------------------------------------------------------------------------
# F1 — routes that had NO authentication at all before this task.
# ---------------------------------------------------------------------------
class TestNoTokenIsUnauthenticated:
async def test_plan_propose_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post("/api/plan/propose", json={"workspace_id": str(uuid4())})
assert resp.status_code == 401
async def test_plan_apply_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post(
"/api/plan/apply", json={"workspace_id": str(uuid4()), "days": []}
)
assert resp.status_code == 401
async def test_plan_sync_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post("/api/plan/sync", json={"workspace_id": str(uuid4())})
assert resp.status_code == 401
async def test_leave_wishes_create_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post(
"/api/leave-wishes",
json={
"workspace_id": str(uuid4()),
"name": "X",
"window_start_month": 1,
"window_start_day": 1,
"window_end_month": 1,
"window_end_day": 7,
"target_weeks": 1,
"first_year": 2030,
},
)
assert resp.status_code == 401
async def test_leave_wishes_list_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.get(f"/api/leave-wishes/{uuid4()}")
assert resp.status_code == 401
async def test_day_overrides_create_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post(
"/api/day-overrides",
json={"workspace_id": str(uuid4()), "date": "2030-01-01", "is_working_day": False},
)
assert resp.status_code == 401
async def test_day_overrides_list_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.get(f"/api/day-overrides/workspace/{uuid4()}")
assert resp.status_code == 401
async def test_vacation_days_create_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post(
"/api/vacation-days",
json={
"personal_data_uuid": str(uuid4()),
"workspace_id": str(uuid4()),
"year": 2030,
"total_days": 25,
},
)
assert resp.status_code == 401
async def test_vacation_days_list_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.get(f"/api/vacation-days/workspace/{uuid4()}")
assert resp.status_code == 401
async def test_day_assertions_create_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.post(
"/api/day-assertions",
json={
"workspace_id": str(uuid4()),
"child_id": str(uuid4()),
"parent_personal_data_uuid": str(uuid4()),
"date": "2030-01-01",
"day_type": "vacation",
},
)
assert resp.status_code == 401
async def test_day_assertions_list_requires_auth(self, anon_client: AsyncClient) -> None:
resp = await anon_client.get(f"/api/day-assertions/workspace/{uuid4()}")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# F3 — a real id from a DIFFERENT workspace must 404, never 403.
# ---------------------------------------------------------------------------
class TestForeignWorkspaceIdIs404:
async def test_workspace_get(self, client: AsyncClient, two_workspaces: _TwoWorkspaces) -> None:
resp = await client.get(f"/api/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_workspace_members(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/workspace/{two_workspaces.ws_b.id}/members")
assert resp.status_code == 404
async def test_tips_get(self, client: AsyncClient, two_workspaces: _TwoWorkspaces) -> None:
resp = await client.get(f"/api/tips/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_children_by_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/children/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_children_get_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/children/{two_workspaces.child_b.id}")
assert resp.status_code == 404
async def test_settings_update_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.put(
f"/api/settings/child/{two_workspaces.child_b.id}", json={"name": "Hijacked"}
)
assert resp.status_code == 404
async def test_settings_delete_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.delete(f"/api/settings/{two_workspaces.child_b.id}")
assert resp.status_code == 404
async def test_settings_update_foreign_parent(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.put(
f"/api/settings/parent/{two_workspaces.pd_b.uuid}",
json={"employment_status": "none"},
)
assert resp.status_code == 404
async def test_calendar_days_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(
"/api/calendar/days",
params={"workspace_id": str(two_workspaces.ws_b.id), "start_year": 2030},
)
assert resp.status_code == 404
async def test_analytics_overview_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(
f"/api/analytics/{two_workspaces.ws_b.id}/overview",
params={"child_id": str(two_workspaces.child_b.id)},
)
assert resp.status_code == 404
async def test_analytics_overview_foreign_child_in_own_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
"""R1: `workspace_id` is the caller's own (workspace A), but
`child_id` belongs to workspace B — the IDOR the router-level
`scope.check_child` guard closes (see `analytics.get_overview`).
"""
resp = await client.get(
f"/api/analytics/{two_workspaces.ws_a.id}/overview",
params={"child_id": str(two_workspaces.child_b.id)},
)
assert resp.status_code == 404
async def test_analytics_projection_foreign_child_in_own_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
"""Same IDOR as above, on the sibling `/projection` route."""
resp = await client.get(
f"/api/analytics/{two_workspaces.ws_a.id}/projection",
params={"child_id": str(two_workspaces.child_b.id)},
)
assert resp.status_code == 404
async def test_analytics_age4_foreign_parent(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(
f"/api/analytics/parent/{two_workspaces.pd_b.uuid}"
f"/child/{two_workspaces.child_b.id}/age4"
)
assert resp.status_code == 404
async def test_plan_propose_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/plan/propose", json={"workspace_id": str(two_workspaces.ws_b.id)}
)
assert resp.status_code == 404
async def test_plan_apply_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/plan/apply",
json={
"workspace_id": str(two_workspaces.ws_a.id),
"days": [
{
"date": "2030-06-01",
"parent_personal_data_uuid": str(_CALLER_PD_UUID),
"child_id": str(two_workspaces.child_b.id),
"day_type": "vacation",
}
],
},
)
assert resp.status_code == 404
async def test_plan_sync_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/plan/sync", json={"workspace_id": str(two_workspaces.ws_b.id)}
)
assert resp.status_code == 404
async def test_leave_wishes_create_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/leave-wishes",
json={
"workspace_id": str(two_workspaces.ws_b.id),
"name": "X",
"window_start_month": 1,
"window_start_day": 1,
"window_end_month": 1,
"window_end_day": 7,
"target_weeks": 1,
"first_year": 2030,
},
)
assert resp.status_code == 404
async def test_leave_wishes_create_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/leave-wishes",
json={
"workspace_id": str(two_workspaces.ws_a.id),
"child_id": str(two_workspaces.child_b.id),
"name": "X",
"window_start_month": 1,
"window_start_day": 1,
"window_end_month": 1,
"window_end_day": 7,
"target_weeks": 1,
"first_year": 2030,
},
)
assert resp.status_code == 404
async def test_leave_wishes_list_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/leave-wishes/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_leave_wishes_update_foreign_wish(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
wish_repo = SQLiteLeaveWishRepository(connection)
wish = await wish_repo.create(
LeaveWish(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
child_id=None,
name="Foreign",
window_start_month=1,
window_start_day=1,
window_end_month=1,
window_end_day=7,
target_weeks=1,
first_year=2030,
last_year=None,
priority=1,
parent_mode="either",
parent_personal_data_uuid=None,
)
)
resp = await client.put(f"/api/leave-wishes/{wish.id}", json={"name": "Hijacked"})
assert resp.status_code == 404
async def test_leave_wishes_delete_foreign_wish(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
wish_repo = SQLiteLeaveWishRepository(connection)
wish = await wish_repo.create(
LeaveWish(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
child_id=None,
name="Foreign",
window_start_month=1,
window_start_day=1,
window_end_month=1,
window_end_day=7,
target_weeks=1,
first_year=2030,
last_year=None,
priority=1,
parent_mode="either",
parent_personal_data_uuid=None,
)
)
resp = await client.delete(f"/api/leave-wishes/{wish.id}")
assert resp.status_code == 404
async def test_day_overrides_create_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/day-overrides",
json={
"workspace_id": str(two_workspaces.ws_b.id),
"date": "2030-01-01",
"is_working_day": False,
},
)
assert resp.status_code == 404
async def test_day_overrides_get_foreign(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
repo = SQLiteDayOverrideRepository(connection)
row = await repo.create(
DayOverride(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
date=date(2030, 1, 1),
is_working_day=False,
)
)
resp = await client.get(f"/api/day-overrides/{row.id}")
assert resp.status_code == 404
async def test_day_overrides_list_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/day-overrides/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_day_overrides_delete_foreign(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
repo = SQLiteDayOverrideRepository(connection)
row = await repo.create(
DayOverride(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
date=date(2030, 1, 2),
is_working_day=False,
)
)
resp = await client.delete(f"/api/day-overrides/{row.id}")
assert resp.status_code == 404
async def test_vacation_days_create_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/vacation-days",
json={
"personal_data_uuid": str(_CALLER_PD_UUID),
"workspace_id": str(two_workspaces.ws_b.id),
"year": 2030,
"total_days": 25,
},
)
assert resp.status_code == 404
async def test_vacation_days_create_foreign_member(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/vacation-days",
json={
"personal_data_uuid": str(two_workspaces.pd_b.uuid),
"workspace_id": str(two_workspaces.ws_a.id),
"year": 2030,
"total_days": 25,
},
)
assert resp.status_code == 404
async def test_vacation_days_get_foreign(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
repo = SQLiteVacationDaysRepository(connection)
row = await repo.create(
VacationDays(
id=uuid4(),
personal_data_uuid=two_workspaces.pd_b.uuid,
workspace_id=two_workspaces.ws_b.id,
year=2030,
total_days=25,
)
)
resp = await client.get(f"/api/vacation-days/{row.id}")
assert resp.status_code == 404
async def test_vacation_days_list_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/vacation-days/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_planned_days_list_by_child_foreign(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(
"/api/planned-days", params={"child_id": str(two_workspaces.child_b.id)}
)
assert resp.status_code == 404
async def test_planned_days_get_foreign(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
repo = SQLitePlannedLeaveRepository(connection)
row = await repo.create(
PlannedLeaveDay(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
child_id=two_workspaces.child_b.id,
parent_personal_data_uuid=two_workspaces.pd_b.uuid,
date=date(2030, 6, 1),
day_type="vacation",
)
)
resp = await client.get(f"/api/planned-days/{row.id}")
assert resp.status_code == 404
async def test_planned_days_by_workspace_foreign(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/planned-days/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_day_assertions_create_foreign_child(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/day-assertions",
json={
"workspace_id": str(two_workspaces.ws_a.id),
"child_id": str(two_workspaces.child_b.id),
"parent_personal_data_uuid": str(_CALLER_PD_UUID),
"date": "2030-06-01",
"day_type": "vacation",
},
)
assert resp.status_code == 404
async def test_day_assertions_create_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/day-assertions",
json={
"workspace_id": str(two_workspaces.ws_b.id),
"child_id": str(two_workspaces.child_b.id),
"parent_personal_data_uuid": str(two_workspaces.pd_b.uuid),
"date": "2030-06-01",
"day_type": "vacation",
},
)
assert resp.status_code == 404
async def test_day_assertions_list_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(f"/api/day-assertions/workspace/{two_workspaces.ws_b.id}")
assert resp.status_code == 404
async def test_day_assertions_clear_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.request(
"DELETE",
"/api/day-assertions",
params={
"workspace_id": str(two_workspaces.ws_b.id),
"child_id": str(two_workspaces.child_b.id),
"parent_personal_data_uuid": str(two_workspaces.pd_b.uuid),
"date": "2030-06-01",
},
)
assert resp.status_code == 404
async def test_fetch_data_manual_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.post(
"/api/parental-leave/manual",
json={
"personal_data_uuid": str(_CALLER_PD_UUID),
"workspace_id": str(two_workspaces.ws_b.id),
"child_id": str(two_workspaces.child_a.id),
"total_days": 480,
"sjukpenning_left": 390,
"lagsta_niva_left": 90,
"dubbeldagar_left": 60,
"reserved_days_left": 90,
},
)
assert resp.status_code == 404
async def test_fetch_data_get_latest_foreign_pd(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
resp = await client.get(
f"/api/parental-leave/{two_workspaces.pd_b.uuid}/{two_workspaces.child_a.id}"
)
assert resp.status_code == 404
async def test_invitations_revoke_foreign_workspace(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
"""The invitation belongs to workspace B; the caller (workspace A)
must not be able to revoke it, even by guessing/enumerating its id.
"""
# `invited_by_user_id` FKs to a real `user` row.
user_repo = SQLiteUserRepository(connection)
inviter = await user_repo.create(
User(
id=uuid4(),
personal_data_uuid=two_workspaces.pd_b.uuid,
email="inviter-b@example.com",
hashed_password="not-a-real-hash",
created_at=datetime.now(),
)
)
inv_repo = SQLiteWorkspaceInvitationRepository(connection)
inv = await inv_repo.create(
WorkspaceInvitation(
id=uuid4(),
workspace_id=two_workspaces.ws_b.id,
token_hash="deadbeef",
invited_by_user_id=inviter.id,
status="pending",
expires_at=datetime.now() + timedelta(hours=1),
created_at=datetime.now(),
)
)
resp = await client.delete(f"/api/invitations/{inv.id}")
assert resp.status_code == 404
async def test_list_invitations_foreign_workspace(
self, client: AsyncClient, two_workspaces: _TwoWorkspaces
) -> None:
"""`GET /{workspace_id}/invitations` is the only way to discover an
id for `DELETE /api/invitations/{id}` -- it must not leak workspace
B's invitations (or their existence) to a caller in workspace A."""
resp = await client.get(f"/api/workspace/{two_workspaces.ws_b.id}/invitations")
assert resp.status_code == 404
async def test_list_invitations_never_returns_token_hash(
self, client: AsyncClient, connection: aiosqlite.Connection, two_workspaces: _TwoWorkspaces
) -> None:
user_repo = SQLiteUserRepository(connection)
inviter = await user_repo.create(
User(
id=uuid4(),
personal_data_uuid=_CALLER_PD_UUID,
email="inviter-a@example.com",
hashed_password="not-a-real-hash",
created_at=datetime.now(),
)
)
inv_repo = SQLiteWorkspaceInvitationRepository(connection)
await inv_repo.create(
WorkspaceInvitation(
id=uuid4(),
workspace_id=two_workspaces.ws_a.id,
token_hash="super-secret-hash",
invited_by_user_id=inviter.id,
status="pending",
expires_at=datetime.now() + timedelta(hours=1),
created_at=datetime.now(),
)
)
resp = await client.get(f"/api/workspace/{two_workspaces.ws_a.id}/invitations")
assert resp.status_code == 200
body = resp.json()
assert len(body) == 1
assert "token_hash" not in body[0]
assert "super-secret-hash" not in resp.text
Reference in New Issue View Git Blame Copy Permalink