Audit Logging for Location Data Access

← Back to Securing Geospatial APIs

Location data has a property that ordinary business records do not: reading it is itself a sensitive act. Knowing that an account queried a 200-metre box around a particular address at 03:00 tells you something even if no row was returned. That makes the audit trail for a spatial API a first-class security control rather than a compliance checkbox — and it makes the naive implementation, “copy every returned feature into a log table”, actively harmful.

This page covers a trail that answers the questions an incident actually asks — who read what area, when, under which authorisation, and how much came back — while holding less sensitive data than the table it protects. It builds on the identity plumbing from Setting Tenant Context in asyncpg Connections and the scope model in JWT Authentication for Spatial Scopes.

# Prerequisites & Environment

PostgreSQL 14+ with PostGIS 3.3+, FastAPI 0.110+, asyncpg 0.29. The application must already authenticate callers and set a per-session identity; everything below assumes app.subject_id, app.tenant_id and app.request_id are available through current_setting.

-- What the audit layer expects to find on every session
SELECT current_setting('app.subject_id', true)  AS subject,
       current_setting('app.tenant_id',  true)  AS tenant,
       current_setting('app.request_id', true)  AS request;

If any of those return NULL in production traffic, fix that before building the trail — an audit record without an actor is a timestamp.

# What to record, and what not to

The design decision that matters is the granularity of the record. Three levels are common, and the middle one is almost always right.

# Step-by-Step Implementation

# 1. The audit table

Store the envelope of what was read as a geometry, so the trail is itself spatially queryable — “show me everything anyone read within 1 km of this address” becomes an indexed query rather than a log grep.

CREATE TABLE access_audit (
    id           bigserial,
    occurred_at  timestamptz NOT NULL DEFAULT now(),
    subject_id   text        NOT NULL,
    tenant_id    text,
    request_id   uuid,
    action       text        NOT NULL CHECK (action IN ('read','create','update','delete','export')),
    layer        text        NOT NULL,
    -- The AREA that was requested, not the rows that came back
    envelope     geometry(Polygon, 4326),
    row_count    integer,
    filters      jsonb       NOT NULL DEFAULT '{}'::jsonb,
    status_code  smallint,
    PRIMARY KEY (id, occurred_at)
) PARTITION BY RANGE (occurred_at);

CREATE INDEX access_audit_env_gix   ON access_audit USING GIST (envelope);
CREATE INDEX access_audit_subject   ON access_audit (subject_id, occurred_at DESC);
CREATE INDEX access_audit_filters   ON access_audit USING GIN (filters);

Partitioning is not incidental here — the trail outgrows the feature table within months on a busy API, and expiry has to be cheap. The mechanics are the same as in Table Partitioning for Large Spatial Datasets.

Then take away the ability to rewrite it:

GRANT INSERT ON access_audit TO api_rw;
REVOKE UPDATE, DELETE, TRUNCATE ON access_audit FROM api_rw;
GRANT SELECT ON access_audit TO auditor;

# 2. Capture writes with a trigger

A trigger is the only mechanism that also covers migrations, admin scripts and a developer with psql.

CREATE OR REPLACE FUNCTION audit_feature_write()
RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER AS $$
DECLARE
    target geometry;
BEGIN
    target := COALESCE(NEW.geom, OLD.geom);
    INSERT INTO access_audit (
        subject_id, tenant_id, request_id, action, layer,
        envelope, row_count, filters
    ) VALUES (
        COALESCE(current_setting('app.subject_id', true), 'system'),
        current_setting('app.tenant_id', true),
        NULLIF(current_setting('app.request_id', true), '')::uuid,
        lower(TG_OP),
        TG_TABLE_NAME,
        -- Envelope only: the trail never holds the precise geometry
        CASE WHEN target IS NULL THEN NULL
             ELSE ST_Envelope(ST_Buffer(target::geography, 50)::geometry) END,
        1,
        jsonb_build_object('feature_id', COALESCE(NEW.id, OLD.id))
    );
    RETURN COALESCE(NEW, OLD);
END $$;

CREATE TRIGGER features_audit
    AFTER INSERT OR UPDATE OR DELETE ON features
    FOR EACH ROW EXECUTE FUNCTION audit_feature_write();

SECURITY DEFINER lets the trigger insert into a table the calling role cannot otherwise write to directly — which is what stops an application bug from forging records. The 50-metre buffer before taking the envelope means a point feature produces a real polygon rather than a degenerate one, and coarsens the recorded location slightly on purpose.

# 3. Capture reads in middleware

There is no SELECT trigger, and there should not be — reads are far more frequent and the interesting context lives in the API layer.

