conn.go

v0.3.0

package transport

import (
	"bytes"
	"context"
	"io"
	"os"
	"path"
	"strconv"
	"time"

	"github.com/pkg/sftp"
	xssh "golang.org/x/crypto/ssh"

	"go.bigb.es/auxilia/culpa"
	"go.bigb.es/auxilia/legatus"
)

// sshConn is a live SSH connection to one host, implementing legatus.Conn over
// golang.org/x/crypto/ssh and github.com/pkg/sftp. It carries no user identity:
// sudo-wrapping is driven entirely by the per-call Cmd.Sudo / PutOptions.Sudo
// flags (the privileged action, not the login identity, decides). No key
// material or secret is logged here (IV5).
type sshConn struct {
	client *xssh.Client
	host   string
	// jumps are the ProxyJump clients this connection tunnels through, in dial
	// order. They must outlive the target client and are closed in reverse on
	// Close — closing a jump tears down the tunnel beneath it.
	jumps []*xssh.Client
}

var _ legatus.Conn = (*sshConn)(nil)

// Exec runs cmd on the remote host. The command is rendered by buildRemote,
// which individually quotes every argument and env value (IV2) and applies sudo
// + env(1) wrapping when cmd.Sudo is set (IV6). stdout/stderr are captured in
// full and the exit code is reported in ExecResult.
//
// A command that runs to completion is NOT an error, even on a non-zero exit:
// the exit status is reported via ExecResult.ExitCode and err is nil. Callers
// decide what a non-zero exit means — a step's Apply treats it as failure, while
// a probe (`test -e`, `systemctl is-active`, `dpkg-query`) treats it as a signal.
// err is reserved for failures to run the command at all: no session, a lost
// signal, an IO error, or context cancellation (which sends SIGTERM, tears down
// the remote process, and surfaces ctx.Err()).
func (c *sshConn) Exec(ctx context.Context, cmd legatus.Cmd) (legatus.ExecResult, error) {
	return c.ExecStream(ctx, cmd, nil, nil)
}

// ExecStream runs cmd, streaming each non-nil writer live and buffering each nil
// writer into the returned ExecResult (the buffered contract Exec relies on). The
// exit-code and cancellation behavior is identical to Exec.
func (c *sshConn) ExecStream(ctx context.Context, cmd legatus.Cmd, stdout, stderr io.Writer) (legatus.ExecResult, error) {
	if cmd.Path == "" {
		return legatus.ExecResult{}, culpa.WithCode(
			culpa.New("ssh: empty command path"), "SSH_EXEC")
	}

	sess, err := c.client.NewSession()
	if err != nil {
		return legatus.ExecResult{}, culpa.WithCode(
			culpa.Wrapf(err, "ssh: new session on %s", c.host), "SSH_EXEC")
	}
	defer func() { _ = sess.Close() }()

	// A nil writer falls back to a buffer copied into ExecResult; a non-nil writer
	// receives the stream live and leaves the ExecResult field nil.
	var outBuf, errBuf bytes.Buffer
	if stdout != nil {
		sess.Stdout = stdout
	} else {
		sess.Stdout = &outBuf
	}
	if stderr != nil {
		sess.Stderr = stderr
	} else {
		sess.Stderr = &errBuf
	}
	if cmd.Stdin != nil {
		sess.Stdin = cmd.Stdin
	}

	remote, err := buildRemote(cmd)
	if err != nil {
		return legatus.ExecResult{}, err
	}

	if err := sess.Start(remote); err != nil {
		return legatus.ExecResult{}, culpa.WithCode(
			culpa.Wrapf(err, "ssh: start command on %s", c.host), "SSH_EXEC")
	}

	result := func() legatus.ExecResult {
		r := legatus.ExecResult{}
		if stdout == nil {
			r.Stdout = outBuf.Bytes()
		}
		if stderr == nil {
			r.Stderr = errBuf.Bytes()
		}
		return r
	}

	done := make(chan error, 1)
	go func() { done <- sess.Wait() }()

	select {
	case <-ctx.Done():
		_ = sess.Signal(xssh.SIGTERM)
		_ = sess.Close()
		<-done
		return result(), culpa.WithCode(
			culpa.Wrapf(ctx.Err(), "ssh: command on %s cancelled", c.host), "SSH_EXEC")
	case werr := <-done:
		res := result()
		if werr == nil {
			return res, nil
		}
		var ee *xssh.ExitError
		if culpa.As(werr, &ee) {
			// Ran to completion, non-zero exit: report via ExitCode, not error.
			res.ExitCode = ee.ExitStatus()
			return res, nil
		}
		// Non-exit failure (lost signal, IO error, …): the command did not run to
		// a clean exit, so this is a hard error.
		return res, culpa.WithCode(
			culpa.Wrapf(werr, "ssh: command on %s", c.host), "SSH_EXEC")
	}
}

