Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 17 additions & 9 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,21 +29,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
proxy behaves identically to psql direct. Rationale recorded in
`docs/adr/0002-transparent-proxy-ssl-refusal.md`.
- Protocol codec (Phase 3): `internal/protocol` decodes the wire messages
pgpilot routes on (Query, Parse, Bind, Describe, Execute, Sync, Terminate,
ReadyForQuery, CommandComplete, ErrorResponse, RowDescription, DataRow) via
`jackc/pgx/v5/pgproto3`, with round-trip tests and a panic-safe, fuzzed
decoder. The proxy now relays messages frame-by-frame instead of as opaque
bytes and tracks each session's transaction status from ReadyForQuery
(`I`/`T`/`E`). `cmd/pgpilot` gains a `-log-level` flag. Rationale recorded in
pgpilot routes on via `jackc/pgx/v5/pgproto3`, with round-trip tests and a
panic-safe, fuzzed decoder. The proxy now relays messages frame-by-frame and
tracks each session's transaction status from ReadyForQuery. `cmd/pgpilot`
gains a `-log-level` flag. Rationale in
`docs/adr/0003-message-aware-relay.md`.
- Connection pool (Phase 4a): `internal/pool` is a bounded, health-checked pool
of backend connections — configurable max size, acquire timeout, idle timeout
with a background reaper, and per-connection health checks — that applies
backpressure (a bounded wait, or an immediate refusal once a waiter limit is
reached) instead of queueing acquirers without bound. It is the primitive the
session- and transaction-pooling layers will build on.
backpressure instead of queueing acquirers without bound.
- SCRAM authentication and pooling foundation (Phase 4b): `internal/scram`
implements SCRAM-SHA-256 for both the client and server roles;
`internal/backend` opens and authenticates pgpilot's own connections to the
primary with SCRAM, resets them for reuse (`ROLLBACK`, then `DISCARD ALL`),
and manages one connection pool per `(user, database)`; `internal/config`
loads pgpilot's JSON configuration (backend, users, pool sizing). The SCRAM
client is validated against real PostgreSQL. Wiring the client-facing SCRAM
authenticator and session pooling into the proxy is the remaining half of this
phase. Rationale recorded in
`docs/adr/0004-auth-termination-and-pooling.md`.

### Dependencies

- Added `github.com/jackc/pgx/v5` v5.7.1 (pinned to keep the module's `go 1.22`
floor) for its `pgproto3` wire-protocol codec.
- Promoted `golang.org/x/crypto` to a direct dependency for `pbkdf2`, used by
the SCRAM implementation.
67 changes: 67 additions & 0 deletions docs/adr/0004-auth-termination-and-pooling.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
# 4. Terminate authentication to enable connection pooling

Date: 2026-07-21

## Status

Accepted. Supersedes the connection-handling decision in
[ADR 0002](0002-transparent-proxy-ssl-refusal.md), which passed authentication
through untouched.

## Context

Session and transaction pooling reuse a backend connection across *different*
clients. A backend connection can only be reused if it is already authenticated,
which means the pooler — not the client — has to own the backend's
authentication, and therefore has to authenticate the client itself. ADR 0002's
pass-through model, where the client authenticates end-to-end with the backend,
makes reuse impossible: a pooled connection is already past its startup and
cannot replay a new client's handshake. Pooling forces this to change.

## Decision

**pgpilot terminates authentication on both sides, pgbouncer-style.**

- **Backend side (client role):** pgpilot opens its own connections to the
primary and authenticates them with **SCRAM-SHA-256** (`internal/scram`
client, driven by `internal/backend`). These connections are what the pool
holds.
- **Client side (server role):** pgpilot verifies a connecting client with
**SCRAM-SHA-256** (`internal/scram` server) before handing it a pooled
backend. (This half is wired into the proxy in the next step of the phase.)

**Credentials come from a static config file** (`internal/config`, JSON): a list
of users with passwords, used both to verify clients and to open backend
connections. This is pgbouncer's "userlist" model. We rejected two alternatives:

- **`auth_query`** (fetch each user's SCRAM verifier from the backend at
runtime) is more production-faithful but adds a bootstrap connection, verifier
caching, and failure modes; overkill for this project's scope.
- **Cleartext client auth** would avoid implementing the SCRAM *server* but is a
weak posture and would not exercise the interesting half of the protocol.

**Pools are keyed by `(user, database)`.** A connection authenticated as one user
against one database cannot serve another; separate pools keep them isolated,
matching how a client's identity maps to a backend session.

