The Security Baseline

Security is not a feature you add. It is a baseline you establish and then maintain. Every chapter after this one builds on the assumption that the security baseline is in place. HTTPS is enforced. Rate limiting is active. Input validation rejects malformed data before it reaches business logic. SQL queries are parameterized. Multi-tenant data isolation is enforced at the database level.

This is not a comprehensive application security guide. It is the minimum set of defenses that a production SaaS must have before accepting its first user. These defenses block the most common attack categories: injection, broken authentication, and data exposure. Advanced topics like penetration testing, dependency vulnerability scanning, and incident response are beyond the scope of a book about shipping a product, but they are worth pursuing once the product is live.

The Feature

Marketflow rejects malformed input, limits request rates on sensitive endpoints, enforces HTTPS on all connections, prevents SQL injection by construction, and isolates each organizer’s data at the database level using PostgreSQL row-level security policies.

The Decision

The security baseline uses existing infrastructure wherever possible. HTTPS is handled by Cloudflare, not by self-managed certificates. Rate limiting uses a Redis-backed middleware, not an application-level counter. RLS policies run in PostgreSQL, not in application code. The principle is the same as the rest of the book: use what exists, write only what must be custom.

The Implementation

HTTPS Enforcement

Cloudflare terminates TLS at its edge. Traffic between the user’s browser and Cloudflare is encrypted. Traffic between Cloudflare and the Hetzner VPS travels through the Cloudflare Tunnel, which is also encrypted. No self-signed certificates. No Let’s Encrypt renewal cron jobs.

In the Cloudflare dashboard, set SSL/TLS encryption mode to “Full (strict)” and enable “Always Use HTTPS.” This redirects all HTTP requests to HTTPS at Cloudflare’s edge before they reach the application.

Rate Limiting

# backend/app/middleware/rate_limit.py
import time
from collections.abc import Callable

from fastapi import HTTPException, Request, Response
from redis.asyncio import Redis
from starlette.middleware.base import BaseHTTPMiddleware

from app.config import settings


class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: Callable, redis: Redis) -> None:
        super().__init__(app)
        self.redis = redis

    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # Only rate limit specific paths
        if not self._should_rate_limit(request.url.path):
            return await call_next(request)

        client_ip = request.client.host if request.client else "unknown"
        key = f"rate_limit:{client_ip}:{request.url.path}"

        limit, window = self._get_limit(request.url.path)

        current = await self.redis.get(key)
        if current and int(current) >= limit:
            raise HTTPException(
                status_code=429,
                detail="Too many requests. Try again later.",
            )

        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, window)
        await pipe.execute()

        return await call_next(request)

    def _should_rate_limit(self, path: str) -> bool:
        rate_limited_prefixes = [
            "/api/auth/",
            "/api/public/",
            "/api/webhooks/",
        ]
        return any(path.startswith(prefix) for prefix in rate_limited_prefixes)

    def _get_limit(self, path: str) -> tuple[int, int]:
        """Returns (max_requests, window_seconds)."""
        if path.startswith("/api/auth/"):
            return 10, 60  # 10 requests per minute for auth
        if path.startswith("/api/webhooks/"):
            return 100, 60  # 100 per minute for webhooks
        return 30, 60  # 30 per minute for public endpoints

# backend/app/main.py (additions)
from redis.asyncio import Redis
from app.middleware.rate_limit import RateLimitMiddleware

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = Redis.from_url(settings.redis_url)
    yield
    await app.state.redis.close()

# After creating the app:
app.add_middleware(RateLimitMiddleware, redis=app.state.redis)

Auth endpoints get 10 requests per minute per IP. This prevents brute force login attempts without blocking legitimate users who mistype their password. The limit is generous enough for normal use and tight enough to make automated attacks impractical.

Input Validation with Pydantic v2

# backend/app/schemas/market.py
from pydantic import BaseModel, ConfigDict, Field


class MarketCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")

    name: str = Field(min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=5000)
    location: str = Field(min_length=1, max_length=500)

Three lines of configuration prevent three categories of attack:

• extra="forbid" rejects requests with unexpected fields. Without this, an attacker can send {"name": "...", "organizer_id": "someone-elses-id"} and overwrite fields they should not control.
• max_length prevents denial-of-service via oversized payloads. Without length limits, a single request with a 100 MB description field consumes memory and database storage.
• min_length=1 prevents empty strings that pass required validation but create records with blank names.

# TRAP: Accepting extra fields
class MarketCreate(BaseModel):
    # Default is extra="ignore", which silently drops unknown fields.
    # extra="allow" stores them, which is worse.
    name: str
    description: str | None = None
    location: str
    # An attacker sends: {"name": "...", "location": "...", "is_active": false}
    # With SQLAlchemy, this could overwrite database fields if the model
    # is updated with **body.model_dump()