// Put uploads content to remote with mode. When opts request sudo, the file is
// staged to a private per-invocation /tmp path (the SFTP subsystem runs as the
// login user and can't write a root-owned destination), then the destination
// directory is created and the file moved into place with sudo (IV6); an owner,
// if given, is applied with `sudo chown`. Without sudo the file is written
// straight to the destination via SFTP.
func (c *sshConn) Put(ctx context.Context, content io.Reader, remote string, mode legatus.FileMode, opts ...legatus.PutOption) error {
	if remote == "" {
		return culpa.WithCode(culpa.New("ssh: empty remote path"), "SSH_PUT")
	}
	var o legatus.PutOptions
	for _, opt := range opts {
		opt(&o)
	}

	sc, err := sftp.NewClient(c.client)
	if err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: open sftp to %s", c.host), "SSH_PUT")
	}
	defer func() { _ = sc.Close() }()

	if !o.Sudo {
		if err := sftpWrite(sc, content, remote, mode); err != nil {
			return err
		}
		if o.Owner != "" {
			if err := c.chown(ctx, o.Owner, remote, false); err != nil {
				return err
			}
		}
		return nil
	}

	// Sudo path: stage to a private temp file, then privileged-move into place.
	tmp := remoteTempPath(remote)
	if err := sftpWrite(sc, content, tmp, mode); err != nil {
		return err
	}
	if err := c.checkedExec(ctx,
		legatus.Cmd{Path: "mkdir", Args: []string{"-p", path.Dir(remote)}, Sudo: true},
		"sudo mkdir remote dir"); err != nil {
		_ = sc.Remove(tmp)
		return err
	}
	if err := c.checkedExec(ctx,
		legatus.Cmd{Path: "mv", Args: []string{tmp, remote}, Sudo: true},
		"sudo mv into place"); err != nil {
		_ = sc.Remove(tmp)
		return err
	}
	if o.Owner != "" {
		if err := c.chown(ctx, o.Owner, remote, true); err != nil {
			return err
		}
	}
	return nil
}

// chown sets the owner of remote, sudo-wrapped when sudo is set.
func (c *sshConn) chown(ctx context.Context, owner, remote string, sudo bool) error {
	return c.checkedExec(ctx,
		legatus.Cmd{Path: "chown", Args: []string{owner, remote}, Sudo: sudo},
		"chown "+remote)
}

// checkedExec runs a Put-path command that must succeed. The Conn.Exec contract
// reports a clean non-zero exit via ExitCode with a nil error, so requiring
// success means checking ExitCode too — otherwise a failed sudo mv/chown would
// be silently swallowed, leaving the remote partially written or mis-owned. The
// failure (transport error or non-zero exit with its output) is wrapped SSH_PUT.
func (c *sshConn) checkedExec(ctx context.Context, cmd legatus.Cmd, what string) error {
	res, err := c.Exec(ctx, cmd)
	if err == nil && res.ExitCode != 0 {
		err = culpa.Errorf("exit %d: %s", res.ExitCode, bytes.TrimSpace(res.Combined()))
	}
	if err != nil {
		return culpa.WithCode(culpa.Wrapf(err, "ssh: %s on %s", what, c.host), "SSH_PUT")
	}
	return nil
}

