Validating Spatial Scope Claims in FastAPI Dependencies

← Back to JWT Authentication for Spatial Scopes

Build one reusable dependency that confirms a requested geometry or bounding box falls inside the token’s spatial scope, returns 403 otherwise, and never verifies the same JWT twice in a single request.

# Context & when to use

The scope check belongs in a FastAPI dependency, not in each route handler, for one blunt reason: a check that lives in the handler is a check a future handler will forget. A dependency attaches to every route that declares it, runs before the handler body, and short-circuits the request with 403 if the requested area is out of bounds — so the expensive PostGIS read never runs for an unauthorised caller. This is the enforcement half of the pattern whose token design is covered in JWT authentication for spatial scopes; here we focus purely on the FastAPI wiring: decode once, cache, compare, fail closed.

Use a dedicated scope dependency whenever more than one endpoint reads geometry, when the requested area arrives in different shapes (a bbox query, an uploaded polygon, a single point), or when the scope encoding might change — bbox today, H3 cell set tomorrow — and you want that swap to touch one function. Prefer a plain inline check only for a throwaway single-route service where reuse and composition buy you nothing.

# Dependency composition

The chain is four dependencies deep. Each layer does one thing and hands a narrower, validated value to the next; FastAPI resolves the graph and caches each dependency’s result within the request.

# Runnable implementation

One module with the full chain. current_claims verifies and memoises; require_geometry_in_scope is a dependency factory so the same logic guards a bbox route, a point route, or a polygon upload.

# app/security/scope_dep.py
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
import jwt
from jwt import ExpiredSignatureError, InvalidTokenError

from app.dependencies import get_db

oauth2 = OAuth2PasswordBearer(tokenUrl="token")
PUBLIC_KEY = open("jwt-public.pem").read()


async def current_claims(request: Request, token: str = Depends(oauth2)) -> dict:
    """Decode the JWT exactly once; reuse the result for every dependency
    in the same request via request.state (FastAPI also caches the dependency,
    but state makes the intent explicit and survives sub-dependency reuse)."""
    if (cached := getattr(request.state, "claims", None)) is not None:
        return cached
    try:
        claims = jwt.decode(
            token, PUBLIC_KEY,
            algorithms=["RS256"],                     # pinned; never "none"
            audience="geospatial-api",
            issuer="https://auth.geospatial-api.com",
            options={"require": ["exp", "iss", "aud", "sub"]},
            leeway=10,
        )
    except ExpiredSignatureError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Token expired")
    except InvalidTokenError as exc:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"Invalid token: {exc}")
    request.state.claims = claims
    return claims


def _scope_to_geometry_sql(claim: dict) -> tuple[str, dict]:
    """Return an SQL fragment producing the scope geometry + its bind params.
    Supports bbox and WKT-ref today; extend the dispatch for h3/geohash."""
    if not claim:                                      # null / missing => fail closed
        raise HTTPException(status.HTTP_403_FORBIDDEN, "No spatial scope on token")
    enc, srid = claim.get("encoding"), int(claim.get("srid", 4326))
    if enc == "bbox":
        # union of envelopes -> one MultiPolygon scope geometry
        regions = claim["regions"]
        parts = ",".join(
            f"ST_MakeEnvelope({b[0]},{b[1]},{b[2]},{b[3]},{srid})" for b in regions
        )
        return f"ST_Collect(ARRAY[{parts}])", {}
    if enc == "wkt":
        return "ST_GeomFromText(:scope_wkt, :srid)", {
            "scope_wkt": claim["regions"][0], "srid": srid}
    raise HTTPException(status.HTTP_403_FORBIDDEN, f"Unsupported encoding: {enc}")


def require_geometry_in_scope(request_geom_sql: str):
    """Dependency factory. `request_geom_sql` is an SQL expression (using bind
    params supplied by the route) that yields the requested geometry."""
    async def _dep(
        claims: dict = Depends(current_claims),
        db: AsyncSession = Depends(get_db),
        request: Request = None,
    ):
        scope_sql, scope_params = _scope_to_geometry_sql(claims.get("geo_scope"))
        # bbox binds come from the route via request.query_params
        qp = dict(request.query_params)
        params = {**scope_params,
                  **{k: float(qp[k]) for k in ("minx", "miny", "maxx", "maxy")
                     if k in qp}}
        row = await db.execute(text(f"""
            SELECT ST_Within(({request_geom_sql}), ({scope_sql})) AS ok
        """), params)
        if not row.scalar():
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                "Requested geometry is outside your authorised spatial scope.",
            )
    return _dep


