Securing PostGIS Connections in Workflows

A fan-out changes what a database connection is. One flow becomes four hundred tasks, each of which will happily open its own connection with whatever credential it was handed, and the two failure modes that follow — an exhausted connection pool and a single powerful password copied everywhere — are both structural rather than accidental. The fix is three decisions made once: verify TLS properly, issue short-lived credentials scoped to a role that can do exactly one job, and pool at the worker instead of at the task.

When to Use This Pattern#

• A mapped fan-out writes to PostGIS, which is every ledger, every feature load, every footprint update.
• The database is shared with an application or with analysts whose connections must not be starved.
• Credentials are currently a long-lived password in an environment variable or a secret file.
• The pipeline crosses a network boundary to reach the database, which makes TLS verification a real control rather than a formality.

Complete Working Example#

Three parts: a verified connection string, a credential that expires, and one pool per worker process.

from __future__ import annotations

import functools
import os
from datetime import datetime, timedelta, timezone

import boto3
import psycopg
from psycopg_pool import ConnectionPool
from prefect import task


def _auth_token(host: str, port: int, user: str) -> str:
    """A 15-minute token from cloud identity. No password exists to leak."""
    return boto3.client("rds").generate_db_auth_token(
        DBHostname=host, Port=port, DBUsername=user
    )


def _conninfo(role: str) -> str:
    host, port = os.environ["PGHOST"], int(os.environ.get("PGPORT", 5432))
    return psycopg.conninfo.make_conninfo(
        host=host,
        port=port,
        dbname=os.environ["PGDATABASE"],
        user=role,
        password=_auth_token(host, port, role),
        # verify-full, not require: 'require' encrypts and authenticates nothing.
        sslmode="verify-full",
        sslrootcert=os.environ["PG_CA_BUNDLE"],
        application_name=f"tiles/{os.environ.get('PREFECT_FLOW_RUN_ID', 'local')}"[:63],
        connect_timeout=5,
    )


@functools.lru_cache(maxsize=4)
def pool_for(role: str) -> ConnectionPool:
    """One pool per role per worker PROCESS — never one connection per task."""
    return ConnectionPool(
        conninfo=_conninfo(role),
        min_size=1,
        max_size=4,                          # worker concurrency, not fan-out width
        max_lifetime=13 * 60,                # under the token's 15-minute validity
        reconnect_timeout=30,
        kwargs={"autocommit": False},
    )


@task(retries=2, timeout_seconds=120, tags=["tile"])
def record_tile(tile: Tile, key: str, valid_pct: float) -> None:
    with pool_for("pipeline_writer").connection() as conn:
        conn.execute(LEDGER_UPSERT, {"z": tile.z, "x": tile.x, "y": tile.y,
                                     "key": key, "valid_pct": valid_pct})

The roles themselves are where least privilege actually lives, and they are three statements:

-- Reads sources. Cannot write anything, anywhere.
CREATE ROLE pipeline_reader LOGIN;
GRANT USAGE ON SCHEMA sources TO pipeline_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA sources TO pipeline_reader;

-- Writes the ledger and footprints. Cannot read the sensitive schema at all.
CREATE ROLE pipeline_writer LOGIN;
GRANT USAGE ON SCHEMA products TO pipeline_writer;
GRANT SELECT, INSERT, UPDATE ON products.tile_ledger TO pipeline_writer;
REVOKE ALL ON SCHEMA restricted FROM pipeline_writer;

-- Dashboards. Read-only, and rate-limited by its own connection budget.
CREATE ROLE dashboard_reader LOGIN CONNECTION LIMIT 5;
GRANT USAGE ON SCHEMA products TO dashboard_reader;
GRANT SELECT ON products.tile_ledger TO dashboard_reader;

sslmode=verify-full rather than require is the single most misunderstood setting here. require encrypts the connection and verifies nothing about who is on the other end, so it defends against passive interception and not against an attacker who can answer for the host — which, on a network where that matters at all, is the interesting case. verify-full checks the certificate chain and the hostname, and the only cost is distributing a CA bundle to the workers. Where a managed database publishes a bundle, this is a file in the image and a path in the environment.

Parameter & Option Reference#

Verification & Testing#

def test_connection_verifies_the_host() -> None:
    info = psycopg.conninfo.conninfo_to_dict(_conninfo("pipeline_writer"))
    assert info["sslmode"] == "verify-full"
    assert Path(info["sslrootcert"]).exists(), "CA bundle missing from the worker image"


def test_writer_cannot_read_the_restricted_schema(db) -> None:
    with connect_as("pipeline_writer") as conn:
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("SELECT * FROM restricted.households LIMIT 1")


def test_pool_is_shared_not_per_task(monkeypatch) -> None:
    opened = []
    monkeypatch.setattr(ConnectionPool, "__init__", counting_init(opened))
    with prefect_test_harness():
        tile_build(make_tiles(50))
    assert len(opened) <= 2, f"{len(opened)} pools created — one per task"


def test_pooled_connection_does_not_outlive_its_token() -> None:
    pool = pool_for("pipeline_writer")
    assert pool.max_lifetime < TOKEN_TTL_SECONDS, "connections outlive their credential"


def test_no_password_appears_in_a_log_record(caplog) -> None:
    record_tile.fn(TILE, key="k", valid_pct=0.99)
    assert not any("password" in r.getMessage().lower() for r in caplog.records)

The max_lifetime test prevents a failure that is genuinely hard to diagnose from the outside. A pooled connection authenticated with a fifteen-minute token stays open indefinitely — the token authenticates the handshake, not the session — until something causes a reconnect, at which point the token has long expired and the reconnect fails. The symptom is a pipeline that works for hours and then produces authentication errors on a subset of workers at unpredictable times, which looks like an identity-provider problem and is not.

Two smaller points round out the connection story, both of which come up the first time somebody profiles a slow fan-out. The first is that opening a verified TLS connection is not free — a handshake plus certificate validation is tens of milliseconds, which is negligible once per worker and significant four hundred times. That is another reason the pool is not merely a resource control but a performance one, and it is why the connect-per-task pattern often looks fine in a test with ten tiles and poor in production with four hundred.

The second is application_name. It costs nothing, it is carried into pg_stat_activity, and it turns the question “what are these forty connections doing” from an investigation into a query. Setting it to the flow run identifier means a database administrator can attribute load to a specific run without knowing anything about the pipeline, and can tell a colleague exactly which run to stop. During an incident that is worth more than most of the monitoring built for the purpose.

Common Pitfalls#

• sslmode=require. It encrypts and verifies nothing, which is the security posture people believe verify-full gives them.
• A connection per mapped task. Four hundred tasks exhaust a shared database and take the application with them.
• Pool sized from the fan-out. The queued tasks do not hold connections; only the concurrent ones do.
• Pooled connections outliving short-lived tokens. The failure arrives hours later and looks like an identity problem.
• One role for everything. A writer that can read the restricted schema means a compromised tile task can too.
• No application_name. During an incident pg_stat_activity is the fastest attribution tool available, and only if the connections identify themselves.

Frequently Asked Questions#

Related#

• Security boundaries for spatial data — where the credential boundary sits among the three
• Scoping object storage credentials per tile job — the same discipline for object storage
• Storing flow state in PostGIS versus object storage — what the writer role is writing
• Streaming large GeoPackage loads into PostGIS — the heaviest user of these connections
• Limiting DAG fan-out with concurrency groups — bounding writers at the orchestrator as well

← Security Boundaries for Spatial Data