feat: ssh connection to remote docker daemon (#594)

Signed-off-by: Matej Vasek <mvasek@redhat.com>

buildpacks/builder.go

@@ -14,8 +14,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/Masterminds/semver"
+	fn "knative.dev/kn-plugin-func"
+	"knative.dev/kn-plugin-func/docker"
 
+	"github.com/Masterminds/semver"
+	"github.com/buildpacks/pack"
+	"github.com/buildpacks/pack/logging"
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/events"
@@ -25,14 +29,8 @@ import (
 	"github.com/docker/docker/api/types/registry"
 	"github.com/docker/docker/api/types/swarm"
 	"github.com/docker/docker/api/types/volume"
-	v1 "github.com/opencontainers/image-spec/specs-go/v1"
-
-	"github.com/buildpacks/pack"
-	"github.com/buildpacks/pack/logging"
-
 	dockerClient "github.com/docker/docker/client"
-
-	fn "knative.dev/kn-plugin-func"
+	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
 //Builder holds the configuration that will be passed to
@@ -95,16 +93,12 @@ func (builder *Builder) Build(ctx context.Context, f fn.Function) (err error) {
 		logWriter = &bytes.Buffer{}
 	}
 
-	// Client with a logger which is enabled if in Verbose mode.
-	dockerClient, err := dockerClient.NewClientWithOpts(
-		dockerClient.FromEnv,
-		dockerClient.WithVersion("1.38"),
-	)
+	cli, dockerHost, err := docker.NewDockerClient()
 	if err != nil {
 		return err
 	}
 
-	version, err := dockerClient.ServerVersion(ctx)
+	version, err := cli.ServerVersion(ctx)
 	if err != nil {
 		return err
 	}
@@ -141,14 +135,15 @@ func (builder *Builder) Build(ctx context.Context, f fn.Function) (err error) {
 		TrustBuilder: !daemonIsPodmanBeforeV330 &&
 			(strings.HasPrefix(packBuilder, "quay.io/boson") ||
 				strings.HasPrefix(packBuilder, "gcr.io/paketo-buildpacks")),
-		DockerHost: os.Getenv("DOCKER_HOST"),
+		DockerHost: dockerHost,
 		ContainerConfig: struct {
 			Network string
 			Volumes []string
 		}{Network: network, Volumes: nil},
 	}
 
-	dockerClientWrapper := &clientWrapper{dockerClient}
+	dockerClientWrapper := &clientWrapper{cli}
+	// Client with a logger which is enabled if in Verbose mode.
 	packClient, err := pack.NewClient(pack.WithLogger(logging.New(logWriter)), pack.WithDockerClient(dockerClientWrapper))
 	if err != nil {
 		return

docker/docker_client.go

@@ -0,0 +1,49 @@
+package docker
+
+import (
+	"net/http"
+	"net/url"
+	"os"
+
+	"knative.dev/kn-plugin-func/ssh"
+
+	"github.com/docker/docker/client"
+)
+
+func NewDockerClient() (dockerClient client.CommonAPIClient, dockerHost string, err error) {
+	dockerHost = os.Getenv("DOCKER_HOST")
+	_url, err := url.Parse(dockerHost)
+	isSSH := err == nil && _url.Scheme == "ssh"
+
+	if !isSSH {
+		dockerClient, err = client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+		return
+	}
+
+	credentialsConfig := ssh.Config{
+		Identity:           os.Getenv("DOCKER_HOST_SSH_IDENTITY"),
+		PassPhrase:         os.Getenv("DOCKER_HOST_SSH_IDENTITY_PASSPHRASE"),
+		PasswordCallback:   ssh.NewPasswordCbk(),
+		PassPhraseCallback: ssh.NewPassPhraseCbk(),
+		HostKeyCallback:    ssh.NewHostKeyCbk(),
+	}
+	dialContext, dockerHost, err := ssh.NewDialContext(_url, credentialsConfig)
+	if err != nil {
+		return
+	}
+
+	httpClient := &http.Client{
+		// No tls
+		// No proxy
+		Transport: &http.Transport{
+			DialContext: dialContext,
+		},
+	}
+
+	dockerClient, err = client.NewClientWithOpts(
+		client.WithAPIVersionNegotiation(),
+		client.WithHTTPClient(httpClient),
+		client.WithHost("http://placeholder/"))
+
+	return dockerClient, dockerHost, err
+}

docker/pusher.go

@@ -18,7 +18,6 @@ import (
 	"github.com/containers/image/v5/pkg/docker/config"
 	containersTypes "github.com/containers/image/v5/types"
 	"github.com/docker/docker/api/types"
-	"github.com/docker/docker/client"
 	"github.com/docker/docker/errdefs"
 )
 
@@ -40,7 +39,7 @@ var ErrUnauthorized = errors.New("bad credentials")
 type VerifyCredentialsCallback func(ctx context.Context, username, password, registry string) error
 
 func CheckAuth(ctx context.Context, username, password, registry string) error {
-	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	cli, _, err := NewDockerClient()
 	if err != nil {
 		return err
 	}
@@ -243,7 +242,7 @@ func (n *Pusher) Push(ctx context.Context, f fn.Function) (digest string, err er
 		return "", err
 	}
 
-	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	cli, _, err := NewDockerClient()
 	if err != nil {
 		return "", fmt.Errorf("failed to create docker api client: %w", err)
 	}

docker/runner.go

@@ -14,8 +14,6 @@ import (
 	"github.com/docker/go-connections/nat"
 	"github.com/pkg/errors"
 
-	"github.com/docker/docker/client"
-
 	fn "knative.dev/kn-plugin-func"
 )
 
@@ -35,7 +33,7 @@ func (n *Runner) Run(ctx context.Context, f fn.Function) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	cli, _, err := NewDockerClient()
 	if err != nil {
 		return errors.Wrap(err, "failed to create docker api client")
 	}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/buildpacks/pack v0.21.1
 	github.com/cloudevents/sdk-go/v2 v2.5.0
 	github.com/containers/image/v5 v5.10.6
+	github.com/docker/cli v20.10.7+incompatible
 	github.com/docker/docker v20.10.8+incompatible
 	github.com/docker/docker-credential-helpers v0.6.4
 	github.com/docker/go-connections v0.4.0
@@ -24,9 +25,11 @@ require (
 	github.com/ory/viper v1.7.5
 	github.com/pkg/errors v0.9.1
 	github.com/spf13/cobra v1.2.1
+	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a
 	golang.org/x/mod v0.5.1 // indirect
 	golang.org/x/net v0.0.0-20210929193557-e81a3d93ecf6 // indirect
 	golang.org/x/sys v0.0.0-20211002104244-808efd93c36d // indirect
+	golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.7 // indirect
 	gopkg.in/yaml.v2 v2.4.0

ssh/ssh_dialer.go

@@ -0,0 +1,427 @@
+// NOTE: this code is based on "github.com/containers/podman/v3/pkg/bindings"
+
+package ssh
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net"
+	urlPkg "net/url"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/docker/cli/cli/connhelper"
+	"github.com/docker/docker/pkg/homedir"
+	"golang.org/x/crypto/ssh"
+	"golang.org/x/crypto/ssh/agent"
+	"golang.org/x/crypto/ssh/knownhosts"
+)
+
+type PasswordCallback func() (string, error)
+type PassPhraseCallback func() (string, error)
+type HostKeyCallback func(hostPort string, pubKey ssh.PublicKey) error
+
+type Config struct {
+	Identity           string
+	PassPhrase         string
+	PasswordCallback   PasswordCallback
+	PassPhraseCallback PassPhraseCallback
+	HostKeyCallback    HostKeyCallback
+}
+
+type DialContextFn = func(ctx context.Context, network, addr string) (net.Conn, error)
+
+func NewDialContext(url *urlPkg.URL, config Config) (DialContextFn, string, error) {
+	sshConfig, err := NewSSHClientConfig(url, config)
+	if err != nil {
+		return nil, "", err
+	}
+
+	port := url.Port()
+	if port == "" {
+		port = "22"
+	}
+	host := url.Hostname()
+
+	sshClient, err := ssh.Dial("tcp", net.JoinHostPort(host, port), sshConfig)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to dial ssh: %w", err)
+	}
+	defer func() {
+		if sshClient != nil {
+			sshClient.Close()
+		}
+	}()
+
+	var remoteDockerHost string
+	if url.Path != "" {
+		remoteDockerHost = fmt.Sprintf(`unix://%s`, url.Path)
+	} else {
+		remoteDockerHost, err = getRemoteDockerHost(sshClient)
+		if err != nil {
+			return nil, "", err
+		}
+	}
+
+	network, addr, err := getNetworkAndAddress(remoteDockerHost)
+	if err != nil {
+		return nil, "", err
+	}
+
+	var dialContext DialContextFn
+
+	if network == "npipe" {
+		// ssh tunneling doesn't support tunneling of Windows' named pipes
+		dialContext, err = stdioDialContext(url, sshClient, config.Identity)
+		return dialContext, remoteDockerHost, err
+	}
+
+	d := dialer{sshClient: sshClient, addr: addr, network: network}
+	sshClient = nil
+	dialContext = d.DialContext
+
+	runtime.SetFinalizer(&d, func(d *dialer) {
+		d.Close()
+	})
+
+	return dialContext, remoteDockerHost, nil
+}
+
+type dialer struct {
+	sshClient *ssh.Client
+	network   string
+	addr      string
+}
+
+func (d *dialer) DialContext(ctx context.Context, n, a string) (net.Conn, error) {
+	conn, err := d.Dial(d.network, d.addr)
+	if err != nil {
+		return nil, err
+	}
+	go func() {
+		if ctx != nil {
+			<-ctx.Done()
+			conn.Close()
+		}
+	}()
+	return conn, nil
+}
+
+func (d *dialer) Dial(n, a string) (net.Conn, error) {
+	return d.sshClient.Dial(d.network, d.addr)
+}
+
+func (d *dialer) Close() error {
+	return d.sshClient.Close()
+}
+
+func isWindowsMachine(sshClient *ssh.Client) (bool, error) {
+	session, err := sshClient.NewSession()
+	if err != nil {
+		return false, err
+	}
+	defer session.Close()
+
+	out, err := session.CombinedOutput("systeminfo")
+	if err == nil && strings.Contains(string(out), "Windows") {
+		return true, nil
+	}
+	return false, nil
+}
+
+func getRemoteDockerHost(sshClient *ssh.Client) (remoteDockerHost string, err error) {
+	session, err := sshClient.NewSession()
+	if err != nil {
+		return
+	}
+	defer session.Close()
+
+	out, err := session.CombinedOutput("set")
+	if err != nil {
+		return
+	}
+
+	remoteDockerHost = "unix:///var/run/docker.sock"
+	isWin, err := isWindowsMachine(sshClient)
+	if err != nil {
+		return
+	}
+
+	if isWin {
+		remoteDockerHost = "npipe:////./pipe/docker_engine"
+	}
+
+	scanner := bufio.NewScanner(bytes.NewBuffer(out))
+	for scanner.Scan() {
+		if strings.HasPrefix(scanner.Text(), "DOCKER_HOST=") {
+			parts := strings.SplitN(scanner.Text(), "=", 2)
+			remoteDockerHost = strings.Trim(parts[1], `"'`)
+			break
+		}
+	}
+
+	return remoteDockerHost, err
+}
+
+func getNetworkAndAddress(remoteDockerHost string) (network string, addr string, err error) {
+	remoteDockerHostURL, err := urlPkg.Parse(remoteDockerHost)
+	if err != nil {
+		return
+	}
+	switch remoteDockerHostURL.Scheme {
+	case "unix", "npipe":
+		addr = remoteDockerHostURL.Path
+	case "fd":
+		remoteDockerHostURL.Scheme = "tcp" // don't know why it works that way
+		fallthrough
+	case "tcp":
+		addr = remoteDockerHostURL.Host
+	default:
+		return "", "", errors.New("scheme is not supported")
+	}
+	network = remoteDockerHostURL.Scheme
+
+	return network, addr, err
+}
+
+func stdioDialContext(url *urlPkg.URL, sshClient *ssh.Client, identity string) (DialContextFn, error) {
+	session, err := sshClient.NewSession()
+	if err != nil {
+		return nil, err
+	}
+	defer session.Close()
+
+	out, err := session.CombinedOutput("docker system dial-stdio --help")
+	if err != nil {
+		return nil, fmt.Errorf("cannot use dial-stdio: %w (%q)", err, out)
+	}
+
+	var opts []string
+	if identity != "" {
+		opts = append(opts, "-i", identity)
+	}
+
+	connHelper, err := connhelper.GetConnectionHelperWithSSHOpts(url.String(), opts)
+	if err != nil {
+		return nil, err
+	}
+
+	return connHelper.Dialer, nil
+}
+
+// Default key names.
+var knownKeyNames = []string{"id_rsa", "id_dsa", "id_ecdsa", "id_ecdsa_sk", "id_ed25519", "id_ed25519_sk"}
+
+func NewSSHClientConfig(url *urlPkg.URL, credentialsConfig Config) (*ssh.ClientConfig, error) {
+	var (
+		authMethods []ssh.AuthMethod
+		signers     []ssh.Signer
+		err         error
+	)
+
+	if pw, found := url.User.Password(); found {
+		authMethods = append(authMethods, ssh.Password(pw))
+	}
+
+	// add signer from explicit identity parameter
+	if credentialsConfig.Identity != "" {
+		s, err := publicKey(credentialsConfig.Identity, []byte(credentialsConfig.Identity), credentialsConfig.PassPhraseCallback)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse identity file: %w", err)
+		}
+		signers = append(signers, s)
+	}
+
+	// add signers from ssh-agent
+	if sock, found := os.LookupEnv("SSH_AUTH_SOCK"); found && sock != "" {
+		var agentSigners []ssh.Signer
+		var agentConn net.Conn
+		agentConn, err = net.Dial("unix", sock)
+		if err != nil {
+			return nil, fmt.Errorf("failed to connect to ssh-agent's socket: %w", err)
+		}
+		agentSigners, err = agent.NewClient(agentConn).Signers()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get signers from ssh-agent: %w", err)
+		}
+		signers = append(signers, agentSigners...)
+	}
+
+	// if there is no explicit identity file nor keys from ssh-agent then
+	// add keys with standard name from ~/.ssh/
+	if len(signers) == 0 {
+		var defaultKeyPaths []string
+		if home, err := os.UserHomeDir(); err == nil {
+			for _, keyName := range knownKeyNames {
+				p := filepath.Join(home, ".ssh", keyName)
+
+				fi, err := os.Stat(p)
+				if err != nil {
+					continue
+				}
+				if fi.Mode().IsRegular() {
+					defaultKeyPaths = append(defaultKeyPaths, p)
+				}
+			}
+		}
+
+		if len(defaultKeyPaths) == 1 {
+			s, err := publicKey(defaultKeyPaths[0], []byte(credentialsConfig.PassPhrase), credentialsConfig.PassPhraseCallback)
+			if err != nil {
+				return nil, err
+			}
+			signers = append(signers, s)
+		}
+	}
+
+	if len(signers) > 0 {
+		var dedup = make(map[string]ssh.Signer)
+		// Dedup signers based on fingerprint, ssh-agent keys override explicit identity
+		for _, s := range signers {
+			fp := ssh.FingerprintSHA256(s.PublicKey())
+			//if _, found := dedup[fp]; found {
+			//	key updated
+			//}
+			dedup[fp] = s
+		}
+
+		var uniq []ssh.Signer
+		for _, s := range dedup {
+			uniq = append(uniq, s)
+		}
+		authMethods = append(authMethods, ssh.PublicKeysCallback(func() ([]ssh.Signer, error) {
+			return uniq, nil
+		}))
+	}
+
+	if len(authMethods) == 0 && credentialsConfig.PasswordCallback != nil {
+		authMethods = append(authMethods, ssh.PasswordCallback(credentialsConfig.PasswordCallback))
+	}
+
+	const sshTimeout = 5
+	clientConfig := &ssh.ClientConfig{
+		User:            url.User.Username(),
+		Auth:            authMethods,
+		HostKeyCallback: createHostKeyCallback(credentialsConfig.HostKeyCallback),
+		HostKeyAlgorithms: []string{
+			ssh.KeyAlgoECDSA256,
+			ssh.KeyAlgoECDSA384,
+			ssh.KeyAlgoECDSA521,
+			ssh.KeyAlgoED25519,
+			ssh.SigAlgoRSASHA2512,
+			ssh.SigAlgoRSASHA2256,
+			ssh.KeyAlgoRSA,
+			ssh.KeyAlgoDSA,
+		},
+		Timeout: sshTimeout * time.Second,
+	}
+
+	return clientConfig, nil
+}
+
+func publicKey(path string, passphrase []byte, passPhraseCallback PassPhraseCallback) (ssh.Signer, error) {
+	key, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read key file: %w", err)
+	}
+
+	signer, err := ssh.ParsePrivateKey(key)
+	if err != nil {
+		var missingPhraseError *ssh.PassphraseMissingError
+		if ok := errors.As(err, &missingPhraseError); !ok {
+			return nil, fmt.Errorf("failed to parse private key: %w", err)
+		}
+
+		if len(passphrase) == 0 && passPhraseCallback != nil {
+			b, err := passPhraseCallback()
+			if err != nil {
+				return nil, err
+			}
+			passphrase = []byte(b)
+		}
+
+		return ssh.ParsePrivateKeyWithPassphrase(key, passphrase)
+	}
+
+	return signer, nil
+}
+
+func createHostKeyCallback(hostKeyCallback HostKeyCallback) func(hostPort string, remote net.Addr, key ssh.PublicKey) error {
+	return func(hostPort string, remote net.Addr, pubKey ssh.PublicKey) error {
+		host, port := hostPort, "22"
+		if _h, _p, err := net.SplitHostPort(host); err == nil {
+			host, port = _h, _p
+		}
+
+		knownHosts := filepath.Join(homedir.Get(), ".ssh", "known_hosts")
+
+		_, err := os.Stat(knownHosts)
+		if err != nil && errors.Is(err, os.ErrNotExist) {
+			if hostKeyCallback != nil && hostKeyCallback(hostPort, pubKey) == nil {
+				return nil
+			}
+			return errUnknownServerKey
+		}
+
+		f, err := os.Open(knownHosts)
+		if err != nil {
+			return fmt.Errorf("failed to open known_hosts: %w", err)
+		}
+		defer f.Close()
+
+		hashhost := knownhosts.HashHostname(host)
+
+		var errs []error
+		scanner := bufio.NewScanner(f)
+		for scanner.Scan() {
+			_, hostPorts, _key, _, _, err := ssh.ParseKnownHosts(scanner.Bytes())
+			if err != nil {
+				errs = append(errs, err)
+				continue
+			}
+
+			for _, hp := range hostPorts {
+				h, p := hp, "22"
+				if _h, _p, err := net.SplitHostPort(hp); err == nil {
+					h, p = _h, _p
+				}
+
+				if (h == host || h == hashhost) && port == p {
+					if pubKey.Type() != _key.Type() {
+						errs = append(errs, fmt.Errorf("missmatch in type of a key"))
+						continue
+					}
+					if bytes.Equal(_key.Marshal(), pubKey.Marshal()) {
+						return nil
+					}
+
+					return errBadServerKey
+				}
+			}
+		}
+
+		if hostKeyCallback != nil && hostKeyCallback(hostPort, pubKey) == nil {
+			return nil
+		}
+
+		if len(errs) > 0 {
+			return fmt.Errorf("server is not trusted (%v)", errs)
+		}
+
+		return errUnknownServerKey
+	}
+}
+
+var ErrBadServerKeyMsg = "server key for given host differs from key in known_host"
+var ErrUnknownServerKeyMsg = "server key not found in known_hosts"
+
+// I would expose those but since ssh pkg doesn't do correct error wrapping it would be entirely futile
+var errBadServerKey = errors.New(ErrBadServerKeyMsg)
+var errUnknownServerKey = errors.New(ErrUnknownServerKeyMsg)

