From 1bfb048c4bb85ae06d051fd9945f757ca1f56523 Mon Sep 17 00:00:00 2001 From: Andy Bonventre Date: Wed, 15 Jul 2026 18:21:57 -0400 Subject: [PATCH 1/2] feat(proxy): multiplex HTTP and Postgres on one port when their ports coincide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ships as v0.19.0. When postgres.port equals proxy.port (and their hosts resolve to the same address), gatekeeper now multiplexes the HTTP/CONNECT proxy and the Postgres data-plane listener onto one real listener instead of binding two — no new config flag; the port equality itself is the declaration. Distinct ports (today's default, and every existing config) keep the two-listener path byte-for-byte unchanged. Mechanism: proxy.Demux (proxy/demux.go) owns the one real listener and classifies each accepted connection by its first 8 bytes in its own goroutine — never in the shared accept loop, so a silent or slow client can't stall Accept for other connections. Classification positively matches Postgres startup signatures (SSLRequest, GSSENCRequest, v3 StartupMessage) and defaults everything else to HTTP, mirroring Caddy-L4's postgres matcher. The sniffed bytes are replayed on the classified connection (mirroring proxyProtoLogConn's hold-then-replay pattern) before it's pushed onto one of two in-memory virtual listeners; http.Server.Serve and PostgresServer.StartListener then run completely unmodified against those. Config validation (resolveListenTopology, config.go) runs at New() and Start() so the two can never disagree: equal ports with different hosts, or with proxy.proxy_protocol != postgres.proxy_protocol, are fatal startup errors — a shared listener has exactly one PROXY protocol setting, owned by proxy.proxy_protocol. Port 0 on both sides never triggers multiplexing, since two independent ephemeral binds aren't "the same port" — this preserves every existing test and config that leaves the port unset on both listeners. Docs: config reference ("Sharing one listener with the HTTP proxy"), load-balancer guide ("Single load balancer, one shared port", including health-check behavior on the shared listener), a concepts note, and a commented example in examples/gatekeeper-postgres.yaml. --- CHANGELOG.md | 9 + config.go | 69 ++ config_topology_test.go | 163 ++++ .../concepts/08-postgres-data-plane.md | 2 + .../guides/11-load-balancer-proxy-protocol.md | 36 +- docs/content/reference/02-config-file.md | 26 +- examples/gatekeeper-postgres.yaml | 12 + gatekeeper.go | 130 ++- gatekeeper_multiplex_test.go | 401 ++++++++ proxy/demux.go | 336 +++++++ proxy/demux_test.go | 864 ++++++++++++++++++ 11 files changed, 2009 insertions(+), 39 deletions(-) create mode 100644 config_topology_test.go create mode 100644 gatekeeper_multiplex_test.go create mode 100644 proxy/demux.go create mode 100644 proxy/demux_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index d1795ba..ddef53d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ Gatekeeper is a standalone credential-injecting TLS-intercepting proxy. It trans Gatekeeper is pre-1.0. The configuration schema and credential source interface may change between minor versions. +## v0.19.0 — 2026-07-15 + +### Added + +- **The HTTP/CONNECT proxy and the Postgres data-plane listener can now share a single port** — an operator putting gatekeeper behind a load balancer previously needed one backend service, one health check, and one firewall rule per listener, even when the two planes were fronted by the same LB. Standing up two of everything for what is conceptually one gatekeeper endpoint was pure ceremony; this removes it. The trigger is topology, not a new flag: when `postgres.port` equals `proxy.port` (and their hosts, after defaults, also match), gatekeeper multiplexes both planes onto one real listener instead of binding two — the port equality itself *is* the declaration, there is no `multiplex: true` to set. Distinct ports (the default, and every existing config) leave the two-listener path byte-for-byte unchanged; `postgres` being absent never engages the shared-listener path either. + Mechanism: a new `proxy.Demux` (`proxy/demux.go`) owns the one real listener — wrapped with `WrapProxyProtocolListener` first when `proxy.proxy_protocol` is set, exactly as today — and reads the first 8 bytes of each accepted connection *in a per-connection goroutine, never in the shared accept loop*, so a silent or slow client (an LB health check that opens a socket and waits, or a slow-loris) can never stall `Accept` for every other pending connection — the same discipline `WrapProxyProtocolListener`'s own `Accept` already follows, and a prior review finding this change deliberately doesn't regress. Those 8 bytes positively identify a Postgres startup signature — `SSLRequest` (length `00 00 00 08`, code `04 d2 16 2f`), `GSSENCRequest` (same length, code `04 d2 16 30`), or a v3 `StartupMessage` (any 4-byte length, then protocol `00 03 00 00`) — and default everything else (HTTP methods, the h2 client preface, or any other prefix) to the HTTP plane, the same positive-match-Postgres approach as Caddy-L4's postgres matcher. The classified connection is wrapped so those 8 bytes are replayed on its first `Read` calls before any further bytes from the underlying socket — the same hold-then-replay pattern `proxyProtoLogConn` already uses for the PROXY header — then pushed onto one of two in-memory virtual listeners (a bounded, non-blocking channel of conns implementing `net.Listener`). `http.Server.Serve` and `PostgresServer.StartListener` run completely unmodified against those virtual listeners, unaware they aren't backed by a real socket. Once classification succeeds, the sniff deadline (10s, matching `WrapProxyProtocolListener`'s own PROXY-header-read timeout) is cleared before handoff, so only the downstream server's own timeout — `http.Server`'s `ReadHeaderTimeout`, or `PostgresServer`'s handshake timeout — ever applies from that point on. A connection that times out, closes early, or otherwise sends fewer than 8 bytes is dropped with a DEBUG log line naming the failure, never the bytes themselves; a successfully classified connection logs a second DEBUG line naming only `http` or `postgres`. + Because one physical listener means one PROXY protocol setting, gatekeeper now validates listener topology at startup (`resolveListenTopology`, `config.go`) before either listener binds: equal ports with `proxy.proxy_protocol != postgres.proxy_protocol` is a fatal, descriptive config error (`proxy.proxy_protocol` is the setting actually applied to the shared listener), as is equal ports with different hosts (there's only one socket to bind, and gatekeeper won't guess which address you meant). Port `0` on both sides — "give me any available port," gatekeeper's own default — deliberately never triggers multiplexing on its own, even though the values compare equal; two independent ephemeral-port binds aren't "the same port," and treating them as such would have silently changed the two-listener behavior of every config (and most of this repo's own test suite) that leaves the port unset. `gatekeeper.Server` gains a `demux *proxy.Demux` field; `Stop` calls `Demux.StopAccepting` (not `Close`) to close only the real listener immediately, deliberately leaving both virtual listeners open for `PostgresServer.Shutdown`/`http.Server.Shutdown` to close as part of their own graceful drain — closing them out from under those servers first would race their own closed-flag bookkeeping and, for `PostgresServer`, surface as a spurious error log on an entirely ordinary shutdown. + Documented in the config reference (["Sharing one listener with the HTTP proxy"](docs/content/reference/02-config-file.md#sharing-one-listener-with-the-http-proxy)) and the load-balancer guide (["Single load balancer, one shared port"](docs/content/guides/11-load-balancer-proxy-protocol.md#single-load-balancer-one-shared-port), including health-check behavior on the shared listener: an HTTP health check classifies as HTTP traffic and reaches `/healthz` normally, and a bare TCP connect-then-close probe is simply dropped, cleanly and silently, once the classification read times out) + ## v0.18.0 — 2026-07-15 ### Added diff --git a/config.go b/config.go index 8b40b47..191c638 100644 --- a/config.go +++ b/config.go @@ -1,6 +1,7 @@ package gatekeeper import ( + "fmt" "os" "gopkg.in/yaml.v3" @@ -171,3 +172,71 @@ func LoadConfig(path string) (*Config, error) { } return ParseConfig(data) } + +// listenTopology is the resolved result of applying gatekeeper's host +// default rules to Config.Proxy and Config.Postgres, and deciding whether +// they collapse onto one shared listener. See resolveListenTopology. +type listenTopology struct { + proxyHost string + proxyPort int + pgHost string // "" when no Postgres listener is configured + pgPort int + multiplex bool +} + +// resolveListenTopology applies gatekeeper's host-default rules +// (proxy.host defaults to defaultProxyHost; postgres.host defaults to the +// resolved proxy host) and decides whether the proxy and Postgres listeners +// collapse onto one shared port. There is no separate flag for this — an +// operator declares "one listener" purely by pointing both configs at the +// same port and host; resolveListenTopology and Server.Start both call this +// function so the declaration and the actual wiring can never disagree. +// +// Multiplexing triggers only when postgres.port is a real, non-zero port +// and it equals proxy.port, and the two configs' hosts resolve to the same +// address. Port 0 (ask the OS for an ephemeral port) never triggers +// multiplexing, even when both sides leave it unset: two independent +// net.Listen(...:0) calls are not "the same port," and treating them as +// such would silently change today's two-listener behavior for every +// config — and every test in this suite — that leaves the port unset on +// both listeners. +// +// It returns a fatal, user-facing error for the two configurations +// gatekeeper refuses to start with once ports are declared equal: different +// hosts (ambiguous — which address should the one shared listener bind?), +// and proxy.proxy_protocol != postgres.proxy_protocol (a single shared +// listener can only have one PROXY protocol setting, owned by +// proxy.proxy_protocol since proxy.proxy_protocol is the listener owner). +func resolveListenTopology(cfg *Config) (listenTopology, error) { + proxyHost := cfg.Proxy.Host + if proxyHost == "" { + proxyHost = defaultProxyHost + } + topo := listenTopology{proxyHost: proxyHost, proxyPort: cfg.Proxy.Port} + if cfg.Postgres == nil { + return topo, nil + } + + pgHost := cfg.Postgres.Host + if pgHost == "" { + pgHost = proxyHost + } + topo.pgHost = pgHost + topo.pgPort = cfg.Postgres.Port + + if cfg.Postgres.Port == 0 || cfg.Postgres.Port != cfg.Proxy.Port { + return topo, nil + } + + // Equal, non-zero ports: this is the multiplex declaration. + if pgHost != proxyHost { + return topo, fmt.Errorf("postgres.port (%d) equals proxy.port but postgres.host (%q) differs from proxy.host (%q); a shared listener can only bind one address — use the same host on both, or different ports", + cfg.Postgres.Port, pgHost, proxyHost) + } + if cfg.Postgres.ProxyProtocol != cfg.Proxy.ProxyProtocol { + return topo, fmt.Errorf("postgres.port (%d) equals proxy.port but postgres.proxy_protocol (%v) differs from proxy.proxy_protocol (%v); a shared listener has one PROXY protocol setting, owned by proxy.proxy_protocol", + cfg.Postgres.Port, cfg.Postgres.ProxyProtocol, cfg.Proxy.ProxyProtocol) + } + topo.multiplex = true + return topo, nil +} diff --git a/config_topology_test.go b/config_topology_test.go new file mode 100644 index 0000000..5c5facb --- /dev/null +++ b/config_topology_test.go @@ -0,0 +1,163 @@ +package gatekeeper + +// config_topology_test.go tests resolveListenTopology, the pure function +// that decides whether the proxy and Postgres listeners collapse onto one +// shared, single-port listener (see demux.go in the proxy package) purely +// from port/host equality — there is no separate config flag. + +import ( + "strings" + "testing" +) + +func TestResolveListenTopology_NoPostgresNeverMultiplexes(t *testing.T) { + cfg := &Config{Proxy: ProxyConfig{Port: 8080, Host: "127.0.0.1"}} + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if topo.multiplex { + t.Error("multiplex = true with no postgres configured, want false") + } +} + +func TestResolveListenTopology_EqualNonZeroPortsAndHostsMultiplex(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1"}, + Postgres: &PostgresConfig{Port: 5432}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if !topo.multiplex { + t.Error("multiplex = false with equal proxy.port/postgres.port and matching hosts, want true") + } +} + +func TestResolveListenTopology_EqualPortsExplicitSameHostMultiplexes(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "0.0.0.0"}, + Postgres: &PostgresConfig{Port: 5432, Host: "0.0.0.0"}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if !topo.multiplex { + t.Error("multiplex = false with explicit matching hosts, want true") + } +} + +// TestResolveListenTopology_EqualPortsDifferentHostsErrors pins the fatal, +// ambiguous-bind error: two listeners can't share one port but bind +// different addresses. +func TestResolveListenTopology_EqualPortsDifferentHostsErrors(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1"}, + Postgres: &PostgresConfig{Port: 5432, Host: "10.0.0.5"}, + } + _, err := resolveListenTopology(cfg) + if err == nil { + t.Fatal("resolveListenTopology succeeded with equal ports but different hosts, want an error") + } + if !strings.Contains(err.Error(), "127.0.0.1") || !strings.Contains(err.Error(), "10.0.0.5") { + t.Errorf("error = %q, want it to name both conflicting hosts", err) + } +} + +// TestResolveListenTopology_MismatchedProxyProtocolErrors pins the fatal +// error for a shared listener with disagreeing PROXY protocol settings: one +// physical listener can only have one PROXY protocol setting. +func TestResolveListenTopology_MismatchedProxyProtocolErrors(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1", ProxyProtocol: true}, + Postgres: &PostgresConfig{Port: 5432, ProxyProtocol: false}, + } + _, err := resolveListenTopology(cfg) + if err == nil { + t.Fatal("resolveListenTopology succeeded with mismatched proxy_protocol settings, want an error") + } + if !strings.Contains(err.Error(), "proxy_protocol") { + t.Errorf("error = %q, want it to mention proxy_protocol", err) + } +} + +// TestResolveListenTopology_MatchingProxyProtocolMultiplexes is the +// converse of the mismatch case: when both sides explicitly agree, the +// shared listener is allowed and inherits that setting. +func TestResolveListenTopology_MatchingProxyProtocolMultiplexes(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1", ProxyProtocol: true}, + Postgres: &PostgresConfig{Port: 5432, ProxyProtocol: true}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if !topo.multiplex { + t.Error("multiplex = false with matching proxy_protocol settings, want true") + } +} + +// TestResolveListenTopology_DistinctPortsNeverMultiplex pins today's default +// two-listener path, unchanged. +func TestResolveListenTopology_DistinctPortsNeverMultiplex(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 8080, Host: "127.0.0.1"}, + Postgres: &PostgresConfig{Port: 5432}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if topo.multiplex { + t.Error("multiplex = true with distinct ports, want false") + } +} + +// TestResolveListenTopology_BothPortsZeroNeverMultiplex is a regression pin +// for a subtle trap: Port: 0 on both sides means "ask the OS for an +// ephemeral port" on each listener independently, not "these are the same +// port." Treating 0 == 0 as a multiplex trigger would silently change +// behavior for every config (and every test in this suite) that leaves the +// port unset on both listeners, in violation of "distinct ports leave the +// two-listener path unchanged" — port 0 on both sides isn't a declared +// shared port at all, so it must never trigger multiplexing. +func TestResolveListenTopology_BothPortsZeroNeverMultiplex(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}, + Postgres: &PostgresConfig{Port: 0}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if topo.multiplex { + t.Error("multiplex = true with both ports left at 0, want false (0 means \"OS-assigned\", not \"shared\")") + } +} + +// TestResolveListenTopology_PostgresHostDefaultsToResolvedProxyHost mirrors +// the existing gatekeeper.go rule (postgres.host defaults to the same host +// the proxy listener resolved to, including the 127.0.0.1 default when +// proxy.host is empty) so multiplex detection agrees with it exactly. +func TestResolveListenTopology_PostgresHostDefaultsToResolvedProxyHost(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432}, // Host left empty -> defaultProxyHost + Postgres: &PostgresConfig{Port: 5432}, + } + topo, err := resolveListenTopology(cfg) + if err != nil { + t.Fatalf("resolveListenTopology: %v", err) + } + if !topo.multiplex { + t.Error("multiplex = false when both hosts default to the same resolved address, want true") + } + if topo.proxyHost != defaultProxyHost { + t.Errorf("proxyHost = %q, want %q", topo.proxyHost, defaultProxyHost) + } + if topo.pgHost != defaultProxyHost { + t.Errorf("pgHost = %q, want %q", topo.pgHost, defaultProxyHost) + } +} diff --git a/docs/content/concepts/08-postgres-data-plane.md b/docs/content/concepts/08-postgres-data-plane.md index 4414666..f3d55a7 100644 --- a/docs/content/concepts/08-postgres-data-plane.md +++ b/docs/content/concepts/08-postgres-data-plane.md @@ -8,6 +8,8 @@ keywords: ["gatekeeper", "postgres", "neon", "SCRAM", "credential injection", "d Gatekeeper runs a second listener that speaks the Postgres wire protocol. A client connects to it with the real database hostname and an authentication token in place of the database password. Gatekeeper resolves the real password, authenticates upstream, and relays the connection. The database password never reaches the client. +This is a genuinely separate listener from the HTTP/CONNECT proxy — except when `postgres.port` is configured to the same value as `proxy.port`, in which case gatekeeper multiplexes both planes onto one shared listener, classifying each connection by its first bytes. See [Sharing one listener with the HTTP proxy](../reference/02-config-file.md#sharing-one-listener-with-the-http-proxy) in the config reference. + This mirrors the HTTP credential-injection plane: the client presents a weak identity (its run token), and Gatekeeper substitutes the real credential before talking to the upstream. The two planes share configuration, network policy, per-run context scoping, and audit logging. ## What it solves diff --git a/docs/content/guides/11-load-balancer-proxy-protocol.md b/docs/content/guides/11-load-balancer-proxy-protocol.md index 9eea7fa..f369aa6 100644 --- a/docs/content/guides/11-load-balancer-proxy-protocol.md +++ b/docs/content/guides/11-load-balancer-proxy-protocol.md @@ -8,7 +8,7 @@ keywords: ["gatekeeper", "load balancer", "PROXY protocol", "GCP", "client IP", Recover the real client IP when gatekeeper runs behind a TCP-terminating load balancer, such as GCP's global TCP Proxy load balancer. Without this, every request's `client_ip` log attribute shows the load balancer's hop instead of the actual client. -PROXY protocol support is configured per listener: `proxy.proxy_protocol` for the HTTP/CONNECT listener, and `postgres.proxy_protocol` for the [Postgres data-plane listener](../concepts/08-postgres-data-plane.md) (if configured). They are independent toggles — enabling one does not enable the other — because the two listeners are commonly fronted by different load balancers, or only one of them is exposed publicly at all. Everything below applies identically to both; the config examples call out where the field name differs. +PROXY protocol support is configured per listener: `proxy.proxy_protocol` for the HTTP/CONNECT listener, and `postgres.proxy_protocol` for the [Postgres data-plane listener](../concepts/08-postgres-data-plane.md) (if configured). They are independent toggles — enabling one does not enable the other — because the two listeners are commonly fronted by different load balancers, or only one of them is exposed publicly at all. Everything below applies identically to both; the config examples call out where the field name differs. (The one exception: when `proxy.port` and `postgres.port` are set to the same value, the two listeners collapse into one shared listener behind a single load balancer, and the two `proxy_protocol` settings must then agree — see [Single load balancer, one shared port](#single-load-balancer-one-shared-port) below.) ## Prerequisites @@ -69,9 +69,39 @@ postgres: proxy_protocol: true ``` +## Single load balancer, one shared port + +The two listeners above don't have to be separate ports. If `proxy.port` and `postgres.port` are set to the *same* value (and the same host), gatekeeper multiplexes both planes onto one real listener instead of binding two — useful when you only want to stand up one backend service, one health check, and one firewall rule in front of gatekeeper, rather than one per listener. + +There's no separate flag for this: the port equality itself is the declaration. Each accepted connection is classified by its first bytes — a Postgres startup signature (`SSLRequest`, `GSSENCRequest`, or the v3 `StartupMessage`) routes to the Postgres data plane, and everything else (HTTP methods, the h2 preface) routes to the HTTP/CONNECT plane — before being handed to the same HTTP server or Postgres server code that runs the ordinary, separate-port configuration. Classification happens per-connection, not in the shared accept loop, so a silent or slow-to-classify client (an LB health check that opens a socket and never sends anything, for example) can't stall every other pending connection. + +```yaml +proxy: + host: 0.0.0.0 + port: 5432 + proxy_protocol: true + +tls: + ca_cert: ca.crt + ca_key: ca.key + +postgres: + port: 5432 + proxy_protocol: true +``` + +A shared listener has exactly one PROXY protocol setting, since there's only one physical socket to wrap: `proxy.proxy_protocol` and `postgres.proxy_protocol` must agree (both `true` or both `false`, or simply leave `postgres.proxy_protocol` unset — see the [config reference](../reference/02-config-file.md#postgres) for the exact default). Gatekeeper refuses to start otherwise, with an error naming the mismatch — this is checked at startup, before either listener binds, so a typo doesn't surface as a mysterious connection-classification bug later. `proxy.proxy_protocol` is the setting that's actually applied to the shared listener; `postgres.proxy_protocol` exists in this mode purely so the two fields can't silently drift apart in a config that's meant to describe one listener. + +Two configuration mistakes are also rejected at startup for the same reason: + +- **Equal ports, different hosts** — e.g. `proxy.host: 127.0.0.1` with `postgres.host: 0.0.0.0` and the same port. There's only one socket to bind, and gatekeeper can't guess which address you meant. +- **Equal ports, mismatched `proxy_protocol`** — described above. + +If `postgres` isn't configured at all, or the two ports differ, gatekeeper binds two listeners exactly as it always has — the shared-port path only ever engages when you've explicitly pointed both configs at the same port and host. + ## Configuring the GCP backend service -Enabling `proxy_protocol` on gatekeeper is only half the change — the load balancer must also be told to send the header. For a global TCP Proxy load balancer, this is the backend service's `proxyHeader` field. Configure this on whichever backend service fronts the listener you enabled — the HTTP/CONNECT port, the Postgres port, or both, if both are load-balanced. +Enabling `proxy_protocol` on gatekeeper is only half the change — the load balancer must also be told to send the header. For a global TCP Proxy load balancer, this is the backend service's `proxyHeader` field. Configure this on whichever backend service fronts the listener you enabled — the HTTP/CONNECT port, the Postgres port, or both, if both are load-balanced. On the shared-port setup above, there's only one backend service and one health check to configure, since there's only one listener. Update an existing backend service: @@ -164,6 +194,8 @@ level=DEBUG msg="dropping connection: malformed PROXY protocol header" peer=35.1 Gatekeeper's fail-open policy means a load balancer health check keeps working whether or not it carries a PROXY header: a probe that includes one is parsed normally, and a probe that doesn't falls back to the raw TCP peer address. Neither case is rejected, so flipping `proxy_protocol` on and off does not require a matching change to the health check configuration. This holds for both the HTTP listener's `/healthz` endpoint and a bare TCP connect check against the Postgres listener. +On the shared-port setup, `/healthz` still works exactly as it does on a dedicated HTTP listener: an HTTP health check's request line is classified as HTTP traffic (it isn't a Postgres startup signature) and routed to the same handler that serves `/healthz` on a distinct port. A bare TCP connect-then-close probe — the kind a plain TCP health check performs — never sends enough bytes to be classified either way, so it's simply dropped once the classification read times out, the same clean, silent outcome as connecting and disconnecting from any other gatekeeper listener. + ## Next steps - [Postgres data plane](../concepts/08-postgres-data-plane.md) — how the Postgres listener authenticates clients and resolves upstream credentials diff --git a/docs/content/reference/02-config-file.md b/docs/content/reference/02-config-file.md index a8f34fe..d9fd9c0 100644 --- a/docs/content/reference/02-config-file.md +++ b/docs/content/reference/02-config-file.md @@ -71,6 +71,8 @@ proxy: - **Required:** No - **Default:** `0` (random available port) +If this equals `postgres.port` (and `proxy.host` equals `postgres.host`, after defaults), the two listeners collapse onto one shared physical listener instead of binding separately — see [Sharing one listener with the HTTP proxy](#sharing-one-listener-with-the-http-proxy) under `postgres` below. There is no separate flag for this: setting both ports the same **is** the declaration. + ### proxy.host Bind address for the proxy listener. @@ -217,7 +219,29 @@ Semantics are identical to `proxy.proxy_protocol`, applied to this listener inst Because the header is honored from any peer, only enable this when the Postgres port is reachable solely through the load balancer, and never use `client_ip` for security decisions. -This is a separate toggle from `proxy.proxy_protocol` — enabling one does not enable the other, since the two listeners are typically fronted by different load balancers (or only one of them is exposed at all). +This is a separate toggle from `proxy.proxy_protocol` — enabling one does not enable the other, since the two listeners are typically fronted by different load balancers (or only one of them is exposed at all). **Exception:** when `postgres.port` equals `proxy.port` (see below), the two settings must agree. + +### Sharing one listener with the HTTP proxy + +When `postgres.port` equals `proxy.port` — and `proxy.host` and `postgres.host` are the same string, after their defaults are applied — gatekeeper multiplexes both planes onto **one** real listener instead of binding two. (The host comparison is a literal string match, not address resolution: `proxy.host: localhost` and `postgres.host: 127.0.0.1` are treated as *different* and rejected as an ambiguous bind rather than guessed to be equivalent.) Each accepted connection is classified by its first bytes: a recognized Postgres startup signature (`SSLRequest`, `GSSENCRequest`, or the v3 `StartupMessage`) routes to the Postgres data plane, and everything else — HTTP methods, the h2 preface — routes to the HTTP/CONNECT plane, unchanged. There is no dedicated config flag for this; the port equality itself is the trigger. + +```yaml +proxy: + host: 0.0.0.0 + port: 5432 + +postgres: + port: 5432 # same host (defaulted from proxy.host) and same port as above -> shared listener +``` + +Gatekeeper refuses to start, with a fatal, descriptive error, in two situations once the ports are equal: + +- **Different hosts.** `postgres.port == proxy.port` but the two configs' hosts (after defaults) don't match — there's only one socket to bind, and which address to use is ambiguous. +- **Mismatched `proxy_protocol`.** `postgres.proxy_protocol != proxy.proxy_protocol` — a single shared listener can only have one PROXY protocol setting. `proxy.proxy_protocol` is the setting actually applied; set both fields (or leave `postgres.proxy_protocol` unset, which mirrors `proxy.proxy_protocol`'s own default of `false`) so they agree. + +Port `0` (letting the OS assign an ephemeral port) never triggers multiplexing on its own, even if both `proxy.port` and `postgres.port` are left unset — two independent ephemeral-port binds are not "the same port." Multiplexing requires an explicit, matching, non-zero port on both listeners. + +If `postgres` isn't configured at all, or the two ports differ, gatekeeper binds two listeners exactly as it always has. See [Deploying behind a TCP load balancer](../guides/11-load-balancer-proxy-protocol.md#single-load-balancer-one-shared-port) for the full walkthrough, including PROXY protocol and health checks on the shared listener. --- diff --git a/examples/gatekeeper-postgres.yaml b/examples/gatekeeper-postgres.yaml index 7180eef..c5ed48c 100644 --- a/examples/gatekeeper-postgres.yaml +++ b/examples/gatekeeper-postgres.yaml @@ -40,6 +40,18 @@ proxy: auth_token: local-test-token # Postgres data-plane listener. Omit this block to disable the data plane. +# +# Setting postgres.port to the SAME value as proxy.port above (with a +# matching host, and a matching proxy_protocol setting) collapses both +# listeners onto one shared port instead of binding two — no separate flag, +# the port equality alone is the declaration. Useful behind a single load +# balancer / firewall rule. See the "Sharing one listener with the HTTP +# proxy" section of the config reference and the load-balancer guide's +# "Single load balancer, one shared port" section for the full rules +# (equal ports + different hosts, or mismatched proxy_protocol, both fail +# fast at startup instead of binding). +# postgres: +# port: 9080 # == proxy.port -> one shared listener, not two postgres: host: 127.0.0.1 # optional; defaults to the proxy host port: 5432 diff --git a/gatekeeper.go b/gatekeeper.go index e3295ec..415ccb2 100644 --- a/gatekeeper.go +++ b/gatekeeper.go @@ -211,6 +211,7 @@ type Server struct { proxyServer *http.Server pgServer *proxy.PostgresServer // postgres data-plane listener, if configured pgAddr string // actual postgres listener address after Start + demux *proxy.Demux // non-nil when proxy and postgres share one listener (see resolveListenTopology) logCleanup func() // closes log file if output is a file path pendingRefreshes []pendingRefresh @@ -269,6 +270,14 @@ func New(ctx context.Context, cfg *Config, version string) (*Server, error) { return nil, fmt.Errorf("postgres listener requires tls.ca_cert and tls.ca_key to be configured") } + // Fail fast on an invalid or ambiguous proxy/postgres listener topology + // (e.g. equal ports but different hosts) rather than discovering it only + // once Start tries to bind. Start recomputes the same topology right + // before wiring listeners, so the two can never disagree. + if _, err := resolveListenTopology(cfg); err != nil { + return nil, err + } + // Load credentials from config and set directly on the proxy. // Credentials are fetched once at startup. For sources like // aws-secretsmanager, restart the process to pick up rotated values. @@ -793,11 +802,17 @@ func (s *Server) Start(ctx context.Context) error { s.started = true s.mu.Unlock() - // Default to localhost if no host is configured. - host := s.cfg.Proxy.Host - if host == "" { - host = defaultProxyHost + // Resolve host defaults and decide whether proxy and postgres share one + // listener (see resolveListenTopology's doc comment): equal, non-zero + // ports on matching hosts is the entire declaration — there's no + // separate flag. New already validated this at construction time from + // the same function, so an error here would only surface if cfg were + // mutated after New returned. + topology, err := resolveListenTopology(s.cfg) + if err != nil { + return err } + host := topology.proxyHost // Start proxy listener. addr := fmt.Sprintf("%s:%d", host, s.cfg.Proxy.Port) @@ -817,12 +832,17 @@ func (s *Server) Start(ctx context.Context) error { // which all log from the outer request's RemoteAddr) ever sees it. See // proxy.WrapProxyProtocolListener for the fail-open USE policy, the 10s // header-read timeout, and the malformed-header debug log — the Postgres - // data-plane listener below shares this exact same helper. + // data-plane listener (or, when topology.multiplex is true, this same + // shared listener) uses this exact same helper. if s.cfg.Proxy.ProxyProtocol { ln = proxy.WrapProxyProtocolListener(ln) } - slog.Info("gatekeeper listening", "addr", ln.Addr().String(), "version", s.version) + if topology.multiplex { + slog.Info("proxy and postgres multiplexed on one listener", "addr", ln.Addr().String(), "version", s.version) + } else { + slog.Info("gatekeeper listening", "addr", ln.Addr().String(), "version", s.version) + } s.mu.Lock() s.proxyLn = ln @@ -838,38 +858,26 @@ func (s *Server) Start(ctx context.Context) error { // CONNECT tunnels are long-lived, and a write timeout would kill // idle but valid connections. } - go func() { _ = s.proxyServer.Serve(ln) }() - // Start the Postgres data-plane listener if configured. - if s.cfg.Postgres != nil { - pgHost := s.cfg.Postgres.Host - if pgHost == "" { - pgHost = host // same default the HTTP listener resolved - } - pgAddr := fmt.Sprintf("%s:%d", pgHost, s.cfg.Postgres.Port) - pgLn, err := net.Listen("tcp", pgAddr) - if err != nil { - // Tear down the already-running HTTP server so a postgres bind - // failure doesn't leak the HTTP listener. http.Server.Shutdown - // closes the listener it was Serve-ing, so closing the server is - // sufficient — closing ln again here would race the Serve goroutine. - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - _ = s.proxyServer.Shutdown(shutdownCtx) - cancel() - return fmt.Errorf("starting postgres listener: %w", err) - } - // Wrap before StartListener: the PROXY header arrives as the very - // first bytes on the wire, ahead of the client's SSLRequest, so the - // listener must be PROXY-protocol-aware from the first Accept. See - // proxy.WrapProxyProtocolListener — same fail-open USE policy, 10s - // header-read timeout, and malformed-header debug log as the HTTP - // listener above. - if s.cfg.Postgres.ProxyProtocol { - pgLn = proxy.WrapProxyProtocolListener(pgLn) - } + if topology.multiplex { + // resolveListenTopology already confirmed proxy.port == postgres.port, + // matching hosts, and matching proxy_protocol settings, so ln alone + // (already PROXY-protocol-wrapped above if configured) carries both + // planes. NewDemux classifies each connection by its first bytes + // (proxy/demux.go) and routes it to one of two virtual listeners; + // http.Server.Serve and PostgresServer.StartListener run completely + // unmodified against those, unaware they aren't backed by a real + // socket. + dx := proxy.NewDemux(ln) + s.demux = dx + + go func() { _ = s.proxyServer.Serve(dx.HTTPListener()) }() + pg := proxy.NewPostgresServer(s.proxy) - if err := pg.StartListener(pgLn); err != nil { - _ = pgLn.Close() + if err := pg.StartListener(dx.PostgresListener()); err != nil { + // Tear down the already-running HTTP server and the demux's real + // listener so a postgres wiring failure doesn't leak either. + _ = dx.Close() shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) _ = s.proxyServer.Shutdown(shutdownCtx) cancel() @@ -880,6 +888,46 @@ func (s *Server) Start(ctx context.Context) error { s.pgAddr = pg.Addr() s.mu.Unlock() slog.Info("gatekeeper postgres listener", "addr", pg.Addr(), "subsystem", "proxy") + } else { + go func() { _ = s.proxyServer.Serve(ln) }() + + // Start the Postgres data-plane listener if configured. + if s.cfg.Postgres != nil { + pgAddr := fmt.Sprintf("%s:%d", topology.pgHost, s.cfg.Postgres.Port) + pgLn, err := net.Listen("tcp", pgAddr) + if err != nil { + // Tear down the already-running HTTP server so a postgres bind + // failure doesn't leak the HTTP listener. http.Server.Shutdown + // closes the listener it was Serve-ing, so closing the server is + // sufficient — closing ln again here would race the Serve goroutine. + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = s.proxyServer.Shutdown(shutdownCtx) + cancel() + return fmt.Errorf("starting postgres listener: %w", err) + } + // Wrap before StartListener: the PROXY header arrives as the very + // first bytes on the wire, ahead of the client's SSLRequest, so the + // listener must be PROXY-protocol-aware from the first Accept. See + // proxy.WrapProxyProtocolListener — same fail-open USE policy, 10s + // header-read timeout, and malformed-header debug log as the HTTP + // listener above. + if s.cfg.Postgres.ProxyProtocol { + pgLn = proxy.WrapProxyProtocolListener(pgLn) + } + pg := proxy.NewPostgresServer(s.proxy) + if err := pg.StartListener(pgLn); err != nil { + _ = pgLn.Close() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = s.proxyServer.Shutdown(shutdownCtx) + cancel() + return fmt.Errorf("starting postgres listener: %w", err) + } + s.mu.Lock() + s.pgServer = pg + s.pgAddr = pg.Addr() + s.mu.Unlock() + slog.Info("gatekeeper postgres listener", "addr", pg.Addr(), "subsystem", "proxy") + } } // Start background refresh goroutines for any RefreshingSource credentials. @@ -911,6 +959,16 @@ func (s *Server) Stop(ctx context.Context) error { if s.logCleanup != nil { defer s.logCleanup() } + if s.demux != nil { + // Stop accepting new connections on the shared listener immediately. + // This deliberately does not close the demux's virtual listeners + // (Demux.StopAccepting, not Close): pgServer.Shutdown and + // proxyServer.Shutdown below each close the virtual listener they own + // as part of their own graceful drain, and closing it out from under + // them here first would race their own closed-flag bookkeeping — see + // StopAccepting's doc comment in proxy/demux.go. + _ = s.demux.StopAccepting() + } // Drain both planes concurrently so each gets the full ctx budget rather // than the Postgres drain eating into the HTTP server's deadline. var wg sync.WaitGroup diff --git a/gatekeeper_multiplex_test.go b/gatekeeper_multiplex_test.go new file mode 100644 index 0000000..6d16215 --- /dev/null +++ b/gatekeeper_multiplex_test.go @@ -0,0 +1,401 @@ +package gatekeeper + +// gatekeeper_multiplex_test.go tests single-port multiplexing at the +// Server/Config level: when proxy.port and postgres.port coincide (and +// their hosts and proxy_protocol settings agree), gatekeeper.Server wires +// both planes onto one shared listener (see resolveListenTopology in +// config.go and proxy.Demux in the proxy package) instead of binding two. + +import ( + "bufio" + "bytes" + "context" + "crypto/tls" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "net/url" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/majorcontext/gatekeeper/proxy" + + "github.com/jackc/pgx/v5/pgproto3" +) + +// freeTCPPort returns a currently-unused TCP port on 127.0.0.1 by binding +// briefly and releasing it, so a caller can configure the SAME port number +// for two independent listeners (proxy.port == postgres.port) before either +// is bound. There is an inherent, very small TOCTOU race between releasing +// the port here and the caller binding it; this is the same technique +// TestServerPostgresStartFailureCleansUpHTTP already uses to occupy a port +// deliberately. +func freeTCPPort(t *testing.T) int { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + if err := ln.Close(); err != nil { + t.Fatalf("close: %v", err) + } + return port +} + +// syncLogBuffer is a concurrency-safe io.Writer for capturing slog output. +type syncLogBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncLogBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// captureDefaultSlog redirects the global slog default logger to a buffer +// for the duration of the test, restoring the previous default on cleanup. +// gatekeeper.Server's startup log lines (e.g. "gatekeeper listening", +// "proxy and postgres multiplexed on one listener") go through the package +// default logger, not a per-server logger, so this is the only way to +// observe them in a test. +func captureDefaultSlog(t *testing.T) *syncLogBuffer { + t.Helper() + buf := &syncLogBuffer{} + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return buf +} + +// newMultiplexTestConfig builds a Config with proxy.port == postgres.port +// (the multiplex trigger) pointing at a fresh CA on disk, ready for +// New/Start. It returns the *proxy.CA too, so a caller can mint a backend +// certificate with it via startTLSBackend, the same pattern +// TestServerProxyProtocol_ConnectIntercepted uses. +func newMultiplexTestConfig(t *testing.T, port int, proxyProtocol bool) (*Config, *proxy.CA) { + t.Helper() + caDir := t.TempDir() + ca, err := proxy.NewCA(caDir) + if err != nil { + t.Fatalf("NewCA: %v", err) + } + cfg := &Config{ + Proxy: ProxyConfig{Port: port, Host: "127.0.0.1", ProxyProtocol: proxyProtocol}, + TLS: TLSConfig{ + CACert: filepath.Join(caDir, "ca.crt"), + CAKey: filepath.Join(caDir, "ca.key"), + }, + Postgres: &PostgresConfig{ + Port: port, + ProxyProtocol: proxyProtocol, + }, + } + return cfg, ca +} + +// TestServerMultiplexesOnSharedPort is the core scenario this feature +// exists for at the Server/Config level: with proxy.port == postgres.port, +// gatekeeper.Server serves both a real HTTP CONNECT request (with TLS +// interception and credential injection) and a real Postgres wire-protocol +// handshake on the SAME address, and logs the multiplex startup line +// instead of the plain "gatekeeper listening" line. +func TestServerMultiplexesOnSharedPort(t *testing.T) { + port := freeTCPPort(t) + cfg, ca := newMultiplexTestConfig(t, port, false) + + var backendAuth string + var backendMu sync.Mutex + _, backendPort, caCertPool := startTLSBackend(t, ca, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + backendMu.Lock() + backendAuth = r.Header.Get("Authorization") + backendMu.Unlock() + w.WriteHeader(http.StatusOK) + })) + + cfg.Credentials = []CredentialConfig{ + { + Host: "127.0.0.1", + Header: "Authorization", + Source: SourceConfig{Type: "static", Value: "shared-port-token"}, + }, + } + cfg.Network = NetworkConfig{Policy: "permissive"} + + srv, err := New(context.Background(), cfg, "") + if err != nil { + t.Fatalf("New: %v", err) + } + srv.proxy.SetUpstreamCAs(caCertPool) + + // New already installed its own slog default (via configureLogging); only + // capture starting now so we observe Start's own log lines, not New's. + logBuf := captureDefaultSlog(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = srv.Start(ctx) }() + waitForProxy(t, srv, 2*time.Second) + + proxyAddr := srv.ProxyAddr() + pgAddr := srv.PostgresAddr() + if proxyAddr == "" || pgAddr == "" { + t.Fatalf("ProxyAddr() = %q, PostgresAddr() = %q, want both non-empty", proxyAddr, pgAddr) + } + if proxyAddr != pgAddr { + t.Fatalf("ProxyAddr() = %q, PostgresAddr() = %q, want equal — they share one listener", proxyAddr, pgAddr) + } + + log := logBuf.String() + if !strings.Contains(log, "proxy and postgres multiplexed on one listener") { + t.Errorf("startup log = %q, want it to contain the multiplex log line", log) + } + if strings.Contains(log, "gatekeeper listening\"") { + t.Errorf("startup log = %q, want the plain two-listener \"gatekeeper listening\" line NOT to appear when multiplexed", log) + } + + // --- HTTP CONNECT + TLS interception + credential injection --- + proxyURL, _ := url.Parse("http://" + proxyAddr) + client := &http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyURL(proxyURL), + TLSClientConfig: &tls.Config{RootCAs: caCertPool, MinVersion: tls.VersionTLS12}, + }, + } + resp, err := client.Get("https://127.0.0.1:" + backendPort + "/data") + if err != nil { + t.Fatalf("GET through shared listener: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + backendMu.Lock() + gotAuth := backendAuth + backendMu.Unlock() + if gotAuth != "Bearer shared-port-token" { + t.Errorf("backend Authorization = %q, want %q", gotAuth, "Bearer shared-port-token") + } + + // --- Postgres wire protocol on the SAME address --- + conn, err := net.DialTimeout("tcp", pgAddr, 2*time.Second) + if err != nil { + t.Fatalf("dial postgres on shared listener: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(2 * time.Second)) + + fe := pgproto3.NewFrontend(conn, conn) + fe.Send(&pgproto3.SSLRequest{}) + if err := fe.Flush(); err != nil { + t.Fatalf("send SSLRequest: %v", err) + } + sslResp := make([]byte, 1) + if _, err := io.ReadFull(conn, sslResp); err != nil { + t.Fatalf("read SSLRequest response: %v", err) + } + if sslResp[0] != 'S' { + t.Fatalf("SSLRequest response = %q, want 'S' (proving the shared listener speaks Postgres, not just HTTP)", sslResp[0]) + } +} + +// TestServerMultiplexed_ProxyProtocolAdvertisedAddr verifies the PROXY +// protocol interaction on a shared listener: one shared listener has one +// PROXY protocol setting (proxy.proxy_protocol, required equal to +// postgres.proxy_protocol by resolveListenTopology), and a PROXY v1 header +// sent ahead of either plane's traffic surfaces the advertised client +// address — not the raw loopback test-dialer address — in that plane's +// canonical log line. This combines TestServerProxyProtocol_ConnectIntercepted +// and TestServerPostgresProxyProtocol onto the one shared address. +func TestServerMultiplexed_ProxyProtocolAdvertisedAddr(t *testing.T) { + port := freeTCPPort(t) + cfg, ca := newMultiplexTestConfig(t, port, true) + + _, backendPort, caCertPool := startTLSBackend(t, ca, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + cfg.Network = NetworkConfig{Policy: "permissive"} + cfg.Credentials = []CredentialConfig{ + { + Host: "*.neon.tech", + Postgres: &PostgresCredentialConfig{Resolver: "static"}, + Source: SourceConfig{Type: "static", Value: "pw"}, + }, + } + + srv, err := New(context.Background(), cfg, "") + if err != nil { + t.Fatalf("New: %v", err) + } + srv.proxy.SetUpstreamCAs(caCertPool) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = srv.Start(ctx) }() + waitForProxy(t, srv, 2*time.Second) + + addr := srv.ProxyAddr() + if addr != srv.PostgresAddr() { + t.Fatalf("ProxyAddr() = %q, PostgresAddr() = %q, want equal", addr, srv.PostgresAddr()) + } + + proxyHeader := "PROXY TCP4 100.52.56.181 10.0.0.1 51234 443\r\n" + + t.Run("http", func(t *testing.T) { + waitLog := captureServerLog(t, srv) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if _, err := conn.Write([]byte(proxyHeader)); err != nil { + t.Fatalf("write PROXY header: %v", err) + } + + backendAddr := "127.0.0.1:" + backendPort + fmt.Fprintf(conn, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", backendAddr, backendAddr) + connectResp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read CONNECT response: %v", err) + } + if connectResp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT status = %d, want 200", connectResp.StatusCode) + } + + tlsConn := tls.Client(conn, &tls.Config{RootCAs: caCertPool, ServerName: "127.0.0.1", MinVersion: tls.VersionTLS12}) + if err := tlsConn.Handshake(); err != nil { + t.Fatalf("TLS handshake: %v", err) + } + defer tlsConn.Close() + + fmt.Fprintf(tlsConn, "GET /inner HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n\r\n", backendAddr) + innerResp, err := http.ReadResponse(bufio.NewReader(tlsConn), nil) + if err != nil { + t.Fatalf("read inner response: %v", err) + } + io.Copy(io.Discard, innerResp.Body) + innerResp.Body.Close() + + logged := waitLog() + host, _, err := net.SplitHostPort(logged.ClientAddr) + if err != nil { + t.Fatalf("ClientAddr = %q: SplitHostPort: %v", logged.ClientAddr, err) + } + if host != "100.52.56.181" { + t.Errorf("ClientAddr host = %q, want 100.52.56.181 (PROXY-header source)", host) + } + }) + + t.Run("postgres", func(t *testing.T) { + waitLog := captureServerLog(t, srv) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(2 * time.Second)) + + if _, err := conn.Write([]byte(proxyHeader)); err != nil { + t.Fatalf("write PROXY header: %v", err) + } + + fe := pgproto3.NewFrontend(conn, conn) + fe.Send(&pgproto3.SSLRequest{}) + if err := fe.Flush(); err != nil { + t.Fatalf("send SSLRequest: %v", err) + } + sslResp := make([]byte, 1) + if _, err := io.ReadFull(conn, sslResp); err != nil { + t.Fatalf("read SSLRequest response: %v", err) + } + if sslResp[0] != 'S' { + t.Fatalf("SSLRequest response = %q, want 'S'", sslResp[0]) + } + + tlsConn := tls.Client(conn, &tls.Config{ServerName: "db.test.local", RootCAs: caCertPool}) + if err := tlsConn.Handshake(); err != nil { + t.Fatalf("TLS handshake: %v", err) + } + defer tlsConn.Close() + + pfe := pgproto3.NewFrontend(tlsConn, tlsConn) + pfe.Send(&pgproto3.StartupMessage{ + ProtocolVersion: pgproto3.ProtocolVersionNumber, + Parameters: map[string]string{"user": "app", "database": "appdb"}, + }) + if err := pfe.Flush(); err != nil { + t.Fatalf("send startup: %v", err) + } + if _, err := pfe.Receive(); err != nil { + t.Fatalf("receive auth request: %v", err) + } + pfe.Send(&pgproto3.PasswordMessage{Password: "any-token"}) + if err := pfe.Flush(); err != nil { + t.Fatalf("send password: %v", err) + } + if _, err := pfe.Receive(); err != nil { + t.Fatalf("receive auth result: %v", err) + } + + logged := waitLog() + host, _, err := net.SplitHostPort(logged.ClientAddr) + if err != nil { + t.Fatalf("ClientAddr = %q: SplitHostPort: %v", logged.ClientAddr, err) + } + if host != "100.52.56.181" { + t.Errorf("ClientAddr host = %q, want 100.52.56.181 (PROXY-header source)", host) + } + }) +} + +// --- config validation, surfaced through New --- + +// TestNewRejectsEqualPortsDifferentHosts pins the fatal, ambiguous-bind +// error at the point a real caller would hit it: New, before any listener +// is ever bound. +func TestNewRejectsEqualPortsDifferentHosts(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1"}, + TLS: newTestCAConfig(t), + Postgres: &PostgresConfig{Port: 5432, Host: "10.0.0.5"}, + } + _, err := New(context.Background(), cfg, "") + if err == nil { + t.Fatal("New succeeded with equal ports but different hosts, want an error") + } + if !strings.Contains(err.Error(), "127.0.0.1") || !strings.Contains(err.Error(), "10.0.0.5") { + t.Errorf("error = %q, want it to name both conflicting hosts", err) + } +} + +// TestNewRejectsMismatchedProxyProtocolOnSharedPort pins the fatal error for +// a shared listener with disagreeing PROXY protocol settings. +func TestNewRejectsMismatchedProxyProtocolOnSharedPort(t *testing.T) { + cfg := &Config{ + Proxy: ProxyConfig{Port: 5432, Host: "127.0.0.1", ProxyProtocol: true}, + TLS: newTestCAConfig(t), + Postgres: &PostgresConfig{Port: 5432, ProxyProtocol: false}, + } + _, err := New(context.Background(), cfg, "") + if err == nil { + t.Fatal("New succeeded with mismatched proxy_protocol settings on a shared port, want an error") + } + if !strings.Contains(err.Error(), "proxy_protocol") { + t.Errorf("error = %q, want it to mention proxy_protocol", err) + } +} diff --git a/proxy/demux.go b/proxy/demux.go new file mode 100644 index 0000000..279d6cc --- /dev/null +++ b/proxy/demux.go @@ -0,0 +1,336 @@ +package proxy + +// demux.go implements single-port multiplexing: when gatekeeper's HTTP/ +// CONNECT proxy listener and its Postgres data-plane listener are configured +// on the same address, one real net.Listener carries both. Each accepted +// connection is classified by its first bytes in its own goroutine — never +// in the shared accept loop, so a silent or slow client can never stall +// Accept for every other pending connection — and routed to one of two +// in-memory virtual listeners. http.Server.Serve and +// PostgresServer.StartListener then run completely unmodified against those +// virtual listeners, unaware they aren't backed by a real socket. +// +// This is a hand-rolled, minimal cmux-style demultiplexer: gatekeeper is +// dependency-conscious, so rather than adding a cmux dependency this reuses +// the existing connection-wrapping pattern from proxyProtoLogConn +// (proxyproto.go), which already holds and replays bytes consumed ahead of +// a connection's real payload. + +import ( + "fmt" + "io" + "log/slog" + "net" + "sync" + "sync/atomic" + "time" +) + +// demuxSniffLen is the number of leading bytes the demux reads from a newly +// accepted connection before routing it. It is exactly enough to positively +// identify every Postgres startup signature gatekeeper's data plane +// recognizes: a 4-byte big-endian header followed by either a well-known +// magic code (SSLRequest, GSSENCRequest) or protocol version 3.0 +// (StartupMessage). Anything else — an HTTP/1.x request line, the h2 client +// preface, or any other prefix — is routed to the HTTP plane by default. +const demuxSniffLen = 8 + +// demuxSniffDeadline bounds how long the demux waits for the first +// demuxSniffLen bytes of a new connection before giving up on it. It reuses +// the same 10s window as WrapProxyProtocolListener's PROXY header read (see +// proxyproto.go), so a silent client — an LB health check that opens a +// socket and waits, or a slow-loris — is dropped on the same timescale +// wherever gatekeeper reads a connection's opening bytes. Once +// classification succeeds, this deadline is cleared before the connection +// is handed to a downstream server, so only that server's own timeouts +// (http.Server's ReadHeaderTimeout, or PostgresServer's handshake timeout) +// ever apply from that point on. +const demuxSniffDeadline = 10 * time.Second + +// demuxBacklog bounds each virtual listener's queue of classified, +// not-yet-Accepted connections. It absorbs a burst of concurrent handshakes +// arriving faster than the downstream server's Accept loop drains them; a +// connection that can't be enqueued (backlog full, or the listener already +// closed) is closed by the dispatcher instead of blocking the accept loop +// that's classifying other connections. +const demuxBacklog = 64 + +// demuxProtocol identifies which plane a connection belongs to. +type demuxProtocol int + +const ( + demuxHTTP demuxProtocol = iota + demuxPostgres +) + +// String returns the protocol label used in the per-connection DEBUG log +// line. It never includes any connection content — only "http" or +// "postgres". +func (p demuxProtocol) String() string { + if p == demuxPostgres { + return "postgres" + } + return "http" +} + +// classifyPrefix reports which plane a connection belongs to, given its +// first demuxSniffLen bytes. Classification is positive-match Postgres — +// the same approach as Caddy-L4's postgres matcher: only a recognized +// Postgres startup signature routes to demuxPostgres, and every other +// prefix defaults to demuxHTTP. +func classifyPrefix(prefix []byte) demuxProtocol { + if isPostgresStartup(prefix) { + return demuxPostgres + } + return demuxHTTP +} + +// isPostgresStartup reports whether prefix opens with one of the three +// Postgres startup signatures gatekeeper's data plane accepts: +// +// - SSLRequest: length 00 00 00 08, code 04 d2 16 2f (80877103) +// - GSSENCRequest: length 00 00 00 08, code 04 d2 16 30 (80877104) +// - v3 StartupMessage: any 4-byte length, then protocol 00 03 00 00 +// +// A prefix shorter than demuxSniffLen can never match. +func isPostgresStartup(prefix []byte) bool { + if len(prefix) < demuxSniffLen { + return false + } + if prefix[0] == 0x00 && prefix[1] == 0x00 && prefix[2] == 0x00 && prefix[3] == 0x08 && + prefix[4] == 0x04 && prefix[5] == 0xd2 && prefix[6] == 0x16 && + (prefix[7] == 0x2f || prefix[7] == 0x30) { + return true + } + return prefix[4] == 0x00 && prefix[5] == 0x03 && prefix[6] == 0x00 && prefix[7] == 0x00 +} + +// demuxConn wraps a newly accepted connection so the demuxSniffLen bytes +// consumed while classifying it are replayed on the first Read calls, ahead +// of any further bytes from the underlying conn. This mirrors +// proxyProtoLogConn's hold-then-replay pattern in proxyproto.go, which +// exists for the same reason: net.Conn has no way to "un-read" bytes once +// they're consumed, so the bytes read to classify a connection must be +// spliced back in front of it for its actual owner (http.Server or +// PostgresServer) to see the whole, unmodified byte stream. +type demuxConn struct { + net.Conn + prefix []byte +} + +func (c *demuxConn) Read(b []byte) (int, error) { + if len(c.prefix) > 0 { + n := copy(b, c.prefix) + c.prefix = c.prefix[n:] + return n, nil + } + return c.Conn.Read(b) +} + +// Raw unwraps to the innermost connection, so a caller that needs the real +// transport conn — e.g. postgres.go's underlyingTCPConn, for TCP keep-alive +// setup — can reach it through a demuxConn exactly as it already does +// through a proxyProtoLogConn (proxyproto.go). It recurses through any +// further Raw()-implementing wrapper beneath it, e.g. a proxyProtoLogConn +// when the shared listener is also PROXY-protocol-wrapped. +func (c *demuxConn) Raw() net.Conn { + if rc, ok := c.Conn.(interface{ Raw() net.Conn }); ok { + return rc.Raw() + } + return c.Conn +} + +// virtualListener is an in-memory net.Listener fed by a Demux's dispatcher +// goroutine instead of a real socket. http.Server.Serve and +// PostgresServer.StartListener run against it completely unmodified. +type virtualListener struct { + addr net.Addr + + mu sync.Mutex + closed bool + conns chan net.Conn + done chan struct{} +} + +func newVirtualListener(addr net.Addr) *virtualListener { + return &virtualListener{ + addr: addr, + conns: make(chan net.Conn, demuxBacklog), + done: make(chan struct{}), + } +} + +// push enqueues conn for a future Accept call. It returns false — without +// blocking — when the listener is already closed or its backlog is full, so +// the dispatcher can close conn itself instead of leaking it or stalling on +// a slow Accept loop. +func (l *virtualListener) push(conn net.Conn) bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.closed { + return false + } + select { + case l.conns <- conn: + return true + default: + return false + } +} + +func (l *virtualListener) Accept() (net.Conn, error) { + select { + case conn := <-l.conns: + return conn, nil + case <-l.done: + return nil, net.ErrClosed + } +} + +// Close marks the listener closed and closes any connection still queued +// but never Accepted, so it doesn't leak a file descriptor with nobody left +// to close it. Nothing can push after closed is set (push takes the same +// lock), so the drain below is race-free. +func (l *virtualListener) Close() error { + l.mu.Lock() + if l.closed { + l.mu.Unlock() + return nil + } + l.closed = true + close(l.done) + l.mu.Unlock() + + for { + select { + case conn := <-l.conns: + conn.Close() + default: + return nil + } + } +} + +func (l *virtualListener) Addr() net.Addr { return l.addr } + +// Demux owns a single real listener carrying both HTTP/CONNECT proxy +// traffic and Postgres data-plane traffic. It classifies each accepted +// connection by its first bytes (see classifyPrefix) in its own goroutine — +// never in the accept loop — and routes it to one of two virtual listeners, +// so a silent or slow client can never stall Accept for every other pending +// connection (the same discipline WrapProxyProtocolListener's Accept +// follows, and for the same reason: see proxyproto.go). +type Demux struct { + ln net.Listener + httpVL *virtualListener + pgVL *virtualListener + closed atomic.Bool +} + +// NewDemux begins accepting connections on ln — which may already be +// wrapped with WrapProxyProtocolListener, in which case its PROXY header (if +// any) is consumed lazily on the demux's own sniff Read, ahead of +// classification — and routes each one to HTTPListener or PostgresListener +// based on its first bytes. +func NewDemux(ln net.Listener) *Demux { + d := &Demux{ + ln: ln, + httpVL: newVirtualListener(ln.Addr()), + pgVL: newVirtualListener(ln.Addr()), + } + go d.acceptLoop() + return d +} + +// HTTPListener returns the virtual listener carrying HTTP/CONNECT proxy +// traffic. Pass it to http.Server.Serve unmodified. +func (d *Demux) HTTPListener() net.Listener { return d.httpVL } + +// PostgresListener returns the virtual listener carrying Postgres +// data-plane traffic. Pass it to PostgresServer.StartListener unmodified. +func (d *Demux) PostgresListener() net.Listener { return d.pgVL } + +// Close closes the real listener and both virtual listeners. It's meant for +// a Demux used standalone (e.g. tests): an embedder that separately owns +// graceful shutdown of the two downstream servers — as gatekeeper.go does — +// should call StopAccepting instead and let each server close its own +// virtual listener as part of its own Shutdown/Stop. +func (d *Demux) Close() error { + err := d.StopAccepting() + d.httpVL.Close() + d.pgVL.Close() + return err +} + +// StopAccepting closes only the real listener, so no further connections +// are accepted and classified. It deliberately leaves both virtual +// listeners open: PostgresServer.Shutdown/Stop and http.Server.Shutdown each +// close the virtual listener they were started on as part of their own +// graceful drain, and closing it out from under them here first would race +// their own closed-flag bookkeeping — for PostgresServer that would surface +// as a spurious "postgres accept loop exited" error log on an entirely +// ordinary shutdown. +func (d *Demux) StopAccepting() error { + d.closed.Store(true) + return d.ln.Close() +} + +func (d *Demux) acceptLoop() { + for { + conn, err := d.ln.Accept() + if err != nil { + if !d.closed.Load() { + slog.Error("demux accept loop exited", "subsystem", "proxy", "error", err) + } + return + } + go d.classifyAndDispatch(conn) + } +} + +// classifyAndDispatch runs in its own goroutine per connection — never in +// acceptLoop — so a silent or slow-to-classify client blocks only itself, +// not Accept for every other pending connection. +func (d *Demux) classifyAndDispatch(conn net.Conn) { + proto, sniffed, err := sniffProtocol(conn) + if err != nil { + // Never log connection content — only that classification failed and + // why (a timeout, a short read, or a closed peer), never any bytes. + slog.Debug("demux: dropping connection", "subsystem", "proxy", "err", err) + conn.Close() + return + } + + vl := d.httpVL + if proto == demuxPostgres { + vl = d.pgVL + } + slog.Debug("demux: classified connection", "subsystem", "proxy", "protocol", proto.String()) + if !vl.push(sniffed) { + sniffed.Close() + } +} + +// sniffProtocol reads the first demuxSniffLen bytes of conn, bounded by +// demuxSniffDeadline, and classifies them. On success it returns a conn +// that replays those bytes before reading any more from the underlying +// connection (see demuxConn), with the read deadline it set cleared so only +// the downstream server's own timeouts apply from here on. On failure — the +// deadline expires with nothing sent (a silent client), the peer closes +// early (a short or malformed opener), or any other read error — it returns +// a non-nil error and the caller drops the connection. +func sniffProtocol(conn net.Conn) (demuxProtocol, net.Conn, error) { + if err := conn.SetReadDeadline(time.Now().Add(demuxSniffDeadline)); err != nil { + return demuxHTTP, nil, fmt.Errorf("set sniff deadline: %w", err) + } + prefix := make([]byte, demuxSniffLen) + n, readErr := io.ReadFull(conn, prefix) + clearErr := conn.SetReadDeadline(time.Time{}) + if readErr != nil { + return demuxHTTP, nil, fmt.Errorf("read sniff prefix (%d/%d bytes): %w", n, demuxSniffLen, readErr) + } + if clearErr != nil { + return demuxHTTP, nil, fmt.Errorf("clear sniff deadline: %w", clearErr) + } + return classifyPrefix(prefix), &demuxConn{Conn: conn, prefix: prefix}, nil +} diff --git a/proxy/demux_test.go b/proxy/demux_test.go new file mode 100644 index 0000000..6db92bb --- /dev/null +++ b/proxy/demux_test.go @@ -0,0 +1,864 @@ +package proxy + +// demux_test.go tests the single-port multiplexer: classification of a +// connection's first bytes as HTTP/CONNECT proxy traffic or Postgres +// data-plane traffic, the virtual listeners that feed http.Server and +// PostgresServer unmodified, and the end-to-end wiring that lets both +// planes share one real listener. + +import ( + "bufio" + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgproto3" +) + +// --- classification ------------------------------------------------------- + +func TestClassifyPrefix_PostgresSSLRequest(t *testing.T) { + // length 00 00 00 08, code 04 d2 16 2f (80877103). + prefix := []byte{0x00, 0x00, 0x00, 0x08, 0x04, 0xd2, 0x16, 0x2f} + if got := classifyPrefix(prefix); got != demuxPostgres { + t.Errorf("classifyPrefix(SSLRequest) = %v, want demuxPostgres", got) + } +} + +func TestClassifyPrefix_PostgresGSSENCRequest(t *testing.T) { + // length 00 00 00 08, code 04 d2 16 30 (80877104). + prefix := []byte{0x00, 0x00, 0x00, 0x08, 0x04, 0xd2, 0x16, 0x30} + if got := classifyPrefix(prefix); got != demuxPostgres { + t.Errorf("classifyPrefix(GSSENCRequest) = %v, want demuxPostgres", got) + } +} + +func TestClassifyPrefix_PostgresV3StartupMessage(t *testing.T) { + tests := []struct { + name string + length [4]byte + }{ + // A real StartupMessage's length reflects the full message + // (parameters included), which varies per connection — the + // classifier must not care what it is, only that the protocol + // version that follows is 3.0. + {"typical length", [4]byte{0x00, 0x00, 0x00, 0x29}}, + {"minimal length", [4]byte{0x00, 0x00, 0x00, 0x08}}, + {"large length", [4]byte{0x00, 0x00, 0x01, 0x00}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + prefix := append(tt.length[:], 0x00, 0x03, 0x00, 0x00) + if got := classifyPrefix(prefix); got != demuxPostgres { + t.Errorf("classifyPrefix(v3 StartupMessage, %s) = %v, want demuxPostgres", tt.name, got) + } + }) + } +} + +func TestClassifyPrefix_HTTPDefaults(t *testing.T) { + tests := []struct { + name string + prefix string + }{ + {"GET", "GET / HTTP/1.1\r\n"}, + {"POST", "POST /x HTTP/1.1\r\n"}, + {"PUT", "PUT /x HTTP/1.1\r\n"}, + {"DELETE", "DELETE /x HTTP/1.1\r\n"}, + {"PATCH", "PATCH /x HTTP/1.1\r\n"}, + {"HEAD", "HEAD / HTTP/1.1\r\n"}, + {"OPTIONS", "OPTIONS * HTTP/1.1\r\n"}, + {"CONNECT", "CONNECT example.com:443 HTTP/1.1\r\n"}, + {"TRACE", "TRACE / HTTP/1.1\r\n"}, + // h2 client connection preface (RFC 7540 3.5). + {"h2 preface", "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"}, + // Bytes that don't match any Postgres startup signature and aren't + // even a plausible HTTP method still default to HTTP — the + // classifier only positively matches Postgres. + {"garbage", "\x01\x02\x03\x04\x05\x06\x07\x08"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + prefix := []byte(tt.prefix) + if len(prefix) < demuxSniffLen { + t.Fatalf("test prefix %q shorter than demuxSniffLen (%d)", tt.prefix, demuxSniffLen) + } + if got := classifyPrefix(prefix[:demuxSniffLen]); got != demuxHTTP { + t.Errorf("classifyPrefix(%q) = %v, want demuxHTTP", tt.name, got) + } + }) + } +} + +// TestClassifyPrefix_LengthAloneIsNotEnoughToMatch guards against a +// classifier that keys only on the length field (00 00 00 08) shared by +// SSLRequest and GSSENCRequest: a StartupMessage-shaped 8-byte message (an +// implausibly small one, but not impossible) must not be misclassified just +// because its length happens to also be 8. +func TestClassifyPrefix_LengthAloneIsNotEnoughToMatch(t *testing.T) { + prefix := []byte{0x00, 0x00, 0x00, 0x08, 0x00, 0x03, 0x00, 0x00} + if got := classifyPrefix(prefix); got != demuxPostgres { + t.Errorf("classifyPrefix(8-byte v3 StartupMessage) = %v, want demuxPostgres (still a valid v3 startup signature)", got) + } + + // But an SSLRequest-length message with neither the SSLRequest/GSSENCRequest + // code nor the v3 protocol version must default to HTTP. + prefix2 := []byte{0x00, 0x00, 0x00, 0x08, 0xff, 0xff, 0xff, 0xff} + if got := classifyPrefix(prefix2); got != demuxHTTP { + t.Errorf("classifyPrefix(length-8, unrecognized code) = %v, want demuxHTTP", got) + } +} + +// --- byte replay ------------------------------------------------------ + +// TestSniffProtocol_ReplaysPeekedBytesBeforeUnderlyingConn verifies that +// sniffProtocol's returned conn replays the sniffed prefix on Read before +// any further bytes reach the caller — the same hold-then-replay contract +// proxyProtoLogConn upholds for the PROXY protocol header. +func TestSniffProtocol_ReplaysPeekedBytesBeforeUnderlyingConn(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + + payload := "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n" + go func() { + client.Write([]byte(payload)) + }() + + proto, sniffed, err := sniffProtocol(server) + if err != nil { + t.Fatalf("sniffProtocol: %v", err) + } + defer sniffed.Close() + if proto != demuxHTTP { + t.Fatalf("proto = %v, want demuxHTTP", proto) + } + + got := make([]byte, len(payload)) + if _, err := io.ReadFull(sniffed, got); err != nil { + t.Fatalf("ReadFull: %v", err) + } + if string(got) != payload { + t.Errorf("replayed+continued bytes = %q, want %q", got, payload) + } +} + +// deadlineRecordingConn wraps a net.Conn and records every SetReadDeadline +// call, so a test can assert on the final deadline value without depending +// on real-time timing. +type deadlineRecordingConn struct { + net.Conn + mu sync.Mutex + calls []time.Time +} + +func (c *deadlineRecordingConn) SetReadDeadline(t time.Time) error { + c.mu.Lock() + c.calls = append(c.calls, t) + c.mu.Unlock() + return c.Conn.SetReadDeadline(t) +} + +func (c *deadlineRecordingConn) deadlineCalls() []time.Time { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]time.Time, len(c.calls)) + copy(out, c.calls) + return out +} + +// TestSniffProtocol_ClearsDeadlineOnSuccess verifies that a successful sniff +// clears the read deadline it set (the last SetReadDeadline call is the zero +// value), so the downstream server's own timeout governs subsequent reads +// instead of the sniff window silently persisting. +func TestSniffProtocol_ClearsDeadlineOnSuccess(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + defer server.Close() + + rec := &deadlineRecordingConn{Conn: server} + + go func() { client.Write([]byte("GET / HTTP/1.1\r\n")) }() + + _, sniffed, err := sniffProtocol(rec) + if err != nil { + t.Fatalf("sniffProtocol: %v", err) + } + defer sniffed.Close() + + calls := rec.deadlineCalls() + if len(calls) == 0 { + t.Fatal("sniffProtocol never called SetReadDeadline") + } + last := calls[len(calls)-1] + if !last.IsZero() { + t.Errorf("last SetReadDeadline call = %v, want the zero value (no deadline), so the sniff window doesn't leak into the downstream server's own timeout handling", last) + } +} + +// TestSniffProtocol_ClearsDeadlineOnFailure verifies the deadline is cleared +// even when the sniff itself fails (a short connection), so a caller that +// reuses the raw conn after logging the drop never inherits a stale +// deadline either. This uses a real TCP loopback conn rather than +// net.Pipe: net.Pipe ties both ends' deadline machinery together, so once +// one side closes, SetReadDeadline on the other side starts failing too — +// which would mask the very call sequence this test verifies. +func TestSniffProtocol_ClearsDeadlineOnFailure(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + acceptedCh := make(chan net.Conn, 1) + go func() { + c, err := ln.Accept() + if err != nil { + close(acceptedCh) + return + } + acceptedCh <- c + }() + + client, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + // Fewer than demuxSniffLen bytes, then close: a short/malformed opener. + if _, err := client.Write([]byte{0x01, 0x02}); err != nil { + t.Fatalf("write: %v", err) + } + client.Close() + + server := <-acceptedCh + if server == nil { + t.Fatal("Accept failed") + } + defer server.Close() + + rec := &deadlineRecordingConn{Conn: server} + _, _, sniffErr := sniffProtocol(rec) + if sniffErr == nil { + t.Fatal("sniffProtocol succeeded on a short connection, want an error") + } + + calls := rec.deadlineCalls() + if len(calls) < 2 { + t.Fatalf("SetReadDeadline called %d times, want at least 2 (set, then clear)", len(calls)) + } + last := calls[len(calls)-1] + if !last.IsZero() { + t.Errorf("last SetReadDeadline call = %v, want the zero value even on a failed sniff", last) + } +} + +// --- virtualListener -------------------------------------------------- + +type fakeConn struct { + net.Conn + closed bool +} + +func (c *fakeConn) Close() error { + c.closed = true + return nil +} + +func TestVirtualListener_PushThenAccept(t *testing.T) { + vl := newVirtualListener(&net.TCPAddr{}) + c := &fakeConn{} + if !vl.push(c) { + t.Fatal("push returned false, want true") + } + got, err := vl.Accept() + if err != nil { + t.Fatalf("Accept: %v", err) + } + if got != net.Conn(c) { + t.Error("Accept returned a different conn than was pushed") + } +} + +func TestVirtualListener_CloseUnblocksAccept(t *testing.T) { + vl := newVirtualListener(&net.TCPAddr{}) + errCh := make(chan error, 1) + go func() { + _, err := vl.Accept() + errCh <- err + }() + // Give Accept a moment to block before closing. + time.Sleep(20 * time.Millisecond) + vl.Close() + select { + case err := <-errCh: + if err != net.ErrClosed { + t.Errorf("Accept error = %v, want net.ErrClosed", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Accept did not unblock after Close") + } +} + +func TestVirtualListener_PushAfterCloseReturnsFalseAndDoesNotLeak(t *testing.T) { + vl := newVirtualListener(&net.TCPAddr{}) + vl.Close() + c := &fakeConn{} + if vl.push(c) { + t.Error("push after Close returned true, want false") + } +} + +func TestVirtualListener_CloseClosesQueuedButUnacceptedConns(t *testing.T) { + vl := newVirtualListener(&net.TCPAddr{}) + c := &fakeConn{} + if !vl.push(c) { + t.Fatal("push returned false, want true") + } + vl.Close() + if !c.closed { + t.Error("conn queued but never Accepted was not closed by Close, want it closed to avoid leaking the file descriptor") + } +} + +func TestVirtualListener_BacklogFullDropsInsteadOfBlocking(t *testing.T) { + vl := newVirtualListener(&net.TCPAddr{}) + // Fill the backlog. + for i := 0; i < demuxBacklog; i++ { + if !vl.push(&fakeConn{}) { + t.Fatalf("push %d failed before backlog was full", i) + } + } + // One more must not block and must report failure. + done := make(chan bool, 1) + go func() { done <- vl.push(&fakeConn{}) }() + select { + case ok := <-done: + if ok { + t.Error("push into a full backlog returned true, want false") + } + case <-time.After(2 * time.Second): + t.Fatal("push into a full backlog blocked, want a non-blocking false") + } +} + +// --- Demux: routing ----------------------------------------------------- + +func newTestDemux(t *testing.T) (*Demux, net.Listener) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + d := NewDemux(ln) + t.Cleanup(func() { d.Close() }) + return d, ln +} + +func TestDemux_RoutesHTTPConnToHTTPListener(t *testing.T) { + d, ln := newTestDemux(t) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")); err != nil { + t.Fatalf("write: %v", err) + } + + accepted := acceptWithTimeout(t, d.HTTPListener(), 2*time.Second) + defer accepted.Close() + + buf := make([]byte, len("GET / HTTP/1.1\r\n")) + if _, err := io.ReadFull(accepted, buf); err != nil { + t.Fatalf("read replayed bytes: %v", err) + } + if !bytes.HasPrefix(buf, []byte("GET / HTTP/1.1")) { + t.Errorf("replayed bytes = %q, want prefix %q", buf, "GET / HTTP/1.1") + } + + select { + case c := <-acceptAsync(d.PostgresListener()): + if c != nil { + c.Close() + } + t.Fatal("HTTP conn was also routed to the Postgres listener") + case <-time.After(200 * time.Millisecond): + // Expected: nothing arrives on the Postgres listener. + } +} + +func TestDemux_RoutesPostgresConnToPostgresListener(t *testing.T) { + d, ln := newTestDemux(t) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + + fe := pgproto3.NewFrontend(conn, conn) + fe.Send(&pgproto3.SSLRequest{}) + if err := fe.Flush(); err != nil { + t.Fatalf("send SSLRequest: %v", err) + } + + accepted := acceptWithTimeout(t, d.PostgresListener(), 2*time.Second) + defer accepted.Close() + + // SSLRequest is exactly 8 bytes on the wire: length(4) + code(4). + buf := make([]byte, 8) + if _, err := io.ReadFull(accepted, buf); err != nil { + t.Fatalf("read replayed bytes: %v", err) + } + want := []byte{0x00, 0x00, 0x00, 0x08, 0x04, 0xd2, 0x16, 0x2f} + if !bytes.Equal(buf, want) { + t.Errorf("replayed bytes = %x, want %x", buf, want) + } + + select { + case c := <-acceptAsync(d.HTTPListener()): + if c != nil { + c.Close() + } + t.Fatal("Postgres conn was also routed to the HTTP listener") + case <-time.After(200 * time.Millisecond): + } +} + +func acceptWithTimeout(t *testing.T, ln net.Listener, timeout time.Duration) net.Conn { + t.Helper() + ch := acceptAsync(ln) + select { + case c := <-ch: + if c == nil { + t.Fatal("Accept returned a nil conn") + } + return c + case <-time.After(timeout): + t.Fatalf("Accept on %v timed out after %v", ln.Addr(), timeout) + return nil + } +} + +func acceptAsync(ln net.Listener) <-chan net.Conn { + ch := make(chan net.Conn, 1) + go func() { + c, err := ln.Accept() + if err != nil { + ch <- nil + return + } + ch <- c + }() + return ch +} + +// TestDemux_SilentClientDoesNotBlockOtherAccepts guards the same class of bug +// TestWrapProxyProtocolListenerAcceptDoesNotBlockOnSilentClient guards in +// proxyproto_test.go: classification must happen in a per-connection +// goroutine, never in the shared accept loop, so one silent client can never +// stall Accept for every other pending connection. +func TestDemux_SilentClientDoesNotBlockOtherAccepts(t *testing.T) { + d, ln := newTestDemux(t) + + silent, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial (silent): %v", err) + } + defer silent.Close() + // Deliberately write nothing. + + normal, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial (normal): %v", err) + } + defer normal.Close() + if _, err := normal.Write([]byte("GET / HTTP/1.1\r\n\r\n")); err != nil { + t.Fatalf("write: %v", err) + } + + // The 2s bound is well clear of demuxSniffDeadline (10s): a correct demux + // classifies the normal connection immediately, independent of the + // silent one still being sniffed in its own goroutine. + accepted := acceptWithTimeout(t, d.HTTPListener(), 2*time.Second) + accepted.Close() +} + +// TestDemux_MalformedShortConnDroppedNoPanic verifies that a connection +// which sends fewer than demuxSniffLen bytes and then closes is dropped — +// routed to neither virtual listener — without panicking the dispatcher, and +// that the drop is logged at DEBUG. +func TestDemux_MalformedShortConnDroppedNoPanic(t *testing.T) { + buf := captureSlogText(t) + d, ln := newTestDemux(t) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + if _, err := conn.Write([]byte{0x01, 0x02}); err != nil { + t.Fatalf("write: %v", err) + } + conn.Close() + + select { + case c := <-acceptAsync(d.HTTPListener()): + if c != nil { + c.Close() + } + t.Fatal("short/malformed conn was routed to the HTTP listener, want dropped") + case <-time.After(300 * time.Millisecond): + } + select { + case c := <-acceptAsync(d.PostgresListener()): + if c != nil { + c.Close() + } + t.Fatal("short/malformed conn was routed to the Postgres listener, want dropped") + case <-time.After(300 * time.Millisecond): + } + + got := waitForLogContaining(buf, "demux") + if !strings.Contains(got, "demux") { + t.Errorf("log output = %q, want a DEBUG line mentioning the demux dropping the connection", got) + } +} + +// --- Demux: lifecycle ----------------------------------------------------- + +func TestDemux_CloseClosesRealListenerAndBothVirtualListeners(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + d := NewDemux(ln) + addr := ln.Addr().String() + + if err := d.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + if _, err := net.DialTimeout("tcp", addr, time.Second); err == nil { + t.Error("dial succeeded after Close, want the real listener closed") + } + if _, err := d.HTTPListener().Accept(); err != net.ErrClosed { + t.Errorf("HTTPListener().Accept() after Close = %v, want net.ErrClosed", err) + } + if _, err := d.PostgresListener().Accept(); err != net.ErrClosed { + t.Errorf("PostgresListener().Accept() after Close = %v, want net.ErrClosed", err) + } +} + +// TestDemux_StopAcceptingLeavesVirtualListenersOpen verifies StopAccepting's +// narrower contract (used by gatekeeper.go, which owns the two downstream +// servers' own graceful shutdown): the real listener stops taking new +// connections, but both virtual listeners stay open for their owners to +// close in their own time. +func TestDemux_StopAcceptingLeavesVirtualListenersOpen(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + d := NewDemux(ln) + addr := ln.Addr().String() + + if err := d.StopAccepting(); err != nil { + t.Fatalf("StopAccepting: %v", err) + } + + if _, err := net.DialTimeout("tcp", addr, time.Second); err == nil { + t.Error("dial succeeded after StopAccepting, want the real listener closed") + } + + // Both virtual listeners must still be open — pushing into them and + // Accepting must still work — until their owners close them. + c := &fakeConn{} + if !d.HTTPListener().(*virtualListener).push(c) { + t.Fatal("push into HTTP virtual listener failed after StopAccepting, want it still open") + } + got, err := d.HTTPListener().Accept() + if err != nil { + t.Fatalf("Accept on HTTP virtual listener after StopAccepting: %v", err) + } + if got != net.Conn(c) { + t.Error("Accept returned a different conn than was pushed") + } + + d.HTTPListener().Close() + d.PostgresListener().Close() +} + +// --- end-to-end: both planes share one port ------------------------------- + +// newDemuxE2ESetup wires an http.Server and a PostgresServer onto one shared +// Demux listener — exactly how gatekeeper.go wires them when +// postgres.port == proxy.port — backed by a real HTTPS backend (for +// CONNECT + TLS interception + credential injection) and a real fake +// Postgres server (for SCRAM upstream auth). ln may already be wrapped +// (e.g. with WrapProxyProtocolListener); the caller retains it to dial +// against. +type demuxE2ESetup struct { + CA *CA + Proxy *Proxy + Backend *httptest.Server + Postgres *PostgresServer + FakePostgres *fakePostgresServer + Addr string + ReceivedAuth func() string +} + +func newDemuxE2ESetup(t *testing.T, ln net.Listener) *demuxE2ESetup { + t.Helper() + + ca, err := generateCA() + if err != nil { + t.Fatalf("generateCA: %v", err) + } + + var mu sync.Mutex + var receivedAuth string + backend := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + receivedAuth = r.Header.Get("Authorization") + mu.Unlock() + w.Write([]byte("ok")) + })) + t.Cleanup(backend.Close) + + fake := startFakePostgres(t, "ep-foo-123.aws.neon.tech", "app_rw", "real-password") + + p := NewProxy() + p.SetCA(ca) + // Trust both upstream TLS identities the shared listener will need to + // reach: the HTTPS backend (CONNECT-relayed) and the fake Postgres + // server (SCRAM upstream). + combinedCAs := x509.NewCertPool() + combinedCAs.AddCert(backend.Certificate()) + combinedCAs.AddCert(fake.cert.Leaf) + p.SetUpstreamCAs(combinedCAs) + // No proxy.SetAuthToken: leaving it unset means both planes accept any + // token ("localhost trust", matching HTTP's and Postgres's behavior when + // no auth is configured) — the run-token literal below exists only to + // exercise the Postgres password-carries-the-token wire format, not to + // be validated against anything. + p.SetPostgresResolver("*.neon.tech", NewStaticPostgresResolver("real-password")) + p.SetCredentialWithGrant(mustParseURL(backend.URL).Hostname(), "Authorization", "Bearer test-token-123", "test-grant") + + dx := NewDemux(ln) + t.Cleanup(func() { dx.Close() }) + + httpServer := &http.Server{Handler: p} + go func() { _ = httpServer.Serve(dx.HTTPListener()) }() + t.Cleanup(func() { httpServer.Close() }) + + pg := NewPostgresServer(p) + pg.dialUpstream = func(_ context.Context, _ string) (string, error) { + return fake.addr, nil + } + if err := pg.StartListener(dx.PostgresListener()); err != nil { + t.Fatalf("StartListener: %v", err) + } + t.Cleanup(pg.Stop) + + return &demuxE2ESetup{ + CA: ca, + Proxy: p, + Backend: backend, + Postgres: pg, + FakePostgres: fake, + Addr: ln.Addr().String(), + ReceivedAuth: func() string { + mu.Lock() + defer mu.Unlock() + return receivedAuth + }, + } +} + +// TestDemuxEndToEnd_HTTPAndPostgresShareOnePort is the core scenario this +// feature exists for: a real HTTP CONNECT request with TLS interception and +// credential injection, and a real Postgres handshake authenticated and +// relayed upstream, both succeed against the SAME listener address — proving +// http.Server and PostgresServer genuinely run unmodified against their +// virtual listeners. +func TestDemuxEndToEnd_HTTPAndPostgresShareOnePort(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + setup := newDemuxE2ESetup(t, ln) + + // --- HTTP CONNECT + TLS interception + credential injection --- + clientCAs := x509.NewCertPool() + clientCAs.AppendCertsFromPEM(setup.CA.CertPEM()) + client := &http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyURL(mustParseURL("http://" + setup.Addr)), + TLSClientConfig: &tls.Config{RootCAs: clientCAs}, + }, + } + resp, err := client.Get(setup.Backend.URL + "/api/data") + if err != nil { + t.Fatalf("HTTP request through shared listener: %v", err) + } + defer resp.Body.Close() + io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("HTTP status = %d, want 200", resp.StatusCode) + } + if got := setup.ReceivedAuth(); got != "Bearer test-token-123" { + t.Errorf("backend Authorization = %q, want %q", got, "Bearer test-token-123") + } + + // --- Postgres handshake + SCRAM upstream + query relay --- + conn, err := connectThroughGatekeeper(t, setup.Postgres, caTrustPool(t, setup.CA), + "ep-foo-123.aws.neon.tech", "app_rw", "appdb", "run-token") + if err != nil { + t.Fatalf("connect through shared listener (postgres): %v", err) + } + res, err := conn.Exec(context.Background(), "SELECT 1").ReadAll() + if err != nil { + t.Fatalf("Exec: %v", err) + } + if len(res) == 0 || res[0].Err != nil { + t.Fatalf("query result = %+v", res) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + conn.Close(ctx) +} + +// TestDemuxEndToEnd_ProxyProtocolAdvertisedAddrOnBothPlanes verifies the +// PROXY protocol interaction: a shared listener wrapped with +// WrapProxyProtocolListener strips a leading PROXY v1 header lazily on the +// demux's own sniff Read (before classification), for both planes — so the +// advertised client address, not the raw loopback test-dialer address, +// reaches each plane's request log. +func TestDemuxEndToEnd_ProxyProtocolAdvertisedAddrOnBothPlanes(t *testing.T) { + base, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + ln := WrapProxyProtocolListener(base) + setup := newDemuxE2ESetup(t, ln) + + proxyHeader := "PROXY TCP4 100.52.56.181 10.0.0.1 51234 443\r\n" + + t.Run("http", func(t *testing.T) { + // A fresh logCapture per subtest: the two subtests share setup.Proxy, + // and reinstalling the logger here (rather than sharing one capture + // across both) keeps each subtest's "exactly one entry" assertion + // from racing the other subtest's request. + cap := &logCapture{} + setup.Proxy.SetLogger(cap.log) + + conn, err := net.Dial("tcp", setup.Addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if _, err := conn.Write([]byte(proxyHeader)); err != nil { + t.Fatalf("write PROXY header: %v", err) + } + + backendAddr := mustParseURL(setup.Backend.URL).Host + fmt.Fprintf(conn, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", backendAddr, backendAddr) + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read CONNECT response: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT status = %d, want 200", resp.StatusCode) + } + + clientCAs := x509.NewCertPool() + clientCAs.AppendCertsFromPEM(setup.CA.CertPEM()) + tlsConn := tls.Client(conn, &tls.Config{RootCAs: clientCAs, ServerName: mustParseURL(setup.Backend.URL).Hostname()}) + if err := tlsConn.Handshake(); err != nil { + t.Fatalf("TLS handshake: %v", err) + } + defer tlsConn.Close() + + fmt.Fprintf(tlsConn, "GET /api/data HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n\r\n", backendAddr) + innerResp, err := http.ReadResponse(bufio.NewReader(tlsConn), nil) + if err != nil { + t.Fatalf("read inner response: %v", err) + } + io.ReadAll(innerResp.Body) + innerResp.Body.Close() + + var entries []RequestLogData + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + entries = cap.snapshot() + if len(entries) >= 1 { + break + } + time.Sleep(20 * time.Millisecond) + } + if len(entries) != 1 { + t.Fatalf("got %d log entries, want exactly 1", len(entries)) + } + host, _, err := net.SplitHostPort(entries[0].ClientAddr) + if err != nil { + t.Fatalf("ClientAddr = %q: SplitHostPort: %v", entries[0].ClientAddr, err) + } + if host != "100.52.56.181" { + t.Errorf("ClientAddr host = %q, want 100.52.56.181 (PROXY-header source)", host) + } + }) + + t.Run("postgres", func(t *testing.T) { + cap := &logCapture{} + setup.Proxy.SetLogger(cap.log) + + raw, err := net.Dial("tcp", setup.Addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + if _, err := raw.Write([]byte(proxyHeader)); err != nil { + raw.Close() + t.Fatalf("write PROXY header: %v", err) + } + + msg, conn := pgClientHandshakeOnConn(t, raw, "ep-foo-123.aws.neon.tech", caTrustPool(t, setup.CA), "app_rw", "appdb", "run-token") + if _, ok := msg.(*pgproto3.AuthenticationOk); !ok { + conn.Close() + t.Fatalf("expected AuthenticationOk, got %T", msg) + } + // The audit log entry is written when the relay completes, after the + // client disconnects — close now so the log-wait loop below doesn't + // race a still-open, still-relaying connection. + conn.Close() + + var entries []RequestLogData + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + entries = cap.snapshot() + if len(entries) >= 1 { + break + } + time.Sleep(20 * time.Millisecond) + } + if len(entries) != 1 { + t.Fatalf("got %d log entries, want exactly 1", len(entries)) + } + host, _, err := net.SplitHostPort(entries[0].ClientAddr) + if err != nil { + t.Fatalf("ClientAddr = %q: SplitHostPort: %v", entries[0].ClientAddr, err) + } + if host != "100.52.56.181" { + t.Errorf("ClientAddr host = %q, want 100.52.56.181 (PROXY-header source)", host) + } + }) +} From 95357604aa32225560a82f6b6f76c6b732f2e9d0 Mon Sep 17 00:00:00 2001 From: Andy Bonventre Date: Wed, 15 Jul 2026 18:49:33 -0400 Subject: [PATCH 2/2] fix(proxy): retry transient Accept errors in the demux loop instead of exiting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Demux.acceptLoop is the sole caller of Accept on the real socket in multiplex mode — http.Server.Serve only ever sees the virtual listener, which never surfaces an OS-level error, so http.Server's own accept-retry loop can neither see nor recover from a transient failure on the real socket. The loop previously returned on ANY Accept error, so one transient error (EMFILE/ENFILE under fd exhaustion, ECONNABORTED — realistic for a proxy holding many long-lived CONNECT tunnels and Postgres relays) permanently killed accept for BOTH planes until process restart. Before multiplexing, http.Server.Serve tolerated exactly these errors, so this silently removed the HTTP plane's prior resilience. Mirror net/http.Server.Serve's accept-error handling: exit cleanly only when the demux is shutting down (Close/StopAccepting set the closed flag before closing the listener), otherwise back off with a capped exponential delay (5ms doubling to a 1s cap, logged at WARN) and retry while the listener is live, resetting the delay after a successful Accept. Unlike net/http, the retry is not gated on the deprecated, unreliable net.Error.Temporary(): any error while the listener is open is retried; a genuinely dead-but-unclosed listener retries once per second forever with a WARN each time — the same visible, capped pathological case net/http tolerates, not a zero-delay spin. PostgresServer.acceptLoop (proxy/postgres.go) has the same unconditional-exit pattern, but its blast radius is unchanged by this PR (non-multiplex Postgres plane only), so it is left out of scope here. --- CHANGELOG.md | 1 + proxy/demux.go | 54 +++++++++++++++- proxy/demux_test.go | 148 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ddef53d..4628ccd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ Gatekeeper is pre-1.0. The configuration schema and credential source interface - **The HTTP/CONNECT proxy and the Postgres data-plane listener can now share a single port** — an operator putting gatekeeper behind a load balancer previously needed one backend service, one health check, and one firewall rule per listener, even when the two planes were fronted by the same LB. Standing up two of everything for what is conceptually one gatekeeper endpoint was pure ceremony; this removes it. The trigger is topology, not a new flag: when `postgres.port` equals `proxy.port` (and their hosts, after defaults, also match), gatekeeper multiplexes both planes onto one real listener instead of binding two — the port equality itself *is* the declaration, there is no `multiplex: true` to set. Distinct ports (the default, and every existing config) leave the two-listener path byte-for-byte unchanged; `postgres` being absent never engages the shared-listener path either. Mechanism: a new `proxy.Demux` (`proxy/demux.go`) owns the one real listener — wrapped with `WrapProxyProtocolListener` first when `proxy.proxy_protocol` is set, exactly as today — and reads the first 8 bytes of each accepted connection *in a per-connection goroutine, never in the shared accept loop*, so a silent or slow client (an LB health check that opens a socket and waits, or a slow-loris) can never stall `Accept` for every other pending connection — the same discipline `WrapProxyProtocolListener`'s own `Accept` already follows, and a prior review finding this change deliberately doesn't regress. Those 8 bytes positively identify a Postgres startup signature — `SSLRequest` (length `00 00 00 08`, code `04 d2 16 2f`), `GSSENCRequest` (same length, code `04 d2 16 30`), or a v3 `StartupMessage` (any 4-byte length, then protocol `00 03 00 00`) — and default everything else (HTTP methods, the h2 client preface, or any other prefix) to the HTTP plane, the same positive-match-Postgres approach as Caddy-L4's postgres matcher. The classified connection is wrapped so those 8 bytes are replayed on its first `Read` calls before any further bytes from the underlying socket — the same hold-then-replay pattern `proxyProtoLogConn` already uses for the PROXY header — then pushed onto one of two in-memory virtual listeners (a bounded, non-blocking channel of conns implementing `net.Listener`). `http.Server.Serve` and `PostgresServer.StartListener` run completely unmodified against those virtual listeners, unaware they aren't backed by a real socket. Once classification succeeds, the sniff deadline (10s, matching `WrapProxyProtocolListener`'s own PROXY-header-read timeout) is cleared before handoff, so only the downstream server's own timeout — `http.Server`'s `ReadHeaderTimeout`, or `PostgresServer`'s handshake timeout — ever applies from that point on. A connection that times out, closes early, or otherwise sends fewer than 8 bytes is dropped with a DEBUG log line naming the failure, never the bytes themselves; a successfully classified connection logs a second DEBUG line naming only `http` or `postgres`. Because one physical listener means one PROXY protocol setting, gatekeeper now validates listener topology at startup (`resolveListenTopology`, `config.go`) before either listener binds: equal ports with `proxy.proxy_protocol != postgres.proxy_protocol` is a fatal, descriptive config error (`proxy.proxy_protocol` is the setting actually applied to the shared listener), as is equal ports with different hosts (there's only one socket to bind, and gatekeeper won't guess which address you meant). Port `0` on both sides — "give me any available port," gatekeeper's own default — deliberately never triggers multiplexing on its own, even though the values compare equal; two independent ephemeral-port binds aren't "the same port," and treating them as such would have silently changed the two-listener behavior of every config (and most of this repo's own test suite) that leaves the port unset. `gatekeeper.Server` gains a `demux *proxy.Demux` field; `Stop` calls `Demux.StopAccepting` (not `Close`) to close only the real listener immediately, deliberately leaving both virtual listeners open for `PostgresServer.Shutdown`/`http.Server.Shutdown` to close as part of their own graceful drain — closing them out from under those servers first would race their own closed-flag bookkeeping and, for `PostgresServer`, surface as a spurious error log on an entirely ordinary shutdown. + The demux accept loop retries transient `Accept` errors instead of exiting on them, mirroring `net/http.Server.Serve`: on an error it exits cleanly only when the demux is shutting down (its own `Close`/`StopAccepting` set the closed flag before closing the listener), and otherwise backs off with a capped exponential delay (5ms doubling to a 1s cap, logged at WARN) and retries while the listener is live, resetting the delay after a successful `Accept`. This matters because in multiplex mode the demux loop is the *sole* caller of `Accept` on the real socket — `http.Server` only ever sees the virtual listener, which never surfaces an OS-level error — so exiting on a transient failure (EMFILE/ENFILE under fd exhaustion, ECONNABORTED, realistic for a proxy holding many long-lived tunnels and relays) would have killed accept for *both* planes until process restart, silently removing the resilience `http.Server.Serve`'s own accept-retry loop gave the HTTP plane before it was multiplexed. Unlike `net/http` it does not gate the retry on the deprecated, unreliable `net.Error.Temporary()`; the trade is that a genuinely dead-but-unclosed listener retries once per second forever with a WARN each time — the same visible, capped pathological case `net/http` tolerates, not a zero-delay spin. Documented in the config reference (["Sharing one listener with the HTTP proxy"](docs/content/reference/02-config-file.md#sharing-one-listener-with-the-http-proxy)) and the load-balancer guide (["Single load balancer, one shared port"](docs/content/guides/11-load-balancer-proxy-protocol.md#single-load-balancer-one-shared-port), including health-check behavior on the shared listener: an HTTP health check classifies as HTTP traffic and reaches `/healthz` normally, and a bare TCP connect-then-close probe is simply dropped, cleanly and silently, once the classification read times out) ## v0.18.0 — 2026-07-15 diff --git a/proxy/demux.go b/proxy/demux.go index 279d6cc..6b7f546 100644 --- a/proxy/demux.go +++ b/proxy/demux.go @@ -55,6 +55,15 @@ const demuxSniffDeadline = 10 * time.Second // that's classifying other connections. const demuxBacklog = 64 +// demuxAcceptRetryBaseDelay and demuxAcceptRetryMaxDelay bound the +// exponential backoff acceptLoop applies after a transient Accept error, +// mirroring net/http.Server.Serve's own accept-retry delays (5ms doubling +// to a 1s cap). See acceptLoop. +const ( + demuxAcceptRetryBaseDelay = 5 * time.Millisecond + demuxAcceptRetryMaxDelay = 1 * time.Second +) + // demuxProtocol identifies which plane a connection belongs to. type demuxProtocol int @@ -275,15 +284,54 @@ func (d *Demux) StopAccepting() error { return d.ln.Close() } +// acceptLoop accepts connections on the real listener and dispatches each to +// classification. It mirrors net/http.Server.Serve's accept-error handling: +// a transient Accept error is retried after a capped exponential backoff +// rather than tearing down the listener, and only an intentional shutdown +// (d.closed set) exits the loop. +// +// This resilience is load-bearing precisely because acceptLoop is the SOLE +// caller of Accept on the real socket in multiplex mode: http.Server.Serve +// runs against the virtual listener (see virtualListener), which never +// surfaces an OS-level Accept error, so http.Server's own accept-retry loop +// can no longer see — let alone recover from — a transient failure on the +// real socket. Returning here on the first transient error (EMFILE/ENFILE +// under fd exhaustion, ECONNABORTED — realistic for a proxy holding many +// long-lived CONNECT tunnels and Postgres relays) would kill accept for BOTH +// planes until process restart, silently dropping the resilience the HTTP +// plane had before it was multiplexed. +// +// Unlike net/http, this does not gate the retry on the deprecated and +// unreliable net.Error.Temporary(): any error while the listener is still +// live is treated as transient and retried. On a genuinely dead-but-unclosed +// listener that means retrying once per second forever, each attempt logged +// at WARN — the same acceptable, visible pathological case net/http tolerates +// via its own capped backoff, not a tight zero-delay spin. func (d *Demux) acceptLoop() { + var backoff time.Duration for { conn, err := d.ln.Accept() if err != nil { - if !d.closed.Load() { - slog.Error("demux accept loop exited", "subsystem", "proxy", "error", err) + if d.closed.Load() { + // Intentional shutdown: StopAccepting sets closed before + // closing the listener, so this error is our own doing. Exit + // cleanly and silently. + return + } + if backoff == 0 { + backoff = demuxAcceptRetryBaseDelay + } else { + backoff *= 2 + } + if backoff > demuxAcceptRetryMaxDelay { + backoff = demuxAcceptRetryMaxDelay } - return + slog.Warn("demux: transient accept error; retrying", + "subsystem", "proxy", "error", err, "retry_in", backoff) + time.Sleep(backoff) + continue } + backoff = 0 go d.classifyAndDispatch(conn) } } diff --git a/proxy/demux_test.go b/proxy/demux_test.go index 6db92bb..edc40ad 100644 --- a/proxy/demux_test.go +++ b/proxy/demux_test.go @@ -12,6 +12,7 @@ import ( "context" "crypto/tls" "crypto/x509" + "errors" "fmt" "io" "net" @@ -598,6 +599,153 @@ func TestDemux_StopAcceptingLeavesVirtualListenersOpen(t *testing.T) { d.PostgresListener().Close() } +// --- accept-loop resilience to transient Accept errors -------------------- + +// scriptedAcceptListener is a fake net.Listener whose Accept returns a +// transient error its first failN times, then hands out conns from a +// channel, and otherwise blocks until Close. It records how many times +// Accept is called after Close so a test can prove the demux accept loop +// exits cleanly on shutdown (exactly one post-close Accept) rather than +// spin-retrying the closed-listener error. +type scriptedAcceptListener struct { + mu sync.Mutex + failsRemaining int + transientErr error + acceptsAfterClose int + + conns chan net.Conn + closed chan struct{} + once sync.Once +} + +func newScriptedAcceptListener(failN int, transientErr error) *scriptedAcceptListener { + return &scriptedAcceptListener{ + failsRemaining: failN, + transientErr: transientErr, + conns: make(chan net.Conn, 1), + closed: make(chan struct{}), + } +} + +func (l *scriptedAcceptListener) Accept() (net.Conn, error) { + l.mu.Lock() + select { + case <-l.closed: + l.acceptsAfterClose++ + l.mu.Unlock() + return nil, net.ErrClosed + default: + } + if l.failsRemaining > 0 { + l.failsRemaining-- + l.mu.Unlock() + return nil, l.transientErr + } + l.mu.Unlock() + + select { + case c := <-l.conns: + return c, nil + case <-l.closed: + l.mu.Lock() + l.acceptsAfterClose++ + l.mu.Unlock() + return nil, net.ErrClosed + } +} + +func (l *scriptedAcceptListener) Close() error { + l.once.Do(func() { close(l.closed) }) + return nil +} + +func (l *scriptedAcceptListener) Addr() net.Addr { return &net.TCPAddr{} } + +func (l *scriptedAcceptListener) acceptsAfterCloseCount() int { + l.mu.Lock() + defer l.mu.Unlock() + return l.acceptsAfterClose +} + +// TestDemux_AcceptLoopRetriesTransientErrors is the regression guard for the +// availability bug PR #56's review surfaced: Demux.acceptLoop is the SOLE +// caller of Accept on the real socket in multiplex mode (http.Server only +// ever sees the virtual listener, which never surfaces OS-level errors), so +// if the loop exits on the first transient Accept error — EMFILE/ENFILE +// under fd exhaustion, ECONNABORTED — accept dies for BOTH planes until +// process restart, silently removing the resilience http.Server.Serve's own +// accept-retry loop gave the HTTP plane before it was multiplexed. The loop +// must instead back off and retry (mirroring net/http.Server.Serve) while +// the listener is live, so a good connection arriving after a burst of +// transient errors is still dispatched. +func TestDemux_AcceptLoopRetriesTransientErrors(t *testing.T) { + // Three transient failures, then a real, HTTP-classifiable connection. + transient := errors.New("simulated EMFILE: too many open files") + l := newScriptedAcceptListener(3, transient) + + serverConn, clientConn := net.Pipe() + // Exactly demuxSniffLen bytes so sniffProtocol's ReadFull completes and + // the pipe writer isn't left blocked: "GET /xxx" classifies as HTTP. + go func() { _, _ = clientConn.Write([]byte("GET /xxx")) }() + l.conns <- serverConn + + d := NewDemux(l) + t.Cleanup(func() { d.Close() }) + + // If the loop had exited after the first transient error (the bug), the + // good conn is never Accepted from the scripted listener and never + // dispatched, so this Accept times out. The 2s bound is far above the + // ~35ms the three 5ms/10ms/20ms backoffs take. + accepted := acceptWithTimeout(t, d.HTTPListener(), 2*time.Second) + defer accepted.Close() + + // The dispatched conn replays the sniffed prefix: prove it's the good + // connection that survived the transient-error burst. + buf := make([]byte, len("GET /xxx")) + if _, err := io.ReadFull(accepted, buf); err != nil { + t.Fatalf("read replayed bytes: %v", err) + } + if string(buf) != "GET /xxx" { + t.Errorf("replayed bytes = %q, want %q", buf, "GET /xxx") + } +} + +// TestDemux_AcceptLoopExitsCleanlyOnClose guards the other half of the fix: +// the retry path must not swallow shutdown. When the demux's own close sets +// closed before closing the real listener, the resulting Accept error is +// gatekeeper's intentional shutdown, not a transient failure — the loop must +// return immediately without logging and without spin-retrying the +// closed-listener error. +func TestDemux_AcceptLoopExitsCleanlyOnClose(t *testing.T) { + logBuf := captureSlogText(t) + l := newScriptedAcceptListener(0, nil) + d := NewDemux(l) + + // Let the accept loop reach its blocking Accept before shutting down. + time.Sleep(20 * time.Millisecond) + + if err := d.StopAccepting(); err != nil { + t.Fatalf("StopAccepting: %v", err) + } + + // Give the loop time to observe the close and return. A correct loop + // calls Accept exactly once more (getting net.ErrClosed), sees closed, + // and returns; a loop that treated the closed-listener error as transient + // would keep calling Accept on a ~5ms backoff, so the post-close count + // would climb past 1 within this window. + time.Sleep(80 * time.Millisecond) + + if got := l.acceptsAfterCloseCount(); got != 1 { + t.Errorf("Accept called %d times after close, want exactly 1: >1 means the loop spin-retried the closed-listener error instead of exiting on shutdown", got) + } + if s := logBuf.String(); strings.Contains(s, "transient accept error") || strings.Contains(s, "accept loop exited") { + t.Errorf("clean shutdown logged an accept error/retry line, want none: %q", s) + } + + d.HTTPListener().Close() + d.PostgresListener().Close() +} + // --- end-to-end: both planes share one port ------------------------------- // newDemuxE2ESetup wires an http.Server and a PostgresServer onto one shared