// Get streams the remote file to w via SFTP.
func (c *sshConn) Get(ctx context.Context, remote string, w io.Writer) error {
	if remote == "" {
		return culpa.WithCode(culpa.New("ssh: empty remote path"), "SSH_GET")
	}
	sc, err := sftp.NewClient(c.client)
	if err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: open sftp to %s", c.host), "SSH_GET")
	}
	defer func() { _ = sc.Close() }()

	f, err := sc.Open(remote)
	if err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: open remote %s", remote), "SSH_GET")
	}
	defer func() { _ = f.Close() }()

	if _, err := io.Copy(w, f); err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: read remote %s", remote), "SSH_GET")
	}
	return nil
}

// Close releases the underlying SSH connection, then any ProxyJump clients in
// reverse dial order (a jump must outlive the hop it carries). The first error
// is returned; later closes still run.
func (c *sshConn) Close() error {
	var firstErr error
	if c.client != nil {
		if err := c.client.Close(); err != nil {
			firstErr = culpa.WithCode(
				culpa.Wrapf(err, "ssh: close connection to %s", c.host), "SSH_CLOSE")
		}
	}
	for i := len(c.jumps) - 1; i >= 0; i-- {
		if err := c.jumps[i].Close(); err != nil && firstErr == nil {
			firstErr = culpa.WithCode(
				culpa.Wrapf(err, "ssh: close jump connection"), "SSH_CLOSE")
		}
	}
	return firstErr
}

// sftpWrite writes src to remote via SFTP, creating parent dirs and applying
// mode. It is the unprivileged write path shared by Put's direct and staging
// branches.
func sftpWrite(sc *sftp.Client, src io.Reader, remote string, mode legatus.FileMode) error {
	if dir := path.Dir(remote); dir != "." && dir != "/" {
		if err := sc.MkdirAll(dir); err != nil {
			return culpa.WithCode(
				culpa.Wrapf(err, "ssh: mkdir remote dir %s", dir), "SSH_PUT")
		}
	}
	dst, err := sc.OpenFile(remote, os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
	if err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: open remote %s", remote), "SSH_PUT")
	}
	if _, err := io.Copy(dst, src); err != nil {
		_ = dst.Close()
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: write remote %s", remote), "SSH_PUT")
	}
	if err := dst.Close(); err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: close remote %s", remote), "SSH_PUT")
	}
	if err := sc.Chmod(remote, mode); err != nil {
		return culpa.WithCode(
			culpa.Wrapf(err, "ssh: chmod remote %s", remote), "SSH_PUT")
	}
	return nil
}

// remoteTempPath derives a private staging path under /tmp from the destination
// base name plus a pid-and-time suffix, unique enough for concurrent uploads;
// the file is moved or removed immediately after.
func remoteTempPath(remote string) string {
	base := path.Base(remote)
	return "/tmp/legatus." + strconv.Itoa(os.Getpid()) + "." +
		strconv.FormatInt(time.Now().UnixNano(), 10) + "." + base
}


