Isolate authenticated single-cluster catalogs and sessions

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Unreleased
 
 BUG FIXES
+- security: authenticate the exact `/openapi` schema route, reject authenticated database requests without usable tokens, and isolate single-cluster HTTP/SSE dynamic tools and schemas by effective request credentials; preserve sessions on unchanged reload polls, safely update logging levels during requests, and reject SSE messages from other credentials or retired generations (fixes [#182](https://github.com/Altinity/altinity-mcp/issues/182))
 - security: require self-contained JWE ClickHouse connections with host and username; prevent inheritance of operator endpoints, credentials, headers, roles, and TLS material; restrict token TLS file paths to `server.jwe.tls_material_dir` (fixes [#179](https://github.com/Altinity/altinity-mcp/issues/179)). Missing request tokens now fail closed in JWE-only and combined JWE/OAuth client paths, including discovery, instead of using static operator credentials. Regenerate tokens that relied on static connection defaults.
 - security: OpenAPI `execute_query` always rejects write statements and enforces `max_query_length`; dynamic-tool JSON bodies use the same input-size limit (fixes [#183](https://github.com/Altinity/altinity-mcp/issues/183))
 - security: isolate per-request ClickHouse headers, settings, and roles; prevent concurrent OAuth bearer forwarding from crashing the server, confusing caller identities, or leaking a stale bearer into later JWE or static-credential requests (fixes [#181](https://github.com/Altinity/altinity-mcp/issues/181))

README.md

@@ -342,10 +342,14 @@ OpenAI MCP/tooling references:
 
 ### Authentication
 - **With JWE**: Add the JWE token to either:
-  1. Path parameter: `/{jwe_token}/openapi/...` (now required)
+  1. Path parameter: `/{jwe_token}/openapi/...`
   2. Authorization header: `Bearer {token}` (alternative)
   3. `x-altinity-mcp-key` header (alternative)
-- **Without JWE**: Use server-configured credentials (no auth needed in requests)
+- **With OAuth**: Send `Authorization: Bearer {token}`.
+- **With both**: A self-contained JWE takes priority; otherwise OAuth supplies the credentials.
+- **With both disabled**: Use server-configured ClickHouse credentials.
+
+The exact `/openapi` schema route also requires authentication when either mode is enabled. Dynamic HTTP/SSE tools and OpenAPI schemas use catalogs isolated by the caller's effective credential; server credentials are never a fallback under authentication. An effective configuration change resets catalog generations and requires existing SSE clients to reconnect. Unchanged reload polls preserve sessions. See [dynamic discovery](docs/tools.md#dynamic-discovery).
 
 ### Available Actions
 

cmd/altinity-mcp/credential_transport.go

@@ -0,0 +1,227 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+
+	"github.com/altinity/altinity-mcp/pkg/config"
+	altinitymcp "github.com/altinity/altinity-mcp/pkg/server"
+	"github.com/modelcontextprotocol/go-sdk/mcp"
+	"github.com/rs/zerolog/log"
+)
+
+// withServerSnapshot captures the immutable parent and its converted tool rules
+// together. Authentication, discovery and execution all use this generation.
+func (a *application) withServerSnapshot(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		a.configMutex.RLock()
+		parent := a.mcpServer
+		a.configMutex.RUnlock()
+		if parent == nil {
+			http.Error(w, "Server unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		ctx := context.WithValue(r.Context(), altinitymcp.CHJWEServerKey, parent)
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}
+
+func (a *application) authenticateSnapshot(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		parent := altinitymcp.GetClickHouseJWEServerFromContext(r.Context())
+		cfg := parent.Config
+		// Reuse OAuth challenges and expiry checks with a request-scoped adapter;
+		// its auth injector cannot accidentally read a later global parent.
+		if cfg.Server.OAuth.Enabled {
+			adapter := &application{config: cfg, mcpServer: parent}
+			adapter.createMCPAuthInjector(cfg)(next).ServeHTTP(w, r)
+			return
+		}
+		if cfg.Server.JWE.Enabled {
+			token, claims, _, _, err := parent.ValidateAuth(r)
+			if errors.Is(err, altinitymcp.ErrJWEIncompleteConnection) {
+				http.Error(w, altinitymcp.ErrJWEIncompleteConnection.Error(), http.StatusUnauthorized)
+				return
+			}
+			if err != nil {
+				http.Error(w, "Missing or invalid authentication token", http.StatusUnauthorized)
+				return
+			}
+			if !parent.JWEClaimsHaveCredentials(claims) {
+				http.Error(w, altinitymcp.ErrJWEIncompleteConnection.Error(), http.StatusUnauthorized)
+				return
+			}
+			ctx := context.WithValue(r.Context(), altinitymcp.JWETokenKey, token)
+			ctx = context.WithValue(ctx, altinitymcp.JWEClaimsKey, claims)
+			r = r.WithContext(ctx)
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+func (a *application) buildSingleClusterHandler(cfg config.Config, sse bool) http.Handler {
+	var sdkHandler http.Handler
+	transport := ""
+	if sse {
+		transport = "sse"
+		a.configMutex.Lock()
+		pool := &credentialSSEPool{parent: a.mcpServer, buckets: make(map[string]*credentialSSEBucket)}
+		a.ssePools = append(a.ssePools, pool)
+		a.configMutex.Unlock()
+		sdkHandler = pool
+	} else {
+		sdkHandler = mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server {
+			return altinitymcp.GetClickHouseJWEServerFromContext(r.Context()).GetServer(r)
+		}, statelessStreamableOptions())
+	}
+	mux := http.NewServeMux()
+	openapiHandler := a.withServerSnapshot(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// The reserved prefix can match /{token}/openapi when a tool itself is
+		// named openapi. It carries no path credential; use the request headers.
+		if strings.HasPrefix(r.URL.Path, "/openapi/") {
+			r.SetPathValue("token", "")
+		}
+		parent := altinitymcp.GetClickHouseJWEServerFromContext(r.Context())
+		parent.OpenAPIHandler(w, r)
+	}))
+	transportHandler := a.withServerSnapshot(a.authenticateSnapshot(sdkHandler))
+	// /openapi/ and /{token}/openapi/ cannot coexist as Go mux patterns.
+	// Combined mode reserves the pathless OpenAPI prefix in its fallback
+	// instead, with the same authentication and body guards as explicit routes.
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if cfg.Server.OpenAPI.Enabled && cfg.Server.JWE.Enabled && cfg.Server.OAuth.Enabled && strings.HasPrefix(r.URL.Path, "/openapi/") {
+			openapiHandler.ServeHTTP(w, r)
+			return
+		}
+		transportHandler.ServeHTTP(w, r)
+	})
+	for _, pattern := range transportRoutePatterns(cfg.Server.JWE.Enabled, cfg.Server.OAuth.Enabled, transport) {
+		mux.Handle(pattern, handler)
+	}
+	if cfg.Server.OpenAPI.Enabled {
+		for _, pattern := range openAPIRoutePatterns(cfg.Server.JWE.Enabled, cfg.Server.OAuth.Enabled) {
+			mux.Handle(pattern, openapiHandler)
+		}
+		if sse && cfg.Server.JWE.Enabled && cfg.Server.OAuth.Enabled {
+			mux.Handle("/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if strings.HasPrefix(r.URL.Path, "/openapi/") {
+					openapiHandler.ServeHTTP(w, r)
+					return
+				}
+				http.NotFound(w, r)
+			}))
+		}
+		protocol := "http"
+		if cfg.Server.OpenAPI.TLS {
+			protocol = "https"
+		}
+		path := "/openapi"
+		if cfg.Server.JWE.Enabled && !cfg.Server.OAuth.Enabled {
+			path = "/{token}/openapi"
+		}
+		log.Info().Str("url", fmt.Sprintf("%s://%s:%d%s", protocol, cfg.Server.Address, cfg.Server.Port, path)).Msg("OpenAPI server listening")
+	}
+	mux.HandleFunc("/health", a.healthHandler)
+	mux.HandleFunc("/livez", a.livenessHandler)
+	mux.HandleFunc("/jwe-token-generator", a.jweTokenGeneratorHandler)
+	a.registerOAuthHTTPRoutes(mux)
+	return finalizeTransportHandler(mux, cfg)
+}
+
+// Each SDK SSE handler owns the sessions of one effective credential. The SDK
+// invokes getServer only on GET and retains that GET's context for POST work.
+// A POST must find its caller's existing bucket, and then its session inside
+// that bucket; misses never allocate or discover anything.
+type credentialSSEPool struct {
+	mu      sync.Mutex
+	parent  *altinitymcp.ClickHouseJWEServer
+	buckets map[string]*credentialSSEBucket
+	active  int
+	closed  bool
+}
+type credentialSSEBucket struct {
+	handler *mcp.SSEHandler
+	refs    int
+	cancels map[*http.Request]context.CancelFunc
+}
+
+func (p *credentialSSEPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	parent := altinitymcp.GetClickHouseJWEServerFromContext(r.Context())
+	key := "static"
+	if parent.Config.Server.JWE.Enabled || parent.Config.Server.OAuth.Enabled {
+		var err error
+		key, _, err = parent.CredentialCatalogKey(r.Context())
+		if err != nil {
+			http.Error(w, "Missing authentication", http.StatusUnauthorized)
+			return
+		}
+	}
+	p.mu.Lock()
+	if p.closed || p.parent != parent {
+		p.mu.Unlock()
+		http.NotFound(w, r)
+		return
+	}
+	bucket := p.buckets[key]
+	if r.Method != http.MethodGet {
+		p.mu.Unlock()
+		if r.Method != http.MethodPost {
+			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		if bucket == nil {
+			http.NotFound(w, r)
+			return
+		}
+		bucket.handler.ServeHTTP(w, r)
+		return
+	}
+	if bucket == nil {
+		bucket = &credentialSSEBucket{handler: mcp.NewSSEHandler(parent.GetServer, nil), cancels: make(map[*http.Request]context.CancelFunc)}
+		p.buckets[key] = bucket
+	}
+	ctx, cancel := context.WithCancel(r.Context())
+	bucket.refs++
+	bucket.cancels[r] = cancel
+	p.active++
+	p.mu.Unlock()
+	defer func() {
+		cancel()
+		p.mu.Lock()
+		delete(bucket.cancels, r)
+		bucket.refs--
+		p.active--
+		if bucket.refs == 0 && p.buckets[key] == bucket {
+			delete(p.buckets, key)
+		}
+		p.mu.Unlock()
+	}()
+	bucket.handler.ServeHTTP(w, r.WithContext(ctx))
+}
+
+func (p *credentialSSEPool) Retire(parent *altinitymcp.ClickHouseJWEServer) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for _, bucket := range p.buckets {
+		for _, cancel := range bucket.cancels {
+			cancel()
+		}
+	}
+	p.buckets = make(map[string]*credentialSSEBucket)
+	p.parent = parent
+}
+func (p *credentialSSEPool) Close() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.closed = true
+	for _, bucket := range p.buckets {
+		for _, cancel := range bucket.cancels {
+			cancel()
+		}
+	}
+	p.buckets = make(map[string]*credentialSSEBucket)
+}