81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
"""C3.1 — give user_roles a tenant_id, so isolation is structural
|
|
|
|
`user_roles` was created without one. That is not a cosmetic omission: both
|
|
isolation layers *derive* their coverage from the presence of that column —
|
|
`app/core/tenant_filter.py` scopes any model that has it, and the RLS policy
|
|
compares it against the session variable. A table without it is protected by
|
|
neither, no matter how careful the queries are.
|
|
|
|
It is the same shape of gap as `tenants`, which is where "any holder of
|
|
`superadmin.tenant.delete` can delete any tenant" lived: a table outside the
|
|
mechanism, guarded only by whatever each call site remembered to write. Three
|
|
places currently remember. Three is a number that only goes up, and the failure
|
|
mode when one forgets is a grant resolving across tenants.
|
|
|
|
After this the explicit predicates stay — they are cheap and they document the
|
|
intent — but they stop being the only thing standing there.
|
|
|
|
**Backfill from the user, not the role.** `roles.tenant_id` is nullable: a null
|
|
means a shared/global role template, and a grant of a shared role still belongs
|
|
to the tenant of the person holding it. Rows for tenant-less users (superadmins)
|
|
stay null, which the RLS policy already treats as globally visible — the same
|
|
way their `users` row is treated.
|
|
|
|
Revision ID: c1_1_user_roles_tenant_id
|
|
Revises: c1_0_org_units_and_scoped_roles
|
|
"""
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
|
|
revision = "c1_1_user_roles_tenant_id"
|
|
down_revision = "c1_0_org_units_and_scoped_roles"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
TENANT_TABLES = ["user_roles"]
|
|
POLICY = "tenant_isolation"
|
|
USING = """
|
|
coalesce(current_setting('docqube.bypass', true), '') = 'on'
|
|
OR tenant_id IS NULL
|
|
OR tenant_id = nullif(current_setting('docqube.tenant_id', true), '')::uuid
|
|
"""
|
|
|
|
|
|
def upgrade() -> None:
|
|
op.add_column(
|
|
"user_roles",
|
|
sa.Column(
|
|
"tenant_id",
|
|
UUID(as_uuid=True),
|
|
sa.ForeignKey("tenants.id", ondelete="CASCADE"),
|
|
nullable=True,
|
|
),
|
|
)
|
|
op.create_index("ix_user_roles_tenant_id", "user_roles", ["tenant_id"])
|
|
|
|
op.execute(
|
|
"""
|
|
UPDATE user_roles ur
|
|
SET tenant_id = u.tenant_id
|
|
FROM users u
|
|
WHERE u.id = ur.user_id
|
|
"""
|
|
)
|
|
|
|
# Left nullable on purpose. A superadmin has no tenant, so their grants have
|
|
# none either; forcing NOT NULL would mean inventing one.
|
|
op.execute(f"ALTER TABLE user_roles ENABLE ROW LEVEL SECURITY")
|
|
op.execute(f"ALTER TABLE user_roles FORCE ROW LEVEL SECURITY")
|
|
op.execute(f"DROP POLICY IF EXISTS {POLICY} ON user_roles")
|
|
op.execute(f"CREATE POLICY {POLICY} ON user_roles USING ({USING}) WITH CHECK ({USING})")
|
|
|
|
|
|
def downgrade() -> None:
|
|
op.execute(f"DROP POLICY IF EXISTS {POLICY} ON user_roles")
|
|
op.execute("ALTER TABLE user_roles NO FORCE ROW LEVEL SECURITY")
|
|
op.execute("ALTER TABLE user_roles DISABLE ROW LEVEL SECURITY")
|
|
op.drop_index("ix_user_roles_tenant_id", table_name="user_roles")
|
|
op.drop_column("user_roles", "tenant_id")
|