feat(multicluster): URL-path routing across N ClickHouse clusters (#132)

README.md

@@ -7,7 +7,7 @@ A Model Context Protocol (MCP) server that provides tools for interacting with C
 ## Features
 
 - **Multiple Transport Options**: Support for STDIO, HTTP, and Server-Sent Events (SSE) transports
-- **OAuth 2.0 Authorization**: Forward Bearer tokens to ClickHouse for token-based authentication via OIDC providers (see [OAuth 2.0 Documentation](docs/oauth_authorization.md))
+- **OAuth 2.0 Authorization**: Broker OAuth flows from MCP clients against any OIDC provider; auto-detects Bearer vs Basic auth for ClickHouse (see [OAuth 2.0 Documentation](docs/oauth_authorization.md))
 - **JWE Authentication**: Optional JWE-based authentication with encryption for secure database access
 - **TLS Support**: Full TLS encryption support for both ClickHouse® connections and MCP server endpoints
 - **Comprehensive Tools**: Built-in tools for listing tables, describing schemas, and executing queries
@@ -236,7 +236,7 @@ export LOG_LEVEL=debug
 # OAuth — env-var injection lets operators pull secrets from a Kubernetes
 # Secret via `valueFrom.secretKeyRef` instead of committing them to YAML.
 export MCP_OAUTH_ENABLED=true
-export MCP_OAUTH_MODE=gating
+export MCP_OAUTH_BROKER=true
 export MCP_OAUTH_ISSUER=https://accounts.example.com
 export MCP_OAUTH_SIGNING_SECRET=...
 
@@ -357,7 +357,7 @@ JWE takes priority — if present and valid and has valid credentials, use it an
 
 ### OAuth 2.0 Authorization
 
-The MCP server supports OAuth 2.0/OpenID Connect authentication, with token forwarding to ClickHouse or token verification locally. This enables MCP clients to authenticate via an Identity Provider (Keycloak, Azure AD, Google, AWS Cognito) and have their Bearer tokens forwarded to ClickHouse for token-based authentication via `token_processors`.
+The MCP server supports OAuth 2.0/OpenID Connect authentication, with Bearer tokens presented to ClickHouse or verified locally. This enables MCP clients to authenticate via an Identity Provider (Keycloak, Azure AD, Google, AWS Cognito) and use token-based ClickHouse authentication via `token_processors`.
 
 For full setup instructions, provider-specific guides, and ClickHouse configuration, see the [OAuth 2.0 Authorization Documentation](docs/oauth_authorization.md).
 

cmd/altinity-mcp/client_assertion_test.go

@@ -360,7 +360,7 @@ func TestHandleOAuthTokenAuthCode_LenientPrivateKeyJWT(t *testing.T) {
 	// helper directly. parseCIMDMetadata only fails on bad shape, so
 	// reaching here proves the parser accepts both methods (covered);
 	// the lenient runtime branch is exercised by integration tests in
-	// oauth_regression_test.go via the broker_upstream flow. This unit
+	// oauth_regression_test.go via the broker flow. This unit
 	// test guards the parser-side accept of "private_key_jwt" against a
 	// future revert to strict-mode-only.
 }

cmd/altinity-mcp/main_test.go

@@ -364,6 +364,26 @@ func (m *mockCommand) IsSet(name string) bool {
 	return m.setFlags[name]
 }
 
+func TestBuildConfigAppliesMulticlusterDefaultsAfterFlags(t *testing.T) {
+	t.Parallel()
+	cmd := &mockCommand{
+		flags: map[string]interface{}{
+			"multicluster-enabled": true,
+		},
+		setFlags: map[string]bool{
+			"multicluster-enabled": true,
+		},
+		stringMaps: make(map[string]map[string]string),
+	}
+
+	cfg, err := buildConfig(cmd)
+	require.NoError(t, err)
+	require.True(t, cfg.Multicluster.Enabled)
+	require.Equal(t, 10000, cfg.Multicluster.CatalogCacheMax)
+	require.Equal(t, 15*time.Minute, cfg.Multicluster.CatalogTTLFallback)
+	require.Equal(t, 60*time.Second, cfg.Multicluster.CatalogNegativeTTL)
+}
+
 // TestBuildServerTLSConfig tests server TLS configuration building
 func TestBuildServerTLSConfig(t *testing.T) {
 	t.Parallel()
@@ -3228,197 +3248,166 @@ func generateSelfSignedCert() ([]byte, []byte, error) {
 func TestValidateOAuthRuntimeConfig(t *testing.T) {
 	t.Parallel()
 
+	resourceServerBase := func() config.OAuthConfig {
+		return config.OAuthConfig{
+			Enabled:  true,
+			Issuer:   "https://altinity.auth0.com/",
+			Audience: "https://example-mcp.test/",
+		}
+	}
+	brokerBase := func() config.OAuthConfig {
+		return config.OAuthConfig{
+			Enabled:       true,
+			Broker:        true,
+			SigningSecret: "test-signing-secret-32-byte-key!!",
+			Issuer:        "https://accounts.google.com",
+			Audience:      "some-google-client-id.apps.googleusercontent.com",
+			ClientID:      "some-google-client-id.apps.googleusercontent.com",
+			ClientSecret:  "GOCSPX-redacted",
+			AuthURL:       "https://accounts.google.com/o/oauth2/v2/auth",
+			TokenURL:      "https://oauth2.googleapis.com/token",
+		}
+	}
+
 	t.Run("disabled_returns_nil", func(t *testing.T) {
 		t.Parallel()
 		cfg := config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{Enabled: false}}}
 		require.NoError(t, validateOAuthRuntimeConfig(cfg))
 	})
 
-	t.Run("unsupported_mode", func(t *testing.T) {
+	t.Run("resource_server_requires_http", func(t *testing.T) {
 		t.Parallel()
-		cfg := config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{
-			Enabled:       true,
-			Mode:          "custom",
-			SigningSecret: "test-signing-secret-32-byte-key!!",
-		}}}
+		cfg := config.Config{
+			Server:     config.ServerConfig{OAuth: resourceServerBase()},
+			ClickHouse: config.ClickHouseConfig{Protocol: config.TCPProtocol},
+		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.Error(t, err)
-		require.Contains(t, err.Error(), "unsupported oauth mode")
+		require.ErrorContains(t, err, "requires clickhouse protocol http")
 	})
 
-	t.Run("short_signing_secret_rejected", func(t *testing.T) {
+	t.Run("valid_resource_server_config", func(t *testing.T) {
 		t.Parallel()
-		cfg := config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{
-			Enabled:       true,
-			Mode:          "gating",
-			SigningSecret: "too-short",
-		}}}
-		err := validateOAuthRuntimeConfig(cfg)
-		require.Error(t, err)
-		require.Contains(t, err.Error(), "at least 32 bytes")
-	})
-
-	t.Run("missing_gating_secret", func(t *testing.T) {
-		t.Parallel()
-		cfg := config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{
-			Enabled:       true,
-			Mode:          "gating",
-			SigningSecret: "",
-		}}}
-		err := validateOAuthRuntimeConfig(cfg)
-		require.Error(t, err)
-		require.Contains(t, err.Error(), "signing_secret is required")
+		cfg := config.Config{
+			Server:     config.ServerConfig{OAuth: resourceServerBase()},
+			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
+		}
+		require.NoError(t, validateOAuthRuntimeConfig(cfg))
 	})
 
-	t.Run("forward_mode_requires_http", func(t *testing.T) {
+	t.Run("resource_server_requires_issuer", func(t *testing.T) {
 		t.Parallel()
+		o := resourceServerBase()
+		o.Issuer = ""
 		cfg := config.Config{
-			Server: config.ServerConfig{OAuth: config.OAuthConfig{
-				Enabled:       true,
-				Mode:          "forward",
-				SigningSecret: "test-signing-secret-32-byte-key!!",
-			}},
-			ClickHouse: config.ClickHouseConfig{Protocol: config.TCPProtocol},
+			Server:     config.ServerConfig{OAuth: o},
+			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.Error(t, err)
-		require.Contains(t, err.Error(), "requires clickhouse protocol http")
+		require.ErrorContains(t, err, "broker=false requires oauth.issuer")
 	})
 
-	t.Run("valid_gating_config", func(t *testing.T) {
+	t.Run("resource_server_requires_audience", func(t *testing.T) {
 		t.Parallel()
+		o := resourceServerBase()
+		o.Audience = ""
 		cfg := config.Config{
-			Server: config.ServerConfig{OAuth: config.OAuthConfig{
-				Enabled:       true,
-				Mode:          "gating",
-				SigningSecret: "test-signing-secret-32-byte-key!!",
-				Issuer:        "https://example.auth0.com/",
-				Audience:      "https://example-mcp.test/",
-			}},
+			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
-		require.NoError(t, validateOAuthRuntimeConfig(cfg))
+		err := validateOAuthRuntimeConfig(cfg)
+		require.ErrorContains(t, err, "broker=false requires oauth.audience")
 	})
 
-	t.Run("valid_forward_config", func(t *testing.T) {
+	t.Run("broker_valid_full_config", func(t *testing.T) {
 		t.Parallel()
 		cfg := config.Config{
-			Server: config.ServerConfig{OAuth: config.OAuthConfig{
-				Enabled:       true,
-				Mode:          "forward",
-				SigningSecret: "test-signing-secret-32-byte-key!!",
-			}},
+			Server:     config.ServerConfig{OAuth: brokerBase()},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		require.NoError(t, validateOAuthRuntimeConfig(cfg))
 	})
 
-	// Gating-mode with broker_upstream: opt-in DCR-via-MCP hybrid. The pure-
-	// resource-server gating shape stays the default; broker shape unlocks
-	// the AS handlers and requires the upstream-IdP fields to be present.
-	gatingBrokerBase := func() config.OAuthConfig {
-		return config.OAuthConfig{
-			Enabled:        true,
-			Mode:           "gating",
-			SigningSecret:  "test-signing-secret-32-byte-key!!",
-			Issuer:         "https://accounts.google.com",
-			Audience:       "some-google-client-id.apps.googleusercontent.com",
-			BrokerUpstream: true,
-			ClientID:       "some-google-client-id.apps.googleusercontent.com",
-			ClientSecret:   "GOCSPX-redacted",
-			AuthURL:        "https://accounts.google.com/o/oauth2/v2/auth",
-			TokenURL:       "https://oauth2.googleapis.com/token",
-			UserInfoURL:    "https://openidconnect.googleapis.com/v1/userinfo",
-		}
-	}
-
-	t.Run("gating_broker_valid_full_config", func(t *testing.T) {
+	t.Run("broker_requires_http", func(t *testing.T) {
 		t.Parallel()
 		cfg := config.Config{
-			Server:     config.ServerConfig{OAuth: gatingBrokerBase()},
-			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
+			Server:     config.ServerConfig{OAuth: brokerBase()},
+			ClickHouse: config.ClickHouseConfig{Protocol: config.TCPProtocol},
 		}
-		require.NoError(t, validateOAuthRuntimeConfig(cfg))
+		err := validateOAuthRuntimeConfig(cfg)
+		require.ErrorContains(t, err, "requires clickhouse protocol http")
 	})
 
-	t.Run("gating_broker_missing_client_id_rejected", func(t *testing.T) {
+	t.Run("broker_missing_signing_secret_rejected", func(t *testing.T) {
 		t.Parallel()
-		o := gatingBrokerBase()
-		o.ClientID = ""
+		o := brokerBase()
+		o.SigningSecret = ""
 		cfg := config.Config{
 			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "broker_upstream=true requires upstream-IdP fields")
-		require.ErrorContains(t, err, "client_id")
+		require.ErrorContains(t, err, "signing_secret is required")
 	})
 
-	t.Run("gating_broker_missing_client_secret_rejected", func(t *testing.T) {
+	t.Run("broker_short_signing_secret_rejected", func(t *testing.T) {
 		t.Parallel()
-		o := gatingBrokerBase()
-		o.ClientSecret = ""
+		o := brokerBase()
+		o.SigningSecret = "too-short"
 		cfg := config.Config{
 			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "client_secret")
+		require.ErrorContains(t, err, "at least 32 bytes")
 	})
 
-	t.Run("gating_broker_missing_auth_url_rejected", func(t *testing.T) {
+	t.Run("broker_missing_client_id_rejected", func(t *testing.T) {
 		t.Parallel()
-		o := gatingBrokerBase()
-		o.AuthURL = ""
+		o := brokerBase()
+		o.ClientID = ""
 		cfg := config.Config{
 			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "auth_url")
+		require.ErrorContains(t, err, "broker=true requires")
+		require.ErrorContains(t, err, "client_id")
 	})
 
-	t.Run("gating_broker_missing_token_url_rejected", func(t *testing.T) {
+	t.Run("broker_missing_client_secret_rejected", func(t *testing.T) {
 		t.Parallel()
-		o := gatingBrokerBase()
-		o.TokenURL = ""
+		o := brokerBase()
+		o.ClientSecret = ""
 		cfg := config.Config{
 			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "token_url")
+		require.ErrorContains(t, err, "client_secret")
 	})
 
-	t.Run("gating_non_broker_still_forbids_client_id", func(t *testing.T) {
+	t.Run("broker_missing_auth_url_rejected", func(t *testing.T) {
 		t.Parallel()
-		// Default gating (broker_upstream=false) must still reject the
-		// upstream-IdP fields per #109; broker_upstream is an explicit opt-in.
+		o := brokerBase()
+		o.AuthURL = ""
 		cfg := config.Config{
-			Server: config.ServerConfig{OAuth: config.OAuthConfig{
-				Enabled:       true,
-				Mode:          "gating",
-				SigningSecret: "test-signing-secret-32-byte-key!!",
-				Issuer:        "https://altinity.auth0.com/",
-				Audience:      "https://example-mcp.test/",
-				ClientID:      "some-client",
-			}},
+			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "without broker_upstream")
+		require.ErrorContains(t, err, "auth_url")
 	})
 
-	t.Run("gating_broker_still_requires_audience", func(t *testing.T) {
+	t.Run("broker_missing_token_url_rejected", func(t *testing.T) {
 		t.Parallel()
-		o := gatingBrokerBase()
-		o.Audience = ""
+		o := brokerBase()
+		o.TokenURL = ""
 		cfg := config.Config{
 			Server:     config.ServerConfig{OAuth: o},
 			ClickHouse: config.ClickHouseConfig{Protocol: config.HTTPProtocol},
 		}
 		err := validateOAuthRuntimeConfig(cfg)
-		require.ErrorContains(t, err, "oauth.audience")
+		require.ErrorContains(t, err, "token_url")
 	})
 }
 
@@ -3430,28 +3419,6 @@ func TestWarnOAuthMisconfiguration(t *testing.T) {
 		// Should not panic
 		warnOAuthMisconfiguration(config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{Enabled: false}}})
 	})
-
-	t.Run("gating_mode_missing_public_auth_server_url", func(t *testing.T) {
-		t.Parallel()
-		// Should log warning but not panic
-		warnOAuthMisconfiguration(config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{
-			Enabled:             true,
-			Mode:                "gating",
-			Issuer:              "https://issuer.example.com",
-			PublicAuthServerURL: "",
-		}}})
-	})
-
-	t.Run("gating_mode_with_public_auth_server_url", func(t *testing.T) {
-		t.Parallel()
-		// Should not warn
-		warnOAuthMisconfiguration(config.Config{Server: config.ServerConfig{OAuth: config.OAuthConfig{
-			Enabled:             true,
-			Mode:                "gating",
-			Issuer:              "https://issuer.example.com",
-			PublicAuthServerURL: "https://public.example.com",
-		}}})
-	})
 }
 
 func TestTransportRoutePatterns(t *testing.T) {
@@ -3698,11 +3665,11 @@ func TestLivenessHandler(t *testing.T) {
 	})
 }
 
-func TestHealthHandler_OAuthForwardMode(t *testing.T) {
+func TestHealthHandler_OAuth(t *testing.T) {
 	t.Parallel()
 	cfg := config.Config{
 		Server: config.ServerConfig{
-			OAuth: config.OAuthConfig{Enabled: true, Mode: "forward"},
+			OAuth: config.OAuthConfig{Enabled: true},
 		},
 	}
 	app := &application{