DockMV/internal/sshx/dialstdio.go

package sshx

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"strings"
	"sync"
	"time"

	"golang.org/x/crypto/ssh"
)

// DialAPI opens one connection to the remote daemon's API by running
// `docker system dial-stdio` over SSH and treating that session's stdin and
// stdout as a socket. It is the same mechanism `docker -H ssh://…` uses, so the
// remote host still needs nothing but sshd and the docker CLI.
//
// The returned connection is what dkr.NewTunnel dials through: from there on the
// whole Docker Engine API — inventory, archive streams, image save — is
// available on a remote source host exactly as it is on a local one.
func (r *RemoteDocker) DialAPI(ctx context.Context) (net.Conn, error) {
	sess, err := r.c.conn.NewSession()
	if err != nil {
		return nil, fmt.Errorf("open ssh session: %w", err)
	}
	stdin, err := sess.StdinPipe()
	if err != nil {
		sess.Close()
		return nil, fmt.Errorf("attach to remote stdin: %w", err)
	}
	stdout, err := sess.StdoutPipe()
	if err != nil {
		sess.Close()
		return nil, fmt.Errorf("attach to remote stdout: %w", err)
	}
	errBuf := &syncBuffer{}
	sess.Stderr = errBuf

	cmd := r.Cmd("system", "dial-stdio")
	if err := sess.Start(cmd); err != nil {
		sess.Close()
		return nil, fmt.Errorf("start %q: %w", cmd, err)
	}

	cfg := r.c.Config()
	// The connection deliberately outlives ctx: the HTTP transport keeps it in
	// its idle pool between API calls, and closes it itself when a request is
	// cancelled or the client is closed.
	return &apiConn{
		sess: sess, stdin: stdin, stdout: stdout, stderr: errBuf,
		remote: apiAddr(fmt.Sprintf("%s@%s", cfg.User, cfg.addr())),
	}, nil
}

// ProbeCLI checks that the remote docker CLI is usable before the API is
// tunnelled through it, so a missing binary or a permission problem is reported
// as itself rather than as a broken socket. It returns the daemon version.
func (r *RemoteDocker) ProbeCLI(ctx context.Context) (string, error) {
	out, ok, err := r.Try(ctx, "version", "--format", "{{.Server.Version}}")
	if err != nil {
		return "", err
	}
	if ok {
		if v := strings.TrimSpace(out); v != "" {
			return v, nil
		}
	}
	res, err := r.c.Run(ctx, r.Cmd("version"))
	if err != nil {
		return "", err
	}
	msg := strings.TrimSpace(res.Stderr)
	switch {
	case strings.Contains(msg, "permission denied"):
		return "", errors.New("the login user cannot talk to the docker daemon; " +
			"add it to the docker group, or enable sudo -n for this source")
	case msg != "":
		return "", errors.New("docker is not usable on that host: " + firstLine(msg))
	default:
		return "", errors.New("docker is not installed or not on PATH on that host")
	}
}

// apiConn adapts an SSH session to net.Conn.
type apiConn struct {
	sess   *ssh.Session
	stdin  io.WriteCloser
	stdout io.Reader
	stderr *syncBuffer
	remote apiAddr

	once sync.Once
	err  error
}

func (c *apiConn) Read(p []byte) (int, error) {
	n, err := c.stdout.Read(p)
	if err != nil {
		return n, c.wrap(err)
	}
	return n, nil
}

func (c *apiConn) Write(p []byte) (int, error) {
	n, err := c.stdin.Write(p)
	if err != nil {
		return n, c.wrap(err)
	}
	return n, nil
}

// wrap replaces the bare EOF a failed remote command produces with whatever it
// printed on stderr, which is the only place the reason appears.
func (c *apiConn) wrap(err error) error {
	if msg := strings.TrimSpace(c.stderr.String()); msg != "" {
		return fmt.Errorf("docker system dial-stdio on the remote host failed: %s", firstLine(msg))
	}
	return err
}

func (c *apiConn) Close() error {
	c.once.Do(func() {
		// Closing stdin lets the remote docker exit cleanly; the session is torn
		// down straight after either way.
		_ = c.stdin.Close()
		c.err = c.sess.Close()
		if errors.Is(c.err, io.EOF) {
			c.err = nil
		}
	})
	return c.err
}

func (c *apiConn) LocalAddr() net.Addr  { return apiAddr("dockmv") }
func (c *apiConn) RemoteAddr() net.Addr { return c.remote }

// The deadline calls are no-ops: an SSH channel has no deadline of its own, and
// the Docker client relies on context cancellation rather than on these. This
// mirrors what the docker CLI's own ssh:// transport does.
func (c *apiConn) SetDeadline(time.Time) error      { return nil }
func (c *apiConn) SetReadDeadline(time.Time) error  { return nil }
func (c *apiConn) SetWriteDeadline(time.Time) error { return nil }

type apiAddr string

func (a apiAddr) Network() string { return "ssh" }
func (a apiAddr) String() string  { return string(a) }

// syncBuffer collects remote stderr, which the ssh session writes from its own
// goroutine while the connection is being read.
type syncBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *syncBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *syncBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.String()
}