conn.go

v0.6.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
	// Empty Dir ⇒ today's cwd. A missing Dir makes Run() fail to start with a
	// non-ExitError, which the final error branch wraps as a LOCAL_EXEC run error
	// (not a non-zero exit) — IV5 for the local transport, no new code path.
	ec.Dir = cmd.Dir
	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	// Empty Dir ⇒ today's cwd. A missing Dir makes Run() fail to start with a
68	// non-ExitError, which the final error branch wraps as a LOCAL_EXEC run error
69	// (not a non-zero exit) — IV5 for the local transport, no new code path.
70	ec.Dir = cmd.Dir
71	if env != nil {
72	ec.Env = env
73	}
74	if cmd.Stdin != nil {
75	ec.Stdin = cmd.Stdin
76	}
77	// A nil writer falls back to a buffer copied into ExecResult; a non-nil writer
78	// receives the stream live and leaves the ExecResult field nil.
79	var outBuf, errBuf bytes.Buffer
80	if stdout != nil {
81	ec.Stdout = stdout
82	} else {
83	ec.Stdout = &outBuf
84	}
85	if stderr != nil {
86	ec.Stderr = stderr
87	} else {
88	ec.Stderr = &errBuf
89	}
90
91	err := ec.Run()
92	res := legatus.ExecResult{}
93	if stdout == nil {
94	res.Stdout = outBuf.Bytes()
95	}
96	if stderr == nil {
97	res.Stderr = errBuf.Bytes()
98	}
99	if err != nil {
100	// A cancelled ctx kills the process, which surfaces as an ExitError; the
101	// contract (IV4, matching SSH) is to report ctx.Err(), not a non-zero exit.
102	if ctx.Err() != nil {
103	return res, culpa.WithCode(culpa.Wrapf(ctx.Err(), "local: run %s cancelled", cmd.Path), "LOCAL_EXEC")
104	}
105	var ee *exec.ExitError
106	if errors.As(err, &ee) {
107	res.ExitCode = ee.ExitCode()
108	return res, nil
109	}
110	return res, culpa.WithCode(culpa.Wrapf(err, "local: run %s", cmd.Path), "LOCAL_EXEC")
111	}
112	return res, nil
113	}
114
115	// Put writes content to remote with mode. Without sudo it writes atomically
116	// (temp file in the destination directory, then rename). With sudo it stages to
117	// a private /tmp file then `sudo mv` into place, mirroring the SSH transport. An
118	// owner, if given, is applied with `chown` (sudo-wrapped when sudo is set).
119	func (c *Conn) Put(ctx context.Context, content io.Reader, remote string, mode legatus.FileMode, opts ...legatus.PutOption) error {
120	if remote == "" {
121	return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_PUT")
122	}
123	var o legatus.PutOptions
124	for _, opt := range opts {
125	opt(&o)
126	}
127	data, err := io.ReadAll(content)
128	if err != nil {
129	return culpa.WithCode(culpa.Wrap(err, "local: read content"), "LOCAL_PUT")
130	}
131
132	if !o.Sudo {
133	if err := atomicWrite(remote, data, mode); err != nil {
134	return err
135	}
136	if o.Owner != "" {
137	return c.chown(ctx, o.Owner, remote, false)
138	}
139	return nil
140	}
141
142	tmp := localTempPath(remote)
143	if err := atomicWrite(tmp, data, mode); err != nil {
144	return err
145	}
146	if err := c.runChecked(ctx, legatus.Cmd{Path: "mkdir", Args: []string{"-p", filepath.Dir(remote)}, Sudo: true}); err != nil {
147	_ = os.Remove(tmp)
148	return culpa.WithCode(culpa.Wrap(err, "local: sudo mkdir"), "LOCAL_PUT")
149	}
150	if err := c.runChecked(ctx, legatus.Cmd{Path: "mv", Args: []string{tmp, remote}, Sudo: true}); err != nil {
151	_ = os.Remove(tmp)
152	return culpa.WithCode(culpa.Wrap(err, "local: sudo mv into place"), "LOCAL_PUT")
153	}
154	if o.Owner != "" {
155	return c.chown(ctx, o.Owner, remote, true)
156	}
157	return nil
158	}
159
160	// Get streams the local file at remote to w.
161	func (c *Conn) Get(_ context.Context, remote string, w io.Writer) error {
162	if remote == "" {
163	return culpa.WithCode(culpa.New("local: empty remote path"), "LOCAL_GET")
164	}
165	f, err := os.Open(remote)
166	if err != nil {
167	return culpa.WithCode(culpa.Wrapf(err, "local: open %s", remote), "LOCAL_GET")
168	}
169	defer func() { _ = f.Close() }()
170	if _, err := io.Copy(w, f); err != nil {
171	return culpa.WithCode(culpa.Wrapf(err, "local: read %s", remote), "LOCAL_GET")
172	}
173	return nil
174	}
175
176	// Close is a no-op: a local Conn holds no connection.
177	func (c *Conn) Close() error { return nil }
178
179	// chown sets the owner of remote via the chown command (sudo-wrapped when sudo),
180	// matching how the SSH transport applies ownership (handles user:group names).
181	func (c *Conn) chown(ctx context.Context, owner, remote string, sudo bool) error {
182	return c.runChecked(ctx, legatus.Cmd{Path: "chown", Args: []string{owner, remote}, Sudo: sudo})
183	}
184
185	// runChecked runs a command that must succeed, turning a non-zero exit into an
186	// error so a failed sudo mv/chown is never silently swallowed.
187	func (c *Conn) runChecked(ctx context.Context, cmd legatus.Cmd) error {
188	res, err := c.Exec(ctx, cmd)
189	if err == nil && res.ExitCode != 0 {
190	err = culpa.Errorf("exit %d: %s", res.ExitCode, bytes.TrimSpace(res.Combined()))
191	}
192	return err
193	}
194
195	// sudoArgv builds the argv for a sudo-wrapped command: `sudo -- [env KEY=val …]
196	// path args`. env(1) re-exports the environment under sudo (sudo strips it),
197	// mirroring the SSH transport. No shell quoting is needed — os/exec passes argv
198	// structurally, so values can never be re-split.
199	func sudoArgv(cmd legatus.Cmd) []string {
200	argv := []string{"sudo", "--"}
201	if len(cmd.Env) > 0 {
202	argv = append(argv, "env")
203	argv = append(argv, sortedEnv(cmd.Env)...)
204	}
205	argv = append(argv, cmd.Path)
206	return append(argv, cmd.Args...)
207	}
208
209	// sortedEnv renders env as deterministic KEY=value tokens.
210	func sortedEnv(env map[string]string) []string {
211	keys := make([]string, 0, len(env))
212	for k := range env {
213	keys = append(keys, k)
214	}
215	sort.Strings(keys)
216	out := make([]string, 0, len(keys))
217	for _, k := range keys {
218	out = append(out, k+"="+env[k])
219	}
220	return out
221	}
222
223	// atomicWrite writes data to dst via a temp file in the same directory then
224	// rename, so a failed or interrupted write never leaves a partial target.
225	func atomicWrite(dst string, data []byte, mode legatus.FileMode) error {
226	dir := filepath.Dir(dst)
227	if err := os.MkdirAll(dir, 0o755); err != nil {
228	return culpa.WithCode(culpa.Wrapf(err, "local: mkdir %s", dir), "LOCAL_PUT")
229	}
230	tmp, err := os.CreateTemp(dir, ".legatus-*")
231	if err != nil {
232	return culpa.WithCode(culpa.Wrapf(err, "local: temp in %s", dir), "LOCAL_PUT")
233	}
234	tmpName := tmp.Name()
235	if _, err := tmp.Write(data); err != nil {
236	_ = tmp.Close()
237	_ = os.Remove(tmpName)
238	return culpa.WithCode(culpa.Wrapf(err, "local: write %s", tmpName), "LOCAL_PUT")
239	}
240	if err := tmp.Close(); err != nil {
241	_ = os.Remove(tmpName)
242	return culpa.WithCode(culpa.Wrapf(err, "local: close %s", tmpName), "LOCAL_PUT")
243	}
244	if err := os.Chmod(tmpName, mode); err != nil {
245	_ = os.Remove(tmpName)
246	return culpa.WithCode(culpa.Wrapf(err, "local: chmod %s", tmpName), "LOCAL_PUT")
247	}
248	if err := os.Rename(tmpName, dst); err != nil {
249	_ = os.Remove(tmpName)
250	return culpa.WithCode(culpa.Wrapf(err, "local: rename into %s", dst), "LOCAL_PUT")
251	}
252	return nil
253	}
254
255	// localTempPath derives a private /tmp staging path for the sudo move path.
256	func localTempPath(remote string) string {
257	return filepath.Join(os.TempDir(), "legatus."+
258	strconv.Itoa(os.Getpid())+"."+
259	strconv.FormatInt(time.Now().UnixNano(), 10)+"."+
260	filepath.Base(remote))
261	}
262

conn.go

Source Files