From b792f9a328a7eb1979b0b33d3296e18b91e9d2ea Mon Sep 17 00:00:00 2001 From: Evan Phoenix Date: Thu, 29 Oct 2026 01:00:57 +0000 Subject: [PATCH 1/4] Select direct or cloud cluster routing per connection Retain advertised addresses and CA pins when adding remote clusters, probe verified direct routes on connection, and briefly cache failed direct attempts before using cloud. Preserve explicit and legacy forced-cloud entries. --- cli/commands/cluster_add.go | 24 +- cli/commands/cluster_add_byname_test.go | 21 ++ cli/commands/cluster_add_viacloud_test.go | 48 ---- clientconfig/clientconfig.go | 28 +-- clientconfig/local.go | 18 +- clientconfig/routing.go | 180 +++++++++++++++ clientconfig/routing_test.go | 208 ++++++++++++++++++ .../docs/miren-cloud/cloud-routed-clusters.md | 44 +++- 8 files changed, 483 insertions(+), 88 deletions(-) create mode 100644 clientconfig/routing.go create mode 100644 clientconfig/routing_test.go diff --git a/cli/commands/cluster_add.go b/cli/commands/cluster_add.go index 426304442..e82cd364e 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.") @@ -596,7 +592,7 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { Name: clusterName, XID: clusterXID, Address: address, - ViaCloud: opts.viaCloud, + ViaCloud: 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..84e812efd 100644 --- a/cli/commands/cluster_add_byname_test.go +++ b/cli/commands/cluster_add_byname_test.go @@ -71,6 +71,27 @@ 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 + }) + added, err := addCluster(presenceContext(t), addClusterOptions{clusterName: "prod"}) + require.NoError(t, err) + require.False(t, added.ViaCloud, "a successful fallback is not a forced route") + 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) +} + // 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_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/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..76cc2fb8b 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 != "" { @@ -284,7 +293,8 @@ 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)) } diff --git a/clientconfig/routing.go b/clientconfig/routing.go new file mode 100644 index 000000000..3ccfe57f9 --- /dev/null +++ b/clientconfig/routing.go @@ -0,0 +1,180 @@ +package clientconfig + +import ( + "context" + "crypto/sha1" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "encoding/json" + "encoding/pem" + "fmt" + "net" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "github.com/quic-go/quic-go" + "miren.dev/runtime/pkg/rpc" +) + +const cloudRouteCacheTTL = 30 * time.Second + +// Only failed direct attempts are cached. A working direct route is checked +// again on the next connection, so leaving its network never pins us to it. +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 + } + // 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() + cachePath := filepath.Join(cacheDir, fmt.Sprintf("cloud_route_%x", key)) + if cacheDir != "" { + if info, err := os.Stat(cachePath); err == nil && time.Since(info.ModTime()) < cloudRouteCacheTTL { + return c.viaCloudOptions(ctx, config) + } + } + + addresses := slices.Clone(c.AllAddresses) + if c.Hostname != "" && !slices.Contains(addresses, c.Hostname) { + addresses = append(addresses, c.Hostname) + } + type result struct { + cluster *ClusterConfig + err error + } + results := make(chan result, len(addresses)) + probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + for _, address := range addresses { + go func() { + cluster, err := c.probeDirectRoute(probeCtx, address) + results <- result{cluster, err} + }() + } + for range addresses { + select { + case res := <-results: + if res.err == nil { + cancel() + return res.cluster.directRPCOptions(ctx, config, "") + } + case <-probeCtx.Done(): + goto relay + } + } + +relay: + if err := ctx.Err(); err != nil { + return nil, err + } + opts, err := c.viaCloudOptions(ctx, config) + if err == nil && cacheDir != "" { + _ = os.WriteFile(cachePath, nil, 0600) + } + return opts, err +} + +// 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 != "" { + // Replace system-root validation with a pinned-root chain validation, + // not an unauthenticated connectivity check. + tlsConfig.InsecureSkipVerify = true + tlsConfig.VerifyConnection = func(state tls.ConnectionState) error { + root := state.PeerCertificates[len(state.PeerCertificates)-1] + sum := sha1.Sum(root.Raw) + if !strings.EqualFold(hex.EncodeToString(sum[:]), c.CACertFingerprint) { + return fmt.Errorf("cluster CA fingerprint mismatch") + } + roots := x509.NewCertPool() + roots.AddCert(root) + intermediates := x509.NewCertPool() + for _, cert := range state.PeerCertificates[1:] { + intermediates.AddCert(cert) + } + name := tlsConfig.ServerName + leaf := state.PeerCertificates[0] + if c.TLSServerName == "" && leaf.VerifyHostname(name) != nil && slices.Contains(leaf.DNSNames, APIServerName) { + name = APIServerName + } + if _, err := leaf.Verify(x509.VerifyOptions{Roots: roots, Intermediates: intermediates, DNSName: name}); err != nil { + return err + } + direct.CACert = string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: root.Raw})) + direct.TLSServerName = name + return nil + } + } + 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() + conn, err := transport.Dial(ctx, &net.UDPAddr{IP: ip.IP, Port: port, Zone: ip.Zone}, tlsConfig, &quic.Config{ + InitialPacketSize: rpc.InitialPacketSize, HandshakeIdleTimeout: 2 * time.Second, + }) + 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..118d7fde0 --- /dev/null +++ b/clientconfig/routing_test.go @@ -0,0 +1,208 @@ +package clientconfig + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" + "miren.dev/runtime/pkg/caauth" + "miren.dev/runtime/pkg/rpc" +) + +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) + connect := func() error { + state, err := cluster.State(t.Context(), 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" + require.ErrorContains(t, connect(), "503", "a mismatched CA must fall back to cloud") + cluster.CACertFingerprint = ca.Fingerprint() + 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() + + // 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") + }) + + // Cache expiry must restore the direct attempt with unchanged config. + 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) + require.ErrorContains(t, connect(), "503", "cloud hint skips probing until it expires") + cacheDir, err := getCacheDir() + require.NoError(t, err) + paths, err := filepath.Glob(filepath.Join(cacheDir, "cloud_route_*")) + require.NoError(t, err) + for _, path := range paths { + old := time.Now().Add(-time.Minute) + require.NoError(t, os.Chtimes(path, old, old)) + } + assertDirect() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = cluster.State(ctx, 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 { + _ = state.Close() + } + 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") +} diff --git a/docs/docs/miren-cloud/cloud-routed-clusters.md b/docs/docs/miren-cloud/cloud-routed-clusters.md index b4544119d..5d07eb719 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,23 @@ 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. +Each subsequent connection tries the saved direct addresses first, then falls +back to cloud. Failed direct attempts are cached for 30 seconds, so repeated +commands off-network do not each wait for a timeout. After that, direct routing +is tried again. Moving between networks does not permanently change your config. +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. + 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 +206,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[Route selection precedes RPC] + +Automatic routing checks direct connectivity with a verified TLS handshake. +Once that succeeds, the command uses a separate direct RPC connection. If that +connection subsequently fails, the command fails rather than replaying it +through cloud. In-flight operations are never automatically replayed. + +::: + +:::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 From b58ecbb1a43e9b99d7a80a1a00510b581fa7c3e5 Mon Sep 17 00:00:00 2001 From: Evan Phoenix Date: Thu, 29 Oct 2026 23:25:47 +0000 Subject: [PATCH 2/4] Preserve TLS verification during cluster route discovery Recover pinned roots from failed verified handshakes and retry with normal TLS validation. Add routing diagnostics and distinguish add-time cloud fallback from forced routing in CLI output. --- cli/commands/cluster.go | 2 + cli/commands/cluster_add.go | 8 +- cli/commands/cluster_add_byname_test.go | 14 +++- cli/commands/cluster_add_json.go | 10 ++- cli/commands/cluster_add_json_test.go | 21 ++++++ clientconfig/routing.go | 75 +++++++++++-------- clientconfig/routing_test.go | 39 +++++++++- docs/docs/command/cluster-add.md | 2 +- .../docs/miren-cloud/cloud-routed-clusters.md | 9 +++ 9 files changed, 139 insertions(+), 41 deletions(-) diff --git a/cli/commands/cluster.go b/cli/commands/cluster.go index 9dd532da2..be8b1c371 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 e82cd364e..a1f506456 100644 --- a/cli/commands/cluster_add.go +++ b/cli/commands/cluster_add.go @@ -569,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 { @@ -592,7 +595,8 @@ func addCluster(ctx *Context, opts addClusterOptions) (*addedCluster, error) { Name: clusterName, XID: clusterXID, Address: address, - ViaCloud: clusterConfig.ViaCloud, + 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 84e812efd..693be0ea1 100644 --- a/cli/commands/cluster_add_byname_test.go +++ b/cli/commands/cluster_add_byname_test.go @@ -80,9 +80,13 @@ func TestAddClusterFallbackPreservesAutomaticRouting(t *testing.T) { withCloudRouteProbe(t, func(*Context, *clientconfig.Config, string, *clientconfig.IdentityConfig, *ClusterResponse) bool { return true }) - added, err := addCluster(presenceContext(t), addClusterOptions{clusterName: "prod"}) + ctx := presenceContext(t) + added, err := addCluster(ctx, addClusterOptions{clusterName: "prod"}) require.NoError(t, err) - require.False(t, added.ViaCloud, "a successful fallback is not a forced route") + 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") @@ -90,6 +94,12 @@ func TestAddClusterFallbackPreservesAutomaticRouting(t *testing.T) { 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 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/clientconfig/routing.go b/clientconfig/routing.go index 3ccfe57f9..42f1c4a5f 100644 --- a/clientconfig/routing.go +++ b/clientconfig/routing.go @@ -9,6 +9,7 @@ import ( "encoding/hex" "encoding/json" "encoding/pem" + "errors" "fmt" "net" "os" @@ -17,6 +18,7 @@ import ( "strings" "time" + "github.com/containerd/log" "github.com/quic-go/quic-go" "miren.dev/runtime/pkg/rpc" ) @@ -44,6 +46,8 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp cachePath := filepath.Join(cacheDir, fmt.Sprintf("cloud_route_%x", key)) if cacheDir != "" { if info, err := os.Stat(cachePath); err == nil && time.Since(info.ModTime()) < cloudRouteCacheTTL { + reason, _ := os.ReadFile(cachePath) + log.G(ctx).WithField("cluster", c.XID).WithField("direct_errors", string(reason)).Debug("Using cached direct-route failure; trying cloud") return c.viaCloudOptions(ctx, config) } } @@ -55,6 +59,7 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp type result struct { cluster *ClusterConfig err error + address string } results := make(chan result, len(addresses)) probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second) @@ -62,9 +67,10 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp for _, address := range addresses { go func() { cluster, err := c.probeDirectRoute(probeCtx, address) - results <- result{cluster, err} + results <- result{cluster, err, address} }() } + var failures []string for range addresses { select { case res := <-results: @@ -72,7 +78,11 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp cancel() return res.cluster.directRPCOptions(ctx, config, "") } + failure := fmt.Sprintf("%s: %v", res.address, 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.Sprintf("probing %v: %v", addresses, probeCtx.Err())) goto relay } } @@ -81,9 +91,10 @@ relay: if err := ctx.Err(); err != nil { return nil, err } + log.G(ctx).WithField("cluster", c.XID).WithField("direct_errors", strings.Join(failures, "; ")).Debug("No verified direct route; trying cloud") opts, err := c.viaCloudOptions(ctx, config) if err == nil && cacheDir != "" { - _ = os.WriteFile(cachePath, nil, 0600) + _ = os.WriteFile(cachePath, []byte(strings.Join(failures, "; ")), 0600) } return opts, err } @@ -130,33 +141,8 @@ func (c *ClusterConfig) probeDirectRoute(ctx context.Context, address string) (* return nil, fmt.Errorf("invalid cluster CA") } } else if c.CACertFingerprint != "" { - // Replace system-root validation with a pinned-root chain validation, - // not an unauthenticated connectivity check. - tlsConfig.InsecureSkipVerify = true - tlsConfig.VerifyConnection = func(state tls.ConnectionState) error { - root := state.PeerCertificates[len(state.PeerCertificates)-1] - sum := sha1.Sum(root.Raw) - if !strings.EqualFold(hex.EncodeToString(sum[:]), c.CACertFingerprint) { - return fmt.Errorf("cluster CA fingerprint mismatch") - } - roots := x509.NewCertPool() - roots.AddCert(root) - intermediates := x509.NewCertPool() - for _, cert := range state.PeerCertificates[1:] { - intermediates.AddCert(cert) - } - name := tlsConfig.ServerName - leaf := state.PeerCertificates[0] - if c.TLSServerName == "" && leaf.VerifyHostname(name) != nil && slices.Contains(leaf.DNSNames, APIServerName) { - name = APIServerName - } - if _, err := leaf.Verify(x509.VerifyOptions{Roots: roots, Intermediates: intermediates, DNSName: name}); err != nil { - return err - } - direct.CACert = string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: root.Raw})) - direct.TLSServerName = name - return nil - } + // Require pinned-root recovery even if the peer has a system-trusted CA. + tlsConfig.RootCAs = x509.NewCertPool() } network := "udp6" if ip.IP.To4() != nil { @@ -169,9 +155,36 @@ func (c *ClusterConfig) probeDirectRoute(ctx context.Context, address string) (* defer socket.Close() transport := &quic.Transport{Conn: socket} defer transport.Close() - conn, err := transport.Dial(ctx, &net.UDPAddr{IP: ip.IP, Port: port, Zone: ip.Zone}, tlsConfig, &quic.Config{ + 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 } diff --git a/clientconfig/routing_test.go b/clientconfig/routing_test.go index 118d7fde0..d4fd0a51b 100644 --- a/clientconfig/routing_test.go +++ b/clientconfig/routing_test.go @@ -1,9 +1,11 @@ package clientconfig import ( + "bytes" "context" "errors" "fmt" + "log/slog" "net" "net/http" "net/http/httptest" @@ -17,6 +19,7 @@ import ( "gopkg.in/yaml.v3" "miren.dev/runtime/pkg/caauth" "miren.dev/runtime/pkg/rpc" + "miren.dev/runtime/pkg/slogrus" ) type routingCredentialRecorder struct { @@ -52,8 +55,10 @@ func TestAutomaticClusterRouting(t *testing.T) { 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(t.Context(), config) + state, err := cluster.State(ctx, config) require.NoError(t, err) defer state.Close() _, err = state.Client("missing") @@ -73,10 +78,23 @@ func TestAutomaticClusterRouting(t *testing.T) { 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(), "Using cached direct-route failure") + 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 = "" @@ -103,6 +121,21 @@ func TestAutomaticClusterRouting(t *testing.T) { 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") @@ -151,9 +184,9 @@ insecure: true } assertDirect() - ctx, cancel := context.WithCancel(t.Context()) + canceledCtx, cancel := context.WithCancel(t.Context()) cancel() - _, err = cluster.State(ctx, config) + _, err = cluster.State(canceledCtx, config) require.ErrorIs(t, err, context.Canceled) } 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 5d07eb719..1c4d52d72 100644 --- a/docs/docs/miren-cloud/cloud-routed-clusters.md +++ b/docs/docs/miren-cloud/cloud-routed-clusters.md @@ -44,6 +44,15 @@ commands off-network do not each wait for a timeout. After that, direct routing is tried again. Moving between networks does not permanently change your config. 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, 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. This adds a +handshake attempt to the direct probe; entries with a saved CA do not need it. + +The cache records a failed direct attempt, not proof that cloud is available. +If the cluster has no working cloud link, commands can fail through cloud until +the hint expires. Use `miren -vv app list` to see why direct routing failed; +debug output includes the attempted addresses and errors, even on cache hits. 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 From cf26a97b838f85544f45ba874781a72bdd6f5ef5 Mon Sep 17 00:00:00 2001 From: Evan Phoenix Date: Tue, 3 Nov 2026 23:06:25 +0000 Subject: [PATCH 3/4] Reuse successful cluster routes and preserve fallback diagnostics Avoid repeated direct probes by remembering successful routes and recovered trust. Rediscover only during connection setup, expire successful cloud routes, and retain both failures in the normal CLI diagnostic. --- cli/commands/rpc_errors.go | 24 +++ cli/commands/rpc_errors_test.go | 45 +++++ clientconfig/routing.go | 176 +++++++++++++++--- clientconfig/routing_test.go | 149 +++++++++++++-- .../docs/miren-cloud/cloud-routed-clusters.md | 33 ++-- pkg/rpc/client.go | 27 +++ pkg/rpc/client_routing_test.go | 66 +++++++ pkg/rpc/state.go | 120 +++++++++--- pkg/rpc/transport_msg.go | 10 +- 9 files changed, 568 insertions(+), 82 deletions(-) create mode 100644 pkg/rpc/client_routing_test.go 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..ae554a905 100644 --- a/cli/commands/rpc_errors_test.go +++ b/cli/commands/rpc_errors_test.go @@ -1,11 +1,15 @@ package commands import ( + "bytes" "context" "errors" "fmt" "io" "log/slog" + "net" + "net/http" + "net/http/httptest" "strings" "testing" "time" @@ -19,6 +23,47 @@ func diagnosticContext() *Context { return &Context{ClusterName: "local", CommandName: "route list"} } +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/routing.go b/clientconfig/routing.go index 42f1c4a5f..40b5f9177 100644 --- a/clientconfig/routing.go +++ b/clientconfig/routing.go @@ -25,8 +25,43 @@ import ( const cloudRouteCacheTTL = 30 * time.Second -// Only failed direct attempts are cached. A working direct route is checked -// again on the next connection, so leaving its network never pins us to it. +// 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 @@ -43,60 +78,153 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp }{c, cloudURL}) key := sha256.Sum256(data) cacheDir, _ := getCacheDir() - cachePath := filepath.Join(cacheDir, fmt.Sprintf("cloud_route_%x", key)) + var cachePath string if cacheDir != "" { - if info, err := os.Stat(cachePath); err == nil && time.Since(info.ModTime()) < cloudRouteCacheTTL { - reason, _ := os.ReadFile(cachePath) - log.G(ctx).WithField("cluster", c.XID).WithField("direct_errors", string(reason)).Debug("Using cached direct-route failure; trying cloud") - return c.viaCloudOptions(ctx, config) + 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 + var err error + if route.Cloud { + opts, err = c.viaCloudOptions(ctx, config) + } else { + direct := *c + direct.Hostname, direct.AllAddresses = route.Address, nil + direct.CACert, direct.TLSServerName = route.CACert, route.TLSServerName + opts, err = direct.directRPCOptions(ctx, config, "") + } + if err != nil { + return nil, err + } + 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} + results <- result{cluster, err, address, time.Since(started)} }() } - var failures []string + var failures []error for range addresses { select { case res := <-results: if res.err == nil { cancel() - return res.cluster.directRPCOptions(ctx, config, "") + 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) } - failure := fmt.Sprintf("%s: %v", res.address, 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.Sprintf("probing %v: %v", addresses, probeCtx.Err())) - goto relay + failures = append(failures, fmt.Errorf("probing %v: %w", addresses, probeCtx.Err())) + return nil, errors.Join(failures...) } } - -relay: - if err := ctx.Err(); err != nil { - return nil, err - } - log.G(ctx).WithField("cluster", c.XID).WithField("direct_errors", strings.Join(failures, "; ")).Debug("No verified direct route; trying cloud") - opts, err := c.viaCloudOptions(ctx, config) - if err == nil && cacheDir != "" { - _ = os.WriteFile(cachePath, []byte(strings.Join(failures, "; ")), 0600) - } - return opts, err + return nil, errors.Join(failures...) } // Probe without sending credentials. If add ran off-network, cloud's CA diff --git a/clientconfig/routing_test.go b/clientconfig/routing_test.go index d4fd0a51b..258754894 100644 --- a/clientconfig/routing_test.go +++ b/clientconfig/routing_test.go @@ -3,18 +3,24 @@ 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" @@ -84,7 +90,7 @@ func TestAutomaticClusterRouting(t *testing.T) { require.Contains(t, diagnostics.String(), "cluster CA fingerprint mismatch") diagnostics.Reset() require.ErrorContains(t, connect(), "503") - require.Contains(t, diagnostics.String(), "Using cached direct-route failure") + 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() @@ -167,21 +173,12 @@ insecure: true require.ErrorContains(t, connect(), "503") }) - // Cache expiry must restore the direct attempt with unchanged config. + // 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) - require.ErrorContains(t, connect(), "503", "cloud hint skips probing until it expires") - cacheDir, err := getCacheDir() - require.NoError(t, err) - paths, err := filepath.Glob(filepath.Join(cacheDir, "cloud_route_*")) - require.NoError(t, err) - for _, path := range paths { - old := time.Now().Add(-time.Minute) - require.NoError(t, os.Chtimes(path, old, old)) - } assertDirect() canceledCtx, cancel := context.WithCancel(t.Context()) @@ -220,7 +217,10 @@ func TestAutomaticClusterRoutingCancelsDNS(t *testing.T) { go func() { state, err := cluster.State(ctx, config) if state != nil { - _ = state.Close() + defer state.Close() + } + if err == nil { + _, err = state.Client("missing") } result <- err }() @@ -239,3 +239,128 @@ func TestAutomaticClusterRoutingCancelsDNS(t *testing.T) { 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()) + 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/docs/docs/miren-cloud/cloud-routed-clusters.md b/docs/docs/miren-cloud/cloud-routed-clusters.md index 1c4d52d72..ddd3d3869 100644 --- a/docs/docs/miren-cloud/cloud-routed-clusters.md +++ b/docs/docs/miren-cloud/cloud-routed-clusters.md @@ -38,21 +38,28 @@ 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. -Each subsequent connection tries the saved direct addresses first, then falls -back to cloud. Failed direct attempts are cached for 30 seconds, so repeated -commands off-network do not each wait for a timeout. After that, direct routing -is tried again. Moving between networks does not permanently change your config. +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. + +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, 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. This adds a -handshake attempt to the direct probe; entries with a saved CA do not need it. - -The cache records a failed direct attempt, not proof that cloud is available. -If the cluster has no working cloud link, commands can fail through cloud until -the hint expires. Use `miren -vv app list` to see why direct routing failed; -debug output includes the attempted addresses and errors, even on cache hits. +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 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 { From 63c60335a75514e789a5c3555ae7d9223f144635 Mon Sep 17 00:00:00 2001 From: Evan Phoenix Date: Tue, 3 Nov 2026 23:34:11 +0000 Subject: [PATCH 4/4] Resolve cluster credentials before selecting a route Keep authentication failures out of route rediscovery and preserve login and rate-limit diagnostics. Reuse the resolved bearer token across connection attempts, retain refresh retry timing, and correct the connection-setup retry documentation. --- cli/commands/rpc_errors_test.go | 57 +++++++++++++++++++ clientconfig/local.go | 6 +- clientconfig/routing.go | 22 +++++-- clientconfig/routing_test.go | 12 ++++ clientconfig/token.go | 4 +- .../docs/miren-cloud/cloud-routed-clusters.md | 14 +++-- 6 files changed, 102 insertions(+), 13 deletions(-) diff --git a/cli/commands/rpc_errors_test.go b/cli/commands/rpc_errors_test.go index ae554a905..a0aa2265b 100644 --- a/cli/commands/rpc_errors_test.go +++ b/cli/commands/rpc_errors_test.go @@ -10,10 +10,14 @@ import ( "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" @@ -23,6 +27,59 @@ 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") diff --git a/clientconfig/local.go b/clientconfig/local.go index 76cc2fb8b..ef76f0a08 100644 --- a/clientconfig/local.go +++ b/clientconfig/local.go @@ -278,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{ @@ -298,7 +302,7 @@ func (c *ClusterConfig) viaCloudOptions(ctx context.Context, config *Config) ([] 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 index 40b5f9177..90a75d205 100644 --- a/clientconfig/routing.go +++ b/clientconfig/routing.go @@ -70,6 +70,16 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp 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 { @@ -88,17 +98,17 @@ func (c *ClusterConfig) routedOptions(ctx context.Context, config *Config) ([]rp 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 - var err error if route.Cloud { - opts, err = c.viaCloudOptions(ctx, config) + 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, err = direct.directRPCOptions(ctx, config, "") - } - if err != nil { - return nil, err + opts = append(direct.dialOptions(route.Address), rpc.WithBearerToken(token)) } return connect(opts...) } diff --git a/clientconfig/routing_test.go b/clientconfig/routing_test.go index 258754894..6a78b94a5 100644 --- a/clientconfig/routing_test.go +++ b/clientconfig/routing_test.go @@ -326,6 +326,18 @@ func TestSuccessfulClusterRoutes(t *testing.T) { 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 } 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/miren-cloud/cloud-routed-clusters.md b/docs/docs/miren-cloud/cloud-routed-clusters.md index ddd3d3869..5f835c037 100644 --- a/docs/docs/miren-cloud/cloud-routed-clusters.md +++ b/docs/docs/miren-cloud/cloud-routed-clusters.md @@ -45,6 +45,10 @@ failed connection discards the cached route, discovers a route again, and retrie 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. @@ -222,12 +226,12 @@ 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. -:::note[Route selection precedes RPC] +:::note[Only connection setup is retried] -Automatic routing checks direct connectivity with a verified TLS handshake. -Once that succeeds, the command uses a separate direct RPC connection. If that -connection subsequently fails, the command fails rather than replaying it -through cloud. In-flight operations are never automatically replayed. +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. :::