diff --git a/.golangci.yml b/.golangci.yml index c63d09d..496de7e 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -27,8 +27,8 @@ linters: misspell: locale: US -issues: - exclude-rules: - - text: "misspelling of" - linters: - - misspell + exclusions: + rules: + - text: "misspelling of" + linters: + - misspell diff --git a/AGENT.md b/AGENT.md index b21c2b2..afa3aa2 100644 --- a/AGENT.md +++ b/AGENT.md @@ -312,7 +312,9 @@ tests. 4. **Sudo auto-fill only supports leading `sudo`.** `CommandUsesSudo` returns true only when the remote command starts with `sudo`, matching the exact form `sudoStdinCommand` can safely rewrite. Non-leading sudo inside shell - wrappers or pipelines is left untouched. + wrappers or pipelines is left untouched and is never silently ignored: sshx + warns on stderr before connecting and again when the remote reports + `sudo: a password is required`, suggesting `sudo sh -c ""`. 5. **Command safety checks.** Destructive operations (`rm -rf /`, `mkfs`, `dd`, fork bombs, `curl | sh`, critical file edits, shutdown/reboot) are blocked unless `--force`/`-f` or `--no-safety-check` is given. Matching happens on diff --git a/CHANGELOG.md b/CHANGELOG.md index 57fa9cb..13a46ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- `sshx text` reads remote files through a pipelined SFTP path: read-ahead + aperture plus `UseConcurrentReads`, so the SFTP layer keeps multiple requests + in flight for one file instead of one round trip per read. On the reporting + host the same 8 MiB window went from a median 83.0 s to 26.5 s (88.0/77.9 s → + 23.8/29.1 s, alternating runs, identical bytes and lines scanned). `--max-scan-bytes` + still bounds both the scan and the read-ahead, and the truncation probe reads + exactly one byte past the budget. + +### Added + +- `sshx text` narrates scan progress on stderr after a short grace period + (bytes, percentage, lines, elapsed, matches) and warns when a scan stops at + its `--max-scan-bytes` budget, naming `--offset`/`--tail`/`--pattern`/ + `--max-scan-bytes` as the ways to narrow it. A slow scan closes with the same + advice. stdout stays exactly one JSON document; progress never enters it. +- `sshx.text.v1` `stats.expected_scan_bytes` (additive): the byte budget of the + scanned window, so a caller can size a scan and spot a partial one before + trusting `total_hits_exact`. Omitted when the source size is unknown (journal). +- A policy block now mirrors its reason to stderr in `--json` mode + (`sshx: blocked by safety policy (phase=admission, error_kind=blocked, + executed=false, exit_code=-1)`) plus the flattened reason with the guarded + alternative, so a caller that only prints stdout/stderr no longer sees a + silent refusal. The block predicate and the `sshx sql --docker=` alternative + are documented in README/README_CN/usage, with unit and E2E assertions. +- `sshx` warns when a command contains a non-leading `sudo` (for example + `cd /data/app && sudo docker compose up -d`): password auto-fill only rewrites + a leading `sudo`, so it announces the boundary before connecting and explains + the refusal afterwards, suggesting `sudo sh -c ""`. The auto-fill + scope is unchanged. + +### Fixed + +- `sshx text` no longer looks like a hang: a 60+ second SFTP window used to emit + nothing at all, and a budget-limited scan returned partial results without + saying so. + ## [0.17.0] - 2026-09-16 ### Added diff --git a/README.md b/README.md index 0eac7f4..5b753c2 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,8 @@ It remains a single binary with one-shot invocations and no resident component o 14. Session-bound jump hosts (`via` / `--via=`): nested SSH to private targets with per-hop trust and audit, without local tunnels. 15. `sshx text`: bounded remote log/text dissection (exception blocks, presets, - line windows) over SFTP or sshx-owned journalctl. + line windows) over SFTP or sshx-owned journalctl, with pipelined reads and + stderr scan progress so a wide window is never mistaken for a hang. ## Installation @@ -260,6 +261,22 @@ By default: | `1..254` | Remote command's exit status, propagated verbatim | | `255` | `sshx`-level failure (connect / auth / host-key / timeout / blocked) | +A policy block reaches stderr as well as stdout, so a caller that only prints +the streams sees the reason instead of a silent refusal: + +```text +$ sshx -h=prod-db --json "docker exec pg psql -U app -d app -c 'select 1'" +sshx: blocked by safety policy (phase=admission, error_kind=blocked, executed=false, exit_code=-1); no remote command ran +sshx: block reason: ⚠️ Dangerous command blocked | Command: docker exec pg psql … | Reason: Direct PostgreSQL client execution ("psql") bypasses the guarded SQL pipeline. Use: sshx sql -h= --db= [--docker=] "" (adds classification, backups, and audit) | If you are sure, use --force or -f flag +``` + +The block predicate is machine-readable, and all four fields appear together: +`exit_code=-1`, `error_kind=blocked`, `phase=admission`, `executed=false`. A +blocked command never reaches the network. Use the guarding path instead: +`sshx sql -h= --db= [--docker=] ""`. stdout still +carries exactly one JSON document; the mirror lines are additional, not a +replacement. + ### `--json` structured output Add `--json` to get a single JSON object on stdout (diagnostics still go to @@ -786,6 +803,23 @@ sshx -h=192.168.1.101 -pk=server-B "sudo systemctl restart nginx" sshx -h=192.168.1.102 -pk=server-C "sudo systemctl restart nginx" ``` +**Auto-fill only rewrites the first token.** sshx detects a password prompt and +feeds the stored password only when `sudo` is the command's leading token. A +`sudo` that appears later runs without a password: + +```bash +# NOT filled — sudo is not the leading token; the remote fails with +# "sudo: a password is required" +sshx -h=prod-web "cd /data/app && sudo docker compose up -d" + +# Filled — wrap the whole privileged command +sshx -h=prod-web "sudo sh -c 'cd /data/app && docker compose up -d'" +``` + +sshx warns on stderr when it sees the first form, both before connecting and +again if the remote reports that sudo wanted a password. Auto-fill never +changes the command text in ways the caller did not ask for. + ## Host Key Verification 🔐 `sshx` now enforces strict host key verification just like the OpenSSH client. Instead of silently trusting unknown hosts, the tool reads the trust store from `~/.ssh/known_hosts` (or the path you provide) and aborts the connection if the host is missing or the key changes. diff --git a/README_CN.md b/README_CN.md index 67063a2..a5b4920 100644 --- a/README_CN.md +++ b/README_CN.md @@ -77,7 +77,7 @@ Agent 需要的不是另一个交互式 SSH shell，而是一份稳定、可组 12. 人类专用的 `sshx login`：登录已配置主机，可选 `--sudo` 进入特权壳。不属于 Agent/MCP 契约。 13. 源地址绑定（`--bind=`），对齐 OpenSSH `-b` / `BindAddress` / `BindInterface`。 14. 会话内跳板（`via` / `--via=`）：嵌套 SSH 进入私网目标，每跳独立信任与审计，不在本机开隧道。 -15. `sshx text`：有界远端日志/文本解剖（异常块、preset、行窗），经 SFTP 或 sshx 持有的 journalctl。 +15. `sshx text`：有界远端日志/文本解剖（异常块、preset、行窗），经 SFTP 或 sshx 持有的 journalctl；读取走并发流水线，扫描进度写 stderr，宽窗口不再像卡死。 ## 安装 @@ -259,6 +259,16 @@ sshx --transfer=192.168.1.100:/var/log/app.log --to=192.168.1.101:/backup/app.lo | `1..254` | 远程命令的退出码，原样透传 | | `255` | `sshx` 层面的失败（连接 / 认证 / 主机密钥 / 超时 / 被拦截） | +被安全策略拦截时，除 stdout 的 JSON 外，stderr 也会给出人类可读的原因，只打印两个流的调用方不会把拦截误读成「静默拒绝」： + +```text +$ sshx -h=prod-db --json "docker exec pg exec -U app -d app -c 'select 1'" +sshx: blocked by safety policy (phase=admission, error_kind=blocked, executed=false, exit_code=-1); no remote command ran +sshx: block reason: ⚠️ Dangerous command blocked | ... | Reason: Direct PostgreSQL client execution ("psql") bypasses the guarded SQL pipeline. Use: sshx sql -h= --db= [--docker=] "" ... +``` + +拦截判定是机器可读的，四个字段同时出现：`exit_code=-1`、`error_kind=blocked`、`phase=admission`、`executed=false`。被拦截的命令不会触网。请改用受管路径：`sshx sql -h= --db= [--docker=] ""`。stdout 仍只有一份 JSON 文档，stderr 的两行是补充信息。 + ### `--json` 结构化输出 加上 `--json` 即可在 stdout 得到单个 JSON 对象（诊断日志仍走 stderr，保证 stdout 纯净）： @@ -570,6 +580,18 @@ sshx -h=192.168.1.101 -pk=server-B "sudo systemctl restart nginx" sshx -h=192.168.1.102 -pk=server-C "sudo systemctl restart nginx" ``` +**自动填密只作用于命令的首个 token。** 只有命令以 `sudo` 开头时，sshx 才会识别密码提示并注入已保存的密码；出现在 `&&` 之后、脚本内部或管道中的 `sudo` 不会被填密： + +```bash +# 不会填密 —— sudo 不是首个 token，远端会以 "sudo: a password is required" 失败 +sshx -h=prod-web "cd /data/app && sudo docker compose up -d" + +# 会填密 —— 把整条特权命令包起来 +sshx -h=prod-web "sudo sh -c 'cd /data/app && docker compose up -d'" +``` + +sshx 遇到前一种写法会在 stderr 给出提示：连接前给一次，远端确实报出 sudo 索要密码时再给一次。填密范围本身没有变化。 + ### 密码键名说明 - **master**: 默认的 sudo 密码键名,用于 sudo 命令 diff --git a/docs/text.md b/docs/text.md index 6b3d6fc..914a2c4 100644 --- a/docs/text.md +++ b/docs/text.md @@ -37,3 +37,39 @@ JSON schema `sshx.text.v1`. Branch on `success`, `hits[].kind`, `stats.total_hits` vs `stats.returned`, `truncated`, `truncated_reason`, `redacted`, and `line_origin` (`file` or `scanned_window`). Secret-shaped spans are redacted unless `--no-redact`. Audit events omit hit text. + +`stats` also carries the window accounting a caller needs to judge coverage: + +| Field | Meaning | +| --------------------- | ---------------------------------------------------------------- | +| `file_size` | Remote file size when the source is a file (omitted for journal) | +| `window_start_byte` | First byte the window reads | +| `expected_scan_bytes` | Window budget: the window capped by `--max-scan-bytes` | + +`expected_scan_bytes` is additive and omitted when the source size is unknown +(journal). Compare it with `bytes_scanned`: a scan that stopped early reports +less than it expected, and a scan that hit its budget sets +`truncated_reason=max_scan_bytes` with `total_hits_exact=false`. + +## Monitoring + +A wide window over a slow link used to look like a hang. `sshx text` now +narrates progress on **stderr** while it scans: + +```text +sshx text: scanning 2.0MiB/2097152B of 61.3MiB/64296766B (3%) lines=10240 elapsed=4.1s matches=0 +sshx text: warning: scan stopped at its --max-scan-bytes budget after 12.0s (8.0MiB/8388608B scanned, 8.0MiB/8388608B in window); results are partial (total_hits_exact=false). Narrow with --offset=N or --tail=N, pre-filter with --pattern=..., or raise --max-scan-bytes=N. +``` + +- Progress appears only after a grace period, at most once per second, and + never for a scan that finishes quickly. +- A scan that stops at its byte budget always warns, so partial results are + never mistaken for complete ones. A slow scan closes with the same + `--offset`/`--tail` advice. +- **stdout stays exactly one JSON document.** Progress and advice go to stderr + only, so `--json` parsing is unaffected. + +```bash +# Narrow the window instead of scanning 8MiB to answer a small question +sshx text -h=prod-web --path=/var/log/app.log --offset=41000 --limit=200 --json +``` diff --git a/internal/app/app.go b/internal/app/app.go index 999cdf1..651788a 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "net" "os" "os/signal" @@ -339,6 +340,10 @@ func RunContext(ctx context.Context, args []string) (err error) { config.SudoPassword = password logger.GetLogger().Success("Sudo password will be auto-filled when prompted") } + } else if hint, needed := sshclient.NonLeadingSudoHint(config.Command); needed && config.SudoKey != "" { + // Warn before connecting: this command will prompt for a password that + // no non-interactive session can answer. + logger.GetLogger().Warning("sudo password auto-fill skipped for this command: %s", hint) } // Create SSH client @@ -377,6 +382,7 @@ func runCommand(client *sshclient.SSHClient, config *sshclient.Config, audit *au res, execErr := client.RunCommand(config.JSONOutput) dur := time.Since(start) audit.recordCommandResult(config, client.AuthMethodUsed(), res, dur, classifyError(execErr), execErr) + reportSudoPromptFailure(os.Stderr, config.Command, res.Stderr+res.Stdout) if config.JSONOutput { if outputErr := emitCommandJSON(config, client.AuthMethodUsed(), res, dur, classifyError(execErr), execErr); outputErr != nil { @@ -428,6 +434,19 @@ func reportSSHFailure(config *sshclient.Config, audit *auditRecorder, authMethod return err } +// reportSudoPromptFailure explains a sudo refusal sshx cannot help with: the +// stored password is only injected for a leading sudo, so a mid-command sudo +// stops with "a password is required". It writes straight to stderr rather than +// through the logger, because it explains a failed run to a caller that may be +// running with diagnostics quieted. +func reportSudoPromptFailure(w io.Writer, command, output string) { + hint, needed := sshclient.NonLeadingSudoHint(command) + if !needed || !sshclient.SudoPasswordPromptFailure(output) { + return + } + writeDiagnosticNote(w, "sshx: %s\n", hint) +} + // emitCommandJSON writes a single JSON result line to stdout. Diagnostic logs go // to stderr, so stdout stays a pure machine-readable stream. func emitCommandJSON(config *sshclient.Config, authMethod sshclient.AuthMethod, res sshclient.ExecResult, dur time.Duration, errKind string, execErr error) error { diff --git a/internal/app/diagnostics.go b/internal/app/diagnostics.go new file mode 100644 index 0000000..17286c1 --- /dev/null +++ b/internal/app/diagnostics.go @@ -0,0 +1,19 @@ +package app + +import ( + "fmt" + "io" + + "github.com/talkincode/sshx/pkg/logger" +) + +// writeDiagnosticNote writes a best-effort diagnostic note to w. +// +// Notes explain a result that has already been reported on stdout, so a failed +// write must never change the command's outcome and has nowhere better to go +// than the debug log (which is itself stderr-only, keeping --json stdout pure). +func writeDiagnosticNote(w io.Writer, format string, args ...any) { + if _, err := fmt.Fprintf(w, format, args...); err != nil { + logger.GetLogger().Debug("failed to write diagnostic note: %v", err) + } +} diff --git a/internal/app/lifecycle.go b/internal/app/lifecycle.go index fc841bb..5cdc58b 100644 --- a/internal/app/lifecycle.go +++ b/internal/app/lifecycle.go @@ -4,6 +4,9 @@ import ( "bytes" "encoding/json" "fmt" + "io" + "os" + "strings" "time" "github.com/talkincode/sshx/internal/execution" @@ -111,12 +114,64 @@ func emitLifecycleJSON(config *sshclient.Config, value any) error { if err != nil { return err } + // Write the human-readable mirror first so the JSON document stays the last + // line a caller that merges stderr into stdout would have to parse. + reportPolicyRejection(os.Stderr, document) if err := encodeJSON(document); err != nil { return fmt.Errorf("%w: deliver execution result: %w", execution.ErrLocalIO, err) } return nil } +// reportPolicyRejection mirrors a blocked admission decision to stderr. The +// rejection predicate is machine-readable (exit_code=-1, error_kind=blocked, +// phase=admission, executed=false) but lives inside the stdout document, which +// callers that surface only stdout/stderr used to read as a silent refusal. +// stdout keeps exactly one JSON document, so --json stays machine-parsable. +func reportPolicyRejection(w io.Writer, document map[string]json.RawMessage) { + if documentString(document, "error_kind") != execution.ErrorKindBlocked { + return + } + phase := documentString(document, "phase") + if phase == "" { + phase = execution.PhaseAdmission + } + executed := strings.TrimSpace(string(document["executed"])) + if executed != "true" { + executed = "false" + } + exitCode := -1 + if raw, ok := document["exit_code"]; ok { + _ = json.Unmarshal(raw, &exitCode) //nolint:errcheck // only reads the optional numeric projection + } + writeDiagnosticNote(w, "sshx: blocked by safety policy (phase=%s, error_kind=%s, executed=%s, exit_code=%d); no remote command ran\n", + phase, execution.ErrorKindBlocked, executed, exitCode) + if reason := flattenPolicyReason(documentString(document, "error")); reason != "" { + writeDiagnosticNote(w, "sshx: block reason: %s\n", reason) + } +} + +// flattenPolicyReason collapses a multi-line policy message into one greppable +// line. +func flattenPolicyReason(message string) string { + parts := strings.Split(message, "\n") + kept := make([]string, 0, len(parts)) + for _, part := range parts { + if trimmed := strings.TrimSpace(part); trimmed != "" { + kept = append(kept, trimmed) + } + } + return strings.Join(kept, " | ") +} + +func documentString(document map[string]json.RawMessage, key string) string { + var text string + if raw, ok := document[key]; ok { + _ = json.Unmarshal(raw, &text) //nolint:errcheck // only reads optional string projections + } + return text +} + func finalizeLifecycle(config *sshclient.Config, value any) (map[string]json.RawMessage, error) { data, err := marshalLifecycleValue(value) if err != nil { @@ -158,11 +213,7 @@ func finalizeLifecycle(config *sshclient.Config, value any) (map[string]json.Raw } } stringField := func(key string) string { - var text string - if raw, ok := document[key]; ok { - _ = json.Unmarshal(raw, &text) //nolint:errcheck // only reads optional string projections - } - return text + return documentString(document, key) } code := -1 if raw, ok := document["exit_code"]; ok { diff --git a/internal/app/lifecycle_policy_test.go b/internal/app/lifecycle_policy_test.go new file mode 100644 index 0000000..e2d855b --- /dev/null +++ b/internal/app/lifecycle_policy_test.go @@ -0,0 +1,178 @@ +package app + +import ( + "bytes" + "encoding/json" + "errors" + "io" + "os" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/talkincode/sshx/internal/sshclient" +) + +// captureStreams splits os.Stdout and os.Stderr into separate pipes so a test +// can assert on each stream independently. +func captureStreams(t *testing.T, fn func()) (stdout, stderr []byte) { + t.Helper() + + oldStdout, oldStderr := os.Stdout, os.Stderr + outR, outW, err := os.Pipe() + require.NoError(t, err) + errR, errW, err := os.Pipe() + require.NoError(t, err) + + type result struct { + data []byte + err error + } + read := func(r *os.File) chan result { + ch := make(chan result, 1) + go func() { + var buf bytes.Buffer + _, copyErr := io.Copy(&buf, r) + ch <- result{data: buf.Bytes(), err: copyErr} + }() + return ch + } + outCh, errCh := read(outR), read(errR) + + os.Stdout, os.Stderr = outW, errW + func() { + defer func() { + os.Stdout, os.Stderr = oldStdout, oldStderr + _ = outW.Close() //nolint:errcheck // test cleanup + _ = errW.Close() //nolint:errcheck // test cleanup + }() + fn() + }() + + outRes := <-outCh + errRes := <-errCh + require.NoError(t, outRes.err) + require.NoError(t, errRes.err) + require.NoError(t, outR.Close()) + require.NoError(t, errR.Close()) + return outRes.data, errRes.data +} + +// TestReportPolicyRejectionMirrorsBlockedDecision is the regression guard for +// the silent-refusal report: a blocked admission decision must reach stderr, +// where callers that only print the streams can see it. +func TestReportPolicyRejectionMirrorsBlockedDecision(t *testing.T) { + document := map[string]json.RawMessage{ + "error_kind": json.RawMessage(`"blocked"`), + "phase": json.RawMessage(`"admission"`), + "executed": json.RawMessage(`false`), + "exit_code": json.RawMessage(`-1`), + "error": json.RawMessage(`"⚠️ Dangerous command blocked\nCommand: docker exec c psql -c 'select 1'\nReason: bypasses the guarded SQL pipeline. Use: sshx sql -h= --db= [--docker=] \"\""`), + } + + var out bytes.Buffer + reportPolicyRejection(&out, document) + lines := strings.Split(strings.TrimRight(out.String(), "\n"), "\n") + require.Len(t, lines, 2, "expected a decision line and a reason line") + assert.Contains(t, lines[0], "blocked by safety policy") + assert.Contains(t, lines[0], "phase=admission") + assert.Contains(t, lines[0], "error_kind=blocked") + assert.Contains(t, lines[0], "executed=false") + assert.Contains(t, lines[0], "exit_code=-1") + assert.Contains(t, lines[1], "bypasses the guarded SQL pipeline") + assert.Contains(t, lines[1], "|", "multi-line reasons must be flattened to one line") + assert.Contains(t, lines[1], "sshx sql", "the reason must keep the guarded alternative") +} + +// TestReportPolicyRejectionIgnoresOtherFailures keeps the mirror narrow: only a +// policy block earns it, so ordinary remote failures are not annotated. +func TestReportPolicyRejectionIgnoresOtherFailures(t *testing.T) { + for _, document := range []map[string]json.RawMessage{ + {"error_kind": json.RawMessage(`"remote_error"`), "error": json.RawMessage(`"boom"`)}, + {"error_kind": json.RawMessage(`""`)}, + {}, + } { + var out bytes.Buffer + reportPolicyRejection(&out, document) + assert.Empty(t, out.String()) + } +} + +// TestReportPolicyRejectionDefaultsMissingFields: a blocked decision without +// the optional projections still reads as "nothing ran". +func TestReportPolicyRejectionDefaultsMissingFields(t *testing.T) { + var out bytes.Buffer + reportPolicyRejection(&out, map[string]json.RawMessage{ + "error_kind": json.RawMessage(`"blocked"`), + }) + line := out.String() + assert.Contains(t, line, "phase=admission") + assert.Contains(t, line, "executed=false") + assert.Contains(t, line, "exit_code=-1") + assert.Contains(t, line, "no remote command ran") +} + +// TestBlockedJSONKeepsStdoutPureAndMirrorsStderr runs the real blocked +// admission path: stdout must stay exactly one JSON document while stderr +// carries the human-readable rejection. +func TestBlockedJSONKeepsStdoutPureAndMirrorsStderr(t *testing.T) { + config := &sshclient.Config{ + Mode: "ssh", + Host: "db1.example.net", + Port: "22", + User: "operator", + Command: `docker exec teamsacs_pgdb18 psql -U teamsacs -d teamsacs -At -c 'select 1'`, + JSONOutput: true, + } + blocked := &sshclient.CommandBlockedError{ + Command: config.Command, + Reason: `Direct PostgreSQL client execution ("psql") bypasses the guarded SQL pipeline. Use: sshx sql -h= --db= [--docker=] "" (adds classification, backups, and audit)`, + } + + var runErr error + stdout, stderr := captureStreams(t, func() { + runErr = reportSSHFailure(config, nil, sshclient.AuthMethodUnknown, "blocked", blocked) + }) + require.ErrorIs(t, runErr, ErrReported) + + // stdout: exactly one JSON document, nothing else. + assert.True(t, json.Valid(bytes.TrimSpace(stdout)), "stdout must be a single JSON document: %q", stdout) + assert.Equal(t, 1, strings.Count(strings.TrimSpace(string(stdout)), "\n")+1, "stdout must hold one JSON line") + + var document map[string]any + require.NoError(t, json.Unmarshal(stdout, &document)) + assert.Equal(t, "blocked", document["error_kind"]) + assert.Equal(t, "admission", document["phase"]) + assert.Equal(t, false, document["executed"]) + assert.EqualValues(t, -1, document["exit_code"]) + + // stderr: the rejection a caller can print. + require.NotEmpty(t, stderr, "a blocked --json run must explain itself on stderr") + assert.Contains(t, string(stderr), "blocked by safety policy") + assert.Contains(t, string(stderr), "phase=admission") + assert.Contains(t, string(stderr), "sshx sql") + assert.NotContains(t, string(stdout), "blocked by safety policy", "the mirror must not leak into stdout") +} + +// TestNonBlockedJSONFailureHasNoMirror: a plain remote failure keeps today's +// behavior and writes nothing extra to stderr. +func TestNonBlockedJSONFailureHasNoMirror(t *testing.T) { + config := &sshclient.Config{ + Mode: "ssh", + Host: "db1.example.net", + Port: "22", + User: "operator", + Command: "systemctl restart api", + JSONOutput: true, + } + + var runErr error + stdout, stderr := captureStreams(t, func() { + runErr = reportSSHFailure(config, nil, sshclient.AuthMethodUnknown, "connect", errors.New("dial tcp: connection refused")) + }) + require.ErrorIs(t, runErr, ErrReported) + assert.NotContains(t, string(stderr), "blocked by safety policy") + assert.True(t, json.Valid(bytes.TrimSpace(stdout))) +} diff --git a/internal/app/run.go b/internal/app/run.go index d31f76e..1683e5f 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -121,12 +121,14 @@ func HandleRun(config *sshclient.Config, audit *auditRecorder) error { if execErr != nil { if outcome.RunID != "" { recordRunAudit(audit, config, req, snap, outcome) + reportRunSudoPromptFailures(outcome, req.Action.Command) return execErr } return reportRunRequestFailure(config, audit, execErr) } recordRunAudit(audit, config, req, snap, outcome) + reportRunSudoPromptFailures(outcome, req.Action.Command) // Single-target --json emits one versioned result document. if req.JSONOutput && !req.JSONLOutput && outcome.Single != nil { @@ -146,6 +148,28 @@ func HandleRun(config *sshclient.Config, audit *auditRecorder) error { } } +// reportRunSudoPromptFailures explains a per-target sudo refusal that sshx +// cannot auto-fill: the stored password only rewrites a leading sudo, so a +// mid-command sudo stops with "a password is required". It writes straight to +// stderr because the caller explaining a failure may be running with +// diagnostics quieted, and stdout must keep exactly one result document. +func reportRunSudoPromptFailures(outcome execution.RunOutcome, command string) { + hint, needed := sshclient.NonLeadingSudoHint(command) + if !needed { + return + } + for _, res := range outcome.Results { + if !sshclient.SudoPasswordPromptFailure(res.Stderr + res.Stdout) { + continue + } + if label := res.Target.Alias; label != "" { + fmt.Fprintf(os.Stderr, "sshx: [%s] %s\n", label, hint) + continue + } + fmt.Fprintf(os.Stderr, "sshx: %s\n", hint) + } +} + type executionNoopEvents struct{} func (executionNoopEvents) WriteEvent(execution.Event) error { return nil } @@ -389,6 +413,11 @@ func buildRunRequest(config *sshclient.Config) (*execution.Request, *execution.P req.Action.Intent = execution.IntentRead } } + if hint, needed := sshclient.NonLeadingSudoHint(req.Action.Command); needed { + // Non-interactive sessions cannot answer a mid-command sudo prompt and + // auto-fill only rewrites the leading token. Say so before connecting. + logger.GetLogger().Warning("%s", hint) + } if req.Policy.FailureMode == "" { req.Policy.FailureMode = execution.FailureContinue } diff --git a/internal/app/text.go b/internal/app/text.go index c3c9a9e..663b134 100644 --- a/internal/app/text.go +++ b/internal/app/text.go @@ -2,6 +2,7 @@ package app import ( "fmt" + "os" "strings" "time" @@ -43,13 +44,15 @@ type textJSONResult struct { } type textRun struct { - config *sshclient.Config - audit *auditRecorder - start time.Time - phase string - client *sshclient.SSHClient - req textsafe.Request - scan textsafe.Result + config *sshclient.Config + audit *auditRecorder + start time.Time + phase string + client *sshclient.SSHClient + req textsafe.Request + scan textsafe.Result + reporter *textScanReporter + scanStart time.Time } func textRequestFrom(config *sshclient.Config) (textsafe.Request, error) { @@ -155,6 +158,11 @@ func HandleText(config *sshclient.Config, audit *auditRecorder) (err error) { } func (r *textRun) collect(req textsafe.Request) (textsafe.Result, error) { + // A streamed SFTP window is the only source that can idle for minutes, so it + // is the one that narrates progress. stdout stays untouched because the + // reporter writes to stderr. + r.reporter = newTextScanReporter(os.Stderr, req.Pattern != "") + r.scanStart = time.Now() switch req.Kind { case textsafe.SourceFile: if r.config.TextUseSudo { @@ -178,7 +186,7 @@ func (r *textRun) scanSFTP(req textsafe.Request) (textsafe.Result, error) { req.FileSize = meta.Size req.WindowStartByte = meta.StartByte req.SkipPartialFirst = meta.SkipPartial - return textsafe.Scan(file, req) + return textsafe.ScanWithProgress(file, req, r.reporter) } func (r *textRun) scanPrivilegedFile(req textsafe.Request) (textsafe.Result, error) { @@ -220,6 +228,7 @@ func (r *textRun) succeed() error { r.phase = "complete" r.recordAudit(0, "", nil) result := r.baseResult(true, 0, "", nil) + r.reportScanAdvice() if r.config.JSONOutput { return emitTextJSON(r.config, result) } @@ -230,6 +239,16 @@ func (r *textRun) succeed() error { return nil } +// reportScanAdvice closes a streamed scan with the guidance a caller needs when +// the window was too wide or the scan stopped at its byte budget. It writes to +// stderr only. +func (r *textRun) reportScanAdvice() { + if r.reporter == nil { + return + } + r.reporter.Finish(r.scan, time.Since(r.scanStart)) +} + func (r *textRun) fail(kind string, failErr error) error { r.recordAudit(-1, kind, failErr) result := r.baseResult(false, -1, kind, failErr) diff --git a/internal/app/text_progress.go b/internal/app/text_progress.go new file mode 100644 index 0000000..d46470b --- /dev/null +++ b/internal/app/text_progress.go @@ -0,0 +1,121 @@ +package app + +import ( + "fmt" + "io" + "time" + + "github.com/talkincode/sshx/internal/textsafe" +) + +// Long remote scans used to look like a hang: a 90 s SFTP window produced no +// output at all. The reporter narrates progress on stderr, which keeps stdout a +// single machine-readable JSON document for --json callers. +const ( + // textProgressGrace hides progress for quick scans. + textProgressGrace = 3 * time.Second + // textProgressInterval limits progress lines to one per second. + textProgressInterval = time.Second + // textSlowScanAdvice is the elapsed scan time worth advising the caller to + // narrow the window. + textSlowScanAdvice = 10 * time.Second +) + +// textScanReporter writes bounded scan progress and closing advice to a single +// stream (stderr in production). +type textScanReporter struct { + out io.Writer + pattern bool + start time.Time + lastLine time.Time + sample textsafe.ScanProgress + + grace time.Duration + interval time.Duration + advice time.Duration +} + +func newTextScanReporter(out io.Writer, patternSet bool) *textScanReporter { + return &textScanReporter{ + out: out, + pattern: patternSet, + start: time.Now(), + grace: textProgressGrace, + interval: textProgressInterval, + advice: textSlowScanAdvice, + } +} + +// Progress implements textsafe.ProgressObserver. +func (r *textScanReporter) Progress(p textsafe.ScanProgress) { + r.sample = p + now := time.Now() + if now.Sub(r.start) < r.grace { + return + } + if !r.lastLine.IsZero() && now.Sub(r.lastLine) < r.interval { + return + } + r.lastLine = now + writeDiagnosticNote(r.out, "%s\n", r.progressLine(now)) +} + +func (r *textScanReporter) progressLine(now time.Time) string { + p := r.sample + line := fmt.Sprintf("sshx text: scanning %s", formatByteCount(p.Bytes)) + if p.FileSize > 0 { + line += fmt.Sprintf(" of %s (%d%%)", formatByteCount(p.FileSize), percentOf(p.Bytes, p.FileSize)) + } + line += fmt.Sprintf(" lines=%d elapsed=%s", p.Lines, formatSeconds(now.Sub(r.start))) + if r.pattern { + line += fmt.Sprintf(" matches=%d", p.MatchedLines) + } + return line +} + +// Finish prints advice when a scan was slow or stopped at its byte budget, so +// partial results are never silently mistaken for complete ones. +func (r *textScanReporter) Finish(result textsafe.Result, elapsed time.Duration) { + if r.out == nil { + return + } + if result.Truncated && textsafe.HasTruncationReason(result.TruncatedReason, "max_scan_bytes") { + writeDiagnosticNote(r.out, "sshx text: warning: scan stopped at its --max-scan-bytes budget after %s (%s scanned, %s in window); results are partial "+ + "(total_hits_exact=false). Narrow with --offset=N or --tail=N, pre-filter with --pattern=..., or raise --max-scan-bytes=N.\n", + formatSeconds(elapsed), formatByteCount(result.Stats.BytesScanned), formatByteCount(result.Stats.ExpectedScanBytes)) + return + } + if elapsed >= r.advice { + writeDiagnosticNote(r.out, "sshx text: notice: scan took %s for %s; narrow the window with --offset=N or --tail=N when a smaller window answers the question.\n", + formatSeconds(elapsed), formatByteCount(result.Stats.BytesScanned)) + } +} + +// formatByteCount renders a byte count as an exact integer with a binary unit. +func formatByteCount(n int64) string { + switch { + case n >= 1<<30: + return fmt.Sprintf("%.1fGiB/%dB", float64(n)/(1<<30), n) + case n >= 1<<20: + return fmt.Sprintf("%.1fMiB/%dB", float64(n)/(1<<20), n) + case n >= 1<<10: + return fmt.Sprintf("%.1fKiB/%dB", float64(n)/(1<<10), n) + default: + return fmt.Sprintf("%dB", n) + } +} + +func percentOf(part, whole int64) int { + if whole <= 0 { + return 0 + } + pct := part * 100 / whole + if pct > 100 { + return 100 + } + return int(pct) +} + +func formatSeconds(d time.Duration) string { + return fmt.Sprintf("%.1fs", d.Seconds()) +} diff --git a/internal/app/text_progress_test.go b/internal/app/text_progress_test.go new file mode 100644 index 0000000..22a6a01 --- /dev/null +++ b/internal/app/text_progress_test.go @@ -0,0 +1,274 @@ +package app + +import ( + "bytes" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/talkincode/sshx/internal/execution" + "github.com/talkincode/sshx/internal/textsafe" +) + +func newTestReporter(out *bytes.Buffer, patternSet bool) *textScanReporter { + r := newTextScanReporter(out, patternSet) + r.grace = 0 + r.interval = 0 + r.advice = time.Hour + return r +} + +// TestTextScanReporterStaysQuietDuringGrace: a fast scan must not print +// anything, so ordinary runs keep their output unchanged. +func TestTextScanReporterStaysQuietDuringGrace(t *testing.T) { + var out bytes.Buffer + reporter := newTextScanReporter(&out, true) + reporter.Progress(textsafe.ScanProgress{Bytes: 4096, Lines: 10, FileSize: 1 << 20}) + assert.Empty(t, out.String()) +} + +// TestTextScanReporterPrintsBoundedProgress covers the monitored shape: one +// line per interval with bytes, percentage, lines, elapsed, and matches. +func TestTextScanReporterPrintsBoundedProgress(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, true) + reporter.interval = time.Hour // only the first sample may print + + reporter.Progress(textsafe.ScanProgress{Bytes: 512 << 10, Lines: 2048, MatchedLines: 7, FileSize: 2 << 20}) + + lines := strings.Split(strings.TrimRight(out.String(), "\n"), "\n") + require.Len(t, lines, 1, "progress must be throttled to one line per interval") + assert.Contains(t, lines[0], "sshx text: scanning") + assert.Contains(t, lines[0], "512.0KiB/524288B") + assert.Contains(t, lines[0], " of 2.0MiB/2097152B") + assert.Contains(t, lines[0], "(25%)") + assert.Contains(t, lines[0], "lines=2048") + assert.Contains(t, lines[0], "matches=7") + + reporter.Progress(textsafe.ScanProgress{Bytes: 1 << 20, Lines: 4096, FileSize: 2 << 20}) + assert.Len(t, strings.Split(strings.TrimRight(out.String(), "\n"), "\n"), 1) +} + +// TestTextScanReporterOmitsMatchesWithoutPattern keeps the line honest for a +// preset-only scan, where no match count exists. +func TestTextScanReporterOmitsMatchesWithoutPattern(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.Progress(textsafe.ScanProgress{Bytes: 1 << 20, Lines: 10, FileSize: 1 << 20}) + assert.NotContains(t, out.String(), "matches=") + assert.Contains(t, out.String(), "(100%)") +} + +// TestTextScanReporterUnknownSizeOmitsPercentage: a journal scan has no file +// size, so progress must not invent a percentage. +func TestTextScanReporterUnknownSizeOmitsPercentage(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.Progress(textsafe.ScanProgress{Bytes: 1 << 20, Lines: 10}) + assert.NotContains(t, out.String(), "%") + assert.Contains(t, out.String(), "1.0MiB/1048576B") +} + +// TestTextScanReporterWarnsOnByteBudgetTruncation is the "partial results" +// guard: stopping at --max-scan-bytes must say so and name the narrower +// alternatives. +func TestTextScanReporterWarnsOnByteBudgetTruncation(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.Finish(textsafe.Result{ + Truncated: true, + TruncatedReason: "max_scan_bytes", + Stats: textsafe.Stats{BytesScanned: 1 << 20, ExpectedScanBytes: 1 << 20}, + }, 4*time.Second) + + line := out.String() + assert.Contains(t, line, "warning") + assert.Contains(t, line, "--max-scan-bytes") + assert.Contains(t, line, "--offset") + assert.Contains(t, line, "--tail") + assert.Contains(t, line, "partial") + assert.Contains(t, line, "total_hits_exact=false") +} + +// TestTextScanReporterWarnsWhenReasonsAreJoined guards the multi-reason form of +// truncated_reason: a scan that also capped its hit list reports +// "max_scan_bytes,max_hits", and the byte-budget warning must still appear. +func TestTextScanReporterWarnsWhenReasonsAreJoined(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.Finish(textsafe.Result{ + Truncated: true, + TruncatedReason: "max_scan_bytes,max_hits", + Stats: textsafe.Stats{BytesScanned: 1 << 20, ExpectedScanBytes: 1 << 20}, + }, 4*time.Second) + + assert.Contains(t, out.String(), "warning") + assert.Contains(t, out.String(), "--max-scan-bytes") +} + +// TestTextScanReporterSkipsOtherTruncationReasons: a capped hit list is not a +// bounded scan, so it must not claim the window was cut short. +func TestTextScanReporterSkipsOtherTruncationReasons(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.Finish(textsafe.Result{ + Truncated: true, + TruncatedReason: "max_hits", + Stats: textsafe.Stats{BytesScanned: 1 << 20, ExpectedScanBytes: 1 << 20}, + }, 4*time.Second) + + assert.NotContains(t, out.String(), "warning") +} + +// TestTextScanReporterAdvisesOnSlowScan covers the second trigger: a scan that +// covered its window but took long enough that a smaller window is worth +// suggesting next time. +func TestTextScanReporterAdvisesOnSlowScan(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.advice = time.Second + reporter.Finish(textsafe.Result{Stats: textsafe.Stats{BytesScanned: 8 << 20}}, 12*time.Second) + + line := out.String() + assert.Contains(t, line, "notice") + assert.Contains(t, line, "8.0MiB/8388608B") + assert.Contains(t, line, "--offset") +} + +// TestTextScanReporterQuietOnFastCompleteScan: no warning, no notice. +func TestTextScanReporterQuietOnFastCompleteScan(t *testing.T) { + var out bytes.Buffer + reporter := newTestReporter(&out, false) + reporter.advice = time.Minute + reporter.Finish(textsafe.Result{Stats: textsafe.Stats{BytesScanned: 4096}}, 10*time.Millisecond) + assert.Empty(t, out.String()) +} + +func TestTextProgressFormatters(t *testing.T) { + assert.Equal(t, "512B", formatByteCount(512)) + assert.Equal(t, "1.5KiB/1536B", formatByteCount(1536)) + assert.Equal(t, "1.5MiB/1572864B", formatByteCount(1536<<10)) + assert.Equal(t, "2.0GiB/2147483648B", formatByteCount(2<<30)) + assert.Equal(t, 50, percentOf(1<<20, 2<<20)) + assert.Equal(t, 100, percentOf(9<<20, 2<<20)) + assert.Equal(t, 0, percentOf(10, 0)) + assert.Equal(t, "1.5s", formatSeconds(1500*time.Millisecond)) +} + +// TestReportSudoPromptFailureExplainsTheBoundary: when the remote refuses for +// want of a password that auto-fill never supplies, the run must say why. +func TestReportSudoPromptFailureExplainsTheBoundary(t *testing.T) { + tests := []struct { + name string + command string + output string + want bool + }{ + { + name: "mid-command sudo refused", + command: `cd /data/appdata/teamsacs && sudo docker compose up -d`, + output: "sudo: a password is required\n", + want: true, + }, + { + name: "leading sudo is auto-filled", + command: `sudo docker compose up -d`, + output: "sudo: a password is required\n", + want: false, + }, + { + name: "mid-command sudo succeeded", + command: `cd /srv && sudo systemctl restart api`, + output: "", + want: false, + }, + { + name: "unrelated failure", + command: `cd /srv && sudo systemctl restart api`, + output: "bash: systemctl: command not found\n", + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var out bytes.Buffer + reportSudoPromptFailure(&out, tc.command, tc.output) + if !tc.want { + assert.Empty(t, out.String()) + return + } + assert.Contains(t, out.String(), "sudo is not the first token") + assert.Contains(t, out.String(), `sudo sh -c`) + }) + } +} + +func TestReportRunSudoPromptFailuresPerTarget(t *testing.T) { + command := `cd /data/appdata/teamsacs && sudo docker compose up -d` + outcome := execution.RunOutcome{ + RunID: "run-test", + Results: []execution.TargetResult{ + { + Target: execution.ResolvedTarget{Alias: "prod-web", Address: "192.0.2.10"}, + Stderr: "sudo: a password is required\n", + }, + { + Target: execution.ResolvedTarget{Alias: "prod-edge", Address: "192.0.2.11"}, + Stderr: "container started\n", + }, + }, + } + + _, stderr := captureStreams(t, func() { + reportRunSudoPromptFailures(outcome, command) + }) + + // Only the target that actually hit the prompt is named. + assert.Contains(t, string(stderr), "[prod-web]") + assert.NotContains(t, string(stderr), "[prod-edge]") + assert.Contains(t, string(stderr), "sudo is not the first token") +} + +func TestReportRunSudoPromptFailuresStaysQuiet(t *testing.T) { + refusal := execution.RunOutcome{ + RunID: "run-test", + Results: []execution.TargetResult{{ + Target: execution.ResolvedTarget{Alias: "prod-web"}, + Stderr: "sudo: a password is required\n", + }}, + } + + tests := []struct { + name string + outcome execution.RunOutcome + command string + }{ + {name: "leading sudo is auto-filled", outcome: refusal, command: `sudo docker compose up -d`}, + {name: "no sudo at all", outcome: refusal, command: `docker compose ps`}, + { + name: "no sudo prompt in the output", + outcome: execution.RunOutcome{ + RunID: "run-test", + Results: []execution.TargetResult{{ + Target: execution.ResolvedTarget{Alias: "prod-web"}, + Stderr: "bash: docker: command not found\n", + }}, + }, + command: `cd /srv && sudo docker compose up -d`, + }, + {name: "no targets ran", outcome: execution.RunOutcome{RunID: "run-test"}, command: `cd /srv && sudo id`}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, stderr := captureStreams(t, func() { + reportRunSudoPromptFailures(tc.outcome, tc.command) + }) + assert.Empty(t, string(stderr)) + }) + } +} diff --git a/internal/app/usage.go b/internal/app/usage.go index 1c63be6..be0fff1 100644 --- a/internal/app/usage.go +++ b/internal/app/usage.go @@ -132,6 +132,12 @@ Agent / Scripting Mode: error_kind (timeout, auth, host_key, connect, blocked, exit_missing, config, error), so it is always distinguishable from a remote exit 255. + A policy block also writes its reason to stderr + (exit_code=-1, error_kind=blocked, phase=admission, executed=false), so a + caller that only prints the streams sees why nothing ran. Use the guarded + alternative instead: sshx sql -h= --db= [--docker=] + "". stdout still carries exactly one JSON document. + Trust note: high-risk bypasses (force, no-safety-check, accept-unknown-host, insecure-hostkey) require explicit CLI flags. Inherited env values and working-directory .env files are ignored for those decisions. @@ -145,6 +151,10 @@ Sudo Auto-fill: sshx -h=host "sh -c 'sudo whoami'" sshx -h=host "echo sudo" + sshx warns on stderr when it sees a non-leading sudo, because the remote then + stops with "sudo: a password is required". Wrap the privileged part instead: + sshx -h=host "sudo sh -c 'cd /data/app && docker compose up -d'" + This keeps keyring lookup, stdin password injection, and future audit fields on one clear rule. Put sudo at the beginning of the remote command when you want sshx to auto-fill it. @@ -783,6 +793,14 @@ There is no --command. JSON schema is sshx.text.v1. Branch on success, hits[].kind, stats.total_hits vs returned, truncated, truncated_reason, and line_origin (file vs scanned_window). +Monitoring: + A scan that runs longer than a few seconds narrates progress on stderr + (bytes, percentage, lines, elapsed, matches). Long or budget-limited scans + close with advice naming --offset/--tail/--max-scan-bytes. stderr carries + these lines only: stdout stays exactly one JSON document. stats gains + expected_scan_bytes (the window budget) alongside file_size/window_start_byte, + so a caller can size a scan before trusting total_hits_exact. + Examples: sshx text -h=prod-web --path=/var/log/nginx/error.log --preset=exception --json sshx text -h=prod-web --journal=nginx.service --since=1h --preset=error --json diff --git a/internal/sshclient/remote_state.go b/internal/sshclient/remote_state.go index 9ce220f..21b9931 100644 --- a/internal/sshclient/remote_state.go +++ b/internal/sshclient/remote_state.go @@ -15,9 +15,9 @@ import ( // RemoteHome resolves the authenticated user's home directory through SFTP. func (c *SSHClient) RemoteHome() (string, error) { - client, err := sftp.NewClient(c.client) + client, err := c.newSFTPClient() if err != nil { - return "", fmt.Errorf("open SFTP session: %w", err) + return "", err } defer func() { _ = client.Close() }() //nolint:errcheck // best-effort close after read-only query home, err := client.RealPath(".") @@ -36,9 +36,9 @@ func (c *SSHClient) ReadRemoteFile(remotePath string, limit int64, expectedUID s if err := validateAbsoluteRemotePath(remotePath); err != nil { return nil, err } - client, clientErr := sftp.NewClient(c.client) + client, clientErr := c.newSFTPClient() if clientErr != nil { - return nil, fmt.Errorf("open SFTP session: %w", clientErr) + return nil, clientErr } defer func() { _ = client.Close() }() //nolint:errcheck // best-effort close info, statErr := client.Lstat(remotePath) @@ -127,9 +127,9 @@ func (c *SSHClient) WriteRemoteFileAtomic(remotePath string, data []byte) error if err := validateAbsoluteRemotePath(remotePath); err != nil { return err } - client, clientErr := sftp.NewClient(c.client) + client, clientErr := c.newSFTPClient() if clientErr != nil { - return fmt.Errorf("open SFTP session: %w", clientErr) + return clientErr } defer func() { _ = client.Close() }() //nolint:errcheck // best-effort close dir := path.Dir(remotePath) diff --git a/internal/sshclient/sftp_read.go b/internal/sshclient/sftp_read.go new file mode 100644 index 0000000..e3c9b03 --- /dev/null +++ b/internal/sshclient/sftp_read.go @@ -0,0 +1,103 @@ +package sshclient + +import ( + "errors" + "io" + + "github.com/pkg/sftp" +) + +// SFTP read pipelining. +// +// The SFTP protocol is request/response: every read the caller asks for costs a +// round trip unless several requests for the same file are in flight. pkg/sftp +// only overlaps round trips when it is handed a buffer larger than its 32 KiB +// packet size — see sftp.File.readAt, which splits such a read into +// maxPacket-sized requests bounded by MaxConcurrentRequestsPerFile. +// +// A line scanner asks for a few kilobytes at a time, so a plain file.Read loop +// is RTT-bound: on a ~350 ms link an 8 MiB window costs ~256 sequential round +// trips (observed: 90 s). Handing the layer 1 MiB reads keeps up to +// sftpMaxConcurrentRequests requests in flight, which turns the same window +// into ~8 waves. +const ( + // sftpReadApertureBytes is the read-ahead buffer handed to the SFTP layer. + // It is deliberately equal to sftpMaxConcurrentRequests * 32 KiB so the + // pipelining window, not the buffer, stays the limiting factor. + sftpReadApertureBytes = 1 << 20 + // sftpMinPipelinedReadBytes is the smallest caller read worth pipelining. + // Smaller reads are served straight from the source so the truncation probe + // (a single byte past --max-scan-bytes) never over-fetches. + sftpMinPipelinedReadBytes = 32 << 10 + // sftpMaxConcurrentRequests bounds in-flight SFTP requests per file. + sftpMaxConcurrentRequests = 32 +) + +// sftpClientOptions makes the pipelining window explicit instead of relying on +// package defaults, so a dependency upgrade cannot silently degrade a scan back +// to one round trip per read. +func sftpClientOptions() []sftp.ClientOption { + return []sftp.ClientOption{ + sftp.UseConcurrentReads(true), + sftp.MaxConcurrentRequestsPerFile(sftpMaxConcurrentRequests), + } +} + +// pipelinedReader reads through a read-ahead aperture so the SFTP layer can +// overlap round trips. Reads smaller than sftpMinPipelinedReadBytes bypass the +// aperture, and budget bounds how much read-ahead may ever pull, so the bytes +// offered to a caller are identical to a plain file.Read loop. +type pipelinedReader struct { + src io.Reader + buf []byte + start int + end int + budget int64 + pending error +} + +func newPipelinedReader(src io.Reader, aperture int, budget int64) *pipelinedReader { + if aperture <= 0 { + aperture = sftpReadApertureBytes + } + return &pipelinedReader{src: src, buf: make([]byte, aperture), budget: budget} +} + +func (p *pipelinedReader) Read(b []byte) (int, error) { + if p.start == p.end { + if p.pending != nil { + err := p.pending + p.pending = nil + return 0, err + } + if len(b) < sftpMinPipelinedReadBytes || p.budget <= 0 { + return p.src.Read(b) + } + if err := p.fill(); err != nil { + return 0, err + } + } + n := copy(b, p.buf[p.start:p.end]) + p.start += n + return n, nil +} + +// fill tops the aperture up from the source. Bytes already buffered are served +// before any deferred error, matching io.Reader semantics. +func (p *pipelinedReader) fill() error { + want := int64(len(p.buf)) + if p.budget < want { + want = p.budget + } + n, err := p.src.Read(p.buf[:want]) + p.start, p.end = 0, n + p.budget -= int64(n) + if n > 0 { + p.pending = err + return nil + } + if errors.Is(err, io.EOF) { + return io.EOF + } + return err +} diff --git a/internal/sshclient/sftp_read_test.go b/internal/sshclient/sftp_read_test.go new file mode 100644 index 0000000..3461cdc --- /dev/null +++ b/internal/sshclient/sftp_read_test.go @@ -0,0 +1,189 @@ +package sshclient + +import ( + "bytes" + "io" + "testing" +) + +// recordingReader counts how many bytes were pulled from the source and the +// size of each request, which is what the pipelining window depends on. +type recordingReader struct { + data []byte + pos int + pulls int + pullMax int +} + +func (r *recordingReader) Read(p []byte) (int, error) { + if r.pos >= len(r.data) { + return 0, io.EOF + } + n := copy(p, r.data[r.pos:]) + r.pos += n + r.pulls++ + if n > r.pullMax { + r.pullMax = n + } + return n, nil +} + +// TestPipelinedReaderServesSameBytes feeds the same source through the +// pipelined reader and compares with the original: the bytes and the EOF +// position must match exactly, otherwise read-ahead would change scan results. +func TestPipelinedReaderServesSameBytes(t *testing.T) { + source := make([]byte, 3*(1<<20)+12345) + for i := range source { + source[i] = byte(i % 251) + } + + pr := newPipelinedReader(&recordingReader{data: source}, 1<<20, int64(len(source))) + var got bytes.Buffer + buf := make([]byte, 64<<10) + for { + n, err := pr.Read(buf) + got.Write(buf[:n]) + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read failed: %v", err) + } + } + if !bytes.Equal(got.Bytes(), source) { + t.Fatalf("pipelined reader returned %d bytes differing from the source", got.Len()) + } +} + +// TestPipelinedReaderSmallReadsBypassAperture pins the truncation-probe +// guarantee: a caller read below the pipelining threshold must pull exactly +// what it asked for, never a read-ahead block. +func TestPipelinedReaderSmallReadsBypassAperture(t *testing.T) { + src := &recordingReader{data: make([]byte, 1<<20)} + pr := newPipelinedReader(src, 1<<20, 1<<20) + + one := make([]byte, 1) + if _, err := pr.Read(one); err != nil { + t.Fatalf("read failed: %v", err) + } + if src.pos != 1 { + t.Fatalf("1-byte read pulled %d bytes from the source, want 1", src.pos) + } + + two := make([]byte, 4096) + if _, err := pr.Read(two); err != nil { + t.Fatalf("read failed: %v", err) + } + if src.pos != 1+len(two) { + t.Fatalf("4 KiB read pulled %d bytes total, want %d", src.pos, 1+len(two)) + } +} + +// TestPipelinedReaderLargeReadsUseAperture verifies the read-ahead path is +// actually taken, and that its requests are aperture-sized rather than +// line-sized, which is what lets pkg/sftp overlap round trips. +func TestPipelinedReaderLargeReadsUseAperture(t *testing.T) { + const aperture = 1 << 20 + src := &recordingReader{data: make([]byte, 5<<20)} + pr := newPipelinedReader(src, aperture, 5<<20) + + buf := make([]byte, 64<<10) + if _, err := pr.Read(buf); err != nil { + t.Fatalf("read failed: %v", err) + } + if src.pulls != 1 { + t.Fatalf("source pulled %d times, want a single read-ahead", src.pulls) + } + if src.pullMax != aperture { + t.Fatalf("read-ahead was %d bytes, want the %d-byte aperture", src.pullMax, aperture) + } + // The caller asked for 64 KiB and got it from the aperture. + if src.pos != aperture { + t.Fatalf("source advanced to %d, want %d", src.pos, aperture) + } +} + +// TestPipelinedReaderBudgetBoundsReadAhead is the --max-scan-bytes guard: the +// read-ahead may never pull past the window budget, so a scan cannot fetch +// remote bytes the caller was not allowed to read. +func TestPipelinedReaderBudgetBoundsReadAhead(t *testing.T) { + const budget = 2 << 20 + src := &recordingReader{data: make([]byte, 8<<20)} + pr := newPipelinedReader(src, 1<<20, budget) + + remaining := int64(budget) + buf := make([]byte, 64<<10) + for remaining > 0 { + want := int64(len(buf)) + if remaining < want { + want = remaining + } + n, err := pr.Read(buf[:want]) + if err != nil { + t.Fatalf("read failed: %v", err) + } + remaining -= int64(n) + } + if int64(src.pos) != budget { + t.Fatalf("source advanced to %d after the budget was consumed, want %d", src.pos, budget) + } + + // Past the budget the reader serves the caller directly, which is how the + // truncation probe sees one byte beyond --max-scan-bytes and no more. + if _, err := pr.Read(buf[:64<<10]); err != nil { + t.Fatalf("read past budget failed: %v", err) + } + if src.pos != budget+(64<<10) { + t.Fatalf("read past the budget pulled %d bytes, want %d", src.pos, budget+(64<<10)) + } +} + +// TestPipelinedReaderPreservesFinalPartialRead covers a source that returns +// data and io.EOF together: the bytes must be delivered before the error. +func TestPipelinedReaderPreservesFinalPartialRead(t *testing.T) { + pr := newPipelinedReader(&eofWithDataReader{data: []byte("tail-bytes")}, 64<<10, 1<<20) + + var got bytes.Buffer + buf := make([]byte, 64<<10) + for { + n, err := pr.Read(buf) + got.Write(buf[:n]) + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read failed: %v", err) + } + } + if got.String() != "tail-bytes" { + t.Fatalf("got %q, want %q", got.String(), "tail-bytes") + } +} + +type eofWithDataReader struct { + data []byte + done bool +} + +func (r *eofWithDataReader) Read(p []byte) (int, error) { + if r.done { + return 0, io.EOF + } + r.done = true + return copy(p, r.data), io.EOF +} + +// TestSFTPClientOptionsEnablePipelining guards the dependency-dependent part of +// the fix: the options that keep pkg/sftp overlapping reads must stay wired. +func TestSFTPClientOptionsEnablePipelining(t *testing.T) { + opts := sftpClientOptions() + if len(opts) != 2 { + t.Fatalf("sftpClientOptions returned %d options, want 2 (UseConcurrentReads + MaxConcurrentRequestsPerFile)", len(opts)) + } + if sftpMaxConcurrentRequests <= 1 { + t.Fatalf("sftpMaxConcurrentRequests = %d, want a pipelining window", sftpMaxConcurrentRequests) + } + if sftpReadApertureBytes < sftpMaxConcurrentRequests<<10 { + t.Fatalf("aperture %d cannot keep %d requests in flight", sftpReadApertureBytes, sftpMaxConcurrentRequests) + } +} diff --git a/internal/sshclient/text.go b/internal/sshclient/text.go index a8776cb..1e07d33 100644 --- a/internal/sshclient/text.go +++ b/internal/sshclient/text.go @@ -28,9 +28,9 @@ func (c *SSHClient) OpenRemoteText(remotePath string, fromEnd bool, maxBytes int if maxBytes <= 0 { maxBytes = 8 << 20 } - client, err := sftp.NewClient(c.client) + client, err := c.newSFTPClient() if err != nil { - return nil, meta, fmt.Errorf("open SFTP session: %w", err) + return nil, meta, err } info, statErr := client.Lstat(remotePath) if statErr != nil { @@ -56,14 +56,26 @@ func (c *SSHClient) OpenRemoteText(remotePath string, fromEnd bool, maxBytes int } meta.SkipPartial = true } - return &sftpReadCloser{File: file, session: client}, meta, nil + // maxBytes bounds read-ahead too: a scan never pulls more than its window + // from the remote file, except for the single probe byte that detects + // --max-scan-bytes truncation. + return &sftpReadCloser{ + reader: newPipelinedReader(file, sftpReadApertureBytes, maxBytes), + File: file, + session: client, + }, meta, nil } type sftpReadCloser struct { + reader io.Reader *sftp.File session *sftp.Client } +func (f *sftpReadCloser) Read(p []byte) (int, error) { + return f.reader.Read(p) +} + func (f *sftpReadCloser) Close() error { fileErr := f.File.Close() sessErr := f.session.Close() diff --git a/internal/sshclient/transport.go b/internal/sshclient/transport.go index 0166dec..82276bb 100644 --- a/internal/sshclient/transport.go +++ b/internal/sshclient/transport.go @@ -91,7 +91,7 @@ func (c *SSHClient) newSFTPClient() (*sftp.Client, error) { if err != nil { return nil, err } - client, err := sftp.NewClient(conn) + client, err := sftp.NewClient(conn, sftpClientOptions()...) if err != nil { return nil, c.transportError("remote_io", "open SFTP session", err) } diff --git a/internal/sshclient/validate.go b/internal/sshclient/validate.go index cdfc251..0943e30 100644 --- a/internal/sshclient/validate.go +++ b/internal/sshclient/validate.go @@ -116,6 +116,102 @@ func isCommandWhitespace(ch byte) bool { return ch == ' ' || ch == '\t' || ch == '\r' || ch == '\n' } +// NonLeadingSudoHint reports a `sudo` invocation in command position that sshx +// cannot feed a password to, because auto-fill only rewrites a leading `sudo` +// token (see sudoStdinCommand). `cd /data/app && sudo docker compose up -d` +// therefore runs verbatim and sudo prompts for a password it can never read +// over a non-interactive session. +// +// The returned hint tells the operator how to restructure the command. The +// auto-fill scope itself is unchanged: sshx still fills only the leading form. +func NonLeadingSudoHint(command string) (string, bool) { + trimmed := strings.TrimSpace(command) + if trimmed == "" { + return "", false + } + if _, ok := leadingSudoRemainder(trimmed); ok { + return "", false + } + if !sudoInCommandPosition(trimmed, 0) { + return "", false + } + return "sudo is not the first token, so sshx cannot auto-fill a stored sudo password " + + "(auto-fill only rewrites a leading `sudo`). Run the privileged part as " + + "`sudo sh -c \"\"`, or move `sudo` to the front.", true +} + +// sudoInCommandPosition reports whether any shell segment executes sudo as its +// command, recursing into `sh -c '