Row Level Security

[sternalon]

Hi All.

I have set up a multi-tenant app where different companies (tenants) can use the app.

To ensure that one company can not access the data of another by mistake, I have implemented row level security. This was done by creating a get_session() method, which only allows access to rows that contain information for that tenant (method is pasted below).

I am now hoping to use Starlette Admin all allow users from a given company to access their data (but no one else's data)

Is there a way with Starlette-admin to intercept the session used to access the data, and apply this row level security?

Alternately, can you suggest other ways to ensure that users from one tenant can not access data from another tenant?

Thanks!

async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
try:
tenant_id = get_global_tenant_id()
query = text(f"SET app.current_tenant={tenant_id};")
await session.execute(text("SET SESSION ROLE tenant_user;"))
await session.execute(query)
yield session
except:
await session.rollback()
raise
finally:
await session.execute(text("RESET ROLE;"))
await session.commit()

[sternalon]

I just found the get_session method in starlette_admin:
https://github.com/jowilf/starlette-admin/blob/main/starlette_admin/contrib/sqla/middleware.py

I am going to try override this, and I suspect that should work. I will update here if I am able to get it working.

[sternalon]

I managed to get this working. I found that the easiest way was to top apply the row level security within the is_authenticated method within the AuthProvider. I think this first since it an authorization (or sort of).

My is_authenticated method looks something like this:

async def is_authenticated(self, request: Request) -> bool:
try:
token = request.session.get("token")
tenant_id = validate_token(token)
is_valid = tenant_id is not None

        if is_valid:
            global_tenant_id.set(tenant_id)
            username = request.session["username"]
            permissions = request.session["permissions"]
            avatar = None
            user = create_user_obj(username, permissions, avatar)
            request.state.user = create_user_obj(username, permissions, avatar)

            await self.apply_row_level_security(request, tenant_id)

            return True
        return False
    except:
        return False

And my apply_row_level_security method is:

async def apply_row_level_security(self, request, tenant_id):
session = request.state.session
company = await find_company_by_global_tenant_id(session)
query = text(f"SET app.current_tenant={company.id};")
await session.execute(text("SET SESSION ROLE tenant_user;"))
await session.execute(query)
request.state.session = session

All seems to be working nicely now.

[regisin]

Could you explain the global_tenant_id.set(tenant_id) and find_company_by_global_tenant_id?

Edit: is this it? https://adityamattos.com/multi-tenancy-in-python-fastapi-and-sqlalchemy-using-postgres-row-level-security