conn.go

v0.4.0

// Package local implements legatus.Conn/legatus.Dialer on the local machine, so
// workstation-local steps run through the same Flow/Runner model as remote ones.
// Exec runs via os/exec; Put/Get use local file IO with atomic writes. No remote
// shell is involved, so arguments are passed structurally (os/exec argv) with no
// quoting; sudo wrapping mirrors the SSH transport's `sudo -- [env KEY=val …]`
// shape so a step behaves the same whether it targets a remote host or this one.
package local

import (
	"bytes"
	"context"
	"errors"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strconv"
	"time"

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

// cancelWaitDelay bounds how long Run waits for I/O after a ctx-cancel kill. A
// killed shell can leave a child (e.g. `sleep`) holding the stdout pipe open, so
// without this Run would block until that child exits; the delay force-closes the
// pipes so a streamed command (logs -f on Ctrl-C) returns promptly.
const cancelWaitDelay = 2 * time.Second

// Conn is a legatus.Conn that operates on the local machine.
type Conn struct{}

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

// NewConn returns a local Conn.
func NewConn() *Conn { return &Conn{} }

// Exec runs cmd locally. A process that runs to completion is not an error even
// on a non-zero exit: the status is reported via ExecResult.ExitCode and err is
// nil. err is reserved for failure to launch (binary not found, IO error) or
// context cancellation — matching the SSH transport's Conn.Exec contract so the
// engine cannot tell a local Conn from a remote one.
func (c *Conn) 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 launch-error contract is identical to Exec.
func (c *Conn) ExecStream(ctx context.Context, cmd legatus.Cmd, stdout, stderr io.Writer) (legatus.ExecResult, error) {
	if cmd.Path == "" {
		return legatus.ExecResult{}, culpa.WithCode(culpa.New("local: empty command path"), "LOCAL_EXEC")
	}

	name, args := cmd.Path, cmd.Args
	var env []string
	if cmd.Sudo {
		argv := sudoArgv(cmd)
		name, args = argv[0], argv[1:]
	} else if len(cmd.Env) > 0 {
		env = append(os.Environ(), sortedEnv(cmd.Env)...)
	}

	ec := exec.CommandContext(ctx, name, args...)
	ec.WaitDelay = cancelWaitDelay
	if env != nil {
		ec.Env = env
	}
	if cmd.Stdin != nil {
		ec.Stdin = cmd.Stdin
	}
	// 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 {
		ec.Stdout = stdout
	} else {
		ec.Stdout = &outBuf
	}
	if stderr != nil {
		ec.Stderr = stderr
	} else {
		ec.Stderr = &errBuf
	}

	err := ec.Run()
	res := legatus.ExecResult{}
	if stdout == nil {
		res.Stdout = outBuf.Bytes()
	}
	if stderr == nil {
		res.Stderr = errBuf.Bytes()
	}
	if err != nil {
		// A cancelled ctx kills the process, which surfaces as an ExitError; the
		// contract (IV4, matching SSH) is to report ctx.Err(), not a non-zero exit.
		if ctx.Err() != nil {
			return res, culpa.WithCode(culpa.Wrapf(ctx.Err(), "local: run %s cancelled", cmd.Path), "LOCAL_EXEC")
		}
		var ee *exec.ExitError
		if errors.As(err, &ee) {
			res.ExitCode = ee.ExitCode()
			return res, nil
		}
		return res, culpa.WithCode(culpa.Wrapf(err, "local: run %s", cmd.Path), "LOCAL_EXEC")
	}
	return res, nil
}

