From d28e29e49c1b3eb3d114b94247cc5bcb58b98fe7 Mon Sep 17 00:00:00 2001
From: Manas Srivastava <[email protected]>
Date: Wed, 12 Aug 2026 23:36:53 +0530
Subject: [PATCH] fix(provision): authenticate to password-protected shared
 Redis + stop leaking cluster DNS into mongo/nats customer URLs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two customer-facing breakages found by live-testing the Azure cluster.

BUG 1 — /cache/new 503s on every call.
redis-provision now runs with --requirepass, but the local backend's admin
client was built as goredis.Options{Addr: REDIS_PROVISION_HOST} — no password
field anywhere in the path — so ACL SETUSER failed with "ERR Protocol error:
unauthenticated multibulk length" and Provision correctly fell closed rather
than handing out a credential-less shared URL.

New env REDIS_PROVISION_URL ("redis://[user]:password@host:port[/db]", parsed
with goredis.ParseURL) supplies the credentials. Unset → the existing
REDIS_PROVISION_HOST Addr form, byte-for-byte today's behaviour. Malformed →
logged (with the userinfo redacted, since net/url echoes the whole URL back in
its error) and the same fallback, because the factory chain has no error
channel and refusing to boot over one typo is worse than the 503.

Threaded as a separate constructor argument rather than read from env inside
the backend, mirroring mongo.NewBackend(backendType, adminURI, mongoHost):
credentialed admin endpoint and customer-facing host stay distinct inputs, and
the precedence rule lives in exactly one function. The backend records
opts.Addr as its host, never the admin URL — otherwise the admin password would
be interpolated into customer connection strings.