ssh/terminal.go

@@ -0,0 +1,118 @@
+package ssh
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"golang.org/x/crypto/ssh"
+	"golang.org/x/term"
+)
+
+// readSecret prompts for a secret and returns value input by user from stdin
+// Unlike terminal.ReadPassword(), $(echo $SECRET | podman...) is supported.
+// Additionally, all input after `<secret>/n` is queued to podman command.
+//
+// NOTE: this code is based on "github.com/containers/podman/v3/pkg/terminal"
+func readSecret(prompt string) (pw []byte, err error) {
+	fd := int(os.Stdin.Fd())
+	if term.IsTerminal(fd) {
+		fmt.Fprint(os.Stderr, prompt)
+		pw, err = term.ReadPassword(fd)
+		fmt.Fprintln(os.Stderr)
+		return
+	}
+
+	var b [1]byte
+	for {
+		n, err := os.Stdin.Read(b[:])
+		// terminal.readSecret discards any '\r', so we do the same
+		if n > 0 && b[0] != '\r' {
+			if b[0] == '\n' {
+				return pw, nil
+			}
+			pw = append(pw, b[0])
+			// limit size, so that a wrong input won't fill up the memory
+			if len(pw) > 1024 {
+				err = errors.New("password too long, 1024 byte limit")
+			}
+		}
+		if err != nil {
+			// terminal.readSecret accepts EOF-terminated passwords
+			// if non-empty, so we do the same
+			if err == io.EOF && len(pw) > 0 {
+				err = nil
+			}
+			return pw, err
+		}
+	}
+}
+
+func NewPasswordCbk() PasswordCallback {
+	var pwdSet bool
+	var pwd string
+	return func() (string, error) {
+		if pwdSet {
+			return pwd, nil
+		}
+
+		p, err := readSecret("please enter password:")
+		if err != nil {
+			return "", err
+		}
+		pwdSet = true
+		pwd = string(p)
+
+		return pwd, err
+	}
+}
+
+func NewPassPhraseCbk() PassPhraseCallback {
+	var pwdSet bool
+	var pwd string
+	return func() (string, error) {
+		if pwdSet {
+			return pwd, nil
+		}
+
+		p, err := readSecret("please enter passphrase to private key:")
+		if err != nil {
+			return "", err
+		}
+		pwdSet = true
+		pwd = string(p)
+
+		return pwd, err
+	}
+}
+
+func NewHostKeyCbk() HostKeyCallback {
+	var trust []byte
+	return func(hostPort string, pubKey ssh.PublicKey) error {
+		if bytes.Equal(trust, pubKey.Marshal()) {
+			return nil
+		}
+		msg := `The authenticity of host %s cannot be established.
+%s key fingerprint is %s
+Are you sure you want to continue connecting (yes/no)? `
+		fmt.Fprintf(os.Stdout, msg, hostPort, pubKey.Type(), ssh.FingerprintSHA256(pubKey))
+		reader := bufio.NewReader(os.Stdin)
+		answer, err := reader.ReadString('\n')
+		if err != nil {
+			return err
+		}
+		answer = strings.TrimRight(answer, "\r\n")
+		answer = strings.ToLower(answer)
+
+		if answer == "yes" || answer == "y" {
+			trust = pubKey.Marshal()
+			return nil
+		}
+
+		return errors.New("key rejected")
+	}
+}