// Put writes content to remote with mode. Without sudo it writes atomically
// (temp file in the destination directory, then rename). With sudo it stages to
// a private /tmp file then `sudo mv` into place, mirroring the SSH transport. An
// owner, if given, is applied with `chown` (sudo-wrapped when sudo is set).
func (c *Conn) Put(ctx context.Context, content io.Reader, remote string, mode legatus.FileMode, opts ...legatus.PutOption) error {
	if remote == "" {
		return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_PUT")
	}
	var o legatus.PutOptions
	for _, opt := range opts {
		opt(&o)
	}
	data, err := io.ReadAll(content)
	if err != nil {
		return culpa.WithCode(culpa.Wrap(err, "local: read content"), "LOCAL_PUT")
	}

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

	tmp := localTempPath(remote)
	if err := atomicWrite(tmp, data, mode); err != nil {
		return err
	}
	if err := c.runChecked(ctx, legatus.Cmd{Path: "mkdir", Args: []string{"-p", filepath.Dir(remote)}, Sudo: true}); err != nil {
		_ = os.Remove(tmp)
		return culpa.WithCode(culpa.Wrap(err, "local: sudo mkdir"), "LOCAL_PUT")
	}
	if err := c.runChecked(ctx, legatus.Cmd{Path: "mv", Args: []string{tmp, remote}, Sudo: true}); err != nil {
		_ = os.Remove(tmp)
		return culpa.WithCode(culpa.Wrap(err, "local: sudo mv into place"), "LOCAL_PUT")
	}
	if o.Owner != "" {
		return c.chown(ctx, o.Owner, remote, true)
	}
	return nil
}

// Get streams the local file at remote to w.
func (c *Conn) Get(_ context.Context, remote string, w io.Writer) error {
	if remote == "" {
		return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_GET")
	}
	f, err := os.Open(remote)
	if err != nil {
		return culpa.WithCode(culpa.Wrapf(err, "local: open %s", remote), "LOCAL_GET")
	}
	defer func() { _ = f.Close() }()
	if _, err := io.Copy(w, f); err != nil {
		return culpa.WithCode(culpa.Wrapf(err, "local: read %s", remote), "LOCAL_GET")
	}
	return nil
}

// Close is a no-op: a local Conn holds no connection.
func (c *Conn) Close() error { return nil }

// chown sets the owner of remote via the chown command (sudo-wrapped when sudo),
// matching how the SSH transport applies ownership (handles user:group names).
func (c *Conn) chown(ctx context.Context, owner, remote string, sudo bool) error {
	return c.runChecked(ctx, legatus.Cmd{Path: "chown", Args: []string{owner, remote}, Sudo: sudo})
}

// runChecked runs a command that must succeed, turning a non-zero exit into an
// error so a failed sudo mv/chown is never silently swallowed.
func (c *Conn) runChecked(ctx context.Context, cmd legatus.Cmd) 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()))
	}
	return err
}

// sudoArgv builds the argv for a sudo-wrapped command: `sudo -- [env KEY=val …]
// path args`. env(1) re-exports the environment under sudo (sudo strips it),
// mirroring the SSH transport. No shell quoting is needed — os/exec passes argv
// structurally, so values can never be re-split.
func sudoArgv(cmd legatus.Cmd) []string {
	argv := []string{"sudo", "--"}
	if len(cmd.Env) > 0 {
		argv = append(argv, "env")
		argv = append(argv, sortedEnv(cmd.Env)...)
	}
	argv = append(argv, cmd.Path)
	return append(argv, cmd.Args...)
}

// sortedEnv renders env as deterministic KEY=value tokens.
func sortedEnv(env map[string]string) []string {
	keys := make([]string, 0, len(env))
	for k := range env {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	out := make([]string, 0, len(keys))
	for _, k := range keys {
		out = append(out, k+"="+env[k])
	}
	return out
}

// atomicWrite writes data to dst via a temp file in the same directory then
// rename, so a failed or interrupted write never leaves a partial target.
func atomicWrite(dst string, data []byte, mode legatus.FileMode) error {
	dir := filepath.Dir(dst)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return culpa.WithCode(culpa.Wrapf(err, "local: mkdir %s", dir), "LOCAL_PUT")
	}
	tmp, err := os.CreateTemp(dir, ".legatus-*")
	if err != nil {
		return culpa.WithCode(culpa.Wrapf(err, "local: temp in %s", dir), "LOCAL_PUT")
	}
	tmpName := tmp.Name()
	if _, err := tmp.Write(data); err != nil {
		_ = tmp.Close()
		_ = os.Remove(tmpName)
		return culpa.WithCode(culpa.Wrapf(err, "local: write %s", tmpName), "LOCAL_PUT")
	}
	if err := tmp.Close(); err != nil {
		_ = os.Remove(tmpName)
		return culpa.WithCode(culpa.Wrapf(err, "local: close %s", tmpName), "LOCAL_PUT")
	}
	if err := os.Chmod(tmpName, mode); err != nil {
		_ = os.Remove(tmpName)
		return culpa.WithCode(culpa.Wrapf(err, "local: chmod %s", tmpName), "LOCAL_PUT")
	}
	if err := os.Rename(tmpName, dst); err != nil {
		_ = os.Remove(tmpName)
		return culpa.WithCode(culpa.Wrapf(err, "local: rename into %s", dst), "LOCAL_PUT")
	}
	return nil
}