# SAFE: Forbid extra fields, constrain all strings
class MarketCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")

    name: str = Field(min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=5000)
    location: str = Field(min_length=1, max_length=500)

SQL Injection Prevention

SQLAlchemy’s query builder generates parameterized queries by construction. There is no way to accidentally concatenate user input into a SQL string when using the select(), where(), and insert() builders.

# TRAP: String concatenation in a raw SQL query
from sqlalchemy import text

async def search_markets(db: AsyncSession, query: str):
    result = await db.execute(
        text(f"SELECT * FROM markets WHERE name LIKE '%{query}%'")
    )
    # query = "'; DROP TABLE markets; --" destroys the database

# SAFE: Parameterized query
async def search_markets(db: AsyncSession, query: str):
    result = await db.execute(
        text("SELECT * FROM markets WHERE name LIKE :query"),
        {"query": f"%{query}%"},
    )

# SAFEST: Use the SQLAlchemy query builder
async def search_markets(db: AsyncSession, query: str):
    result = await db.execute(
        select(Market).where(Market.name.ilike(f"%{query}%"))
    )

The SQLAlchemy query builder is the default for all Marketflow queries. Raw text() queries are used only when the builder cannot express the query, which has not happened yet in this book and is unlikely to happen in a typical SaaS application.

Row-Level Security

-- Run in Supabase SQL editor or as an Alembic migration

-- Enable RLS on all tenant-scoped tables
ALTER TABLE markets ENABLE ROW LEVEL SECURITY;
ALTER TABLE stalls ENABLE ROW LEVEL SECURITY;
ALTER TABLE applications ENABLE ROW LEVEL SECURITY;
ALTER TABLE bookings ENABLE ROW LEVEL SECURITY;
ALTER TABLE market_days ENABLE ROW LEVEL SECURITY;

-- Organizers can only see their own markets
CREATE POLICY organizer_markets ON markets
  FOR ALL
  USING (organizer_id = auth.uid());

-- Stalls are visible if the user owns the market
CREATE POLICY organizer_stalls ON stalls
  FOR ALL
  USING (
    market_id IN (
      SELECT id FROM markets WHERE organizer_id = auth.uid()
    )
  );

-- Applications are visible to the market organizer and the vendor
CREATE POLICY application_access ON applications
  FOR ALL
  USING (
    market_id IN (
      SELECT id FROM markets WHERE organizer_id = auth.uid()
    )
    OR
    vendor_id IN (
      SELECT id FROM vendors WHERE user_id = auth.uid()
    )
  );

-- Vendors can see their own profile
CREATE POLICY vendor_own_profile ON vendors
  FOR ALL
  USING (user_id = auth.uid());

RLS policies are the database-level safety net. Even if the application code has a bug that omits the tenant filter, PostgreSQL enforces the policy. A query that accidentally selects all markets returns only the current user’s markets because the RLS policy adds WHERE organizer_id = auth.uid() to every query.

The Trap

RLS policies use auth.uid(), which is a Supabase function that extracts the user ID from the JWT passed in the database connection. When the FastAPI backend connects to Supabase’s PostgreSQL with the service key, RLS is bypassed because the service role has superuser privileges. This is intentional for administrative operations. For user-facing operations, the backend must set the JWT in the database session:

# TRAP: Service key bypasses RLS
# All queries run as superuser, RLS policies are ignored
engine = create_async_engine(supabase_connection_string)

# SAFE: Set the user's JWT for RLS enforcement
async def get_db_with_rls(
    request: Request,
) -> AsyncSession:
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    async with async_session() as session:
        await session.execute(
            text("SET LOCAL request.jwt.claim.sub = :sub"),
            {"sub": token},
        )
        yield session

The simpler approach, which Marketflow uses, is to enforce tenant isolation at the application level (Chapter 5’s get_market_for_organizer dependency) and use RLS as a defense-in-depth layer. Both approaches prevent data leakage. Using both means a single bug in either layer does not expose data.

The Cost

Security Component	Cost	Implementation Time
HTTPS (Cloudflare)	$0	5 minutes
Rate limiting (Redis)	$0 (Redis already running)	2 hours
Input validation (Pydantic)	$0	Ongoing (per schema)
SQL injection prevention	$0 (SQLAlchemy)	Already done
RLS policies	$0 (Supabase)	3 hours

The total implementation cost is approximately one day of work. The alternative, recovering from a security breach, costs weeks of engineering time, legal consultation, customer notifications, and reputation damage that cannot be quantified.