diff --git a/README.md b/README.md index 679286b..febba53 100644 --- a/README.md +++ b/README.md @@ -112,6 +112,8 @@ To serve over HTTP and take the S3M token from each incoming request's `Authoriz s3m mcp --http --stateless-auth ``` +The public S3M MCP server is available at `https://s3m.olcf.ornl.gov/olcf/mcp`. Until it receives a token with usable S3M service scopes, it only exposes one tool describing how to create an S3M token in [myOLCF](https://my.olcf.ornl.gov/) and configure the server. Create a token with the scopes for the S3M services you want to use, then configure your MCP client to connect to the endpoint with `Authorization: `. + The S3M CLI also serves documentation tools via MCP and OpenAPI, to help AI agents understand how to use the other tools. These are populated from a docs archive fetched over HTTP (`--docs-url`, default the public `olcf/s3m-mcp-docs` archive). diff --git a/VERSION b/VERSION index 0ea3a94..0d91a54 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.2.0 +0.3.0 diff --git a/build/containers/deployment/release.Dockerfile b/build/containers/deployment/release.Dockerfile index 8c9c43c..995ade9 100644 --- a/build/containers/deployment/release.Dockerfile +++ b/build/containers/deployment/release.Dockerfile @@ -1,6 +1,6 @@ # syntax=docker/dockerfile:1.7 -ARG GO_VERSION=1.27.1 +ARG GO_VERSION=1.27.2 FROM golang:${GO_VERSION} AS builder WORKDIR /src diff --git a/go.mod b/go.mod index 40d3897..cf98f93 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/olcf/s3m-cli -go 1.27.1 +go 1.27.2 require ( github.com/jedib0t/go-pretty/v6 v6.8.3 @@ -20,7 +20,7 @@ require ( github.com/segmentio/asm v1.2.1 // indirect github.com/segmentio/encoding v0.5.4 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect - golang.org/x/net v0.59.0 // indirect + golang.org/x/net v0.60.0 // indirect golang.org/x/oauth2 v0.37.0 // indirect golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect diff --git a/go.sum b/go.sum index 34b2629..dddad92 100644 --- a/go.sum +++ b/go.sum @@ -28,8 +28,8 @@ github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zI github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= -golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= -golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU= +golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98= golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58= golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= diff --git a/internal/cmd/servercmd/helpers.go b/internal/cmd/servercmd/helpers.go index b52b641..39130f1 100644 --- a/internal/cmd/servercmd/helpers.go +++ b/internal/cmd/servercmd/helpers.go @@ -168,38 +168,6 @@ func prepareStatelessConnection( return allowed, conn, nil } -func wrapWithAuthTokenExtraction(handler http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method == http.MethodOptions { - handler.ServeHTTP(w, r) - return - } - - authorization := r.Header.Get("Authorization") - if len(authorization) > maxAuthorizationHeaderBytes { - http.Error(w, "Authorization header too large", http.StatusRequestHeaderFieldsTooLarge) - - return - } - - token := grpcclient.TokenFromAuthorizationHeader(authorization) - - logAuthTokenExtraction(r, token) - - if token == "" { - logMissingAuthorization(r) - http.Error(w, "missing Authorization header", http.StatusUnauthorized) - - return - } - - ctx := grpcclient.ContextWithAuthToken(r.Context(), token) - handler.ServeHTTP(w, r.WithContext(ctx)) - }) -} - -const maxAuthorizationHeaderBytes = 16 << 10 - func registerStatelessOpenAPIHandlers( mux *http.ServeMux, routes []toolset.HTTPRouteSpec, diff --git a/internal/cmd/servercmd/helpers_test.go b/internal/cmd/servercmd/helpers_test.go index ff62e42..47243ca 100644 --- a/internal/cmd/servercmd/helpers_test.go +++ b/internal/cmd/servercmd/helpers_test.go @@ -100,93 +100,6 @@ func TestHasAllowedMethodNames(t *testing.T) { } } -func TestWrapWithAuthTokenExtractionSuccess(t *testing.T) { - var capturedToken string - - inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - token, ok := grpcclient.AuthTokenFromContext(r.Context()) - if ok { - capturedToken = token - } - - w.WriteHeader(http.StatusOK) - }) - - handler := wrapWithAuthTokenExtraction(inner) - - req := httptest.NewRequest(http.MethodPost, "/", nil) - req.Header.Set("Authorization", "Bearer my-secret-token") - - rr := httptest.NewRecorder() - handler.ServeHTTP(rr, req) - - if rr.Code != http.StatusOK { - t.Fatalf("expected status 200, got %d", rr.Code) - } - - if capturedToken != "my-secret-token" { - t.Fatalf("expected token 'my-secret-token', got %q", capturedToken) - } -} - -func TestWrapWithAuthTokenExtractionMissingHeader(t *testing.T) { - inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - t.Fatal("inner handler should not be called") - }) - - handler := wrapWithAuthTokenExtraction(inner) - - req := httptest.NewRequest(http.MethodPost, "/", nil) - - rr := httptest.NewRecorder() - handler.ServeHTTP(rr, req) - - if rr.Code != http.StatusUnauthorized { - t.Fatalf("expected status 401, got %d", rr.Code) - } -} - -func TestWrapWithAuthTokenExtractionRejectsOversizedHeader(t *testing.T) { - inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - t.Fatal("inner handler should not be called") - }) - - handler := wrapWithAuthTokenExtraction(inner) - req := httptest.NewRequest(http.MethodPost, "/", nil) - req.Header.Set("Authorization", strings.Repeat("x", maxAuthorizationHeaderBytes+1)) - - rr := httptest.NewRecorder() - handler.ServeHTTP(rr, req) - - if rr.Code != http.StatusRequestHeaderFieldsTooLarge { - t.Fatalf("expected status 431, got %d", rr.Code) - } -} - -func TestWrapWithAuthTokenExtractionOptionsPassthrough(t *testing.T) { - called := false - - inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - called = true - w.WriteHeader(http.StatusNoContent) - }) - - handler := wrapWithAuthTokenExtraction(inner) - - req := httptest.NewRequest(http.MethodOptions, "/", nil) - - rr := httptest.NewRecorder() - handler.ServeHTTP(rr, req) - - if !called { - t.Fatal("OPTIONS request should pass through without auth check") - } - - if rr.Code != http.StatusNoContent { - t.Fatalf("expected status 204, got %d", rr.Code) - } -} - func TestWrapStatelessOpenAPIRouteAllowsAuthorizedMethod(t *testing.T) { file := slurmv0042pb.File_proto_slurm_v0042_slurm_proto service := file.Services().ByName("SlurmIndirect") diff --git a/internal/cmd/servercmd/logging.go b/internal/cmd/servercmd/logging.go index a52ffae..3d0130e 100644 --- a/internal/cmd/servercmd/logging.go +++ b/internal/cmd/servercmd/logging.go @@ -39,21 +39,6 @@ func truncateForLog(value string) string { return sanitizeLogValue(value[:8]) + "..." } -func logAuthTokenExtraction(r *http.Request, token string) { - slog.LogAttrs(r.Context(), slog.LevelDebug, "wrapWithAuthTokenExtraction", - safeStringAttr("method", r.Method), - safeStringAttr("path", r.URL.Path), - slog.Bool("hasToken", token != ""), - safeStringAttr("tokenPrefix", truncateForLog(token)), - ) -} - -func logMissingAuthorization(r *http.Request) { - slog.LogAttrs(r.Context(), slog.LevelWarn, "Rejecting request: missing Authorization header", - safeStringAttr("path", r.URL.Path), - ) -} - func logMCPServerFactory(r *http.Request, token string) { slog.LogAttrs(r.Context(), slog.LevelDebug, "MCP server factory called", slog.Bool("hasToken", token != ""), diff --git a/internal/cmd/servercmd/server.go b/internal/cmd/servercmd/server.go index efe96e2..c12dff4 100644 --- a/internal/cmd/servercmd/server.go +++ b/internal/cmd/servercmd/server.go @@ -35,7 +35,8 @@ func BuildMCPCommand(rt *runtime.Runtime) *cli.Command { Usage: "Use S3M token from Authorization header for MCP over HTTP and OpenAPI servers"}, &cli.BoolFlag{Name: "stdio", Usage: "Also serve MCP over STDIO (default when --http is not set)"}, - }, docFlags()...) + }, tokenValidationFlags()...) + flags = append(flags, docFlags()...) return &cli.Command{ Name: "mcp", @@ -77,23 +78,7 @@ func BuildMCPCommand(rt *runtime.Runtime) *cli.Command { defer closeConn(conn) - cache := newTokenCache(rt, conn) - handler := newStatelessMCPHTTPHandler(rt, conn, cache) - - addr := c.String("http-addr") - slog.Info("listening for MCP HTTP (stateless, per-token filtering)", "addr", addr) - - srv := &http.Server{ - Addr: addr, - Handler: handler, - ReadHeaderTimeout: 10 * time.Second, - ReadTimeout: 30 * time.Second, - WriteTimeout: 30 * time.Second, - IdleTimeout: 60 * time.Second, - } - - // no STDIO in stateless mode - return srv.ListenAndServe() + return serveStatelessMCPHTTP(rt, c, conn) } // Non-stateless mode: single shared server @@ -161,6 +146,28 @@ func BuildMCPCommand(rt *runtime.Runtime) *cli.Command { } } +func serveStatelessMCPHTTP(rt *runtime.Runtime, c *cli.Command, conn *grpc.ClientConn) error { + cache, err := newConfiguredTokenCache(rt, c, conn) + if err != nil { + return err + } + + handler := newStatelessMCPHTTPHandler(rt, conn, cache) + addr := c.String("http-addr") + slog.Info("listening for MCP HTTP (stateless, per-token filtering)", "addr", addr) + + srv := &http.Server{ + Addr: addr, + Handler: handler, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 60 * time.Second, + } + + return srv.ListenAndServe() +} + func newStatelessMCPHTTPHandler( rt *runtime.Runtime, conn *grpc.ClientConn, @@ -190,6 +197,12 @@ func newStatelessMCPHTTPHandler( }) requestHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if len(r.Header.Get("Authorization")) > maxAuthorizationHeaderBytes { + http.Error(w, "Authorization header too large", http.StatusRequestHeaderFieldsTooLarge) + + return + } + if !validStatelessMCPRequest(w, r) { return } @@ -199,9 +212,11 @@ func newStatelessMCPHTTPHandler( streamableHandler.ServeHTTP(w, r.WithContext(ctx)) }) - return wrapWithAuthTokenExtraction(requestHandler) + return requestHandler } +const maxAuthorizationHeaderBytes = 16 << 10 + func buildStatelessMCPServer( rt *runtime.Runtime, conn *grpc.ClientConn, @@ -212,16 +227,29 @@ func buildStatelessMCPServer( logMCPServerFactory(r, token) - perms := cache.GetPermissions(r.Context(), token) - allowed := filterAllowed(perms, rt.Methods) - - ts := buildToolSet(rt, allowed, conn, cache.Vars) - srv := mcp.NewServer(&mcp.Implementation{ Name: "s3m", Version: buildinfo.Version, }, nil) + perms, introspectionErr := cache.getPermissions(r.Context(), token) + if !perms.Known { + fallback := toolset.BuildUnauthenticatedToolSet() + if introspectionErr != nil { + fallback = toolset.BuildAuthenticationErrorToolSet() + } + + for _, spec := range fallback.MCP { + srv.AddTool(spec.Tool, spec.Handler) + } + + return srv + } + + allowed := filterAllowed(perms, rt.Methods) + + ts := buildToolSet(rt, allowed, conn, cache.Vars) + for _, spec := range ts.MCP { originalHandler := spec.Handler capturedToken := token @@ -285,7 +313,8 @@ func BuildOpenAPICommand(rt *runtime.Runtime) *cli.Command { &cli.StringFlag{Name: "addr", Value: cmd.DefaultOpenAPIAddr, Usage: "Address to bind"}, &cli.BoolFlag{Name: "stateless-auth", Usage: "Use stateless auth: require S3M token from Authorization header and do not require a stored login"}, - }, docFlags()...) + }, tokenValidationFlags()...) + flags = append(flags, docFlags()...) return &cli.Command{ Name: "openapi", @@ -319,7 +348,11 @@ func BuildOpenAPICommand(rt *runtime.Runtime) *cli.Command { defer closeConn(conn) - cache := newTokenCache(rt, conn) + cache, err := newConfiguredTokenCache(rt, c, conn) + if err != nil { + return err + } + ts := buildToolSet(rt, allowed, conn, cache.Vars) mux := http.NewServeMux() @@ -365,6 +398,43 @@ func BuildOpenAPICommand(rt *runtime.Runtime) *cli.Command { } } +func newConfiguredTokenCache( + rt *runtime.Runtime, + c *cli.Command, + connections ...grpc.ClientConnInterface, +) (*tokenCache, error) { + return newTokenCacheWithValidation( + rt, + c.String("stateless-token-regex"), + c.Int("stateless-token-min-length"), + c.Int("stateless-token-max-length"), + connections..., + ) +} + +func tokenValidationFlags() []cli.Flag { + return []cli.Flag{ + &cli.StringFlag{ + Name: "stateless-token-regex", + Value: defaultS3MJWTRegex, + Sources: cli.EnvVars("S3M_STATELESS_TOKEN_REGEX"), + Usage: "Regular expression required for tokens before stateless introspection", + }, + &cli.IntFlag{ + Name: "stateless-token-min-length", + Value: defaultS3MTokenMinLength, + Sources: cli.EnvVars("S3M_STATELESS_TOKEN_MIN_LENGTH"), + Usage: "Minimum token length allowed for stateless introspection", + }, + &cli.IntFlag{ + Name: "stateless-token-max-length", + Value: defaultS3MTokenMaxLength, + Sources: cli.EnvVars("S3M_STATELESS_TOKEN_MAX_LENGTH"), + Usage: "Maximum token length allowed for stateless introspection", + }, + } +} + func configureGitDocs(ctx context.Context, rt *runtime.Runtime, c *cli.Command) error { docsURL := c.String("docs-url") if docsURL == "" { diff --git a/internal/cmd/servercmd/server_test.go b/internal/cmd/servercmd/server_test.go index cedbb51..518f03d 100644 --- a/internal/cmd/servercmd/server_test.go +++ b/internal/cmd/servercmd/server_test.go @@ -22,6 +22,7 @@ import ( func TestStatelessMCPHTTPHandlerBuildsServerOncePerRequest(t *testing.T) { rt := &runtime.Runtime{} cache := newTokenCache(rt) + token := testS3MToken() var introspections int cache.introspect = func(context.Context, *runtime.Runtime, string) (auth.TokenRecord, error) { @@ -32,7 +33,7 @@ func TestStatelessMCPHTTPHandlerBuildsServerOncePerRequest(t *testing.T) { handler := newStatelessMCPHTTPHandler(rt, nil, cache) req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{}`)) - req.Header.Set("Authorization", "Bearer invalid-token") + req.Header.Set("Authorization", token) req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json, text/event-stream") @@ -82,7 +83,137 @@ func TestStatelessMCPHTTPHandlerRejectsInvalidTransportBeforeIntrospection(t *te } } -func TestStatelessMCPHTTPHandlerUsesBearerForVisibleDocs(t *testing.T) { +func TestStatelessMCPHTTPHandlerRejectsOversizedAuthorizationBeforeIntrospection(t *testing.T) { + rt := &runtime.Runtime{} + cache := newTokenCache(rt) + + var introspections int + cache.introspect = func(context.Context, *runtime.Runtime, string) (auth.TokenRecord, error) { + introspections++ + + return auth.TokenRecord{}, errors.New("invalid token") + } + + handler := newStatelessMCPHTTPHandler(rt, nil, cache) + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{}`)) + req.Header.Set("Authorization", strings.Repeat("x", maxAuthorizationHeaderBytes+1)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, req) + + if recorder.Code != http.StatusRequestHeaderFieldsTooLarge { + t.Fatalf("expected status 431, got %d", recorder.Code) + } + if introspections != 0 { + t.Fatalf("expected oversized authorization to skip introspection, got %d calls", introspections) + } +} + +func TestStatelessMCPHTTPHandlerProvidesAccessToolWithoutAuthorization(t *testing.T) { + server := httptest.NewServer(newStatelessMCPHTTPHandler(&runtime.Runtime{}, nil, newTokenCache(nil))) + defer server.Close() + + client := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "1"}, nil) + session, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: server.URL, + HTTPClient: server.Client(), + DisableStandaloneSSE: true, + }, nil) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer func() { _ = session.Close() }() + + tools, err := session.ListTools(context.Background(), nil) + if err != nil { + t.Fatalf("ListTools: %v", err) + } + if len(tools.Tools) != 1 || !hasTool(tools.Tools, "get_s3m_access") { + t.Fatalf("expected only get_s3m_access, got %+v", toolNames(tools.Tools)) + } + + result, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: "get_s3m_access"}) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if result.IsError { + t.Fatalf("expected get_s3m_access to succeed, got %+v", result) + } + + text, ok := result.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("expected text content, got %T", result.Content[0]) + } + if !strings.Contains(text.Text, "https://my.olcf.ornl.gov/") || + !strings.Contains(text.Text, "scopes for the S3M services") || + !strings.Contains(text.Text, `"mcpServers"`) || + !strings.Contains(text.Text, "${S3M_TOKEN}") || + !strings.Contains(text.Text, "[mcp_servers.s3m]") || + !strings.Contains(text.Text, `env_http_headers = { Authorization = "S3M_TOKEN" }`) || + !strings.Contains(text.Text, "~/.copilot/mcp-config.json") || + !strings.Contains(text.Text, `"tools": ["*"]`) || + !strings.Contains(text.Text, "# claude|codex|copilot") { + t.Fatalf("unexpected access instructions: %q", text.Text) + } +} + +func TestStatelessMCPHTTPHandlerReportsAuthenticationError(t *testing.T) { + rt := &runtime.Runtime{} + cache := newTokenCache(rt) + cache.introspect = func(context.Context, *runtime.Runtime, string) (auth.TokenRecord, error) { + return auth.TokenRecord{}, errors.New("unavailable") + } + + server := httptest.NewServer(newStatelessMCPHTTPHandler(rt, nil, cache)) + defer server.Close() + + httpClient := server.Client() + httpClient.Transport = authHeaderRoundTripper{ + base: httpClient.Transport, + token: testS3MToken(), + } + + client := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "1"}, nil) + session, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: server.URL, + HTTPClient: httpClient, + DisableStandaloneSSE: true, + }, nil) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer func() { _ = session.Close() }() + + tools, err := session.ListTools(context.Background(), nil) + if err != nil { + t.Fatalf("ListTools: %v", err) + } + if len(tools.Tools) != 1 || !hasTool(tools.Tools, "get_s3m_access") { + t.Fatalf("expected only get_s3m_access, got %+v", toolNames(tools.Tools)) + } + if tools.Tools[0].Title != "Resolve S3M Authentication" { + t.Fatalf("unexpected fallback title: %q", tools.Tools[0].Title) + } + + result, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: "get_s3m_access"}) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + + text, ok := result.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("expected text content, got %T", result.Content[0]) + } + if !strings.Contains(text.Text, "could not be validated") || + !strings.Contains(text.Text, "temporarily unavailable") || + strings.Contains(text.Text, `"mcpServers"`) { + t.Fatalf("unexpected authentication error instructions: %q", text.Text) + } +} + +func TestStatelessMCPHTTPHandlerProvidesAccessToolForUnusableToken(t *testing.T) { file := slurmv0042pb.File_proto_slurm_v0042_slurm_proto service := file.Services().ByName("SlurmIndirect") method := service.Methods().ByName("GetJobs") @@ -97,9 +228,8 @@ func TestStatelessMCPHTTPHandlerUsesBearerForVisibleDocs(t *testing.T) { Desc: "Get jobs", }} - store := loadToolDocStore(t, methods, toolName, "Project {{project}}") rt := &runtime.Runtime{Methods: methods} - rt.SetDocs(store) + rt.SetDocs(loadToolDocStore(t, methods, toolName, "Project {{project}}")) cache := newTokenCache(rt) cache.storeRecord("unknown-token", auth.TokenRecord{ @@ -129,12 +259,68 @@ func TestStatelessMCPHTTPHandlerUsesBearerForVisibleDocs(t *testing.T) { } defer func() { _ = session.Close() }() + tools, err := session.ListTools(context.Background(), nil) + if err != nil { + t.Fatalf("ListTools: %v", err) + } + if len(tools.Tools) != 1 || !hasTool(tools.Tools, "get_s3m_access") { + t.Fatalf("expected only get_s3m_access, got %+v", toolNames(tools.Tools)) + } +} + +func TestStatelessMCPHTTPHandlerUsesAuthorizationHeaderForVisibleDocs(t *testing.T) { + file := slurmv0042pb.File_proto_slurm_v0042_slurm_proto + service := file.Services().ByName("SlurmIndirect") + method := service.Methods().ByName("GetJobs") + toolName := proto.ToolNameForMethod(file, service, method) + + methods := []proto.MethodInfo{{ + File: file, + Service: service, + Method: method, + ToolName: toolName, + Path: "/jobs", + Desc: "Get jobs", + }} + + store := loadToolDocStore(t, methods, toolName, "Project {{project}}") + rt := &runtime.Runtime{Methods: methods} + rt.SetDocs(store) + + cache := newTokenCache(rt) + cache.storeRecord("known-token", auth.TokenRecord{ + Token: "known-token", + Project: "proj-x", + Enclave: "enc", + Scopes: []string{"*"}, + }) + + server := httptest.NewServer(newStatelessMCPHTTPHandler(rt, nil, cache)) + defer server.Close() + + httpClient := server.Client() + httpClient.Transport = authHeaderRoundTripper{ + base: httpClient.Transport, + token: "known-token", + } + + client := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "1"}, nil) + session, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: server.URL, + HTTPClient: httpClient, + DisableStandaloneSSE: true, + }, nil) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer func() { _ = session.Close() }() + tools, err := session.ListTools(context.Background(), nil) if err != nil { t.Fatalf("ListTools: %v", err) } if !hasTool(tools.Tools, toolName) { - t.Fatalf("expected fail-open stateless surface to include %q, got %+v", toolName, toolNames(tools.Tools)) + t.Fatalf("expected permitted stateless surface to include %q, got %+v", toolName, toolNames(tools.Tools)) } if !hasTool(tools.Tools, "doc_lookup") { t.Fatalf("expected doc_lookup to be exposed, got %+v", toolNames(tools.Tools)) @@ -215,6 +401,15 @@ func TestServerCommandsDefaultToPublicGitDocs(t *testing.T) { if got := stringFlagDefault(t, cmd, "docs-path"); got != defaultDocsPath { t.Fatalf("docs-path default = %q, want %q", got, defaultDocsPath) } + if got := stringFlagDefault(t, cmd, "stateless-token-regex"); got != defaultS3MJWTRegex { + t.Fatalf("stateless-token-regex default = %q, want %q", got, defaultS3MJWTRegex) + } + if got := intFlagDefault(t, cmd, "stateless-token-min-length"); got != defaultS3MTokenMinLength { + t.Fatalf("stateless-token-min-length default = %d, want %d", got, defaultS3MTokenMinLength) + } + if got := intFlagDefault(t, cmd, "stateless-token-max-length"); got != defaultS3MTokenMaxLength { + t.Fatalf("stateless-token-max-length default = %d, want %d", got, defaultS3MTokenMaxLength) + } if !hasFlag(cmd, "docs-token") || !hasFlag(cmd, "docs-poll") { t.Fatalf("expected docs override flags on %s command", tt.name) } @@ -230,7 +425,7 @@ type authHeaderRoundTripper struct { func (rt authHeaderRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { cloned := req.Clone(req.Context()) cloned.Header = req.Header.Clone() - cloned.Header.Set("Authorization", "Bearer "+rt.token) + cloned.Header.Set("Authorization", rt.token) return rt.base.RoundTrip(cloned) } @@ -308,6 +503,19 @@ func stringFlagDefault(t *testing.T, cmd *cli.Command, name string) string { return "" } +func intFlagDefault(t *testing.T, cmd *cli.Command, name string) int { + t.Helper() + + for _, flag := range cmd.Flags { + if intFlag, ok := flag.(*cli.IntFlag); ok && intFlag.Name == name { + return intFlag.Value + } + } + + t.Fatalf("int flag %q not found", name) + return 0 +} + func hasFlag(cmd *cli.Command, name string) bool { for _, flag := range cmd.Flags { switch typed := flag.(type) { diff --git a/internal/cmd/servercmd/tokencache.go b/internal/cmd/servercmd/tokencache.go index fda7e8e..9cdb476 100644 --- a/internal/cmd/servercmd/tokencache.go +++ b/internal/cmd/servercmd/tokencache.go @@ -4,7 +4,9 @@ import ( "container/list" "context" "errors" + "fmt" "log/slog" + "regexp" "sync" "time" @@ -26,6 +28,9 @@ type tokenCache struct { cache map[string]*tokenCacheEntry order *list.List inflight map[string]*tokenLookup + tokenRegex *regexp.Regexp + minLength int + maxLength int } type tokenCacheEntry struct { @@ -38,18 +43,62 @@ type tokenLookup struct { done chan struct{} record auth.TokenRecord ok bool + err error } const tokenCacheTTL = 2 * time.Hour const tokenCacheMaxEntries = 2048 +const ( + defaultS3MJWTRegex = `^eyJh[A-Za-z0-9_-]*\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$` + defaultS3MTokenMinLength = 300 + defaultS3MTokenMaxLength = 4096 +) + func newTokenCache(rt *runtime.Runtime, connections ...grpc.ClientConnInterface) *tokenCache { + cache, err := newTokenCacheWithValidation( + rt, + defaultS3MJWTRegex, + defaultS3MTokenMinLength, + defaultS3MTokenMaxLength, + connections..., + ) + if err != nil { + panic(err) + } + + return cache +} + +func newTokenCacheWithValidation( + rt *runtime.Runtime, + tokenRegex string, + minLength int, + maxLength int, + connections ...grpc.ClientConnInterface, +) (*tokenCache, error) { + if minLength < 0 { + return nil, errors.New("S3M token minimum length must not be negative") + } + + if maxLength < minLength { + return nil, errors.New("S3M token maximum length must be at least the minimum length") + } + + compiledTokenRegex, err := regexp.Compile(tokenRegex) + if err != nil { + return nil, fmt.Errorf("compile S3M token regex: %w", err) + } + cache := &tokenCache{ rt: rt, introspect: introspectTokenRecord, cache: make(map[string]*tokenCacheEntry), order: list.New(), inflight: make(map[string]*tokenLookup), + tokenRegex: compiledTokenRegex, + minLength: minLength, + maxLength: maxLength, } if len(connections) > 0 && connections[0] != nil { @@ -66,7 +115,7 @@ func newTokenCache(rt *runtime.Runtime, connections ...grpc.ClientConnInterface) } } - return cache + return cache, nil } func introspectTokenRecord( @@ -121,12 +170,18 @@ func (c *tokenCache) GetRecord(ctx context.Context, token string) (auth.TokenRec // GetPermissions returns a permissions snapshot for the given token. func (c *tokenCache) GetPermissions(ctx context.Context, token string) permissions.Snapshot { - rec, ok := c.GetRecord(ctx, token) + perms, _ := c.getPermissions(ctx, token) + + return perms +} + +func (c *tokenCache) getPermissions(ctx context.Context, token string) (permissions.Snapshot, error) { + rec, ok, err := c.recordForTokenResult(ctx, token) if !ok { - return permissions.New(nil, false) + return permissions.New(nil, false), err } - return permissions.New(rec.Scopes, rec.IntrospectionOK()) + return permissions.New(rec.Scopes, rec.IntrospectionOK()), nil } func (c *tokenCache) resolveRecord(ctx context.Context) (auth.TokenRecord, bool) { @@ -143,12 +198,18 @@ func (c *tokenCache) resolveRecord(ctx context.Context) (auth.TokenRecord, bool) } func (c *tokenCache) recordForToken(ctx context.Context, token string) (auth.TokenRecord, bool) { + rec, ok, _ := c.recordForTokenResult(ctx, token) + + return rec, ok +} + +func (c *tokenCache) recordForTokenResult(ctx context.Context, token string) (auth.TokenRecord, bool, error) { if token == "" { - return auth.TokenRecord{}, false + return auth.TokenRecord{}, false, nil } if cached, ok := c.loadCachedRecord(token); ok { - return cached, true + return cached, true, nil } c.mu.Lock() @@ -157,9 +218,9 @@ func (c *tokenCache) recordForToken(ctx context.Context, token string) (auth.Tok select { case <-pending.done: - return pending.record, pending.ok + return pending.record, pending.ok, pending.err case <-ctx.Done(): - return auth.TokenRecord{}, false + return auth.TokenRecord{}, false, ctx.Err() } } @@ -167,38 +228,49 @@ func (c *tokenCache) recordForToken(ctx context.Context, token string) (auth.Tok c.inflight[token] = pending c.mu.Unlock() - rec, ok := c.lookupToken(ctx, token) + rec, ok, err := c.lookupToken(ctx, token) c.mu.Lock() pending.record = rec pending.ok = ok + pending.err = err delete(c.inflight, token) close(pending.done) c.mu.Unlock() - return rec, ok + return rec, ok, err } -func (c *tokenCache) lookupToken(ctx context.Context, token string) (auth.TokenRecord, bool) { +func (c *tokenCache) lookupToken(ctx context.Context, token string) (auth.TokenRecord, bool, error) { if cached, ok := c.loadCachedRecord(token); ok { - return cached, true + return cached, true, nil } if rec, ok := c.lookupStoredRecord(token); ok { c.storeRecord(token, rec) - return rec, true + return rec, true, nil + } + + if !c.looksLikeS3MToken(token) { + return auth.TokenRecord{}, false, nil } rec, err := c.introspectToken(ctx, token) if err != nil { slog.Debug("token introspection failed", "error", err) - return auth.TokenRecord{}, false + return auth.TokenRecord{}, false, err } c.storeRecord(token, rec) - return rec, true + return rec, true, nil +} + +func (c *tokenCache) looksLikeS3MToken(token string) bool { + return len(token) >= c.minLength && + len(token) <= c.maxLength && + c.tokenRegex.MatchString(token) } func (c *tokenCache) loadCachedRecord(token string) (auth.TokenRecord, bool) { diff --git a/internal/cmd/servercmd/tokencache_test.go b/internal/cmd/servercmd/tokencache_test.go index b162d25..c8b9fe9 100644 --- a/internal/cmd/servercmd/tokencache_test.go +++ b/internal/cmd/servercmd/tokencache_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "testing" "time" @@ -14,6 +15,16 @@ import ( "github.com/olcf/s3m-cli/internal/runtime" ) +func testS3MToken() string { + return testS3MTokenWithLength(defaultS3MTokenMinLength) +} + +func testS3MTokenWithLength(length int) string { + const fixed = "eyJh.a.a" + + return "eyJh" + strings.Repeat("a", length-len(fixed)) + ".a.a" +} + func TestTokenCacheStoreAndLoad(t *testing.T) { cache := newTokenCache(nil) @@ -146,21 +157,33 @@ func TestGetPermissionsWithScopes(t *testing.T) { } func TestGetPermissionsUnknownToken(t *testing.T) { - cache := newTokenCache(nil) + rt := &runtime.Runtime{Target: "unused-target"} + cache := newTokenCache(rt) + + var calls int + cache.introspect = func(context.Context, *runtime.Runtime, string) (auth.TokenRecord, error) { + calls++ + + return auth.TokenRecord{}, nil + } perms := cache.GetPermissions(context.Background(), "unknown-token") if perms.Known { t.Fatal("expected permissions to be unknown for missing token") } + if calls != 0 { + t.Fatalf("expected malformed token to skip introspection, got %d calls", calls) + } } func TestGetPermissionsFallsBackToIntrospection(t *testing.T) { rt := &runtime.Runtime{Target: "unused-target"} cache := newTokenCache(rt) + token := testS3MToken() introspected := auth.TokenRecord{ - Token: "unknown-token", + Token: token, Project: "proj", Enclave: "enc", Scopes: []string{"/test.Service/*"}, @@ -176,14 +199,14 @@ func TestGetPermissionsFallsBackToIntrospection(t *testing.T) { if gotRT != rt { t.Fatal("expected introspection to receive the cache runtime") } - if token != "unknown-token" { - t.Fatalf("expected token unknown-token, got %q", token) + if token != testS3MToken() { + t.Fatalf("expected plausible S3M token, got %q", token) } return introspected, nil } - perms := cache.GetPermissions(context.Background(), "unknown-token") + perms := cache.GetPermissions(context.Background(), token) if !perms.Known { t.Fatal("expected introspected permissions to be known") @@ -195,7 +218,7 @@ func TestGetPermissionsFallsBackToIntrospection(t *testing.T) { t.Fatalf("expected one introspection call, got %d", calls) } - perms = cache.GetPermissions(context.Background(), "unknown-token") + perms = cache.GetPermissions(context.Background(), token) if !perms.Known { t.Fatal("expected cached introspected permissions to stay known") @@ -208,21 +231,26 @@ func TestGetPermissionsFallsBackToIntrospection(t *testing.T) { func TestGetPermissionsReturnsUnknownWhenIntrospectionFails(t *testing.T) { rt := &runtime.Runtime{Target: "unused-target"} cache := newTokenCache(rt) + wantErr := errors.New("boom") cache.introspect = func(context.Context, *runtime.Runtime, string) (auth.TokenRecord, error) { - return auth.TokenRecord{}, errors.New("boom") + return auth.TokenRecord{}, wantErr } - perms := cache.GetPermissions(context.Background(), "unknown-token") + perms, err := cache.getPermissions(context.Background(), testS3MToken()) if perms.Known { t.Fatal("expected failed introspection to leave permissions unknown") } + if !errors.Is(err, wantErr) { + t.Fatalf("expected introspection error %v, got %v", wantErr, err) + } } func TestTokenCacheCoalescesConcurrentIntrospection(t *testing.T) { rt := &runtime.Runtime{Target: "unused-target"} cache := newTokenCache(rt) + token := testS3MToken() started := make(chan struct{}) release := make(chan struct{}) @@ -239,14 +267,14 @@ func TestTokenCacheCoalescesConcurrentIntrospection(t *testing.T) { <-release - return auth.TokenRecord{Token: "shared-token", Scopes: []string{"/test/*"}}, nil + return auth.TokenRecord{Token: token, Scopes: []string{"/test/*"}}, nil } const goroutines = 8 results := make(chan permissions.Snapshot, goroutines) for range goroutines { go func() { - results <- cache.GetPermissions(context.Background(), "shared-token") + results <- cache.GetPermissions(context.Background(), token) }() } @@ -267,6 +295,76 @@ func TestTokenCacheCoalescesConcurrentIntrospection(t *testing.T) { } } +func TestLooksLikeS3MToken(t *testing.T) { + cache := newTokenCache(nil) + + tests := []struct { + name string + token string + want bool + }{ + {name: "minimum length", token: testS3MToken(), want: true}, + {name: "maximum length", token: testS3MTokenWithLength(defaultS3MTokenMaxLength), want: true}, + {name: "too short", token: testS3MTokenWithLength(defaultS3MTokenMinLength - 1)}, + {name: "too long", token: testS3MTokenWithLength(defaultS3MTokenMaxLength + 1)}, + {name: "wrong prefix", token: "eyJb" + testS3MToken()[len("eyJh"):]}, + {name: "missing segments", token: "eyJh" + strings.Repeat("a", defaultS3MTokenMinLength-len("eyJh"))}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := cache.looksLikeS3MToken(tt.token); got != tt.want { + t.Fatalf("looksLikeS3MToken() = %t, want %t", got, tt.want) + } + }) + } +} + +func TestTokenCacheUsesConfiguredTokenValidation(t *testing.T) { + cache, err := newTokenCacheWithValidation(nil, `^custom-[0-9]+$`, 8, 10) + if err != nil { + t.Fatalf("newTokenCacheWithValidation: %v", err) + } + + tests := []struct { + token string + want bool + }{ + {token: "custom-1", want: true}, + {token: "custom-123", want: true}, + {token: "custom-", want: false}, + {token: "custom-1234", want: false}, + {token: "other-12", want: false}, + } + + for _, tt := range tests { + if got := cache.looksLikeS3MToken(tt.token); got != tt.want { + t.Errorf("looksLikeS3MToken(%q) = %t, want %t", tt.token, got, tt.want) + } + } +} + +func TestTokenCacheRejectsInvalidTokenValidation(t *testing.T) { + tests := []struct { + name string + regex string + minLength int + maxLength int + }{ + {name: "invalid regex", regex: `[`, minLength: 1, maxLength: 2}, + {name: "negative minimum", regex: `.`, minLength: -1, maxLength: 2}, + {name: "maximum below minimum", regex: `.`, minLength: 2, maxLength: 1}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := newTokenCacheWithValidation(nil, tt.regex, tt.minLength, tt.maxLength); err == nil { + t.Fatal("expected invalid token validation configuration to fail") + } + }) + } +} + func TestVarsFromToken(t *testing.T) { cache := newTokenCache(nil) diff --git a/internal/toolset/access.go b/internal/toolset/access.go new file mode 100644 index 0000000..6f0f21f --- /dev/null +++ b/internal/toolset/access.go @@ -0,0 +1,53 @@ +package toolset + +import ( + "context" + _ "embed" + "encoding/json" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +//go:embed access_instructions.txt +var accessInstructions string + +func BuildUnauthenticatedToolSet() *ToolSet { + return buildAccessToolSet( + "Get S3M Access", + "Explain how to get an S3M token and configure this MCP server. Use if no other S3M tools "+ + "are available but the user is asking for facility operations (submitting a job, starting a "+ + "streaming cluster, etc.)", + accessInstructions, + ) +} + +func BuildAuthenticationErrorToolSet() *ToolSet { + return buildAccessToolSet( + "Resolve S3M Authentication", + "Explain that the supplied S3M token could not be validated and how to recover.", + "The supplied S3M token could not be validated. The authentication service may be temporarily "+ + "unavailable, or the token may be invalid or expired. Retry the request. If it continues to fail, "+ + "create a new token in myOLCF and use it for this MCP server. Do not paste the token into this "+ + "conversation.", + ) +} + +func buildAccessToolSet(title, description, instructions string) *ToolSet { + ts := New() + + ts.MCP = append(ts.MCP, MCPToolSpec{ + Tool: &mcp.Tool{ + Name: "get_s3m_access", + Title: title, + Description: description, + InputSchema: json.RawMessage(`{"type":"object","additionalProperties":false}`), + }, + Handler: func(_ context.Context, _ *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: instructions}}, + }, nil + }, + }) + + return ts +} diff --git a/internal/toolset/access_instructions.txt b/internal/toolset/access_instructions.txt new file mode 100644 index 0000000..e152c19 --- /dev/null +++ b/internal/toolset/access_instructions.txt @@ -0,0 +1,58 @@ +## Using S3M tools + +1. Log in to https://my.olcf.ornl.gov/ and select the S3M-enabled project you want to work in. +2. Create an S3M token with the scopes for the S3M services you want to use. Scopes are enabled per project; if your project cannot select a scope you need, contact User Assistance or the S3M team. +3. Configure your MCP server registration to read the raw `Authorization` header from the environment variable `S3M_TOKEN`. (Your token expires, so loading it dynamically is both easier and more secure.) + + For Claude Code, create `.mcp.json` in your project directory: + + ```json + { + "mcpServers": { + "s3m": { + "type": "http", + "url": "https://s3m.olcf.ornl.gov/olcf/mcp", + "headers": { + "Authorization": "${S3M_TOKEN}" + } + } + } + } + ``` + + For Codex, add this to `~/.codex/config.toml`: + + ```toml + [mcp_servers.s3m] + url = "https://s3m.olcf.ornl.gov/olcf/mcp" + env_http_headers = { Authorization = "S3M_TOKEN" } + ``` + + For GitHub Copilot CLI, add this to `~/.copilot/mcp-config.json`: + + ```json + { + "mcpServers": { + "s3m": { + "type": "http", + "url": "https://s3m.olcf.ornl.gov/olcf/mcp", + "headers": { + "Authorization": "${S3M_TOKEN}" + }, + "tools": ["*"] + } + } + } + ``` + +4. Start your agent without putting the token in your shell history (for Bash; agents, translate for the user's shell): + + read -s S3M_TOKEN + export S3M_TOKEN + # claude|codex|copilot + +**Do not paste your token into this conversation or pass it to this tool.** + +**Never post your token online or commit it to Git. If you do, revoke it immediately in myOLCF.** + +Agents: please help the user understand how to authenticate to S3M's MCP server safely. Advocate for using `read -s`. Carefully help the user set up their `.mcp.json`, Codex `config.toml`, Copilot CLI `mcp-config.json`, or other configurations securely. Ensure you do not duplicate existing S3M MCP registrations. diff --git a/spack.yaml b/spack.yaml index e4d3fe6..7df9226 100644 --- a/spack.yaml +++ b/spack.yaml @@ -6,5 +6,5 @@ spack: unify: true specs: - - go@1.27.1 + - go@1.27.2 - protobuf@3.29.3 diff --git a/vendor/golang.org/x/net/http2/flow.go b/vendor/golang.org/x/net/http2/flow.go index b7dbd18..5066a39 100644 --- a/vendor/golang.org/x/net/http2/flow.go +++ b/vendor/golang.org/x/net/http2/flow.go @@ -10,6 +10,9 @@ package http2 // flow control window update. const inflowMinRefresh = 4 << 10 +// maxFlowWindow is the maximum size of a flow control window. +const maxFlowWindow = (1 << 31) - 1 + // inflow accounts for an inbound flow control window. // It tracks both the latest window sent to the peer (used for enforcement) // and the accumulated unsent window. @@ -74,47 +77,99 @@ func takeInflows(f1, f2 *inflow, n uint32) bool { return true } -// outflow is the outbound flow control window's size. +// connOutflow is connection-level outbound flow control. +type connOutflow struct { + initial int32 // SETTINGS_INITIAL_WINDOW_SIZE, changes with settings updates + n int32 // connection-level flow control window + flowErr bool // set when a flow control error is encountered +} + +func (f *connOutflow) init() { + f.initial = initialWindowSize // initial stream window size + f.n = initialWindowSize // current connection window size +} + +func (f *connOutflow) changeInitialWindowSize(size int64) bool { + if size > maxFlowWindow { + f.flowErr = true + return false + } + f.initial = int32(size) + return true +} + +func (f *connOutflow) add(n int32) bool { + sum := int64(f.n) + int64(n) + if sum > maxFlowWindow { + f.flowErr = true + return false + } + f.n += n + return true +} + +// outflow is the stream-level outbound flow control window's size. type outflow struct { _ incomparable - // n is the number of DATA bytes we're allowed to send. - // An outflow is kept both on a conn and a per-stream. - n int32 + // delta is the difference between the stream's flow control window and + // the connection's initial window size (conn.initial). + // + // Another view is that delta is the number of flow control bytes provided to this + // stream in WINDOW_UPDATE frames, less the number of bytes sent on the stream. + delta int32 // conn points to the shared connection-level outflow that is - // shared by all streams on that conn. It is nil for the outflow - // that's on the conn directly. - conn *outflow + // shared by all streams on that conn. + conn *connOutflow } -func (f *outflow) setConnFlow(cf *outflow) { f.conn = cf } - -func (f *outflow) available() int32 { - n := f.n - if f.conn != nil && f.conn.n < n { - n = f.conn.n +func (f *outflow) available() (int32, bool) { + if f.conn == nil { + return maxFlowWindow, true // only happens in tests } - return n + if f.conn.flowErr { + // Block all sending once any stream observes a flow control error. + return 0, false + } + n := int64(f.conn.initial) + int64(f.delta) + if n > maxFlowWindow { + f.conn.flowErr = true + return 0, false + } + return min(int32(n), f.conn.n), true } func (f *outflow) take(n int32) { - if n > f.available() { - panic("internal error: took too much") + if f.conn == nil { + return // only happens in tests } - f.n -= n - if f.conn != nil { - f.conn.n -= n + avail, _ := f.available() + if n > avail { + panic("internal error: took too much") } + f.delta -= n + f.conn.n -= n } // add adds n bytes (positive or negative) to the flow control window. // It returns false if the sum would exceed 2^31-1. func (f *outflow) add(n int32) bool { - sum := f.n + n - if (sum > n) == (f.n > 0) { - f.n = sum - return true + if f.conn == nil { + return true // only happens in tests + } + avail := int64(f.conn.initial) + int64(f.delta) + if avail > maxFlowWindow { + // An earlier change to the initial window pushed this stream over the limit. + // This is a connection-level flow control error. + f.conn.flowErr = true + return false } - return false + if avail+int64(n) > maxFlowWindow { + // This update would push the stream over the limit. + // This is a stream-level flow control error. + return false + } + f.delta += n + return true } diff --git a/vendor/golang.org/x/net/http2/frame.go b/vendor/golang.org/x/net/http2/frame.go index be75bad..3363bba 100644 --- a/vendor/golang.org/x/net/http2/frame.go +++ b/vendor/golang.org/x/net/http2/frame.go @@ -1711,7 +1711,8 @@ func (fr *Framer) readMetaFrame(hf *HeadersFrame) (Frame, error) { mh := &MetaHeadersFrame{ HeadersFrame: hf, } - var remainSize = fr.maxHeaderListSize() + var headersRemainSize = fr.maxHeaderListSize() + var trailersRemainSize = fr.maxHeaderListSize() var sawRegular bool var invalid error // pseudo header field errors @@ -1743,14 +1744,29 @@ func (fr *Framer) readMetaFrame(hf *HeadersFrame) (Frame, error) { return } - size := hf.Size() - if size > remainSize { + var remainSize *uint32 + var size uint32 + if hf.Name == "trailer" { + remainSize = &trailersRemainSize + fieldCount := strings.Count(hf.Value, ",") + 1 + // Rather than actually constructing hpack.HeaderField for each + // trailer field and cumulatively adding its Size, just do the math + // manually to avoid unnecessary work. This does make it so + // whitespaces after comma are counted against the budget, but that + // should be innocuous. + size = uint32(len(hf.Value)-fieldCount+1) + uint32(fieldCount)*hpack.HeaderField{}.Size() + } else { + remainSize = &headersRemainSize + size = hf.Size() + } + if size > *remainSize { hdec.SetEmitEnabled(false) mh.Truncated = true - remainSize = 0 + headersRemainSize = 0 + trailersRemainSize = 0 return } - remainSize -= size + *remainSize -= size mh.Fields = append(mh.Fields, hf) }) @@ -1766,10 +1782,10 @@ func (fr *Framer) readMetaFrame(hf *HeadersFrame) (Frame, error) { // skip parsing the fragment and close the connection. // // "Too much" is either any CONTINUATION frame after we've already - // exceeded the max header list size (in which case remainSize is 0), - // or a frame whose encoded size is more than twice the remaining - // header list bytes we're willing to accept. - if int64(len(frag)) > int64(2*remainSize) { + // exceeded the max header list size (if so, both budgets are 0), or a + // frame whose encoded size is more than twice the remaining header + // list bytes we're willing to accept. + if int64(len(frag)) > 2*int64(headersRemainSize+trailersRemainSize) { if VerboseLogs { log.Printf("http2: header list too large") } diff --git a/vendor/golang.org/x/net/http2/server.go b/vendor/golang.org/x/net/http2/server.go index a7d2053..a6cd662 100644 --- a/vendor/golang.org/x/net/http2/server.go +++ b/vendor/golang.org/x/net/http2/server.go @@ -280,7 +280,6 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon doneServing: make(chan struct{}), clientMaxStreams: math.MaxUint32, // Section 6.5.2: "Initially, there is no limit to this value" advMaxStreams: conf.MaxConcurrentStreams, - initialStreamSendWindowSize: initialWindowSize, initialStreamRecvWindowSize: conf.MaxUploadBufferPerStream, maxFrameSize: initialMaxFrameSize, pingTimeout: conf.PingTimeout, @@ -320,7 +319,7 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon // These start at the RFC-specified defaults. If there is a higher // configured value for inflow, that will be updated when we send a // WINDOW_UPDATE shortly after sending SETTINGS. - sc.flow.add(initialWindowSize) + sc.flow.init() sc.inflow.init(initialWindowSize) sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf) sc.hpackEncoder.SetMaxDynamicTableSizeLimit(conf.MaxEncoderHeaderTableSize) @@ -427,7 +426,7 @@ type serverConn struct { wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes bodyReadCh chan bodyReadMsg // from handlers -> serve serveMsgCh chan interface{} // misc messages & code to send to / run on the serve loop - flow outflow // conn-wide (not stream-specific) outbound flow control + flow connOutflow // conn-wide (not stream-specific) outbound flow control inflow inflow // conn-wide inbound flow control tlsState *tls.ConnectionState // shared by all handlers, like net/http remoteAddrStr string @@ -441,6 +440,9 @@ type serverConn struct { sawFirstSettings bool // got the initial SETTINGS frame after the preface needToSendSettingsAck bool unackedSettings int // how many SETTINGS have we sent without ACKs? + pendingEncoderTableSize bool // peer changed SETTINGS_HEADER_TABLE_SIZE; apply to hpackEncoder before the next frame write + encoderTableSizeMin uint32 // smallest SETTINGS_HEADER_TABLE_SIZE since the last apply + encoderTableSize uint32 // latest SETTINGS_HEADER_TABLE_SIZE queuedControlFrames int // control frames in the writeSched queue clientMaxStreams uint32 // SETTINGS_MAX_CONCURRENT_STREAMS from client (our PUSH_PROMISE limit) advMaxStreams uint32 // our SETTINGS_MAX_CONCURRENT_STREAMS advertised the client @@ -451,7 +453,6 @@ type serverConn struct { maxPushPromiseID uint32 // ID of the last push promise (even), or 0 if there have been no pushes streams map[uint32]*stream unstartedHandlers []unstartedHandler - initialStreamSendWindowSize int32 initialStreamRecvWindowSize int32 maxFrameSize int32 peerMaxHeaderListSize uint32 // zero means unknown (default) @@ -521,7 +522,8 @@ type stream struct { // immutable: sc *serverConn id uint32 - body *pipe // non-nil if expecting DATA frames + body *pipe // non-nil if expecting DATA frames + reqBody *requestBody cw closeWaiter // closed wait stream transitions to closed state ctx context.Context cancelCtx func() @@ -1172,6 +1174,16 @@ func (sc *serverConn) startFrameWrite(wr FrameWriteRequest) { sc.writingFrame = true sc.needsFrameFlush = true + if sc.pendingEncoderTableSize { + // hpackEncoder may be in use by writeFrameAsync, so SETTINGS + // changes to it are deferred until no frame is being written. + // Replaying the smallest size before the latest one keeps the + // encoder's view identical to having applied every change + // (RFC 7541, Section 4.2). + sc.pendingEncoderTableSize = false + sc.hpackEncoder.SetMaxDynamicTableSize(sc.encoderTableSizeMin) + sc.hpackEncoder.SetMaxDynamicTableSize(sc.encoderTableSize) + } if wr.write.staysWithinBuffer(sc.bw.Available()) { sc.writingFrameAsync = false err := wr.write.writeFrame(sc) @@ -1271,6 +1283,11 @@ func (sc *serverConn) scheduleFrameWrite() { } sc.inFrameScheduleLoop = true for !sc.writingFrameAsync { + if sc.flow.flowErr && (!sc.inGoAway || sc.goAwayCode == ErrCodeNo) { + sc.inGoAway = true + sc.needToSendGoAway = true + sc.goAwayCode = ErrCodeFlowControl + } if sc.needToSendGoAway { sc.needToSendGoAway = false sc.startFrameWrite(FrameWriteRequest{ @@ -1294,6 +1311,9 @@ func (sc *serverConn) scheduleFrameWrite() { sc.startFrameWrite(wr) continue } + if sc.flow.flowErr { + continue + } } if sc.needsFrameFlush { sc.startFrameWrite(FrameWriteRequest{write: flushFrameWriter{}}) @@ -1526,6 +1546,10 @@ func (sc *serverConn) processWindowUpdate(f *WindowUpdateFrame) error { return nil } if !st.flow.add(int32(f.Increment)) { + if st.flow.conn.flowErr { + // This is a lazily-detected connection-level flow control error. + return sc.countError("bad_flow", ConnectionError(ErrCodeFlowControl)) + } return sc.countError("bad_flow", streamError(f.StreamID, ErrCodeFlowControl)) } default: // connection-level flow control @@ -1584,10 +1608,6 @@ func (sc *serverConn) closeStream(st *stream, err error) { } } if p := st.body; p != nil { - // Return any buffered unread bytes worth of conn-level flow control. - // See golang.org/issue/16481 - sc.sendWindowUpdate(nil, p.Len()) - p.CloseWithError(err) } if e, ok := err.(StreamError); ok { @@ -1641,7 +1661,12 @@ func (sc *serverConn) processSetting(s Setting) error { } switch s.ID { case SettingHeaderTableSize: - sc.hpackEncoder.SetMaxDynamicTableSize(s.Val) + // Applied by startFrameWrite; see comment there. + if !sc.pendingEncoderTableSize || s.Val < sc.encoderTableSizeMin { + sc.encoderTableSizeMin = s.Val + } + sc.encoderTableSize = s.Val + sc.pendingEncoderTableSize = true case SettingEnablePush: sc.pushEnabled = s.Val != 0 case SettingMaxConcurrentStreams: @@ -1672,28 +1697,14 @@ func (sc *serverConn) processSetting(s Setting) error { func (sc *serverConn) processSettingInitialWindowSize(val uint32) error { sc.serveG.check() - // Note: val already validated to be within range by - // processSetting's Valid call. - - // "A SETTINGS frame can alter the initial flow control window - // size for all current streams. When the value of - // SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST - // adjust the size of all stream flow control windows that it - // maintains by the difference between the new value and the - // old value." - old := sc.initialStreamSendWindowSize - sc.initialStreamSendWindowSize = int32(val) - growth := int32(val) - old // may be negative - for _, st := range sc.streams { - if !st.flow.add(growth) { - // 6.9.2 Initial Flow Control Window Size - // "An endpoint MUST treat a change to - // SETTINGS_INITIAL_WINDOW_SIZE that causes any flow - // control window to exceed the maximum size as a - // connection error (Section 5.4.1) of type - // FLOW_CONTROL_ERROR." - return sc.countError("setting_win_size", ConnectionError(ErrCodeFlowControl)) - } + if !sc.flow.changeInitialWindowSize(int64(val)) { + // 6.9.2 Initial Flow Control Window Size + // "An endpoint MUST treat a change to + // SETTINGS_INITIAL_WINDOW_SIZE that causes any flow + // control window to exceed the maximum size as a + // connection error (Section 5.4.1) of type + // FLOW_CONTROL_ERROR." + return sc.countError("setting_win_size", ConnectionError(ErrCodeFlowControl)) } return nil } @@ -1966,7 +1977,7 @@ func (sc *serverConn) processHeaders(f *MetaHeadersFrame) error { if st.reqTrailer != nil { st.trailer = make(http.Header) } - st.body = req.Body.(*requestBody).pipe // may be nil + st.body = st.reqBody.pipe // may be nil st.declBodyBytes = req.ContentLength handler := sc.handler.ServeHTTP @@ -1989,7 +2000,7 @@ func (sc *serverConn) processHeaders(f *MetaHeadersFrame) error { st.readDeadline = time.AfterFunc(sc.hs.ReadTimeout, st.onReadTimeout) } - return sc.scheduleHandler(id, rw, req, handler) + return sc.scheduleHandler(st, rw, req, handler) } func (sc *serverConn) upgradeRequest(req *http.Request) { @@ -2102,7 +2113,6 @@ func (sc *serverConn) newStream(id, pusherID uint32, state streamState, priority } st.cw.Init() st.flow.conn = &sc.flow // link to conn-level counter - st.flow.add(sc.initialStreamSendWindowSize) st.inflow.init(sc.initialStreamRecvWindowSize) if sc.hs.WriteTimeout > 0 { st.writeDeadline = time.AfterFunc(sc.hs.WriteTimeout, st.onWriteTimeout) @@ -2184,7 +2194,7 @@ func (sc *serverConn) newWriterAndRequest(st *stream, f *MetaHeadersFrame) (*res } else { req.ContentLength = -1 } - req.Body.(*requestBody).pipe = &pipe{ + st.reqBody.pipe = &pipe{ b: &dataBuffer{expected: req.ContentLength}, } } @@ -2204,7 +2214,7 @@ func (sc *serverConn) newWriterAndRequestNoBody(st *stream, rp httpcommon.Server return nil, nil, sc.countError(res.InvalidReason, streamError(st.id, ErrCodeProtocol)) } - body := &requestBody{ + st.reqBody = &requestBody{ conn: sc, stream: st, needsContinue: res.NeedsContinue, @@ -2220,7 +2230,7 @@ func (sc *serverConn) newWriterAndRequestNoBody(st *stream, rp httpcommon.Server ProtoMinor: 0, TLS: tlsState, Host: rp.Authority, - Body: body, + Body: st.reqBody, Trailer: res.Trailer, }).WithContext(st.ctx) rw := sc.newResponseWriter(st, req) @@ -2244,11 +2254,12 @@ type unstartedHandler struct { rw *responseWriter req *http.Request handler func(http.ResponseWriter, *http.Request) + body *pipe } // scheduleHandler starts a handler goroutine, // or schedules one to start as soon as an existing handler finishes. -func (sc *serverConn) scheduleHandler(streamID uint32, rw *responseWriter, req *http.Request, handler func(http.ResponseWriter, *http.Request)) error { +func (sc *serverConn) scheduleHandler(st *stream, rw *responseWriter, req *http.Request, handler func(http.ResponseWriter, *http.Request)) error { sc.serveG.check() maxHandlers := sc.advMaxStreams if sc.curHandlers < maxHandlers { @@ -2260,10 +2271,11 @@ func (sc *serverConn) scheduleHandler(streamID uint32, rw *responseWriter, req * return sc.countError("too_many_early_resets", ConnectionError(ErrCodeEnhanceYourCalm)) } sc.unstartedHandlers = append(sc.unstartedHandlers, unstartedHandler{ - streamID: streamID, + streamID: st.id, rw: rw, req: req, handler: handler, + body: st.body, }) return nil } @@ -2277,6 +2289,10 @@ func (sc *serverConn) handlerDone() { u := sc.unstartedHandlers[i] if sc.streams[u.streamID] == nil { // This stream was reset before its goroutine had a chance to start. + if u.body != nil { + u.body.BreakWithError(errClosedBody) + sc.sendWindowUpdate(nil, u.body.Len()) + } continue } if sc.curHandlers >= maxHandlers { @@ -2298,6 +2314,12 @@ func (sc *serverConn) runHandler(rw *responseWriter, req *http.Request, handler didPanic := true defer func() { rw.rws.stream.cancelCtx() + if b := rw.rws.stream.reqBody; b != nil { + // Closing the body refunds flow control credit for any unconsumed data. + // (reqBody is nil for Upgrade: h2c requests, but those do not use flow + // control for the request body.) + b.Close() + } if req.MultipartForm != nil { req.MultipartForm.RemoveAll() } @@ -2396,7 +2418,7 @@ func (sc *serverConn) noteBodyReadFromHandler(st *stream, n int, err error) { func (sc *serverConn) noteBodyRead(st *stream, n int) { sc.serveG.check() sc.sendWindowUpdate(nil, n) // conn-level - if st.state != stateHalfClosedRemote && st.state != stateClosed { + if st != nil && st.state != stateHalfClosedRemote && st.state != stateClosed { // Don't send this WINDOW_UPDATE if the stream is closed // remotely. sc.sendWindowUpdate(st, n) @@ -2444,6 +2466,9 @@ func (b *requestBody) Close() error { b.closeOnce.Do(func() { if b.pipe != nil { b.pipe.BreakWithError(errClosedBody) + if unread := b.pipe.Len(); unread > 0 { + b.conn.noteBodyReadFromHandler(nil, unread, errClosedBody) + } } }) return nil @@ -2461,9 +2486,6 @@ func (b *requestBody) Read(p []byte) (n int, err error) { if err == io.EOF { b.sawEOF = true } - if b.conn == nil { - return - } b.conn.noteBodyReadFromHandler(b.stream, n, err) return } diff --git a/vendor/golang.org/x/net/http2/transport.go b/vendor/golang.org/x/net/http2/transport.go index 088eb9b..d894c60 100644 --- a/vendor/golang.org/x/net/http2/transport.go +++ b/vendor/golang.org/x/net/http2/transport.go @@ -27,6 +27,7 @@ import ( "net/http" "net/http/httptrace" "net/textproto" + "slices" "strconv" "sync" "sync/atomic" @@ -203,7 +204,7 @@ type ClientConn struct { mu sync.Mutex // guards following cond *sync.Cond // hold mu; broadcast on flow/closed changes - flow outflow // our conn-level flow control quota (cs.outflow is per stream) + flow connOutflow // our conn-level flow control quota (cs.outflow is per stream) inflow inflow // peer's conn-level flow control doNotReuse bool // whether conn is marked to not be reused for any future requests closing bool @@ -227,7 +228,6 @@ type ClientConn struct { maxConcurrentStreams uint32 peerMaxHeaderListSize uint64 peerMaxHeaderTableSize uint32 - initialWindowSize uint32 initialStreamRecvWindowSize int32 readIdleTimeout time.Duration pingTimeout time.Duration @@ -472,7 +472,6 @@ func (t *Transport) newClientConn(c net.Conn, singleUse bool, internalStateHook readerDone: make(chan struct{}), nextStreamID: 1, maxFrameSize: 16 << 10, // spec default - initialWindowSize: 65535, // spec default initialStreamRecvWindowSize: conf.MaxUploadBufferPerStream, maxConcurrentStreams: initialMaxConcurrentStreams, // "infinite", per spec. Use a smaller value until we have received server settings. strictMaxConcurrentStreams: conf.StrictMaxConcurrentRequests, @@ -497,7 +496,7 @@ func (t *Transport) newClientConn(c net.Conn, singleUse bool, internalStateHook } cc.cond = sync.NewCond(&cc.mu) - cc.flow.add(int32(initialWindowSize)) + cc.flow.init() // TODO: adjust this writer size to account for frame size + // MTU + crypto/tls record padding. @@ -876,6 +875,33 @@ func (cc *ClientConn) sendGoAway() error { return nil } +func (cc *ClientConn) goAwayAndClose(code ErrCode) { + cc.mu.Lock() + closed := cc.closed + cc.closing = true + cc.closed = true + cc.mu.Unlock() + if closed { + return + } + if f := cc.fr.countError; f != nil { + f(fmt.Sprintf("conn_close_error_%s", code.stringToken())) + } + done := make(chan struct{}) + go func() { + defer close(done) + cc.wmu.Lock() + cc.fr.WriteGoAway(0, code, nil) + cc.bw.Flush() + cc.wmu.Unlock() + }() + select { + case <-done: + case <-time.After(250 * time.Millisecond): + } + cc.closeForError(fmt.Errorf("http2: closing connection with %v", code)) +} + // closes the client connection immediately. In-flight requests are interrupted. // err is sent to streams. func (cc *ClientConn) closeForError(err error) { @@ -1319,7 +1345,9 @@ func (cs *clientStream) cleanupWriteRequest(err error) { } if err != nil { cs.abortStream(err) // possibly redundant, but harmless - if cs.sentHeaders { + if ce, ok := err.(ConnectionError); ok { + cc.goAwayAndClose(ErrCode(ce)) + } else if cs.sentHeaders { if se, ok := err.(StreamError); ok { if se.Cause != errFromPeer { cc.writeStreamReset(cs.ID, se.Code, false, err) @@ -1648,8 +1676,12 @@ func (cs *clientStream) awaitFlowControl(maxBytes int) (taken int32, err error) return 0, errRequestCanceled default: } - if a := cs.flow.available(); a > 0 { - take := a + avail, ok := cs.flow.available() + if !ok { + return 0, ConnectionError(ErrCodeFlowControl) + } + if avail > 0 { + take := avail if int(take) > maxBytes { take = int32(maxBytes) // can't truncate int; take is int32 @@ -1710,8 +1742,7 @@ type resAndError struct { // requires cc.mu be held. func (cc *ClientConn) addStreamLocked(cs *clientStream) { - cs.flow.add(int32(cc.initialWindowSize)) - cs.flow.setConnFlow(&cc.flow) + cs.flow.conn = &cc.flow cs.inflow.init(cc.initialStreamRecvWindowSize) cs.ID = cc.nextStreamID cc.nextStreamID += 2 @@ -2095,18 +2126,42 @@ func (rl *clientConnReadLoop) handleResponse(cs *clientStream, f *MetaHeadersFra return nil, nil } + // Delete various headers that might mess up framing for HTTP/1. This is + // not a problem for HTTP/2, but someone might use HTTP/2 transport as a + // reverse proxy which forwards the response to an HTTP/1 client. Our + // HTTP/1 client transport will properly reject improper headers such as + // multiple conflicting Content-Length headers, but other implementations + // might not. + // TODO: just reject such responses? We deleted them for compatibility + // since this was done in a security fix (go.dev/issue/81115). However, + // rejecting them seems entirely reasonable and relatively safe. + + // Connection-specific header fields must not appear in an HTTP/2 message, + // and any message containing them is malformed. RFC 9113, Section 8.2.2. + for _, k := range connHeaders { + delete(res.Header, k) + } res.ContentLength = -1 - if clens := res.Header["Content-Length"]; len(clens) == 1 { - if cl, err := strconv.ParseUint(clens[0], 10, 63); err == nil { - res.ContentLength = int64(cl) + if clens, ok := res.Header["Content-Length"]; ok { + // Repeated Content-Length values may be collapsed into one only if + // they are identical per RFC 9110 Section 8.6. + // No need to trim whitespace, HTTP/2 header values must not have + // extraneous whitespace per RFC 9113 Section 8.2.1. + // Non-canonical headers are already rejected by our framer at this + // point. + conflicting := slices.ContainsFunc(clens[1:], func(clen string) bool { + return clen != clens[0] + }) + cl, err := strconv.ParseUint(clens[0], 10, 63) + if conflicting || err != nil { + delete(res.Header, "Content-Length") } else { - // TODO: care? unlike http/1, it won't mess up our framing, so it's - // more safe smuggling-wise to ignore. + res.Header["Content-Length"] = clens[:1] + res.ContentLength = int64(cl) } - } else if len(clens) > 1 { - // TODO: care? unlike http/1, it won't mess up our framing, so it's - // more safe smuggling-wise to ignore. - } else if f.StreamEnded() && !cs.isHead { + } + + if res.ContentLength < 0 && f.StreamEnded() && !cs.isHead { res.ContentLength = 0 } @@ -2417,6 +2472,10 @@ const ( func (rl *clientConnReadLoop) streamByID(id uint32, headerOrData bool) *clientStream { rl.cc.mu.Lock() defer rl.cc.mu.Unlock() + return rl.streamByIDLocked(id, headerOrData) +} + +func (rl *clientConnReadLoop) streamByIDLocked(id uint32, headerOrData bool) *clientStream { if headerOrData { // Work around an unfortunate gRPC behavior. // See comment on ClientConn.rstStreamPingsBlocked for details. @@ -2499,24 +2558,10 @@ func (rl *clientConnReadLoop) processSettingsNoWrite(f *SettingsFrame) error { case SettingMaxHeaderListSize: cc.peerMaxHeaderListSize = uint64(s.Val) case SettingInitialWindowSize: - // Values above the maximum flow-control - // window size of 2^31-1 MUST be treated as a - // connection error (Section 5.4.1) of type - // FLOW_CONTROL_ERROR. - if s.Val > math.MaxInt32 { + if !cc.flow.changeInitialWindowSize(int64(s.Val)) { return ConnectionError(ErrCodeFlowControl) } - - // Adjust flow control of currently-open - // frames by the difference of the old initial - // window size and this one. - delta := int32(s.Val) - int32(cc.initialWindowSize) - for _, cs := range cc.streams { - cs.flow.add(delta) - } cc.cond.Broadcast() - - cc.initialWindowSize = s.Val case SettingHeaderTableSize: cc.henc.SetMaxDynamicTableSize(s.Val) cc.peerMaxHeaderTableSize = s.Val @@ -2558,29 +2603,30 @@ func (rl *clientConnReadLoop) processSettingsNoWrite(f *SettingsFrame) error { func (rl *clientConnReadLoop) processWindowUpdate(f *WindowUpdateFrame) error { cc := rl.cc - cs := rl.streamByID(f.StreamID, notHeaderOrDataFrame) - if f.StreamID != 0 && cs == nil { - return nil - } - cc.mu.Lock() defer cc.mu.Unlock() - - fl := &cc.flow - if cs != nil { - fl = &cs.flow - } - if !fl.add(int32(f.Increment)) { - // For stream, the sender sends RST_STREAM with an error code of FLOW_CONTROL_ERROR - if cs != nil { + if f.StreamID == 0 { + if !cc.flow.add(int32(f.Increment)) { + return ConnectionError(ErrCodeFlowControl) + } + } else { + cs := rl.streamByIDLocked(f.StreamID, notHeaderOrDataFrame) + if cs == nil { + return nil + } + if !cs.flow.add(int32(f.Increment)) { + if cs.flow.conn.flowErr { + // This is a lazily-detected connection-level flow control error. + return ConnectionError(ErrCodeFlowControl) + } + // For stream, the sender sends RST_STREAM with + // an error code of FLOW_CONTROL_ERROR. rl.endStreamErrorLocked(cs, StreamError{ StreamID: f.StreamID, Code: ErrCodeFlowControl, }) return nil } - - return ConnectionError(ErrCodeFlowControl) } cc.cond.Broadcast() return nil diff --git a/vendor/golang.org/x/net/http2/transport_wrap.go b/vendor/golang.org/x/net/http2/transport_wrap.go index 34dfb3c..eeaa95f 100644 --- a/vendor/golang.org/x/net/http2/transport_wrap.go +++ b/vendor/golang.org/x/net/http2/transport_wrap.go @@ -119,33 +119,27 @@ type http2TransportContextKey struct{} // DialFromContext dials a new connection using the http2.Transport's DialTLS/DialTLSContext. func (t transportConfig) DialFromContext(ctx context.Context, network, address string) (net.Conn, error) { - if ctx.Value(http2TransportContextKey{}) == nil { + dial, _ := ctx.Value(http2TransportContextKey{}).(*transportRoundTripState) + if dial == nil { // We're being called from a RoundTrip that did not start with an http2.Transport. // Use the http.Transport's dialer. return nil, errors.ErrUnsupported } - - tlsConf := t.t.TLSClientConfig - if tlsConf == nil { - tlsConf = &tls.Config{} - } else { - tlsConf = tlsConf.Clone() - } - if !slices.Contains(tlsConf.NextProtos, "h2") { - tlsConf.NextProtos = append([]string{"h2"}, tlsConf.NextProtos...) - } - if tlsConf.ServerName == "" { - host, _, err := net.SplitHostPort(address) - if err == nil { - tlsConf.ServerName = host - } - } - return t.t.dialTLS(ctx, network, address, tlsConf) + return t.t.dialForRoundTrip(ctx, dial, network, address) } type transportInternal struct { initOnce sync.Once lazyt1 *http.Transport + + dialMu sync.Mutex + dials map[string]*transportDialState +} + +type transportDialState struct { + dialc chan struct{} // close when dial returns + dialErr error // dial result, set before dialc is closed + rtdonec chan struct{} // closed when RoundTrip initiating the dial returns } func (t *Transport) init() *http.Transport { @@ -167,6 +161,52 @@ func (t *Transport) configure(t1 *http.Transport) { } } +// transportRoundTripState is the state of the dial for an http2.Transport.RoundTrip. +type transportRoundTripState struct { + mu sync.Mutex + rtdone bool // set when RoundTrip returns + rtdonec chan struct{} // closed when RoundTrip returns + gotconnc chan struct{} // closed when GotConn hook is called +} + +// startDial is called when a dial starts. +// +// It returns done=true if the RoundTrip has already returned, +// in which case we should skip dialing. +func (dial *transportRoundTripState) startDial() (gotconnc chan struct{}, done bool) { + dial.mu.Lock() + defer dial.mu.Unlock() + if dial.rtdone { + return nil, true + } + if dial.rtdonec == nil { + dial.rtdonec = make(chan struct{}) + } + dial.gotconnc = make(chan struct{}) + return dial.gotconnc, false +} + +// gotConn is called when RoundTrip gets a connection. +// This may happen multiple times per RoundTrip, if a request is retried. +func (dial *transportRoundTripState) gotConn() { + dial.mu.Lock() + defer dial.mu.Unlock() + if dial.gotconnc != nil { + close(dial.gotconnc) + dial.gotconnc = nil + } +} + +// roundTripDone is called when RoundTrip returns. +func (dial *transportRoundTripState) roundTripDone() { + dial.mu.Lock() + defer dial.mu.Unlock() + dial.rtdone = true + if dial.rtdonec != nil { + close(dial.rtdonec) + } +} + func (t *Transport) roundTripOpt(req *http.Request, opt RoundTripOpt) (*http.Response, error) { t1 := t.init() @@ -186,12 +226,146 @@ func (t *Transport) roundTripOpt(req *http.Request, opt RoundTripOpt) (*http.Res // Both http.Transport and http2.Transport allow the user to provide a custom // dial function, and historically you only get the dial function from the // Transport you're calling RoundTrip on. - ctx := context.WithValue(req.Context(), http2TransportContextKey{}, t) + // + // In addition, http2.Transport coalesces dials, which http.Transport historically + // has not. + dial := &transportRoundTripState{} + defer dial.roundTripDone() + ctx := context.WithValue(req.Context(), http2TransportContextKey{}, dial) + ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{ + GotConn: func(httptrace.GotConnInfo) { + // Tell dialForRoundTrip that we have received a connection. + dial.gotConn() + }, + }) req = req.WithContext(ctx) - return t1.RoundTrip(req) } +var errCoalescedDialAbandoned = errors.New("http2: abandoned coalesced dial") + +const coalescedDialRetryTimeout = 50 * time.Millisecond + +// dialForRoundTrip is called (indirectly) by net/http when dialing a new connection +// for a request call which originated as an http2.Transport.RoundTrip call. +// +// It uses the http2.Transport's Dial hooks and coalesces dials. +func (t *Transport) dialForRoundTrip(ctx context.Context, dial *transportRoundTripState, network, address string) (net.Conn, error) { + gotconnc, rtdone := dial.startDial() + if rtdone { + // RoundTrip returned, no need for this dial to proceed. + return nil, errCoalescedDialAbandoned + } + + var state *transportDialState + for { + // The first dial to an address registers itself with t.dials. + // When the dial finishes, it records the outcome and removes itself from t.dials + // so future dials will start a new round of coalescing. + // + // Subsequent dials observe that a dial is in progress, and skip dialing. + t.dialMu.Lock() + if t.dials == nil { + t.dials = make(map[string]*transportDialState) + } + leader := t.dials[address] + if leader == nil { + // No entry in t.dials to coalesce with. Add ourselves as leader. + state = &transportDialState{ + dialc: make(chan struct{}), + rtdonec: dial.rtdonec, + } + t.dials[address] = state + } + t.dialMu.Unlock() + if leader == nil { + // We are the leader, so we should dial for real. + break + } + + // Coalesce with a previous dial. + var ( + done = false + leaderdialc = leader.dialc + leaderrtdonec = leader.rtdonec + timerc <-chan time.Time + ) + for !done { + if leaderdialc == nil && leaderrtdonec == nil && timerc == nil { + // The leader finished dialing, and its RoundTrip finished. + // + // There are several possibilities, which reduce to: + // - The RoundTrip we are dialing for is about to receive + // a connection (possibly the one the leader just dialed), + // but we haven't observed it yet. + // - There's something wrong with the leader's connection, + // such as a TLS handshake failure. + // + // We have no good way to distinguish between these cases. + // + // Wait a short time for the leader's connection (or some + // other usable connection) to be delivered to our RoundTrip. + // If it is not, start a dial of our own. + timerc = time.After(coalescedDialRetryTimeout) + } + select { + case <-leaderdialc: + // The leader's dial completed. + if leader.dialErr != nil { + return nil, leader.dialErr + } + leaderdialc = nil + case <-leaderrtdonec: + // The leader's RoundTrip completed. + leaderrtdonec = nil + case <-timerc: + // The leader's dial and RoundTrip completed, and some time + // has passed. Assume we're not getting a connection and + // restart the dial process. + done = true + case <-dial.rtdonec: + // Our RoundTrip returned. + return nil, errCoalescedDialAbandoned + case <-gotconnc: + // Our RoundTrip got a connection. + return nil, errCoalescedDialAbandoned + case <-ctx.Done(): + // net/http abandoned the dial. + return nil, ctx.Err() + } + } + } + + // Dial for real. + tlsConf := t.TLSClientConfig + if tlsConf == nil { + tlsConf = &tls.Config{} + } else { + tlsConf = tlsConf.Clone() + } + if !slices.Contains(tlsConf.NextProtos, "h2") { + tlsConf.NextProtos = append([]string{"h2"}, tlsConf.NextProtos...) + } + if tlsConf.ServerName == "" { + host, _, err := net.SplitHostPort(address) + if err == nil { + tlsConf.ServerName = host + } + } + nc, err := t.dialTLS(ctx, network, address, tlsConf) + + // Remove this dial from t.dials, so future dial attempts will not coalesce with it. + t.dialMu.Lock() + delete(t.dials, address) + t.dialMu.Unlock() + + // Notify any dial coalesced with this one that we are done. + state.dialErr = err + close(state.dialc) + + return nc, err +} + func (t *Transport) closeIdleConnections() { t1 := t.init() t1.CloseIdleConnections() diff --git a/vendor/golang.org/x/net/http2/writesched.go b/vendor/golang.org/x/net/http2/writesched.go index 36ad5e3..f1903a8 100644 --- a/vendor/golang.org/x/net/http2/writesched.go +++ b/vendor/golang.org/x/net/http2/writesched.go @@ -79,10 +79,11 @@ func (wr FrameWriteRequest) Consume(n int32) (FrameWriteRequest, FrameWriteReque } // Might need to split after applying limits. - allowed := wr.stream.flow.available() - if n < allowed { - allowed = n + avail, ok := wr.stream.flow.available() + if !ok { + return empty, empty, 0 } + allowed := min(n, avail) if wr.stream.sc.maxFrameSize < allowed { allowed = wr.stream.sc.maxFrameSize } diff --git a/vendor/golang.org/x/net/internal/httpcommon/gzip.go b/vendor/golang.org/x/net/internal/httpcommon/gzip.go new file mode 100644 index 0000000..819dd65 --- /dev/null +++ b/vendor/golang.org/x/net/internal/httpcommon/gzip.go @@ -0,0 +1,132 @@ +// Copyright 2026 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package httpcommon + +import ( + "compress/flate" + "compress/gzip" + "errors" + "io" + "io/fs" + "sync" +) + +var errConcurrentRead = errors.New("http: concurrent read on response body") + +// incomparable is a zero-width, non-comparable type. Adding it to a struct +// makes that struct also non-comparable, and generally doesn't add +// any size (as long as it's first). +type incomparable [0]func() + +// GzipReader wraps a response body so it can lazily +// get gzip.Reader from the pool on the first call to Read. +// After Close is called it puts gzip.Reader to the pool immediately +// if there is no Read in progress or later when Read completes. +type GzipReader struct { + _ incomparable + Body io.ReadCloser // underlying Response.Body + mu sync.Mutex // guards zr and zerr + zr *gzip.Reader // stores gzip reader from the pool between reads + zerr error // sticky gzip reader init error or sentinel value to detect concurrent read and read after close +} + +type eofReader struct{} + +func (eofReader) Read([]byte) (int, error) { return 0, io.EOF } +func (eofReader) ReadByte() (byte, error) { return 0, io.EOF } + +var gzipPool = sync.Pool{New: func() any { return new(gzip.Reader) }} + +// gzipPoolGet gets a gzip.Reader from the pool and resets it to read from r. +func gzipPoolGet(r io.Reader) (*gzip.Reader, error) { + zr := gzipPool.Get().(*gzip.Reader) + if err := zr.Reset(r); err != nil { + gzipPoolPut(zr) + return nil, err + } + return zr, nil +} + +// gzipPoolPut puts a gzip.Reader back into the pool. +func gzipPoolPut(zr *gzip.Reader) { + // Reset will allocate bufio.Reader if we pass it anything + // other than a flate.Reader, so ensure that it's getting one. + var r flate.Reader = eofReader{} + zr.Reset(r) + gzipPool.Put(zr) +} + +// acquire returns a gzip.Reader for reading response body. +// The reader must be released after use. +func (gz *GzipReader) acquire() (*gzip.Reader, error) { + gz.mu.Lock() + defer gz.mu.Unlock() + if gz.zerr != nil { + return nil, gz.zerr + } + if gz.zr == nil { + // gzipPoolGet might block indefinitely since it reads the gzip header. + // Therefore, drop mu temporarily when using gzipPoolGet. + // We set zerr to errConcurrentRead to prevent concurrent read + // even when mu is temporarily dropped. + gz.zerr = errConcurrentRead + gz.mu.Unlock() + zr, err := gzipPoolGet(gz.Body) + gz.mu.Lock() + // Guard against Close being called while gzipPoolGet is running. + if gz.zerr != errConcurrentRead { + if zr != nil { + gzipPoolPut(zr) + } + return nil, gz.zerr + } + gz.zr, gz.zerr = zr, err + if gz.zerr != nil { + return nil, gz.zerr + } + } + ret := gz.zr + gz.zr, gz.zerr = nil, errConcurrentRead + return ret, nil +} + +// release returns the gzip.Reader to the pool if Close was called during Read. +func (gz *GzipReader) release(zr *gzip.Reader) { + gz.mu.Lock() + defer gz.mu.Unlock() + if gz.zerr == errConcurrentRead { + gz.zr, gz.zerr = zr, nil + } else { // fs.ErrClosed + gzipPoolPut(zr) + } +} + +// close returns the gzip.Reader to the pool immediately or +// signals release to do so after Read completes. +func (gz *GzipReader) close() { + gz.mu.Lock() + defer gz.mu.Unlock() + if gz.zerr == nil && gz.zr != nil { + gzipPoolPut(gz.zr) + gz.zr = nil + } + gz.zerr = fs.ErrClosed +} + +func (gz *GzipReader) Read(p []byte) (n int, err error) { + zr, err := gz.acquire() + if err != nil { + return 0, err + } + defer gz.release(zr) + + return zr.Read(p) +} + +func (gz *GzipReader) Close() error { + gz.close() + + return gz.Body.Close() +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 51df1a6..fa5c925 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -60,7 +60,7 @@ github.com/urfave/cli/v3 # github.com/yosida95/uritemplate/v3 v3.0.2 ## explicit; go 1.14 github.com/yosida95/uritemplate/v3 -# golang.org/x/net v0.59.0 +# golang.org/x/net v0.60.0 ## explicit; go 1.26.0 golang.org/x/net/http/httpguts golang.org/x/net/http2