// localTempPath derives a private /tmp staging path for the sudo move path.
func localTempPath(remote string) string {
	return filepath.Join(os.TempDir(), "legatus."+
		strconv.Itoa(os.Getpid())+"."+
		strconv.FormatInt(time.Now().UnixNano(), 10)+"."+
		filepath.Base(remote))
}


1	// Package local implements legatus.Conn/legatus.Dialer on the local machine, so
2	// workstation-local steps run through the same Flow/Runner model as remote ones.
3	// Exec runs via os/exec; Put/Get use local file IO with atomic writes. No remote
4	// shell is involved, so arguments are passed structurally (os/exec argv) with no
5	// quoting; sudo wrapping mirrors the SSH transport's `sudo -- [env KEY=val …]`
6	// shape so a step behaves the same whether it targets a remote host or this one.
7	package local
8
9	import (
10	"bytes"
11	"context"
12	"errors"
13	"io"
14	"os"
15	"os/exec"
16	"path/filepath"
17	"sort"
18	"strconv"
19	"time"
20
21	"go.bigb.es/auxilia/culpa"
22	"go.bigb.es/auxilia/legatus"
23	)
24
25	// cancelWaitDelay bounds how long Run waits for I/O after a ctx-cancel kill. A
26	// killed shell can leave a child (e.g. `sleep`) holding the stdout pipe open, so
27	// without this Run would block until that child exits; the delay force-closes the
28	// pipes so a streamed command (logs -f on Ctrl-C) returns promptly.
29	const cancelWaitDelay = 2 * time.Second
30
31	// Conn is a legatus.Conn that operates on the local machine.
32	type Conn struct{}
33
34	var _ legatus.Conn = (*Conn)(nil)
35
36	// NewConn returns a local Conn.
37	func NewConn() *Conn { return &Conn{} }
38
39	// Exec runs cmd locally. A process that runs to completion is not an error even
40	// on a non-zero exit: the status is reported via ExecResult.ExitCode and err is
41	// nil. err is reserved for failure to launch (binary not found, IO error) or
42	// context cancellation — matching the SSH transport's Conn.Exec contract so the
43	// engine cannot tell a local Conn from a remote one.
44	func (c *Conn) Exec(ctx context.Context, cmd legatus.Cmd) (legatus.ExecResult, error) {
45	return c.ExecStream(ctx, cmd, nil, nil)
46	}
47
48	// ExecStream runs cmd, streaming each non-nil writer live and buffering each nil
49	// writer into the returned ExecResult (the buffered contract Exec relies on). The
50	// exit-code and launch-error contract is identical to Exec.
51	func (c *Conn) ExecStream(ctx context.Context, cmd legatus.Cmd, stdout, stderr io.Writer) (legatus.ExecResult, error) {
52	if cmd.Path == "" {
53	return legatus.ExecResult{}, culpa.WithCode(culpa.New("local: empty command path"), "LOCAL_EXEC")
54	}
55
56	name, args := cmd.Path, cmd.Args
57	var env []string
58	if cmd.Sudo {
59	argv := sudoArgv(cmd)
60	name, args = argv[0], argv[1:]
61	} else if len(cmd.Env) > 0 {
62	env = append(os.Environ(), sortedEnv(cmd.Env)...)
63	}
64
65	ec := exec.CommandContext(ctx, name, args...)
66	ec.WaitDelay = cancelWaitDelay
67	if env != nil {
68	ec.Env = env
69	}
70	if cmd.Stdin != nil {
71	ec.Stdin = cmd.Stdin
72	}
73	// A nil writer falls back to a buffer copied into ExecResult; a non-nil writer
74	// receives the stream live and leaves the ExecResult field nil.
75	var outBuf, errBuf bytes.Buffer
76	if stdout != nil {
77	ec.Stdout = stdout
78	} else {
79	ec.Stdout = &outBuf
80	}
81	if stderr != nil {
82	ec.Stderr = stderr
83	} else {
84	ec.Stderr = &errBuf
85	}
86
87	err := ec.Run()
88	res := legatus.ExecResult{}
89	if stdout == nil {
90	res.Stdout = outBuf.Bytes()
91	}
92	if stderr == nil {
93	res.Stderr = errBuf.Bytes()
94	}
95	if err != nil {
96	// A cancelled ctx kills the process, which surfaces as an ExitError; the
97	// contract (IV4, matching SSH) is to report ctx.Err(), not a non-zero exit.
98	if ctx.Err() != nil {
99	return res, culpa.WithCode(culpa.Wrapf(ctx.Err(), "local: run %s cancelled", cmd.Path), "LOCAL_EXEC")
100	}
101	var ee *exec.ExitError
102	if errors.As(err, &ee) {
103	res.ExitCode = ee.ExitCode()
104	return res, nil
105	}
106	return res, culpa.WithCode(culpa.Wrapf(err, "local: run %s", cmd.Path), "LOCAL_EXEC")
107	}
108	return res, nil
109	}
110
111	// Put writes content to remote with mode. Without sudo it writes atomically
112	// (temp file in the destination directory, then rename). With sudo it stages to
113	// a private /tmp file then `sudo mv` into place, mirroring the SSH transport. An
114	// owner, if given, is applied with `chown` (sudo-wrapped when sudo is set).
115	func (c *Conn) Put(ctx context.Context, content io.Reader, remote string, mode legatus.FileMode, opts ...legatus.PutOption) error {
116	if remote == "" {
117	return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_PUT")
118	}
119	var o legatus.PutOptions
120	for _, opt := range opts {
121	opt(&o)
122	}
123	data, err := io.ReadAll(content)
124	if err != nil {
125	return culpa.WithCode(culpa.Wrap(err, "local: read content"), "LOCAL_PUT")
126	}
127
128	if !o.Sudo {
129	if err := atomicWrite(remote, data, mode); err != nil {
130	return err
131	}
132	if o.Owner != "" {
133	return c.chown(ctx, o.Owner, remote, false)
134	}
135	return nil
136	}
137
138	tmp := localTempPath(remote)
139	if err := atomicWrite(tmp, data, mode); err != nil {
140	return err
141	}
142	if err := c.runChecked(ctx, legatus.Cmd{Path: "mkdir", Args: []string{"-p", filepath.Dir(remote)}, Sudo: true}); err != nil {
143	_ = os.Remove(tmp)
144	return culpa.WithCode(culpa.Wrap(err, "local: sudo mkdir"), "LOCAL_PUT")
145	}
146	if err := c.runChecked(ctx, legatus.Cmd{Path: "mv", Args: []string{tmp, remote}, Sudo: true}); err != nil {
147	_ = os.Remove(tmp)
148	return culpa.WithCode(culpa.Wrap(err, "local: sudo mv into place"), "LOCAL_PUT")
149	}
150	if o.Owner != "" {
151	return c.chown(ctx, o.Owner, remote, true)
152	}
153	return nil
154	}
155
156	// Get streams the local file at remote to w.
157	func (c *Conn) Get(_ context.Context, remote string, w io.Writer) error {
158	if remote == "" {
159	return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_GET")
160	}
161	f, err := os.Open(remote)
162	if err != nil {
163	return culpa.WithCode(culpa.Wrapf(err, "local: open %s", remote), "LOCAL_GET")
164	}
165	defer func() { _ = f.Close() }()
166	if _, err := io.Copy(w, f); err != nil {
167	return culpa.WithCode(culpa.Wrapf(err, "local: read %s", remote), "LOCAL_GET")
168	}
169	return nil
170	}
171
172	// Close is a no-op: a local Conn holds no connection.
173	func (c *Conn) Close() error { return nil }
174
175	// chown sets the owner of remote via the chown command (sudo-wrapped when sudo),
176	// matching how the SSH transport applies ownership (handles user:group names).
177	func (c *Conn) chown(ctx context.Context, owner, remote string, sudo bool) error {
178	return c.runChecked(ctx, legatus.Cmd{Path: "chown", Args: []string{owner, remote}, Sudo: sudo})
179	}
180
181	// runChecked runs a command that must succeed, turning a non-zero exit into an
182	// error so a failed sudo mv/chown is never silently swallowed.
183	func (c *Conn) runChecked(ctx context.Context, cmd legatus.Cmd) error {
184	res, err := c.Exec(ctx, cmd)
185	if err == nil && res.ExitCode != 0 {
186	err = culpa.Errorf("exit %d: %s", res.ExitCode, bytes.TrimSpace(res.Combined()))
187	}
188	return err
189	}
190
191	// sudoArgv builds the argv for a sudo-wrapped command: `sudo -- [env KEY=val …]
192	// path args`. env(1) re-exports the environment under sudo (sudo strips it),
193	// mirroring the SSH transport. No shell quoting is needed — os/exec passes argv
194	// structurally, so values can never be re-split.
195	func sudoArgv(cmd legatus.Cmd) []string {
196	argv := []string{"sudo", "--"}
197	if len(cmd.Env) > 0 {
198	argv = append(argv, "env")
199	argv = append(argv, sortedEnv(cmd.Env)...)
200	}
201	argv = append(argv, cmd.Path)
202	return append(argv, cmd.Args...)
203	}
204
205	// sortedEnv renders env as deterministic KEY=value tokens.
206	func sortedEnv(env map[string]string) []string {
207	keys := make([]string, 0, len(env))
208	for k := range env {
209	keys = append(keys, k)
210	}
211	sort.Strings(keys)
212	out := make([]string, 0, len(keys))
213	for _, k := range keys {
214	out = append(out, k+"="+env[k])
215	}
216	return out
217	}
218
219	// atomicWrite writes data to dst via a temp file in the same directory then
220	// rename, so a failed or interrupted write never leaves a partial target.
221	func atomicWrite(dst string, data []byte, mode legatus.FileMode) error {
222	dir := filepath.Dir(dst)
223	if err := os.MkdirAll(dir, 0o755); err != nil {
224	return culpa.WithCode(culpa.Wrapf(err, "local: mkdir %s", dir), "LOCAL_PUT")
225	}
226	tmp, err := os.CreateTemp(dir, ".legatus-*")
227	if err != nil {
228	return culpa.WithCode(culpa.Wrapf(err, "local: temp in %s", dir), "LOCAL_PUT")
229	}
230	tmpName := tmp.Name()
231	if _, err := tmp.Write(data); err != nil {
232	_ = tmp.Close()
233	_ = os.Remove(tmpName)
234	return culpa.WithCode(culpa.Wrapf(err, "local: write %s", tmpName), "LOCAL_PUT")
235	}
236	if err := tmp.Close(); err != nil {
237	_ = os.Remove(tmpName)
238	return culpa.WithCode(culpa.Wrapf(err, "local: close %s", tmpName), "LOCAL_PUT")
239	}
240	if err := os.Chmod(tmpName, mode); err != nil {
241	_ = os.Remove(tmpName)
242	return culpa.WithCode(culpa.Wrapf(err, "local: chmod %s", tmpName), "LOCAL_PUT")
243	}
244	if err := os.Rename(tmpName, dst); err != nil {
245	_ = os.Remove(tmpName)
246	return culpa.WithCode(culpa.Wrapf(err, "local: rename into %s", dst), "LOCAL_PUT")
247	}
248	return nil
249	}
250
251	// localTempPath derives a private /tmp staging path for the sudo move path.
252	func localTempPath(remote string) string {
253	return filepath.Join(os.TempDir(), "legatus."+
254	strconv.Itoa(os.Getpid())+"."+
255	strconv.FormatInt(time.Now().UnixNano(), 10)+"."+
256	filepath.Base(remote))
257	}
258

conn.go

Source Files