80 lines
3 KiB
Python
80 lines
3 KiB
Python
|
"""
|
||
|
|
Bevis at flertenant-isolasjonen (RLS på organization_id) faktisk holder --
|
||
|
|
automatiserer det som til nå kun er sjekket manuelt via test_isolation.sql.
|
||
|
|
|
||
|
|
Kjører som `teecup_app`-rollen (NOSUPERUSER/NOBYPASSRLS), akkurat som
|
||
|
|
produksjon -- ikke som en superbruker som ville hoppet over RLS uansett.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app import db as app_db
|
||
|
|
from app.auth import CurrentUser, get_authorized_org
|
||
|
|
from app.errors import app_error # noqa: F401 (dokumenterer avhengigheten)
|
||
|
|
from fastapi import HTTPException
|
||
|
|
|
||
|
|
from tests.conftest import add_membership, create_org, create_user
|
||
|
|
|
||
|
|
|
||
|
|
async def test_org_scoped_read_never_leaks_across_orgs(pool):
|
||
|
|
org_a = await create_org()
|
||
|
|
org_b = await create_org()
|
||
|
|
|
||
|
|
# Les ALLE rader i en org-scopet tabell mens app.current_org peker på A --
|
||
|
|
# ingen eksplisitt WHERE organization_id i selve spørringen. Hvis RLS
|
||
|
|
# ikke virket ville dette returnert BEGGE organisasjonene.
|
||
|
|
async with app_db.org_connection(org_a) as conn:
|
||
|
|
rows = await conn.fetch("SELECT id::text FROM organization")
|
||
|
|
|
||
|
|
ids = {r["id"] for r in rows}
|
||
|
|
assert org_a in ids
|
||
|
|
assert org_b not in ids
|
||
|
|
|
||
|
|
|
||
|
|
async def test_org_scoped_write_cannot_touch_other_org(pool):
|
||
|
|
org_a = await create_org()
|
||
|
|
org_b = await create_org()
|
||
|
|
|
||
|
|
async with app_db.org_connection(org_a) as conn:
|
||
|
|
result = await conn.execute(
|
||
|
|
"UPDATE organization SET name = 'kapret' WHERE id = $1", org_b
|
||
|
|
)
|
||
|
|
# asyncpg execute() returnerer "UPDATE <n>" -- n MÅ være 0: RLS sin
|
||
|
|
# USING-klausul filtrerer bort org B-raden før UPDATE-en i det hele tatt
|
||
|
|
# vurderer den, selv om spørringen eksplisitt ba om akkurat den raden.
|
||
|
|
assert result == "UPDATE 0"
|
||
|
|
|
||
|
|
async with app_db.org_connection(org_b) as conn:
|
||
|
|
row = await conn.fetchrow("SELECT name FROM organization WHERE id = $1", org_b)
|
||
|
|
assert row["name"] != "kapret"
|
||
|
|
|
||
|
|
|
||
|
|
async def test_no_org_context_set_yields_zero_rows_not_all_rows(pool):
|
||
|
|
"""Regresjonsvern for migrasjon 005 (RLS NULL-guard): en tilkobling UTEN
|
||
|
|
`app.current_org` satt i det hele tatt skal se INGENTING i en org-scopet
|
||
|
|
tabell -- IKKE (feilaktig) alt, som ville skjedd hvis policyen
|
||
|
|
sammenlignet mot en tom streng i stedet for å garantere NULL != noe."""
|
||
|
|
await create_org()
|
||
|
|
await create_org()
|
||
|
|
|
||
|
|
async with app_db.plain_connection() as conn:
|
||
|
|
rows = await conn.fetch("SELECT id FROM organization")
|
||
|
|
assert rows == []
|
||
|
|
|
||
|
|
|
||
|
|
async def test_membership_verified_before_org_context_is_ever_set(pool):
|
||
|
|
org = await create_org()
|
||
|
|
member = await create_user()
|
||
|
|
outsider = await create_user()
|
||
|
|
await add_membership(org, member)
|
||
|
|
|
||
|
|
# Medlem: får org_id tilbake.
|
||
|
|
result = await get_authorized_org(org, user=CurrentUser(user_id=member))
|
||
|
|
assert result == org
|
||
|
|
|
||
|
|
# Ikke-medlem: 403, IKKE en stille tom respons -- og det skjer FØR
|
||
|
|
# noen org-scopet spørring i det hele tatt kjøres.
|
||
|
|
with pytest.raises(HTTPException) as exc_info:
|
||
|
|
await get_authorized_org(org, user=CurrentUser(user_id=outsider))
|
||
|
|
assert exc_info.value.status_code == 403
|