Add bridge metrics and broker API key rotation

.env.example

@@ -24,6 +24,11 @@ ENCRYPTION_KEY=
 STATE_KEY=
 
 API_KEY=nexus-admin-key
+# Optional: mounted secret files for API key rotation without broker restart.
+# API_KEY_FILE may contain one key; API_KEYS_FILE may contain comma/newline-separated keys.
+API_KEY_FILE=
+API_KEYS_FILE=
+API_KEY_RELOAD_INTERVAL=30s
 
 # --- Policies ---
 ALLOWED_CIDRS=0.0.0.0/0

docs/concepts/broker.md

@@ -60,4 +60,6 @@ openssl rand -base64 32
 
 ## Network isolation
 
-The Broker should only accept connections from the Gateway's IP range. It should not be reachable from the public internet or from agent processes. The `API_KEY` on the Broker should be known only to the Gateway.
+The Broker should only accept connections from the Gateway's IP range. It should not be reachable from the public internet or from agent processes. The Broker API key should be known only to the Gateway and can be supplied through `API_KEY`, `API_KEYS`, `API_KEY_FILE`, or `API_KEYS_FILE`.
+
+For production rotation, mount API keys as secret files and configure `API_KEY_FILE` or `API_KEYS_FILE`. The Broker reloads those files on `API_KEY_RELOAD_INTERVAL` without a process restart.

docs/concepts/security-model.md

@@ -50,6 +50,8 @@ API keys are not sufficient on their own. Layer network controls on top:
 
 CIDR allowlisting is configurable on both the Broker and the Gateway.
 
+Broker API keys can be rotated without restarting the Broker by mounting key files with `API_KEY_FILE` or `API_KEYS_FILE`. The files are reloaded according to `API_KEY_RELOAD_INTERVAL`.
+
 ## Audit log
 
 Every significant event is written to `audit_events`: consents created, tokens issued, refreshes succeeded or failed, connections cleaned up. Each row includes IP address and User-Agent.

docs/getting-started/configuration.md

@@ -26,7 +26,11 @@ Both the Broker and the Gateway must receive the same value for `STATE_KEY`. If
 | `REDIS_URL` | Yes | Redis URL for caching and peer discovery. Example: `redis://localhost:6379` |
 | `ENCRYPTION_KEY` | Yes | 32-byte Base64 string for AES-GCM 256-bit token encryption. Generate with `openssl rand -base64 32`. This key must never change while connections exist in the database. |
 | `STATE_KEY` | Yes | Same as the shared `STATE_KEY`. Must match the Gateway exactly. |
-| `API_KEY` | Yes | Key that the Gateway and admin callers use to authenticate with the Broker. |
+| `API_KEY` | Conditional | Single key that the Gateway and admin callers use to authenticate with the Broker. Required unless `API_KEYS`, `API_KEY_FILE`, or `API_KEYS_FILE` supplies keys. |
+| `API_KEYS` | No | Comma-separated Broker API keys. Useful during manual key rollover. |
+| `API_KEY_FILE` | No | Path to a mounted secret file containing one Broker API key. Reloaded without process restart. |
+| `API_KEYS_FILE` | No | Path to a mounted secret file containing comma- or newline-separated Broker API keys. Reloaded without process restart. |
+| `API_KEY_RELOAD_INTERVAL` | No | How often the Broker re-reads `API_KEY_FILE` and `API_KEYS_FILE`. Default: `30s` |
 | `BASE_URL` | Yes | The public URL of the Broker, used to construct the OAuth callback URL. Example: `https://broker.example.com` |
 | `REDIRECT_PATH` | No | The path appended to `BASE_URL` for the OAuth callback. Default: `/auth/callback` |
 | `ALLOWED_CIDRS` | No | Comma-separated list of IP ranges allowed to reach the Broker. In production, restrict this to the Gateway's IP. Example: `10.0.0.0/8` |

docs/infrastructure/deploying-nexus.md

