Skip to content

Commit 7875f5b

Browse files
authored
acc: run ssh/connect-serverless-gpu locally (#5878)
## Changes Local test coverage for `databricks ssh connect`, which was disabled on cloud and local in #4838, across the two layers we can exercise without compute: 1. **Bootstrap job submission** — `acceptance/ssh/connect-serverless-gpu` runs against the in-process testserver and asserts the serverless-GPU job the CLI submits (`GPU_1xA10`, serverless env, secret scope, notebook task). This needs new `libs/testserver` stubs: `secrets/list` → `RESOURCE_DOES_NOT_EXIST` for a missing scope, a synthesized `metadata.json` on `ssh-server-bootstrap-*` submit, driver-proxy `metadata`/`logs`, and `workspace/export` → `404` for a missing object. 2. **Real SSH handshake through the proxy** — `client_server_sshd_test.go` stands up a real `sshd` behind `proxy.NewProxyServer` and drives a real SSH client (`x/crypto/ssh`) through `RunClientProxy`. The `test-exp-ssh` job installs `openssh-server` on Linux to support it. ## Why `ssh connect` currently has no automated coverage. These tests exercise the two layers reachable locally: the control-plane setup (the bootstrap job the CLI submits) and the tunnel transport carrying the real SSH protocol. A full local handshake via the connect command can't complete (no compute, a `wss://` driver-proxy dial, real `ssh` spawn), so the acceptance test stops at the connection attempt — failure output goes to `LOG.stderr` (excluded from goldens), and the cloud `Connection successful` path is preserved for when cloud is re-enabled. The handshake test is Linux-only (`//go:build linux`): running `sshd -i` as non-root is reliable there but not on macOS; it skips when `sshd` is absent. The end-to-end path through the real driver proxy (bootstrap notebook → `databricks ssh server` on compute) still needs an integration test and is out of scope. ## Tests - `acceptance/ssh/connect-serverless-gpu` (asserts the submitted bootstrap job payload) - `TestClientServerRealSSHD` in `experimental/ssh/internal/proxy` (real SSH handshake, pubkey auth, and remote command over the proxy tunnel), running on Linux in the `test-exp-ssh` job
1 parent 3713667 commit 7875f5b

10 files changed

Lines changed: 360 additions & 6 deletions

File tree

.github/workflows/push.yml

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -262,6 +262,13 @@ jobs:
262262
with:
263263
cache-key: test-exp-ssh
264264

265+
# The proxy tests exercise a real OpenSSH server, which the runner image
266+
# ships only the client half of. Install it on Linux (the OS the sshd
267+
# test is gated to); other platforms skip that test.
268+
- name: Install OpenSSH server
269+
if: runner.os == 'Linux'
270+
run: sudo apt-get update && sudo apt-get install -y openssh-server
271+
265272
- name: Run tests
266273
run: |
267274
go tool -modfile=tools/task/go.mod task test-exp-ssh
Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +0,0 @@
1-
Connection successful

acceptance/ssh/connect-serverless-gpu/out.test.toml

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.
Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
2+
>>> print_requests.py //api/2.2/jobs/runs/submit
3+
{
4+
"method": "POST",
5+
"path": "/api/2.2/jobs/runs/submit",
6+
"body": {
7+
"environments": [
8+
{
9+
"environment_key": "ssh_tunnel_serverless",
10+
"spec": {
11+
"environment_version": "4"
12+
}
13+
}
14+
],
15+
"run_name": "ssh-server-bootstrap-serverless-gpu-test",
16+
"tasks": [
17+
{
18+
"compute": {
19+
"hardware_accelerator": "GPU_1xA10"
20+
},
21+
"environment_key": "ssh_tunnel_serverless",
22+
"notebook_task": {
23+
"base_parameters": {
24+
"authorizedKeySecretName": "client-public-key",
25+
"maxClients": "10",
26+
"secretScopeName": "[USERNAME]-serverless-gpu-test-ssh-tunnel-keys",
27+
"serverless": "true",
28+
"sessionId": "serverless-gpu-test",
29+
"shutdownDelay": "10m0s",
30+
"version": "[CLI_VERSION]"
31+
},
32+
"notebook_path": "/Workspace/Users/[USERNAME]/.databricks/ssh-tunnel/[CLI_VERSION]/serverless-gpu-test/ssh-server-bootstrap"
33+
},
34+
"task_key": "start_ssh_server",
35+
"timeout_seconds": 86400
36+
}
37+
],
38+
"timeout_seconds": 86400
39+
}
40+
}
Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,22 @@
1+
if [ -z "${CLOUD_ENV:-}" ]; then
2+
# Local runs have no release artifacts and never execute the uploaded
3+
# binaries, so stand in empty archives (dev name omits the version).
4+
mkdir -p releases
5+
: >releases/databricks_cli_linux_amd64.zip
6+
: >releases/databricks_cli_linux_arm64.zip
7+
CLI_RELEASES_DIR="$PWD/releases"
8+
fi
9+
110
errcode $CLI ssh connect --name serverless-gpu-test --accelerator=GPU_1xA10 --releases-dir=$CLI_RELEASES_DIR -- "echo 'Connection successful'" >out.stdout.txt 2>LOG.stderr
211

3-
if ! grep -q "Connection successful" out.stdout.txt; then
4-
run_id=$(cat LOG.stderr | grep -o "Job submitted successfully with run ID: [0-9]*" | grep -o "[0-9]*$")
5-
trace $CLI jobs get-run "$run_id" > LOG.job
12+
if [ -n "${CLOUD_ENV:-}" ]; then
13+
# On cloud the connection runs the remote command; dump the run on failure.
14+
if ! grep -q "Connection successful" out.stdout.txt; then
15+
run_id=$(cat LOG.stderr | grep -o "Job submitted successfully with run ID: [0-9]*" | grep -o "[0-9]*$")
16+
trace $CLI jobs get-run "$run_id" > LOG.job
17+
fi
18+
else
19+
# Locally the handshake can't complete (no compute); just assert the CLI
20+
# submitted the right serverless GPU bootstrap job.
21+
trace print_requests.py //api/2.2/jobs/runs/submit
622
fi

acceptance/ssh/connect-serverless-gpu/test.toml

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,17 @@
1-
Local = false
1+
Local = true
22
Cloud = false
33

44
# Serverless GPU is only available in newer environments
55
RequiresUnityCatalog = true
66

7+
# Assert the serverless GPU bootstrap job the CLI submits on a local run.
8+
RecordRequests = true
9+
10+
# Local runs stage stand-in release archives here; not part of the golden output.
11+
Ignore = [
12+
"releases",
13+
]
14+
715
# Serverless GPU is not available in GCP yet
816
[CloudEnvs]
917
gcp = false
Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,191 @@
1+
//go:build linux
2+
3+
package proxy
4+
5+
import (
6+
"context"
7+
"crypto/ed25519"
8+
"crypto/rand"
9+
"encoding/pem"
10+
"fmt"
11+
"io"
12+
"net"
13+
"net/http/httptest"
14+
"os"
15+
"os/exec"
16+
"os/user"
17+
"path/filepath"
18+
"testing"
19+
"time"
20+
21+
"github.com/databricks/cli/libs/cmdio"
22+
"github.com/gorilla/websocket"
23+
"github.com/stretchr/testify/assert"
24+
"github.com/stretchr/testify/require"
25+
"golang.org/x/crypto/ssh"
26+
)
27+
28+
// findSSHD returns the path to the OpenSSH server binary, or "" if it isn't installed.
29+
func findSSHD() string {
30+
if p, err := exec.LookPath("sshd"); err == nil {
31+
return p
32+
}
33+
// LookPath misses sshd because it lives in sbin, which is usually not on a
34+
// non-root user's PATH; probe the conventional locations directly.
35+
for _, p := range []string{"/usr/sbin/sshd", "/usr/local/sbin/sshd", "/sbin/sshd"} {
36+
if _, err := os.Stat(p); err == nil {
37+
return p
38+
}
39+
}
40+
return ""
41+
}
42+
43+
// pipeConn adapts a pair of pipes into a net.Conn so an in-process SSH client can speak the SSH
44+
// protocol over the proxy transport (RunClientProxy reads/writes the other ends of these pipes).
45+
type pipeConn struct {
46+
r *io.PipeReader
47+
w *io.PipeWriter
48+
}
49+
50+
func (c pipeConn) Read(p []byte) (int, error) { return c.r.Read(p) }
51+
func (c pipeConn) Write(p []byte) (int, error) { return c.w.Write(p) }
52+
53+
func (c pipeConn) Close() error {
54+
_ = c.r.Close()
55+
return c.w.Close()
56+
}
57+
58+
func (pipeConn) LocalAddr() net.Addr { return pipeAddr{} }
59+
func (pipeConn) RemoteAddr() net.Addr { return pipeAddr{} }
60+
func (pipeConn) SetDeadline(_ time.Time) error { return nil }
61+
func (pipeConn) SetReadDeadline(_ time.Time) error { return nil }
62+
func (pipeConn) SetWriteDeadline(_ time.Time) error { return nil }
63+
64+
type pipeAddr struct{}
65+
66+
func (pipeAddr) Network() string { return "pipe" }
67+
func (pipeAddr) String() string { return "pipe" }
68+
69+
func writeOpenSSHPrivateKey(t *testing.T, path string, key ed25519.PrivateKey) {
70+
block, err := ssh.MarshalPrivateKey(key, "")
71+
require.NoError(t, err)
72+
require.NoError(t, os.WriteFile(path, pem.EncodeToMemory(block), 0o600))
73+
}
74+
75+
// writeSSHDConfig writes a minimal sshd_config that lets a real sshd run as a non-root user in
76+
// inetd mode (-i) and authenticate the current user by public key. StrictModes/UsePAM are off so
77+
// sshd doesn't reject the temp-dir key files or try to switch users.
78+
func writeSSHDConfig(t *testing.T, dir, hostKeyPath, authKeysPath string) string {
79+
cfg := fmt.Sprintf(
80+
"HostKey %s\n"+
81+
"AuthorizedKeysFile %s\n"+
82+
"PidFile none\n"+
83+
"StrictModes no\n"+
84+
"UsePAM no\n"+
85+
"PubkeyAuthentication yes\n"+
86+
"PasswordAuthentication no\n"+
87+
"LogLevel ERROR\n",
88+
hostKeyPath, authKeysPath)
89+
path := filepath.Join(dir, "sshd_config")
90+
require.NoError(t, os.WriteFile(path, []byte(cfg), 0o600))
91+
return path
92+
}
93+
94+
// TestClientServerRealSSHD runs a real OpenSSH server (sshd) behind the websocket proxy and a real
95+
// SSH client through RunClientProxy, exercising the actual SSH handshake and a remote command
96+
// end-to-end without any Databricks compute. It is the local stand-in for the ssh-connectivity
97+
// coverage that otherwise requires a cluster: the tunnel must carry the SSH protocol faithfully in
98+
// both directions. Sibling tests in client_server_test.go cover the transport with a `cat` echo
99+
// server; this one adds a genuine sshd so the handshake, auth, and channel exec are validated too.
100+
//
101+
// The test is Linux-only: running sshd -i as a non-root user with a throwaway config is reliable
102+
// there (CI installs openssh-server), but not on macOS. It still skips gracefully when sshd is
103+
// absent so the general `test` job, which doesn't install it, doesn't fail.
104+
func TestClientServerRealSSHD(t *testing.T) {
105+
sshdPath := findSSHD()
106+
if sshdPath == "" {
107+
t.Skip("sshd (openssh-server) not installed; skipping real SSH handshake test")
108+
}
109+
110+
currentUser, err := user.Current()
111+
require.NoError(t, err)
112+
113+
dir := t.TempDir()
114+
115+
// Host key: identifies the server; the client pins it via FixedHostKey.
116+
_, hostPriv, err := ed25519.GenerateKey(rand.Reader)
117+
require.NoError(t, err)
118+
hostKeyPath := filepath.Join(dir, "host_key")
119+
writeOpenSSHPrivateKey(t, hostKeyPath, hostPriv)
120+
hostSigner, err := ssh.NewSignerFromSigner(hostPriv)
121+
require.NoError(t, err)
122+
123+
// Client key: authorized on the server via authorized_keys, presented by the SSH client to log in.
124+
_, clientPriv, err := ed25519.GenerateKey(rand.Reader)
125+
require.NoError(t, err)
126+
clientSigner, err := ssh.NewSignerFromSigner(clientPriv)
127+
require.NoError(t, err)
128+
authKeysPath := filepath.Join(dir, "authorized_keys")
129+
require.NoError(t, os.WriteFile(authKeysPath, ssh.MarshalAuthorizedKey(clientSigner.PublicKey()), 0o600))
130+
131+
sshdConfigPath := writeSSHDConfig(t, dir, hostKeyPath, authKeysPath)
132+
133+
ctx := cmdio.MockDiscard(t.Context())
134+
135+
// Server proxy: each websocket connection spawns `sshd -i`, which speaks SSH over its stdio.
136+
connections := NewConnectionsManager(2, time.Hour)
137+
server := httptest.NewServer(NewProxyServer(ctx, connections, func(ctx context.Context) *exec.Cmd {
138+
return exec.CommandContext(ctx, sshdPath, "-f", sshdConfigPath, "-i", "-e")
139+
}))
140+
defer server.Close()
141+
142+
wsURL := "ws" + server.URL[4:]
143+
createConn := func(ctx context.Context, connID string) (*websocket.Conn, error) {
144+
conn, _, err := websocket.DefaultDialer.Dial(fmt.Sprintf("%s?id=%s", wsURL, connID), nil) // nolint:bodyclose
145+
return conn, err
146+
}
147+
148+
// Wire the SSH client's transport through RunClientProxy:
149+
// ssh client --writes--> clientToProxy --> proxy sends over ws --> sshd stdin
150+
// ssh client <--reads--- proxyToClient <-- proxy recvs from ws <-- sshd stdout
151+
clientToProxyR, clientToProxyW := io.Pipe()
152+
proxyToClientR, proxyToClientW := io.Pipe()
153+
154+
proxyDone := make(chan error, 1)
155+
go func() {
156+
requestHandoverTick := func() <-chan time.Time { return time.After(time.Hour) }
157+
proxyDone <- RunClientProxy(ctx, clientToProxyR, proxyToClientW, requestHandoverTick, createConn)
158+
}()
159+
160+
conn := pipeConn{r: proxyToClientR, w: clientToProxyW}
161+
sshConfig := &ssh.ClientConfig{
162+
User: currentUser.Username,
163+
Auth: []ssh.AuthMethod{ssh.PublicKeys(clientSigner)},
164+
HostKeyCallback: ssh.FixedHostKey(hostSigner.PublicKey()),
165+
Timeout: 30 * time.Second,
166+
}
167+
168+
clientConn, chans, reqs, err := ssh.NewClientConn(conn, "pipe", sshConfig)
169+
require.NoError(t, err, "SSH handshake failed")
170+
sshClient := ssh.NewClient(clientConn, chans, reqs)
171+
defer sshClient.Close()
172+
173+
session, err := sshClient.NewSession()
174+
require.NoError(t, err)
175+
defer session.Close()
176+
177+
out, err := session.Output("echo 'Connection successful'")
178+
require.NoError(t, err)
179+
assert.Equal(t, "Connection successful\n", string(out))
180+
181+
// Tear down the client side and confirm the proxy shuts down cleanly rather than hanging.
182+
require.NoError(t, sshClient.Close())
183+
_ = conn.Close()
184+
185+
select {
186+
case err := <-proxyDone:
187+
require.NoError(t, err)
188+
case <-time.After(10 * time.Second):
189+
t.Fatal("client proxy did not shut down after the SSH session ended")
190+
}
191+
}

libs/testserver/handlers.go

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,15 @@ func AddDefaultHandlers(server *Server) {
105105
path := req.URL.Query().Get("path")
106106
data := req.Workspace.WorkspaceExport(path)
107107

108+
// A missing object returns 404, matching the real API; returning the nil
109+
// body otherwise trips the response normalizer.
110+
if data == nil {
111+
return Response{
112+
StatusCode: 404,
113+
Body: map[string]string{"message": fmt.Sprintf("Path (%s) doesn't exist.", path)},
114+
}
115+
}
116+
108117
// The filer reads the raw object body via ?direct_download=true, while
109118
// the SDK's Workspace.Export (used by `databricks workspace export`)
110119
// requests JSON and expects the base64-encoded content field.
@@ -730,6 +739,20 @@ func AddDefaultHandlers(server *Server) {
730739
return req.Workspace.SecretsGet(req)
731740
})
732741

742+
server.Handle("GET", "/api/2.0/secrets/list", func(req Request) any {
743+
return req.Workspace.SecretsList(req)
744+
})
745+
746+
// SSH tunnel server behind the driver proxy: /metadata returns the remote
747+
// login user, /logs is a best-effort error tail fetched on failure.
748+
server.Handle("GET", "/driver-proxy-api/o/{workspace_id}/{cluster_id}/{port}/metadata", func(req Request) any {
749+
return Response{Body: sshTunnelRemoteUser}
750+
})
751+
752+
server.Handle("GET", "/driver-proxy-api/o/{workspace_id}/{cluster_id}/{port}/logs", func(req Request) any {
753+
return Response{Body: ""}
754+
})
755+
733756
// Secrets ACLs:
734757
server.Handle("GET", "/api/2.0/secrets/acls/get", func(req Request) any {
735758
return req.Workspace.SecretsAclsGet(req)

libs/testserver/jobs.go

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515

1616
"github.com/databricks/databricks-sdk-go/service/compute"
1717
"github.com/databricks/databricks-sdk-go/service/jobs"
18+
"github.com/databricks/databricks-sdk-go/service/workspace"
1819
)
1920

2021
const missingJobGitProviderMessage = "git_source.git_provider must be one of: github,gitlab,bitbucketcloud,gitlabenterpriseedition,bitbucketserver,azuredevopsservices,githubenterprise,awscodecommit"
@@ -519,9 +520,45 @@ func (s *FakeWorkspace) JobsSubmit(req Request) Response {
519520
Tasks: tasks,
520521
}
521522

523+
// No tunnel server runs locally, so synthesize the metadata.json it would
524+
// publish; `ssh connect` polls for it before connecting.
525+
if strings.HasPrefix(runName, sshTunnelBootstrapRunPrefix) {
526+
s.writeSSHTunnelMetadata(request)
527+
}
528+
522529
return Response{Body: jobs.SubmitRunResponse{RunId: runId}}
523530
}
524531

532+
const (
533+
sshTunnelBootstrapRunPrefix = "ssh-server-bootstrap-"
534+
sshTunnelBootstrapNotebook = "ssh-server-bootstrap"
535+
sshTunnelServerPort = 7772
536+
sshTunnelClusterID = "1234-567890-serverless"
537+
sshTunnelRemoteUser = "spark"
538+
)
539+
540+
// writeSSHTunnelMetadata publishes the metadata.json a real tunnel server would
541+
// write next to the bootstrap notebook. Callers must hold the workspace lock.
542+
func (s *FakeWorkspace) writeSSHTunnelMetadata(request jobs.SubmitRun) {
543+
for _, t := range request.Tasks {
544+
if t.NotebookTask == nil {
545+
continue
546+
}
547+
metadataPath := strings.TrimSuffix(t.NotebookTask.NotebookPath, sshTunnelBootstrapNotebook) + "metadata.json"
548+
metadata, err := json.Marshal(map[string]any{
549+
"port": sshTunnelServerPort,
550+
"cluster_id": sshTunnelClusterID,
551+
})
552+
if err != nil {
553+
continue
554+
}
555+
s.files[metadataPath] = FileEntry{
556+
Info: workspace.ObjectInfo{ObjectType: "FILE", Path: metadataPath},
557+
Data: metadata,
558+
}
559+
}
560+
}
561+
525562
// executePythonWheelTask runs a python wheel task locally using uv.
526563
// For tasks using existing_cluster_id, the venv is cached per cluster to match
527564
// cloud behavior where libraries are cached on running clusters.

0 commit comments

Comments
 (0)