vendor/github.com/docker/cli/cli/connhelper/commandconn/commandconn.go

@@ -0,0 +1,283 @@
+// Package commandconn provides a net.Conn implementation that can be used for
+// proxying (or emulating) stream via a custom command.
+//
+// For example, to provide an http.Client that can connect to a Docker daemon
+// running in a Docker container ("DIND"):
+//
+//  httpClient := &http.Client{
+//  	Transport: &http.Transport{
+//  		DialContext: func(ctx context.Context, _network, _addr string) (net.Conn, error) {
+//  			return commandconn.New(ctx, "docker", "exec", "-it", containerID, "docker", "system", "dial-stdio")
+//  		},
+//  	},
+//  }
+package commandconn
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"runtime"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	exec "golang.org/x/sys/execabs"
+)
+
+// New returns net.Conn
+func New(ctx context.Context, cmd string, args ...string) (net.Conn, error) {
+	var (
+		c   commandConn
+		err error
+	)
+	c.cmd = exec.CommandContext(ctx, cmd, args...)
+	// we assume that args never contains sensitive information
+	logrus.Debugf("commandconn: starting %s with %v", cmd, args)
+	c.cmd.Env = os.Environ()
+	c.cmd.SysProcAttr = &syscall.SysProcAttr{}
+	setPdeathsig(c.cmd)
+	createSession(c.cmd)
+	c.stdin, err = c.cmd.StdinPipe()
+	if err != nil {
+		return nil, err
+	}
+	c.stdout, err = c.cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	c.cmd.Stderr = &stderrWriter{
+		stderrMu:    &c.stderrMu,
+		stderr:      &c.stderr,
+		debugPrefix: fmt.Sprintf("commandconn (%s):", cmd),
+	}
+	c.localAddr = dummyAddr{network: "dummy", s: "dummy-0"}
+	c.remoteAddr = dummyAddr{network: "dummy", s: "dummy-1"}
+	return &c, c.cmd.Start()
+}
+
+// commandConn implements net.Conn
+type commandConn struct {
+	cmd           *exec.Cmd
+	cmdExited     bool
+	cmdWaitErr    error
+	cmdMutex      sync.Mutex
+	stdin         io.WriteCloser
+	stdout        io.ReadCloser
+	stderrMu      sync.Mutex
+	stderr        bytes.Buffer
+	stdioClosedMu sync.Mutex // for stdinClosed and stdoutClosed
+	stdinClosed   bool
+	stdoutClosed  bool
+	localAddr     net.Addr
+	remoteAddr    net.Addr
+}
+
+// killIfStdioClosed kills the cmd if both stdin and stdout are closed.
+func (c *commandConn) killIfStdioClosed() error {
+	c.stdioClosedMu.Lock()
+	stdioClosed := c.stdoutClosed && c.stdinClosed
+	c.stdioClosedMu.Unlock()
+	if !stdioClosed {
+		return nil
+	}
+	return c.kill()
+}
+
+// killAndWait tries sending SIGTERM to the process before sending SIGKILL.
+func killAndWait(cmd *exec.Cmd) error {
+	var werr error
+	if runtime.GOOS != "windows" {
+		werrCh := make(chan error)
+		go func() { werrCh <- cmd.Wait() }()
+		cmd.Process.Signal(syscall.SIGTERM)
+		select {
+		case werr = <-werrCh:
+		case <-time.After(3 * time.Second):
+			cmd.Process.Kill()
+			werr = <-werrCh
+		}
+	} else {
+		cmd.Process.Kill()
+		werr = cmd.Wait()
+	}
+	return werr
+}
+
+// kill returns nil if the command terminated, regardless to the exit status.
+func (c *commandConn) kill() error {
+	var werr error
+	c.cmdMutex.Lock()
+	if c.cmdExited {
+		werr = c.cmdWaitErr
+	} else {
+		werr = killAndWait(c.cmd)
+		c.cmdWaitErr = werr
+		c.cmdExited = true
+	}
+	c.cmdMutex.Unlock()
+	if werr == nil {
+		return nil
+	}
+	wExitErr, ok := werr.(*exec.ExitError)
+	if ok {
+		if wExitErr.ProcessState.Exited() {
+			return nil
+		}
+	}
+	return errors.Wrapf(werr, "commandconn: failed to wait")
+}
+
+func (c *commandConn) onEOF(eof error) error {
+	// when we got EOF, the command is going to be terminated
+	var werr error
+	c.cmdMutex.Lock()
+	if c.cmdExited {
+		werr = c.cmdWaitErr
+	} else {
+		werrCh := make(chan error)
+		go func() { werrCh <- c.cmd.Wait() }()
+		select {
+		case werr = <-werrCh:
+			c.cmdWaitErr = werr
+			c.cmdExited = true
+		case <-time.After(10 * time.Second):
+			c.cmdMutex.Unlock()
+			c.stderrMu.Lock()
+			stderr := c.stderr.String()
+			c.stderrMu.Unlock()
+			return errors.Errorf("command %v did not exit after %v: stderr=%q", c.cmd.Args, eof, stderr)
+		}
+	}
+	c.cmdMutex.Unlock()
+	if werr == nil {
+		return eof
+	}
+	c.stderrMu.Lock()
+	stderr := c.stderr.String()
+	c.stderrMu.Unlock()
+	return errors.Errorf("command %v has exited with %v, please make sure the URL is valid, and Docker 18.09 or later is installed on the remote host: stderr=%s", c.cmd.Args, werr, stderr)
+}
+
+func ignorableCloseError(err error) bool {
+	errS := err.Error()
+	ss := []string{
+		os.ErrClosed.Error(),
+	}
+	for _, s := range ss {
+		if strings.Contains(errS, s) {
+			return true
+		}
+	}
+	return false
+}
+
+func (c *commandConn) CloseRead() error {
+	// NOTE: maybe already closed here
+	if err := c.stdout.Close(); err != nil && !ignorableCloseError(err) {
+		logrus.Warnf("commandConn.CloseRead: %v", err)
+	}
+	c.stdioClosedMu.Lock()
+	c.stdoutClosed = true
+	c.stdioClosedMu.Unlock()
+	if err := c.killIfStdioClosed(); err != nil {
+		logrus.Warnf("commandConn.CloseRead: %v", err)
+	}
+	return nil
+}
+
+func (c *commandConn) Read(p []byte) (int, error) {
+	n, err := c.stdout.Read(p)
+	if err == io.EOF {
+		err = c.onEOF(err)
+	}
+	return n, err
+}
+
+func (c *commandConn) CloseWrite() error {
+	// NOTE: maybe already closed here
+	if err := c.stdin.Close(); err != nil && !ignorableCloseError(err) {
+		logrus.Warnf("commandConn.CloseWrite: %v", err)
+	}
+	c.stdioClosedMu.Lock()
+	c.stdinClosed = true
+	c.stdioClosedMu.Unlock()
+	if err := c.killIfStdioClosed(); err != nil {
+		logrus.Warnf("commandConn.CloseWrite: %v", err)
+	}
+	return nil
+}
+
+func (c *commandConn) Write(p []byte) (int, error) {
+	n, err := c.stdin.Write(p)
+	if err == io.EOF {
+		err = c.onEOF(err)
+	}
+	return n, err
+}
+
+func (c *commandConn) Close() error {
+	var err error
+	if err = c.CloseRead(); err != nil {
+		logrus.Warnf("commandConn.Close: CloseRead: %v", err)
+	}
+	if err = c.CloseWrite(); err != nil {
+		logrus.Warnf("commandConn.Close: CloseWrite: %v", err)
+	}
+	return err
+}
+
+func (c *commandConn) LocalAddr() net.Addr {
+	return c.localAddr
+}
+func (c *commandConn) RemoteAddr() net.Addr {
+	return c.remoteAddr
+}
+func (c *commandConn) SetDeadline(t time.Time) error {
+	logrus.Debugf("unimplemented call: SetDeadline(%v)", t)
+	return nil
+}
+func (c *commandConn) SetReadDeadline(t time.Time) error {
+	logrus.Debugf("unimplemented call: SetReadDeadline(%v)", t)
+	return nil
+}
+func (c *commandConn) SetWriteDeadline(t time.Time) error {
+	logrus.Debugf("unimplemented call: SetWriteDeadline(%v)", t)
+	return nil
+}
+
+type dummyAddr struct {
+	network string
+	s       string
+}
+
+func (d dummyAddr) Network() string {
+	return d.network
+}
+
+func (d dummyAddr) String() string {
+	return d.s
+}
+
+type stderrWriter struct {
+	stderrMu    *sync.Mutex
+	stderr      *bytes.Buffer
+	debugPrefix string
+}
+
+func (w *stderrWriter) Write(p []byte) (int, error) {
+	logrus.Debugf("%s%s", w.debugPrefix, string(p))
+	w.stderrMu.Lock()
+	if w.stderr.Len() > 4096 {
+		w.stderr.Reset()
+	}
+	n, err := w.stderr.Write(p)
+	w.stderrMu.Unlock()
+	return n, err
+}

