Bootcamp i backendutveckling med FastAPI · Lektion

Säkerhet på radnivå och datapartitionering

Upprätthåll tydliga datagränser med PostgreSQL row-level security och klientbegränsade query guards.

Lektion 3 av 413 steg

Säkerhet på radnivå och datapartitionering är en gratis lektion i Bootcamp i backendutveckling med FastAPI på CoddyKit. Detta är lektion 3 av 4. Du kan läsa vilka 3 lektioner som helst i den här lärvägen kostnadsfritt i sin helhet – därefter låser CoddyKit PRO upp alla lektioner, plus praktisk övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Den ingår i lärvägen för Bootcamp i backendutveckling med FastAPI, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Varför tydliga tenant-gränser är viktiga

I en SaaS-tjänst med flera tenants är den mest katastrofala typen av buggar dataläckage mellan tenants: att tenant A läser tenant B:s rader. Ett enda saknat WHERE tenant_id = ? i en endpoint räcker för att läcka allt.

Det finns tre försvarslager:

Skydd i applikationen — varje fråga begränsas med tenant_id.
Row-level security (RLS) i databasen — PostgreSQL vägrar returnera rader som inte matchar en sessionspolicy, även om applikationen glömmer det.
Partitionering — separera tenant-data fysiskt för bättre prestanda och begränsad skadeomfattning.

Den här lektionen kombinerar alla tre i FastAPI + PostgreSQL. Grundregeln är: lita aldrig enbart på applikationen. RLS är skyddsnätet som omvandlar ett läckage till ett tomt resultat.

Tenant-modellen med gemensamt schema

Vi använder modellen med gemensamt schema: en uppsättning tabeller där varje rad innehåller en kolumn med tenant_id. Den är billigast att drifta och enklast att migrera, men har ingen isolering som standard — isoleringen är helt och hållet ditt ansvar.

Varje tenant-ägd tabell följer samma struktur: en indexerad främmande nyckel tenant_id, som helst även ingår i en sammansatt primärnyckel så att anrop mellan tenants blir omöjliga redan genom konstruktionen.

from sqlalchemy import Column, BigInteger, String, ForeignKey, Index
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class Invoice(Base):
    __tablename__ = "invoices"

    # Composite PK (tenant_id, id) makes a row globally addressable
    # only WITH its tenant -> no accidental cross-tenant lookups.
    tenant_id = Column(BigInteger, ForeignKey("tenants.id"), primary_key=True)
    id = Column(BigInteger, primary_key=True)
    customer = Column(String, nullable=False)
    amount_cents = Column(BigInteger, nullable=False)

    __table_args__ = (
        Index("ix_invoices_tenant", "tenant_id"),
    )

Identifiera tenant från begäran

Varje begäran måste kopplas till exakt en tenant innan någon fråga körs. Vanliga källor är en subdomän (acme.app.com), en header (X-Tenant-ID) eller — säkrast — ett claim i den autentiserade JWT:n, så att klienten inte kan förfalska det.

Föredra JWT-claimet. En header eller subdomän styrs av angriparen, medan ett signerat token-claim inte gör det. Exponera den identifierade tenant via ett FastAPI-beroende så att alla endpoints delar samma betrodda källa.

from fastapi import Depends, HTTPException, Request


async def get_current_tenant(request: Request) -> int:
    # Set earlier by the auth dependency after verifying the JWT signature.
    tenant_id = getattr(request.state, "tenant_id", None)
    if tenant_id is None:
        # Fail closed: no tenant context => refuse, never default to "all".
        raise HTTPException(status_code=401, detail="No tenant context")
    return tenant_id

Frågeskydd på applikationsnivå

Det första lagret är disciplinerad begränsning av frågor. Centralisera den så att ingen utvecklare behöver komma ihåg satsen WHERE tenant_id = ? manuellt. En tenant-begränsad session eller ett repository injicerar filtret automatiskt.

Här kapslar ett litet repository in SQLAlchemy och tvingar fram tenant-filtret vid varje läsning. Poängen är att göra den säkra vägen till standardvägen.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession


class InvoiceRepository:
    def __init__(self, session: AsyncSession, tenant_id: int):
        self.session = session
        self.tenant_id = tenant_id

    async def list(self):
        stmt = select(Invoice).where(Invoice.tenant_id == self.tenant_id)
        result = await self.session.execute(stmt)
        return result.scalars().all()

    async def get(self, invoice_id: int):
        stmt = select(Invoice).where(
            Invoice.tenant_id == self.tenant_id,
            Invoice.id == invoice_id,
        )
        return (await self.session.execute(stmt)).scalar_one_or_none()