@@ -32,7 +32,11 @@ Store these in your secret manager (AWS Secrets Manager, Azure Key Vault, HashiC
 | `ENCRYPTION_KEY` | yes | 32-byte base64 key for AES-GCM token encryption |
 | `STATE_KEY` | yes | 32-byte base64 key for OAuth state HMAC signing — must match Gateway |
 | `BASE_URL` | yes | Public URL of the Broker, e.g. `https://broker.internal.example.com` |
-| `API_KEY` | yes | Key the Gateway uses to authenticate with the Broker |
+| `API_KEY` | conditional | Single key the Gateway uses to authenticate with the Broker. Required unless `API_KEYS`, `API_KEY_FILE`, or `API_KEYS_FILE` supplies keys. |
+| `API_KEYS` | no | Comma-separated Broker API keys for key rollover. |
+| `API_KEY_FILE` | no | Mounted secret file containing one Broker API key; reloaded without process restart. |
+| `API_KEYS_FILE` | no | Mounted secret file containing comma- or newline-separated Broker API keys; reloaded without process restart. |
+| `API_KEY_RELOAD_INTERVAL` | no | File reload interval for `API_KEY_FILE` and `API_KEYS_FILE` (default: `30s`). |
 | `REDIRECT_PATH` | no | OAuth callback path (default: `/auth/callback`) |
 | `ALLOWED_CIDRS` | no | Comma-separated CIDRs for IP allowlisting, e.g. `10.0.0.0/8` |
 | `ALLOWED_RETURN_DOMAINS` | no | Comma-separated allowed domains for `return_url` validation |
@@ -133,6 +137,8 @@ volumes:
 
 Both keys must be identical across all instances of the same service. In Kubernetes, use a single `Secret` object mounted into both the Broker and Gateway pods for `STATE_KEY`.
 
+For Broker API key rotation, prefer `API_KEY_FILE` or `API_KEYS_FILE` backed by a mounted secret. The Broker reloads those files on `API_KEY_RELOAD_INTERVAL`, so updating the secret volume can add or remove accepted keys without restarting the pod.
+
 ## Health checks
 
 The Broker exposes `GET /health` and the Gateway exposes `GET /health`. Both return `200 OK` when the service is ready. Configure your load balancer to use these endpoints.

docs/reference/security-model.md

@@ -23,6 +23,9 @@ Security in Nexus relies on three primary environment variables that must be gua
 ### 3. The API Key (`API_KEY` / `BROKER_API_KEY`)
 - **Role:** Authenticates the Gateway to the Broker and the Admin to the Broker.
 - **Impact:** Controls access to provider registration and token retrieval.
+- **Rotation:** For runtime rotation without restarting the Broker, mount API
+  keys as secret files and configure `API_KEY_FILE` or `API_KEYS_FILE`. The
+  Broker reloads those files on `API_KEY_RELOAD_INTERVAL`.
 
 ## Usage Secrets vs. Master Secrets
 

docs/services/broker.md

@@ -69,5 +69,9 @@ See [Health Checks Architecture](../healthchecks.md) for details on the monitori
 | `REDIS_URL` | Redis URL for caching discovery and state. | Required |
 | `ENCRYPTION_KEY` | 32-byte Base64 key for AES-GCM. | Required |
 | `STATE_KEY` | 32-byte Base64 key for signing state. Must match the Gateway. The Broker will **fatal-exit** on startup if absent. | Required |
-| `API_KEY` | Key for Gateway-to-Broker authentication. | Required |
+| `API_KEY` | Single key for Gateway-to-Broker authentication. | Conditional |
+| `API_KEYS` | Comma-separated Gateway-to-Broker keys for rollover. | Optional |
+| `API_KEY_FILE` | Mounted secret file containing one Broker API key; reloaded without restart. | Optional |
+| `API_KEYS_FILE` | Mounted secret file containing comma- or newline-separated Broker API keys; reloaded without restart. | Optional |
+| `API_KEY_RELOAD_INTERVAL` | File reload interval for API key secret files. | `30s` |
 

nexus-bridge/README.md

@@ -181,6 +181,8 @@ type Metrics interface {
 	IncConnections()
 	IncDisconnects()
 	IncTokenRefreshes()
+	ObserveConnectionDuration(duration time.Duration)
+	ObserveTokenRefreshLatency(duration time.Duration)
 	SetConnectionStatus(status float64)
 }
-```
+```

nexus-bridge/bridge.go

@@ -145,10 +145,12 @@ func (b *Bridge) MaintainGRPCConnection(
 		b.metrics.IncConnections()
 		b.metrics.SetConnectionStatus(1)
 		b.logger.Info("gRPC connection established", "target", target)
+		connectedAt := time.Now()
 
 		err = run(ctx, conn)
 
 		conn.Close()
+		b.metrics.ObserveConnectionDuration(time.Since(connectedAt))
 		b.metrics.SetConnectionStatus(0)
 		b.metrics.IncDisconnects()
 
@@ -193,10 +195,12 @@ func (b *Bridge) manageConnection(parentCtx context.Context, connectionID string
 		return err
 	}
 	defer conn.Close()
+	connectedAt := time.Now()
+	defer b.metrics.ObserveConnectionDuration(time.Since(connectedAt))
 
 	// --- Concurrency and Shutdown Management ---
-	done := make(chan struct{})       // Channel to signal shutdown to goroutines
-	defer close(done) // Ensure done is closed when manageConnection exits to clean up pumps
+	done := make(chan struct{}) // Channel to signal shutdown to goroutines
+	defer close(done)           // Ensure done is closed when manageConnection exits to clean up pumps
 
 	// Explicitly tie connection closure to context cancellation to prevent goroutine
 	// leaks if ReadMessage blocks indefinitely without a set read deadline.
@@ -224,7 +228,7 @@ func (b *Bridge) manageConnection(parentCtx context.Context, connectionID string
 	handler.OnConnect(sendFunc)
 
 	readErrChan := make(chan error, 1)
-	
+
 	b.startPumps(ctx, conn, handler, writeChan, readErrChan, done, connectionID)
 
 	return b.runEventLoop(ctx, connectionID, token, handler, readErrChan, done)
@@ -376,7 +380,9 @@ func (b *Bridge) runEventLoop(ctx context.Context, connectionID string, token *a
 			b.metrics.IncTokenRefreshes()
 			b.logger.Info("Starting background token refresh", "connectionID", connectionID)
 			go func() {
+				refreshStartedAt := time.Now()
 				refreshedToken, refreshErr := b.oauthClient.RefreshConnection(ctx, connectionID)
+				b.metrics.ObserveTokenRefreshLatency(time.Since(refreshStartedAt))
 				if refreshErr != nil {
 					select {
 					case refreshErrChan <- refreshErr:

nexus-bridge/bridge_test.go

@@ -64,15 +64,25 @@ func (h *mockHandler) OnDisconnect(err error) {
 
 // mockMetrics is a mock implementation of the Metrics interface for testing.
 type mockMetrics struct {
-	connections      int32
-	disconnects      int32
-	tokenRefreshes   int32
-	connectionStatus atomic.Value
+	connections           int32
+	disconnects           int32
+	tokenRefreshes        int32
+	connectionDurations   int32
+	tokenRefreshLatencies int32
+	connectionStatus      atomic.Value
 }
 
-func (m *mockMetrics) IncConnections()                    { atomic.AddInt32(&m.connections, 1) }
-func (m *mockMetrics) IncDisconnects()                    { atomic.AddInt32(&m.disconnects, 1) }
-func (m *mockMetrics) IncTokenRefreshes()                 { atomic.AddInt32(&m.tokenRefreshes, 1) }
+func (m *mockMetrics) IncConnections() { atomic.AddInt32(&m.connections, 1) }
+func (m *mockMetrics) IncDisconnects() { atomic.AddInt32(&m.disconnects, 1) }
+func (m *mockMetrics) IncTokenRefreshes() {
+	atomic.AddInt32(&m.tokenRefreshes, 1)
+}
+func (m *mockMetrics) ObserveConnectionDuration(time.Duration) {
+	atomic.AddInt32(&m.connectionDurations, 1)
+}
+func (m *mockMetrics) ObserveTokenRefreshLatency(time.Duration) {
+	atomic.AddInt32(&m.tokenRefreshLatencies, 1)
+}
 func (m *mockMetrics) SetConnectionStatus(status float64) { m.connectionStatus.Store(status) }
 
 // testLogger is a mock implementation of the Logger interface for testing.
@@ -90,6 +100,26 @@ func (l *testLogger) Error(err error, msg string, keysAndValues ...interface{})
 
 var upgrader = websocket.Upgrader{}
 
+func waitForAtomicCount(t *testing.T, value *int32, expected int32) {
+	t.Helper()
+
+	deadline := time.After(2 * time.Second)
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		if atomic.LoadInt32(value) == expected {
+			return
+		}
+
+		select {
+		case <-deadline:
+			t.Fatalf("timed out waiting for metric count %d, got %d", expected, atomic.LoadInt32(value))
+		case <-ticker.C:
+		}
+	}
+}
+
 // --- Tests ---
 
 func TestBridge_PermanentTokenError(t *testing.T) {
@@ -152,6 +182,9 @@ func TestBridge_PermanentCloseCode(t *testing.T) {
 	if atomic.LoadInt32(&metrics.disconnects) != 1 {
 		t.Errorf("Expected 1 disconnect, got %d", metrics.disconnects)
 	}
+	if atomic.LoadInt32(&metrics.connectionDurations) != 1 {
+		t.Errorf("Expected 1 connection duration observation, got %d", metrics.connectionDurations)
+	}
 }
 
 func TestBridge_ConnectionDropAndReconnect(t *testing.T) {
@@ -221,7 +254,9 @@ func TestBridge_ContextCancellation(t *testing.T) {
 		},
 	}
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		upgrader.Upgrade(w, r, nil)
+		conn, _ := upgrader.Upgrade(w, r, nil)
+		defer conn.Close()
+		<-r.Context().Done()
 	}))
 	defer server.Close()
 
@@ -250,6 +285,9 @@ func TestBridge_ContextCancellation(t *testing.T) {
 	if metrics.connectionStatus.Load() != 0.0 {
 		t.Errorf("Expected connection status to be 0, but got %v", metrics.connectionStatus.Load())
 	}
+	if atomic.LoadInt32(&metrics.connectionDurations) != 1 {
+		t.Errorf("Expected 1 connection duration observation, got %d", metrics.connectionDurations)
+	}
 }
 
 func TestBridge_HappyPath(t *testing.T) {
@@ -442,6 +480,7 @@ func TestBridge_TokenRefreshWithoutDisconnect(t *testing.T) {
 	if atomic.LoadInt32(&metrics.tokenRefreshes) != 1 {
 		t.Errorf("Expected 1 token refresh, got %d", metrics.tokenRefreshes)
 	}
+	waitForAtomicCount(t, &metrics.tokenRefreshLatencies, 1)
 	if metrics.connectionStatus.Load() != 1.0 {
 		t.Errorf("Expected connection status to be 1, but got %v", metrics.connectionStatus.Load())
 	}
@@ -484,6 +523,9 @@ func TestGRPC_CleanExit(t *testing.T) {
 	if atomic.LoadInt32(&metrics.connections) != 1 {
 		t.Errorf("expected 1 connection, got %d", metrics.connections)
 	}
+	if atomic.LoadInt32(&metrics.connectionDurations) != 1 {
+		t.Errorf("expected 1 connection duration observation, got %d", metrics.connectionDurations)
+	}
 }
 
 func TestGRPC_PermanentError(t *testing.T) {

nexus-bridge/options.go

@@ -19,6 +19,8 @@ type Metrics interface {
 	IncConnections()
 	IncDisconnects()
 	IncTokenRefreshes()
+	ObserveConnectionDuration(duration time.Duration)
+	ObserveTokenRefreshLatency(duration time.Duration)
 	SetConnectionStatus(status float64)
 }
 
@@ -31,9 +33,13 @@ func (l *nopLogger) Error(err error, msg string, keysAndValues ...interface{}) {
 
 type nopMetrics struct{}
 
-func (m *nopMetrics) IncConnections()                    {}
-func (m *nopMetrics) IncDisconnects()                    {}
-func (m *nopMetrics) IncTokenRefreshes()                 {}
+func (m *nopMetrics) IncConnections()    {}
+func (m *nopMetrics) IncDisconnects()    {}
+func (m *nopMetrics) IncTokenRefreshes() {}
+func (m *nopMetrics) ObserveConnectionDuration(time.Duration) {
+}
+func (m *nopMetrics) ObserveTokenRefreshLatency(time.Duration) {
+}
 func (m *nopMetrics) SetConnectionStatus(status float64) {}
 
 // --- Configuration ---

nexus-bridge/telemetry/metrics.go

@@ -1,15 +1,19 @@
 package telemetry
 
 import (
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 // PromMetrics implements the bridge.Metrics interface using Prometheus.
 type PromMetrics struct {
-	connections    prometheus.Counter
-	disconnects    prometheus.Counter
-	tokenRefreshes prometheus.Counter
-	connStatus     prometheus.Gauge
+	connections         prometheus.Counter
+	disconnects         prometheus.Counter
+	tokenRefreshes      prometheus.Counter
+	connStatus          prometheus.Gauge
+	connectionDuration  prometheus.Histogram
+	tokenRefreshLatency prometheus.Histogram
 }
 
 // NewMetrics creates and registers standard bridge metrics.
@@ -44,12 +48,28 @@ func NewMetrics(registry prometheus.Registerer, agentLabels map[string]string) *
 			Help:        "Current status of the connection (1 = connected, 0 = disconnected).",
 			ConstLabels: agentLabels,
 		}),
+		connectionDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace:   "bridge",
+			Name:        "connection_duration_seconds",
+			Help:        "Duration a bridge connection remained alive before disconnecting.",
+			ConstLabels: agentLabels,
+			Buckets:     []float64{1, 5, 15, 30, 60, 300, 900, 1800, 3600, 7200, 21600, 43200, 86400},
+		}),
+		tokenRefreshLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace:   "bridge",
+			Name:        "token_refresh_latency_seconds",
+			Help:        "Latency of token refresh operations.",
+			ConstLabels: agentLabels,
+			Buckets:     []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30},
+		}),
 	}
 
 	registry.MustRegister(m.connections)
 	registry.MustRegister(m.disconnects)
 	registry.MustRegister(m.tokenRefreshes)
 	registry.MustRegister(m.connStatus)
+	registry.MustRegister(m.connectionDuration)
+	registry.MustRegister(m.tokenRefreshLatency)
 
 	return m
 }
@@ -66,6 +86,14 @@ func (m *PromMetrics) IncTokenRefreshes() {
 	m.tokenRefreshes.Inc()
 }
 
+func (m *PromMetrics) ObserveConnectionDuration(duration time.Duration) {
+	m.connectionDuration.Observe(duration.Seconds())
+}
+
+func (m *PromMetrics) ObserveTokenRefreshLatency(duration time.Duration) {
+	m.tokenRefreshLatency.Observe(duration.Seconds())
+}
+
 func (m *PromMetrics) SetConnectionStatus(status float64) {
 	m.connStatus.Set(status)
 }

nexus-bridge/telemetry/metrics_test.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -75,3 +76,43 @@ func TestNewMetrics_NilLabels(t *testing.T) {
 		}
 	}
 }
