fix(security/dsgvo): RLS für hours_payouts nachgezogen (Mandantentrennung)

hours_payouts war die einzige firmenbezogene Tabelle ohne Row-Level-Security
(Migration 0030 hatte keinen RLS-Block). Die Endpunkte filtern zwar applikativ
nach company_id (kein akutes Leck), aber das DB-seitige Schutznetz – das im
ganzen System (FORCE RLS, 0024/0034) die Mandantentrennung garantiert – fehlte.

Migration 0039 aktiviert ENABLE+FORCE RLS + company_id-Policies (analog 0024).
conftest.py-RLS-Replik + neuer Cross-Tenant-Test test_rls_hours_payouts_tenant_isolation.
Verifiziert auf 137+164 (rls=True, force=True, 4 Policies). 191/191 Tests grün.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-23 14:06:19 +02:00
co-authored by Claude Opus 4.8
parent 2b74f95e68
commit 3034c6c55a
3 changed files with 72 additions and 1 deletions
+28
View File
@@ -188,3 +188,31 @@ async def test_rls_insert_blocked_for_wrong_tenant(db_session):
f"Unerwarteter Fehler (kein RLS-Fehler): {e}"
finally:
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
async def test_rls_hours_payouts_tenant_isolation(client: AsyncClient, db_session):
"""hours_payouts: Mandant B darf Auszahlungen von Mandant A nicht sehen (DB-Ebene)."""
a = await register_company(client, "PAYOUT-A")
b = await register_company(client, "PAYOUT-B")
cid_a, uid_a = str(a["user"]["company_id"]), str(a["user"]["id"])
cid_b = str(b["user"]["company_id"])
# Auszahlung für Mandant A direkt einfügen (unter Bypass)
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
await db_session.execute(text(
"INSERT INTO hours_payouts (id, company_id, user_id, hours, created_by) "
"VALUES (gen_random_uuid(), :cid, :uid, 8.0, :uid)"
), {"cid": cid_a, "uid": uid_a})
await db_session.commit()
async def payouts_as(cid: str) -> set[str]:
await db_session.execute(text("SET LOCAL app.bypass_rls = 'off'"))
await db_session.execute(text(f"SET LOCAL app.company_id = '{cid}'"))
res = await db_session.execute(text("SELECT company_id FROM hours_payouts"))
seen = {str(r[0]) for r in res.fetchall()}
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
return seen
assert cid_a in await payouts_as(cid_a), "Mandant A sieht eigene Auszahlung nicht"
assert cid_a not in await payouts_as(cid_b), \
"RLS BLOCKIERT NICHT: Mandant B sieht Auszahlung von Mandant A!"