mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
remote: fall back to scripts when the server refuses the sftp subsystem
pkg/sftp reports the refusal as an unnamed error string, so the subsystem is now requested in sftpfs.New and a server reply of 'failure' becomes ErrSubsystemRefused. Client.Volume prefers SFTP and, only on that sentinel, serves the connection from execfs over the exec channel and remembers the answer until the connection is replaced. Any other failure (dead link, refused channel) is returned unchanged. The exec runner quotes the script and every argument whole into one /bin/sh command line.
This commit is contained in:
1 parent
5bbad0c76b
commit
4139ddb27f
5 files changed
+173
-33
No files matched your search
@@ -612,7 +612,7 @@ func remoteFSLs(operand string) int {
|
||||
return 2
|
||||
}
|
||||
return withRemoteFS(host, func(ctx context.Context, client *remote.Client) int {
|
||||
fsys, err := client.SFTP()
|
||||
fsys, err := client.Volume()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, i18n.M.ErrorPrefix, err)
|
||||
return 1
|
||||
@@ -648,7 +648,7 @@ func remoteFSGet(args []string) int {
|
||||
localPath = args[1]
|
||||
}
|
||||
return withRemoteFS(host, func(ctx context.Context, client *remote.Client) int {
|
||||
fsys, err := client.SFTP()
|
||||
fsys, err := client.Volume()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, i18n.M.ErrorPrefix, err)
|
||||
return 1
|
||||
@@ -692,7 +692,7 @@ func remoteFSPut(args []string) int {
|
||||
}
|
||||
defer in.Close()
|
||||
return withRemoteFS(host, func(ctx context.Context, client *remote.Client) int {
|
||||
fsys, err := client.SFTP()
|
||||
fsys, err := client.Volume()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, i18n.M.ErrorPrefix, err)
|
||||
return 1
|
||||
|
||||
@@ -43,17 +43,9 @@ type sshLink struct{ *remote.Client }
|
||||
func (l sshLink) Forwards() Forwards { return l.Client.Forwards() }
|
||||
|
||||
func (l sshLink) Dirs(ctx context.Context, dir string) (string, []sftpfs.Entry, error) {
|
||||
fsys, err := l.SFTP()
|
||||
vol, err := l.Volume()
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
// The far machine resolves it: ~ belongs to the account that logged in, and
|
||||
// a symlinked project must be listed under the path it really has, or
|
||||
// stepping up from it would leave the tree it was reached through.
|
||||
at, err := fsys.RealPath(ctx, dir)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
entries, err := fsys.List(ctx, at)
|
||||
return at, entries, err
|
||||
return sftpfs.Dirs(ctx, vol, dir)
|
||||
}
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
||||
"reasonix/internal/base/netclient"
|
||||
"reasonix/internal/platform/remote/execfs"
|
||||
"reasonix/internal/platform/remote/forward"
|
||||
"reasonix/internal/platform/remote/sftpfs"
|
||||
)
|
||||
@@ -41,7 +42,7 @@ type JumpHostOptions struct {
|
||||
|
||||
// Client is a supervised SSH connection: it dials, verifies the host key,
|
||||
// attaches forwards, keeps the link alive, and reconnects with backoff. It
|
||||
// shares one SFTP session and one forward.Set across its lifetime.
|
||||
// shares one file layer and one forward.Set across its lifetime.
|
||||
type Client struct {
|
||||
opts Options
|
||||
clock Clock
|
||||
@@ -52,8 +53,8 @@ type Client struct {
|
||||
mu sync.Mutex
|
||||
ssh *ssh.Client
|
||||
hops []*ssh.Client
|
||||
sftp *sftpfs.FS
|
||||
generation uint64 // bumps on every (re)connect; SFTP handles carry it
|
||||
volume sftpfs.Volume // SFTP, or scripts when the server refuses the subsystem
|
||||
generation uint64 // bumps on every (re)connect; file handles carry it
|
||||
status Status
|
||||
closed bool
|
||||
hopHosts map[string]ResolvedHost
|
||||
@@ -173,33 +174,48 @@ func (c *Client) SSH() (*ssh.Client, error) {
|
||||
return c.ssh, nil
|
||||
}
|
||||
|
||||
// SFTP returns the shared SFTP handle, lazily created for the current
|
||||
// connection. Handles obtained before a reconnect return ErrDisconnected on
|
||||
// use; re-fetch via SFTP.
|
||||
func (c *Client) SFTP() (*sftpfs.FS, error) {
|
||||
// Volume returns the connection's file layer, created on first use. SFTP is
|
||||
// preferred; a server that answers the subsystem request with a refusal gets
|
||||
// the script-based layer instead, and the connection keeps that answer until
|
||||
// it is replaced. Any other failure to open SFTP (a dead link, a refused
|
||||
// channel) is returned as it is: the scripts would ride the same channel.
|
||||
// Handles obtained before a reconnect return ErrDisconnected on use; re-fetch.
|
||||
func (c *Client) Volume() (sftpfs.Volume, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.ssh == nil {
|
||||
return nil, ErrNotConnected
|
||||
}
|
||||
if c.sftp != nil {
|
||||
return c.sftp, nil
|
||||
if c.volume != nil {
|
||||
return c.volume, nil
|
||||
}
|
||||
fsys, err := sftpfs.New(c.ssh)
|
||||
if err != nil {
|
||||
switch {
|
||||
case err == nil:
|
||||
c.volume = fsys
|
||||
case errors.Is(err, sftpfs.ErrSubsystemRefused):
|
||||
c.volume = execfs.New(scriptRunner{c})
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
c.sftp = fsys
|
||||
return c.sftp, nil
|
||||
return c.volume, nil
|
||||
}
|
||||
|
||||
// Files is the SFTP handle as the file layer a kernel bootstrap needs.
|
||||
// Files is the connection's file layer as a kernel bootstrap needs it.
|
||||
func (c *Client) Files() (sftpfs.Files, error) {
|
||||
fsys, err := c.SFTP()
|
||||
v, err := c.Volume()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fsys, nil
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// scriptRunner runs execfs scripts over the connection's exec channel.
|
||||
type scriptRunner struct{ c *Client }
|
||||
|
||||
func (r scriptRunner) Script(ctx context.Context, stdin io.Reader, body string, args ...string) (execfs.Result, error) {
|
||||
res, err := r.c.execIn(ctx, stdin, execfs.CommandLine(body, args...))
|
||||
return execfs.Result{Stdout: res.Stdout, Stderr: res.Stderr, ExitCode: res.ExitCode}, err
|
||||
}
|
||||
|
||||
// ExecResult is the outcome of a one-shot remote command.
|
||||
@@ -274,6 +290,11 @@ func openExecSession(ctx context.Context, cl io.Closer, open execSessionOpener)
|
||||
// opening, the shared client is closed to unblock NewSession; the supervisor
|
||||
// reconnects it.
|
||||
func (c *Client) Exec(ctx context.Context, cmd string) (ExecResult, error) {
|
||||
return c.execIn(ctx, nil, cmd)
|
||||
}
|
||||
|
||||
// execIn is Exec with stdin fed to the command; nil leaves it empty.
|
||||
func (c *Client) execIn(ctx context.Context, stdin io.Reader, cmd string) (ExecResult, error) {
|
||||
cl, err := c.SSH()
|
||||
if err != nil {
|
||||
return ExecResult{}, err
|
||||
@@ -293,6 +314,9 @@ func (c *Client) Exec(ctx context.Context, cmd string) (ExecResult, error) {
|
||||
_ = sess.Close()
|
||||
return ExecResult{}, ctxErr
|
||||
}
|
||||
if in, ok := sess.(interface{ feed(io.Reader) }); ok && stdin != nil {
|
||||
in.feed(stdin)
|
||||
}
|
||||
return runExec(ctx, sess, cmd)
|
||||
}
|
||||
|
||||
@@ -305,6 +329,8 @@ type execSession interface {
|
||||
|
||||
type sshExecSession struct{ *ssh.Session }
|
||||
|
||||
func (s *sshExecSession) feed(r io.Reader) { s.Stdin = r }
|
||||
|
||||
func (s *sshExecSession) capture(stdout, stderr *bytes.Buffer) {
|
||||
s.Stdout, s.Stderr = stdout, stderr
|
||||
}
|
||||
@@ -534,18 +560,18 @@ func (c *Client) installConn(cl *ssh.Client, hops []*ssh.Client) {
|
||||
c.mu.Lock()
|
||||
c.ssh = cl
|
||||
c.hops = hops
|
||||
c.sftp = nil
|
||||
c.volume = nil
|
||||
c.generation++
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
func (c *Client) teardownConn() {
|
||||
c.mu.Lock()
|
||||
sftpH, cl, hops := c.sftp, c.ssh, c.hops
|
||||
c.sftp, c.ssh, c.hops = nil, nil, nil
|
||||
vol, cl, hops := c.volume, c.ssh, c.hops
|
||||
c.volume, c.ssh, c.hops = nil, nil, nil
|
||||
c.mu.Unlock()
|
||||
if sftpH != nil {
|
||||
_ = sftpH.Close()
|
||||
if vol != nil {
|
||||
_ = vol.Close()
|
||||
}
|
||||
if cl != nil {
|
||||
_ = cl.Close()
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package remote
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os/exec"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/platform/remote/execfs/fstest"
|
||||
"reasonix/internal/platform/remote/sftpfs"
|
||||
"reasonix/internal/platform/remote/sshtest"
|
||||
)
|
||||
|
||||
func startedClient(t *testing.T, srv *sshtest.Server) *Client {
|
||||
t.Helper()
|
||||
c := newTestClient(t, srv, Options{Auth: AuthOptions{
|
||||
DisableAgent: true,
|
||||
Password: func() (string, error) { return "pw", nil },
|
||||
}})
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
if err := c.Start(ctx); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
return c
|
||||
}
|
||||
|
||||
func TestAServerThatServesSFTPKeepsTheSFTPLayer(t *testing.T) {
|
||||
c := startedClient(t, sshtest.Start(t, sshtest.Options{Password: "pw", SFTPRoot: t.TempDir()}))
|
||||
v, err := c.Volume()
|
||||
if err != nil || v.Mode() != sftpfs.ModeSFTP {
|
||||
t.Fatalf("Volume = %v, %v; want sftp", v, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAServerThatRefusesTheSubsystemGetsTheScriptLayerAndKeepsIt(t *testing.T) {
|
||||
c := startedClient(t, sshtest.Start(t, sshtest.Options{Password: "pw"}))
|
||||
v, err := c.Volume()
|
||||
if err != nil {
|
||||
t.Fatalf("Volume: %v", err)
|
||||
}
|
||||
if v.Mode() != sftpfs.ModeExec {
|
||||
t.Fatalf("mode = %q, want exec", v.Mode())
|
||||
}
|
||||
again, err := c.Volume()
|
||||
if err != nil || again != v {
|
||||
t.Fatalf("a second call = %v, %v; want the layer the first one chose, not another probe", again, err)
|
||||
}
|
||||
files, err := c.Files()
|
||||
if err != nil || sftpfs.ModeOf(files) != sftpfs.ModeExec {
|
||||
t.Fatalf("Files = %v, %v; bootstrap must get the same layer", files, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAReconnectAsksTheServerAgain(t *testing.T) {
|
||||
c := startedClient(t, sshtest.Start(t, sshtest.Options{Password: "pw"}))
|
||||
if _, err := c.Volume(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.mu.Lock()
|
||||
cl := c.ssh
|
||||
c.mu.Unlock()
|
||||
c.installConn(cl, nil)
|
||||
c.mu.Lock()
|
||||
remembered := c.volume
|
||||
c.mu.Unlock()
|
||||
if remembered != nil {
|
||||
t.Fatal("a new connection kept the old connection's answer")
|
||||
}
|
||||
}
|
||||
|
||||
func TestADeadLinkIsNotMistakenForARefusal(t *testing.T) {
|
||||
c := startedClient(t, sshtest.Start(t, sshtest.Options{Password: "pw"}))
|
||||
cl, err := c.SSH()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = cl.Close()
|
||||
if v, err := c.Volume(); err == nil {
|
||||
t.Fatalf("Volume on a closed link = %v, want its error", v)
|
||||
}
|
||||
c.mu.Lock()
|
||||
remembered := c.volume
|
||||
c.mu.Unlock()
|
||||
if remembered != nil {
|
||||
t.Fatal("a failure that was not a refusal was remembered as one")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheScriptLayerThroughARealShellOverSSH(t *testing.T) {
|
||||
if err := exec.Command("stat", "-c", "%s", "/").Run(); err != nil {
|
||||
t.Skip("needs a GNU userland")
|
||||
}
|
||||
c := startedClient(t, sshtest.Start(t, sshtest.Options{Password: "pw", Shell: true}))
|
||||
v, err := c.Volume()
|
||||
if err != nil || v.Mode() != sftpfs.ModeExec {
|
||||
t.Fatalf("Volume = %v, %v", v, err)
|
||||
}
|
||||
fstest.RunScenario(t, v, t.TempDir())
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
@@ -26,6 +27,7 @@ type Server struct {
|
||||
execFunc func(cmd string) (stdout string, stderr string, exit int)
|
||||
sftpRoot string
|
||||
enableSFT bool
|
||||
shell bool
|
||||
|
||||
mu sync.Mutex
|
||||
conns []net.Conn
|
||||
@@ -49,6 +51,9 @@ type Options struct {
|
||||
Exec func(cmd string) (stdout string, stderr string, exit int)
|
||||
// SFTPRoot enables the SFTP subsystem rooted at this directory.
|
||||
SFTPRoot string
|
||||
// Shell runs each exec request under this machine's /bin/sh, with the
|
||||
// channel as its stdin, stdout and stderr, instead of answering from Exec.
|
||||
Shell bool
|
||||
}
|
||||
|
||||
// Start launches a server on 127.0.0.1:0.
|
||||
@@ -102,6 +107,7 @@ func Start(t *testing.T, opts Options) *Server {
|
||||
execFunc: opts.Exec,
|
||||
sftpRoot: opts.SFTPRoot,
|
||||
enableSFT: opts.SFTPRoot != "",
|
||||
shell: opts.Shell,
|
||||
}
|
||||
s.wg.Add(1)
|
||||
go s.serve()
|
||||
@@ -232,6 +238,21 @@ func (s *Server) handleSession(newCh ssh.NewChannel) {
|
||||
}
|
||||
|
||||
func (s *Server) runExec(ch ssh.Channel, cmd string) {
|
||||
if s.shell {
|
||||
c := exec.Command("/bin/sh", "-c", cmd)
|
||||
c.Stdin, c.Stdout, c.Stderr = ch, ch, ch.Stderr()
|
||||
code := 0
|
||||
if err := c.Run(); err != nil {
|
||||
var ee *exec.ExitError
|
||||
if !errors.As(err, &ee) {
|
||||
code = 127
|
||||
} else {
|
||||
code = ee.ExitCode()
|
||||
}
|
||||
}
|
||||
sendExitStatus(ch, code)
|
||||
return
|
||||
}
|
||||
stdout, stderr, exit := "", "", 0
|
||||
if s.execFunc != nil {
|
||||
stdout, stderr, exit = s.execFunc(cmd)
|
||||
|
||||
Reference in new issue
Block a user