Varför enbart applikationsskydd inte räcker

Skydd i applikationen kan misslyckas utan att det märks. En rå SQL-rapport, ett glömt filter i en admin-endpoint, en ORM-relation som läser över tenant-gränser eller en snabbfix av en junior utvecklare — allt detta kan kringgå repositoryt.

Du behöver ett lager som applikationen inte kan glömma. Det är PostgreSQL Row-Level Security: databasen lägger själv till ett osynligt predikat i varje fråga mot en skyddad tabell. Till och med SELECT * FROM invoices returnerar endast den aktuella tenantens rader.

Aktivera Row-Level Security i PostgreSQL

RLS fungerar genom att binda policies till en sessionsvariabel. Applikationen anger app.current_tenant i början av varje begäran, och policies jämför varje rads tenant_id med den.

Stegen är: aktivera RLS på tabellen och skapa sedan en USING-policy (styr synlighet vid läsningar, uppdateringar och borttagningar) samt en WITH CHECK-policy (styr vilka värden som får skrivas vid infogningar och uppdateringar).

-- Run once per protected table (migration)
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoices FORCE ROW LEVEL SECURITY;  -- applies even to table owner

CREATE POLICY tenant_isolation ON invoices
    USING (tenant_id = current_setting('app.current_tenant')::bigint)
    WITH CHECK (tenant_id = current_setting('app.current_tenant')::bigint);

Ange tenant-kontext per begäran

För att RLS ska fungera måste varje anslutning ha rätt app.current_tenant innan någon fråga körs. Med en anslutningspool är detta känsligt: en poolad anslutning återanvänds, så du måste ange variabeln när begäran börjar och återställa den när begäran slutar. Annars kan en läckt variabel orsaka läsningar över tenant-gränser.

Använd set_config(key, value, true) — flaggan true gör variabeln transaktionslokal, så att den återställs automatiskt när transaktionen avslutas. Det här är det säkraste mönstret med poolade anslutningar.

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


async def scoped_session(session: AsyncSession, tenant_id: int):
    # Transaction-local: third arg `true` ties the setting to the current tx,
    # so it cannot leak to the next request reusing this pooled connection.
    await session.execute(
        text("SELECT set_config('app.current_tenant', :tid, true)"),
        {"tid": str(tenant_id)},
    )
    return session

Koppla in RLS i ett FastAPI-beroende

Knyt ihop delarna: ett beroende returnerar en session som redan har fått tenant-kontexten angiven i en öppen transaktion. Endpoints kör sedan vanliga frågor medan RLS i bakgrunden upprätthåller isoleringen.

Observera att anslutningen inte får köras som PostgreSQL-superanvändare eller tabellägare utan FORCE ROW LEVEL SECURITY — superanvändare och ägare kringgår RLS som standard. Använd en dedikerad applikationsroll med låga behörigheter.

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant),
    session: AsyncSession = Depends(get_session),
):
    async with session.begin():  # one transaction => set_config stays scoped
        await session.execute(
            text("SELECT set_config('app.current_tenant', :tid, true)"),
            {"tid": str(tenant_id)},
        )
        yield session  # every query inside is RLS-filtered automatically

Partitionera efter tenant för skalning

När du har många tenants blir en enda enorm tabell en belastning: vacuum, svällande index och frågeplaner som påverkas av högljudda grannar. Deklarativ partitionering efter tenant_id (LIST eller HASH) delar upp tabellen i fysiska partitioner.

Fördelarna är att frågor som innehåller tenant_id får partition pruning (endast den relevanta partitionen genomsöks), underhåll körs per partition och att en tenant kan tas bort snabbt med DETACH/DROP i stället för en långsam massborttagning.

-- Hash partitioning spreads tenants across N partitions evenly
CREATE TABLE invoices (
    tenant_id  bigint NOT NULL,
    id         bigint NOT NULL,
    customer   text   NOT NULL,
    amount_cents bigint NOT NULL,
    PRIMARY KEY (tenant_id, id)
) PARTITION BY HASH (tenant_id);

CREATE TABLE invoices_p0 PARTITION OF invoices
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE invoices_p1 PARTITION OF invoices
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
-- ... p2, p3

RLS och partitionering tillsammans

RLS-policies på en partitionerad överordnad tabell ärvs automatiskt av alla partitioner i moderna versioner av PostgreSQL, så du definierar policyn en gång på den överordnade tabellen. Partition pruning och RLS-predikatet samverkar: frågeplaneraren begränsar sökningen till en partition och RLS filtrerar sedan raderna i den.