**Connections are reset before reuse** with `ROLLBACK` followed by `DISCARD ALL`,
sent as *separate* simple queries — combined into one multi-statement query they
would run inside an implicit transaction block, which `DISCARD ALL` forbids.
DISCARD ALL clears prepared statements, temp tables, LISTEN registrations, and
session GUCs, so a reused connection looks fresh. A connection that fails to
reset is discarded rather than reused.

## Consequences

- The SCRAM-SHA-256 implementation is validated against real PostgreSQL: the
backend connector authenticates to the primary, and wrong passwords are
rejected by the server.
- pgpilot now needs a config file (`-config`); the `-primary` flag is replaced.
Config hot-reload is a Phase 6 concern.
- Because pgpilot authenticates clients itself, `sslmode=require` still fails
(TLS termination remains future work, per ADR 0002), and cancel-request
handling is deferred — a client's CancelRequest cannot yet be mapped to a
pooled backend.
- Reset-based reuse assumes a well-behaved client that waits for its final
ReadyForQuery before disconnecting; an abrupt mid-response disconnect causes
the connection to be discarded rather than reused.
2 changes: 2 additions & 0 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,5 @@ module github.com/sachhg/pgpilot
go 1.22

require github.com/jackc/pgx/v5 v5.7.1

require golang.org/x/crypto v0.27.0
191 changes: 191 additions & 0 deletions internal/backend/backend.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,191 @@
// Package backend opens and authenticates pgpilot's own connections to a
// PostgreSQL backend using SCRAM-SHA-256, leaving them ready to relay client
// traffic. The connections are what the pool manages.
package backend

import (
"context"
"fmt"
"net"
"slices"
"time"

"github.com/jackc/pgx/v5/pgproto3"

"github.com/sachhg/pgpilot/internal/scram"
)

// protocolVersion3 is the PostgreSQL 3.0 protocol version.
const protocolVersion3 = 196608

const scramMechanism = "SCRAM-SHA-256"

// Conn is an authenticated backend connection, quiescent at ReadyForQuery. It
// implements pool.Conn.
type Conn struct {
raw net.Conn
params map[string]string
pid uint32
key uint32
}

// Close closes the underlying connection.
func (c *Conn) Close() error { return c.raw.Close() }

// NetConn returns the underlying connection for byte-level relaying. The backend
// is quiescent at ReadyForQuery when Dial returns, so no protocol bytes are
// buffered ahead.
func (c *Conn) NetConn() net.Conn { return c.raw }

// Params returns the ParameterStatus values the backend reported during startup.
func (c *Conn) Params() map[string]string { return c.params }

// Dial opens a TCP connection to addr and authenticates as user/password for the
// given database with SCRAM-SHA-256, returning once the backend reaches
// ReadyForQuery.
func Dial(ctx context.Context, addr, user, password, database string) (*Conn, error) {
var d net.Dialer
raw, err := d.DialContext(ctx, "tcp", addr)
if err != nil {
return nil, fmt.Errorf("backend: dial %s: %w", addr, err)
}
if deadline, ok := ctx.Deadline(); ok {
_ = raw.SetDeadline(deadline)
}
c := &Conn{raw: raw, params: make(map[string]string)}
if err := c.authenticate(user, password, database); err != nil {
_ = raw.Close()
return nil, err
}
_ = raw.SetDeadline(time.Time{}) // clear the handshake deadline before relaying
return c, nil
}

func (c *Conn) authenticate(user, password, database string) error {
fe := pgproto3.NewFrontend(c.raw, c.raw)
fe.Send(&pgproto3.StartupMessage{
ProtocolVersion: protocolVersion3,
Parameters: map[string]string{"user": user, "database": database},
})
if err := fe.Flush(); err != nil {
return fmt.Errorf("backend: send startup: %w", err)
}

var sc *scram.Client
for {
msg, err := fe.Receive()
if err != nil {
return fmt.Errorf("backend: receive during startup: %w", err)
}
switch m := msg.(type) {
case *pgproto3.AuthenticationSASL:
if !slices.Contains(m.AuthMechanisms, scramMechanism) {
return fmt.Errorf("backend: server offered unsupported SASL mechanisms %v", m.AuthMechanisms)
}
sc, err = scram.NewClient(password)
if err != nil {
return err
}
fe.Send(&pgproto3.SASLInitialResponse{AuthMechanism: scramMechanism, Data: []byte(sc.FirstMessage())})
if err := fe.Flush(); err != nil {
return err
}
case *pgproto3.AuthenticationSASLContinue:
if sc == nil {
return fmt.Errorf("backend: unexpected SASL continue")
}
final, err := sc.HandleServerFirst(string(m.Data))
if err != nil {
return err
}
fe.Send(&pgproto3.SASLResponse{Data: []byte(final)})
if err := fe.Flush(); err != nil {
return err
}
case *pgproto3.AuthenticationSASLFinal:
if sc == nil {
return fmt.Errorf("backend: unexpected SASL final")
}
if err := sc.HandleServerFinal(string(m.Data)); err != nil {
return err
}
case *pgproto3.AuthenticationOk:
// Authentication done; keep reading to ReadyForQuery.
case *pgproto3.ParameterStatus:
c.params[m.Name] = m.Value
case *pgproto3.BackendKeyData:
c.pid, c.key = m.ProcessID, m.SecretKey
case *pgproto3.ReadyForQuery:
return nil
case *pgproto3.ErrorResponse:
return fmt.Errorf("backend: server rejected connection: %s %s", m.Code, m.Message)
case *pgproto3.NoticeResponse:
// ignore
default:
return fmt.Errorf("backend: unexpected message %T during startup", msg)
}
}
}

