1package remote
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "log/slog"
11 "os/exec"
12 "strings"
13 "time"
14 "unicode"
15 "unicode/utf8"
16
17 "github.com/bnema/vev/internal/adapters/sshstdio"
18 "github.com/bnema/vev/internal/domain"
19 "github.com/bnema/vev/internal/ports"
20)
21
22const (
23 maxCatalogBytes = 1 << 20
24 maxCatalogDiagnosticBytes = 4 << 10
25 catalogCommandTimeout = 30 * time.Second
26 catalogCommandWaitDelay = 500 * time.Millisecond
27)
28
29var (
30 errCatalogDecode = errors.New("remote catalog: invalid JSON")
31 errCatalogTrailing = errors.New("remote catalog: trailing bytes after JSON")
32 errCatalogRequired = errors.New("remote catalog: missing required fields")
33 errCatalogSession = errors.New("remote catalog: invalid session")
34 errCatalogSSH = errors.New("remote catalog: ssh command failed")
35 errCatalogTooLarge = errors.New("remote catalog: output exceeds size limit")
36)
37
38// CatalogClient fetches remote session catalogs over SSH.
39type CatalogClient struct {
40 command func(ctx context.Context, name string, args ...string) *exec.Cmd
41 // timeout bounds each List when > 0; otherwise catalogCommandTimeout is used.
42 timeout time.Duration
43}
44
45var _ ports.RemoteCatalogClient = (*CatalogClient)(nil)
46
47// NewCatalogClient returns a RemoteCatalogClient that shells out to ssh.
48func NewCatalogClient() ports.RemoteCatalogClient {
49 return &CatalogClient{command: exec.CommandContext}
50}
51
52func (c *CatalogClient) listTimeout() time.Duration {
53 if c != nil && c.timeout > 0 {
54 return c.timeout
55 }
56 return catalogCommandTimeout
57}
58
59// List runs `ssh -- <target> 'vev' 'cmd' 'remote-catalog' '--json'` and decodes
60// exactly one versioned catalog envelope from stdout. List always applies a
61// bounded command timeout derived from ctx so a caller with no deadline cannot
62// hang indefinitely, while still honoring caller cancellation.
63func (c *CatalogClient) List(ctx context.Context, target string) (ports.RemoteCatalog, error) {
64 if err := ctx.Err(); err != nil {
65 return ports.RemoteCatalog{}, err
66 }
67 if err := domain.ValidateRemoteHostTarget(target); err != nil {
68 slog.Debug("remote catalog rejected target", "target", target, "err", err)
69 return ports.RemoteCatalog{}, err
70 }
71
72 runCtx, cancel := context.WithTimeout(ctx, c.listTimeout())
73 defer cancel()
74
75 command := c.command
76 if command == nil {
77 command = exec.CommandContext
78 }
79 spec := sshstdio.BuildCommandForRemoteCommand(target, "vev", "cmd", "remote-catalog", "--json")
80 cmd := command(runCtx, spec.Path, spec.Args...)
81 stdout := boundedBuffer{limit: maxCatalogBytes}
82 stderr := boundedBuffer{limit: maxCatalogDiagnosticBytes}
83 cmd.Stdout = &stdout
84 cmd.Stderr = &stderr
85 cmd.WaitDelay = catalogCommandWaitDelay
86 err := cmd.Run()
87 if ctxErr := runCtx.Err(); ctxErr != nil {
88 return ports.RemoteCatalog{}, ctxErr
89 }
90 if err != nil {
91 if stdout.overflow || stderr.overflow {
92 slog.Debug("remote catalog output too large", "target", target, "stdout_limit", maxCatalogBytes, "stderr_limit", maxCatalogDiagnosticBytes)
93 return ports.RemoteCatalog{}, errCatalogTooLarge
94 }
95 stderrText := sanitizeCatalogDiagnostic(string(stderr.Bytes()))
96 slog.Debug("remote catalog ssh failed", "target", target, "err", err, "stderr", stderrText)
97 if stderrText != "" {
98 return ports.RemoteCatalog{}, fmt.Errorf("%w: %w: %s", errCatalogSSH, err, stderrText)
99 }
100 return ports.RemoteCatalog{}, fmt.Errorf("%w: %w", errCatalogSSH, err)
101 }
102 if stdout.overflow || stderr.overflow {
103 slog.Debug("remote catalog output too large", "target", target, "stdout_limit", maxCatalogBytes, "stderr_limit", maxCatalogDiagnosticBytes)
104 return ports.RemoteCatalog{}, errCatalogTooLarge
105 }
106
107 catalog, err := decodeRemoteCatalog(stdout.Bytes())
108 if err != nil {
109 slog.Debug("remote catalog decode failed", "target", target, "err", err)
110 return ports.RemoteCatalog{}, err
111 }
112 return catalog, nil
113}
114
115type catalogEnvelope struct {
116 ProtocolVersion *uint16 `json:"protocol_version"`
117 SchemaVersion *uint16 `json:"schema_version"`
118 Sessions *[]ports.RemoteCatalogSession `json:"sessions"`
119}
120
121func decodeRemoteCatalog(raw []byte) (ports.RemoteCatalog, error) {
122 dec := json.NewDecoder(bytes.NewReader(raw))
123 var envelope catalogEnvelope
124 if err := dec.Decode(&envelope); err != nil {
125 return ports.RemoteCatalog{}, fmt.Errorf("%w: %w", errCatalogDecode, err)
126 }
127 if envelope.ProtocolVersion == nil || envelope.Sessions == nil {
128 return ports.RemoteCatalog{}, errCatalogRequired
129 }
130 if err := rejectTrailingJSON(dec); err != nil {
131 return ports.RemoteCatalog{}, err
132 }
133
134 schemaVersion := uint16(0)
135 if envelope.SchemaVersion != nil {
136 schemaVersion = *envelope.SchemaVersion
137 }
138 if *envelope.ProtocolVersion != ports.ProtocolVersion {
139 return ports.RemoteCatalog{}, &ports.RemoteCatalogVersionMismatchError{
140 Got: *envelope.ProtocolVersion,
141 Want: ports.ProtocolVersion,
142 Kind: "protocol",
143 }
144 }
145 if schemaVersion != 0 && schemaVersion != ports.RemoteCatalogSchemaVersion {
146 return ports.RemoteCatalog{}, &ports.RemoteCatalogVersionMismatchError{
147 Got: schemaVersion,
148 Want: ports.RemoteCatalogSchemaVersion,
149 Kind: "catalog",
150 }
151 }
152
153 sessions := *envelope.Sessions
154 if sessions == nil {
155 sessions = []ports.RemoteCatalogSession{}
156 }
157 catalog := ports.RemoteCatalog{
158 ProtocolVersion: *envelope.ProtocolVersion,
159 SchemaVersion: schemaVersion,
160 Sessions: sessions,
161 }
162 if err := ports.ValidateRemoteCatalog(catalog); err != nil {
163 if schemaVersion == 0 {
164 // Keep the old adapter error taxonomy for pre-parity peers and
165 // direct CLI compatibility. Current-schema payloads retain the
166 // typed validation error for callers that need to classify it.
167 return ports.RemoteCatalog{}, fmt.Errorf("%w: %w", errCatalogSession, err)
168 }
169 return ports.RemoteCatalog{}, err
170 }
171 return catalog, nil
172}
173
174func rejectTrailingJSON(dec *json.Decoder) error {
175 var extra json.RawMessage
176 err := dec.Decode(&extra)
177 if errors.Is(err, io.EOF) {
178 return nil
179 }
180 if err == nil {
181 return errCatalogTrailing
182 }
183 return fmt.Errorf("%w: %w", errCatalogTrailing, err)
184}
185
186func validCatalogState(state string) bool {
187 switch state {
188 case "up", "down", "broken":
189 return true
190 default:
191 return false
192 }
193}
194
195// sanitizeCatalogDiagnostic strips terminal escapes and control bytes while
196// preserving useful printable diagnostic text for errors and logs.
197func sanitizeCatalogDiagnostic(raw string) string {
198 var b strings.Builder
199 b.Grow(len(raw))
200 for i := 0; i < len(raw); {
201 if raw[i] == 0x1b {
202 i = skipTerminalEscape(raw, i+1)
203 continue
204 }
205 r, size := utf8.DecodeRuneInString(raw[i:])
206 if r == utf8.RuneError && size == 1 {
207 i++
208 continue
209 }
210 if unicode.IsPrint(r) || r == '\t' || r == '\n' {
211 b.WriteRune(r)
212 }
213 i += size
214 }
215 return strings.TrimSpace(b.String())
216}
217
218func skipTerminalEscape(s string, i int) int {
219 if i >= len(s) {
220 return i
221 }
222 switch s[i] {
223 case '[': // CSI: ESC [ ... final byte in 0x40-0x7E
224 i++
225 for i < len(s) {
226 ch := s[i]
227 i++
228 if ch >= 0x40 && ch <= 0x7e {
229 break
230 }
231 }
232 return i
233 case ']': // OSC: ESC ] ... BEL or ST (ESC \)
234 i++
235 for i < len(s) {
236 if s[i] == 0x07 {
237 return i + 1
238 }
239 if s[i] == 0x1b && i+1 < len(s) && s[i+1] == '\\' {
240 return i + 2
241 }
242 i++
243 }
244 return i
245 default:
246 // Two-byte Fe escape or unknown: drop ESC and the next byte.
247 return i + 1
248 }
249}
250
251// boundedBuffer captures stream output up to an explicit byte limit and records overflow.
252type boundedBuffer struct {
253 buf bytes.Buffer
254 limit int
255 overflow bool
256}
257
258func (b *boundedBuffer) Write(p []byte) (int, error) {
259 if b.overflow {
260 return len(p), nil
261 }
262 remaining := b.limit - b.buf.Len()
263 if remaining <= 0 {
264 b.overflow = true
265 return len(p), nil
266 }
267 if len(p) <= remaining {
268 return b.buf.Write(p)
269 }
270 n, err := b.buf.Write(p[:remaining])
271 b.overflow = true
272 if err != nil {
273 return n, err
274 }
275 return len(p), nil
276}
277
278func (b *boundedBuffer) Bytes() []byte {
279 return b.buf.Bytes()
280}