Välja LIST eller HASH:

HASH — jämn fördelning och inga hotspots, men du kan inte isolera en enda stor tenant.
LIST — placera specifika stora tenants i dedikerade partitioner; utmärkt för att isolera högljudda grannar och för säkerhetskopior per tenant.

Verifiera isolering i tester

Isolering måste testas som en säkerhetskontroll, inte tas för given. Skriv ett test som anger tenant A:s kontext, infogar rader, byter till tenant B och kontrollerar att B inte ser något. Den här rena Python-modellen visar exakt den invarians som din RLS-policy upprätthåller.

class RLSSimulator:
    """Mimics how a USING policy filters rows by session tenant."""

    def __init__(self):
        self.rows = []
        self.current_tenant = None

    def set_tenant(self, tenant_id):
        self.current_tenant = tenant_id

    def insert(self, row):
        # WITH CHECK: can only write rows for the active tenant.
        if row["tenant_id"] != self.current_tenant:
            raise PermissionError("WITH CHECK violation")
        self.rows.append(row)

    def select_all(self):
        # USING: only rows matching the active tenant are visible.
        return [r for r in self.rows if r["tenant_id"] == self.current_tenant]


db = RLSSimulator()
db.set_tenant(1)
db.insert({"tenant_id": 1, "id": 10, "customer": "Acme"})

db.set_tenant(2)
db.insert({"tenant_id": 2, "id": 11, "customer": "Globex"})

print("Tenant 2 sees:", db.select_all())
db.set_tenant(1)
print("Tenant 1 sees:", db.select_all())

Snabbkontroll: poolade anslutningar och RLS

Du angav app.current_tenant med set_config på en anslutning från en pool, men använde formen på sessionsnivå (det tredje argumentet false) i stället för den transaktionslokala formen. Vilken produktionskonsekvens är mest sannolik?

Besvara frågan i den interaktiva lektionen

Sammanfattning: försvar i flera lager för tenant-data

Ni har nu en strategi med flera isoleringslager för FastAPI med flera tenants:

Skydd i appen — ett tenant-avgränsat repository gör säkra frågor till standardalternativet, men det kan glömmas bort.
Row-Level Security — ENABLE + FORCE RLS, en USING/WITH CHECK-policy baserad på app.current_tenant samt en roll med låga behörigheter, så att databasen inte kan returnera rader från andra tenants.
Transaktionslokal kontext — ange tenant med set_config(..., true) i den transaktion som hanterar begäran, så att poolade anslutningar aldrig läcker kontext.
Partitionering — använd LIST/HASH efter tenant_id för bortgallring, underhåll per tenant och snabb avveckling; RLS-policyer ärvs av alla partitioner.

Grundtanken är att behandla isolering som en säkerhetskontroll med överlappande lager och verifiera den med tester. Lita aldrig enbart på applikationen.

Gratis att börja

Lär dig Bootcamp i backendutveckling med FastAPI med en AI-lärare – gratis

Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.

Kurser: 21
Lektioner: 84

Börja lära dig gratis Visa hela kursen

Vanliga frågor

Är lektionen ”Säkerhet på radnivå och datapartitionering” gratis?

Ja – du kan läsa vilka 3 lektioner som helst i lärvägen Bootcamp i backendutveckling med FastAPI, inklusive ”Säkerhet på radnivå och datapartitionering”, kostnadsfritt i sin helhet här på webben. Därefter låser CoddyKit PRO upp alla lektioner, plus interaktiv övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Vad lär jag mig i ”Säkerhet på radnivå och datapartitionering”?

Upprätthåll tydliga datagränser med PostgreSQL row-level security och klientbegränsade query guards. Ni övar på Bootcamp i backendutveckling med FastAPI med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.

Behöver jag någon erfarenhet för att börja lära mig Bootcamp i backendutveckling med FastAPI?

Du behöver inga förkunskaper. Utbildningen i Bootcamp i backendutveckling med FastAPI på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 3 av 4.

Hur lång tid tar lektionen ”Säkerhet på radnivå och datapartitionering”?

De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.

Kan jag skriva och köra kod i den här Bootcamp i backendutveckling med FastAPI-lektionen?

Ja. Varje Bootcamp i backendutveckling med FastAPI-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.

Alla lektioner i den här kursen

Strategier och avvägningar för klientisolering
Identifiering av klientkontext och middleware
Säkerhet på radnivå och datapartitionering
Användningsmätning, kvoter och billing-hooks

← Tillbaka till Bootcamp i backendutveckling med FastAPI