// Reset returns the connection to a clean state for reuse by another client: it
// rolls back any open transaction and discards session state (prepared
// statements, temp tables, LISTEN registrations, session GUCs), reading to
// ReadyForQuery. Callers should reuse the connection only when Reset succeeds.
//
// ROLLBACK and DISCARD ALL are sent as separate simple queries: combined into
// one multi-statement query they would run inside an implicit transaction block,
// which DISCARD ALL forbids.
func (c *Conn) Reset(ctx context.Context) error {
fe, restore := c.frontend(ctx)
defer restore()
if err := runSimple(fe, "ROLLBACK"); err != nil {
return fmt.Errorf("backend: reset rollback: %w", err)
}
if err := runSimple(fe, "DISCARD ALL"); err != nil {
return fmt.Errorf("backend: reset discard: %w", err)
}
return nil
}

// Ping checks the connection is alive and at a clean ReadyForQuery state by
// running a no-op query. It is suitable as a pool health check.
func (c *Conn) Ping(ctx context.Context) error {
fe, restore := c.frontend(ctx)
defer restore()
if err := runSimple(fe, ";"); err != nil {
return fmt.Errorf("backend: ping: %w", err)
}
return nil
}

// frontend returns a pgproto3 Frontend over the connection, applying ctx's
// deadline for the duration; restore clears it.
func (c *Conn) frontend(ctx context.Context) (fe *pgproto3.Frontend, restore func()) {
fe = pgproto3.NewFrontend(c.raw, c.raw)
if deadline, ok := ctx.Deadline(); ok {
_ = c.raw.SetDeadline(deadline)
return fe, func() { _ = c.raw.SetDeadline(time.Time{}) }
}
return fe, func() {}
}

// runSimple sends one simple query and reads until ReadyForQuery, returning an
// error only on ErrorResponse (notices and command tags are ignored).
func runSimple(fe *pgproto3.Frontend, sql string) error {
fe.Send(&pgproto3.Query{String: sql})
if err := fe.Flush(); err != nil {
return err
}
for {
msg, err := fe.Receive()
if err != nil {
return err
}
switch m := msg.(type) {
case *pgproto3.ReadyForQuery:
return nil
case *pgproto3.ErrorResponse:
return fmt.Errorf("%s %s", m.Code, m.Message)
}
}
}
41 changes: 41 additions & 0 deletions internal/backend/backend_integration_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
//go:build integration

package backend

import (
"context"
"testing"
"time"
)

const primaryAddr = "127.0.0.1:55432"

// TestDial_AuthenticatesAgainstPrimary proves the SCRAM-SHA-256 client is
// correct against PostgreSQL's reference implementation: a successful Dial means
// the real server accepted our proof. Requires `make up`.
func TestDial_AuthenticatesAgainstPrimary(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

c, err := Dial(ctx, primaryAddr, "pgpilot", "pgpilot", "pgpilot")
if err != nil {
t.Fatalf("Dial: %v (is `make up` running?)", err)
}
defer func() { _ = c.Close() }()

if c.Params()["server_version"] == "" {
t.Error("no server_version ParameterStatus captured during startup")
}
if err := c.Ping(ctx); err != nil {
t.Errorf("Ping: %v", err)
}
}

func TestDial_WrongPasswordRejected(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

if _, err := Dial(ctx, primaryAddr, "pgpilot", "not-the-password", "pgpilot"); err == nil {
t.Fatal("Dial succeeded with a wrong password")
}
}
Loading
Loading