import json
import time
import uuid
from typing import Any, Callable

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

AUDITED_PREFIXES = ("/v1/features", "/v1/positions", "/v1/exports")
COARSE_DP = 2   # ~1.1 km — enough to investigate, not enough to identify


class SpatialAuditMiddleware(BaseHTTPMiddleware):
    """Record one envelope row per audited read, after the response is known."""

    def __init__(self, app, pool_factory: Callable[[], Any]) -> None:
        super().__init__(app)
        self._pool_factory = pool_factory

    async def dispatch(self, request: Request, call_next):
        if not request.url.path.startswith(AUDITED_PREFIXES):
            return await call_next(request)

        request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
        request.state.request_id = request_id
        started = time.perf_counter()
        response: Response = await call_next(request)
        elapsed_ms = (time.perf_counter() - started) * 1000

        subject = getattr(request.state, "subject_id", None)
        if subject is None:              # unauthenticated: nothing to attribute
            return response

        bbox = _parse_bbox(request.query_params.get("bbox"))
        pool = self._pool_factory()
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO access_audit (
                    subject_id, tenant_id, request_id, action, layer,
                    envelope, row_count, filters, status_code
                ) VALUES ($1, $2, $3, 'read', $4,
                          CASE WHEN $5::float8 IS NULL THEN NULL
                               ELSE ST_MakeEnvelope($5, $6, $7, $8, 4326) END,
                          $9, $10::jsonb, $11)
                """,
                subject,
                getattr(request.state, "tenant_id", None),
                uuid.UUID(request_id),
                request.url.path.strip("/").split("/")[-1],
                *(bbox or (None, None, None, None)),
                int(response.headers.get("x-result-count", 0) or 0),
                json.dumps({
                    k: v for k, v in request.query_params.items()
                    if k not in {"bbox", "token"}
                }),
                response.status_code,
            )

        response.headers["x-request-id"] = request_id
        response.headers["server-timing"] = f"app;dur={elapsed_ms:.1f}"
        return response


def _parse_bbox(raw: str | None) -> tuple[float, float, float, float] | None:
    if not raw:
        return None
    try:
        minx, miny, maxx, maxy = (round(float(v), COARSE_DP) for v in raw.split(","))
    except ValueError:
        return None
    return minx, miny, maxx, maxy

Two deliberate choices. The record is written after the response, so audit latency never appears in the client’s timing, and the row count is taken from a response header the route sets. And the bounding box is rounded before storage — the trail knows the neighbourhood, not the doorstep.

# 4. Keep precise coordinates out of application logs

The audit table is access-controlled. The log pipeline usually is not. Redact at the formatter so no route has to remember.

import logging
import re

COORD_RE = re.compile(r"(-?\d{1,3}\.\d{3})\d+")


class CoarsenCoordinates(logging.Filter):
    """Truncate any decimal degree in a log line to 3 dp (~110 m)."""

    def filter(self, record: logging.LogRecord) -> bool:
        if isinstance(record.msg, str):
            record.msg = COORD_RE.sub(r"\1", record.msg)
        if record.args:
            record.args = tuple(
                COORD_RE.sub(r"\1", a) if isinstance(a, str) else a for a in record.args
            )
        return True

# 5. Know which questions the trail can answer

An audit design is only as good as the questions it can answer under pressure. Before shipping, write down the queries an incident will actually ask and check that each one is answerable from the columns you chose — most trails fail this test on the third or fourth question, and the missing column is usually the row count or the request id that ties an access back to a specific call.

The four questions below are the ones that come up in practice. Two are answered by the subject index, one by the spatial index on the envelope, and one only by having stored the row count. None of them need the returned geometry, which is the whole argument for not storing it.

# Production Code Example

The query an investigation actually runs — everything read near a location in a window, grouped by who read it:

SELECT a.subject_id,
       count(*)                                   AS accesses,
       min(a.occurred_at)                         AS first_seen,
       max(a.occurred_at)                         AS last_seen,
       sum(a.row_count)                           AS rows_returned,
       array_agg(DISTINCT a.layer)                AS layers,
       -- How tightly the accesses cluster on the point of interest
       round(min(ST_Distance(
           ST_Centroid(a.envelope)::geography,
           ST_SetSRID(ST_MakePoint($1, $2), 4326)::geography
       ))::numeric, 0)                            AS closest_m
FROM   access_audit a
WHERE  a.occurred_at >= $3
  AND  a.occurred_at <  $4
  AND  a.envelope && ST_Buffer(
           ST_SetSRID(ST_MakePoint($1, $2), 4326)::geography, $5
       )::geometry
GROUP  BY a.subject_id
HAVING count(*) > 1
ORDER  BY accesses DESC
LIMIT  50;

Because the envelope is a real geometry with a GiST index, this runs in single-digit milliseconds over hundreds of millions of audit rows — the same index behaviour described in Bounding Box & Spatial Index Queries, applied to the trail rather than the data.

A second query worth having ready is the inverse: everything one subject touched, ordered by how unusual it was. Sorting by row_count descending surfaces bulk extractions immediately, and joining on request_id ties each access back to the trace captured by the observability layer, so an investigator can see the endpoint, the latency and the audit record as one story rather than three systems to reconcile.

# Verification & Testing

Test that the trail cannot be bypassed, not merely that it records.

import pytest


@pytest.mark.asyncio
async def test_direct_sql_write_is_still_audited(db_conn):
    """A write that bypasses the API must still leave a record."""
    before = await db_conn.fetchval("SELECT count(*) FROM access_audit")
    await db_conn.execute(
        "INSERT INTO features (layer, geom) "
        "VALUES ('roads', ST_SetSRID(ST_MakePoint(-0.12, 51.50), 4326))"
    )
    after = await db_conn.fetchval("SELECT count(*) FROM access_audit")
    assert after == before + 1


@pytest.mark.asyncio
async def test_application_role_cannot_rewrite_history(api_conn):
    with pytest.raises(Exception) as exc:
        await api_conn.execute("DELETE FROM access_audit WHERE true")
    assert "permission denied" in str(exc.value).lower()


@pytest.mark.asyncio
async def test_audit_envelope_is_coarsened(client, auth_headers):
    await client.get(
        "/v1/features",
        params={"bbox": "-0.127761,51.507351,-0.127700,51.507400"},
        headers=auth_headers,
    )
    row = await _latest_audit_row()
    # 2 dp rounding: the stored envelope must not resolve the original box
    assert abs(row["xmin"] - (-0.13)) < 1e-9

# Failure Modes & Edge Cases

1. Audit writes inside the request transaction. If the middleware’s insert shares the transaction and the request later rolls back, the record disappears with it. Use a separate connection, as above, so an audited failure is still audited.
2. current_setting('app.subject_id') unset. With SET LOCAL on a pooled connection the setting is transaction-scoped; a trigger firing outside that transaction records 'system'. Alert on the rate of 'system' rows rather than ignoring them.
3. The trail becomes the biggest table in the database. At 1 200 reads per second the envelope-level trail grows roughly 30 GB a month. Partition from day one; retrofitting later means a full rewrite.
4. SECURITY DEFINER without a fixed search_path. A definer function is a privilege escalation path if a caller can shadow a table name. Always add SET search_path = pg_catalog, public to the function definition.
5. Audit inserts contending on one index. The (subject_id, occurred_at DESC) index becomes a hot spot when a handful of service accounts generate most traffic. Watch for index page contention and consider dropping to a BRIN index on occurred_at for the append-heavy partitions.
6. Exports treated as ordinary reads. A bulk export is qualitatively different from a map pan — record it with action = 'export' and a row count, and alert above a threshold. Bulk paths are described in Async Bulk Uploads with Celery and deserve their own retention rules.
7. Retention deleted by row. A DELETE FROM access_audit WHERE occurred_at < … on an append-only table produces an enormous vacuum backlog. Detach and drop the partition instead.

# Performance Notes

An envelope-level audit insert costs 0.4–0.9 ms on a warm connection and runs off the response path, so client-visible latency is unchanged. The write trigger adds roughly 0.2 ms per mutated row — negligible for interactive writes, but material for bulk loads: a 500 000 row import produces 500 000 audit rows and doubles the load time. For bulk paths, disable the row trigger inside the load transaction and write one summary record instead.

Storage is the real cost. Measure it early with a projection rather than discovering it:

SELECT pg_size_pretty(pg_total_relation_size('access_audit')) AS total,
       count(*)                                              AS rows,
       pg_size_pretty(
         (pg_total_relation_size('access_audit') / GREATEST(count(*), 1))::bigint
       )                                                     AS per_row
FROM   access_audit;

Roughly 340 bytes per row including the three indexes is typical; the GIN index on filters is the largest single contributor, so drop it if nobody queries by filter.

# Related

• Row-Level Security for Multi-Tenant PostGIS — the isolation the audit trail is evidence for
• Setting Tenant Context in asyncpg Connections — how identity reaches the session the trigger reads
• JWT Authentication for Spatial Scopes — where the subject and scope come from
• Table Partitioning for Large Spatial Datasets — making retention a detach rather than a delete
• Rate Limiting Geofence & Tile Endpoints — throttling the enumeration patterns the trail exposes

← Back to Securing Geospatial APIs