vendor/github.com/docker/cli/cli/connhelper/commandconn/pdeathsig_linux.go

@@ -0,0 +1,10 @@
+package commandconn
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+func setPdeathsig(cmd *exec.Cmd) {
+	cmd.SysProcAttr.Pdeathsig = syscall.SIGKILL
+}

vendor/github.com/docker/cli/cli/connhelper/commandconn/pdeathsig_nolinux.go

@@ -0,0 +1,10 @@
+// +build !linux
+
+package commandconn
+
+import (
+	"os/exec"
+)
+
+func setPdeathsig(cmd *exec.Cmd) {
+}

vendor/github.com/docker/cli/cli/connhelper/commandconn/session_unix.go

@@ -0,0 +1,13 @@
+// +build !windows
+
+package commandconn
+
+import (
+	"os/exec"
+)
+
+func createSession(cmd *exec.Cmd) {
+	// for supporting ssh connection helper with ProxyCommand
+	// https://github.com/docker/cli/issues/1707
+	cmd.SysProcAttr.Setsid = true
+}

vendor/github.com/docker/cli/cli/connhelper/commandconn/session_windows.go

@@ -0,0 +1,8 @@
+package commandconn
+
+import (
+	"os/exec"
+)
+
+func createSession(cmd *exec.Cmd) {
+}

