diff --git a/apps/backend/cmd/kandev/agents.go b/apps/backend/cmd/kandev/agents.go index 7d558d278f..5d6185ace8 100644 --- a/apps/backend/cmd/kandev/agents.go +++ b/apps/backend/cmd/kandev/agents.go @@ -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 diff --git a/apps/backend/go.mod b/apps/backend/go.mod index 03b8eea687..6ef7956cdd 100644 --- a/apps/backend/go.mod +++ b/apps/backend/go.mod @@ -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 diff --git a/apps/backend/go.sum b/apps/backend/go.sum index 143b551178..41dbf8b8a9 100644 --- a/apps/backend/go.sum +++ b/apps/backend/go.sum @@ -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= diff --git a/apps/backend/internal/agent/executor/executor.go b/apps/backend/internal/agent/executor/executor.go index 3f904cc288..44b566c35c 100644 --- a/apps/backend/internal/agent/executor/executor.go +++ b/apps/backend/internal/agent/executor/executor.go @@ -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: diff --git a/apps/backend/internal/agent/runtime/lifecycle/env_preparer_ssh.go b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_ssh.go new file mode 100644 index 0000000000..1a768c9975 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_ssh.go @@ -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 +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go index a74338f455..8fa5449efc 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_backend.go @@ -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, diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go new file mode 100644 index 0000000000..b889f4dc1d --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh.go @@ -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 /tasks/ and /.kandev/sessions/. +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) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection.go new file mode 100644 index 0000000000..4db7cf5052 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection.go @@ -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:` 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) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection_test.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection_test.go new file mode 100644 index 0000000000..b832f02cd9 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_connection_test.go @@ -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) + } + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_credentials.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_credentials.go new file mode 100644 index 0000000000..a864d2b70c --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_credentials.go @@ -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) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_operations.go b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_operations.go new file mode 100644 index 0000000000..068822cd18 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/executor_ssh_operations.go @@ -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 / 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 /.kandev/sessions// 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: +// +// /agentctl.port +// /agentctl.pid +// /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: 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:/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) +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go b/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go index b5b53b15d7..9113f178dc 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go @@ -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 } diff --git a/apps/backend/internal/agentruntime/runtime.go b/apps/backend/internal/agentruntime/runtime.go index 7f3cafd88e..aa6fc1aef7 100644 --- a/apps/backend/internal/agentruntime/runtime.go +++ b/apps/backend/internal/agentruntime/runtime.go @@ -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 diff --git a/apps/backend/internal/orchestrator/executor/executor_state.go b/apps/backend/internal/orchestrator/executor/executor_state.go index f419a421a3..9498573be0 100644 --- a/apps/backend/internal/orchestrator/executor/executor_state.go +++ b/apps/backend/internal/orchestrator/executor/executor_state.go @@ -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. diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index f0fb341662..f7756ddd71 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -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 (