# --- usage: guard a bbox route ------------------------------------------------
from fastapi import FastAPI, Query
app = FastAPI()

bbox_in_scope = require_geometry_in_scope(
    "ST_MakeEnvelope(:minx, :miny, :maxx, :maxy, 4326)"
)

@app.get("/features/bbox", dependencies=[Depends(bbox_in_scope)])
async def features_in_bbox(
    minx: float = Query(...), miny: float = Query(...),
    maxx: float = Query(...), maxy: float = Query(...),
    db: AsyncSession = Depends(get_db),
):
    # Reaches here only if the bbox passed the scope gate.
    result = await db.execute(text("""
        SELECT id, ST_AsGeoJSON(geom)::jsonb AS geometry
        FROM spatial_features
        WHERE geom && ST_MakeEnvelope(:minx,:miny,:maxx,:maxy,4326)
          AND ST_Intersects(geom, ST_MakeEnvelope(:minx,:miny,:maxx,:maxy,4326))
        ORDER BY id LIMIT 200
    """), {"minx": minx, "miny": miny, "maxx": maxx, "maxy": maxy})
    return {"type": "FeatureCollection",
            "features": [{"type": "Feature", "id": r.id, "geometry": r.geometry}
                         for r in result.mappings().all()]}

The gate uses the same GiST-friendly bounding-box query mechanics as the data route — ST_Within for full containment. Swap it for ST_Intersects if a partly-overlapping request should be allowed.

# Key parameters & options

# Gotchas & failure modes

• Checking scope after the DB query. The single worst mistake: fetching features and filtering by scope in Python. An out-of-scope request still runs the full query, holds a connection, and may leak result size via timing. The gate must be a dependency that runs before the handler body — as above — so an unauthorised bbox never reaches spatial_features.
• geo_scope of null silently allowing everything. If _scope_to_geometry_sql returned an unbounded geometry for a missing claim, every token would read the whole planet. Fail closed: null or missing → 403. Add a test that a stripped-scope token is rejected.
• ExpiredSignatureError from a naive exp. exp is a UTC epoch. An issuer that builds it from datetime.now() (local, naive) produces tokens that expire hours early or late depending on the server’s timezone. Mint from datetime.now(UTC) and keep leeway small; a large leeway is not a fix for a timezone bug.
• ERROR: Operation on mixed SRID geometries. The scope claim’s srid differs from the requested geometry’s SRID. Normalise both to one SRID inside _scope_to_geometry_sql (reproject the scope with ST_Transform) rather than letting the mismatch reach ST_Within.
• Dependency cache assumptions across background tasks. request.state lives for the request only. If you enqueue a background task that re-checks scope, re-decode there — the cached claims object is not available once the response is sent.

# Verification

# In-scope bbox → 200
curl -s -o /dev/null -w "%{http_code}\n" -H "Authorization: Bearer $TOKEN" \
  "http://localhost:8000/features/bbox?minx=-2.6&miny=51.35&maxx=-2.4&maxy=51.45"
# → 200

# Out-of-scope bbox → 403
curl -s -o /dev/null -w "%{http_code}\n" -H "Authorization: Bearer $TOKEN" \
  "http://localhost:8000/features/bbox?minx=-0.2&miny=51.45&maxx=0.0&maxy=51.55"
# → 403

# assert the scope gate runs before the handler and fails closed on null scope
import pytest
from httpx import AsyncClient, ASGITransport
from app.security.scope_dep import app

@pytest.mark.asyncio
async def test_null_scope_is_forbidden(token_without_geo_scope):
    async with AsyncClient(transport=ASGITransport(app=app),
                           base_url="http://t") as c:
        r = await c.get("/features/bbox",
                        params={"minx": 0, "miny": 0, "maxx": 1, "maxy": 1},
                        headers={"Authorization": f"Bearer {token_without_geo_scope}"})
    assert r.status_code == 403

# Related

• JWT Authentication for Spatial Scopes — the parent guide covering token signing, RS256 verification, and the scope-encoding decision matrix
• Encoding Geofence Boundaries in JWT Scope Claims — how the geo_scope regions this dependency parses are built compactly
• Bounding-Box Spatial Index Queries — the ST_Within / ST_Intersects index pattern the scope gate reuses

← Back to JWT Authentication for Spatial Scopes