vendor/github.com/docker/cli/cli/connhelper/connhelper.go

@@ -0,0 +1,68 @@
+// Package connhelper provides helpers for connecting to a remote daemon host with custom logic.
+package connhelper
+
+import (
+	"context"
+	"net"
+	"net/url"
+
+	"github.com/docker/cli/cli/connhelper/commandconn"
+	"github.com/docker/cli/cli/connhelper/ssh"
+	"github.com/pkg/errors"
+)
+
+// ConnectionHelper allows to connect to a remote host with custom stream provider binary.
+type ConnectionHelper struct {
+	Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
+	Host   string // dummy URL used for HTTP requests. e.g. "http://docker"
+}
+
+// GetConnectionHelper returns Docker-specific connection helper for the given URL.
+// GetConnectionHelper returns nil without error when no helper is registered for the scheme.
+//
+// ssh://<user>@<host> URL requires Docker 18.09 or later on the remote host.
+func GetConnectionHelper(daemonURL string) (*ConnectionHelper, error) {
+	return getConnectionHelper(daemonURL, nil)
+}
+
+// GetConnectionHelperWithSSHOpts returns Docker-specific connection helper for
+// the given URL, and accepts additional options for ssh connections. It returns
+// nil without error when no helper is registered for the scheme.
+//
+// Requires Docker 18.09 or later on the remote host.
+func GetConnectionHelperWithSSHOpts(daemonURL string, sshFlags []string) (*ConnectionHelper, error) {
+	return getConnectionHelper(daemonURL, sshFlags)
+}
+
+func getConnectionHelper(daemonURL string, sshFlags []string) (*ConnectionHelper, error) {
+	u, err := url.Parse(daemonURL)
+	if err != nil {
+		return nil, err
+	}
+	switch scheme := u.Scheme; scheme {
+	case "ssh":
+		sp, err := ssh.ParseURL(daemonURL)
+		if err != nil {
+			return nil, errors.Wrap(err, "ssh host connection is not valid")
+		}
+		return &ConnectionHelper{
+			Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
+				return commandconn.New(ctx, "ssh", append(sshFlags, sp.Args("docker", "system", "dial-stdio")...)...)
+			},
+			Host: "http://docker",
+		}, nil
+	}
+	// Future version may support plugins via ~/.docker/config.json. e.g. "dind"
+	// See docker/cli#889 for the previous discussion.
+	return nil, err
+}
+
+// GetCommandConnectionHelper returns Docker-specific connection helper constructed from an arbitrary command.
+func GetCommandConnectionHelper(cmd string, flags ...string) (*ConnectionHelper, error) {
+	return &ConnectionHelper{
+		Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return commandconn.New(ctx, cmd, flags...)
+		},
+		Host: "http://docker",
+	}, nil
+}

