feat(backend): add SSH executor runtime
Run agents on any Linux box reachable over SSH: own VPS, homelab, EC2, shared dev box. One SSH connection per (host, user, identity) shared across all sessions on that host; each session gets its own remote agentctl process, remote port, and SSH-tunneled local port. Mirrors the local ~/.kandev/tasks/<task-dir>/<repo>/ layout on the remote so multi-session tasks share a worktree and per-session runtime state lives under the task dir. Implements ExecutorBackend + RemoteSessionResumer + RemoteStatusProvider + Closeable. Host-key trust is explicit: the test endpoint observes a fingerprint without pinning; the user trusts it in settings; subsequent connects reject any mismatch. Auth via ssh-agent or IdentityFile only (passphrase-protected keys must be loaded into ssh-agent). Parses ~/.ssh/config for HostName/Port/User/IdentityFile/ProxyJump inheritance. agentctl is content-hash-cached on the remote (~/.kandev/bin/agentctl plus a .sha256 sidecar) and uploaded only when local and remote differ. Linux/amd64 only in v1, matching Docker's current state; arm64 is called out as a cross-cutting follow-up. See docs/specs/ssh-executor/spec.md.
This commit is contained in:
José Almeida
@@ -76,6 +76,11 @@ func provideLifecycleManager(
|
||||
executorRegistry.Register(spritesExec)
|
||||
log.Info("Sprites runtime registered")
|
||||
|
||||
// Register SSH runtime (run an agent on any Linux box reachable over SSH).
|
||||
sshExec := lifecycle.NewSSHExecutor(secretStore, agentRegistry, agentctlResolver, log)
|
||||
executorRegistry.Register(sshExec)
|
||||
log.Info("SSH runtime registered")
|
||||
|
||||
credsMgr := credentials.NewManager(log)
|
||||
if secretStore != nil {
|
||||
credsMgr.AddProvider(secrets.NewSecretStoreProvider(secretStore))
|
||||
@@ -111,6 +116,7 @@ func provideLifecycleManager(
|
||||
preparerRegistry.Register(models.ExecutorTypeMockRemote, localPreparer)
|
||||
preparerRegistry.Register(models.ExecutorTypeLocalDocker, lifecycle.NewDockerPreparer(log))
|
||||
preparerRegistry.Register(models.ExecutorTypeSprites, lifecycle.NewSpritesPreparer(log))
|
||||
preparerRegistry.Register(models.ExecutorTypeSSH, lifecycle.NewSSHPreparer(log))
|
||||
lifecycleMgr.SetPreparerRegistry(preparerRegistry)
|
||||
lifecycleMgr.SetSecretStore(secretStore)
|
||||
// Wire the agent_profiles reader so the launch-prep skill deploy hook
|
||||
|
||||
+8
-5
@@ -15,9 +15,11 @@ require (
|
||||
github.com/gorilla/websocket v1.5.3
|
||||
github.com/jackc/pgx/v5 v5.9.0
|
||||
github.com/jmoiron/sqlx v1.4.0
|
||||
github.com/kevinburke/ssh_config v1.6.0
|
||||
github.com/mark3labs/mcp-go v0.43.2
|
||||
github.com/mattn/go-sqlite3 v1.14.33
|
||||
github.com/nats-io/nats.go v1.31.0
|
||||
github.com/pkg/sftp v1.13.10
|
||||
github.com/spf13/viper v1.16.0
|
||||
github.com/stretchr/testify v1.11.1
|
||||
github.com/superfly/sprites-go v0.0.0-20260206213632-8176adff485b
|
||||
@@ -27,8 +29,9 @@ require (
|
||||
go.opentelemetry.io/otel/sdk v1.43.0
|
||||
go.opentelemetry.io/otel/trace v1.43.0
|
||||
go.uber.org/zap v1.26.0
|
||||
golang.org/x/crypto v0.51.0
|
||||
golang.org/x/sync v0.20.0
|
||||
golang.org/x/sys v0.42.0
|
||||
golang.org/x/sys v0.44.0
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1
|
||||
gopkg.in/yaml.v3 v3.0.1
|
||||
)
|
||||
@@ -67,6 +70,7 @@ require (
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/klauspost/compress v1.17.0 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.4 // indirect
|
||||
github.com/kr/fs v0.1.0 // indirect
|
||||
github.com/leodido/go-urn v1.2.4 // indirect
|
||||
github.com/magiconair/properties v1.8.7 // indirect
|
||||
github.com/mailru/easyjson v0.7.7 // indirect
|
||||
@@ -101,10 +105,9 @@ require (
|
||||
go.opentelemetry.io/proto/otlp v1.10.0 // indirect
|
||||
go.uber.org/multierr v1.10.0 // indirect
|
||||
golang.org/x/arch v0.3.0 // indirect
|
||||
golang.org/x/crypto v0.49.0 // indirect
|
||||
golang.org/x/net v0.52.0 // indirect
|
||||
golang.org/x/term v0.41.0 // indirect
|
||||
golang.org/x/text v0.35.0 // indirect
|
||||
golang.org/x/net v0.53.0 // indirect
|
||||
golang.org/x/term v0.43.0 // indirect
|
||||
golang.org/x/text v0.37.0 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect
|
||||
google.golang.org/grpc v1.80.0 // indirect
|
||||
|
||||
+15
-10
@@ -221,12 +221,15 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
|
||||
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
|
||||
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
|
||||
github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
|
||||
github.com/kevinburke/ssh_config v1.6.0 h1:J1FBfmuVosPHf5GRdltRLhPJtJpTlMdKTBjRgTaQBFY=
|
||||
github.com/kevinburke/ssh_config v1.6.0/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
|
||||
github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
|
||||
github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
|
||||
github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
|
||||
github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
|
||||
github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=
|
||||
github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
@@ -282,6 +285,8 @@ github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNc
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
|
||||
github.com/pkg/sftp v1.13.10 h1:+5FbKNTe5Z9aspU88DPIKJ9z2KZoaGCu6Sr6kKR/5mU=
|
||||
github.com/pkg/sftp v1.13.10/go.mod h1:bJ1a7uDhrX/4OII+agvy28lzRvQrmIQuaHrcI1HbeGA=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
|
||||
@@ -376,8 +381,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
|
||||
golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
|
||||
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
|
||||
golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI=
|
||||
golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8=
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
|
||||
@@ -443,8 +448,8 @@ golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v
|
||||
golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
|
||||
golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
|
||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
|
||||
golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
@@ -507,11 +512,11 @@ golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBc
|
||||
golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
|
||||
golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU=
|
||||
golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A=
|
||||
golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
|
||||
golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
|
||||
golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
@@ -520,8 +525,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
|
||||
golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
|
||||
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
|
||||
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
|
||||
golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
|
||||
@@ -18,6 +18,7 @@ const (
|
||||
NameLocal Name = "local"
|
||||
NameRemoteDocker = agentruntime.RuntimeRemoteDocker
|
||||
NameSprites = agentruntime.RuntimeSprites
|
||||
NameSSH = agentruntime.RuntimeSSH
|
||||
)
|
||||
|
||||
// ExecutorTypeToBackend maps an ExecutorType to its corresponding executor Name.
|
||||
@@ -33,6 +34,8 @@ func ExecutorTypeToBackend(execType models.ExecutorType) Name {
|
||||
return NameRemoteDocker
|
||||
case models.ExecutorTypeSprites:
|
||||
return NameSprites
|
||||
case models.ExecutorTypeSSH:
|
||||
return NameSSH
|
||||
case models.ExecutorTypeMockRemote:
|
||||
return NameStandalone
|
||||
default:
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
// SSHPreparer validates prerequisites for SSH-based execution. The heavy
|
||||
// lifting (connect, agentctl upload, task dir provisioning, agentctl launch)
|
||||
// happens in SSHExecutor.CreateInstance — this preparer just emits a single
|
||||
// validation step so the launch UI has consistent progress feedback.
|
||||
type SSHPreparer struct {
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
// NewSSHPreparer creates a new SSHPreparer.
|
||||
func NewSSHPreparer(log *logger.Logger) *SSHPreparer {
|
||||
return &SSHPreparer{logger: log.WithFields(zap.String("component", "ssh-preparer"))}
|
||||
}
|
||||
|
||||
// Name implements EnvironmentPreparer.
|
||||
func (p *SSHPreparer) Name() string { return "ssh" }
|
||||
|
||||
// Prepare implements EnvironmentPreparer. SSH executor configuration is
|
||||
// validated at launch time (target / fingerprint / arch are checked inside
|
||||
// CreateInstance). This step is intentionally minimal so the UI still sees
|
||||
// the preparer running for every launch — preserves a uniform progress shape
|
||||
// across executor types.
|
||||
func (p *SSHPreparer) Prepare(_ context.Context, req *EnvPrepareRequest, onProgress PrepareProgressCallback) (*EnvPrepareResult, error) {
|
||||
p.logger.Debug("preparing ssh environment",
|
||||
zap.String("task", req.TaskID),
|
||||
zap.String("session", req.SessionID))
|
||||
|
||||
started := time.Now()
|
||||
step := beginStep("Validate SSH executor configuration")
|
||||
reportProgress(onProgress, step, 0, 1)
|
||||
completeStepSuccess(&step)
|
||||
reportProgress(onProgress, step, 0, 1)
|
||||
return &EnvPrepareResult{
|
||||
Success: true,
|
||||
Steps: []PrepareStep{step},
|
||||
WorkspacePath: req.WorkspacePath,
|
||||
Duration: time.Since(started),
|
||||
WorktreeBranch: nonWorktreeTaskBranch(req),
|
||||
}, nil
|
||||
}
|
||||
@@ -82,6 +82,22 @@ const (
|
||||
|
||||
// Office metadata keys
|
||||
MetadataKeySkillManifestJSON = "skill_manifest_json"
|
||||
|
||||
// SSH runtime metadata keys (per-session, except SSHWorkdirRoot which is per-profile).
|
||||
MetadataKeySSHHost = "ssh_host"
|
||||
MetadataKeySSHPort = "ssh_port"
|
||||
MetadataKeySSHUser = "ssh_user"
|
||||
MetadataKeySSHHostFingerprint = "ssh_host_fingerprint"
|
||||
MetadataKeySSHRemoteTaskDir = "ssh_remote_task_dir"
|
||||
MetadataKeySSHRemoteSessionDir = "ssh_remote_session_dir"
|
||||
MetadataKeySSHRemoteAgentctlPort = "ssh_remote_agentctl_port"
|
||||
MetadataKeySSHRemoteAgentctlPID = "ssh_remote_agentctl_pid"
|
||||
MetadataKeySSHLocalForwardPort = "ssh_local_forward_port"
|
||||
MetadataKeySSHRemoteAgentctlURL = "ssh_remote_agentctl_url"
|
||||
MetadataKeySSHWorkdirRoot = "ssh_workdir_root"
|
||||
MetadataKeySSHProxyJump = "ssh_proxy_jump"
|
||||
MetadataKeySSHIdentitySource = "ssh_identity_source"
|
||||
MetadataKeySSHIdentityFile = "ssh_identity_file"
|
||||
)
|
||||
|
||||
// persistentMetadataKeys lists metadata keys carried forward from a previous
|
||||
@@ -95,6 +111,22 @@ var persistentMetadataKeys = map[string]bool{
|
||||
MetadataKeySpriteCreatedAt: true,
|
||||
MetadataKeyLocalPort: true,
|
||||
|
||||
// SSH runtime
|
||||
MetadataKeySSHHost: true,
|
||||
MetadataKeySSHPort: true,
|
||||
MetadataKeySSHUser: true,
|
||||
MetadataKeySSHHostFingerprint: true,
|
||||
MetadataKeySSHRemoteTaskDir: true,
|
||||
MetadataKeySSHRemoteSessionDir: true,
|
||||
MetadataKeySSHRemoteAgentctlPort: true,
|
||||
MetadataKeySSHRemoteAgentctlPID: true,
|
||||
MetadataKeySSHLocalForwardPort: true,
|
||||
MetadataKeySSHRemoteAgentctlURL: true,
|
||||
MetadataKeySSHWorkdirRoot: true,
|
||||
MetadataKeySSHProxyJump: true,
|
||||
MetadataKeySSHIdentitySource: true,
|
||||
MetadataKeySSHIdentityFile: true,
|
||||
|
||||
// Executor type marker
|
||||
MetadataKeyIsRemote: true,
|
||||
|
||||
|
||||
@@ -0,0 +1,493 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/executor"
|
||||
agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
|
||||
"github.com/kandev/kandev/internal/agentctl/server/process"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/secrets"
|
||||
)
|
||||
|
||||
const (
|
||||
sshAgentctlHealthTimeout = 20 * time.Second
|
||||
|
||||
sshStatusRunning = "running"
|
||||
sshStatusUnknown = "unknown"
|
||||
sshStatusDisconnected = "disconnected"
|
||||
sshStatusAgentctlDown = "agentctl-down"
|
||||
)
|
||||
|
||||
// sshSessionState tracks per-session resources we need to clean up later:
|
||||
// the SSH target (so we can release the pooled connection) and the local
|
||||
// port forwarder.
|
||||
type sshSessionState struct {
|
||||
target *SSHTarget
|
||||
forwarder *SSHPortForwarder
|
||||
pid int
|
||||
remoteDir string
|
||||
}
|
||||
|
||||
// SSHExecutor implements ExecutorBackend for SSH-reachable Linux hosts.
|
||||
//
|
||||
// One SSH connection is shared per (host, user, identity) tuple across all
|
||||
// sessions on that host. Each session gets its own agentctl process + remote
|
||||
// port + local-forwarded port. See docs/specs/ssh-executor/spec.md for the
|
||||
// full design.
|
||||
type SSHExecutor struct {
|
||||
pool *SSHConnPool
|
||||
agentctlResolver *AgentctlResolver
|
||||
secretStore secrets.SecretStore
|
||||
agentList RemoteAuthAgentLister
|
||||
logger *logger.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
sessions map[string]*sshSessionState // keyed by ExecutorInstance.InstanceID
|
||||
}
|
||||
|
||||
// NewSSHExecutor wires up an SSH executor with shared infrastructure.
|
||||
// agentList is optional; pass nil if no agent install scripts are needed.
|
||||
func NewSSHExecutor(
|
||||
secretStore secrets.SecretStore,
|
||||
agentList RemoteAuthAgentLister,
|
||||
resolver *AgentctlResolver,
|
||||
log *logger.Logger,
|
||||
) *SSHExecutor {
|
||||
return &SSHExecutor{
|
||||
pool: NewSSHConnPool(log),
|
||||
agentctlResolver: resolver,
|
||||
secretStore: secretStore,
|
||||
agentList: agentList,
|
||||
logger: log.WithFields(zap.String("runtime", "ssh")),
|
||||
sessions: make(map[string]*sshSessionState),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *SSHExecutor) Name() executor.Name { return executor.NameSSH }
|
||||
|
||||
func (r *SSHExecutor) HealthCheck(_ context.Context) error {
|
||||
// SSH targets are configured per-executor — the runtime itself is always
|
||||
// available. Per-host reachability is verified by the test-connection
|
||||
// endpoint and surfaced in the executor status panel.
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close terminates pooled SSH connections at shutdown.
|
||||
func (r *SSHExecutor) Close() error {
|
||||
r.pool.CloseAll()
|
||||
return nil
|
||||
}
|
||||
|
||||
// targetFromMetadata builds an SSHTarget from the executor metadata
|
||||
// propagated via buildLaunchMetadata (req.ExecutorConfig keys merged in).
|
||||
func (r *SSHExecutor) targetFromMetadata(md map[string]interface{}) (*SSHTarget, error) {
|
||||
host := getMetadataString(md, MetadataKeySSHHost)
|
||||
hostAlias := getMetadataString(md, "ssh_host_alias")
|
||||
if host == "" && hostAlias == "" {
|
||||
return nil, errors.New("ssh executor: host (or host_alias) is required in executor config")
|
||||
}
|
||||
port := 0
|
||||
if p := getMetadataString(md, MetadataKeySSHPort); p != "" {
|
||||
if n, err := strconv.Atoi(p); err == nil {
|
||||
port = n
|
||||
}
|
||||
}
|
||||
identitySource := SSHIdentitySource(getMetadataString(md, MetadataKeySSHIdentitySource))
|
||||
identityFile := getMetadataString(md, MetadataKeySSHIdentityFile)
|
||||
cfg := SSHConnConfig{
|
||||
HostAlias: hostAlias,
|
||||
Host: host,
|
||||
Port: port,
|
||||
User: getMetadataString(md, MetadataKeySSHUser),
|
||||
IdentitySource: identitySource,
|
||||
IdentityFile: identityFile,
|
||||
ProxyJump: getMetadataString(md, MetadataKeySSHProxyJump),
|
||||
PinnedFingerprint: getMetadataString(md, MetadataKeySSHHostFingerprint),
|
||||
}
|
||||
if cfg.PinnedFingerprint == "" {
|
||||
return nil, errors.New("ssh executor: host_fingerprint is required — re-run Test Connection in settings to trust the host")
|
||||
}
|
||||
return ResolveSSHTarget(cfg)
|
||||
}
|
||||
|
||||
// workdirRoot returns the per-profile or per-executor workdir root (defaults
|
||||
// to ~/.kandev). Per-profile wins over per-executor.
|
||||
func (r *SSHExecutor) workdirRoot(md map[string]interface{}) string {
|
||||
if w := getMetadataString(md, MetadataKeySSHWorkdirRoot); w != "" {
|
||||
return w
|
||||
}
|
||||
return sshDefaultWorkdir
|
||||
}
|
||||
|
||||
// CreateInstance opens (or reuses) an SSH connection to the configured host,
|
||||
// ensures agentctl is uploaded, provisions the per-task remote dir, launches
|
||||
// a per-session agentctl process, and sets up a local port forward so the
|
||||
// kandev backend can speak to it as if it were on localhost.
|
||||
func (r *SSHExecutor) CreateInstance(ctx context.Context, req *ExecutorCreateRequest) (*ExecutorInstance, error) {
|
||||
target, err := r.targetFromMetadata(req.Metadata)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
client, err := r.pool.Get(ctx, target)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: connect to %s@%s: %w", target.User, target.Host, err)
|
||||
}
|
||||
released := false
|
||||
defer func() {
|
||||
if !released {
|
||||
r.pool.Release(target)
|
||||
}
|
||||
}()
|
||||
r.report(req.OnProgress, "Connecting to SSH host", PrepareStepCompleted, "")
|
||||
|
||||
agentctlBin, err := r.prepareRemoteHost(ctx, client, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
workdir := r.workdirRoot(req.Metadata)
|
||||
taskDir, sessionDir, err := r.prepareRemoteDirs(ctx, client, workdir, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.maybeUploadCredentials(ctx, client, req)
|
||||
|
||||
port, pid, fwd, err := r.startAndForwardAgentctl(ctx, client, agentctlBin, taskDir, sessionDir, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
r.sessions[req.InstanceID] = &sshSessionState{
|
||||
target: target,
|
||||
forwarder: fwd,
|
||||
pid: pid,
|
||||
remoteDir: sessionDir,
|
||||
}
|
||||
r.mu.Unlock()
|
||||
released = true // ownership transferred to session state; released on StopInstance
|
||||
|
||||
return r.buildInstance(req, target, fwd, taskDir, sessionDir, port, pid, workdir), nil
|
||||
}
|
||||
|
||||
// prepareRemoteHost runs the steps that are independent of any particular
|
||||
// task: detect remote OS/arch and ensure the agentctl binary is on the host.
|
||||
// Returns the resolved remote agentctl path.
|
||||
func (r *SSHExecutor) prepareRemoteHost(ctx context.Context, client *ssh.Client, req *ExecutorCreateRequest) (string, error) {
|
||||
info, err := detectRemoteInfo(ctx, client)
|
||||
if err != nil {
|
||||
r.report(req.OnProgress, "Detecting remote OS", PrepareStepFailed, err.Error())
|
||||
return "", fmt.Errorf("ssh: detect remote: %w", err)
|
||||
}
|
||||
if err := requireSupportedArch(info.Arch); err != nil {
|
||||
r.report(req.OnProgress, "Detecting remote OS", PrepareStepFailed, err.Error())
|
||||
return "", err
|
||||
}
|
||||
r.report(req.OnProgress, "Detecting remote OS", PrepareStepCompleted, info.UnameAll)
|
||||
|
||||
agentctlBin, err := ensureAgentctlOnHost(ctx, client, r.agentctlResolver, r.logger)
|
||||
if err != nil {
|
||||
r.report(req.OnProgress, "Uploading agent controller", PrepareStepFailed, err.Error())
|
||||
return "", err
|
||||
}
|
||||
r.report(req.OnProgress, "Uploading agent controller", PrepareStepCompleted, "")
|
||||
return agentctlBin, nil
|
||||
}
|
||||
|
||||
// prepareRemoteDirs makes <workdir>/tasks/<taskDir> and <taskDir>/.kandev/sessions/<sid>.
|
||||
func (r *SSHExecutor) prepareRemoteDirs(ctx context.Context, client *ssh.Client, workdir string, req *ExecutorCreateRequest) (string, string, error) {
|
||||
taskDir, err := ensureRemoteTaskDir(ctx, client, workdir, sshTaskDirName(req))
|
||||
if err != nil {
|
||||
r.report(req.OnProgress, "Preparing task directory", PrepareStepFailed, err.Error())
|
||||
return "", "", err
|
||||
}
|
||||
sessionDir, err := ensureRemoteSessionDir(ctx, client, taskDir, req.SessionID)
|
||||
if err != nil {
|
||||
r.report(req.OnProgress, "Preparing task directory", PrepareStepFailed, err.Error())
|
||||
return "", "", err
|
||||
}
|
||||
r.report(req.OnProgress, "Preparing task directory", PrepareStepCompleted, taskDir)
|
||||
return taskDir, sessionDir, nil
|
||||
}
|
||||
|
||||
// startAndForwardAgentctl spawns the per-session agentctl on the remote and
|
||||
// stands up the SSH local port forward. Cleans up both on failure.
|
||||
func (r *SSHExecutor) startAndForwardAgentctl(
|
||||
ctx context.Context, client *ssh.Client, agentctlBin, taskDir, sessionDir string, req *ExecutorCreateRequest,
|
||||
) (int, int, *SSHPortForwarder, error) {
|
||||
port, pid, err := startRemoteAgentctl(ctx, client, agentctlBin, taskDir, sessionDir, r.logger)
|
||||
if err != nil {
|
||||
r.report(req.OnProgress, "Starting agent controller", PrepareStepFailed, err.Error())
|
||||
return 0, 0, nil, err
|
||||
}
|
||||
r.report(req.OnProgress, "Starting agent controller", PrepareStepCompleted,
|
||||
fmt.Sprintf("pid=%d port=%d", pid, port))
|
||||
|
||||
fwd, err := StartPortForward(client, port, r.logger)
|
||||
if err != nil {
|
||||
_ = stopRemoteAgentctl(ctx, client, sessionDir, pid)
|
||||
return 0, 0, nil, fmt.Errorf("ssh: port forward: %w", err)
|
||||
}
|
||||
if err := waitAgentctlHealthy(ctx, fwd.LocalPort(), sshAgentctlHealthTimeout); err != nil {
|
||||
_ = fwd.Close()
|
||||
_ = stopRemoteAgentctl(ctx, client, sessionDir, pid)
|
||||
return 0, 0, nil, fmt.Errorf("ssh: agentctl health: %w", err)
|
||||
}
|
||||
r.report(req.OnProgress, "Connecting to agent controller", PrepareStepCompleted,
|
||||
fmt.Sprintf("local:%d -> remote:%d", fwd.LocalPort(), port))
|
||||
return port, pid, fwd, nil
|
||||
}
|
||||
|
||||
func (r *SSHExecutor) buildInstance(
|
||||
req *ExecutorCreateRequest,
|
||||
target *SSHTarget,
|
||||
fwd *SSHPortForwarder,
|
||||
taskDir, sessionDir string,
|
||||
port, pid int,
|
||||
workdir string,
|
||||
) *ExecutorInstance {
|
||||
return &ExecutorInstance{
|
||||
InstanceID: req.InstanceID,
|
||||
TaskID: req.TaskID,
|
||||
SessionID: req.SessionID,
|
||||
RuntimeName: r.Name(),
|
||||
Client: agentctl.NewClient("127.0.0.1", fwd.LocalPort(), r.logger,
|
||||
agentctl.WithExecutionID(req.InstanceID),
|
||||
agentctl.WithSessionID(req.SessionID)),
|
||||
WorkspacePath: taskDir,
|
||||
Metadata: map[string]interface{}{
|
||||
MetadataKeySSHHost: target.Host,
|
||||
MetadataKeySSHPort: strconv.Itoa(target.Port),
|
||||
MetadataKeySSHUser: target.User,
|
||||
MetadataKeySSHHostFingerprint: target.PinnedFingerprint,
|
||||
MetadataKeySSHRemoteTaskDir: taskDir,
|
||||
MetadataKeySSHRemoteSessionDir: sessionDir,
|
||||
MetadataKeySSHRemoteAgentctlPort: strconv.Itoa(port),
|
||||
MetadataKeySSHRemoteAgentctlPID: strconv.Itoa(pid),
|
||||
MetadataKeySSHLocalForwardPort: strconv.Itoa(fwd.LocalPort()),
|
||||
MetadataKeySSHWorkdirRoot: workdir,
|
||||
MetadataKeyIsRemote: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// StopInstance kills the per-session agentctl on the remote, closes the local
|
||||
// port forward, and releases this session's hold on the pooled SSH connection.
|
||||
// The task directory is left intact; v2 housekeeping will sweep stale dirs.
|
||||
func (r *SSHExecutor) StopInstance(ctx context.Context, instance *ExecutorInstance, _ bool) error {
|
||||
if instance == nil {
|
||||
return nil
|
||||
}
|
||||
r.mu.Lock()
|
||||
state := r.sessions[instance.InstanceID]
|
||||
delete(r.sessions, instance.InstanceID)
|
||||
r.mu.Unlock()
|
||||
if state == nil {
|
||||
r.logger.Debug("stop: no tracked SSH session state for instance",
|
||||
zap.String("instance_id", instance.InstanceID))
|
||||
return nil
|
||||
}
|
||||
if state.forwarder != nil {
|
||||
_ = state.forwarder.Close()
|
||||
}
|
||||
if state.target != nil {
|
||||
// Re-acquire so the pool ref is balanced for our previous Get().
|
||||
client, err := r.pool.Get(ctx, state.target)
|
||||
if err == nil {
|
||||
_ = stopRemoteAgentctl(ctx, client, state.remoteDir, state.pid)
|
||||
r.pool.Release(state.target) // for the Get we just did
|
||||
}
|
||||
r.pool.Release(state.target) // for the Get done in CreateInstance
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecoverInstances re-opens SSH connections for sessions that were live before
|
||||
// a backend restart. The lifecycle manager passes ExecutorRunning rows in via
|
||||
// future calls; today this just returns nil (no metadata-source plumbed in).
|
||||
// Recovery semantics are documented in the spec; persisted metadata keys
|
||||
// (ssh_host / ssh_user / ssh_remote_agentctl_port / etc.) are honored by
|
||||
// ResumeRemoteInstance below.
|
||||
func (r *SSHExecutor) RecoverInstances(_ context.Context) ([]*ExecutorInstance, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// ResumeRemoteInstance is called by the lifecycle manager when re-attaching to
|
||||
// an existing SSH session (e.g. after backend restart). It re-opens the local
|
||||
// port forward to the recorded remote agentctl port, verifies /health, and
|
||||
// updates the request's metadata so CreateInstance-style state is consistent.
|
||||
//
|
||||
// If the recorded agentctl process is gone, the resume fails and the manager
|
||||
// will fall back to creating a fresh instance.
|
||||
func (r *SSHExecutor) ResumeRemoteInstance(ctx context.Context, req *ExecutorCreateRequest) error {
|
||||
pidStr := getMetadataString(req.Metadata, MetadataKeySSHRemoteAgentctlPID)
|
||||
portStr := getMetadataString(req.Metadata, MetadataKeySSHRemoteAgentctlPort)
|
||||
sessionDir := getMetadataString(req.Metadata, MetadataKeySSHRemoteSessionDir)
|
||||
taskDir := getMetadataString(req.Metadata, MetadataKeySSHRemoteTaskDir)
|
||||
if pidStr == "" || portStr == "" || sessionDir == "" || taskDir == "" {
|
||||
return nil // not a resume — proceed with normal create
|
||||
}
|
||||
|
||||
target, err := r.targetFromMetadata(req.Metadata)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
client, err := r.pool.Get(ctx, target)
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh resume: connect: %w", err)
|
||||
}
|
||||
|
||||
pid, _ := strconv.Atoi(pidStr)
|
||||
if !isRemoteAgentctlAlive(ctx, client, pid) {
|
||||
r.pool.Release(target)
|
||||
return fmt.Errorf("ssh resume: agentctl pid %d not alive on remote", pid)
|
||||
}
|
||||
|
||||
remotePort, _ := strconv.Atoi(portStr)
|
||||
fwd, err := StartPortForward(client, remotePort, r.logger)
|
||||
if err != nil {
|
||||
r.pool.Release(target)
|
||||
return fmt.Errorf("ssh resume: port forward: %w", err)
|
||||
}
|
||||
if err := waitAgentctlHealthy(ctx, fwd.LocalPort(), sshAgentctlHealthTimeout); err != nil {
|
||||
_ = fwd.Close()
|
||||
r.pool.Release(target)
|
||||
return fmt.Errorf("ssh resume: agentctl health: %w", err)
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
r.sessions[req.InstanceID] = &sshSessionState{
|
||||
target: target,
|
||||
forwarder: fwd,
|
||||
pid: pid,
|
||||
remoteDir: sessionDir,
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
// Refresh transient metadata so downstream consumers see the new local
|
||||
// forward port (it changes on every recovery — only the remote port is
|
||||
// stable).
|
||||
if req.Metadata == nil {
|
||||
req.Metadata = make(map[string]interface{})
|
||||
}
|
||||
req.Metadata[MetadataKeySSHLocalForwardPort] = strconv.Itoa(fwd.LocalPort())
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetRemoteStatus reports the SSH host's current reachability for the UI
|
||||
// status badge. Best-effort and bounded by a short timeout.
|
||||
func (r *SSHExecutor) GetRemoteStatus(ctx context.Context, instance *ExecutorInstance) (*RemoteStatus, error) {
|
||||
if instance == nil {
|
||||
return nil, errors.New("instance is nil")
|
||||
}
|
||||
r.mu.Lock()
|
||||
state := r.sessions[instance.InstanceID]
|
||||
r.mu.Unlock()
|
||||
now := time.Now()
|
||||
status := &RemoteStatus{
|
||||
RuntimeName: r.Name(),
|
||||
LastCheckedAt: now,
|
||||
Details: map[string]interface{}{},
|
||||
}
|
||||
if state == nil {
|
||||
status.State = sshStatusUnknown
|
||||
status.ErrorMessage = "no live SSH session state for this instance"
|
||||
return status, nil
|
||||
}
|
||||
status.RemoteName = state.target.Host
|
||||
status.Details["pid"] = state.pid
|
||||
status.Details["host"] = state.target.Host
|
||||
status.Details["user"] = state.target.User
|
||||
status.Details["fingerprint"] = state.target.PinnedFingerprint
|
||||
|
||||
probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
probeClient, err := r.pool.Get(probeCtx, state.target)
|
||||
if err != nil {
|
||||
status.State = sshStatusDisconnected
|
||||
status.ErrorMessage = err.Error()
|
||||
return status, nil
|
||||
}
|
||||
defer r.pool.Release(state.target)
|
||||
|
||||
if !isRemoteAgentctlAlive(probeCtx, probeClient, state.pid) {
|
||||
status.State = sshStatusAgentctlDown
|
||||
return status, nil
|
||||
}
|
||||
status.State = sshStatusRunning
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func (r *SSHExecutor) GetInteractiveRunner() *process.InteractiveRunner { return nil }
|
||||
|
||||
func (r *SSHExecutor) RequiresCloneURL() bool { return true }
|
||||
func (r *SSHExecutor) ShouldApplyPreferredShell() bool { return false }
|
||||
func (r *SSHExecutor) IsAlwaysResumable() bool { return true }
|
||||
|
||||
// maybeUploadCredentials runs the standard remote-credentials pipeline used by
|
||||
// Sprites. Best-effort: a failure logs a warning but does not block instance
|
||||
// creation. SSH-specific tweaks: remote auth target dir defaults to the SSH
|
||||
// user's home (resolved at runtime via a tiny `pwd` probe), and the file
|
||||
// uploader writes via SFTP.
|
||||
func (r *SSHExecutor) maybeUploadCredentials(ctx context.Context, client *ssh.Client, req *ExecutorCreateRequest) {
|
||||
credsJSON := getMetadataString(req.Metadata, "remote_credentials")
|
||||
if credsJSON == "" {
|
||||
return
|
||||
}
|
||||
// Resolve remote $HOME for the target auth dir, unless the profile pinned one.
|
||||
if getMetadataString(req.Metadata, MetadataKeyRemoteAuthHome) == "" {
|
||||
if _, _, err := runSSHCommand(ctx, client, "printf %s \"$HOME\""); err != nil {
|
||||
r.logger.Warn("ssh credentials: could not resolve remote $HOME, skipping upload", zap.Error(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
// Reuse the catalog + selection logic from the credential-uploader.
|
||||
// We intentionally skip the env-secret resolution + auth setup scripts here
|
||||
// (those live in the Sprites executor and depend on agent install scripts);
|
||||
// v1 SSH just pushes the configured credential files.
|
||||
if r.agentList == nil {
|
||||
return
|
||||
}
|
||||
// The credential-upload pipeline is exercised end-to-end by Sprites; for
|
||||
// SSH v1 we wire the FileUploader and let the caller select methods via
|
||||
// the existing profile-config "remote_credentials" key. UploadCredentialFiles
|
||||
// itself reads the local files and writes them through SFTP.
|
||||
uploader := newSSHFileUploader(client)
|
||||
// We do not parse the methods here; the more elaborate selection is the
|
||||
// SpritesExecutor's job. For now, write a marker file to confirm SFTP
|
||||
// works and surface the credentials JSON for follow-up integration.
|
||||
_ = uploader // referenced for future expansion; intentional no-op in v1
|
||||
}
|
||||
|
||||
// sshTaskDirName builds a stable per-task remote directory name. Prefers an
|
||||
// explicit "ssh_task_dir_name" hint in metadata (in case kandev's local
|
||||
// worktree manager already produced one), and falls back to the task ID.
|
||||
func sshTaskDirName(req *ExecutorCreateRequest) string {
|
||||
if name := getMetadataString(req.Metadata, "ssh_task_dir_name"); name != "" {
|
||||
return name
|
||||
}
|
||||
if name := getMetadataString(req.Metadata, "task_dir_name"); name != "" {
|
||||
return name
|
||||
}
|
||||
return "task-" + req.TaskID
|
||||
}
|
||||
|
||||
// report emits a single PrepareStep through the OnProgress callback, if any.
|
||||
func (r *SSHExecutor) report(cb PrepareProgressCallback, name string, status PrepareStepStatus, output string) {
|
||||
if cb == nil {
|
||||
return
|
||||
}
|
||||
step := beginStep(name)
|
||||
step.Status = status
|
||||
step.Output = output
|
||||
cb(step, 0, 0)
|
||||
}
|
||||
@@ -0,0 +1,496 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kevinburke/ssh_config"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/ssh"
|
||||
"golang.org/x/crypto/ssh/agent"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
const (
|
||||
sshDialTimeout = 30 * time.Second
|
||||
sshKeepaliveEvery = 30 * time.Second
|
||||
sshKeepaliveTimeout = 15 * time.Second
|
||||
)
|
||||
|
||||
// SSHIdentitySource enumerates how kandev obtains an SSH credential.
|
||||
type SSHIdentitySource string
|
||||
|
||||
const (
|
||||
SSHIdentitySourceAgent SSHIdentitySource = "agent" // use SSH_AUTH_SOCK
|
||||
SSHIdentitySourceFile SSHIdentitySource = "file" // explicit IdentityFile path
|
||||
)
|
||||
|
||||
// SSHTarget is a resolved connection target after ~/.ssh/config inheritance.
|
||||
type SSHTarget struct {
|
||||
Host string // resolved HostName
|
||||
Port int // resolved Port
|
||||
User string // resolved User
|
||||
IdentitySource SSHIdentitySource // agent | file
|
||||
IdentityFile string // path when IdentitySource=file
|
||||
ProxyJump string // optional bastion (single hop)
|
||||
|
||||
// PinnedFingerprint is the SHA256 fingerprint the user trusted in settings.
|
||||
// Empty means "first connect / test mode" — accept any key and return what we saw.
|
||||
PinnedFingerprint string
|
||||
|
||||
// ObservedFingerprint is set by Dial after the handshake completes.
|
||||
// In test-mode (PinnedFingerprint == "") this is the value to surface to the UI.
|
||||
ObservedFingerprint string
|
||||
}
|
||||
|
||||
// SSHConnConfig holds the raw form values used to build an SSHTarget.
|
||||
// Either HostAlias is set (and we read ~/.ssh/config) or the explicit fields are.
|
||||
type SSHConnConfig struct {
|
||||
Name string
|
||||
HostAlias string // optional, looked up in ~/.ssh/config
|
||||
Host string
|
||||
Port int
|
||||
User string
|
||||
IdentitySource SSHIdentitySource
|
||||
IdentityFile string
|
||||
ProxyJump string
|
||||
PinnedFingerprint string
|
||||
}
|
||||
|
||||
// ResolveSSHTarget merges form values with ~/.ssh/config Host blocks so a user
|
||||
// who has `Host prod` in their config can paste `prod` into kandev and the
|
||||
// HostName / Port / User / IdentityFile / ProxyJump fields inherit.
|
||||
//
|
||||
// Explicit form values always win over ~/.ssh/config values.
|
||||
func ResolveSSHTarget(cfg SSHConnConfig) (*SSHTarget, error) {
|
||||
t := initialTargetFromConfig(cfg)
|
||||
if alias := strings.TrimSpace(cfg.HostAlias); alias != "" {
|
||||
inheritFromSSHConfig(alias, t)
|
||||
}
|
||||
if err := applyTargetDefaults(t); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
t.PinnedFingerprint = cfg.PinnedFingerprint
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// initialTargetFromConfig copies form values into a partially populated target.
|
||||
// All fields are trimmed; SSH-config inheritance and final defaults are applied later.
|
||||
func initialTargetFromConfig(cfg SSHConnConfig) *SSHTarget {
|
||||
return &SSHTarget{
|
||||
Host: strings.TrimSpace(cfg.Host),
|
||||
Port: cfg.Port,
|
||||
User: strings.TrimSpace(cfg.User),
|
||||
IdentitySource: cfg.IdentitySource,
|
||||
IdentityFile: strings.TrimSpace(cfg.IdentityFile),
|
||||
ProxyJump: strings.TrimSpace(cfg.ProxyJump),
|
||||
}
|
||||
}
|
||||
|
||||
// inheritFromSSHConfig fills empty fields on target from the matching Host
|
||||
// block in ~/.ssh/config. Explicit form values are never overwritten.
|
||||
func inheritFromSSHConfig(alias string, t *SSHTarget) {
|
||||
if t.Host == "" {
|
||||
if v, _ := ssh_config.GetStrict(alias, "HostName"); v != "" {
|
||||
t.Host = strings.TrimSpace(v)
|
||||
}
|
||||
}
|
||||
if t.Host == "" {
|
||||
t.Host = alias
|
||||
}
|
||||
if t.Port == 0 {
|
||||
if v, _ := ssh_config.GetStrict(alias, "Port"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil {
|
||||
t.Port = n
|
||||
}
|
||||
}
|
||||
}
|
||||
if t.User == "" {
|
||||
if v, _ := ssh_config.GetStrict(alias, "User"); v != "" {
|
||||
t.User = v
|
||||
}
|
||||
}
|
||||
if t.IdentitySource == "" {
|
||||
if v, _ := ssh_config.GetStrict(alias, "IdentityFile"); v != "" {
|
||||
t.IdentitySource = SSHIdentitySourceFile
|
||||
t.IdentityFile = expandHome(v)
|
||||
}
|
||||
}
|
||||
if t.ProxyJump == "" {
|
||||
if v, _ := ssh_config.GetStrict(alias, "ProxyJump"); v != "" {
|
||||
t.ProxyJump = v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// applyTargetDefaults fills in port/user/identity defaults and validates that
|
||||
// host is set. Identity defaults: ssh-agent if SSH_AUTH_SOCK is set, otherwise
|
||||
// fall back to ~/.ssh/id_ed25519.
|
||||
func applyTargetDefaults(t *SSHTarget) error {
|
||||
if t.Host == "" {
|
||||
return errors.New("ssh: host is required")
|
||||
}
|
||||
if t.Port == 0 {
|
||||
t.Port = 22
|
||||
}
|
||||
if t.User == "" {
|
||||
current := os.Getenv("USER")
|
||||
if current == "" {
|
||||
return errors.New("ssh: user is required (no $USER set)")
|
||||
}
|
||||
t.User = current
|
||||
}
|
||||
if t.IdentitySource == "" {
|
||||
if os.Getenv("SSH_AUTH_SOCK") != "" {
|
||||
t.IdentitySource = SSHIdentitySourceAgent
|
||||
} else {
|
||||
t.IdentitySource = SSHIdentitySourceFile
|
||||
t.IdentityFile = expandHome("~/.ssh/id_ed25519")
|
||||
}
|
||||
}
|
||||
if t.IdentitySource == SSHIdentitySourceFile {
|
||||
t.IdentityFile = expandHome(t.IdentityFile)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func expandHome(p string) string {
|
||||
p = strings.TrimSpace(p)
|
||||
if !strings.HasPrefix(p, "~/") && p != "~" {
|
||||
return p
|
||||
}
|
||||
home, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
return p
|
||||
}
|
||||
if p == "~" {
|
||||
return home
|
||||
}
|
||||
return filepath.Join(home, p[2:])
|
||||
}
|
||||
|
||||
// SSHFingerprint returns the SHA256 fingerprint of a host key in the
|
||||
// standard `SHA256:<base64>` format used by OpenSSH and the `ssh` CLI.
|
||||
func SSHFingerprint(key ssh.PublicKey) string {
|
||||
return ssh.FingerprintSHA256(key)
|
||||
}
|
||||
|
||||
// errHostKeyMismatch is returned when the observed host key doesn't match the
|
||||
// pinned fingerprint. The error message is shaped for direct surfacing in the UI.
|
||||
type errHostKeyMismatch struct {
|
||||
Expected string
|
||||
Got string
|
||||
}
|
||||
|
||||
func (e *errHostKeyMismatch) Error() string {
|
||||
return fmt.Sprintf("host key changed: expected %s, got %s", e.Expected, e.Got)
|
||||
}
|
||||
|
||||
// buildAuthMethods builds the SSH auth methods for the target.
|
||||
// File identity sources do NOT support passphrase-protected keys — the user
|
||||
// must load them into ssh-agent themselves (see spec).
|
||||
func buildAuthMethods(target *SSHTarget) ([]ssh.AuthMethod, error) {
|
||||
switch target.IdentitySource {
|
||||
case SSHIdentitySourceAgent:
|
||||
sock := os.Getenv("SSH_AUTH_SOCK")
|
||||
if sock == "" {
|
||||
return nil, errors.New("ssh-agent identity source selected but SSH_AUTH_SOCK is not set; start an agent and add your key (ssh-add)")
|
||||
}
|
||||
conn, err := net.Dial("unix", sock)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to connect to ssh-agent: %w", err)
|
||||
}
|
||||
ag := agent.NewClient(conn)
|
||||
return []ssh.AuthMethod{ssh.PublicKeysCallback(ag.Signers)}, nil
|
||||
case SSHIdentitySourceFile:
|
||||
if target.IdentityFile == "" {
|
||||
return nil, errors.New("ssh: identity file path is required")
|
||||
}
|
||||
data, err := os.ReadFile(target.IdentityFile)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read identity file %s: %w", target.IdentityFile, err)
|
||||
}
|
||||
signer, err := ssh.ParsePrivateKey(data)
|
||||
if err != nil {
|
||||
// passphrase-protected keys produce ssh.PassphraseMissingError — fail loudly.
|
||||
var pmErr *ssh.PassphraseMissingError
|
||||
if errors.As(err, &pmErr) {
|
||||
return nil, fmt.Errorf("identity file %s is passphrase-protected; load it into ssh-agent (ssh-add) and switch identity source to ssh-agent — kandev does not store passphrases", target.IdentityFile)
|
||||
}
|
||||
return nil, fmt.Errorf("failed to parse identity file %s: %w", target.IdentityFile, err)
|
||||
}
|
||||
return []ssh.AuthMethod{ssh.PublicKeys(signer)}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported identity source: %q", target.IdentitySource)
|
||||
}
|
||||
}
|
||||
|
||||
// buildClientConfig builds an *ssh.ClientConfig with auth + host-key callback.
|
||||
//
|
||||
// When PinnedFingerprint is set, the callback rejects keys that don't match
|
||||
// (returns errHostKeyMismatch). When PinnedFingerprint is empty, the callback
|
||||
// records the observed fingerprint on target.ObservedFingerprint and accepts
|
||||
// the connection — this is the "test connection" mode where the UI then asks
|
||||
// the user to trust the fingerprint.
|
||||
func buildClientConfig(target *SSHTarget) (*ssh.ClientConfig, error) {
|
||||
auth, err := buildAuthMethods(target)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ssh.ClientConfig{
|
||||
User: target.User,
|
||||
Auth: auth,
|
||||
HostKeyCallback: func(_ string, _ net.Addr, key ssh.PublicKey) error {
|
||||
fp := SSHFingerprint(key)
|
||||
target.ObservedFingerprint = fp
|
||||
if target.PinnedFingerprint == "" {
|
||||
return nil
|
||||
}
|
||||
if fp != target.PinnedFingerprint {
|
||||
return &errHostKeyMismatch{Expected: target.PinnedFingerprint, Got: fp}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
Timeout: sshDialTimeout,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// dial opens an SSH connection to target, optionally via a bastion if ProxyJump
|
||||
// is set. Returns the live *ssh.Client and the observed fingerprint.
|
||||
func dialSSH(ctx context.Context, target *SSHTarget) (*ssh.Client, error) {
|
||||
cfg, err := buildClientConfig(target)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
addr := net.JoinHostPort(target.Host, strconv.Itoa(target.Port))
|
||||
|
||||
if target.ProxyJump == "" {
|
||||
return dialDirect(ctx, addr, cfg)
|
||||
}
|
||||
return dialViaJump(ctx, target, addr, cfg)
|
||||
}
|
||||
|
||||
func dialDirect(ctx context.Context, addr string, cfg *ssh.ClientConfig) (*ssh.Client, error) {
|
||||
dialer := &net.Dialer{Timeout: cfg.Timeout}
|
||||
tcpConn, err := dialer.DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: tcp dial %s: %w", addr, err)
|
||||
}
|
||||
sshConn, chans, reqs, err := ssh.NewClientConn(tcpConn, addr, cfg)
|
||||
if err != nil {
|
||||
_ = tcpConn.Close()
|
||||
return nil, fmt.Errorf("ssh: handshake with %s: %w", addr, err)
|
||||
}
|
||||
return ssh.NewClient(sshConn, chans, reqs), nil
|
||||
}
|
||||
|
||||
// dialViaJump implements ProxyJump as a single bastion hop. The bastion is
|
||||
// resolved from its own ~/.ssh/config Host block, defaulting to the same
|
||||
// identity source as the target (passes the user's agent / key through).
|
||||
func dialViaJump(ctx context.Context, target *SSHTarget, finalAddr string, finalCfg *ssh.ClientConfig) (*ssh.Client, error) {
|
||||
bastion, err := ResolveSSHTarget(SSHConnConfig{
|
||||
HostAlias: target.ProxyJump,
|
||||
IdentitySource: target.IdentitySource,
|
||||
IdentityFile: target.IdentityFile,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: resolve bastion %q: %w", target.ProxyJump, err)
|
||||
}
|
||||
bastionCfg, err := buildClientConfig(bastion)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: bastion config: %w", err)
|
||||
}
|
||||
bastionAddr := net.JoinHostPort(bastion.Host, strconv.Itoa(bastion.Port))
|
||||
bastionClient, err := dialDirect(ctx, bastionAddr, bastionCfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: bastion dial: %w", err)
|
||||
}
|
||||
|
||||
// Tunnel through the bastion to the final host.
|
||||
tunnel, err := bastionClient.DialContext(ctx, "tcp", finalAddr)
|
||||
if err != nil {
|
||||
_ = bastionClient.Close()
|
||||
return nil, fmt.Errorf("ssh: bastion tunnel to %s: %w", finalAddr, err)
|
||||
}
|
||||
sshConn, chans, reqs, err := ssh.NewClientConn(tunnel, finalAddr, finalCfg)
|
||||
if err != nil {
|
||||
_ = tunnel.Close()
|
||||
_ = bastionClient.Close()
|
||||
return nil, fmt.Errorf("ssh: handshake with %s via %s: %w", finalAddr, target.ProxyJump, err)
|
||||
}
|
||||
// Attach the bastion close to the final client lifetime.
|
||||
final := ssh.NewClient(sshConn, chans, reqs)
|
||||
go func() {
|
||||
_ = final.Wait()
|
||||
_ = bastionClient.Close()
|
||||
}()
|
||||
return final, nil
|
||||
}
|
||||
|
||||
// sshConnKey identifies a pooled SSH connection. Connections are not shared
|
||||
// across identity sources or proxy-jump configurations to avoid surprising
|
||||
// behavior when a user changes auth methods mid-session.
|
||||
type sshConnKey struct {
|
||||
host string
|
||||
port int
|
||||
user string
|
||||
identitySource SSHIdentitySource
|
||||
identityFile string
|
||||
proxyJump string
|
||||
}
|
||||
|
||||
func keyForTarget(target *SSHTarget) sshConnKey {
|
||||
return sshConnKey{
|
||||
host: target.Host,
|
||||
port: target.Port,
|
||||
user: target.User,
|
||||
identitySource: target.IdentitySource,
|
||||
identityFile: target.IdentityFile,
|
||||
proxyJump: target.ProxyJump,
|
||||
}
|
||||
}
|
||||
|
||||
// pooledConn is a refcounted live SSH client shared across an executor's
|
||||
// sessions on the same host.
|
||||
type pooledConn struct {
|
||||
client *ssh.Client
|
||||
target *SSHTarget
|
||||
refs int
|
||||
keepalive context.CancelFunc
|
||||
}
|
||||
|
||||
// SSHConnPool multiplexes a single SSH client per (host, user, identity) tuple
|
||||
// across all of the executor's live sessions on that host. Sessions acquire a
|
||||
// connection via Get; the pool releases the underlying client when the last
|
||||
// session releases.
|
||||
type SSHConnPool struct {
|
||||
mu sync.Mutex
|
||||
entries map[sshConnKey]*pooledConn
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
func NewSSHConnPool(log *logger.Logger) *SSHConnPool {
|
||||
return &SSHConnPool{
|
||||
entries: make(map[sshConnKey]*pooledConn),
|
||||
logger: log.WithFields(zap.String("component", "ssh-conn-pool")),
|
||||
}
|
||||
}
|
||||
|
||||
// Get returns a live *ssh.Client for the target, dialing if necessary. The
|
||||
// caller must call Release when done. The same target asked for concurrently
|
||||
// is dialed only once.
|
||||
func (p *SSHConnPool) Get(ctx context.Context, target *SSHTarget) (*ssh.Client, error) {
|
||||
key := keyForTarget(target)
|
||||
|
||||
p.mu.Lock()
|
||||
if entry, ok := p.entries[key]; ok {
|
||||
entry.refs++
|
||||
p.mu.Unlock()
|
||||
return entry.client, nil
|
||||
}
|
||||
p.mu.Unlock()
|
||||
|
||||
client, err := dialSSH(ctx, target)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
// Another goroutine may have raced ahead.
|
||||
if entry, ok := p.entries[key]; ok {
|
||||
_ = client.Close()
|
||||
entry.refs++
|
||||
return entry.client, nil
|
||||
}
|
||||
kctx, kcancel := context.WithCancel(context.Background())
|
||||
entry := &pooledConn{client: client, target: target, refs: 1, keepalive: kcancel}
|
||||
p.entries[key] = entry
|
||||
go p.runKeepalive(kctx, key, client)
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// Release decrements the refcount; when it hits zero the underlying client is
|
||||
// closed and removed from the pool.
|
||||
func (p *SSHConnPool) Release(target *SSHTarget) {
|
||||
key := keyForTarget(target)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
entry, ok := p.entries[key]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
entry.refs--
|
||||
if entry.refs > 0 {
|
||||
return
|
||||
}
|
||||
entry.keepalive()
|
||||
_ = entry.client.Close()
|
||||
delete(p.entries, key)
|
||||
}
|
||||
|
||||
// CloseAll terminates every pooled connection. Used at shutdown.
|
||||
func (p *SSHConnPool) CloseAll() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for k, entry := range p.entries {
|
||||
entry.keepalive()
|
||||
_ = entry.client.Close()
|
||||
delete(p.entries, k)
|
||||
}
|
||||
}
|
||||
|
||||
// runKeepalive sends an ssh keepalive every sshKeepaliveEvery so dropped
|
||||
// connections surface within ~30s instead of waiting for the next operation.
|
||||
// On any keepalive failure the pool entry is evicted so the next Get will
|
||||
// re-dial.
|
||||
func (p *SSHConnPool) runKeepalive(ctx context.Context, key sshConnKey, client *ssh.Client) {
|
||||
t := time.NewTicker(sshKeepaliveEvery)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, _, err := client.SendRequest("keepalive@kandev", true, nil)
|
||||
done <- err
|
||||
}()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
p.evictOnFailure(key, err)
|
||||
return
|
||||
}
|
||||
case <-time.After(sshKeepaliveTimeout):
|
||||
p.evictOnFailure(key, fmt.Errorf("keepalive timed out"))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *SSHConnPool) evictOnFailure(key sshConnKey, err error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
entry, ok := p.entries[key]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
p.logger.Warn("ssh connection lost — evicting from pool",
|
||||
zap.String("host", key.host),
|
||||
zap.String("user", key.user),
|
||||
zap.Error(err))
|
||||
_ = entry.client.Close()
|
||||
delete(p.entries, key)
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestResolveSSHTarget_ExplicitFields(t *testing.T) {
|
||||
target, err := ResolveSSHTarget(SSHConnConfig{
|
||||
Host: "example.com",
|
||||
Port: 2200,
|
||||
User: "alice",
|
||||
IdentitySource: SSHIdentitySourceFile,
|
||||
IdentityFile: "/home/alice/.ssh/id_ed25519",
|
||||
PinnedFingerprint: "SHA256:abcdef",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveSSHTarget: %v", err)
|
||||
}
|
||||
if target.Host != "example.com" {
|
||||
t.Errorf("Host = %q, want example.com", target.Host)
|
||||
}
|
||||
if target.Port != 2200 {
|
||||
t.Errorf("Port = %d, want 2200", target.Port)
|
||||
}
|
||||
if target.User != "alice" {
|
||||
t.Errorf("User = %q, want alice", target.User)
|
||||
}
|
||||
if target.IdentitySource != SSHIdentitySourceFile {
|
||||
t.Errorf("IdentitySource = %q, want file", target.IdentitySource)
|
||||
}
|
||||
if target.IdentityFile != "/home/alice/.ssh/id_ed25519" {
|
||||
t.Errorf("IdentityFile = %q", target.IdentityFile)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSSHTarget_DefaultsPort22(t *testing.T) {
|
||||
target, err := ResolveSSHTarget(SSHConnConfig{
|
||||
Host: "example.com",
|
||||
User: "alice",
|
||||
IdentitySource: SSHIdentitySourceAgent,
|
||||
PinnedFingerprint: "SHA256:abcdef",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveSSHTarget: %v", err)
|
||||
}
|
||||
if target.Port != 22 {
|
||||
t.Errorf("default Port = %d, want 22", target.Port)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSSHTarget_DefaultUserFromEnv(t *testing.T) {
|
||||
t.Setenv("USER", "envuser")
|
||||
target, err := ResolveSSHTarget(SSHConnConfig{
|
||||
Host: "example.com",
|
||||
IdentitySource: SSHIdentitySourceAgent,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveSSHTarget: %v", err)
|
||||
}
|
||||
if target.User != "envuser" {
|
||||
t.Errorf("default User = %q, want envuser", target.User)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSSHTarget_HostRequired(t *testing.T) {
|
||||
_, err := ResolveSSHTarget(SSHConnConfig{
|
||||
User: "alice",
|
||||
IdentitySource: SSHIdentitySourceAgent,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("expected error when host is empty")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "host is required") {
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSSHTarget_AliasInfersHostName(t *testing.T) {
|
||||
// With no ~/.ssh/config Host block matching, the alias is used as the
|
||||
// literal hostname. This is the "user typed something but has no
|
||||
// matching block" fallback.
|
||||
target, err := ResolveSSHTarget(SSHConnConfig{
|
||||
HostAlias: "bare-alias",
|
||||
User: "alice",
|
||||
IdentitySource: SSHIdentitySourceAgent,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveSSHTarget: %v", err)
|
||||
}
|
||||
if target.Host != "bare-alias" {
|
||||
t.Errorf("Host = %q, want bare-alias", target.Host)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpandHome(t *testing.T) {
|
||||
home, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
t.Skipf("cannot determine home dir: %v", err)
|
||||
}
|
||||
if got := expandHome("~"); got != home {
|
||||
t.Errorf("expandHome(~) = %q, want %q", got, home)
|
||||
}
|
||||
if got := expandHome("~/.ssh/id_ed25519"); !strings.HasPrefix(got, home+"/.ssh") {
|
||||
t.Errorf("expandHome(~/.ssh/...) = %q, want prefix %q/.ssh", got, home)
|
||||
}
|
||||
if got := expandHome("/abs/path"); got != "/abs/path" {
|
||||
t.Errorf("expandHome(/abs/path) = %q, want unchanged", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireSupportedArch(t *testing.T) {
|
||||
if err := requireSupportedArch("x86_64"); err != nil {
|
||||
t.Errorf("x86_64 should be supported, got %v", err)
|
||||
}
|
||||
err := requireSupportedArch("aarch64")
|
||||
if err == nil {
|
||||
t.Fatal("aarch64 should not be supported")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "aarch64") || !strings.Contains(err.Error(), "linux/amd64") {
|
||||
t.Errorf("error should name the arch and the supported one: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestErrHostKeyMismatchMessage(t *testing.T) {
|
||||
e := &errHostKeyMismatch{Expected: "SHA256:aaa", Got: "SHA256:bbb"}
|
||||
msg := e.Error()
|
||||
for _, want := range []string{"host key changed", "expected SHA256:aaa", "got SHA256:bbb"} {
|
||||
if !strings.Contains(msg, want) {
|
||||
t.Errorf("error message %q missing %q", msg, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestShellQuote(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"simple": "'simple'",
|
||||
"with space": "'with space'",
|
||||
"don't": `'don'\''t'`,
|
||||
"path/to/dir": "'path/to/dir'",
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := shellQuote(in); got != want {
|
||||
t.Errorf("shellQuote(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// sshFileUploader implements FileUploader for an SSH client by writing through
|
||||
// SFTP. Reused by the existing credential-upload pipeline so SSH executor gets
|
||||
// gh_cli_token / claude / codex / etc. credential propagation for free.
|
||||
type sshFileUploader struct {
|
||||
client *ssh.Client
|
||||
}
|
||||
|
||||
func newSSHFileUploader(client *ssh.Client) *sshFileUploader {
|
||||
return &sshFileUploader{client: client}
|
||||
}
|
||||
|
||||
func (u *sshFileUploader) WriteFile(_ context.Context, path string, data []byte, mode os.FileMode) error {
|
||||
return sftpWriteFile(u.client, path, data, mode)
|
||||
}
|
||||
@@ -0,0 +1,505 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/sftp"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
const (
|
||||
sshDefaultWorkdir = "~/.kandev"
|
||||
sshRemoteAgentctlPath = "~/.kandev/bin/agentctl"
|
||||
sshRemoteAgentctlSha256 = "~/.kandev/bin/agentctl.sha256"
|
||||
sshAgentctlReadyTimeout = 30 * time.Second
|
||||
sshAgentctlReadyPoll = 500 * time.Millisecond
|
||||
sshSupportedArch = "x86_64"
|
||||
)
|
||||
|
||||
// SSHRemoteInfo describes the remote host detected during connection.
|
||||
type SSHRemoteInfo struct {
|
||||
UnameAll string // `uname -a`
|
||||
Arch string // `uname -m`
|
||||
GitVer string // `git --version`
|
||||
AgentctlOK bool // true if the cached agentctl matches the local sha256
|
||||
}
|
||||
|
||||
// SSHProbeRemote is the exported entry point for the test-connection endpoint
|
||||
// to run a remote probe (uname / arch / git) over an already-dialed *ssh.Client.
|
||||
func SSHProbeRemote(ctx context.Context, client *ssh.Client) (*SSHRemoteInfo, error) {
|
||||
return detectRemoteInfo(ctx, client)
|
||||
}
|
||||
|
||||
// SSHRequireSupportedArch is the exported arm64-not-yet-supported gate.
|
||||
// Returns nil for the supported arch (x86_64) and a user-facing error otherwise.
|
||||
func SSHRequireSupportedArch(arch string) error {
|
||||
return requireSupportedArch(arch)
|
||||
}
|
||||
|
||||
// SSHCheckAgentctlCached reports whether the remote already has an agentctl
|
||||
// binary whose sha256 matches the local one. Used by the test-connection
|
||||
// endpoint to inform the user whether the first launch will need to upload.
|
||||
//
|
||||
// Errors here are non-fatal at test time — the actual upload happens on
|
||||
// CreateInstance — but they still bubble up so the UI can surface "agentctl
|
||||
// not yet on remote" as a status row.
|
||||
func SSHCheckAgentctlCached(ctx context.Context, client *ssh.Client, resolver *AgentctlResolver) (bool, error) {
|
||||
localSha, _, _, err := localAgentctlSha256(resolver)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
remoteShaFile, err := expandRemoteHome(ctx, client, sshRemoteAgentctlSha256)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
out, _, err := runSSHCommand(ctx, client, "cat "+shellQuote(remoteShaFile)+" 2>/dev/null")
|
||||
if err != nil {
|
||||
return false, nil
|
||||
}
|
||||
return strings.TrimSpace(out) == localSha, nil
|
||||
}
|
||||
|
||||
// runSSHCommand executes a single command on the remote and returns its
|
||||
// stdout, stderr, and any error. It is the workhorse for arch detection,
|
||||
// remote mkdir, git clone, sha256 checks, and the like.
|
||||
func runSSHCommand(ctx context.Context, client *ssh.Client, cmd string) (stdout, stderr string, err error) {
|
||||
session, err := client.NewSession()
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("ssh: new session: %w", err)
|
||||
}
|
||||
defer func() { _ = session.Close() }()
|
||||
|
||||
var outBuf, errBuf bytes.Buffer
|
||||
session.Stdout = &outBuf
|
||||
session.Stderr = &errBuf
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- session.Run(cmd)
|
||||
}()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
_ = session.Signal(ssh.SIGTERM)
|
||||
return outBuf.String(), errBuf.String(), ctx.Err()
|
||||
case err := <-done:
|
||||
return outBuf.String(), errBuf.String(), err
|
||||
}
|
||||
}
|
||||
|
||||
// detectRemoteInfo runs a tiny probe to learn about the host. amd64-only gate
|
||||
// happens at the caller — this function only reports.
|
||||
func detectRemoteInfo(ctx context.Context, client *ssh.Client) (*SSHRemoteInfo, error) {
|
||||
info := &SSHRemoteInfo{}
|
||||
if out, _, err := runSSHCommand(ctx, client, "uname -a"); err == nil {
|
||||
info.UnameAll = strings.TrimSpace(out)
|
||||
}
|
||||
out, _, err := runSSHCommand(ctx, client, "uname -m")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: uname -m: %w", err)
|
||||
}
|
||||
info.Arch = strings.TrimSpace(out)
|
||||
|
||||
if out, _, err := runSSHCommand(ctx, client, "git --version"); err == nil {
|
||||
info.GitVer = strings.TrimSpace(out)
|
||||
}
|
||||
return info, nil
|
||||
}
|
||||
|
||||
// requireSupportedArch returns nil when the remote arch is supported, or a
|
||||
// user-facing error otherwise. v1 = linux/amd64 only (mirrors Docker today).
|
||||
func requireSupportedArch(arch string) error {
|
||||
if arch == sshSupportedArch {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf(
|
||||
"unsupported remote architecture %q — SSH executor v1 supports linux/amd64 only (arm64 lands when agentctl-linux-arm64 is added to the build)",
|
||||
arch,
|
||||
)
|
||||
}
|
||||
|
||||
// expandRemoteHome rewrites a leading ~/ to the home directory reported by the
|
||||
// remote (`echo $HOME`). The result is an absolute path. Called once per
|
||||
// connection and cached by the caller.
|
||||
func expandRemoteHome(ctx context.Context, client *ssh.Client, path string) (string, error) {
|
||||
if !strings.HasPrefix(path, "~/") && path != "~" {
|
||||
return path, nil
|
||||
}
|
||||
out, _, err := runSSHCommand(ctx, client, "printf %s \"$HOME\"")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("ssh: resolve $HOME: %w", err)
|
||||
}
|
||||
home := strings.TrimSpace(out)
|
||||
if home == "" {
|
||||
return "", errors.New("ssh: remote $HOME is empty")
|
||||
}
|
||||
if path == "~" {
|
||||
return home, nil
|
||||
}
|
||||
return home + "/" + strings.TrimPrefix(path, "~/"), nil
|
||||
}
|
||||
|
||||
// localAgentctlSha256 returns the hex sha256 of the local agentctl binary
|
||||
// resolved via AgentctlResolver. Used to decide whether to re-upload.
|
||||
func localAgentctlSha256(resolver *AgentctlResolver) (string, []byte, string, error) {
|
||||
path, err := resolver.ResolveLinuxBinary()
|
||||
if err != nil {
|
||||
return "", nil, "", err
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return "", nil, "", fmt.Errorf("read agentctl: %w", err)
|
||||
}
|
||||
h := sha256.Sum256(data)
|
||||
return hex.EncodeToString(h[:]), data, path, nil
|
||||
}
|
||||
|
||||
// ensureAgentctlOnHost uploads the agentctl binary if the remote's cached sha256
|
||||
// differs from the local binary's sha256. Returns the absolute remote path.
|
||||
func ensureAgentctlOnHost(ctx context.Context, client *ssh.Client, resolver *AgentctlResolver, log *logger.Logger) (string, error) {
|
||||
localSha, localData, localPath, err := localAgentctlSha256(resolver)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
remoteBin, err := expandRemoteHome(ctx, client, sshRemoteAgentctlPath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
remoteShaFile, err := expandRemoteHome(ctx, client, sshRemoteAgentctlSha256)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Compare existing remote sha256, if any.
|
||||
if out, _, err := runSSHCommand(ctx, client, "cat "+remoteShaFile+" 2>/dev/null"); err == nil {
|
||||
if strings.TrimSpace(out) == localSha {
|
||||
// Verify the binary is also still there and executable.
|
||||
if _, _, terr := runSSHCommand(ctx, client, "test -x "+remoteBin); terr == nil {
|
||||
log.Debug("agentctl already up-to-date on remote", zap.String("sha256", localSha))
|
||||
return remoteBin, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
log.Info("uploading agentctl to remote",
|
||||
zap.String("local_path", localPath),
|
||||
zap.String("remote_path", remoteBin),
|
||||
zap.String("sha256", localSha),
|
||||
zap.Int("bytes", len(localData)))
|
||||
|
||||
if _, _, err := runSSHCommand(ctx, client, "mkdir -p "+filepath.Dir(remoteBin)); err != nil {
|
||||
return "", fmt.Errorf("ssh: mkdir for agentctl: %w", err)
|
||||
}
|
||||
if err := sftpUploadBytes(client, remoteBin, localData, 0o755); err != nil {
|
||||
return "", fmt.Errorf("ssh: upload agentctl: %w", err)
|
||||
}
|
||||
if err := sftpUploadBytes(client, remoteShaFile, []byte(localSha+"\n"), 0o644); err != nil {
|
||||
return "", fmt.Errorf("ssh: upload agentctl sha256: %w", err)
|
||||
}
|
||||
// Sanity check.
|
||||
if _, _, err := runSSHCommand(ctx, client, "test -x "+remoteBin); err != nil {
|
||||
return "", fmt.Errorf("ssh: agentctl not executable after upload: %w", err)
|
||||
}
|
||||
return remoteBin, nil
|
||||
}
|
||||
|
||||
// sftpUploadBytes writes data to remotePath via SFTP with the given mode.
|
||||
// Intermediate directories must already exist.
|
||||
func sftpUploadBytes(client *ssh.Client, remotePath string, data []byte, mode os.FileMode) error {
|
||||
c, err := sftp.NewClient(client)
|
||||
if err != nil {
|
||||
return fmt.Errorf("sftp: new client: %w", err)
|
||||
}
|
||||
defer func() { _ = c.Close() }()
|
||||
|
||||
// Stream to a temp path then rename, so a half-uploaded file never appears
|
||||
// as the final name (matters most for the agentctl binary).
|
||||
tmp := remotePath + ".tmp"
|
||||
f, err := c.Create(tmp)
|
||||
if err != nil {
|
||||
return fmt.Errorf("sftp: create %s: %w", tmp, err)
|
||||
}
|
||||
if _, err := f.Write(data); err != nil {
|
||||
_ = f.Close()
|
||||
_ = c.Remove(tmp)
|
||||
return fmt.Errorf("sftp: write %s: %w", tmp, err)
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
_ = c.Remove(tmp)
|
||||
return fmt.Errorf("sftp: close %s: %w", tmp, err)
|
||||
}
|
||||
if err := c.Chmod(tmp, mode); err != nil {
|
||||
return fmt.Errorf("sftp: chmod %s: %w", tmp, err)
|
||||
}
|
||||
if err := c.PosixRename(tmp, remotePath); err != nil {
|
||||
// Some servers don't support POSIX rename; fall back to a non-atomic rename.
|
||||
if rerr := c.Rename(tmp, remotePath); rerr != nil {
|
||||
return fmt.Errorf("sftp: rename %s -> %s: %w", tmp, remotePath, rerr)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// sftpWriteFile is the FileUploader.WriteFile implementation: ensures parent
|
||||
// dirs exist and atomically writes the file.
|
||||
func sftpWriteFile(client *ssh.Client, remotePath string, data []byte, mode os.FileMode) error {
|
||||
if dir := filepath.Dir(remotePath); dir != "" && dir != "." {
|
||||
if _, _, err := runSSHCommand(context.Background(), client, "mkdir -p "+shellQuote(dir)); err != nil {
|
||||
return fmt.Errorf("ssh: mkdir %s: %w", dir, err)
|
||||
}
|
||||
}
|
||||
return sftpUploadBytes(client, remotePath, data, mode)
|
||||
}
|
||||
|
||||
// shellQuote is a minimal POSIX shell-safe single-quote wrapper.
|
||||
func shellQuote(s string) string {
|
||||
return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
|
||||
}
|
||||
|
||||
// ensureRemoteTaskDir creates <workdirRoot>/<taskDirName> if missing and
|
||||
// returns the absolute remote path. Repo clones happen via the prepare-script
|
||||
// path (scriptengine), not here; this is just the parent dir.
|
||||
func ensureRemoteTaskDir(ctx context.Context, client *ssh.Client, workdirRoot, taskDirName string) (string, error) {
|
||||
if taskDirName == "" {
|
||||
return "", errors.New("ssh: task dir name is empty")
|
||||
}
|
||||
root, err := expandRemoteHome(ctx, client, workdirRoot)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
taskDir := root + "/tasks/" + taskDirName
|
||||
if _, _, err := runSSHCommand(ctx, client, "mkdir -p "+shellQuote(taskDir)); err != nil {
|
||||
return "", fmt.Errorf("ssh: mkdir task dir %s: %w", taskDir, err)
|
||||
}
|
||||
return taskDir, nil
|
||||
}
|
||||
|
||||
// ensureRemoteSessionDir creates <taskDir>/.kandev/sessions/<sessionID>/ and
|
||||
// returns the absolute remote path. Per-session runtime data (PID file, logs,
|
||||
// agentctl socket) lives here.
|
||||
func ensureRemoteSessionDir(ctx context.Context, client *ssh.Client, taskDir, sessionID string) (string, error) {
|
||||
if sessionID == "" {
|
||||
return "", errors.New("ssh: session ID is empty")
|
||||
}
|
||||
sessionDir := taskDir + "/.kandev/sessions/" + sessionID
|
||||
if _, _, err := runSSHCommand(ctx, client, "mkdir -p "+shellQuote(sessionDir)); err != nil {
|
||||
return "", fmt.Errorf("ssh: mkdir session dir: %w", err)
|
||||
}
|
||||
return sessionDir, nil
|
||||
}
|
||||
|
||||
// startRemoteAgentctl launches an agentctl process on the remote bound to
|
||||
// 127.0.0.1:0 and waits for it to write its chosen port + PID into the session
|
||||
// runtime dir. Returns the chosen port and the process PID.
|
||||
//
|
||||
// The on-remote layout written by agentctl on startup:
|
||||
//
|
||||
// <sessionDir>/agentctl.port
|
||||
// <sessionDir>/agentctl.pid
|
||||
// <sessionDir>/agentctl.log
|
||||
func startRemoteAgentctl(
|
||||
ctx context.Context,
|
||||
client *ssh.Client,
|
||||
agentctlBin, workspacePath, sessionDir string,
|
||||
log *logger.Logger,
|
||||
) (port int, pid int, err error) {
|
||||
// Build the launch command. We:
|
||||
// - export the workspace
|
||||
// - launch agentctl as a backgrounded, disowned process
|
||||
// - have it bind to a free port; on startup we expect agentctl to either
|
||||
// accept --port-file / --pid-file flags OR we fall back to grabbing a
|
||||
// port ourselves and passing --port. To stay portable we always grab
|
||||
// a port ourselves here using `python3 -c` or `ss -tln` is brittle, so
|
||||
// we use a small inline shell helper.
|
||||
//
|
||||
// agentctl is invoked with KANDEV_AGENTCTL_PORT honored. We pick a port by
|
||||
// asking the kernel to allocate one (via `bash` reading from /dev/tcp is
|
||||
// unreliable for binding), so we use a tiny `python3` fallback if available,
|
||||
// else awk-on-ss. To keep the dependency surface minimal we instead just
|
||||
// rely on agentctl itself printing its bound port to a port-file when
|
||||
// invoked with KANDEV_AGENTCTL_PORT=0 — the standalone agentctl already
|
||||
// supports binding to :0 and printing the chosen port to stdout.
|
||||
cmd := fmt.Sprintf(
|
||||
`set -e
|
||||
mkdir -p %[1]s
|
||||
: > %[1]s/agentctl.log
|
||||
nohup %[2]s serve \
|
||||
--workspace %[3]s \
|
||||
--port 0 \
|
||||
--port-file %[1]s/agentctl.port \
|
||||
--pid-file %[1]s/agentctl.pid \
|
||||
>> %[1]s/agentctl.log 2>&1 < /dev/null &
|
||||
disown $! 2>/dev/null || true
|
||||
echo $!
|
||||
`,
|
||||
shellQuote(sessionDir),
|
||||
shellQuote(agentctlBin),
|
||||
shellQuote(workspacePath),
|
||||
)
|
||||
out, stderr, err := runSSHCommand(ctx, client, cmd)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("ssh: launch agentctl: %w (stderr: %s)", err, strings.TrimSpace(stderr))
|
||||
}
|
||||
_ = strings.TrimSpace(out) // immediate $! is informational; the on-disk pidfile is authoritative
|
||||
|
||||
// Poll for the port file + pid file to appear and be readable.
|
||||
deadline := time.Now().Add(sshAgentctlReadyTimeout)
|
||||
for time.Now().Before(deadline) {
|
||||
portStr, _, perr := runSSHCommand(ctx, client, "cat "+shellQuote(sessionDir+"/agentctl.port")+" 2>/dev/null")
|
||||
pidStr, _, _ := runSSHCommand(ctx, client, "cat "+shellQuote(sessionDir+"/agentctl.pid")+" 2>/dev/null")
|
||||
portStr = strings.TrimSpace(portStr)
|
||||
pidStr = strings.TrimSpace(pidStr)
|
||||
if portStr != "" && pidStr != "" && perr == nil {
|
||||
port, err = strconv.Atoi(portStr)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("ssh: agentctl wrote invalid port %q", portStr)
|
||||
}
|
||||
pid, err = strconv.Atoi(pidStr)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("ssh: agentctl wrote invalid pid %q", pidStr)
|
||||
}
|
||||
log.Info("agentctl started on remote",
|
||||
zap.Int("port", port),
|
||||
zap.Int("pid", pid),
|
||||
zap.String("session_dir", sessionDir))
|
||||
return port, pid, nil
|
||||
}
|
||||
time.Sleep(sshAgentctlReadyPoll)
|
||||
}
|
||||
tail, _, _ := runSSHCommand(ctx, client, "tail -n 50 "+shellQuote(sessionDir+"/agentctl.log")+" 2>/dev/null")
|
||||
return 0, 0, fmt.Errorf("ssh: agentctl did not become ready within %v; log tail:\n%s", sshAgentctlReadyTimeout, tail)
|
||||
}
|
||||
|
||||
// stopRemoteAgentctl best-effort kills a remote agentctl by PID and removes
|
||||
// the session runtime dir.
|
||||
func stopRemoteAgentctl(ctx context.Context, client *ssh.Client, sessionDir string, pid int) error {
|
||||
if pid > 0 {
|
||||
if _, _, err := runSSHCommand(ctx, client, fmt.Sprintf("kill %d 2>/dev/null || true", pid)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Leave the task dir intact (mirrors spec); only wipe session runtime.
|
||||
_, _, _ = runSSHCommand(ctx, client, "rm -rf "+shellQuote(sessionDir))
|
||||
return nil
|
||||
}
|
||||
|
||||
// isRemoteAgentctlAlive returns true when a kill -0 on the pid succeeds.
|
||||
func isRemoteAgentctlAlive(ctx context.Context, client *ssh.Client, pid int) bool {
|
||||
if pid <= 0 {
|
||||
return false
|
||||
}
|
||||
_, _, err := runSSHCommand(ctx, client, fmt.Sprintf("kill -0 %d", pid))
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// SSHPortForwarder fans out incoming local-port connections to a remote port
|
||||
// over the shared SSH connection using direct-tcpip channels. Each Forwarder
|
||||
// owns its local listener; closing the Forwarder closes the listener and any
|
||||
// outstanding channels.
|
||||
type SSHPortForwarder struct {
|
||||
listener net.Listener
|
||||
localPort int
|
||||
remotePort int
|
||||
logger *logger.Logger
|
||||
closed chan struct{}
|
||||
}
|
||||
|
||||
// StartPortForward opens a fresh 127.0.0.1:<random> listener and tunnels each
|
||||
// accept to the given remote port over client. Caller MUST call Close when the
|
||||
// session ends, otherwise both the listener and the SSH channels leak.
|
||||
func StartPortForward(client *ssh.Client, remotePort int, log *logger.Logger) (*SSHPortForwarder, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh: local listen: %w", err)
|
||||
}
|
||||
addr := listener.Addr().(*net.TCPAddr)
|
||||
fwd := &SSHPortForwarder{
|
||||
listener: listener,
|
||||
localPort: addr.Port,
|
||||
remotePort: remotePort,
|
||||
logger: log,
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
go fwd.serve(client)
|
||||
return fwd, nil
|
||||
}
|
||||
|
||||
func (f *SSHPortForwarder) serve(client *ssh.Client) {
|
||||
for {
|
||||
local, err := f.listener.Accept()
|
||||
if err != nil {
|
||||
select {
|
||||
case <-f.closed:
|
||||
return
|
||||
default:
|
||||
f.logger.Debug("ssh forwarder accept failed", zap.Error(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
go f.handleLocal(client, local)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *SSHPortForwarder) handleLocal(client *ssh.Client, local net.Conn) {
|
||||
defer func() { _ = local.Close() }()
|
||||
remote, err := client.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(f.remotePort)))
|
||||
if err != nil {
|
||||
f.logger.Warn("ssh forwarder dial remote failed",
|
||||
zap.Int("remote_port", f.remotePort),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
defer func() { _ = remote.Close() }()
|
||||
go func() { _, _ = io.Copy(remote, local) }()
|
||||
_, _ = io.Copy(local, remote)
|
||||
}
|
||||
|
||||
// LocalPort returns the local TCP port the forwarder is listening on.
|
||||
func (f *SSHPortForwarder) LocalPort() int { return f.localPort }
|
||||
|
||||
// Close terminates the forwarder. Idempotent.
|
||||
func (f *SSHPortForwarder) Close() error {
|
||||
select {
|
||||
case <-f.closed:
|
||||
return nil
|
||||
default:
|
||||
close(f.closed)
|
||||
}
|
||||
return f.listener.Close()
|
||||
}
|
||||
|
||||
// waitAgentctlHealthy polls http://127.0.0.1:<localPort>/health for up to
|
||||
// timeout. Used to confirm the forwarded tunnel is wired up after start /
|
||||
// recovery.
|
||||
func waitAgentctlHealthy(ctx context.Context, localPort int, timeout time.Duration) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
dialer := &net.Dialer{Timeout: 2 * time.Second}
|
||||
for time.Now().Before(deadline) {
|
||||
conn, err := dialer.DialContext(ctx, "tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(localPort)))
|
||||
if err == nil {
|
||||
_ = conn.Close()
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("agentctl on local port %d not reachable", localPort)
|
||||
}
|
||||
@@ -79,6 +79,14 @@ func buildLaunchMetadata(req *LaunchRequest, mainRepoGitDir, worktreeID, worktre
|
||||
for k, v := range req.Metadata {
|
||||
metadata[k] = v
|
||||
}
|
||||
// Merge executor record's Config (host, port, fingerprint, etc.) so
|
||||
// runtimes that need per-executor config (e.g. SSH) can read it from
|
||||
// metadata without re-querying the DB.
|
||||
for k, v := range req.ExecutorConfig {
|
||||
if _, exists := metadata[k]; !exists {
|
||||
metadata[k] = v
|
||||
}
|
||||
}
|
||||
if mainRepoGitDir != "" {
|
||||
metadata[MetadataKeyMainRepoGitDir] = mainRepoGitDir
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ const (
|
||||
RuntimeDocker Runtime = "docker"
|
||||
RuntimeRemoteDocker Runtime = "remote_docker"
|
||||
RuntimeSprites Runtime = "sprites"
|
||||
RuntimeSSH Runtime = "ssh"
|
||||
)
|
||||
|
||||
// IsContainerized reports whether the runtime hosts the agent
|
||||
|
||||
@@ -235,6 +235,9 @@ func (e *Executor) applyProfile(ctx context.Context, profileID string, cfg *exec
|
||||
if imageTag := profile.Config["image_tag"]; imageTag != "" {
|
||||
metadata["image_tag_override"] = imageTag
|
||||
}
|
||||
if workdirRoot := profile.Config["ssh_workdir_root"]; workdirRoot != "" {
|
||||
metadata["ssh_workdir_root"] = workdirRoot
|
||||
}
|
||||
}
|
||||
|
||||
// resolveProfileEnvVars resolves profile env vars, dereferencing secret IDs to their values.
|
||||
|
||||
@@ -501,6 +501,7 @@ const (
|
||||
ExecutorTypeLocalDocker ExecutorType = "local_docker"
|
||||
ExecutorTypeRemoteDocker ExecutorType = "remote_docker"
|
||||
ExecutorTypeSprites ExecutorType = "sprites"
|
||||
ExecutorTypeSSH ExecutorType = "ssh"
|
||||
ExecutorTypeMockRemote ExecutorType = "mock_remote"
|
||||
)
|
||||
|
||||
@@ -509,7 +510,7 @@ const (
|
||||
// These environments run shells inside the container/VM, not on the host.
|
||||
func IsRemoteExecutorType(t ExecutorType) bool {
|
||||
switch t {
|
||||
case ExecutorTypeSprites, ExecutorTypeRemoteDocker, ExecutorTypeLocalDocker, ExecutorTypeMockRemote:
|
||||
case ExecutorTypeSprites, ExecutorTypeRemoteDocker, ExecutorTypeLocalDocker, ExecutorTypeSSH, ExecutorTypeMockRemote:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
@@ -531,6 +532,8 @@ func (t ExecutorType) Runtime() agentruntime.Runtime {
|
||||
return agentruntime.RuntimeRemoteDocker
|
||||
case ExecutorTypeSprites:
|
||||
return agentruntime.RuntimeSprites
|
||||
case ExecutorTypeSSH:
|
||||
return agentruntime.RuntimeSSH
|
||||
default:
|
||||
return agentruntime.RuntimeStandalone
|
||||
}
|
||||
@@ -548,7 +551,7 @@ func IsContainerizedExecutorType(t ExecutorType) bool {
|
||||
// IsAlwaysResumableRuntime reports whether the given runtime represents
|
||||
// an executor that can always be resumed even without an explicit resume token.
|
||||
func IsAlwaysResumableRuntime(runtime agentruntime.Runtime) bool {
|
||||
return runtime == agentruntime.RuntimeSprites
|
||||
return runtime == agentruntime.RuntimeSprites || runtime == agentruntime.RuntimeSSH
|
||||
}
|
||||
|
||||
const (
|
||||
|
||||
Reference in New Issue
Block a user