tangled.org / core
Monorepo for Tangled tangled.org
1.4k
Fork 229

Configure Feed

Select the types of activity you want to include in your feed.

spindle/mill: proxy debug ssh to executors

Signed-off-by: dawn <dawn@tangled.org>

author ptr.pet committer
Tangled date (Aug 13, 2026, 12:11 PM UTC) commit d1086e3d parent 8aeeb0ae change-id wrukxpps
build.yml
commit-lint.yml
fmt.yml
rust-checks.yml
rust-test-bobbin.yml
rust-test-knot.yml
test.yml
+476 -25
+20 -11
docker-compose.mill.yml
··· 33 33 SPINDLE_NIX_CACHE_UPLOAD_URL: http://ncps:8501/upload 34 34 SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED: "true" 35 35 SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2223 36 - SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: 127.0.0.1 37 - SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_JUMP_HOST: chernobog 36 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: spindle-executor 37 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_JUMP_HOST: debug@spindle.tngl.boltless.dev:2224 38 38 SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD: 10m 39 39 SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST_KEY_PATH: /var/lib/spindle/debug_ssh_host_key 40 40 # dials the mill container directly using ws ··· 78 78 SPINDLE_ROLE: mill 79 79 SPINDLE_ARTIFACT_STORES_DISK_DIR: /var/lib/spindle/artifacts 80 80 SPINDLE_MILL_ARTIFACT_STORE: disk 81 + SPINDLE_MILL_JUMP_LISTEN_ADDR: 0.0.0.0:2224 82 + SPINDLE_MILL_JUMP_HOST_KEY_PATH: /var/lib/spindle/debug_jump_host_key 83 + SPINDLE_MILL_DEBUG_EXECUTOR_PORT: "2223" 84 + SPINDLE_MILL_MAX_JUMP_CONNECTIONS: "128" 81 85 volumes: 82 86 - spindle-artifacts:/var/lib/spindle/artifacts 87 + ports: !override 88 + - "127.0.0.1:2224:2224" 83 89 84 90 mill-tokens: 85 91 profiles: ["linux"] ··· 119 125 SPINDLE_MILL_LABELS: linux,fast 120 126 SPINDLE_MILL_TOKEN_FILE: /shared/executor-a.mill-token 121 127 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11241" 122 - SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2224 123 - ports: 124 - - "127.0.0.1:2224:2224" 128 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: executor-a 125 129 volumes: 126 130 - spindle-executor-a-data:/var/lib/spindle 127 131 - spindle-artifacts:/var/lib/spindle/artifacts 128 132 - ./out/localinfra-spindle-images:/var/lib/spindle/images:ro 129 133 - init-state:/shared:ro 130 134 - ./localinfra/certs/root.crt:/usr/local/share/ca-certificates/caddy.crt:ro 135 + networks: 136 + tngl: 137 + aliases: [executor-a] 131 138 132 139 spindle-executor-b: 133 140 <<: *mill-executor ··· 137 144 SPINDLE_MILL_LABELS: linux,slow 138 145 SPINDLE_MILL_TOKEN_FILE: /shared/executor-b.mill-token 139 146 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11242" 140 - SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2225 141 - ports: 142 - - "127.0.0.1:2225:2225" 147 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: executor-b 143 148 volumes: 144 149 - spindle-executor-b-data:/var/lib/spindle 145 150 - spindle-artifacts:/var/lib/spindle/artifacts 146 151 - ./out/localinfra-spindle-images:/var/lib/spindle/images:ro 147 152 - init-state:/shared:ro 148 153 - ./localinfra/certs/root.crt:/usr/local/share/ca-certificates/caddy.crt:ro 154 + networks: 155 + tngl: 156 + aliases: [executor-b] 149 157 150 158 spindle-executor-c: 151 159 <<: *mill-executor ··· 155 163 SPINDLE_MILL_LABELS: linux,gpu 156 164 SPINDLE_MILL_TOKEN_FILE: /shared/executor-c.mill-token 157 165 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11243" 158 - SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR: 0.0.0.0:2226 159 - ports: 160 - - "127.0.0.1:2226:2226" 166 + SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST: executor-c 161 167 volumes: 162 168 - spindle-executor-c-data:/var/lib/spindle 163 169 - spindle-artifacts:/var/lib/spindle/artifacts 164 170 - ./out/localinfra-spindle-images-alpine:/var/lib/spindle/images:ro 165 171 - init-state:/shared:ro 166 172 - ./localinfra/certs/root.crt:/usr/local/share/ca-certificates/caddy.crt:ro 173 + networks: 174 + tngl: 175 + aliases: [executor-c] 167 176 168 177 volumes: 169 178 spindle-executor-a-data:
+1 -1
docker-compose.yml
··· 203 203 - label=disable 204 204 - seccomp=unconfined 205 205 ports: 206 - - "2223:2223" 206 + - "127.0.0.1:2223:2223" 207 207 volumes: 208 208 - spindle-data:/var/lib/spindle 209 209 - spindle-logs:/var/log/spindle
+4
spindle/server.go
··· 322 322 go s.exec.Connect(ctx) 323 323 } 324 324 325 + if s.mill != nil && s.cfg.Mill.JumpListenAddr != "" { 326 + go s.mill.ServeJump(ctx, s.cfg.Mill.JumpListenAddr, s.cfg.Mill.JumpHostKeyPath, s.cfg.Mill.DebugExecutorPort, s.cfg.Mill.MaxJumpConnections) 327 + } 328 + 325 329 if stopper, ok := s.vault.(secrets.Stopper); ok { 326 330 defer stopper.Stop() 327 331 }
+31
nix/modules/spindle.nix
··· 348 348 }; 349 349 }; 350 350 }; 351 + mill = { 352 + jumpListenAddr = mkOption { 353 + type = types.str; 354 + default = ""; 355 + example = "0.0.0.0:22"; 356 + description = "Address for the mill's restricted debug SSH jump server."; 357 + }; 358 + 359 + jumpHostKeyPath = mkOption { 360 + type = with types; nullOr path; 361 + default = null; 362 + example = "/var/lib/spindle/debug_jump_host_key"; 363 + description = "Path to the debug SSH jump server host key."; 364 + }; 365 + 366 + debugExecutorPort = mkOption { 367 + type = types.port; 368 + default = 2223; 369 + description = "Private debug SSH port shared by executors."; 370 + }; 371 + 372 + maxJumpConnections = mkOption { 373 + type = types.ints.positive; 374 + default = 128; 375 + description = "Maximum concurrent connections to the mill's debug SSH jump server."; 376 + }; 377 + }; 351 378 352 379 environmentFile = mkOption { 353 380 type = with types; nullOr path; ··· 458 485 "SPINDLE_ARTIFACT_STORES_S3_BUCKET=${cfg.artifactStores.s3.bucket}" 459 486 "SPINDLE_ARTIFACT_STORES_S3_REGION=${cfg.artifactStores.s3.region}" 460 487 "SPINDLE_MILL_ARTIFACT_STORE=s3" 488 + "SPINDLE_MILL_JUMP_LISTEN_ADDR=${cfg.mill.jumpListenAddr}" 489 + "SPINDLE_MILL_JUMP_HOST_KEY_PATH=${optionalString (cfg.mill.jumpHostKeyPath != null) (toString cfg.mill.jumpHostKeyPath)}" 490 + "SPINDLE_MILL_DEBUG_EXECUTOR_PORT=${toString cfg.mill.debugExecutorPort}" 491 + "SPINDLE_MILL_MAX_JUMP_CONNECTIONS=${toString cfg.mill.maxJumpConnections}" 461 492 ]; 462 493 ExecStart = "${cfg.package}/bin/spindle"; 463 494 Restart = "always";
+22 -7
spindle/config/config.go
··· 138 138 139 139 // fields are selectively active depending on the role 140 140 type Mill struct { 141 - URL string `env:"URL"` // mill websocket endpoint dialled by the executor 142 - SharedSecret string `env:"SHARED_SECRET"` // the executor's token for dialing the mill 143 - MaxPending int `env:"MAX_PENDING, default=100"` // mill pending job queue limit 144 - ReconnectGrace time.Duration `env:"RECONNECT_GRACE, default=45s"` // reconnect window before leases are failed 145 - Seats int `env:"SEATS, default=4"` // executor seats advertised to the mill 146 - Labels []string `env:"LABELS"` // executor capability labels 147 - ArtifactStore string `env:"ARTIFACT_STORE"` // store shared by mill and its executors 141 + URL string `env:"URL"` // mill websocket endpoint dialled by the executor 142 + SharedSecret string `env:"SHARED_SECRET"` // the executor's token for dialing the mill 143 + MaxPending int `env:"MAX_PENDING, default=100"` // mill pending job queue limit 144 + ReconnectGrace time.Duration `env:"RECONNECT_GRACE, default=45s"` // reconnect window before leases are failed 145 + Seats int `env:"SEATS, default=4"` // executor seats advertised to the mill 146 + Labels []string `env:"LABELS"` // executor capability labels 147 + ArtifactStore string `env:"ARTIFACT_STORE"` // store shared by mill and its executors 148 + JumpListenAddr string `env:"JUMP_LISTEN_ADDR"` 149 + JumpHostKeyPath string `env:"JUMP_HOST_KEY_PATH"` 150 + DebugExecutorPort uint32 `env:"DEBUG_EXECUTOR_PORT, default=2223"` 151 + MaxJumpConnections int `env:"MAX_JUMP_CONNECTIONS, default=128"` 148 152 } 149 153 150 154 type Config struct { ··· 173 177 } 174 178 default: 175 179 return fmt.Errorf("unknown SPINDLE_ROLE %q (want standalone, mill, or executor)", c.Role) 180 + } 181 + if c.Mill.JumpListenAddr != "" { 182 + if c.Role != RoleMill { 183 + return fmt.Errorf("SPINDLE_MILL_JUMP_LISTEN_ADDR requires SPINDLE_ROLE=mill") 184 + } 185 + if c.Mill.JumpHostKeyPath == "" { 186 + return fmt.Errorf("SPINDLE_MILL_JUMP_LISTEN_ADDR requires SPINDLE_MILL_JUMP_HOST_KEY_PATH") 187 + } 188 + if c.Mill.MaxJumpConnections <= 0 { 189 + return fmt.Errorf("SPINDLE_MILL_MAX_JUMP_CONNECTIONS must be greater than zero") 190 + } 176 191 } 177 192 return nil 178 193 }
+45
spindle/config/config_test.go
··· 18 18 t.Fatalf("image directory = %q, want empty", cfg.MicroVMPipelines.ImageDir) 19 19 } 20 20 } 21 + 22 + func TestLoadRequiresJumpHostKey(t *testing.T) { 23 + t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") 24 + t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com") 25 + t.Setenv("SPINDLE_ROLE", "mill") 26 + t.Setenv("SPINDLE_MILL_JUMP_LISTEN_ADDR", "0.0.0.0:22") 27 + 28 + if _, err := Load(context.Background()); err == nil { 29 + t.Fatal("Load accepted a jump listener without a host key path") 30 + } 31 + } 32 + 33 + func TestLoadRejectsJumpListenerOutsideMill(t *testing.T) { 34 + t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") 35 + t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com") 36 + t.Setenv("SPINDLE_ROLE", "standalone") 37 + t.Setenv("SPINDLE_MILL_JUMP_LISTEN_ADDR", "0.0.0.0:22") 38 + t.Setenv("SPINDLE_MILL_JUMP_HOST_KEY_PATH", "/tmp/jump-host-key") 39 + 40 + if _, err := Load(context.Background()); err == nil { 41 + t.Fatal("Load accepted a mill jump listener in standalone mode") 42 + } 43 + } 44 + 45 + func TestLoadValidatesMaxJumpConnections(t *testing.T) { 46 + t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") 47 + t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com") 48 + t.Setenv("SPINDLE_ROLE", "mill") 49 + t.Setenv("SPINDLE_MILL_JUMP_LISTEN_ADDR", "0.0.0.0:22") 50 + t.Setenv("SPINDLE_MILL_JUMP_HOST_KEY_PATH", "/tmp/jump-host-key") 51 + t.Setenv("SPINDLE_MILL_MAX_JUMP_CONNECTIONS", "0") 52 + 53 + if _, err := Load(context.Background()); err == nil { 54 + t.Fatal("Load accepted a non-positive jump connection limit") 55 + } 56 + 57 + t.Setenv("SPINDLE_MILL_MAX_JUMP_CONNECTIONS", "17") 58 + cfg, err := Load(context.Background()) 59 + if err != nil { 60 + t.Fatal(err) 61 + } 62 + if cfg.Mill.MaxJumpConnections != 17 { 63 + t.Fatalf("max jump connections = %d, want 17", cfg.Mill.MaxJumpConnections) 64 + } 65 + }
+170
spindle/mill/jump.go
··· 1 + package mill 2 + 3 + import ( 4 + "context" 5 + "crypto/ed25519" 6 + "crypto/rand" 7 + "encoding/pem" 8 + "fmt" 9 + "net" 10 + "os" 11 + "path/filepath" 12 + "sync" 13 + "time" 14 + 15 + "github.com/gliderlabs/ssh" 16 + gossh "golang.org/x/crypto/ssh" 17 + ) 18 + 19 + const ( 20 + jumpIdleTimeout = 5 * time.Minute 21 + jumpMaxTimeout = 24 * time.Hour 22 + maxJumpConnectionsPerIP = 8 23 + ) 24 + 25 + type jumpContextKey string 26 + 27 + const jumpRouteOpened jumpContextKey = "route-opened" 28 + 29 + func (m *Mill) ServeJump(ctx context.Context, listenAddr, hostKeyPath string, executorPort uint32, maxConnections int) { 30 + if listenAddr == "" { 31 + return 32 + } 33 + srv, err := m.newJumpServer(hostKeyPath, executorPort, maxConnections) 34 + if err != nil { 35 + m.l.Error("setup debug ssh jump server", "err", err) 36 + return 37 + } 38 + 39 + go func() { 40 + <-ctx.Done() 41 + _ = srv.Close() 42 + }() 43 + 44 + m.l.Info("starting debug ssh jump server", "address", listenAddr) 45 + srv.Addr = listenAddr 46 + if err := srv.ListenAndServe(); err != nil && err != ssh.ErrServerClosed { 47 + m.l.Error("debug ssh jump server stopped", "err", err) 48 + } 49 + } 50 + 51 + func (m *Mill) newJumpServer(hostKeyPath string, executorPort uint32, maxConnections int) (*ssh.Server, error) { 52 + if executorPort == 0 { 53 + executorPort = 2223 54 + } 55 + if maxConnections <= 0 { 56 + return nil, fmt.Errorf("max jump connections must be greater than zero") 57 + } 58 + if err := ensureJumpHostKey(hostKeyPath); err != nil { 59 + return nil, fmt.Errorf("prepare jump host key: %w", err) 60 + } 61 + limiter := newJumpConnectionLimiter(maxConnections, maxJumpConnectionsPerIP) 62 + srv := &ssh.Server{ 63 + PublicKeyHandler: func(ctx ssh.Context, _ ssh.PublicKey) bool { 64 + return ctx.User() == "debug" 65 + }, 66 + ConnCallback: func(ctx ssh.Context, conn net.Conn) net.Conn { 67 + if !limiter.acquire(conn.RemoteAddr()) { 68 + _ = conn.Close() 69 + return conn 70 + } 71 + go func() { 72 + <-ctx.Done() 73 + limiter.release(conn.RemoteAddr()) 74 + }() 75 + return conn 76 + }, 77 + LocalPortForwardingCallback: func(ctx ssh.Context, host string, port uint32) bool { 78 + if port != executorPort || !m.hasLiveExecutor(host) { 79 + return false 80 + } 81 + ctx.Lock() 82 + defer ctx.Unlock() 83 + if opened, _ := ctx.Value(jumpRouteOpened).(bool); opened { 84 + return false 85 + } 86 + ctx.SetValue(jumpRouteOpened, true) 87 + return true 88 + }, 89 + ChannelHandlers: map[string]ssh.ChannelHandler{ 90 + "direct-tcpip": ssh.DirectTCPIPHandler, 91 + }, 92 + IdleTimeout: jumpIdleTimeout, 93 + MaxTimeout: jumpMaxTimeout, 94 + } 95 + if err := srv.SetOption(ssh.HostKeyFile(hostKeyPath)); err != nil { 96 + return nil, fmt.Errorf("load jump host key: %w", err) 97 + } 98 + return srv, nil 99 + } 100 + 101 + type jumpConnectionLimiter struct { 102 + mu sync.Mutex 103 + total int 104 + perIP map[string]int 105 + maxTotal int 106 + maxPerIP int 107 + } 108 + 109 + func newJumpConnectionLimiter(maxTotal, maxPerIP int) *jumpConnectionLimiter { 110 + return &jumpConnectionLimiter{ 111 + perIP: make(map[string]int), 112 + maxTotal: maxTotal, 113 + maxPerIP: maxPerIP, 114 + } 115 + } 116 + 117 + func (l *jumpConnectionLimiter) acquire(addr net.Addr) bool { 118 + host := jumpRemoteHost(addr) 119 + l.mu.Lock() 120 + defer l.mu.Unlock() 121 + if l.total >= l.maxTotal || l.perIP[host] >= l.maxPerIP { 122 + return false 123 + } 124 + l.total++ 125 + l.perIP[host]++ 126 + return true 127 + } 128 + 129 + func (l *jumpConnectionLimiter) release(addr net.Addr) { 130 + host := jumpRemoteHost(addr) 131 + l.mu.Lock() 132 + defer l.mu.Unlock() 133 + l.total-- 134 + l.perIP[host]-- 135 + if l.perIP[host] == 0 { 136 + delete(l.perIP, host) 137 + } 138 + } 139 + 140 + func jumpRemoteHost(addr net.Addr) string { 141 + if addr == nil { 142 + return "" 143 + } 144 + host, _, err := net.SplitHostPort(addr.String()) 145 + if err != nil { 146 + return addr.String() 147 + } 148 + return host 149 + } 150 + 151 + func ensureJumpHostKey(path string) error { 152 + if _, err := os.Stat(path); err == nil { 153 + return nil 154 + } else if !os.IsNotExist(err) { 155 + return fmt.Errorf("stat host key: %w", err) 156 + } 157 + 158 + _, privateKey, err := ed25519.GenerateKey(rand.Reader) 159 + if err != nil { 160 + return fmt.Errorf("generate host key: %w", err) 161 + } 162 + block, err := gossh.MarshalPrivateKey(privateKey, "") 163 + if err != nil { 164 + return fmt.Errorf("marshal host key: %w", err) 165 + } 166 + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { 167 + return fmt.Errorf("create host key directory: %w", err) 168 + } 169 + return os.WriteFile(path, pem.EncodeToMemory(block), 0o600) 170 + }
+158
spindle/mill/jump_test.go
··· 1 + package mill 2 + 3 + import ( 4 + "io" 5 + "log/slog" 6 + "net" 7 + "os" 8 + "strconv" 9 + "testing" 10 + "time" 11 + 12 + gossh "golang.org/x/crypto/ssh" 13 + ) 14 + 15 + func TestJumpConnectionLimiter(t *testing.T) { 16 + limiter := newJumpConnectionLimiter(2, 1) 17 + first := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 1000} 18 + sameIP := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 1001} 19 + second := &net.TCPAddr{IP: net.ParseIP("192.0.2.2"), Port: 1000} 20 + third := &net.TCPAddr{IP: net.ParseIP("192.0.2.3"), Port: 1000} 21 + 22 + if !limiter.acquire(first) { 23 + t.Fatal("rejected first connection") 24 + } 25 + if limiter.acquire(sameIP) { 26 + t.Fatal("accepted a second connection from an exhausted IP") 27 + } 28 + if !limiter.acquire(second) { 29 + t.Fatal("rejected connection within the global limit") 30 + } 31 + if limiter.acquire(third) { 32 + t.Fatal("accepted a connection beyond the global limit") 33 + } 34 + limiter.release(first) 35 + if !limiter.acquire(sameIP) { 36 + t.Fatal("did not release the per-IP slot") 37 + } 38 + } 39 + 40 + func TestJumpServerForwardsOnlyLiveExecutorRoute(t *testing.T) { 41 + backend := startJumpBackend(t) 42 + _, portText, err := net.SplitHostPort(backend.Addr().String()) 43 + if err != nil { 44 + t.Fatal(err) 45 + } 46 + port, err := strconv.ParseUint(portText, 10, 32) 47 + if err != nil { 48 + t.Fatal(err) 49 + } 50 + 51 + m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{}) 52 + m.sessions["127.0.0.1"] = newSession("127.0.0.1", "epoch", nil, nil, m.l) 53 + 54 + hostKeyPath := t.TempDir() + "/host-key" 55 + srv, err := m.newJumpServer(hostKeyPath, uint32(port), 2) 56 + if err != nil { 57 + t.Fatal(err) 58 + } 59 + listener, err := net.Listen("tcp", "127.0.0.1:0") 60 + if err != nil { 61 + t.Fatal(err) 62 + } 63 + go func() { _ = srv.Serve(listener) }() 64 + t.Cleanup(func() { _ = srv.Close() }) 65 + 66 + if unauthorized, err := dialJump(t, listener.Addr().String(), "operator"); err == nil { 67 + _ = unauthorized.Close() 68 + t.Fatal("jump server accepted a non-debug user") 69 + } 70 + 71 + client := jumpClient(t, listener.Addr().String()) 72 + defer client.Close() 73 + 74 + if _, err := client.Dial("tcp", net.JoinHostPort("missing", portText)); err == nil { 75 + t.Fatal("forwarded an executor without a live mill session") 76 + } 77 + wrongPort := strconv.FormatUint(port+1, 10) 78 + if _, err := client.Dial("tcp", net.JoinHostPort("127.0.0.1", wrongPort)); err == nil { 79 + t.Fatal("forwarded an executor on an unconfigured port") 80 + } 81 + 82 + conn, err := client.Dial("tcp", net.JoinHostPort("127.0.0.1", portText)) 83 + if err != nil { 84 + t.Fatal(err) 85 + } 86 + defer conn.Close() 87 + if _, err := conn.Write([]byte("hello")); err != nil { 88 + t.Fatal(err) 89 + } 90 + got := make([]byte, 5) 91 + if _, err := io.ReadFull(conn, got); err != nil { 92 + t.Fatal(err) 93 + } 94 + if string(got) != "hello" { 95 + t.Fatalf("forwarded payload = %q", got) 96 + } 97 + if _, err := client.Dial("tcp", net.JoinHostPort("127.0.0.1", portText)); err == nil { 98 + t.Fatal("forwarded a second route on one jump connection") 99 + } 100 + } 101 + 102 + func startJumpBackend(t *testing.T) net.Listener { 103 + t.Helper() 104 + listener, err := net.Listen("tcp", "127.0.0.1:0") 105 + if err != nil { 106 + t.Fatal(err) 107 + } 108 + t.Cleanup(func() { _ = listener.Close() }) 109 + go func() { 110 + for { 111 + conn, err := listener.Accept() 112 + if err != nil { 113 + return 114 + } 115 + go func() { 116 + defer conn.Close() 117 + _, _ = io.Copy(conn, conn) 118 + }() 119 + } 120 + }() 121 + return listener 122 + } 123 + 124 + func jumpClient(t *testing.T, address string) *gossh.Client { 125 + t.Helper() 126 + client, err := dialJump(t, address, "debug") 127 + if err != nil { 128 + t.Fatal(err) 129 + } 130 + return client 131 + } 132 + 133 + func dialJump(t *testing.T, address, user string) (*gossh.Client, error) { 134 + t.Helper() 135 + privateKey, err := gossh.ParsePrivateKey(testPrivateKey(t)) 136 + if err != nil { 137 + t.Fatal(err) 138 + } 139 + return gossh.Dial("tcp", address, &gossh.ClientConfig{ 140 + User: user, 141 + Auth: []gossh.AuthMethod{gossh.PublicKeys(privateKey)}, 142 + HostKeyCallback: gossh.InsecureIgnoreHostKey(), 143 + Timeout: 5 * time.Second, 144 + }) 145 + } 146 + 147 + func testPrivateKey(t *testing.T) []byte { 148 + t.Helper() 149 + path := t.TempDir() + "/key" 150 + if err := ensureJumpHostKey(path); err != nil { 151 + t.Fatal(err) 152 + } 153 + key, err := os.ReadFile(path) 154 + if err != nil { 155 + t.Fatal(err) 156 + } 157 + return key 158 + }
+7
spindle/mill/mill.go
··· 250 250 m.notifyChange() 251 251 } 252 252 253 + func (m *Mill) hasLiveExecutor(nodeID string) bool { 254 + m.mu.Lock() 255 + defer m.mu.Unlock() 256 + sess := m.sessions[nodeID] 257 + return sess != nil && sess.live(m.cfg.ReconnectGrace) 258 + } 259 + 253 260 func (m *Mill) cancelledLeasesForNode(nodeID string) []*RemoteLease { 254 261 m.mu.Lock() 255 262 var candidates []*RemoteLease
+16 -4
spindle/engines/microvm/README.md
··· 243 243 ### Debug ssh 244 244 245 245 When a workflow fails, spindle can keep its microVM alive for a configured grace 246 - window (`MicroVMPipelines.SSH`) and print an `ssh` invocation so you can poke at 247 - the failed VM interactively. Spindle terminates the ssh connection itself and 248 - bridges a pty into the live guest over the agent's vsock; the guest stays 249 - keyless and never runs an ssh daemon. 246 + window (`MicroVMPipelines.DebugSSH.GracePeriod`) and print an `ssh` invocation 247 + so you can poke at the failed VM interactively. Spindle terminates the ssh 248 + connection itself and bridges a pty into the live guest over the agent's 249 + vsock; the guest stays keyless and never runs an ssh daemon. 250 250 251 251 Access mirrors a git push: the ssh username is the job id, and the offered 252 252 public key is sent to the job's repo knot (`sh.tangled.repo.checkPushAllowed`). 253 253 The session is accepted only if that key is allowed to push to the job's repo. 254 + 255 + In a mill fleet, the printed command can use `ssh -J` through the mill's 256 + restricted jump listener. The inner SSH connection still terminates on the 257 + executor, so the mill only forwards an encrypted TCP stream to a live, 258 + operator-registered executor route. 259 + 260 + `SPINDLE_MILL_MAX_JUMP_CONNECTIONS` limits concurrent outer SSH connections. 261 + 262 + Configure each executor's debug host as its registered executor name. The jump 263 + listener checks that the name has a live authenticated mill session and dials 264 + it on the configured private debug port. The registered name must therefore 265 + resolve on the mill's private network. 254 266 255 267 The shell is deliberately not configurable from either end. It always: 256 268 - runs as the `spindle-workflow` user (the ssh username selects the *job*, not a
+2 -2
spindle/engines/microvm/debug_test.go
··· 5 5 import "testing" 6 6 7 7 func TestDebugSSHCommandUsesJumpHost(t *testing.T) { 8 - got := debugSSHCommand("0.0.0.0:2224", "executor-a.internal", "127.0.0.1", "spindle.example", "job-1") 9 - want := "ssh -tt -J spindle.example -p 2224 job-1@127.0.0.1" 8 + got := debugSSHCommand("0.0.0.0:2224", "executor-a.internal", "executor-a", "spindle.example", "job-1") 9 + want := "ssh -tt -J spindle.example -p 2224 job-1@executor-a" 10 10 if got != want { 11 11 t.Fatalf("debug ssh command = %q, want %q", got, want) 12 12 }