patrickandClaude Opus 4.8
3034c6c55a
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>
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""RLS für hours_payouts nachziehen (Mandantentrennung / DSGVO)
|
|
|
|
Revision ID: 0039
|
|
Revises: 0038
|
|
Create Date: 2026-06-23
|
|
|
|
hours_payouts (Migration 0030) hatte als einzige firmenbezogene Tabelle keine
|
|
Row-Level-Security. Die Endpunkte filtern zwar applikativ nach company_id, aber
|
|
das DB-seitige Schutznetz (FORCE RLS, analog 0024) fehlte. Hier nachgezogen.
|
|
"""
|
|
from alembic import op
|
|
from sqlalchemy import text
|
|
|
|
revision = "0039"
|
|
down_revision = "0038"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
_BYPASS = "COALESCE(current_setting('app.bypass_rls', true), 'off') = 'on'"
|
|
_CID = "company_id = NULLIF(current_setting('app.company_id', true), '')::uuid"
|
|
_USING = f"({_BYPASS} OR {_CID})"
|
|
|
|
|
|
def _exec(sql: str) -> None:
|
|
op.execute(text(sql))
|
|
|
|
|
|
def upgrade() -> None:
|
|
_exec("ALTER TABLE hours_payouts ENABLE ROW LEVEL SECURITY")
|
|
_exec("ALTER TABLE hours_payouts FORCE ROW LEVEL SECURITY")
|
|
for cmd in ("select", "insert", "update", "delete"):
|
|
_exec(f"DROP POLICY IF EXISTS rls_hours_payouts_{cmd} ON hours_payouts")
|
|
_exec(f"CREATE POLICY rls_hours_payouts_select ON hours_payouts FOR SELECT USING {_USING}")
|
|
_exec(f"CREATE POLICY rls_hours_payouts_insert ON hours_payouts FOR INSERT WITH CHECK {_USING}")
|
|
_exec(f"CREATE POLICY rls_hours_payouts_update ON hours_payouts FOR UPDATE USING {_USING} WITH CHECK {_USING}")
|
|
_exec(f"CREATE POLICY rls_hours_payouts_delete ON hours_payouts FOR DELETE USING {_USING}")
|
|
|
|
|
|
def downgrade() -> None:
|
|
for cmd in ("select", "insert", "update", "delete"):
|
|
_exec(f"DROP POLICY IF EXISTS rls_hours_payouts_{cmd} ON hours_payouts")
|
|
_exec("ALTER TABLE hours_payouts NO FORCE ROW LEVEL SECURITY")
|
|
_exec("ALTER TABLE hours_payouts DISABLE ROW LEVEL SECURITY")
|