From 49380c132b0f3943bc86037844223bf3e7dd715e Mon Sep 17 00:00:00 2001 From: ashioyajotham Date: Mon, 20 Jul 2026 14:58:54 +0300 Subject: [PATCH] Add bridge metrics and broker key rotation --- .env.example | 5 + docs/concepts/broker.md | 4 +- docs/concepts/security-model.md | 2 + docs/getting-started/configuration.md | 6 +- docs/infrastructure/deploying-nexus.md | 8 +- docs/reference/security-model.md | 3 + docs/services/broker.md | 6 +- nexus-bridge/README.md | 4 +- nexus-bridge/bridge.go | 12 +- nexus-bridge/bridge_test.go | 58 +++++++-- nexus-bridge/options.go | 12 +- nexus-bridge/telemetry/metrics.go | 36 +++++- nexus-bridge/telemetry/metrics_test.go | 41 +++++++ nexus-broker/README.md | 10 +- nexus-broker/cmd/nexus-broker/main.go | 6 +- nexus-broker/pkg/config/broker.go | 34 +++++- nexus-broker/pkg/config/broker_test.go | 43 +++++++ nexus-broker/pkg/server/apikey.go | 157 ++++++++++++++++++++++++- nexus-broker/pkg/server/apikey_test.go | 86 ++++++++++++++ nexus-broker/pkg/vault/encrypt_test.go | 67 +++++++++++ 20 files changed, 569 insertions(+), 31 deletions(-) create mode 100644 nexus-broker/pkg/vault/encrypt_test.go diff --git a/.env.example b/.env.example index 066bee1..9be8921 100644 --- a/.env.example +++ b/.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 diff --git a/docs/concepts/broker.md b/docs/concepts/broker.md index 43677d2..8d8a5b5 100644 --- a/docs/concepts/broker.md +++ b/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. diff --git a/docs/concepts/security-model.md b/docs/concepts/security-model.md index df48d8f..b3d4c9e 100644 --- a/docs/concepts/security-model.md +++ b/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. diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index 05a39ff..1653406 100644 --- a/docs/getting-started/configuration.md +++ b/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` | diff --git a/docs/infrastructure/deploying-nexus.md b/docs/infrastructure/deploying-nexus.md index 7ebac70..2ce10e3 100644 --- a/docs/infrastructure/deploying-nexus.md +++ b/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. diff --git a/docs/reference/security-model.md b/docs/reference/security-model.md index 56ef7fd..aa0570b 100644 --- a/docs/reference/security-model.md +++ b/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 diff --git a/docs/services/broker.md b/docs/services/broker.md index d71c714..2a01d61 100644 --- a/docs/services/broker.md +++ b/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` | diff --git a/nexus-bridge/README.md b/nexus-bridge/README.md index e6a6427..24967ff 100644 --- a/nexus-bridge/README.md +++ b/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) } -``` \ No newline at end of file +``` diff --git a/nexus-bridge/bridge.go b/nexus-bridge/bridge.go index dc8bbd8..f5a6790 100644 --- a/nexus-bridge/bridge.go +++ b/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: diff --git a/nexus-bridge/bridge_test.go b/nexus-bridge/bridge_test.go index 6691499..d263d3e 100644 --- a/nexus-bridge/bridge_test.go +++ b/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) { diff --git a/nexus-bridge/options.go b/nexus-bridge/options.go index 1b71079..6c5a9b8 100644 --- a/nexus-bridge/options.go +++ b/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 --- diff --git a/nexus-bridge/telemetry/metrics.go b/nexus-bridge/telemetry/metrics.go index 81fdc8b..3790c14 100644 --- a/nexus-bridge/telemetry/metrics.go +++ b/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) } diff --git a/nexus-bridge/telemetry/metrics_test.go b/nexus-bridge/telemetry/metrics_test.go index 86dc20c..0f03479 100644 --- a/nexus-bridge/telemetry/metrics_test.go +++ b/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") +} diff --git a/nexus-broker/README.md b/nexus-broker/README.md index c5b838f..2e0564a 100644 --- a/nexus-broker/README.md +++ b/nexus-broker/README.md @@ -37,6 +37,9 @@ ENCRYPTION_KEY=<32-byte base64, stable> STATE_KEY=<32-byte base64, stable> REDIRECT_PATH=/auth/callback API_KEY=dev-api-key-12345 +API_KEY_FILE= +API_KEYS_FILE= +API_KEY_RELOAD_INTERVAL=30s ALLOWED_CIDRS=127.0.0.1/32,::1/128 ALLOWED_RETURN_DOMAINS=localhost,127.0.0.1,::1 PORT=8080 @@ -49,6 +52,11 @@ openssl rand -base64 32 # use output for STATE_KEY Keep ENCRYPTION_KEY and STATE_KEY constant; changing them breaks decrypting stored tokens. +For production API key rotation, mount a secret file and set `API_KEY_FILE` +or `API_KEYS_FILE`. `API_KEY_FILE` may contain one key, while `API_KEYS_FILE` +may contain comma- or newline-separated keys. The broker reloads these files at +`API_KEY_RELOAD_INTERVAL` without requiring a process restart. + ### 3) Run the broker ```bash source .env && go run ./cmd/nexus-broker @@ -275,7 +283,7 @@ OIDC hardening (id_token verification via JWKS, nonce, discovery) is fully imple ## Production Notes - Use managed Postgres (e.g., Azure Flexible Server). Set `sslmode=require`. - Restrict DB network (VNet/private DNS or firewall IPs). Restrict broker sensitive routes by IP. -- Keep keys constant; rotate API key; monitor `/metrics`. +- Keep encryption/state keys constant; rotate API keys through `API_KEY_FILE` or `API_KEYS_FILE`; monitor `/metrics`. - Document each provider in `docs/PROVIDERS.md` when added. --- diff --git a/nexus-broker/cmd/nexus-broker/main.go b/nexus-broker/cmd/nexus-broker/main.go index 50abf2f..522a76a 100644 --- a/nexus-broker/cmd/nexus-broker/main.go +++ b/nexus-broker/cmd/nexus-broker/main.go @@ -103,6 +103,10 @@ func main() { }) auditHandler := handlers.NewAuditHandler(db) connectionsHandler := handlers.NewConnectionsHandler(connSvc) + apiKeySource, err := server.NewReloadingAPIKeySource(cfg.APIKeys, cfg.APIKeyFiles, cfg.APIKeyReloadInterval) + if err != nil { + log.Fatalf("Fatal API key configuration error: %v", err) + } router := srv.Router() router.Get("/auth/callback", callbackHandler.Handle) @@ -111,7 +115,7 @@ func main() { router.Post("/auth/capture-credential", callbackHandler.SaveCredential) protected := router.With( - server.ApiKeyMiddleware(cfg.RequireAPIKey, cfg.APIKeys), + server.ApiKeySourceMiddleware(cfg.RequireAPIKey, apiKeySource), server.AllowlistMiddleware(cfg.RequireAllowlist, cfg.AllowedCIDRs), ) protected.Get("/audit", auditHandler.List) diff --git a/nexus-broker/pkg/config/broker.go b/nexus-broker/pkg/config/broker.go index f57013a..a6f5f74 100644 --- a/nexus-broker/pkg/config/broker.go +++ b/nexus-broker/pkg/config/broker.go @@ -5,6 +5,7 @@ import ( "net/url" "os" "strings" + "time" ) // BrokerConfig holds all configuration for the nexus-broker service. @@ -22,8 +23,10 @@ type BrokerConfig struct { RedirectPath string // API key protection - RequireAPIKey bool - APIKeys map[string]struct{} + RequireAPIKey bool + APIKeys map[string]struct{} + APIKeyFiles []string + APIKeyReloadInterval time.Duration // CIDR allowlist RequireAllowlist bool @@ -50,9 +53,10 @@ func Load() (*BrokerConfig, error) { RedirectPath: envOr("REDIRECT_PATH", "/auth/callback"), - RequireAPIKey: envBool("REQUIRE_API_KEY"), - RequireAllowlist: envBool("REQUIRE_ALLOWLIST"), - AllowedCIDRs: envOr("ALLOWED_CIDRS", "127.0.0.1/32,::1/128"), + RequireAPIKey: envBool("REQUIRE_API_KEY"), + RequireAllowlist: envBool("REQUIRE_ALLOWLIST"), + AllowedCIDRs: envOr("ALLOWED_CIDRS", "127.0.0.1/32,::1/128"), + APIKeyReloadInterval: 30 * time.Second, EnforceReturnURL: envBool("ENFORCE_RETURN_URL"), @@ -84,6 +88,26 @@ func Load() (*BrokerConfig, error) { if v := strings.TrimSpace(os.Getenv("API_KEY")); v != "" { cfg.APIKeys[v] = struct{}{} } + for _, envName := range []string{"API_KEY_FILE", "API_KEYS_FILE"} { + if v := strings.TrimSpace(os.Getenv(envName)); v != "" { + for _, path := range strings.Split(v, ",") { + path = strings.TrimSpace(path) + if path != "" { + cfg.APIKeyFiles = append(cfg.APIKeyFiles, path) + } + } + } + } + if v := strings.TrimSpace(os.Getenv("API_KEY_RELOAD_INTERVAL")); v != "" { + interval, err := time.ParseDuration(v) + if err != nil { + return nil, fmt.Errorf("API_KEY_RELOAD_INTERVAL is invalid: %w", err) + } + if interval <= 0 { + return nil, fmt.Errorf("API_KEY_RELOAD_INTERVAL must be greater than zero") + } + cfg.APIKeyReloadInterval = interval + } // Required fields if cfg.DatabaseURL == "" { diff --git a/nexus-broker/pkg/config/broker_test.go b/nexus-broker/pkg/config/broker_test.go index 4fb576c..09774d4 100644 --- a/nexus-broker/pkg/config/broker_test.go +++ b/nexus-broker/pkg/config/broker_test.go @@ -2,7 +2,9 @@ package config import ( "encoding/base64" + "path/filepath" "testing" + "time" ) func testKey() string { @@ -78,6 +80,47 @@ func TestLoad_Success(t *testing.T) { } } +func TestLoad_APIKeyFilesAndReloadInterval(t *testing.T) { + apiKeyFile := filepath.Join(t.TempDir(), "api-key") + t.Setenv("DATABASE_URL", "postgres://localhost/db") + t.Setenv("BASE_URL", "http://localhost:8080") + t.Setenv("ENCRYPTION_KEY", testKey()) + t.Setenv("STATE_KEY", testKey()) + t.Setenv("API_KEY_FILE", apiKeyFile) + t.Setenv("API_KEYS_FILE", " /var/run/secrets/api-keys ") + t.Setenv("API_KEY_RELOAD_INTERVAL", "5s") + + cfg, err := Load() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.APIKeyReloadInterval != 5*time.Second { + t.Errorf("expected reload interval 5s, got %v", cfg.APIKeyReloadInterval) + } + if len(cfg.APIKeyFiles) != 2 { + t.Fatalf("expected 2 api key files, got %d", len(cfg.APIKeyFiles)) + } + if cfg.APIKeyFiles[0] != apiKeyFile { + t.Errorf("expected first api key file %q, got %q", apiKeyFile, cfg.APIKeyFiles[0]) + } + if cfg.APIKeyFiles[1] != "/var/run/secrets/api-keys" { + t.Errorf("expected trimmed second api key file, got %q", cfg.APIKeyFiles[1]) + } +} + +func TestLoad_InvalidAPIKeyReloadInterval(t *testing.T) { + t.Setenv("DATABASE_URL", "postgres://localhost/db") + t.Setenv("BASE_URL", "http://localhost:8080") + t.Setenv("ENCRYPTION_KEY", testKey()) + t.Setenv("STATE_KEY", testKey()) + t.Setenv("API_KEY_RELOAD_INTERVAL", "0s") + + _, err := Load() + if err == nil { + t.Fatal("expected invalid reload interval error") + } +} + func TestLoad_DBSSLEnforcement(t *testing.T) { t.Setenv("DATABASE_URL", "postgres://localhost/db") t.Setenv("BASE_URL", "http://localhost") diff --git a/nexus-broker/pkg/server/apikey.go b/nexus-broker/pkg/server/apikey.go index 7288a87..49aa990 100644 --- a/nexus-broker/pkg/server/apikey.go +++ b/nexus-broker/pkg/server/apikey.go @@ -1,14 +1,140 @@ package server import ( + "fmt" "net/http" + "os" "strings" + "sync" + "time" "github.com/Prescott-Data/nexus-framework/nexus-broker/pkg/httputil" ) +// APIKeySource checks whether an API key is currently allowed. +type APIKeySource interface { + Contains(key string) bool +} + +type staticAPIKeySource struct { + keys map[string]struct{} +} + +// NewStaticAPIKeySource creates an immutable API key source from an allow-set. +func NewStaticAPIKeySource(allowedKeys map[string]struct{}) APIKeySource { + return &staticAPIKeySource{keys: copyAPIKeys(allowedKeys)} +} + +func (s *staticAPIKeySource) Contains(key string) bool { + _, ok := s.keys[key] + return ok +} + +// ReloadingAPIKeySource keeps inline keys and key files in a single allow-set. +// File contents are re-read at most once per reload interval, enabling mounted +// secret rotation without restarting the broker process. +type ReloadingAPIKeySource struct { + inlineKeys map[string]struct{} + files []string + reloadInterval time.Duration + now func() time.Time + + mu sync.RWMutex + keys map[string]struct{} + lastReload time.Time +} + +// NewReloadingAPIKeySource creates a reloadable source and performs an initial +// load so configuration errors fail during startup instead of on first request. +func NewReloadingAPIKeySource( + inlineKeys map[string]struct{}, + files []string, + reloadInterval time.Duration, +) (*ReloadingAPIKeySource, error) { + if reloadInterval <= 0 { + return nil, fmt.Errorf("api key reload interval must be greater than zero") + } + + source := &ReloadingAPIKeySource{ + inlineKeys: copyAPIKeys(inlineKeys), + files: copyStrings(files), + reloadInterval: reloadInterval, + now: time.Now, + } + if err := source.Reload(); err != nil { + return nil, err + } + return source, nil +} + +// Reload immediately re-reads configured files and swaps in the new allow-set. +func (s *ReloadingAPIKeySource) Reload() error { + next, err := s.load() + if err != nil { + return err + } + + s.mu.Lock() + defer s.mu.Unlock() + s.keys = next + s.lastReload = s.now() + return nil +} + +func (s *ReloadingAPIKeySource) Contains(key string) bool { + s.reloadIfDue() + + s.mu.RLock() + defer s.mu.RUnlock() + _, ok := s.keys[key] + return ok +} + +func (s *ReloadingAPIKeySource) reloadIfDue() { + now := s.now() + + s.mu.RLock() + reloadDue := now.Sub(s.lastReload) >= s.reloadInterval + s.mu.RUnlock() + if !reloadDue { + return + } + + s.mu.Lock() + defer s.mu.Unlock() + if now.Sub(s.lastReload) < s.reloadInterval { + return + } + + next, err := s.load() + s.lastReload = now + if err != nil { + return + } + s.keys = next +} + +func (s *ReloadingAPIKeySource) load() (map[string]struct{}, error) { + keys := copyAPIKeys(s.inlineKeys) + for _, path := range s.files { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("load api key file %s: %w", path, err) + } + for _, key := range parseAPIKeys(string(raw)) { + keys[key] = struct{}{} + } + } + return keys, nil +} + // ApiKeyMiddleware enforces X-API-Key header when requireKey is true. func ApiKeyMiddleware(requireKey bool, allowedKeys map[string]struct{}) func(http.Handler) http.Handler { + return ApiKeySourceMiddleware(requireKey, NewStaticAPIKeySource(allowedKeys)) +} + +// ApiKeySourceMiddleware enforces X-API-Key header using a dynamic key source. +func ApiKeySourceMiddleware(requireKey bool, allowedKeys APIKeySource) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !requireKey { @@ -20,7 +146,7 @@ func ApiKeyMiddleware(requireKey bool, allowedKeys map[string]struct{}) func(htt httputil.WriteError(w, http.StatusUnauthorized, "missing_api_key", "missing api key") return } - if _, ok := allowedKeys[key]; !ok { + if allowedKeys == nil || !allowedKeys.Contains(key) { httputil.WriteError(w, http.StatusForbidden, "invalid_api_key", "invalid api key") return } @@ -28,3 +154,32 @@ func ApiKeyMiddleware(requireKey bool, allowedKeys map[string]struct{}) func(htt }) } } + +func parseAPIKeys(raw string) []string { + parts := strings.FieldsFunc(raw, func(r rune) bool { + return r == ',' || r == '\n' || r == '\r' || r == '\t' + }) + + keys := make([]string, 0, len(parts)) + for _, part := range parts { + key := strings.TrimSpace(part) + if key != "" { + keys = append(keys, key) + } + } + return keys +} + +func copyAPIKeys(keys map[string]struct{}) map[string]struct{} { + copied := make(map[string]struct{}, len(keys)) + for key := range keys { + copied[key] = struct{}{} + } + return copied +} + +func copyStrings(values []string) []string { + copied := make([]string, len(values)) + copy(copied, values) + return copied +} diff --git a/nexus-broker/pkg/server/apikey_test.go b/nexus-broker/pkg/server/apikey_test.go index 363b001..24f3044 100644 --- a/nexus-broker/pkg/server/apikey_test.go +++ b/nexus-broker/pkg/server/apikey_test.go @@ -3,7 +3,10 @@ package server import ( "net/http" "net/http/httptest" + "os" + "path/filepath" "testing" + "time" ) func TestApiKeyMiddleware(t *testing.T) { @@ -70,3 +73,86 @@ func TestApiKeyMiddleware(t *testing.T) { }) } } + +func TestApiKeySourceMiddlewareReloadsFileKeys(t *testing.T) { + keyFile := filepath.Join(t.TempDir(), "api-keys") + if err := os.WriteFile(keyFile, []byte("first-key\n"), 0600); err != nil { + t.Fatalf("failed to write key file: %v", err) + } + + source, err := NewReloadingAPIKeySource(nil, []string{keyFile}, time.Second) + if err != nil { + t.Fatalf("NewReloadingAPIKeySource returned error: %v", err) + } + + now := source.lastReload + source.now = func() time.Time { return now } + + handler := ApiKeySourceMiddleware(true, source)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + + if status := requestWithAPIKey(handler, "first-key"); status != http.StatusOK { + t.Fatalf("expected first key to be accepted, got %d", status) + } + + if err := os.WriteFile(keyFile, []byte("second-key\n"), 0600); err != nil { + t.Fatalf("failed to rotate key file: %v", err) + } + + if status := requestWithAPIKey(handler, "second-key"); status != http.StatusForbidden { + t.Fatalf("expected second key to be rejected before reload interval, got %d", status) + } + + now = now.Add(2 * time.Second) + + if status := requestWithAPIKey(handler, "second-key"); status != http.StatusOK { + t.Fatalf("expected second key to be accepted after reload, got %d", status) + } + if status := requestWithAPIKey(handler, "first-key"); status != http.StatusForbidden { + t.Fatalf("expected first key to be rejected after reload, got %d", status) + } +} + +func TestReloadingAPIKeySourceKeepsLastKeysWhenReloadFails(t *testing.T) { + keyFile := filepath.Join(t.TempDir(), "api-keys") + if err := os.WriteFile(keyFile, []byte("stable-key\n"), 0600); err != nil { + t.Fatalf("failed to write key file: %v", err) + } + + source, err := NewReloadingAPIKeySource(nil, []string{keyFile}, time.Second) + if err != nil { + t.Fatalf("NewReloadingAPIKeySource returned error: %v", err) + } + + now := source.lastReload + source.now = func() time.Time { return now } + + if err := os.Remove(keyFile); err != nil { + t.Fatalf("failed to remove key file: %v", err) + } + now = now.Add(2 * time.Second) + + if !source.Contains("stable-key") { + t.Fatal("expected last known-good key to remain valid after reload failure") + } +} +func TestNewReloadingAPIKeySourceFailsForMissingFile(t *testing.T) { + missingFile := filepath.Join(t.TempDir(), "missing-api-keys") + + _, err := NewReloadingAPIKeySource(nil, []string{missingFile}, time.Second) + if err == nil { + t.Fatal("expected missing key file to fail initial load") + } +} + +func requestWithAPIKey(handler http.Handler, key string) int { + req := httptest.NewRequest("GET", "/", nil) + if key != "" { + req.Header.Set("X-API-Key", key) + } + + rr := httptest.NewRecorder() + handler.ServeHTTP(rr, req) + return rr.Code +} diff --git a/nexus-broker/pkg/vault/encrypt_test.go b/nexus-broker/pkg/vault/encrypt_test.go new file mode 100644 index 0000000..fca7291 --- /dev/null +++ b/nexus-broker/pkg/vault/encrypt_test.go @@ -0,0 +1,67 @@ +package vault + +import ( + "encoding/base64" + "testing" +) + +func TestEncryptDecryptRoundTrip(t *testing.T) { + key := make([]byte, 32) + plaintext := []byte(`{"access_token":"token-123","refresh_token":"refresh-456"}`) + + ciphertext, err := Encrypt(key, plaintext) + if err != nil { + t.Fatalf("Encrypt returned error: %v", err) + } + + decrypted, err := Decrypt(key, ciphertext) + if err != nil { + t.Fatalf("Decrypt returned error: %v", err) + } + if string(decrypted) != string(plaintext) { + t.Fatalf("expected plaintext %q, got %q", plaintext, decrypted) + } +} + +func TestEncryptUsesUniqueNonce(t *testing.T) { + key := make([]byte, 32) + plaintext := []byte("same token payload") + + first, err := Encrypt(key, plaintext) + if err != nil { + t.Fatalf("first Encrypt returned error: %v", err) + } + second, err := Encrypt(key, plaintext) + if err != nil { + t.Fatalf("second Encrypt returned error: %v", err) + } + + if first == second { + t.Fatal("expected different ciphertexts for identical plaintext") + } +} + +func TestDecryptRejectsTamperedCiphertext(t *testing.T) { + key := make([]byte, 32) + ciphertext, err := Encrypt(key, []byte("token payload")) + if err != nil { + t.Fatalf("Encrypt returned error: %v", err) + } + + data, err := base64.StdEncoding.DecodeString(ciphertext) + if err != nil { + t.Fatalf("ciphertext is not valid base64: %v", err) + } + data[len(data)-1] ^= 0x01 + tampered := base64.StdEncoding.EncodeToString(data) + + if _, err := Decrypt(key, tampered); err == nil { + t.Fatal("expected tampered ciphertext to fail authentication") + } +} + +func TestEncryptRejectsInvalidKeyLength(t *testing.T) { + if _, err := Encrypt(make([]byte, 31), []byte("token payload")); err == nil { + t.Fatal("expected invalid key length error") + } +}