BUG 2 — /nosql/new and /queue/new emitted internal cluster DNS
(mongodb://…@mongodb.instant-data.svc.cluster.local:27017/…,
nats://nats.instant-data.svc.cluster.local:4222), unreachable for customers.
The public host was applied only inside the `case "k8s"` branch of each
NewBackend, and the cluster runs MONGO_PROVISION_BACKEND=local /
QUEUE_PROVISION_BACKEND=local.

Both now use the postgres local backend's mechanism verbatim: a buildXURL
helper (mirroring postgres.buildDBURL) over an env-resolved publicHostPort()
that falls back to the in-cluster admin host — never to an empty host. The
resolution order adds K8S_MONGO_PUBLIC_HOST / K8S_NATS_PUBLIC_HOST as a source,
so the already-configured cluster env fixes prod with no ops change, and takes
no built-in default, so a dev box keeps emitting localhost.

Call sites (rg -F, all updated): redis.NewBackend x2 prod (server.New,
pool.NewWithConfig) + 1 test; redis.NewSharedCarveBackend x1 prod + 1 test;
newLocalBackend x3 in the redis factory + 8 in tests. Customer-URL emitters:
one per shared backend (mongo.go Provision, queue/local.go Provision); the k8s
emitters were already correct.

Tests: table-driven over REDIS_PROVISION_URL set/unset/malformed and mongo+nats
public host set/unset, plus registry-iterating guards that every shared-carve
constructor honours the admin URL (rule 18) and that the admin password never
reaches a log line or a customer URL. New funcs are at 100% statement coverage;
make gate green.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 README.md                                     |   4 +
 internal/backend/mongo/mongo.go               |  65 +++-
 internal/backend/mongo/public_host_test.go    | 183 ++++++++++
 internal/backend/queue/local.go               |  68 +++-
 internal/backend/queue/public_host_test.go    | 198 +++++++++++
 internal/backend/redis/admin_url_test.go      | 328 ++++++++++++++++++
 internal/backend/redis/backend.go             |  17 +-
 internal/backend/redis/coverage_test.go       |  28 +-
 internal/backend/redis/dispatch_test.go       |   2 +-
 internal/backend/redis/local.go               |  65 +++-
 internal/config/config.go                     |  14 +
 internal/config/config_test.go                |   5 +
 internal/pool/factory.go                      |   1 +
 internal/server/server.go                     |   4 +-
 internal/server/server_live_roundtrip_test.go |   4 +-
 15 files changed, 955 insertions(+), 31 deletions(-)
 create mode 100644 internal/backend/mongo/public_host_test.go
 create mode 100644 internal/backend/queue/public_host_test.go
 create mode 100644 internal/backend/redis/admin_url_test.go

diff --git a/README.md b/README.md
index dfa1519..90d6f22 100644
--- a/README.md
+++ b/README.md
@@ -67,6 +67,10 @@ The test suite uses mock backends for almost all tests, so `go test ./...` passe
 | `CUSTOMER_REDIS_URL` | Admin URL for redis-provision | `redis://redis-provision.instant-data.svc.cluster.local:6379` |
 | `CUSTOMER_MONGO_URL` | Admin URL for mongodb | `mongodb://root:root@mongodb.instant-data.svc.cluster.local:27017` |
 | `POSTGRES_CLUSTER_URLS` | Comma-separated list of admin DSNs (multi-cluster) | unset |
+| `REDIS_PROVISION_URL` | **Credentialed** admin URL for the shared redis-provision pool: `redis://[user]:password@host:port[/db]`. Required when the pool runs with `--requirepass` — without it `ACL SETUSER` fails and `/cache/new` 503s. Supersedes `REDIS_PROVISION_HOST`; a malformed value logs an error and falls back to it | unset |
+| `REDIS_PROVISION_HOST` | Bare `host:port` admin address for the shared Redis pool. Sends no AUTH — legacy / unauthenticated pools only | `localhost:6379` |
+| `MONGO_PUBLIC_HOST_PORT`, `MONGO_PUBLIC_HOST` (+ `MONGO_PUBLIC_PORT`) | Customer-facing host embedded in `/nosql/new` URLs on the shared backend. Falls back to `K8S_MONGO_PUBLIC_HOST`, then to the in-cluster admin host | unset (port `27017`) |
+| `NATS_PUBLIC_HOST_PORT`, `NATS_PUBLIC_HOST` (+ `NATS_PUBLIC_PORT`) | Customer-facing host embedded in `/queue/new` URLs on the shared backend. Falls back to `K8S_NATS_PUBLIC_HOST`, then to the in-cluster admin host | unset (port `4222`) |
 | `K8S_DEDICATED_BACKEND` | Enable k8s dedicated-pod backend for team / growth tier | `false` |
 | `K8S_EXTERNAL_HOST` | External hostname for dedicated k8s services | unset |
 | `K8S_STORAGE_CLASS` | Storage class for dedicated PVCs | `local-path` |
diff --git a/internal/backend/mongo/mongo.go b/internal/backend/mongo/mongo.go
index cd4f61d..d78c721 100644
--- a/internal/backend/mongo/mongo.go
+++ b/internal/backend/mongo/mongo.go
@@ -14,6 +14,7 @@ import (
 	"encoding/hex"
 	"fmt"
 	"log/slog"
+	"os"
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
@@ -28,6 +29,68 @@ import (
 // Short to fail-fast in tests and when MongoDB is not reachable.
 const connectTimeout = 3 * time.Second
 
+// defaultMongoPort is appended to a public hostname that carries no port of its
+// own. 27017 is the MongoDB wire default and what the mongo-proxy listens on —
+// the same port the k8s backend hardcodes when building customer URLs (k8s.go).
+const defaultMongoPort = "27017"
+
+// buildMongoURL constructs the user-facing connection URL for a provisioned
+// database. Mirrors postgres.buildDBURL (backend/postgres/local.go): the public
+// host wins when configured, otherwise clusterHost — the in-cluster admin
+// address, which is only resolvable from inside the cluster.
+//
+// This is the fix for the leak of internal cluster DNS into customer
+// connection strings: before it, /nosql/new handed out
+// mongodb://…@mongodb.instant-data.svc.cluster.local:27017/… on the shared
+// backend, because the public host was applied only in the "k8s" branch of
+// NewBackend and the cluster runs MONGO_PROVISION_BACKEND=local.
+func buildMongoURL(clusterHost, username, password, dbName string) string {
+	host := publicHostPort()
+	if host == "" {
+		host = clusterHost
+	}
+	return fmt.Sprintf("mongodb://%s:%s@%s/%s?authSource=admin", username, password, host, dbName)
+}
+
+// publicHostPort returns the host:port to embed in user-facing MongoDB URLs, or
+// "" when no public host is configured (the caller then falls back to the
+// cluster-internal mongoHost).
+//
+// Identical mechanism to postgres.publicHostPort (backend/postgres/local.go) and
+// redis.publicHostPort (backend/redis/local.go) — env-resolved at Provision
+// time, never a constructor argument, so the shared/local backend and the
+// dedicated k8s backend agree on the customer-facing hostname.
+//
+// Resolution order:
+//  1. MONGO_PUBLIC_HOST_PORT (e.g. "mongo.instanode.dev:27017")
+//  2. MONGO_PUBLIC_HOST + MONGO_PUBLIC_PORT (port defaults to 27017)
+//  3. K8S_MONGO_PUBLIC_HOST + MONGO_PUBLIC_PORT — the env the k8s branch of
+//     NewBackend already reads. Honouring it here is what makes the fix a pure
+//     code change: a cluster that already advertises mongo.instanode.dev for
+//     dedicated pods now advertises it for shared ones too, no ops change.
+//  4. "" — caller falls back to the in-cluster mongoHost.
+//
+// Deliberately NO built-in default (the k8s branch defaults to
+// "mongo.instanode.dev"): a dev box running the shared backend against
+// localhost:27017 must keep emitting localhost, not a production hostname.
+func publicHostPort() string {
+	if hp := os.Getenv("MONGO_PUBLIC_HOST_PORT"); hp != "" {
+		return hp
+	}
+	host := os.Getenv("MONGO_PUBLIC_HOST")
+	if host == "" {
+		host = os.Getenv("K8S_MONGO_PUBLIC_HOST")
+	}
+	if host == "" {
+		return ""
+	}
+	port := os.Getenv("MONGO_PUBLIC_PORT")
+	if port == "" {
+		port = defaultMongoPort
+	}
+	return host + ":" + port
+}
+
 // decodeStorageSize extracts the dbStats storageSize from a decoded result,
 // tolerating every BSON numeric encoding the server may use across versions
 // (int32 / int64 / float64). Any missing or non-numeric value yields 0 — the
@@ -140,7 +203,7 @@ func (b *LocalBackend) Provision(ctx context.Context, token, tier string) (*Cred
 	}
 
 	// User is created in the admin database; include authSource so clients authenticate correctly.
-	url := fmt.Sprintf("mongodb://%s:%s@%s/%s?authSource=admin", username, password, b.mongoHost, dbName)
+	url := buildMongoURL(b.mongoHost, username, password, dbName)
 	slog.Info("nosql.Provision: provisioned",
 		"token", token,
 		"db", dbName,
diff --git a/internal/backend/mongo/public_host_test.go b/internal/backend/mongo/public_host_test.go
new file mode 100644
index 0000000..8445684
--- /dev/null
+++ b/internal/backend/mongo/public_host_test.go
@@ -0,0 +1,183 @@
+package mongo
+
+// public_host_test.go — the customer-facing hostname in /nosql/new connection
+// strings.
+//
+// COVERAGE BLOCK (CLAUDE.md rule 17):
+//
+//	Symptom:        /nosql/new returned
+//	                mongodb://usr_…:…@mongodb.instant-data.svc.cluster.local:27017/…
+//	                — internal cluster DNS no customer can resolve. The public
+//	                host was applied only inside the `case "k8s"` branch of
+//	                NewBackend (backend.go), and the cluster runs
+//	                MONGO_PROVISION_BACKEND=local.
+//	Enumeration:    rg -F 'mongodb://' / 'b.mongoHost' / 'K8S_MONGO_PUBLIC_HOST'
+//	Sites found:    1 customer-URL emitter on the shared path
+//	                (mongo.go Provision) + 1 on the k8s path (k8s.go, already
+//	                correct) + admin URIs (not customer-facing).
+//	Sites touched:  the shared emitter, via buildMongoURL — the same
+//	                helper+publicHostPort shape as postgres.buildDBURL.
+//	Coverage test:  TestBuildMongoURL below; the unset row pins the fallback to
+//	                the in-cluster host (never an empty host).
+
+import (
+	"strings"
+	"testing"
+)
+
+// mongoPublicHostEnvKeys is every env var publicHostPort consults. Tests clear
+// all of them so a developer's ambient shell env cannot perturb the "unset"
+// rows. A new source added to publicHostPort must be added here.
+var mongoPublicHostEnvKeys = []string{
+	"MONGO_PUBLIC_HOST_PORT",
+	"MONGO_PUBLIC_HOST",
+	"MONGO_PUBLIC_PORT",
+	"K8S_MONGO_PUBLIC_HOST",
+}
+
+func clearMongoPublicHostEnv(t *testing.T) {
+	t.Helper()
+	for _, k := range mongoPublicHostEnvKeys {
+		t.Setenv(k, "")
+	}
+}
+
+// TestPublicHostPort_Mongo exercises every resolution branch of the helper.
+func TestPublicHostPort_Mongo(t *testing.T) {
+	tests := []struct {
+		name string
+		env  map[string]string
+		want string
+	}{
+		{
+			name: "nothing set — empty so the caller falls back to the admin host",
+			env:  map[string]string{},
+			want: "",
+		},
+		{
+			name: "MONGO_PUBLIC_HOST_PORT wins over everything",
+			env: map[string]string{
+				"MONGO_PUBLIC_HOST_PORT": "mongo.instanode.dev:27020",
+				"MONGO_PUBLIC_HOST":      "ignored.example.com",
+				"MONGO_PUBLIC_PORT":      "1111",
+				"K8S_MONGO_PUBLIC_HOST":  "also-ignored.example.com",
+			},
+			want: "mongo.instanode.dev:27020",
+		},
+		{
+			name: "MONGO_PUBLIC_HOST with the default port",
+			env:  map[string]string{"MONGO_PUBLIC_HOST": "mongo.instanode.dev"},
+			want: "mongo.instanode.dev:" + defaultMongoPort,
+		},
+		{
+			name: "MONGO_PUBLIC_HOST with an explicit port",
+			env: map[string]string{
+				"MONGO_PUBLIC_HOST": "mongo.instanode.dev",
+				"MONGO_PUBLIC_PORT": "27099",
+			},
+			want: "mongo.instanode.dev:27099",
+		},
+		{
+			name: "MONGO_PUBLIC_HOST wins over K8S_MONGO_PUBLIC_HOST",
+			env: map[string]string{
+				"MONGO_PUBLIC_HOST":     "explicit.example.com",
+				"K8S_MONGO_PUBLIC_HOST": "k8s.example.com",
+			},
+			want: "explicit.example.com:" + defaultMongoPort,
+		},
+		{
+			// The env the cluster ALREADY sets. Honouring it is what makes the
+			// fix a pure code change with no ops change.
+			name: "K8S_MONGO_PUBLIC_HOST alone — the already-configured prod env",
+			env:  map[string]string{"K8S_MONGO_PUBLIC_HOST": "mongo.instanode.dev"},
+			want: "mongo.instanode.dev:" + defaultMongoPort,
+		},
+		{
+			name: "K8S_MONGO_PUBLIC_HOST with an explicit port",
+			env: map[string]string{
+				"K8S_MONGO_PUBLIC_HOST": "mongo.instanode.dev",
+				"MONGO_PUBLIC_PORT":     "27098",
+			},
+			want: "mongo.instanode.dev:27098",
+		},
+		{
+			name: "port set but no host — still empty (a port alone addresses nothing)",
+			env:  map[string]string{"MONGO_PUBLIC_PORT": "27099"},
+			want: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			clearMongoPublicHostEnv(t)
+			for k, v := range tc.env {
+				t.Setenv(k, v)
+			}
+			if got := publicHostPort(); got != tc.want {
+				t.Errorf("publicHostPort() = %q; want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestBuildMongoURL asserts the customer URL uses the public host when one is
+// configured and the in-cluster admin host otherwise — never an empty host.
+func TestBuildMongoURL(t *testing.T) {
+	const (
+		clusterHost = "mongodb.instant-data.svc.cluster.local:27017"
+		user        = "usr_abc"
+		pass        = "pw123"
+		db          = "db_abc"
+	)
+
+	tests := []struct {
+		name string
+		env  map[string]string
+		want string
+	}{
+		{
+			name: "public host unset — falls back to the cluster host, NOT an empty host",
+			env:  map[string]string{},
+			want: "mongodb://usr_abc:pw123@" + clusterHost + "/db_abc?authSource=admin",
+		},
+		{
+			name: "public host set via K8S_MONGO_PUBLIC_HOST (prod today)",
+			env:  map[string]string{"K8S_MONGO_PUBLIC_HOST": "mongo.instanode.dev"},
+			want: "mongodb://usr_abc:pw123@mongo.instanode.dev:27017/db_abc?authSource=admin",
+		},
+		{
+			name: "public host set via MONGO_PUBLIC_HOST_PORT",
+			env:  map[string]string{"MONGO_PUBLIC_HOST_PORT": "mongo.instanode.dev:27020"},
+			want: "mongodb://usr_abc:pw123@mongo.instanode.dev:27020/db_abc?authSource=admin",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			clearMongoPublicHostEnv(t)
+			for k, v := range tc.env {
+				t.Setenv(k, v)
+			}
+			got := buildMongoURL(clusterHost, user, pass, db)
+			if got != tc.want {
+				t.Errorf("buildMongoURL() = %q; want %q", got, tc.want)
+			}
+			if strings.Contains(got, "@/") {
+				t.Errorf("buildMongoURL() = %q has an empty host", got)
+			}
+		})
+	}
+}
+
+// TestBuildMongoURL_NeverLeaksClusterDNSWhenPublicHostSet is the regression pin:
+// with the public host configured, the internal service DNS must be gone from
+// the customer's connection string entirely.
+func TestBuildMongoURL_NeverLeaksClusterDNSWhenPublicHostSet(t *testing.T) {
+	clearMongoPublicHostEnv(t)
+	t.Setenv("K8S_MONGO_PUBLIC_HOST", "mongo.instanode.dev")
+
+	got := buildMongoURL("mongodb.instant-data.svc.cluster.local:27017", "usr_x", "pw", "db_x")
+	if strings.Contains(got, "svc.cluster.local") {
+		t.Errorf("customer URL still contains internal cluster DNS: %q", got)
+	}
+}
diff --git a/internal/backend/queue/local.go b/internal/backend/queue/local.go
index 2be5853..cbeb775 100644
--- a/internal/backend/queue/local.go
+++ b/internal/backend/queue/local.go
@@ -12,9 +12,72 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"os"
 	"time"
 )
 
+// natsClientPort is the NATS client-protocol port embedded in customer URLs.
+// Matches the port the k8s backend hardcodes in its own customer URLs (k8s.go)
+// and the port the nats-proxy listens on.
+const natsClientPort = "4222"
+
+// buildNATSURL constructs the user-facing NATS URL. Mirrors
+// postgres.buildDBURL / mongo.buildMongoURL: the public host wins when
+// configured, otherwise clusterHost — the in-cluster admin address, which is
+// only resolvable from inside the cluster. clusterHost carries no port (config
+// NATS_HOST is a bare hostname), so the client port is appended.
+//
+// This is the fix for the leak of internal cluster DNS into customer
+// connection strings: before it, /queue/new handed out
+// nats://nats.instant-data.svc.cluster.local:4222 on the shared backend,
+// because the public host was applied only in the "k8s" branch of NewBackend
+// and the cluster runs QUEUE_PROVISION_BACKEND=local.
+func buildNATSURL(clusterHost string) string {
+	host := publicHostPort()
+	if host == "" {
+		host = clusterHost + ":" + natsClientPort
+	}
+	return "nats://" + host
+}
+
+// publicHostPort returns the host:port to embed in user-facing NATS URLs, or ""
+// when no public host is configured (the caller then falls back to the
+// cluster-internal natsHost).
+//
+// Identical mechanism to postgres.publicHostPort (backend/postgres/local.go),
+// redis.publicHostPort (backend/redis/local.go) and mongo.publicHostPort
+// (backend/mongo/mongo.go) — env-resolved at Provision time so the shared/local
+// backend and the dedicated k8s backend agree on the customer-facing hostname.
+//
+// Resolution order:
+//  1. NATS_PUBLIC_HOST_PORT (e.g. "nats.instanode.dev:4222")
+//  2. NATS_PUBLIC_HOST + NATS_PUBLIC_PORT (port defaults to 4222)
+//  3. K8S_NATS_PUBLIC_HOST + NATS_PUBLIC_PORT — the env the k8s branch of
+//     NewBackend already reads, so a cluster that already advertises
+//     nats.instanode.dev for dedicated pods advertises it for shared ones too.
+//  4. "" — caller falls back to the in-cluster natsHost.
+//
+// Deliberately NO built-in default (the k8s branch defaults to
+// "nats.instanode.dev"): a dev box running the shared backend against localhost
+// must keep emitting localhost, not a production hostname.
+func publicHostPort() string {
+	if hp := os.Getenv("NATS_PUBLIC_HOST_PORT"); hp != "" {
+		return hp
+	}
+	host := os.Getenv("NATS_PUBLIC_HOST")
+	if host == "" {
+		host = os.Getenv("K8S_NATS_PUBLIC_HOST")
+	}
+	if host == "" {
+		return ""
+	}
+	port := os.Getenv("NATS_PUBLIC_PORT")
+	if port == "" {
+		port = natsClientPort
+	}
+	return host + ":" + port
+}
+
 // LocalBackend provisions NATS on the shared cluster.
 type LocalBackend struct {
 	natsHost   string
@@ -63,7 +126,10 @@ func (b *LocalBackend) Provision(ctx context.Context, token, tier string) (*Cred
 
 	slog.Info("queue.local.provisioned", "token", token, "subject_prefix", prefix)
 	return &Credentials{
-		URL:           fmt.Sprintf("nats://%s:4222", b.natsHost),
+		// The health check above deliberately keeps using b.natsHost: the
+		// monitor port is cluster-internal. Only the customer-facing URL is
+		// rewritten to the public host.
+		URL:           buildNATSURL(b.natsHost),
 		SubjectPrefix: prefix,
 	}, nil
 }
diff --git a/internal/backend/queue/public_host_test.go b/internal/backend/queue/public_host_test.go
new file mode 100644
index 0000000..86b9474
--- /dev/null
+++ b/internal/backend/queue/public_host_test.go
@@ -0,0 +1,198 @@
+package queue
+
+// public_host_test.go — the customer-facing hostname in /queue/new connection
+// strings.
+//
+// COVERAGE BLOCK (CLAUDE.md rule 17):
+//
+//	Symptom:        /queue/new returned
+//	                nats://nats.instant-data.svc.cluster.local:4222 — internal
+//	                cluster DNS no customer can resolve. The public host was
+//	                applied only inside the `case "k8s"` branch of NewBackend
+//	                (backend.go), and the cluster runs
+//	                QUEUE_PROVISION_BACKEND=local.
+//	Enumeration:    rg -F 'nats://' / 'b.natsHost' / 'K8S_NATS_PUBLIC_HOST'
+//	Sites found:    1 customer-URL emitter on the shared path (local.go
+//	                Provision) + 1 on the k8s path (k8s.go, already correct).
+//	Sites touched:  the shared emitter, via buildNATSURL — the same
+//	                helper+publicHostPort shape as postgres.buildDBURL.
+//	Coverage test:  TestBuildNATSURL below; the unset row pins the fallback to
+//	                the in-cluster host:4222 (never a bare "nats://" or an
+//	                empty host).
+
+import (
+	"strings"
+	"testing"
+)
+
+// natsPublicHostEnvKeys is every env var publicHostPort consults. Tests clear
+// all of them so a developer's ambient shell env cannot perturb the "unset"
+// rows. A new source added to publicHostPort must be added here.
+var natsPublicHostEnvKeys = []string{
+	"NATS_PUBLIC_HOST_PORT",
+	"NATS_PUBLIC_HOST",
+	"NATS_PUBLIC_PORT",
+	"K8S_NATS_PUBLIC_HOST",
+}
+
+func clearNATSPublicHostEnv(t *testing.T) {
+	t.Helper()
+	for _, k := range natsPublicHostEnvKeys {
+		t.Setenv(k, "")
+	}
+}
+
+// TestPublicHostPort_NATS exercises every resolution branch of the helper.
+func TestPublicHostPort_NATS(t *testing.T) {
+	tests := []struct {
+		name string
+		env  map[string]string
+		want string
+	}{
+		{
+			name: "nothing set — empty so the caller falls back to the admin host",
+			env:  map[string]string{},
+			want: "",
+		},
+		{
+			name: "NATS_PUBLIC_HOST_PORT wins over everything",
+			env: map[string]string{
+				"NATS_PUBLIC_HOST_PORT": "nats.instanode.dev:4230",
+				"NATS_PUBLIC_HOST":      "ignored.example.com",
+				"NATS_PUBLIC_PORT":      "1111",
+				"K8S_NATS_PUBLIC_HOST":  "also-ignored.example.com",
+			},
+			want: "nats.instanode.dev:4230",
+		},
+		{
+			name: "NATS_PUBLIC_HOST with the default port",
+			env:  map[string]string{"NATS_PUBLIC_HOST": "nats.instanode.dev"},
+			want: "nats.instanode.dev:" + natsClientPort,
+		},
+		{
+			name: "NATS_PUBLIC_HOST with an explicit port",
+			env: map[string]string{
+				"NATS_PUBLIC_HOST": "nats.instanode.dev",
+				"NATS_PUBLIC_PORT": "4299",
+			},
+			want: "nats.instanode.dev:4299",
+		},
+		{
+			name: "NATS_PUBLIC_HOST wins over K8S_NATS_PUBLIC_HOST",
+			env: map[string]string{
+				"NATS_PUBLIC_HOST":     "explicit.example.com",
+				"K8S_NATS_PUBLIC_HOST": "k8s.example.com",
+			},
+			want: "explicit.example.com:" + natsClientPort,
+		},
+		{
+			// The env the cluster ALREADY sets. Honouring it is what makes the
+			// fix a pure code change with no ops change.
+			name: "K8S_NATS_PUBLIC_HOST alone — the already-configured prod env",
+			env:  map[string]string{"K8S_NATS_PUBLIC_HOST": "nats.instanode.dev"},
+			want: "nats.instanode.dev:" + natsClientPort,
+		},
+		{
+			name: "K8S_NATS_PUBLIC_HOST with an explicit port",
+			env: map[string]string{
+				"K8S_NATS_PUBLIC_HOST": "nats.instanode.dev",
+				"NATS_PUBLIC_PORT":     "4298",
+			},
+			want: "nats.instanode.dev:4298",
+		},
+		{
+			name: "port set but no host — still empty (a port alone addresses nothing)",
+			env:  map[string]string{"NATS_PUBLIC_PORT": "4299"},
+			want: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			clearNATSPublicHostEnv(t)
+			for k, v := range tc.env {
+				t.Setenv(k, v)
+			}
+			if got := publicHostPort(); got != tc.want {
+				t.Errorf("publicHostPort() = %q; want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestBuildNATSURL asserts the customer URL uses the public host when one is
+// configured and the in-cluster admin host otherwise — never an empty host.
+func TestBuildNATSURL(t *testing.T) {
+	const clusterHost = "nats.instant-data.svc.cluster.local"
+
+	tests := []struct {
+		name string
+		env  map[string]string
+		want string
+	}{
+		{
+			name: "public host unset — falls back to clusterHost:4222, NOT an empty host",
+			env:  map[string]string{},
+			want: "nats://" + clusterHost + ":4222",
+		},
+		{
+			name: "public host set via K8S_NATS_PUBLIC_HOST (prod today)",
+			env:  map[string]string{"K8S_NATS_PUBLIC_HOST": "nats.instanode.dev"},
+			want: "nats://nats.instanode.dev:4222",
+		},
+		{
+			name: "public host set via NATS_PUBLIC_HOST_PORT",
+			env:  map[string]string{"NATS_PUBLIC_HOST_PORT": "nats.instanode.dev:4230"},
+			want: "nats://nats.instanode.dev:4230",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			clearNATSPublicHostEnv(t)
+			for k, v := range tc.env {
+				t.Setenv(k, v)
+			}
+			got := buildNATSURL(clusterHost)
+			if got != tc.want {
+				t.Errorf("buildNATSURL() = %q; want %q", got, tc.want)
+			}
+			if got == "nats://" || strings.HasSuffix(got, "://:4222") {
+				t.Errorf("buildNATSURL() = %q has an empty host", got)
+			}
+		})
+	}
+}
+
+// TestBuildNATSURL_NeverLeaksClusterDNSWhenPublicHostSet is the regression pin:
+// with the public host configured, the internal service DNS must be gone from
+// the customer's connection string entirely.
+func TestBuildNATSURL_NeverLeaksClusterDNSWhenPublicHostSet(t *testing.T) {
+	clearNATSPublicHostEnv(t)
+	t.Setenv("K8S_NATS_PUBLIC_HOST", "nats.instanode.dev")
+
+	got := buildNATSURL("nats.instant-data.svc.cluster.local")
+	if strings.Contains(got, "svc.cluster.local") {
+		t.Errorf("customer URL still contains internal cluster DNS: %q", got)
+	}
+}
+
+// TestLocalBackend_Provision_UsesPublicHost closes the loop through Provision
+// itself: the health check must still hit the in-cluster monitor address while
+// the returned URL advertises the public host.
+func TestLocalBackend_Provision_UsesPublicHost(t *testing.T) {
+	clearNATSPublicHostEnv(t)
+	t.Setenv("K8S_NATS_PUBLIC_HOST", "nats.instanode.dev")
+
+	host, port := newHealthTestServer(t, 200)
+	b := newLocalBackend(host)
+	b.monitorPort = port
+
+	creds, err := b.Provision(t.Context(), "abc12345deadbeefcafef00d00112233", "anonymous")
+	if err != nil {
+		t.Fatalf("Provision returned error: %v", err)
+	}
+	if creds.URL != "nats://nats.instanode.dev:4222" {
+		t.Errorf("URL = %q; want nats://nats.instanode.dev:4222", creds.URL)
+	}
+}
diff --git a/internal/backend/redis/admin_url_test.go b/internal/backend/redis/admin_url_test.go
new file mode 100644
index 0000000..4fd51e7
--- /dev/null
+++ b/internal/backend/redis/admin_url_test.go
@@ -0,0 +1,328 @@
+package redis
+
+// admin_url_test.go — REDIS_PROVISION_URL, the credentialed admin connection
+// for the shared redis-provision pool.
+//
+// COVERAGE BLOCK (CLAUDE.md rule 17):
+//
+//	Symptom:        /cache/new 503s with
+//	                "cache.provisionLocal: ACL SETUSER failed on shared
+//	                 multi-tenant Redis … ERR Protocol error: unauthenticated
+//	                 multibulk length"
+//	                because the admin client was built from a bare
+//	                goredis.Options{Addr: REDIS_PROVISION_HOST} and therefore
+//	                never sends AUTH to a pod started with --requirepass.
+//	Enumeration:    rg -F 'newLocalBackend(' / 'NewSharedCarveBackend' /
+//	                'redis.NewBackend' / 'RedisProvisionHost'
+//	Sites found:    3 production constructors of the shared LocalBackend
+//	                (NewBackend default arm, NewBackend k8s-init-failure
+//	                fallback, NewSharedCarveBackend) reached from 3 wiring
+//	                sites (server.New x2, pool.NewWithConfig x1).
+//	Sites touched:  all 3 constructors take adminURL; all 3 wiring sites pass
+//	                cfg.RedisProvisionURL.
+//	Coverage test:  TestSharedBackendConstructors_AllHonourAdminURL below —
+//	                iterates every constructor that yields a shared
+//	                LocalBackend and asserts each one authenticates. A 4th
+//	                constructor added without the adminURL parameter fails it.
+
+import (
+	"strings"
+	"testing"
+
+	goredis "github.com/redis/go-redis/v9"
+)
+
+// TestNewLocalBackend_AdminURL is the table for the three states of
+// REDIS_PROVISION_URL: set, unset, malformed.
+func TestNewLocalBackend_AdminURL(t *testing.T) {
+	tests := []struct {
+		name         string
+		adminURL     string
+		redisHost    string
+		wantAddr     string
+		wantUsername string
+		wantPassword string
+		wantDB       int
+	}{
+		{
+			name:         "url set with user and password — credentials reach the client",
+			adminURL:     "redis://admin:s3cret@redis-provision.instant-data.svc.cluster.local:6379/0",
+			redisHost:    "ignored-when-url-set:6379",
+			wantAddr:     "redis-provision.instant-data.svc.cluster.local:6379",
+			wantUsername: "admin",
+			wantPassword: "s3cret",
+			wantDB:       0,
+		},
+		{
+			name: "url set with password only — the --requirepass shape " +
+				"(default user, no username in the URL)",
+			adminURL:     "redis://:pooLpassw0rd@127.0.0.1:6380/3",
+			redisHost:    "ignored-when-url-set:6379",
+			wantAddr:     "127.0.0.1:6380",
+			wantUsername: "",
+			wantPassword: "pooLpassw0rd",
+			wantDB:       3,
+		},
+		{
+			name:         "url unset — legacy REDIS_PROVISION_HOST Addr form, unchanged",
+			adminURL:     "",
+			redisHost:    "custom:6380",
+			wantAddr:     "custom:6380",
+			wantUsername: "",
+			wantPassword: "",
+			wantDB:       0,
+		},
+		{
+			name:         "url unset and host unset — package default addr",
+			adminURL:     "",
+			redisHost:    "",
+			wantAddr:     defaultRedisAddr,
+			wantUsername: "",
+			wantPassword: "",
+			wantDB:       0,
+		},
+		{
+			name: "url malformed (bad scheme) — falls back to the host form " +
+				"instead of failing the whole provisioner start",
+			adminURL:     "http://admin:s3cret@somewhere:6379",
+			redisHost:    "fallback:6379",
+			wantAddr:     "fallback:6379",
+			wantUsername: "",
+			wantPassword: "",
+			wantDB:       0,
+		},
+		{
+			name: "url malformed (control character, url.Parse failure) — " +
+				"falls back to the host form",
+			adminURL:     "redis://admin:s3cret@some\x7fwhere:6379",
+			redisHost:    "fallback:6379",
+			wantAddr:     "fallback:6379",
+			wantUsername: "",
+			wantPassword: "",
+			wantDB:       0,
+		},
+		{
+			name: "url malformed AND host unset — package default addr, " +
+				"never a half-configured client",
+			adminURL:     "::not a url::",
+			redisHost:    "",
+			wantAddr:     defaultRedisAddr,
+			wantUsername: "",
+			wantPassword: "",
+			wantDB:       0,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			b := newLocalBackend(tc.adminURL, tc.redisHost)
+			opts := b.rdb.Options()
+
+			if opts.Addr != tc.wantAddr {
+				t.Errorf("admin client Addr = %q; want %q", opts.Addr, tc.wantAddr)
+			}
+			if opts.Username != tc.wantUsername {
+				t.Errorf("admin client Username = %q; want %q", opts.Username, tc.wantUsername)
+			}
+			if opts.Password != tc.wantPassword {
+				t.Errorf("admin client Password = %q; want %q", opts.Password, tc.wantPassword)
+			}
+			if opts.DB != tc.wantDB {
+				t.Errorf("admin client DB = %d; want %d", opts.DB, tc.wantDB)
+			}
+
+			// The recorded host is what Provision embeds in the CUSTOMER url
+			// when no REDIS_PUBLIC_HOST* is configured. It must always be a
+			// bare host:port — never the admin URL, which carries the shared
+			// pool's password.
+			if b.redisHost != tc.wantAddr {
+				t.Errorf("redisHost = %q; want %q (bare host:port)", b.redisHost, tc.wantAddr)
+			}
+			if strings.Contains(b.redisHost, "@") || strings.Contains(b.redisHost, "//") {
+				t.Errorf("redisHost = %q leaks admin URL structure into customer URLs", b.redisHost)
+			}
+		})
+	}
+}
+
+// TestNewLocalBackend_AdminURLDoesNotLeakIntoCustomerURL pins the specific
+// regression the redisHost assertion above guards: the password from
+// REDIS_PROVISION_URL must never appear in a credential returned to a customer.
+func TestNewLocalBackend_AdminURLDoesNotLeakIntoCustomerURL(t *testing.T) {
+	const adminPassword = "sup3r-s3cret-pool-pw"
+	b := newLocalBackend("redis://admin:"+adminPassword+"@10.0.0.1:6379/0", "")
+
+	// Provision cannot run without a server, but the host it would interpolate
+	// is fixed at construction time — assert on that.
+	if strings.Contains(b.redisHost, adminPassword) {
+		t.Fatalf("redisHost %q contains the admin password; it is interpolated into customer URLs", b.redisHost)
+	}
+}
+
+// TestParseAdminURL covers the helper's three return shapes directly, including
+// that a parsed URL's non-credential options survive.
+func TestParseAdminURL(t *testing.T) {
+	t.Run("empty returns nil", func(t *testing.T) {
+		if got := parseAdminURL(""); got != nil {
+			t.Errorf("parseAdminURL(\"\") = %+v; want nil", got)
+		}
+	})
+
+	t.Run("malformed returns nil", func(t *testing.T) {
+		if got := parseAdminURL("not-a-redis-url"); got != nil {
+			t.Errorf("parseAdminURL(malformed) = %+v; want nil", got)
+		}
+	})
+
+	t.Run("valid returns options with query params applied", func(t *testing.T) {
+		got := parseAdminURL("redis://:pw@h:6379/2?dial_timeout=7s")
+		if got == nil {
+			t.Fatal("parseAdminURL(valid) = nil; want options")
+		}
+		if got.Password != "pw" || got.Addr != "h:6379" || got.DB != 2 {
+			t.Errorf("options = %+v; want Addr h:6379, Password pw, DB 2", got)
+		}
+		if got.DialTimeout.Seconds() != 7 {
+			t.Errorf("DialTimeout = %v; want 7s (query params must survive)", got.DialTimeout)
+		}
+	})
+}
+
+// TestRedactCredentials asserts the log-sanitiser strips URL userinfo. A
+// net/url parse error stringifies as `parse "<the whole URL>": …`, so logging a
+// ParseURL failure verbatim would publish the shared pool's admin password.
+func TestRedactCredentials(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"empty", "", ""},
+		{
+			"no url — untouched",
+			"redis: invalid URL scheme: http",
+			"redis: invalid URL scheme: http",
+		},
+		{
+			"user:password userinfo redacted",
+			`parse "redis://admin:s3cret@host:6379": net/url: invalid control character in URL`,
+			`parse "redis://***:***@host:6379": net/url: invalid control character in URL`,
+		},
+		{
+			"password-only userinfo redacted",
+			`parse "redis://:s3cret@host:6379": boom`,
+			`parse "redis://***:***@host:6379": boom`,
+		},
+		{
+			"username-only userinfo left alone (no secret in it)",
+			`parse "redis://admin@host:6379": boom`,
+			`parse "redis://admin@host:6379": boom`,
+		},
+		{
+			"every occurrence redacted, not just the first",
+			`a redis://u:p@h1 and b redis://u2:p2@h2`,
+			`a redis://***:***@h1 and b redis://***:***@h2`,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := redactCredentials(tc.in); got != tc.want {
+				t.Errorf("redactCredentials(%q) = %q; want %q", tc.in, got, tc.want)
+			}
+		})
+	}
+}
+
+// TestRedactCredentials_OnRealParseError closes the loop on the redaction: it
+// feeds the sanitiser the ACTUAL error text go-redis produces for a credentialed
+// URL that url.Parse rejects, rather than a hand-written approximation of it.
+func TestRedactCredentials_OnRealParseError(t *testing.T) {
+	const adminPassword = "sup3r-s3cret-pool-pw"
+	_, err := goredisParseURL("redis://admin:" + adminPassword + "@ho\x7fst:6379")
+	if err == nil {
+		t.Fatal("expected a parse error for a URL containing a control character")
+	}
+	if !strings.Contains(err.Error(), adminPassword) {
+		t.Skipf("go-redis no longer echoes the URL in its error (%v) — redaction is belt-and-braces here", err)
+	}
+	if got := redactCredentials(err.Error()); strings.Contains(got, adminPassword) {
+		t.Errorf("redacted error still contains the admin password: %q", got)
+	}
+}
+
+// TestSharedBackendConstructors_AllHonourAdminURL is the rule-18
+// registry-iterating guard: EVERY constructor that yields a shared LocalBackend
+// must thread adminURL through to the admin client. A new constructor added
+// without the parameter (the "two emitters of one broken behaviour" failure
+// mode) fails here rather than silently 503-ing /cache/new in prod.
+func TestSharedBackendConstructors_AllHonourAdminURL(t *testing.T) {
+	const (
+		adminURL = "redis://admin:ctor-pw@ctor-host:6390/0"
+		wantAddr = "ctor-host:6390"
+		wantPass = "ctor-pw"
+	)
+
+	ctors := map[string]func(adminURL, redisHost string) Backend{
+		// Default arm — the backend prod actually runs (REDIS_PROVISION_BACKEND=local).
+		"NewBackend/local": func(a, h string) Backend { return NewBackend("local", a, h) },
+		// Unknown backend names fall through to the same default arm.
+		"NewBackend/unknown": func(a, h string) Backend { return NewBackend("no-such-backend", a, h) },
+		// Non-Team side of tier-aware routing (REDIS_TIER_AWARE_ROUTING_ENABLED).
+		"NewSharedCarveBackend": NewSharedCarveBackend,
+	}
+
+	for name, ctor := range ctors {
+		t.Run(name, func(t *testing.T) {
+			b := ctor(adminURL, "unused:6379")
+			local, ok := b.(*LocalBackend)
+			if !ok {
+				t.Fatalf("%s returned %T; want *LocalBackend", name, b)
+			}
+			opts := local.rdb.Options()
+			if opts.Addr != wantAddr {
+				t.Errorf("%s: Addr = %q; want %q", name, opts.Addr, wantAddr)
+			}
+			if opts.Password != wantPass {
+				t.Errorf("%s: Password = %q; want %q — this constructor ignores REDIS_PROVISION_URL and will 503 on a password-protected pool",
+					name, opts.Password, wantPass)
+			}
+		})
+	}
+}
+
+// TestSharedBackendConstructors_FallBackToHost is the same enumeration for the
+// unset-URL case: no constructor may start requiring REDIS_PROVISION_URL.
+func TestSharedBackendConstructors_FallBackToHost(t *testing.T) {
+	ctors := map[string]func(adminURL, redisHost string) Backend{
+		"NewBackend/local":      func(a, h string) Backend { return NewBackend("local", a, h) },
+		"NewBackend/unknown":    func(a, h string) Backend { return NewBackend("no-such-backend", a, h) },
+		"NewSharedCarveBackend": NewSharedCarveBackend,
+	}
+
+	for name, ctor := range ctors {
+		t.Run(name, func(t *testing.T) {
+			b := ctor("", "legacy-host:6379")
+			local, ok := b.(*LocalBackend)
+			if !ok {
+				t.Fatalf("%s returned %T; want *LocalBackend", name, b)
+			}
+			opts := local.rdb.Options()
+			if opts.Addr != "legacy-host:6379" {
+				t.Errorf("%s: Addr = %q; want legacy-host:6379", name, opts.Addr)
+			}
+			if opts.Password != "" {
+				t.Errorf("%s: Password = %q; want empty (unauthenticated legacy pool)", name, opts.Password)
+			}
+		})
+	}
+}
+
+// TestGoredisNewClient_UsesGivenOptions guards the narrow alias newLocalBackend
+// now depends on: a client built from parsed options must keep them.
+func TestGoredisNewClient_UsesGivenOptions(t *testing.T) {
+	c := goredisNewClient(&goredis.Options{Addr: "a:1", Password: "p"})
+	if c.Options().Addr != "a:1" || c.Options().Password != "p" {
+		t.Errorf("goredisNewClient dropped options: %+v", c.Options())
+	}
+}
diff --git a/internal/backend/redis/backend.go b/internal/backend/redis/backend.go
index 2c32f5f..f688beb 100644
--- a/internal/backend/redis/backend.go
+++ b/internal/backend/redis/backend.go
@@ -78,7 +78,11 @@ type Credentials struct {
 // NewBackend creates a Backend using the given backend type string.
 // "k8s" → K8sBackend (dedicated pod per token, every tier).
 // "local" (default) → LocalBackend (ACL user on shared cluster).
-func NewBackend(backendType, redisHost string) Backend {
+//
+// adminURL (REDIS_PROVISION_URL) carries the credentials for the shared pool and
+// wins over redisHost (REDIS_PROVISION_HOST) when set — see newLocalBackend. The
+// two-input shape mirrors mongo.NewBackend(backendType, adminURI, mongoHost).
+func NewBackend(backendType, adminURL, redisHost string) Backend {
 	switch backendType {
 	case "k8s":
 		// Dedicated-pod-per-resource backend for every tier. Each /cache/new
@@ -103,7 +107,7 @@ func NewBackend(backendType, redisHost string) Backend {
 		b, err := newK8sBackend(kubeconfig, storageClass, image, externalHost, storageSizeGi)
 		if err != nil {
 			slog.Error("redis.k8s_backend_init_failed_fallback_to_local", "error", err)
-			return newLocalBackend(redisHost)
+			return newLocalBackend(adminURL, redisHost)
 		}
 		b.SetPublicHost(publicHost)
 		// Route registry — writes route records per provision so the
@@ -122,15 +126,16 @@ func NewBackend(backendType, redisHost string) Backend {
 		slog.Info("redis.backend_selected", "backend", "k8s", "external_host", externalHost, "public_host", publicHost)
 		return b
 	default:
-		return newLocalBackend(redisHost)
+		return newLocalBackend(adminURL, redisHost)
 	}
 }
 
 // NewSharedCarveBackend creates a LocalBackend: an ACL user + key-prefix carve
 // on a SHARED Redis instance (many tenants per pod). It is the non-Team side of
-// tier-aware routing (see TierDispatchBackend). redisHost is "host:port".
-func NewSharedCarveBackend(redisHost string) Backend {
-	return newLocalBackend(redisHost)
+// tier-aware routing (see TierDispatchBackend). adminURL is the credentialed
+// REDIS_PROVISION_URL (may be empty); redisHost is the "host:port" fallback.
+func NewSharedCarveBackend(adminURL, redisHost string) Backend {
+	return newLocalBackend(adminURL, redisHost)
 }
 
 // NewDedicatedBackend creates a DedicatedProvider for Team-tier Redis provisioning.
diff --git a/internal/backend/redis/coverage_test.go b/internal/backend/redis/coverage_test.go
index 95341aa..68e1047 100644
--- a/internal/backend/redis/coverage_test.go
+++ b/internal/backend/redis/coverage_test.go
@@ -129,11 +129,11 @@ func TestGoredisHelpers(t *testing.T) {
 // TestNewBackend_LocalDefault exercises the default switch arm — any unknown
 // backendType falls back to newLocalBackend.
 func TestNewBackend_LocalDefault(t *testing.T) {
-	b := NewBackend("", "localhost:6379")
+	b := NewBackend("", "", "localhost:6379")
 	if _, ok := b.(*LocalBackend); !ok {
 		t.Errorf("NewBackend(\"\") returned %T; want *LocalBackend", b)
 	}
-	b2 := NewBackend("unknown-backend-name", "localhost:6379")
+	b2 := NewBackend("unknown-backend-name", "", "localhost:6379")
 	if _, ok := b2.(*LocalBackend); !ok {
 		t.Errorf("NewBackend(unknown) returned %T; want *LocalBackend", b2)
 	}
@@ -147,7 +147,7 @@ func TestNewBackend_K8sFallsBackOnInitError(t *testing.T) {
 	// returns a LocalBackend.
 	t.Setenv("K8S_KUBECONFIG", "")
 	// REDIS_URL etc. unset so the route-registry path is not exercised.
-	b := NewBackend("k8s", "localhost:6379")
+	b := NewBackend("k8s", "", "localhost:6379")
 	if _, ok := b.(*LocalBackend); !ok {
 		t.Fatalf("expected LocalBackend fallback on k8s init failure; got %T", b)
 	}
@@ -162,7 +162,7 @@ func TestNewBackend_K8sWithBadKubeconfig(t *testing.T) {
 		t.Fatalf("write tmp kubeconfig: %v", err)
 	}
 	t.Setenv("K8S_KUBECONFIG", tmp)
-	b := NewBackend("k8s", "localhost:6379")
+	b := NewBackend("k8s", "", "localhost:6379")
 	if _, ok := b.(*LocalBackend); !ok {
 		t.Fatalf("expected LocalBackend fallback for bad kubeconfig; got %T", b)
 	}
@@ -190,11 +190,11 @@ func TestNewK8sDedicatedBackend_ErrorWithoutKubeconfig(t *testing.T) {
 
 // TestNewLocalBackend_DefaultAddr verifies the empty-host fallback.
 func TestNewLocalBackend_DefaultAddr(t *testing.T) {
-	b := newLocalBackend("")
+	b := newLocalBackend("", "")
 	if b.redisHost != defaultRedisAddr {
 		t.Errorf("redisHost = %q; want %q (default)", b.redisHost, defaultRedisAddr)
 	}
-	b2 := newLocalBackend("custom:6380")
+	b2 := newLocalBackend("", "custom:6380")
 	if b2.redisHost != "custom:6380" {
 		t.Errorf("redisHost = %q; want custom:6380", b2.redisHost)
 	}
@@ -236,7 +236,7 @@ func TestPublicHostPort(t *testing.T) {
 // returned. Also exercises the publicHost env override.
 func TestLocalBackend_Provision_ACLPath(t *testing.T) {
 	addr := liveRedisAddr(t)
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 
 	t.Setenv("REDIS_PUBLIC_HOST_PORT", "redis.example.com:6379")
@@ -283,7 +283,7 @@ func TestLocalBackend_Provision_NoPublicHost(t *testing.T) {
 	os.Unsetenv("REDIS_PUBLIC_HOST")
 	os.Unsetenv("REDIS_PUBLIC_PORT")
 
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 	token := uniqueToken(t, "covnph-")
 	defer func() { _ = b.Deprovision(context.Background(), token, "") }()
@@ -301,7 +301,7 @@ func TestLocalBackend_Provision_NoPublicHost(t *testing.T) {
 // namespace and asserts StorageBytes returns the per-key memory sum.
 func TestLocalBackend_StorageBytes_PrefixSum(t *testing.T) {
 	addr := liveRedisAddr(t)
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 	token := uniqueToken(t, "covstor-")
 	defer func() { _ = b.Deprovision(context.Background(), token, "") }()
@@ -327,7 +327,7 @@ func TestLocalBackend_StorageBytes_PrefixSum(t *testing.T) {
 // the namespace keys are removed.
 func TestLocalBackend_Deprovision_DeletesACLAndKeys(t *testing.T) {
 	addr := liveRedisAddr(t)
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 	token := uniqueToken(t, "covdep-")
 
@@ -360,7 +360,7 @@ func TestLocalBackend_Deprovision_DeletesACLAndKeys(t *testing.T) {
 // path so Deprovision's SCAN loop exits without ever hitting DEL.
 func TestLocalBackend_Deprovision_NoOpWhenNoKeys(t *testing.T) {
 	addr := liveRedisAddr(t)
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 	tok := uniqueToken(t, "covdepempty-")
 	if err := b.Deprovision(context.Background(), tok, ""); err != nil {
@@ -1176,7 +1176,7 @@ users:
 	t.Setenv("REDIS_URL_FOR_ROUTES", "redis://"+addr)
 	t.Setenv("REDIS_PROXY_ROUTE_PREFIX", "cov_route:")
 	t.Setenv("REDIS_PROXY_PASSWORD_ROUTE_PREFIX", "cov_route_pw:")
-	b := NewBackend("k8s", "")
+	b := NewBackend("k8s", "", "")
 	if _, ok := b.(*K8sBackend); !ok {
 		t.Fatalf("NewBackend(k8s) returned %T; want *K8sBackend", b)
 	}
@@ -1208,7 +1208,7 @@ users: [{name: f, user: {token: t}}]
 	}
 	t.Setenv("K8S_KUBECONFIG", tmp)
 	t.Setenv("REDIS_URL_FOR_ROUTES", "::not-a-url::")
-	b := NewBackend("k8s", "")
+	b := NewBackend("k8s", "", "")
 	if _, ok := b.(*K8sBackend); !ok {
 		t.Fatalf("NewBackend(k8s) returned %T; want *K8sBackend", b)
 	}
@@ -1761,7 +1761,7 @@ func TestLocalBackend_Deprovision_ScanError(t *testing.T) {
 // keys exist (loop never enters).
 func TestLocalBackend_StorageBytes_EmptyNamespace(t *testing.T) {
 	addr := liveRedisAddr(t)
-	b := newLocalBackend(addr)
+	b := newLocalBackend("", addr)
 	defer b.rdb.Close()
 	got, err := b.StorageBytes(context.Background(), "no-such-token-"+uniqueToken(t, ""), "")
 	if err != nil {
diff --git a/internal/backend/redis/dispatch_test.go b/internal/backend/redis/dispatch_test.go
index 747e8d7..1ecdfec 100644
--- a/internal/backend/redis/dispatch_test.go
+++ b/internal/backend/redis/dispatch_test.go
@@ -339,7 +339,7 @@ func TestDispatchImplementsRegrader(t *testing.T) {
 // constructor returns a LocalBackend (ACL carve on a shared Redis) — the
 // non-Team side of tier-aware routing.
 func TestNewSharedCarveBackend_IsLocalBackend(t *testing.T) {
-	b := NewSharedCarveBackend("localhost:6379")
+	b := NewSharedCarveBackend("", "localhost:6379")
 	if _, ok := b.(*LocalBackend); !ok {
 		t.Fatalf("NewSharedCarveBackend returned %T, want *LocalBackend", b)
 	}
diff --git a/internal/backend/redis/local.go b/internal/backend/redis/local.go
index 722d17d..da2d995 100644
--- a/internal/backend/redis/local.go
+++ b/internal/backend/redis/local.go
@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"regexp"
 
 	goredis "github.com/redis/go-redis/v9"
 
@@ -95,19 +96,73 @@ type LocalBackend struct {
 	redisHost string          // Redis host for building connection strings
 }
 
-// newLocalBackend creates a LocalBackend connecting to the given redisHost.
-// redisHost format: "host:port" (e.g. "localhost:6379").
-func newLocalBackend(redisHost string) *LocalBackend {
+// newLocalBackend creates a LocalBackend for the shared redis-provision pool.
+//
+// adminURL — REDIS_PROVISION_URL, "redis://[user]:password@host:port[/db]".
+// When set and parseable it becomes the admin connection. It is the ONLY way to
+// authenticate against a redis-provision pod started with --requirepass: the
+// bare-Addr client below sends no AUTH, so the first ACL SETUSER comes back
+// "ERR Protocol error: unauthenticated multibulk length" and Provision fails
+// closed (503 on /cache/new) rather than handing out a credential-less shared
+// URL. Mirrors mongo's (adminURI, mongoHost) split — the credentialed admin
+// endpoint and the customer-facing host are separate inputs.
+//
+// redisHost — REDIS_PROVISION_HOST, "host:port". The legacy, credential-less
+// form; still the fallback when adminURL is unset, so an unauthenticated pool
+// (dev, or prod before the --requirepass hardening) behaves exactly as before.
+//
+// The host recorded on the backend is always a bare host:port — opts.Addr, never
+// adminURL. Embedding adminURL would leak the admin password into the customer's
+// connection string through the Provision host fallback below.
+func newLocalBackend(adminURL, redisHost string) *LocalBackend {
+	if opts := parseAdminURL(adminURL); opts != nil {
+		return &LocalBackend{rdb: goredisNewClient(opts), redisHost: opts.Addr}
+	}
 	if redisHost == "" {
 		redisHost = defaultRedisAddr
 	}
-	rdb := goredis.NewClient(&goredis.Options{
+	rdb := goredisNewClient(&goredis.Options{
 		Addr: redisHost,
 	})
-	// Extract just the host portion for URL building (strip port if needed for URL).
 	return &LocalBackend{rdb: rdb, redisHost: redisHost}
 }
 
+// parseAdminURL parses REDIS_PROVISION_URL into go-redis client options.
+// Returns nil when the URL is unset (the normal legacy case) or malformed —
+// both leave the caller on the REDIS_PROVISION_HOST Addr fallback.
+//
+// A malformed URL is logged, not returned as an error: the factory chain
+// (NewBackend → newLocalBackend) has no error channel, and refusing to start
+// the provisioner over one typo is a worse outage than the 503 the (still
+// fail-closed) Provision path produces. The log line is the operator's only
+// signal, so it must be unmissable — and it must NOT contain the password,
+// which net/url parse errors echo back verbatim.
+func parseAdminURL(adminURL string) *goredis.Options {
+	if adminURL == "" {
+		return nil
+	}
+	opts, err := goredisParseURL(adminURL)
+	if err != nil {
+		slog.Error("cache.local: REDIS_PROVISION_URL is malformed — falling back to REDIS_PROVISION_HOST with NO credentials; ACL SETUSER will fail on a password-protected pool",
+			"error", redactCredentials(err.Error()))
+		return nil
+	}
+	return opts
+}
+
+// urlCredentialsRe matches the "user:password@" userinfo section of a URL
+// embedded in an arbitrary string. Only userinfo that carries a password (i.e.
+// contains ":") is matched — a bare "//user@host" leaks nothing worth hiding.
+var urlCredentialsRe = regexp.MustCompile(`(//)[^/@\s]*:[^/@\s]*@`)
+
+// redactCredentials strips URL userinfo from a string before it reaches a log
+// sink. net/url's *url.Error stringifies as `parse "<the whole URL>": …`, so
+// logging a ParseURL failure verbatim would publish the shared Redis admin
+// password to stdout — the exact secret the --requirepass hardening added.
+func redactCredentials(s string) string {
+	return urlCredentialsRe.ReplaceAllString(s, "$1***:***@")
+}
+
 // Provision creates a namespaced Redis "database" for the given token.
 // Tries Redis ACL (Redis 6+) first. Falls back to key-namespace isolation
 // if ACL is unavailable or disabled.
diff --git a/internal/config/config.go b/internal/config/config.go
index aebf8a0..9e3293b 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -16,6 +16,15 @@ type Config struct {
 	RedisProvisionBackend    string // REDIS_PROVISION_BACKEND, default "local"
 	RedisProvisionHost       string // REDIS_PROVISION_HOST, default "localhost:6379"
 
+	// RedisProvisionURL is the CREDENTIALED admin connection for the shared
+	// redis-provision pool: "redis://[user]:password@host:port[/db]"
+	// (REDIS_PROVISION_URL, default unset). It supersedes RedisProvisionHost,
+	// which can only express a bare "host:port" and therefore cannot AUTH
+	// against a pod started with --requirepass — every ACL SETUSER then fails
+	// and /cache/new 503s. Unset = unchanged legacy behaviour (Addr-only).
+	// Same role as PostgresCustomersURL / MongoAdminURI for their backends.
+	RedisProvisionURL string // REDIS_PROVISION_URL, default "" (falls back to RedisProvisionHost)
+
 	// RedisTierAwareRoutingEnabled is the kill-switch for tier-aware Redis
 	// backend routing (REDIS_TIER_AWARE_ROUTING_ENABLED). DEFAULT FALSE /
 	// fail-closed: when false (or unset) the shared redisBackend serves every
@@ -115,6 +124,9 @@ func Load() *Config {
 		NeonRegionID:             getenv("NEON_REGION_ID", "aws-us-east-1"),
 		RedisProvisionBackend:    getenv("REDIS_PROVISION_BACKEND", "local"),
 		RedisProvisionHost:       getenv("REDIS_PROVISION_HOST", "localhost:6379"),
+		// No default: an empty REDIS_PROVISION_URL means "use the legacy
+		// REDIS_PROVISION_HOST Addr form", so nothing that works today breaks.
+		RedisProvisionURL: os.Getenv("REDIS_PROVISION_URL"),
 		// Default false: tier-aware routing is opt-in. Any value other than the
 		// exact string "true" leaves it off (fail-closed) — same pattern as
 		// K8sDedicatedBackend.
@@ -170,6 +182,8 @@ func logStartupConfig(cfg *Config) {
 		"neon_api_key_set", cfg.NeonAPIKey != "",
 		"redis_provision_backend", cfg.RedisProvisionBackend,
 		"redis_provision_host", cfg.RedisProvisionHost,
+		// Presence only — the URL embeds the shared pool's admin password.
+		"redis_provision_url_set", cfg.RedisProvisionURL != "",
 		"redis_tier_aware_routing_enabled", cfg.RedisTierAwareRoutingEnabled,
 		"mongo_admin_uri_set", cfg.MongoAdminURI != "",
 		"mongo_host", cfg.MongoHost,
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 03b7573..f806a3f 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -23,6 +23,7 @@ var allConfigEnvKeys = []string{
 	"NEON_REGION_ID",
 	"REDIS_PROVISION_BACKEND",
 	"REDIS_PROVISION_HOST",
+	"REDIS_PROVISION_URL",
 	"REDIS_TIER_AWARE_ROUTING_ENABLED",
 	"MONGO_PROVISION_BACKEND",
 	"MONGO_ADMIN_URI",
@@ -142,6 +143,8 @@ func TestLoad_Defaults(t *testing.T) {
 		{"NeonRegionID", cfg.NeonRegionID, "aws-us-east-1"},
 		{"RedisProvisionBackend", cfg.RedisProvisionBackend, "local"},
 		{"RedisProvisionHost", cfg.RedisProvisionHost, "localhost:6379"},
+		// No default: unset means "use the RedisProvisionHost Addr form".
+		{"RedisProvisionURL", cfg.RedisProvisionURL, ""},
 		{"MongoProvisionBackend", cfg.MongoProvisionBackend, "local"},
 		{"MongoAdminURI", cfg.MongoAdminURI, "mongodb://root:root@localhost:27017"},
 		{"MongoHost", cfg.MongoHost, "localhost:27017"},
@@ -214,6 +217,7 @@ func TestLoad_Overrides(t *testing.T) {
 	t.Setenv("NEON_REGION_ID", "aws-eu-west-1")
 	t.Setenv("REDIS_PROVISION_BACKEND", "upstash")
 	t.Setenv("REDIS_PROVISION_HOST", "redis.example:6380")
+	t.Setenv("REDIS_PROVISION_URL", "redis://:poolpw@redis.example:6380/0")
 	t.Setenv("REDIS_TIER_AWARE_ROUTING_ENABLED", "true")
 	t.Setenv("MONGO_PROVISION_BACKEND", "k8s")
 	t.Setenv("MONGO_ADMIN_URI", "mongodb://admin:pw@m:27017")
@@ -261,6 +265,7 @@ func TestLoad_Overrides(t *testing.T) {
 		{"NeonRegionID", cfg.NeonRegionID, "aws-eu-west-1"},
 		{"RedisProvisionBackend", cfg.RedisProvisionBackend, "upstash"},
 		{"RedisProvisionHost", cfg.RedisProvisionHost, "redis.example:6380"},
+		{"RedisProvisionURL", cfg.RedisProvisionURL, "redis://:poolpw@redis.example:6380/0"},
 		{"MongoProvisionBackend", cfg.MongoProvisionBackend, "k8s"},
 		{"MongoAdminURI", cfg.MongoAdminURI, "mongodb://admin:pw@m:27017"},
 		{"MongoHost", cfg.MongoHost, "m.example:27017"},
diff --git a/internal/pool/factory.go b/internal/pool/factory.go
index 0aa0ec2..f66ce3d 100644
--- a/internal/pool/factory.go
+++ b/internal/pool/factory.go
@@ -23,6 +23,7 @@ func NewWithConfig(db *pgxpool.Pool, aesKey []byte, cfg Config, appCfg *config.C
 	)
 	redisB := redis.NewBackend(
 		appCfg.RedisProvisionBackend,
+		appCfg.RedisProvisionURL,
 		appCfg.RedisProvisionHost,
 	)
 	mongoB := mongo.NewBackend(
diff --git a/internal/server/server.go b/internal/server/server.go
index 650820e..837249c 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -160,9 +160,9 @@ func New(cfg *config.Config, poolMgr *pool.Manager) *Server {
 	// but moves every non-Team tier onto a shared ACL carve so a 5MB cache no
 	// longer costs a whole pod. Off = identical to today; this branch never runs
 	// in prod until an operator flips the flag.
-	redisBackend := redis.NewBackend(cfg.RedisProvisionBackend, cfg.RedisProvisionHost)
+	redisBackend := redis.NewBackend(cfg.RedisProvisionBackend, cfg.RedisProvisionURL, cfg.RedisProvisionHost)
 	if cfg.RedisTierAwareRoutingEnabled {
-		sharedCarve := redis.NewSharedCarveBackend(cfg.RedisProvisionHost)
+		sharedCarve := redis.NewSharedCarveBackend(cfg.RedisProvisionURL, cfg.RedisProvisionHost)
 		redisBackend = redis.NewTierDispatchBackend(sharedCarve, redisBackend)
 		slog.Info("provisioner: redis tier-aware routing ENABLED",
 			"configured_backend", cfg.RedisProvisionBackend,
diff --git a/internal/server/server_live_roundtrip_test.go b/internal/server/server_live_roundtrip_test.go
index 8968ae6..fec9be4 100644
--- a/internal/server/server_live_roundtrip_test.go
+++ b/internal/server/server_live_roundtrip_test.go
@@ -81,7 +81,9 @@ func liveServerWithRealRedis(redisAddr string) *server.Server {
 	return server.NewWithBackends(
 		&config.Config{},
 		nil,
-		redis.NewBackend("", redisAddr), // "" → LocalBackend(redisAddr)
+		// backendType "" → LocalBackend; adminURL "" → the credential-less
+		// REDIS_PROVISION_HOST Addr form, which is what the test's local Redis wants.
+		redis.NewBackend("", "", redisAddr),
 		nil, nil, nil,
 		nil, nil, nil, nil,
 		nil,