1	package transport
2
3	import (
4	"bytes"
5	"context"
6	"io"
7	"os"
8	"path"
9	"strconv"
10	"time"
11
12	"github.com/pkg/sftp"
13	xssh "golang.org/x/crypto/ssh"
14
15	"go.bigb.es/auxilia/culpa"
16	"go.bigb.es/auxilia/legatus"
17	)
18
19	// sshConn is a live SSH connection to one host, implementing legatus.Conn over
20	// golang.org/x/crypto/ssh and github.com/pkg/sftp. It carries no user identity:
21	// sudo-wrapping is driven entirely by the per-call Cmd.Sudo / PutOptions.Sudo
22	// flags (the privileged action, not the login identity, decides). No key
23	// material or secret is logged here (IV5).
24	type sshConn struct {
25	client *xssh.Client
26	host string
27	// jumps are the ProxyJump clients this connection tunnels through, in dial
28	// order. They must outlive the target client and are closed in reverse on
29	// Close — closing a jump tears down the tunnel beneath it.
30	jumps []*xssh.Client
31	}
32
33	var _ legatus.Conn = (*sshConn)(nil)
34
35	// Exec runs cmd on the remote host. The command is rendered by buildRemote,
36	// which individually quotes every argument and env value (IV2) and applies sudo
37	// + env(1) wrapping when cmd.Sudo is set (IV6). stdout/stderr are captured in
38	// full and the exit code is reported in ExecResult.
39	//
40	// A command that runs to completion is NOT an error, even on a non-zero exit:
41	// the exit status is reported via ExecResult.ExitCode and err is nil. Callers
42	// decide what a non-zero exit means — a step's Apply treats it as failure, while
43	// a probe (`test -e`, `systemctl is-active`, `dpkg-query`) treats it as a signal.
44	// err is reserved for failures to run the command at all: no session, a lost
45	// signal, an IO error, or context cancellation (which sends SIGTERM, tears down
46	// the remote process, and surfaces ctx.Err()).
47	func (c *sshConn) Exec(ctx context.Context, cmd legatus.Cmd) (legatus.ExecResult, error) {
48	return c.ExecStream(ctx, cmd, nil, nil)
49	}
50
51	// ExecStream runs cmd, streaming each non-nil writer live and buffering each nil
52	// writer into the returned ExecResult (the buffered contract Exec relies on). The
53	// exit-code and cancellation behavior is identical to Exec.
54	func (c *sshConn) ExecStream(ctx context.Context, cmd legatus.Cmd, stdout, stderr io.Writer) (legatus.ExecResult, error) {
55	if cmd.Path == "" {
56	return legatus.ExecResult{}, culpa.WithCode(
57	culpa.New("ssh: empty command path"), "SSH_EXEC")
58	}
59
60	sess, err := c.client.NewSession()
61	if err != nil {
62	return legatus.ExecResult{}, culpa.WithCode(
63	culpa.Wrapf(err, "ssh: new session on %s", c.host), "SSH_EXEC")
64	}
65	defer func() { _ = sess.Close() }()
66
67	// A nil writer falls back to a buffer copied into ExecResult; a non-nil writer
68	// receives the stream live and leaves the ExecResult field nil.
69	var outBuf, errBuf bytes.Buffer
70	if stdout != nil {
71	sess.Stdout = stdout
72	} else {
73	sess.Stdout = &outBuf
74	}
75	if stderr != nil {
76	sess.Stderr = stderr
77	} else {
78	sess.Stderr = &errBuf
79	}
80	if cmd.Stdin != nil {
81	sess.Stdin = cmd.Stdin
82	}
83
84	remote, err := buildRemote(cmd)
85	if err != nil {
86	return legatus.ExecResult{}, err
87	}
88
89	if err := sess.Start(remote); err != nil {
90	return legatus.ExecResult{}, culpa.WithCode(
91	culpa.Wrapf(err, "ssh: start command on %s", c.host), "SSH_EXEC")
92	}
93
94	result := func() legatus.ExecResult {
95	r := legatus.ExecResult{}
96	if stdout == nil {
97	r.Stdout = outBuf.Bytes()
98	}
99	if stderr == nil {
100	r.Stderr = errBuf.Bytes()
101	}
102	return r
103	}
104
105	done := make(chan error, 1)
106	go func() { done <- sess.Wait() }()
107
108	select {
109	case <-ctx.Done():
110	_ = sess.Signal(xssh.SIGTERM)
111	_ = sess.Close()
112	<-done
113	return result(), culpa.WithCode(
114	culpa.Wrapf(ctx.Err(), "ssh: command on %s cancelled", c.host), "SSH_EXEC")
115	case werr := <-done:
116	res := result()
117	if werr == nil {
118	return res, nil
119	}
120	var ee *xssh.ExitError
121	if culpa.As(werr, &ee) {
122	// Ran to completion, non-zero exit: report via ExitCode, not error.
123	res.ExitCode = ee.ExitStatus()
124	return res, nil
125	}
126	// Non-exit failure (lost signal, IO error, …): the command did not run to
127	// a clean exit, so this is a hard error.
128	return res, culpa.WithCode(
129	culpa.Wrapf(werr, "ssh: command on %s", c.host), "SSH_EXEC")
130	}
131	}
132
133	// Put uploads content to remote with mode. When opts request sudo, the file is
134	// staged to a private per-invocation /tmp path (the SFTP subsystem runs as the
135	// login user and can't write a root-owned destination), then the destination
136	// directory is created and the file moved into place with sudo (IV6); an owner,
137	// if given, is applied with `sudo chown`. Without sudo the file is written
138	// straight to the destination via SFTP.
139	func (c *sshConn) Put(ctx context.Context, content io.Reader, remote string, mode legatus.FileMode, opts ...legatus.PutOption) error {
140	if remote == "" {
141	return culpa.WithCode(culpa.New("ssh: empty remote path"), "SSH_PUT")
142	}
143	var o legatus.PutOptions
144	for _, opt := range opts {
145	opt(&o)
146	}
147
148	sc, err := sftp.NewClient(c.client)
149	if err != nil {
150	return culpa.WithCode(
151	culpa.Wrapf(err, "ssh: open sftp to %s", c.host), "SSH_PUT")
152	}
153	defer func() { _ = sc.Close() }()
154
155	if !o.Sudo {
156	if err := sftpWrite(sc, content, remote, mode); err != nil {
157	return err
158	}
159	if o.Owner != "" {
160	if err := c.chown(ctx, o.Owner, remote, false); err != nil {
161	return err
162	}
163	}
164	return nil
165	}
166
167	// Sudo path: stage to a private temp file, then privileged-move into place.
168	tmp := remoteTempPath(remote)
169	if err := sftpWrite(sc, content, tmp, mode); err != nil {
170	return err
171	}
172	if err := c.checkedExec(ctx,
173	legatus.Cmd{Path: "mkdir", Args: []string{"-p", path.Dir(remote)}, Sudo: true},
174	"sudo mkdir remote dir"); err != nil {
175	_ = sc.Remove(tmp)
176	return err
177	}
178	if err := c.checkedExec(ctx,
179	legatus.Cmd{Path: "mv", Args: []string{tmp, remote}, Sudo: true},
180	"sudo mv into place"); err != nil {
181	_ = sc.Remove(tmp)
182	return err
183	}
184	if o.Owner != "" {
185	if err := c.chown(ctx, o.Owner, remote, true); err != nil {
186	return err
187	}
188	}
189	return nil
190	}
191
192	// chown sets the owner of remote, sudo-wrapped when sudo is set.
193	func (c *sshConn) chown(ctx context.Context, owner, remote string, sudo bool) error {
194	return c.checkedExec(ctx,
195	legatus.Cmd{Path: "chown", Args: []string{owner, remote}, Sudo: sudo},
196	"chown "+remote)
197	}
198
199	// checkedExec runs a Put-path command that must succeed. The Conn.Exec contract
200	// reports a clean non-zero exit via ExitCode with a nil error, so requiring
201	// success means checking ExitCode too — otherwise a failed sudo mv/chown would
202	// be silently swallowed, leaving the remote partially written or mis-owned. The
203	// failure (transport error or non-zero exit with its output) is wrapped SSH_PUT.
204	func (c *sshConn) checkedExec(ctx context.Context, cmd legatus.Cmd, what string) error {
205	res, err := c.Exec(ctx, cmd)
206	if err == nil && res.ExitCode != 0 {
207	err = culpa.Errorf("exit %d: %s", res.ExitCode, bytes.TrimSpace(res.Combined()))
208	}
209	if err != nil {
210	return culpa.WithCode(culpa.Wrapf(err, "ssh: %s on %s", what, c.host), "SSH_PUT")
211	}
212	return nil
213	}
214
215	// Get streams the remote file to w via SFTP.
216	func (c *sshConn) Get(ctx context.Context, remote string, w io.Writer) error {
217	if remote == "" {
218	return culpa.WithCode(culpa.New("ssh: empty remote path"), "SSH_GET")
219	}
220	sc, err := sftp.NewClient(c.client)
221	if err != nil {
222	return culpa.WithCode(
223	culpa.Wrapf(err, "ssh: open sftp to %s", c.host), "SSH_GET")
224	}
225	defer func() { _ = sc.Close() }()
226
227	f, err := sc.Open(remote)
228	if err != nil {
229	return culpa.WithCode(
230	culpa.Wrapf(err, "ssh: open remote %s", remote), "SSH_GET")
231	}
232	defer func() { _ = f.Close() }()
233
234	if _, err := io.Copy(w, f); err != nil {
235	return culpa.WithCode(
236	culpa.Wrapf(err, "ssh: read remote %s", remote), "SSH_GET")
237	}
238	return nil
239	}
240
241	// Close releases the underlying SSH connection, then any ProxyJump clients in
242	// reverse dial order (a jump must outlive the hop it carries). The first error
243	// is returned; later closes still run.
244	func (c *sshConn) Close() error {
245	var firstErr error
246	if c.client != nil {
247	if err := c.client.Close(); err != nil {
248	firstErr = culpa.WithCode(
249	culpa.Wrapf(err, "ssh: close connection to %s", c.host), "SSH_CLOSE")
250	}
251	}
252	for i := len(c.jumps) - 1; i >= 0; i-- {
253	if err := c.jumps[i].Close(); err != nil && firstErr == nil {
254	firstErr = culpa.WithCode(
255	culpa.Wrapf(err, "ssh: close jump connection"), "SSH_CLOSE")
256	}
257	}
258	return firstErr
259	}
260
261	// sftpWrite writes src to remote via SFTP, creating parent dirs and applying
262	// mode. It is the unprivileged write path shared by Put's direct and staging
263	// branches.
264	func sftpWrite(sc *sftp.Client, src io.Reader, remote string, mode legatus.FileMode) error {
265	if dir := path.Dir(remote); dir != "." && dir != "/" {
266	if err := sc.MkdirAll(dir); err != nil {
267	return culpa.WithCode(
268	culpa.Wrapf(err, "ssh: mkdir remote dir %s", dir), "SSH_PUT")
269	}
270	}
271	dst, err := sc.OpenFile(remote, os.O_WRONLY\|os.O_CREATE\|os.O_TRUNC)
272	if err != nil {
273	return culpa.WithCode(
274	culpa.Wrapf(err, "ssh: open remote %s", remote), "SSH_PUT")
275	}
276	if _, err := io.Copy(dst, src); err != nil {
277	_ = dst.Close()
278	return culpa.WithCode(
279	culpa.Wrapf(err, "ssh: write remote %s", remote), "SSH_PUT")
280	}
281	if err := dst.Close(); err != nil {
282	return culpa.WithCode(
283	culpa.Wrapf(err, "ssh: close remote %s", remote), "SSH_PUT")
284	}
285	if err := sc.Chmod(remote, mode); err != nil {
286	return culpa.WithCode(
287	culpa.Wrapf(err, "ssh: chmod remote %s", remote), "SSH_PUT")
288	}
289	return nil
290	}
291
292	// remoteTempPath derives a private staging path under /tmp from the destination
293	// base name plus a pid-and-time suffix, unique enough for concurrent uploads;
294	// the file is moved or removed immediately after.
295	func remoteTempPath(remote string) string {
296	base := path.Base(remote)
297	return "/tmp/legatus." + strconv.Itoa(os.Getpid()) + "." +
298	strconv.FormatInt(time.Now().UnixNano(), 10) + "." + base
299	}
300

conn.go

Source Files