diff --git a/cli/commands/cluster.go b/cli/commands/cluster.go index 8764fc148..7e20409c6 100644 --- a/cli/commands/cluster.go +++ b/cli/commands/cluster.go @@ -101,6 +101,8 @@ func ClusterList(ctx *Context, opts struct { address := ccfg.Hostname if ccfg.ViaCloud { address = lipgloss.NewStyle().Foreground(theme.Muted).Render("via miren cloud") + } else if address == "" && ccfg.XID != "" && ccfg.Identity != "" { + address = lipgloss.NewStyle().Foreground(theme.Muted).Render("direct when reachable, otherwise via cloud") } else if host, port, err := net.SplitHostPort(address); err == nil { // Color the port portion gray for table display grayPort := lipgloss.NewStyle().Foreground(theme.Muted).Render(":" + port) diff --git a/cli/commands/cluster_add.go b/cli/commands/cluster_add.go index 426304442..a1f506456 100644 --- a/cli/commands/cluster_add.go +++ b/cli/commands/cluster_add.go @@ -327,6 +327,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { var clusterCert *clusterCertificate var allAddresses []string var clusterXID string + var selectedCluster *ClusterResponse // What the cluster is called in cloud, and whose it is. Empty in manual // mode, where cloud is never asked and neither is knowable. @@ -351,8 +352,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { } var ( - selectedCluster *ClusterResponse - localName string + localName string // Which of the undialable clusters cloud can still reach. Skipped // when the caller already said --via-cloud, since the answer would // change nothing. @@ -421,9 +421,9 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { ctx.Info("%s advertises no address this machine can dial, and cloud can reach it.", selectedCluster.Name) } + allAddresses = selectedCluster.APIAddresses if opts.viaCloud { - // No address and no certificate: both describe dialing the cluster, - // which is the thing this entry exists to avoid. + // Keep advertised addresses for automatic routing on later connections. // // The route itself is checked when the caller asked for it by name, // rather than when it was chosen above (which already probed). The @@ -440,9 +440,6 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { announceClusterAdd(ctx, selectedCluster.Name, localName, "routed through Miren Cloud") } else { - // Store all available addresses - allAddresses = selectedCluster.APIAddresses - // Try to connect to the cluster workingAddress, cert, err := tryConnectToCluster(ctx, selectedCluster, true) if err != nil { @@ -456,11 +453,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { ctx.Info("Could not reach %s directly, but cloud has a link to it.", selectedCluster.Name) - // The address and certificate belong to a route that does not - // work from here. Keeping either would write an entry that - // looks dialable and is not. opts.viaCloud = true - allAddresses = nil announceClusterAdd(ctx, selectedCluster.Name, localName, "routed through Miren Cloud") } else { @@ -489,7 +482,10 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { AllAddresses: allAddresses, Identity: identityName, XID: clusterXID, - ViaCloud: opts.viaCloud, + ViaCloud: explicitViaCloud, + } + if selectedCluster != nil { + clusterConfig.CACertFingerprint = selectedCluster.CACertFingerprint } // A cloud reached over plain HTTP is a development one, and routing through @@ -498,7 +494,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { // that cannot work, or one that quietly ships a credential in the clear // without recording that anywhere, are both worse than saying it out loud // here and putting the admission in the file. - if opts.viaCloud && identity != nil && strings.HasPrefix(identity.Issuer, "http://") { + if clusterXID != "" && identity != nil && strings.HasPrefix(identity.Issuer, "http://") { clusterConfig.Insecure = true ctx.Warn("%s is not encrypted, so commands to this cluster will send your credentials in the clear.", identity.Issuer) ctx.Warn("Recorded as insecure: true on the cluster. Use an https cloud for anything but local development.") @@ -573,9 +569,12 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { return nil, codedErrorf(codeConfigError, "failed to save cluster configuration: %w", err) } - if opts.viaCloud { + if explicitViaCloud { ctx.Completed("Successfully added cluster %q with identity %q, routed through Miren Cloud", clusterName, identityName) ctx.Info("Commands reach it over the connection it holds open to cloud, so it needs no address here.") + } else if opts.viaCloud { + ctx.Completed("Successfully added cluster %q with identity %q; reached through Miren Cloud while adding", clusterName, identityName) + ctx.Info("Future connections use direct when reachable, otherwise via cloud.") } else if identityName != "" { ctx.Completed("Successfully added cluster %q with identity %q at %s", clusterName, identityName, address) } else { @@ -597,6 +596,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { XID: clusterXID, Address: address, ViaCloud: opts.viaCloud, + ForceCloud: clusterConfig.ViaCloud, Identity: identityName, Insecure: clusterConfig.Insecure, Active: mainConfig.ActiveCluster() == clusterName, diff --git a/cli/commands/cluster_add_byname_test.go b/cli/commands/cluster_add_byname_test.go index 1b09db43f..693be0ea1 100644 --- a/cli/commands/cluster_add_byname_test.go +++ b/cli/commands/cluster_add_byname_test.go @@ -71,6 +71,37 @@ func TestAddClusterByNameRoutedThroughCloud(t *testing.T) { r.Error(err, "the cloud name is not also written") } +func TestAddClusterFallbackPreservesAutomaticRouting(t *testing.T) { + addresses := []string{"invalid::::"} + srv := fakeCloud(t, []ClusterResponse{ + {Name: "prod", XID: "cluster-prod", APIAddresses: addresses, CACertFingerprint: "trusted-pin"}, + }, nil) + configWithIdentity(t, srv.URL) + withCloudRouteProbe(t, func(*Context, *clientconfig.Config, string, *clientconfig.IdentityConfig, *ClusterResponse) bool { + return true + }) + ctx := presenceContext(t) + added, err := addCluster(ctx, addClusterOptions{clusterName: "prod"}) + require.NoError(t, err) + require.True(t, added.ViaCloud, "cloud was used while adding") + require.False(t, added.ForceCloud, "a successful fallback is not a forced route") + require.Contains(t, ctx.Stdout.(*bytes.Buffer).String(), "direct when reachable, otherwise via cloud") + require.NotContains(t, ctx.Stdout.(*bytes.Buffer).String(), "needs no address here") + cfg, err := clientconfig.LoadConfig() + require.NoError(t, err) + cluster, err := cfg.GetCluster("prod") + require.NoError(t, err) + require.False(t, cluster.ViaCloud) + require.Equal(t, addresses, cluster.AllAddresses) + require.Equal(t, "trusted-pin", cluster.CACertFingerprint) + listCtx := presenceContext(t) + require.NoError(t, ClusterList(listCtx, struct { + FormatOptions + ConfigCentric + }{})) + require.Contains(t, listCtx.Stdout.(*bytes.Buffer).String(), "direct when reachable, otherwise via cloud") +} + // A cluster nothing can dial and cloud cannot reach is a dead end. The picker // greys it out; named, it has to be refused outright — falling through would // probe localhost and could pin whatever is listening there under a remote diff --git a/cli/commands/cluster_add_json.go b/cli/commands/cluster_add_json.go index 69dd25914..10548e329 100644 --- a/cli/commands/cluster_add_json.go +++ b/cli/commands/cluster_add_json.go @@ -89,7 +89,7 @@ type commandError struct { Message string `json:"message"` } -// addedCluster describes what was written to the config. +// addedCluster describes the added configuration and the route checked at add time. type addedCluster struct { // Name is the local name, which is what every other command takes. Name string `json:"name"` @@ -104,6 +104,9 @@ type addedCluster struct { // no address of its own. Address string `json:"address,omitempty"` ViaCloud bool `json:"via_cloud"` + // ForceCloud is the persisted override, unlike ViaCloud which reports the + // route used while adding and may change on the next connection. + ForceCloud bool `json:"force_cloud"` Identity string `json:"identity,omitempty"` @@ -159,7 +162,10 @@ const clusterAddJSONDoc = "With `--format json`, the command prints one result d "```\n\n" + "`name` is the local name, which is what every other command takes. `cloud_name` " + "appears alongside it only when `--as` stored the cluster under a different name than " + - "it has in Miren Cloud, and `address` is absent for a cluster reached through cloud.\n\n" + + "it has in Miren Cloud, and `address` is absent for a cluster reached through cloud. " + + "`via_cloud` reports the route selected while adding; `force_cloud` reports whether " + + "future connections are forced through cloud. With `force_cloud: false`, cloud-authenticated " + + "clusters try direct when reachable, otherwise via cloud.\n\n" + "Messages are written for people and will be reworded. The code is the stable part:\n\n" + "| Code | Meaning |\n" + "|------|---------|\n" + diff --git a/cli/commands/cluster_add_json_test.go b/cli/commands/cluster_add_json_test.go index 747d32c52..fc843f91f 100644 --- a/cli/commands/cluster_add_json_test.go +++ b/cli/commands/cluster_add_json_test.go @@ -81,6 +81,7 @@ func TestClusterAddJSONSuccess(t *testing.T) { r.Equal("cluster-prod", added.XID) r.Equal("Acme", added.Organization) r.True(added.ViaCloud) + r.True(added.ForceCloud) r.Empty(added.Address, "a routed cluster is dialed at no address of its own") r.Equal("cloud", added.Identity) r.True(added.Insecure, "the test cloud is http, and the entry says so") @@ -88,6 +89,26 @@ func TestClusterAddJSONSuccess(t *testing.T) { r.Contains(added.ConfigFile, "acme-prod.yaml") } +func TestClusterAddJSONAutomaticFallback(t *testing.T) { + srv := fakeCloud(t, []ClusterResponse{{Name: "prod", XID: "cluster-prod"}}, map[string]bool{"cluster-prod": true}) + configWithIdentity(t, srv.URL) + withCloudRouteProbe(t, func(*Context, *clientconfig.Config, string, *clientconfig.IdentityConfig, *ClusterResponse) bool { + return true + }) + result, printed, exitCode := runClusterAdd(t, jsonAddOpts("prod")) + require.Zero(t, exitCode) + require.True(t, result.OK) + require.True(t, result.Cluster.ViaCloud) + require.False(t, result.Cluster.ForceCloud) + require.Contains(t, printed, `"force_cloud": false`) + require.Empty(t, result.Cluster.Address) + cfg, err := clientconfig.LoadConfig() + require.NoError(t, err) + cluster, err := cfg.GetCluster("prod") + require.NoError(t, err) + require.False(t, cluster.ViaCloud) +} + // A failure has to be legible twice over: as a document a caller can branch on, // and as a non-zero exit status for one that only checks that. func TestClusterAddJSONReportsFailures(t *testing.T) { diff --git a/cli/commands/cluster_add_viacloud_test.go b/cli/commands/cluster_add_viacloud_test.go index 772a66900..f839c042a 100644 --- a/cli/commands/cluster_add_viacloud_test.go +++ b/cli/commands/cluster_add_viacloud_test.go @@ -184,51 +184,3 @@ func TestOnlineButUnansweringClusterIsNotRouted(t *testing.T) { &ClusterResponse{Name: "old-runtime", XID: "cluster-old"}), "presence in cloud is not evidence the cluster will answer over the relay") } - -// What the fallback has to write, and the RFD's claim that a routed entry needs -// no address and no CA. -// -// A cluster that could not be dialed still advertised addresses and may have -// presented a certificate, and keeping either would write an entry that looks -// dialable and is not. The next command would then try the dead address instead -// of the route that works. -func TestRoutedEntryKeepsNoDialableFields(t *testing.T) { - r := require.New(t) - - // The shape addCluster builds once the fallback has fired: viaCloud set, - // and the address, addresses and certificate deliberately left empty. - cfg := &clientconfig.ClusterConfig{ - Hostname: "", - AllAddresses: nil, - Identity: "cloud", - XID: "cluster-abc", - ViaCloud: true, - } - - r.True(cfg.ViaCloud) - r.Empty(cfg.Hostname, "a routed entry is never dialed, so an address would be a trap") - r.Empty(cfg.AllAddresses) - r.Empty(cfg.CACert, "the certificate on the wire belongs to cloud, not the cluster") - r.NotEmpty(cfg.XID, "routing needs the cluster's id in cloud") - r.NotEmpty(cfg.Identity, "and an identity to authenticate with") - - // And it resolves to a relay endpoint rather than a direct dial. - endpoint, err := cfg.CloudEndpoint(viaCloudTestConfig(t, cfg)) - r.NoError(err) - r.NotEmpty(endpoint) -} - -// viaCloudTestConfig wraps a cluster entry in a config holding the identity it -// names, which is what CloudEndpoint resolves the cloud through. -func viaCloudTestConfig(t *testing.T, cluster *clientconfig.ClusterConfig) *clientconfig.Config { - t.Helper() - - cfg := clientconfig.NewConfig() - cfg.SetIdentity("cloud", &clientconfig.IdentityConfig{ - Type: clientconfig.IdentityToken, - Issuer: "https://api.miren.cloud", - Token: freshLoginJWT(t), - }) - cfg.SetCluster("prod", cluster) - return cfg -} diff --git a/cli/commands/rpc_errors.go b/cli/commands/rpc_errors.go index d288c4b25..d02d8c305 100644 --- a/cli/commands/rpc_errors.go +++ b/cli/commands/rpc_errors.go @@ -21,6 +21,30 @@ import ( // user actually typed — "miren route list" is a far better thing to put in an // error than the capability "entities" it happens to resolve. func (c *Context) wrapRPCError(err error) error { + if routes, ok := errors.AsType[*clientconfig.ClusterRouteError](err); ok { + var d *ui.Diagnostic + if direct, ok := errors.AsType[*rpc.ResolveError](routes.Direct); ok { + d = c.diagnoseResolve(direct) + } + if d == nil { + d = &ui.Diagnostic{ + Detail: "Neither the direct addresses nor the cloud relay could establish a connection.", + Actions: []ui.Action{ + {Command: "miren doctor", Note: "check what's reachable"}, + {Command: "miren cluster list", Note: "confirm the addresses"}, + }, + } + } + d.Summary = fmt.Sprintf("couldn't reach %s directly or through Miren Cloud", clusterLabel(c.ClusterName)) + d.Facts = append(d.Facts, + ui.Fact{Label: "Direct", Value: routes.Direct.Error()}, + ui.Fact{Label: "Miren Cloud", Value: routes.Cloud.Error()}, + ) + d.Cause = routes + d.ShowCause = c.Verbose() + return d + } + re, ok := errors.AsType[*rpc.ResolveError](err) if !ok { return err diff --git a/cli/commands/rpc_errors_test.go b/cli/commands/rpc_errors_test.go index ea25a55f3..a0aa2265b 100644 --- a/cli/commands/rpc_errors_test.go +++ b/cli/commands/rpc_errors_test.go @@ -1,15 +1,23 @@ package commands import ( + "bytes" "context" "errors" "fmt" "io" "log/slog" + "net" + "net/http" + "net/http/httptest" + "os" "strings" + "sync/atomic" "testing" "time" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" "miren.dev/runtime/clientconfig" "miren.dev/runtime/pkg/rpc" "miren.dev/runtime/pkg/ui" @@ -19,6 +27,100 @@ func diagnosticContext() *Context { return &Context{ClusterName: "local", CommandName: "route list"} } +func TestRPCClientRoutingCredentialErrors(t *testing.T) { + for _, status := range []int{http.StatusTooManyRequests, http.StatusUnauthorized} { + t.Run(http.StatusText(status), func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + dir := t.TempDir() + t.Setenv(clientconfig.EnvConfigPath, dir) + var signIns atomic.Int32 + cloud := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/auth/refresh", r.URL.Path) + signIns.Add(1) + w.Header().Set("Retry-After", "20") + http.Error(w, http.StatusText(status), status) + })) + defer cloud.Close() + writeTokenIdentity(t, dir, cloud.URL) + cfg, err := clientconfig.LoadConfig() + require.NoError(t, err) + identity, err := cfg.GetIdentity("cloud") + require.NoError(t, err) + identity.Token = "" // Force a refresh, both in memory and under the refresh lock. + data, err := yaml.Marshal(clientconfig.ConfigData{Identities: map[string]*clientconfig.IdentityConfig{"cloud": identity}}) + require.NoError(t, err) + require.NoError(t, os.WriteFile(cfg.GetIdentitySource("cloud"), data, 0600)) + silent, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + defer silent.Close() + cluster := &clientconfig.ClusterConfig{XID: "cluster-auth", Identity: "cloud", CloudURL: cloud.URL, + Hostname: silent.LocalAddr().String(), AllAddresses: []string{silent.LocalAddr().String()}} + ctx := &Context{Context: t.Context(), ClientConfig: cfg, ClusterConfig: cluster, + ClusterName: "wilderness", Stderr: io.Discard, Log: slog.New(slog.NewTextHandler(io.Discard, nil))} + defer ctx.Close() + _, err = ctx.RPCClient("entities") + require.EqualValues(t, 1, signIns.Load(), "credential failures must not retry sign-in on another route") + var diagnostic *ui.Diagnostic + require.ErrorAs(t, err, &diagnostic) + var rendered bytes.Buffer + diagnostic.WriteForTerminal(&rendered) + require.NotContains(t, rendered.String(), "miren doctor") + if status == http.StatusTooManyRequests { + require.Contains(t, rendered.String(), "too many sign-ins") + require.Contains(t, rendered.String(), "waiting will") + require.Contains(t, rendered.String(), "Retry in") + require.NotContains(t, rendered.String(), "miren login") + var limited *clientconfig.CloudRateLimitError + require.ErrorAs(t, err, &limited) + } else { + require.Contains(t, rendered.String(), "miren login") + require.ErrorIs(t, err, clientconfig.ErrLoginRequired) + } + }) + } +} + +func TestRPCClientDiagnosesBothRoutesFailing(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + silent, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer silent.Close() + cloud := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.NotFound(w, nil) + })) + defer cloud.Close() + cfg := clientconfig.NewConfig() + cfg.SetIdentity("cloud", &clientconfig.IdentityConfig{ + Type: clientconfig.IdentityToken, Issuer: cloud.URL, Token: freshLoginJWT(t), + }) + cluster := &clientconfig.ClusterConfig{XID: "cluster-down", Identity: "cloud", CloudURL: cloud.URL, + AllAddresses: []string{silent.LocalAddr().String()}, Insecure: true} + ctx := &Context{Context: t.Context(), ClientConfig: cfg, ClusterConfig: cluster, + ClusterName: "wilderness", CommandName: "app list", Stderr: io.Discard, + Log: slog.New(slog.NewTextHandler(io.Discard, nil))} + defer ctx.Close() + for range 2 { + _, err := ctx.RPCClient("entities") + d, ok := errors.AsType[*ui.Diagnostic](err) + if !ok { + t.Fatalf("expected a diagnostic for both failed routes, got %T: %v", err, err) + } + var rendered bytes.Buffer + d.WriteForTerminal(&rendered) + for _, want := range []string{`couldn't reach cluster "wilderness" directly or through Miren Cloud`, + silent.LocalAddr().String(), "Nothing answered after", "404", "miren doctor"} { + if !strings.Contains(rendered.String(), want) { + t.Errorf("diagnostic lacks %q:\n%s", want, rendered.String()) + } + } + if _, ok := errors.AsType[*rpc.ResolveError](err); !ok { + t.Error("the direct timeout must remain available to callers") + } + } +} + // Each kind has to produce its own diagnosis. The bug this replaces collapsed // every transport failure into one message, so the regression to guard is two // different failures rendering the same text. diff --git a/clientconfig/clientconfig.go b/clientconfig/clientconfig.go index b724ce46d..7598da7bb 100644 --- a/clientconfig/clientconfig.go +++ b/clientconfig/clientconfig.go @@ -59,15 +59,16 @@ type KeyConfig struct { // ClusterConfig holds the configuration for a single cluster type ClusterConfig struct { - Hostname string `yaml:"hostname"` - AllAddresses []string `yaml:"all_addresses,omitempty"` // All available addresses for this cluster - Identity string `yaml:"identity,omitempty"` // Reference to an identity in the Identities section - XID string `yaml:"xid,omitempty"` // Cloud cluster XID for permission checks - CACert string `yaml:"ca_cert,omitempty"` // PEM encoded CA certificate - ClientCert string `yaml:"client_cert,omitempty"` // PEM encoded client certificate (deprecated, use identity) - ClientKey string `yaml:"client_key,omitempty"` // PEM encoded client key (deprecated, use identity) - Insecure bool `yaml:"insecure,omitempty"` // Skip TLS verification - CloudAuth bool `yaml:"cloud_auth,omitempty"` // Use cloud authentication (deprecated, use identity) + Hostname string `yaml:"hostname"` + AllAddresses []string `yaml:"all_addresses,omitempty"` // All available addresses for this cluster + Identity string `yaml:"identity,omitempty"` // Reference to an identity in the Identities section + XID string `yaml:"xid,omitempty"` // Cloud cluster XID for permission checks + CACert string `yaml:"ca_cert,omitempty"` // PEM encoded CA certificate + CACertFingerprint string `yaml:"ca_cert_fingerprint,omitempty"` // Cloud-advertised CA pin, for clusters added remotely + ClientCert string `yaml:"client_cert,omitempty"` // PEM encoded client certificate (deprecated, use identity) + ClientKey string `yaml:"client_key,omitempty"` // PEM encoded client key (deprecated, use identity) + Insecure bool `yaml:"insecure,omitempty"` // Skip TLS verification + CloudAuth bool `yaml:"cloud_auth,omitempty"` // Use cloud authentication (deprecated, use identity) // IdentityTokenPath is a workload identity token file to authenticate with, // re-read on each request because the sandbox controller refreshes it in @@ -78,13 +79,8 @@ type ClusterConfig struct { // against, for when the dial address cannot be a certificate SAN. TLSServerName string `yaml:"tls_server_name,omitempty"` - // ViaCloud routes RPC through Miren Cloud rather than dialing the cluster, - // for a cluster this machine has no route to. It needs XID to name the - // cluster and an identity to authenticate as; Hostname is unused. - // - // Opting in is explicit rather than a fallback from a failed direct dial, - // so that every command's answer comes from a path you chose. A silent - // fallback would also make every command wait out a dial timeout first. + // ViaCloud forces RPC through Miren Cloud. Otherwise cloud-authenticated + // clusters try direct addresses first and fall back to the relay. ViaCloud bool `yaml:"via_cloud,omitempty"` // CloudURL overrides where the relay lives. Normally empty: the identity's diff --git a/clientconfig/local.go b/clientconfig/local.go index bc5ac2bc3..ef76f0a08 100644 --- a/clientconfig/local.go +++ b/clientconfig/local.go @@ -105,13 +105,22 @@ func (c *ClusterConfig) RPCOptions(ctx context.Context, config *Config) ([]rpc.S // RPCOptionsWithName returns RPC options with cluster name for caching func (c *ClusterConfig) RPCOptionsWithName(ctx context.Context, config *Config, clusterName string) ([]rpc.StateOption, error) { - // A cloud-routed cluster settles the matter before anything below: the - // address probing, the cluster CA, and the direct-dial paths all reason - // about a cluster we can reach, which is exactly what this cluster is not. if c.ViaCloud { return c.viaCloudOptions(ctx, config) } + if c.XID != "" && c.Identity != "" && c.IdentityTokenPath == "" { + identity, err := config.GetIdentity(c.Identity) + if err != nil { + return nil, err + } + if (identity.Type == IdentityKeypair || identity.Type == IdentityToken) && (identity.Issuer != "" || c.CloudURL != "") { + return c.routedOptions(ctx, config) + } + } + return c.directRPCOptions(ctx, config, clusterName) +} +func (c *ClusterConfig) directRPCOptions(ctx context.Context, config *Config, clusterName string) ([]rpc.StateOption, error) { // If AllAddresses is set, find a working address hostname := c.Hostname if len(c.AllAddresses) > 0 && clusterName != "" { @@ -269,6 +278,10 @@ func (c *ClusterConfig) viaCloudOptions(ctx context.Context, config *Config) ([] return nil, err } + return c.cloudDialOptions(endpoint, token), nil +} + +func (c *ClusterConfig) cloudDialOptions(endpoint, token string) []rpc.StateOption { // No cluster CA here: the certificate presented is cloud's, verified against // the system roots. The cluster's own identity is not what is on the wire. opts := []rpc.StateOption{ @@ -284,11 +297,12 @@ func (c *ClusterConfig) viaCloudOptions(ctx context.Context, config *Config) ([] if c.Insecure { opts = append(opts, rpc.WithSkipVerify) } - if c.TLSServerName != "" { + // Automatic fallback must verify cloud's hostname, not the cluster's. + if c.ViaCloud && c.TLSServerName != "" { opts = append(opts, rpc.WithTLSServerName(c.TLSServerName)) } - return opts, nil + return opts } // CloudEndpoint returns the cloud a via-cloud cluster is reached through. diff --git a/clientconfig/routing.go b/clientconfig/routing.go new file mode 100644 index 000000000..90a75d205 --- /dev/null +++ b/clientconfig/routing.go @@ -0,0 +1,331 @@ +package clientconfig + +import ( + "context" + "crypto/sha1" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "github.com/containerd/log" + "github.com/quic-go/quic-go" + "miren.dev/runtime/pkg/rpc" +) + +const cloudRouteCacheTTL = 30 * time.Second + +// ClusterRouteError preserves both connection failures when neither route works. +type ClusterRouteError struct { + Direct error + Cloud error +} + +func (e *ClusterRouteError) Error() string { + return fmt.Sprintf("couldn't connect directly (%v) or through Miren Cloud (%v)", e.Direct, e.Cloud) +} + +func (e *ClusterRouteError) Unwrap() []error { return []error{e.Direct, e.Cloud} } + +type clusterRoute struct { + Address string `json:"address,omitempty"` + CACert string `json:"ca_cert,omitempty"` + TLSServerName string `json:"tls_server_name,omitempty"` + Cloud bool `json:"cloud,omitempty"` +} + +func (r clusterRoute) save(path string) { + if path == "" { + return + } + f, err := os.CreateTemp(filepath.Dir(path), ".cluster-route-*") + if err != nil { + return + } + defer os.Remove(f.Name()) + err = json.NewEncoder(f).Encode(r) + closeErr := f.Close() + if err == nil && closeErr == nil { + _ = os.Rename(f.Name(), path) + } +} + +// Successful routes avoid discovery on the next connection. Failed connections +// invalidate the route before rediscovery; no application call is retried. +func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rpc.StateOption, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + cloudURL, err := c.CloudEndpoint(config) + if err != nil { + return nil, err + } + identity, err := config.GetIdentity(c.Identity) + if err != nil { + return nil, err + } + // Authentication is independent of the route. Resolve it before discovery + // so an expired or rate-limited login cannot invalidate a working route. + token, err := config.TokenForIdentity(ctx, c.Identity, identity, cloudURL) + if err != nil { + return nil, fmt.Errorf("failed to authenticate with cloud: %w", err) + } + // Scope the hint to the cloud and connection configuration, not a local + // alias that may be reused for another cluster. + data, _ := json.Marshal(struct { + Cluster *ClusterConfig + Cloud string + }{c, cloudURL}) + key := sha256.Sum256(data) + cacheDir, _ := getCacheDir() + var cachePath string + if cacheDir != "" { + cachePath = filepath.Join(cacheDir, fmt.Sprintf("cluster_route_%x.json", key)) + } + + return []rpc.StateOption{ + rpc.WithBindAddr("[::]:0"), + rpc.WithClientRouting(func(ctx context.Context, connect func(...rpc.StateOption) (*rpc.NetworkClient, error)) (*rpc.NetworkClient, error) { + attempt := func(route clusterRoute) (*rpc.NetworkClient, error) { + var opts []rpc.StateOption + if route.Cloud { + endpoint, err := cloudRPCEndpoint(cloudURL, c.XID, c.Insecure) + if err != nil { + return nil, err + } + opts = c.cloudDialOptions(endpoint, token) + } else { + direct := *c + direct.Hostname, direct.AllAddresses = route.Address, nil + direct.CACert, direct.TLSServerName = route.CACert, route.TLSServerName + opts = append(direct.dialOptions(route.Address), rpc.WithBearerToken(token)) + } + return connect(opts...) + } + var cached clusterRoute + data, _ := os.ReadFile(cachePath) + info, statErr := os.Stat(cachePath) + valid := json.Unmarshal(data, &cached) == nil && + ((cached.Cloud && statErr == nil && time.Since(info.ModTime()) < cloudRouteCacheTTL) || (!cached.Cloud && cached.Address != "")) + // An address with saved trust can be dialed without a preliminary probe. + if !valid && c.Hostname != "" && (c.CACert != "" || c.CACertFingerprint == "") { + cached = clusterRoute{Address: c.Hostname, CACert: c.CACert, TLSServerName: c.TLSServerName} + valid = true + } + var directErr error + if valid { + client, err := attempt(cached) + if err == nil { + // Do not slide a cloud route's expiry on every successful use. + if !cached.Cloud { + cached.save(cachePath) + } + return client, nil + } + if !routeConnectionFailed(err) { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + _ = os.Remove(cachePath) + if !cached.Cloud { + directErr = fmt.Errorf("%s: %w", cached.Address, err) + } + log.G(ctx).WithError(err).Debug("Remembered cluster route failed; rediscovering") + } + if err := ctx.Err(); err != nil { + return nil, err + } + direct, probeErr := c.discoverDirectRoute(ctx) + directErr = errors.Join(directErr, probeErr) + if direct != nil { + route := clusterRoute{Address: direct.Hostname, CACert: direct.CACert, TLSServerName: direct.TLSServerName} + client, err := attempt(route) + if err == nil { + route.save(cachePath) + return client, nil + } + if !routeConnectionFailed(err) { + return nil, err + } + directErr = errors.Join(directErr, fmt.Errorf("%s: %w", route.Address, err)) + } + if err := ctx.Err(); err != nil { + return nil, err + } + route := clusterRoute{Cloud: true} + client, err := attempt(route) + if err == nil { + route.save(cachePath) + return client, nil + } + return nil, &ClusterRouteError{Direct: directErr, Cloud: err} + }), + }, nil +} + +// A server rejecting a capability or credential is not a broken network route. +func routeConnectionFailed(err error) bool { + var re *rpc.ResolveError + if errors.As(err, &re) { + switch re.Kind { + case rpc.ResolveLookupError, rpc.ResolveStatusError, rpc.ResolveDecodeError: + return false + case rpc.ResolveHTTPError, rpc.ResolveUnreachableError, rpc.ResolveWentSilentError, rpc.ResolveNoAnswerError: + return true + } + } + return true +} + +func (c *ClusterConfig) discoverDirectRoute(ctx context.Context) (*ClusterConfig, error) { + addresses := slices.Clone(c.AllAddresses) + if c.Hostname != "" && !slices.Contains(addresses, c.Hostname) { + addresses = append(addresses, c.Hostname) + } + if len(addresses) == 0 { + return nil, fmt.Errorf("no direct addresses advertised") + } + type result struct { + cluster *ClusterConfig + err error + address string + elapsed time.Duration + } + results := make(chan result, len(addresses)) + probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + started := time.Now() + for _, address := range addresses { + go func() { + cluster, err := c.probeDirectRoute(probeCtx, address) + results <- result{cluster, err, address, time.Since(started)} + }() + } + var failures []error + for range addresses { + select { + case res := <-results: + if res.err == nil { + cancel() + return res.cluster, nil + } + failure := fmt.Errorf("%s: %w", res.address, res.err) + var timeout net.Error + var dns *net.DNSError + if errors.As(res.err, &timeout) && timeout.Timeout() && !errors.As(res.err, &dns) { + failure = rpc.NewResolveUnreachableError("", res.address, res.elapsed, res.err) + } + failures = append(failures, failure) + log.G(ctx).WithField("cluster", c.XID).WithField("direct_error", failure).Debug("Direct cluster route failed") + case <-probeCtx.Done(): + failures = append(failures, fmt.Errorf("probing %v: %w", addresses, probeCtx.Err())) + return nil, errors.Join(failures...) + } + } + return nil, errors.Join(failures...) +} + +// Probe without sending credentials. If add ran off-network, cloud's CA +// fingerprint lets us recover the CA securely when a direct route appears. +func (c *ClusterConfig) probeDirectRoute(ctx context.Context, address string) (*ClusterConfig, error) { + if !strings.Contains(address, ":") { + address += ":8443" + } + host, portName, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + port, err := net.LookupPort("udp", portName) + if err != nil { + return nil, err + } + addresses, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil { + return nil, err + } + if len(addresses) == 0 { + return nil, fmt.Errorf("no addresses for %s", host) + } + ip := addresses[0] + // Prefer IPv4 on dual-stack hosts, matching the RPC dialer's resolver. + for _, candidate := range addresses { + if candidate.IP.To4() != nil { + ip = candidate + break + } + } + direct := *c + direct.Hostname = address + direct.AllAddresses = nil + tlsConfig := &tls.Config{ServerName: host, NextProtos: []string{"h3"}} + if c.TLSServerName != "" { + tlsConfig.ServerName = c.TLSServerName + } + if c.CACert != "" { + tlsConfig.RootCAs = x509.NewCertPool() + if !tlsConfig.RootCAs.AppendCertsFromPEM([]byte(c.CACert)) { + return nil, fmt.Errorf("invalid cluster CA") + } + } else if c.CACertFingerprint != "" { + // Require pinned-root recovery even if the peer has a system-trusted CA. + tlsConfig.RootCAs = x509.NewCertPool() + } + network := "udp6" + if ip.IP.To4() != nil { + network = "udp4" + } + socket, err := net.ListenUDP(network, nil) + if err != nil { + return nil, err + } + defer socket.Close() + transport := &quic.Transport{Conn: socket} + defer transport.Close() + remoteAddr := &net.UDPAddr{IP: ip.IP, Port: port, Zone: ip.Zone} + quicConfig := &quic.Config{ + InitialPacketSize: rpc.InitialPacketSize, HandshakeIdleTimeout: 2 * time.Second, + } + conn, err := transport.Dial(ctx, remoteAddr, tlsConfig, quicConfig) + if err != nil && c.CACert == "" && c.CACertFingerprint != "" { + var verification *tls.CertificateVerificationError + if !errors.As(err, &verification) { + return nil, err + } + // A failed handshake exposes the chain without disabling verification. + // Trust its root only after checking the cloud-provided fingerprint. + certs := verification.UnverifiedCertificates + root := certs[len(certs)-1] + sum := sha1.Sum(root.Raw) + if !strings.EqualFold(hex.EncodeToString(sum[:]), c.CACertFingerprint) { + return nil, fmt.Errorf("cluster CA fingerprint mismatch") + } + tlsConfig = tlsConfig.Clone() + tlsConfig.RootCAs = x509.NewCertPool() + tlsConfig.RootCAs.AddCert(root) + if c.TLSServerName == "" && certs[0].VerifyHostname(tlsConfig.ServerName) != nil && slices.Contains(certs[0].DNSNames, APIServerName) { + tlsConfig.ServerName = APIServerName + } + conn, err = transport.Dial(ctx, remoteAddr, tlsConfig, quicConfig) + if err == nil { + direct.CACert = string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: root.Raw})) + direct.TLSServerName = tlsConfig.ServerName + } + } + if err != nil { + return nil, err + } + _ = conn.CloseWithError(0, "route probe complete") + return &direct, nil +} diff --git a/clientconfig/routing_test.go b/clientconfig/routing_test.go new file mode 100644 index 000000000..6a78b94a5 --- /dev/null +++ b/clientconfig/routing_test.go @@ -0,0 +1,378 @@ +package clientconfig + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/qlogwriter" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" + "miren.dev/runtime/pkg/caauth" + "miren.dev/runtime/pkg/rpc" + "miren.dev/runtime/pkg/slogrus" +) + +type routingCredentialRecorder struct { + bearers atomic.Int64 +} + +func (r *routingCredentialRecorder) Authenticate(_ context.Context, creds *rpc.Credentials) (*rpc.Identity, error) { + if creds.BearerToken() != "" { + r.bearers.Add(1) + } + return &rpc.Identity{Subject: "routing-test"}, nil +} + +func TestAutomaticClusterRouting(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + ca := testAuthority(t) + cert, err := ca.IssueCertificate(caauth.Options{ + CommonName: "api.miren", DNSNames: []string{"api.miren"}, ValidFor: time.Hour, + }) + require.NoError(t, err) + recorder := &routingCredentialRecorder{} + server, err := rpc.NewState(t.Context(), rpc.WithBindAddr("127.0.0.1:0"), rpc.WithAuthenticator(recorder), + rpc.WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + require.NoError(t, err) + t.Cleanup(func() { _ = server.Close() }) + cloud := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/api/v1/clusters/cluster-abc/rpc", r.URL.Path) + w.WriteHeader(http.StatusServiceUnavailable) + })) + t.Cleanup(cloud.Close) + cluster := &ClusterConfig{ + XID: "cluster-abc", Identity: "cloud", CloudURL: cloud.URL, Insecure: true, + AllAddresses: []string{"invalid::::", server.ListenAddr()}, CACertFingerprint: ca.Fingerprint(), + } + config := viaCloudConfig(t, cluster) + var diagnostics bytes.Buffer + ctx := slogrus.WithLogger(t.Context(), slog.New(slog.NewJSONHandler(&diagnostics, &slog.HandlerOptions{Level: slog.LevelDebug}))) + connect := func() error { + state, err := cluster.State(ctx, config) + require.NoError(t, err) + defer state.Close() + _, err = state.Client("missing") + return err + } + // Resolving a missing capability on the real QUIC server yields a lookup error. + // The fake cloud instead yields a 503, distinguishing the chosen route. + assertDirect := func() { + err := connect() + var resolve *rpc.ResolveError + require.True(t, errors.As(err, &resolve), "%v", err) + require.Equal(t, rpc.ResolveLookupError, resolve.Kind, "%v", err) + require.ErrorContains(t, err, "unknown object: missing") + } + assertDirect() + require.Positive(t, recorder.bearers.Load(), "the direct peer must observe credentials on an accepted route") + require.Empty(t, cluster.CACert, "discovered trust is per connection, not a config mutation") + + cluster.CACertFingerprint = "wrong-pin" + diagnostics.Reset() + require.ErrorContains(t, connect(), "503", "a mismatched CA must fall back to cloud") + require.Contains(t, diagnostics.String(), server.ListenAddr()) + require.Contains(t, diagnostics.String(), "cluster CA fingerprint mismatch") + diagnostics.Reset() + require.ErrorContains(t, connect(), "503") + require.Contains(t, diagnostics.String(), "Direct cluster route failed") + require.Contains(t, diagnostics.String(), server.ListenAddr()) + require.Contains(t, diagnostics.String(), "cluster CA fingerprint mismatch") + cluster.CACertFingerprint = ca.Fingerprint() + assertDirect() + + cluster.TLSServerName = "other-api.miren" + require.ErrorContains(t, connect(), "503", "pinned-root recovery must respect an explicit server name") + cluster.TLSServerName = "" + assertDirect() + + // Existing direct entries use their saved CA and explicit TLS server name. + cluster.CACert = string(ca.GetCACertificate()) + cluster.CACertFingerprint = "" + cluster.TLSServerName = APIServerName + assertDirect() + cluster.CACert = "" + cluster.CACertFingerprint = ca.Fingerprint() + cluster.TLSServerName = "" + + runnerCert, err := ca.IssueCertificate(caauth.Options{ + CommonName: "runner", DNSNames: []string{"runner.miren"}, ValidFor: time.Hour, + }) + require.NoError(t, err) + runnerRecorder := &routingCredentialRecorder{} + runner, err := rpc.NewState(t.Context(), rpc.WithBindAddr("127.0.0.1:0"), + rpc.WithCertPEMs(runnerCert.CertPEM, runnerCert.KeyPEM), rpc.WithAuthenticator(runnerRecorder)) + require.NoError(t, err) + t.Cleanup(func() { _ = runner.Close() }) + cluster.AllAddresses = []string{runner.ListenAddr()} + require.ErrorContains(t, connect(), "503", "a trusted CA alone does not make an unrelated SAN an API identity") + require.Zero(t, runnerRecorder.bearers.Load(), "an unrelated same-CA peer must not receive the bearer token") + cluster.CACertFingerprint = "wrong-pin" + require.ErrorContains(t, connect(), "503") + require.Zero(t, runnerRecorder.bearers.Load(), "a mismatched CA must not receive the bearer token") + cluster.CACertFingerprint = ca.Fingerprint() + + expiredCert, err := ca.IssueCertificate(caauth.Options{ + CommonName: APIServerName, DNSNames: []string{APIServerName}, ValidFor: -time.Hour, + }) + require.NoError(t, err) + expiredRecorder := &routingCredentialRecorder{} + expiredServer, err := rpc.NewState(t.Context(), rpc.WithBindAddr("127.0.0.1:0"), + rpc.WithCertPEMs(expiredCert.CertPEM, expiredCert.KeyPEM), rpc.WithAuthenticator(expiredRecorder)) + require.NoError(t, err) + t.Cleanup(func() { _ = expiredServer.Close() }) + cluster.AllAddresses = []string{expiredServer.ListenAddr()} + diagnostics.Reset() + require.ErrorContains(t, connect(), "503", "a matching pin and hostname must not bypass certificate expiry") + require.Contains(t, diagnostics.String(), "expired") + require.Zero(t, expiredRecorder.bearers.Load(), "an expired API certificate must not receive the bearer token") + + // Simulate adding while remote, then returning to the direct network. + cluster.AllAddresses = []string{"invalid::::"} + require.ErrorContains(t, connect(), "503") + cluster.AllAddresses = []string{server.ListenAddr()} + assertDirect() + // Returning to the previous config reuses only its short-lived cloud hint. + cluster.AllAddresses = []string{"invalid::::"} + require.ErrorContains(t, connect(), "503") + cluster.AllAddresses = []string{server.ListenAddr()} + cluster.ViaCloud = true + require.ErrorContains(t, connect(), "503", "explicit cloud routing overrides a working direct route") + cluster.ViaCloud = false + + t.Run("legacy cloud entry remains forced", func(t *testing.T) { + original := cluster + defer func() { cluster = original }() + // Old automatic fallbacks had no addresses, CA, fingerprint, or flag + // distinguishing them from an explicit cloud-routing request. + cluster = &ClusterConfig{} + require.NoError(t, yaml.Unmarshal([]byte(fmt.Sprintf(`hostname: "" +via_cloud: true +xid: cluster-abc +identity: cloud +cloud_url: %q +insecure: true +`, cloud.URL)), cluster)) + require.True(t, cluster.ViaCloud) + require.Empty(t, cluster.CACertFingerprint) + require.ErrorContains(t, connect(), "503") + }) + + // A failed relay is never cached: recovery is noticed immediately. + address := server.ListenAddr() + require.NoError(t, server.Close()) + require.ErrorContains(t, connect(), "503") + server, err = rpc.NewState(t.Context(), rpc.WithBindAddr(address), rpc.WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + require.NoError(t, err) + assertDirect() + + canceledCtx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = cluster.State(canceledCtx, config) + require.ErrorIs(t, err, context.Canceled) +} + +func TestAutomaticClusterRoutingCancelsDNS(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + previous := net.DefaultResolver + defer func() { net.DefaultResolver = previous }() + started := make(chan struct{}, 2) + release := make(chan struct{}) + defer close(release) + var active atomic.Int64 + net.DefaultResolver = &net.Resolver{PreferGo: true, Dial: func(ctx context.Context, _, _ string) (net.Conn, error) { + active.Add(1) + defer active.Add(-1) + select { + case started <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-release: + return nil, fmt.Errorf("DNS test shut down") + } + }} + cluster := &ClusterConfig{Hostname: "unresolvable.invalid:8443", XID: "cluster-abc", Identity: "cloud"} + config := viaCloudConfig(t, cluster) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + result := make(chan error, 1) + go func() { + state, err := cluster.State(ctx, config) + if state != nil { + defer state.Close() + } + if err == nil { + _, err = state.Client("missing") + } + result <- err + }() + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("direct routing did not start DNS resolution") + } + cancel() + select { + case err := <-result: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("canceled route selection did not return") + } + require.Eventually(t, func() bool { return active.Load() == 0 }, time.Second, time.Millisecond, + "cancellation must stop the outstanding DNS work, not just abandon its result") +} + +func TestSuccessfulClusterRoutes(t *testing.T) { + for _, savedCA := range []bool{false, true} { + t.Run(fmt.Sprintf("saved_CA_%v", savedCA), func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + var handshakes, calls, cloudCalls atomic.Int64 + tracer := rpc.DefaultQUICConfig.Tracer + rpc.DefaultQUICConfig.Tracer = func(_ context.Context, isClient bool, _ quic.ConnectionID) qlogwriter.Trace { + if !isClient { + handshakes.Add(1) + } + return nil + } + defer func() { rpc.DefaultQUICConfig.Tracer = tracer }() + ca := testAuthority(t) + cert, err := ca.IssueCertificate(caauth.Options{CommonName: APIServerName, DNSNames: []string{APIServerName}, ValidFor: time.Hour}) + require.NoError(t, err) + iface := rpc.NewInterface([]rpc.Method{{Name: "Fail", InterfaceName: "routing", Public: true, + Handler: func(context.Context, rpc.Call) error { + calls.Add(1) + return errors.New("application failed after writing") + }, + }}, nil) + start := func(address string) *rpc.State { + server, err := rpc.NewState(t.Context(), rpc.WithBindAddr(address), rpc.WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + require.NoError(t, err) + t.Cleanup(func() { _ = server.Close() }) + server.Server().ExposeValue("routing", iface) + return server + } + server := start("127.0.0.1:0") + relay, err := rpc.NewState(t.Context(), rpc.WithBindAddr("127.0.0.1:0"), rpc.WithWSBindAddr("127.0.0.1:0"), rpc.WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + require.NoError(t, err) + t.Cleanup(func() { _ = relay.Close() }) + relay.Server().ExposeValue("routing", iface) + target, err := url.Parse("https://" + relay.WSListenAddr()) + require.NoError(t, err) + roots := x509.NewCertPool() + require.True(t, roots.AppendCertsFromPEM(ca.GetCACertificate())) + proxy := httputil.NewSingleHostReverseProxy(target) + proxy.Transport = &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots, ServerName: APIServerName}} + var cloudDown atomic.Bool + cloud := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cloudCalls.Add(1) + if cloudDown.Load() { + http.Error(w, "no link to the cluster", http.StatusServiceUnavailable) + return + } + r.URL.Path = "/_rpc/message" + proxy.ServeHTTP(w, r) + })) + t.Cleanup(cloud.Close) + cluster := &ClusterConfig{XID: "cluster-cache", Identity: "cloud", CloudURL: cloud.URL, + AllAddresses: []string{server.ListenAddr()}, CACertFingerprint: ca.Fingerprint(), Insecure: true} + if savedCA { + cluster.Hostname, cluster.CACert, cluster.TLSServerName = server.ListenAddr(), string(ca.GetCACertificate()), APIServerName + } + connect := func() *rpc.NetworkClient { + // Recreate Config too: the successful route must survive separate CLI invocations. + state, err := cluster.State(t.Context(), viaCloudConfig(t, cluster)) + require.NoError(t, err) + t.Cleanup(func() { _ = state.Close() }) + client, err := state.Client("routing") + require.NoError(t, err) + return client + } + first := connect() + initial := int64(3) + if savedCA { + initial = 1 + } + require.Equal(t, initial, handshakes.Load(), "saved trust needs no probe; fingerprint recovery is paid once") + connect() + connect() + require.Equal(t, initial+2, handshakes.Load(), "each steady-state RPC client needs only its own handshake") + require.ErrorContains(t, first.Call(t.Context(), "Fail", nil, nil), "application failed after writing") + require.EqualValues(t, 1, calls.Load(), "routing must never replay an application call") + require.Zero(t, cloudCalls.Load()) + missing, err := cluster.State(t.Context(), viaCloudConfig(t, cluster)) + require.NoError(t, err) + t.Cleanup(func() { _ = missing.Close() }) + _, err = missing.Client("missing") + var lookup *rpc.ResolveError + require.ErrorAs(t, err, &lookup) + require.Equal(t, rpc.ResolveLookupError, lookup.Kind) + require.Equal(t, initial+3, handshakes.Load(), "a lookup rejection must not rediscover a working route") + require.Zero(t, cloudCalls.Load()) + unauthenticated := viaCloudConfig(t, cluster) + identity, err := unauthenticated.GetIdentity("cloud") + require.NoError(t, err) + identity.Token = "" + state, err := cluster.State(t.Context(), unauthenticated) + if state != nil { + t.Cleanup(func() { _ = state.Close() }) + } + require.ErrorContains(t, err, "failed to authenticate with cloud") + connect() + require.Equal(t, initial+4, handshakes.Load(), "credential failures must preserve the cached direct route and recovered CA") + require.Zero(t, cloudCalls.Load()) + if savedCA { + return + } + + address := server.ListenAddr() + require.NoError(t, server.Close()) + connect() // The cached direct route fails; rediscovery falls back to the working relay. + require.EqualValues(t, 1, cloudCalls.Load()) + cacheDir, err := getCacheDir() + require.NoError(t, err) + paths, err := filepath.Glob(filepath.Join(cacheDir, "cluster_route_*.json")) + require.NoError(t, err) + require.Len(t, paths, 1) + firstCloud, err := os.Stat(paths[0]) + require.NoError(t, err) + server = start(address) + before := handshakes.Load() + connect() + require.EqualValues(t, 2, cloudCalls.Load(), "a working cloud route is reused until expiry") + require.Equal(t, before, handshakes.Load(), "a cached cloud route must not probe") + info, err := os.Stat(paths[0]) + require.NoError(t, err) + require.Equal(t, firstCloud.ModTime(), info.ModTime(), "using a cloud route must not postpone rediscovery") + require.Equal(t, os.FileMode(0600), info.Mode().Perm()) + old := time.Now().Add(-time.Minute) + require.NoError(t, os.Chtimes(paths[0], old, old)) + connect() + require.EqualValues(t, 2, cloudCalls.Load(), "cloud expiry notices the direct network again") + require.NoError(t, server.Close()) + connect() + require.EqualValues(t, 3, cloudCalls.Load()) + start(address) + cloudDown.Store(true) + connect() + require.EqualValues(t, 4, cloudCalls.Load(), "a failed cached cloud route triggers direct rediscovery without waiting for expiry") + }) + } +} diff --git a/clientconfig/token.go b/clientconfig/token.go index b23caa069..083ec6886 100644 --- a/clientconfig/token.go +++ b/clientconfig/token.go @@ -103,8 +103,10 @@ func refreshTokenPair(ctx context.Context, issuer, refreshToken string) (*tokenP // Definitive rejection: token expired, revoked, or already spent by a // concurrent refresh. The refresh endpoint returns plain-text bodies. return nil, ErrLoginRequired + case http.StatusTooManyRequests: + return nil, authResponseError(resp, body) default: - // Transient (5xx, rate-limit, etc.): keep the existing tokens. + // Transient errors such as 5xx keep the existing tokens. return nil, fmt.Errorf("token refresh failed: server returned status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) } diff --git a/docs/docs/command/cluster-add.md b/docs/docs/command/cluster-add.md index 8d4edcd1f..cb5832363 100644 --- a/docs/docs/command/cluster-add.md +++ b/docs/docs/command/cluster-add.md @@ -19,7 +19,7 @@ With `--format json`, the command prints one result document and nothing else on {"ok": false, "error": {"code": "cluster_not_found", "message": "no cluster named ..."}} ``` -`name` is the local name, which is what every other command takes. `cloud_name` appears alongside it only when `--as` stored the cluster under a different name than it has in Miren Cloud, and `address` is absent for a cluster reached through cloud. +`name` is the local name, which is what every other command takes. `cloud_name` appears alongside it only when `--as` stored the cluster under a different name than it has in Miren Cloud, and `address` is absent for a cluster reached through cloud. `via_cloud` reports the route selected while adding; `force_cloud` reports whether future connections are forced through cloud. With `force_cloud: false`, cloud-authenticated clusters try direct when reachable, otherwise via cloud. Messages are written for people and will be reworded. The code is the stable part: diff --git a/docs/docs/miren-cloud/cloud-routed-clusters.md b/docs/docs/miren-cloud/cloud-routed-clusters.md index b4544119d..5f835c037 100644 --- a/docs/docs/miren-cloud/cloud-routed-clusters.md +++ b/docs/docs/miren-cloud/cloud-routed-clusters.md @@ -25,8 +25,7 @@ firewall rule to add. ## Setting one up -Usually you do not have to do anything. `miren cluster add` picks this route on -its own when it is the one that works: +Usually you do not have to do anything. Add your cluster normally: ```bash miren cluster add @@ -35,16 +34,43 @@ miren cluster add If the cluster you pick advertises no address your machine can dial, or advertises addresses that do not answer, the command tries the cloud route itself: it opens a session through cloud and makes a call. If the cluster -answers, the entry is written to route through cloud and it tells you so. A +answers, the entry is saved and it tells you which route worked. A cluster that neither you nor cloud can reach still fails, because that is a real problem rather than a routing choice. +Subsequent connections reuse the last route that successfully established an RPC +client, without probing first. The cache is local to your machine and shared by +commands; moving between networks does not permanently change your config. A +failed connection discards the cached route, discovers a route again, and retries +connection setup. Application calls are never replayed. A server rejecting a +capability or your permissions does not trigger route rediscovery. + +Credentials are resolved before routing starts. An expired or rate-limited login +stops there, without probing, retrying sign-in on another route, or discarding a +working route. The error points you toward logging in or waiting, as appropriate. + +A successful cloud route is cached for 30 seconds without extending its lifetime +on every use, so returning to the direct network is noticed. If the cached cloud +route fails, direct discovery runs immediately rather than waiting for expiry. +Unsuccessful routes are never cached. When both routes fail, the error includes +the direct failures and the cloud failure, along with the usual diagnostic and +`miren doctor` guidance. + +Advertised addresses and the cluster CA fingerprint are retained even when you +add the cluster remotely, so a later direct connection can verify the cluster. +When only the fingerprint is saved and there is no cached direct route, the first +TLS handshake fails verification and supplies the certificate chain. The CLI +checks the CA fingerprint, then retries with that CA trusted and normal TLS +verification enabled. The recovered CA is cached with the successful direct route, +so later connections need only their own RPC handshake. Entries with a saved CA +and address can connect without a preliminary probe even on the first use. + It tries the route rather than asking whether one exists, because those are different questions. Cloud can hold a link to a cluster whose runtime is older than this feature, and that cluster will sit there without answering. Making the call is what tells the two apart. -To skip the direct attempt entirely, ask for it: +To force cloud routing, skipping direct attempts on every connection, ask for it: ```bash miren cluster add --via-cloud @@ -200,8 +226,34 @@ rather than going looking for a fault. **It is slower than a direct connection.** Every frame takes an extra hop, and the relay is not the path to choose when you have a direct one available. -**One cloud, one cluster.** The route is per-cluster. Clusters you can reach -directly should stay that way. +:::note[Only connection setup is retried] + +Automatic routing can rediscover a route and fall back to cloud if direct RPC +connection setup fails. This happens before any application call. Once a call +starts, a failure is returned to the caller; in-flight operations are never +automatically replayed on another route. + +::: + +:::note[Existing forced routes stay forced] + +Older versions wrote `via_cloud: true` for both explicit `--via-cloud` requests +and automatic fallbacks. Both remain forced cloud routes after upgrading: +the saved config cannot distinguish their intent, and automatic fallbacks +discarded the direct addresses and CA. + +Re-add the cluster without `--via-cloud` to enable automatic routing and retrieve +its advertised addresses and CA fingerprint: + +```bash +miren cluster add --cluster NAME --force +``` + +Use the cluster's name in cloud for `NAME`. If its saved local name differs, add +`--as LOCAL_NAME` to replace the existing entry rather than create another one. +If the cloud name exists in multiple organizations, add `--organization ORG`. + +::: ## Configuration diff --git a/pkg/rpc/client.go b/pkg/rpc/client.go index e45d5ea66..68f40827d 100644 --- a/pkg/rpc/client.go +++ b/pkg/rpc/client.go @@ -82,6 +82,7 @@ func (u unreachableClient) Close() error { return nil } type NetworkClient struct { State *State + opts *stateOptions transportKind transportKind @@ -759,6 +760,32 @@ func (c *NetworkClient) derefOID(ctx context.Context, oid OID) error { return nil } +// closeSetupTransport tears down a failed connection attempt without releasing +// a capability (which may not exist yet) or sending another request to the peer. +// The State's shared transport remains available for the next attempt. +func (c *NetworkClient) closeSetupTransport() { + if c.ops != nil { + c.ops.mu.Lock() + if c.ops.ctrlSess != nil { + _ = c.ops.ctrlSess.Close() + c.ops.ctrlSess = nil + } + c.ops.mu.Unlock() + } + if c.httpDrain != nil { + c.httpDrain.mu.Lock() + pool := c.httpDrain.current + c.httpDrain.mu.Unlock() + if pool != nil { + c.httpDrain.retire(pool) + _ = pool.transport.Close() + } + } + if c.remoteAddr != nil && c.State.localMP != nil { + _ = c.State.localMP.RemoveConn(c.remoteAddr) + } +} + func (c *NetworkClient) Close() error { // Close inline client if present if c.inlineClient != nil { diff --git a/pkg/rpc/client_routing_test.go b/pkg/rpc/client_routing_test.go new file mode 100644 index 000000000..6819e09c5 --- /dev/null +++ b/pkg/rpc/client_routing_test.go @@ -0,0 +1,66 @@ +package rpc + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "miren.dev/runtime/pkg/caauth" +) + +func TestClientRoutingIsolatesAttempts(t *testing.T) { + r := require.New(t) + ca, err := caauth.New(caauth.Options{CommonName: "routing-ca", ValidFor: time.Hour}) + r.NoError(err) + cert, err := ca.IssueCertificate(caauth.Options{ + CommonName: "routing-server", DNSNames: []string{"routing.test"}, ValidFor: time.Hour, + }) + r.NoError(err) + server, err := NewState(t.Context(), WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + r.NoError(err) + t.Cleanup(func() { _ = server.Close() }) + emptyServer, err := NewState(t.Context(), WithCertPEMs(cert.CertPEM, cert.KeyPEM)) + r.NoError(err) + t.Cleanup(func() { _ = emptyServer.Close() }) + var calls atomic.Int32 + server.Server().ExposeValue("test", NewInterface([]Method{{ + Name: "fail", InterfaceName: "test", Public: true, + Handler: func(context.Context, Call) error { + calls.Add(1) + return fmt.Errorf("application failure") + }, + }}, nil)) + + routes := 0 + owner, err := NewState(t.Context(), WithSkipVerify, + WithClientRouting(func(_ context.Context, attempt func(...StateOption) (*NetworkClient, error)) (*NetworkClient, error) { + routes++ + _, err := attempt(WithEndpoint(server.ListenAddr()), + WithCertificateVerification(ca.GetCACertificate()), WithTLSServerName("wrong.test")) + r.Error(err) + _, err = attempt(WithEndpoint(emptyServer.ListenAddr()), + WithCertificateVerification(ca.GetCACertificate()), WithTLSServerName("routing.test")) + r.Error(err, "a missing capability fails after the TLS handshake") + _, err = attempt() + r.ErrorContains(err, "no remote address specified", "attempt endpoints must not become defaults") + return attempt(WithEndpoint(server.ListenAddr()), + WithCertificateVerification(ca.GetCACertificate()), WithTLSServerName("routing.test")) + })) + r.NoError(err) + t.Cleanup(func() { _ = owner.Close() }) + + client, err := owner.Client("test") + r.NoError(err) + r.Equal(int32(0), calls.Load(), "routing must not make application calls") + + r.Error(client.Call(t.Context(), "fail", nil, nil)) + r.Equal(int32(1), calls.Load()) + r.Equal(1, routes, "application failure must not reenter routing") + + _, err = owner.Connect(server.ListenAddr(), "test") + r.NoError(err) + r.Equal(1, routes, "explicit Connect must bypass routing") +} diff --git a/pkg/rpc/state.go b/pkg/rpc/state.go index a8237e890..6bb61827d 100644 --- a/pkg/rpc/state.go +++ b/pkg/rpc/state.go @@ -210,7 +210,8 @@ type stateOptions struct { memServerName string - endpoint string + endpoint string + clientRouting func(context.Context, func(...StateOption) (*NetworkClient, error)) (*NetworkClient, error) skipVerify bool caCert []byte @@ -353,6 +354,16 @@ func WithEndpoint(endpoint string) StateOption { } } +// WithClientRouting wraps only State.Client's connection setup, before any +// application call. Each attempt applies options independently on the owning +// State's transport. Explicit State.Connect calls bypass routing; application +// calls are never routed or retried by this handler. +func WithClientRouting(fn func(context.Context, func(...StateOption) (*NetworkClient, error)) (*NetworkClient, error)) StateOption { + return func(o *stateOptions) { + o.clientRouting = fn + } +} + func WithAuthenticator(auth Authenticator) StateOption { return func(o *stateOptions) { o.authenticator = auth @@ -418,31 +429,7 @@ func WithTLSServerName(name string) StateOption { } } -func NewState(ctx context.Context, opts ...StateOption) (*State, error) { - var so stateOptions - - for _, opt := range opts { - opt(&so) - } - - if so.bindAddr == "" { - so.bindAddr = "localhost:0" - } - - if so.log == nil { - so.log = slog.Default() - } - - up, err := net.ResolveUDPAddr("udp", so.bindAddr) - if err != nil { - return nil, err - } - - udpConn, err := net.ListenUDP("udp", up) - if err != nil { - return nil, err - } - +func clientTLSConfig(so *stateOptions) (*tls.Config, error) { tlsCfg := &tls.Config{ InsecureSkipVerify: so.skipVerify, NextProtos: []string{http3.NextProtoH3}, @@ -463,7 +450,10 @@ func NewState(ctx context.Context, opts ...StateOption) (*State, error) { } } - var cert tls.Certificate + var ( + cert tls.Certificate + err error + ) if so.certData != nil && so.keyData != nil { cert, err = tls.X509KeyPair(so.certData, so.keyData) @@ -481,6 +471,40 @@ func NewState(ctx context.Context, opts ...StateOption) (*State, error) { tlsCfg.Certificates = []tls.Certificate{cert} } + return tlsCfg, nil +} + +func NewState(ctx context.Context, opts ...StateOption) (*State, error) { + var so stateOptions + + for _, opt := range opts { + opt(&so) + } + + if so.bindAddr == "" { + so.bindAddr = "localhost:0" + } + + if so.log == nil { + so.log = slog.Default() + } + + up, err := net.ResolveUDPAddr("udp", so.bindAddr) + if err != nil { + return nil, err + } + + udpConn, err := net.ListenUDP("udp", up) + if err != nil { + return nil, err + } + + tlsCfg, err := clientTLSConfig(&so) + if err != nil { + _ = udpConn.Close() + return nil, err + } + pub, priv, err := ed25519.GenerateKey(rand.Reader) if err != nil { return nil, err @@ -1063,6 +1087,22 @@ func (s *State) contextOwnsShutdown(ctx context.Context) bool { } func (s *State) Client(name string) (*NetworkClient, error) { + if s.opts != nil && s.opts.clientRouting != nil { + return s.opts.clientRouting(s.top, func(options ...StateOption) (*NetworkClient, error) { + opts := *s.opts + for _, option := range options { + option(&opts) + } + if opts.endpoint == "" { + return nil, fmt.Errorf("no remote address specified") + } + tlsCfg, err := clientTLSConfig(&opts) + if err != nil { + return nil, err + } + return s.connect(opts.endpoint, name, &opts, tlsCfg) + }) + } if s.defaultEndpoint == "" { return nil, fmt.Errorf("no remote address specified") } @@ -1071,6 +1111,10 @@ func (s *State) Client(name string) (*NetworkClient, error) { } func (s *State) Connect(remote string, name string) (*NetworkClient, error) { + return s.connect(remote, name, s.opts, s.clientTlsCfg) +} + +func (s *State) connect(remote, name string, opts *stateOptions, tlsCfg *tls.Config) (*NetworkClient, error) { var ( client *NetworkClient err error @@ -1086,7 +1130,7 @@ func (s *State) Connect(remote string, name string) (*NetworkClient, error) { client = &NetworkClient{ State: s, transportKind: transportMsg, - tlsCfg: s.clientTlsCfg, + tlsCfg: tlsCfg, remote: remote, } @@ -1110,13 +1154,22 @@ func (s *State) Connect(remote string, name string) (*NetworkClient, error) { client = &NetworkClient{ State: s, transport: s.transport, - tlsCfg: s.clientTlsCfg, + tlsCfg: tlsCfg, remote: remote, } client.setupTransport() } + client.opts = opts + client.tlsCfg = tlsCfg + initialized := false + defer func() { + if !initialized { + client.closeSetupTransport() + } + }() + err = client.resolveCapability(name) if err != nil { s.log.Debug("error resolving capability", "error", err) @@ -1129,6 +1182,7 @@ func (s *State) Connect(remote string, name string) (*NetworkClient, error) { return nil, err } + initialized = true return client, nil } @@ -1159,11 +1213,17 @@ func (c *NetworkClient) newClientUnder(capa *Capability) *NetworkClient { tk = transportMsg } + tlsCfg := c.tlsCfg + if tlsCfg == nil { + tlsCfg = c.State.clientTlsCfg + } + newClient := &NetworkClient{ State: c.State, + opts: c.opts, transportKind: tk, transport: transport, - tlsCfg: c.State.clientTlsCfg.Clone(), + tlsCfg: tlsCfg.Clone(), capa: capa, oid: capa.OID, remote: addr, diff --git a/pkg/rpc/transport_msg.go b/pkg/rpc/transport_msg.go index 37c35c478..bef9699e0 100644 --- a/pkg/rpc/transport_msg.go +++ b/pkg/rpc/transport_msg.go @@ -186,11 +186,15 @@ func nowStamp() string { // rejects it, so the caller's retry picks up a token that has since been // refreshed. Failing hard here would deny that retry the chance. func (c *NetworkClient) bearer() string { - if c.State == nil || c.State.opts == nil { + opts := c.opts + if opts == nil && c.State != nil { + opts = c.State.opts + } + if opts == nil { return "" } - if fn := c.State.opts.bearerTokenFunc; fn != nil { + if fn := opts.bearerTokenFunc; fn != nil { token, err := fn() if err != nil { return "" @@ -198,7 +202,7 @@ func (c *NetworkClient) bearer() string { return token } - return c.State.opts.bearerToken + return opts.bearerToken } func traceMap(ctx context.Context) map[string]string {