vendor/github.com/docker/cli/cli/connhelper/ssh/ssh.go

@@ -0,0 +1,64 @@
+// Package ssh provides the connection helper for ssh:// URL.
+package ssh
+
+import (
+	"net/url"
+
+	"github.com/pkg/errors"
+)
+
+// ParseURL parses URL
+func ParseURL(daemonURL string) (*Spec, error) {
+	u, err := url.Parse(daemonURL)
+	if err != nil {
+		return nil, err
+	}
+	if u.Scheme != "ssh" {
+		return nil, errors.Errorf("expected scheme ssh, got %q", u.Scheme)
+	}
+
+	var sp Spec
+
+	if u.User != nil {
+		sp.User = u.User.Username()
+		if _, ok := u.User.Password(); ok {
+			return nil, errors.New("plain-text password is not supported")
+		}
+	}
+	sp.Host = u.Hostname()
+	if sp.Host == "" {
+		return nil, errors.Errorf("no host specified")
+	}
+	sp.Port = u.Port()
+	if u.Path != "" {
+		return nil, errors.Errorf("extra path after the host: %q", u.Path)
+	}
+	if u.RawQuery != "" {
+		return nil, errors.Errorf("extra query after the host: %q", u.RawQuery)
+	}
+	if u.Fragment != "" {
+		return nil, errors.Errorf("extra fragment after the host: %q", u.Fragment)
+	}
+	return &sp, err
+}
+
+// Spec of SSH URL
+type Spec struct {
+	User string
+	Host string
+	Port string
+}
+
+// Args returns args except "ssh" itself combined with optional additional command args
+func (sp *Spec) Args(add ...string) []string {
+	var args []string
+	if sp.User != "" {
+		args = append(args, "-l", sp.User)
+	}
+	if sp.Port != "" {
+		args = append(args, "-p", sp.Port)
+	}
+	args = append(args, "--", sp.Host)
+	args = append(args, add...)
+	return args
+}