+
+func TestNewMetrics_RegistersTimingHistograms(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	m := NewMetrics(registry, map[string]string{"agent_id": "test-agent"})
+
+	m.ObserveConnectionDuration(2 * time.Minute)
+	m.ObserveTokenRefreshLatency(1500 * time.Millisecond)
+
+	metricFamilies, err := registry.Gather()
+	if err != nil {
+		t.Fatalf("failed to gather metrics: %v", err)
+	}
+
+	assertHistogram := func(name string) {
+		t.Helper()
+
+		for _, mf := range metricFamilies {
+			if mf.GetName() != name {
+				continue
+			}
+
+			for _, metric := range mf.GetMetric() {
+				if metric.GetHistogram().GetSampleCount() != 1 {
+					t.Fatalf("%s expected sample count 1, got %d", name, metric.GetHistogram().GetSampleCount())
+				}
+
+				labels := metric.GetLabel()
+				if len(labels) != 1 || labels[0].GetName() != "agent_id" || labels[0].GetValue() != "test-agent" {
+					t.Fatalf("%s expected agent_id label, got %v", name, labels)
+				}
+			}
+			return
+		}
+
+		t.Fatalf("metric %s not found", name)
+	}
+
+	assertHistogram("bridge_connection_duration_seconds")
+	assertHistogram("bridge_token_refresh_latency_seconds")
+}