vendor/modules.txt

@@ -182,10 +182,14 @@ github.com/davecgh/go-spew/spew
 github.com/dgraph-io/ristretto
 github.com/dgraph-io/ristretto/z
 # github.com/docker/cli v20.10.7+incompatible
+## explicit
 github.com/docker/cli/cli/config
 github.com/docker/cli/cli/config/configfile
 github.com/docker/cli/cli/config/credentials
 github.com/docker/cli/cli/config/types
+github.com/docker/cli/cli/connhelper
+github.com/docker/cli/cli/connhelper/commandconn
+github.com/docker/cli/cli/connhelper/ssh
 # github.com/docker/distribution v2.7.1+incompatible
 github.com/docker/distribution/digestset
 github.com/docker/distribution/reference
@@ -642,6 +646,7 @@ go.uber.org/zap/internal/color
 go.uber.org/zap/internal/exit
 go.uber.org/zap/zapcore
 # golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a
+## explicit
 golang.org/x/crypto/blowfish
 golang.org/x/crypto/cast5
 golang.org/x/crypto/chacha20
@@ -702,6 +707,7 @@ golang.org/x/sys/windows
 golang.org/x/sys/windows/registry
 golang.org/x/sys/windows/svc/eventlog
 # golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b
+## explicit
 golang.org/x/term
 # golang.org/x/text v0.3.7
 ## explicit