From e49c458901db452fd0c555cdf806f1c4908a83b4 Mon Sep 17 00:00:00 2001 From: DivanMe <48186011+Divaaaan@users.noreply.github.com> Date: Tue, 15 Sep 2026 23:25:05 +0300 Subject: [PATCH 1/3] fix: make node checks fail honestly on probe startup --- .github/scripts/workflows.test.mjs | 20 + .github/workflows/ci.yml | 2 + adapters/internal/processlog/writer.go | 94 ++++ adapters/internal/processlog/writer_test.go | 56 ++ adapters/linux/runner.go | 45 +- adapters/linux/runner_completion_test.go | 131 +++++ adapters/macos/runner.go | 45 +- adapters/macos/runner_completion_test.go | 131 +++++ adapters/windows/runner.go | 50 +- adapters/windows/runner_completion_test.go | 138 +++++ core/control/daemon.go | 21 +- core/control/nodecheck.go | 437 ++++++++++++++-- core/control/nodecheck_budget_test.go | 21 +- core/control/nodecheck_test.go | 544 +++++++++++++++++++- core/singbox/check_e2e_test.go | 34 ++ core/singbox/probe.go | 39 +- core/singbox/probe_test.go | 47 ++ 17 files changed, 1662 insertions(+), 193 deletions(-) create mode 100644 adapters/internal/processlog/writer.go create mode 100644 adapters/internal/processlog/writer_test.go create mode 100644 adapters/linux/runner_completion_test.go create mode 100644 adapters/macos/runner_completion_test.go create mode 100644 adapters/windows/runner_completion_test.go diff --git a/.github/scripts/workflows.test.mjs b/.github/scripts/workflows.test.mjs index fd7d6987..7f7dddca 100644 --- a/.github/scripts/workflows.test.mjs +++ b/.github/scripts/workflows.test.mjs @@ -130,6 +130,26 @@ test("every Go setup resolves one exact committed patch", () => { assert.ok(checked >= 3); }); +test("Windows desktop validates the authenticated probe config with its fetched engine", () => { + const desktop = jobs(workflow("ci.yml")).get("desktop"); + assert.ok(desktop, "ci.yml has no desktop job"); + const desktopSteps = steps(desktop); + const fetchIndex = desktopSteps.findIndex((step) => step.includes("name: Fetch sing-box and wintun")); + assert.notEqual(fetchIndex, -1, "desktop job has no resource fetch step"); + + const probeIndexes = desktopSteps + .map((step, index) => ({ step, index })) + .filter(({ step }) => /^\s*run:\s*go test \.\/core\/singbox -run '\^TestProbeConfigPassesSingBoxCheck\$' -count=1\s*$/m.test(step)); + assert.deepEqual( + probeIndexes.map(({ index }) => index), + [fetchIndex + 1], + "the mandatory probe schema check must run exactly once, immediately after fetching the pinned engine", + ); + assert.match(probeIndexes[0].step, /name: Validate authenticated probe config/); + assert.doesNotMatch(probeIndexes[0].step, /^\s*if:/m); + assert.doesNotMatch(probeIndexes[0].step, /continue-on-error:/); +}); + test("the Arch attach step names the repository instead of asking git", () => { // The build step chowns the checkout to `builder` so makepkg can run, and this // step runs as root: gh's own repository resolution shells out to git, git diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dc1c3413..a32318fa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,6 +98,8 @@ jobs: run: cargo fmt --check - name: Fetch sing-box and wintun run: powershell -ExecutionPolicy Bypass -File scripts/fetch-resources.ps1 + - name: Validate authenticated probe config + run: go test ./core/singbox -run '^TestProbeConfigPassesSingBoxCheck$' -count=1 - name: Build core sidecar run: go build -o ui-desktop/src-tauri/binaries/tenebra-core-x86_64-pc-windows-msvc.exe ./cmd/tenebra-core - name: Lint the Rust backend diff --git a/adapters/internal/processlog/writer.go b/adapters/internal/processlog/writer.go new file mode 100644 index 00000000..92249fc5 --- /dev/null +++ b/adapters/internal/processlog/writer.go @@ -0,0 +1,94 @@ +// Package processlog adapts process stdout and stderr to line-oriented log +// callbacks without taking ownership of the process pipes. os/exec can then +// bound pipe-copy completion with Cmd.WaitDelay while callers still receive an +// unterminated final line before process completion is published. +package processlog + +import ( + "bytes" + "sync" +) + +const maxPendingBytes = 1 << 20 + +// Writer buffers a partial line across Write calls and emits complete lines. +// It is safe for concurrent Write and Flush calls, although each process stream +// normally has its own Writer. +type Writer struct { + mu sync.Mutex + pending []byte + lineHasData bool + emit func(string) +} + +// New creates a line writer that calls emit once for each completed line. +func New(emit func(string)) *Writer { + return &Writer{emit: emit} +} + +// Write implements io.Writer. A pathological line is emitted in bounded chunks +// so a child process cannot grow the pending buffer without limit. +func (w *Writer) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + + written := len(p) + for len(p) > 0 { + newline := bytes.IndexByte(p, '\n') + if newline >= 0 { + w.appendBounded(p[:newline]) + if len(w.pending) > 0 || !w.lineHasData { + w.emitPending() + } + w.lineHasData = false + p = p[newline+1:] + continue + } + w.appendBounded(p) + break + } + return written, nil +} + +func (w *Writer) appendBounded(p []byte) { + if len(p) > 0 { + w.lineHasData = true + } + for len(p) > 0 { + room := maxPendingBytes - len(w.pending) + if room > len(p) { + room = len(p) + } + w.pending = append(w.pending, p[:room]...) + p = p[room:] + if len(w.pending) == maxPendingBytes { + w.emitPending() + } + } +} + +// Flush emits an unterminated final line. Call it only after Cmd.Wait returns, +// when os/exec's writer-copy goroutines have completed or WaitDelay closed their +// pipes. +func (w *Writer) Flush() { + w.mu.Lock() + defer w.mu.Unlock() + if len(w.pending) > 0 { + w.emitPending() + } + w.lineHasData = false +} + +func (w *Writer) emitPending() { + w.emitBytes(w.pending) + w.pending = w.pending[:0] +} + +func (w *Writer) emitBytes(line []byte) { + if len(line) > 0 && line[len(line)-1] == '\r' { + line = line[:len(line)-1] + } + if w.emit != nil { + w.emit(string(line)) + } +} diff --git a/adapters/internal/processlog/writer_test.go b/adapters/internal/processlog/writer_test.go new file mode 100644 index 00000000..c13a038a --- /dev/null +++ b/adapters/internal/processlog/writer_test.go @@ -0,0 +1,56 @@ +package processlog + +import ( + "bytes" + "reflect" + "testing" +) + +func TestWriterEmitsCompleteLinesAndFlushesTail(t *testing.T) { + var got []string + w := New(func(line string) { got = append(got, line) }) + + for _, chunk := range []string{"first\r", "\nsec", "ond\nthird"} { + if n, err := w.Write([]byte(chunk)); err != nil || n != len(chunk) { + t.Fatalf("Write(%q) = (%d, %v), want (%d, nil)", chunk, n, err, len(chunk)) + } + } + if want := []string{"first", "second"}; !reflect.DeepEqual(got, want) { + t.Fatalf("before Flush lines = %#v, want %#v", got, want) + } + + w.Flush() + w.Flush() + if want := []string{"first", "second", "third"}; !reflect.DeepEqual(got, want) { + t.Fatalf("after Flush lines = %#v, want %#v", got, want) + } +} + +func TestWriterBoundsAPathologicalLine(t *testing.T) { + var lengths []int + w := New(func(line string) { lengths = append(lengths, len(line)) }) + payload := append(bytes.Repeat([]byte{'x'}, maxPendingBytes+7), '\n') + + if _, err := w.Write(payload); err != nil { + t.Fatalf("Write: %v", err) + } + if want := []int{maxPendingBytes, 7}; !reflect.DeepEqual(lengths, want) { + t.Fatalf("emitted lengths = %v, want bounded chunks %v", lengths, want) + } +} + +func TestWriterDoesNotInventEmptyLineAfterExactBoundary(t *testing.T) { + var lengths []int + w := New(func(line string) { lengths = append(lengths, len(line)) }) + + if _, err := w.Write(bytes.Repeat([]byte{'x'}, maxPendingBytes)); err != nil { + t.Fatalf("write boundary-sized line: %v", err) + } + if _, err := w.Write([]byte{'\n'}); err != nil { + t.Fatalf("write terminating newline: %v", err) + } + + if want := []int{maxPendingBytes}; !reflect.DeepEqual(lengths, want) { + t.Fatalf("emitted lengths = %v, want %v", lengths, want) + } +} diff --git a/adapters/linux/runner.go b/adapters/linux/runner.go index 4c73f7ca..05441fce 100644 --- a/adapters/linux/runner.go +++ b/adapters/linux/runner.go @@ -27,7 +27,6 @@ package linux import ( - "bufio" "bytes" "context" "encoding/json" @@ -42,6 +41,8 @@ import ( "strings" "sync" "time" + + "github.com/Divaaaan/tenebra/adapters/internal/processlog" ) // defaultClashPort matches singbox.TunOptions' default external controller port, @@ -59,6 +60,10 @@ const logRingSize = 200 // is slow or not yet listening. const statsTimeout = 2 * time.Second +// processOutputWaitDelay bounds Cmd.Wait's drain of stdout/stderr after the +// supervised process exits but a descendant keeps inherited handles open. +const processOutputWaitDelay = 500 * time.Millisecond + // maxConnectionsBody bounds the read of the /connections document. It has to be // generous because the body is parsed as one JSON value: the totals live at the // head of the object, but json.Unmarshal still has to walk the whole connection @@ -184,19 +189,11 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) error { runCtx, cancel := context.WithCancel(ctx) cmd := exec.CommandContext(runCtx, bin, "run", "-c", cfgPath) - - stdout, err := cmd.StdoutPipe() - if err != nil { - cancel() - os.Remove(cfgPath) - return fmt.Errorf("linux: stdout pipe: %w", err) - } - stderr, err := cmd.StderrPipe() - if err != nil { - cancel() - os.Remove(cfgPath) - return fmt.Errorf("linux: stderr pipe: %w", err) - } + stdoutLog := processlog.New(r.ring.add) + stderrLog := processlog.New(r.ring.add) + cmd.Stdout = stdoutLog + cmd.Stderr = stderrLog + cmd.WaitDelay = processOutputWaitDelay if err := cmd.Start(); err != nil { cancel() @@ -211,15 +208,13 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) error { r.cfgPath = cfgPath r.clashSecret = secret - // Drain both streams into the ring buffer; the goroutines end when the pipes - // close on process exit. - go r.scan(stdout) - go r.scan(stderr) - // One watcher owns Wait. It publishes the exit on done, closes it, and clears - // the running state so the Runner can be started again. + // the running state so the Runner can be started again. Cmd.Wait owns and + // bounds stdout/stderr copying through WaitDelay. go func() { werr := cmd.Wait() + stdoutLog.Flush() + stderrLog.Flush() cancel() os.Remove(cfgPath) @@ -493,16 +488,6 @@ func (r *Runner) Logs() []string { return ring.snapshot() } -// scan copies a process stream line by line into the ring buffer. -func (r *Runner) scan(rc io.ReadCloser) { - defer rc.Close() - sc := bufio.NewScanner(rc) - sc.Buffer(make([]byte, 0, 64*1024), 1<<20) - for sc.Scan() { - r.ring.add(sc.Text()) - } -} - // singboxVersionTimeout bounds the `sing-box version` call. The binary answers // instantly or not at all; a longer budget would only lengthen the wait when the // file on disk is not actually an executable. diff --git a/adapters/linux/runner_completion_test.go b/adapters/linux/runner_completion_test.go new file mode 100644 index 00000000..38733931 --- /dev/null +++ b/adapters/linux/runner_completion_test.go @@ -0,0 +1,131 @@ +//go:build linux + +package linux + +import ( + "context" + "fmt" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +const runnerHelperModeEnv = "TENEBRA_LINUX_RUNNER_HELPER_MODE" + +func TestMain(m *testing.M) { + switch os.Getenv(runnerHelperModeEnv) { + case "normal": + fmt.Fprintln(os.Stdout, "stdout-final") + fmt.Fprint(os.Stderr, "stderr-tail") + os.Exit(0) + case "orphan-parent": + exe, err := os.Executable() + if err != nil { + fmt.Fprintln(os.Stderr, "locate helper:", err) + os.Exit(2) + } + cmd := exec.Command(exe) + cmd.Env = withRunnerHelperMode(os.Environ(), "orphan-child") + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + fmt.Fprintln(os.Stderr, "start orphan:", err) + os.Exit(2) + } + fmt.Fprintln(os.Stdout, "orphan-ready") + time.Sleep(10 * time.Second) + os.Exit(0) + case "orphan-child": + time.Sleep(2500 * time.Millisecond) + os.Exit(0) + default: + os.Exit(m.Run()) + } +} + +func TestRunnerDoneIncludesFinalStdoutAndStderr(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "normal") + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + r := New() + r.binOverride = exe + + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + select { + case err := <-r.Done(): + if err != nil { + t.Fatalf("Done: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Done did not report helper exit") + } + + logs := strings.Join(r.Logs(), "\n") + for _, want := range []string{"stdout-final", "stderr-tail"} { + if !strings.Contains(logs, want) { + t.Fatalf("Logs() = %q, missing %q", logs, want) + } + } +} + +func TestRunnerStopBoundsOrphanedOutputPipeDrain(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "orphan-parent") + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + r := New() + r.binOverride = exe + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + waitForRunnerLog(t, r, "orphan-ready") + + started := time.Now() + stopped := make(chan error, 1) + go func() { stopped <- r.Stop() }() + select { + case err := <-stopped: + if err != nil { + t.Fatalf("Stop: %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("Stop took %v; orphaned output pipe drain was not bounded", elapsed) + } + case <-time.After(time.Second): + select { + case <-stopped: + case <-time.After(4 * time.Second): + } + t.Fatalf("Stop still blocked after %v on an orphaned output pipe", time.Since(started)) + } +} + +func waitForRunnerLog(t *testing.T, r *Runner, want string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(strings.Join(r.Logs(), "\n"), want) { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("runner log never contained %q; logs=%q", want, r.Logs()) +} + +func withRunnerHelperMode(env []string, mode string) []string { + prefix := runnerHelperModeEnv + "=" + out := make([]string, 0, len(env)+1) + for _, item := range env { + if !strings.HasPrefix(item, prefix) { + out = append(out, item) + } + } + return append(out, prefix+mode) +} diff --git a/adapters/macos/runner.go b/adapters/macos/runner.go index 2479ca66..a8a83d21 100644 --- a/adapters/macos/runner.go +++ b/adapters/macos/runner.go @@ -22,7 +22,6 @@ package macos import ( - "bufio" "bytes" "context" "encoding/json" @@ -37,6 +36,8 @@ import ( "strings" "sync" "time" + + "github.com/Divaaaan/tenebra/adapters/internal/processlog" ) // defaultClashPort matches singbox.TunOptions' default external controller port, @@ -54,6 +55,10 @@ const logRingSize = 200 // is slow or not yet listening. const statsTimeout = 2 * time.Second +// processOutputWaitDelay bounds Cmd.Wait's drain of stdout/stderr after the +// supervised process exits but a descendant keeps inherited handles open. +const processOutputWaitDelay = 500 * time.Millisecond + // maxConnectionsBody bounds the read of the /connections document. It has to be // generous because the body is parsed as one JSON value: the totals live at the // head of the object, but json.Unmarshal still has to walk the whole connection @@ -178,19 +183,11 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) error { runCtx, cancel := context.WithCancel(ctx) cmd := exec.CommandContext(runCtx, bin, "run", "-c", cfgPath) - - stdout, err := cmd.StdoutPipe() - if err != nil { - cancel() - os.Remove(cfgPath) - return fmt.Errorf("macos: stdout pipe: %w", err) - } - stderr, err := cmd.StderrPipe() - if err != nil { - cancel() - os.Remove(cfgPath) - return fmt.Errorf("macos: stderr pipe: %w", err) - } + stdoutLog := processlog.New(r.ring.add) + stderrLog := processlog.New(r.ring.add) + cmd.Stdout = stdoutLog + cmd.Stderr = stderrLog + cmd.WaitDelay = processOutputWaitDelay if err := cmd.Start(); err != nil { cancel() @@ -205,15 +202,13 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) error { r.cfgPath = cfgPath r.clashSecret = secret - // Drain both streams into the ring buffer; the goroutines end when the pipes - // close on process exit. - go r.scan(stdout) - go r.scan(stderr) - // One watcher owns Wait. It publishes the exit on done, closes it, and clears - // the running state so the Runner can be started again. + // the running state so the Runner can be started again. Cmd.Wait owns and + // bounds stdout/stderr copying through WaitDelay. go func() { werr := cmd.Wait() + stdoutLog.Flush() + stderrLog.Flush() cancel() os.Remove(cfgPath) @@ -487,16 +482,6 @@ func (r *Runner) Logs() []string { return ring.snapshot() } -// scan copies a process stream line by line into the ring buffer. -func (r *Runner) scan(rc io.ReadCloser) { - defer rc.Close() - sc := bufio.NewScanner(rc) - sc.Buffer(make([]byte, 0, 64*1024), 1<<20) - for sc.Scan() { - r.ring.add(sc.Text()) - } -} - // singboxVersionTimeout bounds the `sing-box version` call. The binary answers // instantly or not at all; a longer budget would only lengthen the wait when the // file on disk is not actually an executable. diff --git a/adapters/macos/runner_completion_test.go b/adapters/macos/runner_completion_test.go new file mode 100644 index 00000000..72a34df8 --- /dev/null +++ b/adapters/macos/runner_completion_test.go @@ -0,0 +1,131 @@ +//go:build darwin + +package macos + +import ( + "context" + "fmt" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +const runnerHelperModeEnv = "TENEBRA_MACOS_RUNNER_HELPER_MODE" + +func TestMain(m *testing.M) { + switch os.Getenv(runnerHelperModeEnv) { + case "normal": + fmt.Fprintln(os.Stdout, "stdout-final") + fmt.Fprint(os.Stderr, "stderr-tail") + os.Exit(0) + case "orphan-parent": + exe, err := os.Executable() + if err != nil { + fmt.Fprintln(os.Stderr, "locate helper:", err) + os.Exit(2) + } + cmd := exec.Command(exe) + cmd.Env = withRunnerHelperMode(os.Environ(), "orphan-child") + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + fmt.Fprintln(os.Stderr, "start orphan:", err) + os.Exit(2) + } + fmt.Fprintln(os.Stdout, "orphan-ready") + time.Sleep(10 * time.Second) + os.Exit(0) + case "orphan-child": + time.Sleep(2500 * time.Millisecond) + os.Exit(0) + default: + os.Exit(m.Run()) + } +} + +func TestRunnerDoneIncludesFinalStdoutAndStderr(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "normal") + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + r := New() + r.binOverride = exe + + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + select { + case err := <-r.Done(): + if err != nil { + t.Fatalf("Done: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Done did not report helper exit") + } + + logs := strings.Join(r.Logs(), "\n") + for _, want := range []string{"stdout-final", "stderr-tail"} { + if !strings.Contains(logs, want) { + t.Fatalf("Logs() = %q, missing %q", logs, want) + } + } +} + +func TestRunnerStopBoundsOrphanedOutputPipeDrain(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "orphan-parent") + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + r := New() + r.binOverride = exe + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + waitForRunnerLog(t, r, "orphan-ready") + + started := time.Now() + stopped := make(chan error, 1) + go func() { stopped <- r.Stop() }() + select { + case err := <-stopped: + if err != nil { + t.Fatalf("Stop: %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("Stop took %v; orphaned output pipe drain was not bounded", elapsed) + } + case <-time.After(time.Second): + select { + case <-stopped: + case <-time.After(4 * time.Second): + } + t.Fatalf("Stop still blocked after %v on an orphaned output pipe", time.Since(started)) + } +} + +func waitForRunnerLog(t *testing.T, r *Runner, want string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(strings.Join(r.Logs(), "\n"), want) { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("runner log never contained %q; logs=%q", want, r.Logs()) +} + +func withRunnerHelperMode(env []string, mode string) []string { + prefix := runnerHelperModeEnv + "=" + out := make([]string, 0, len(env)+1) + for _, item := range env { + if !strings.HasPrefix(item, prefix) { + out = append(out, item) + } + } + return append(out, prefix+mode) +} diff --git a/adapters/windows/runner.go b/adapters/windows/runner.go index 6b58c516..67f62828 100644 --- a/adapters/windows/runner.go +++ b/adapters/windows/runner.go @@ -10,7 +10,6 @@ package windows import ( - "bufio" "bytes" "context" "encoding/json" @@ -26,6 +25,8 @@ import ( "strings" "sync" "time" + + "github.com/Divaaaan/tenebra/adapters/internal/processlog" ) // defaultClashPort matches singbox.TunOptions' default external controller port, @@ -43,6 +44,11 @@ const logRingSize = 200 // is slow or not yet listening. const statsTimeout = 2 * time.Second +// processOutputWaitDelay bounds Cmd.Wait's drain of stdout/stderr after the +// process exits. A descendant can inherit those handles and keep them open even +// though the supervised process is gone; completion must not hang on it. +const processOutputWaitDelay = 500 * time.Millisecond + // maxConnectionsBody bounds the read of the /connections document. It has to be // generous because the body is parsed as one JSON value: the totals live at the // head of the object, but json.Unmarshal still has to walk the whole connection @@ -154,26 +160,15 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) (startErr error) runCtx, cancel := context.WithCancel(ctx) cmd := exec.CommandContext(runCtx, bin, "run", "-c", cfgPath) - - stdout, err := cmd.StdoutPipe() - if err != nil { - cancel() - os.Remove(cfgPath) - return fmt.Errorf("windows: stdout pipe: %w", err) - } - stderr, err := cmd.StderrPipe() - if err != nil { - _ = stdout.Close() - cancel() - os.Remove(cfgPath) - return fmt.Errorf("windows: stderr pipe: %w", err) - } + stdoutLog := processlog.New(r.ring.add) + stderrLog := processlog.New(r.ring.add) + cmd.Stdout = stdoutLog + cmd.Stderr = stderrLog + cmd.WaitDelay = processOutputWaitDelay releaseProcess, err := startOwnedCommand(cmd) if err != nil { cancel() - _ = stdout.Close() - _ = stderr.Close() os.Remove(cfgPath) return fmt.Errorf("windows: start sing-box: %w", err) } @@ -185,16 +180,15 @@ func (r *Runner) Start(ctx context.Context, configJSON []byte) (startErr error) r.cfgPath = cfgPath r.clashSecret = secret - // Drain both streams into the ring buffer; the goroutines end when the pipes - // close on process exit. - go r.scan(stdout) - go r.scan(stderr) - // One watcher owns Wait. It publishes the exit on done, closes it, and clears - // the running state so the Runner can be started again. + // the running state so the Runner can be started again. Cmd.Wait owns its + // stdout/stderr copy goroutines; WaitDelay prevents inherited pipe handles in + // an orphaned descendant from hanging this watcher forever. go func() { werr := cmd.Wait() werr = errors.Join(werr, releaseProcess()) + stdoutLog.Flush() + stderrLog.Flush() cancel() os.Remove(cfgPath) @@ -468,16 +462,6 @@ func (r *Runner) Logs() []string { return ring.snapshot() } -// scan copies a process stream line by line into the ring buffer. -func (r *Runner) scan(rc io.ReadCloser) { - defer rc.Close() - sc := bufio.NewScanner(rc) - sc.Buffer(make([]byte, 0, 64*1024), 1<<20) - for sc.Scan() { - r.ring.add(sc.Text()) - } -} - // singboxVersionTimeout bounds the `sing-box version` call. The binary answers // instantly or not at all; a longer budget would only lengthen the wait when the // file on disk is not actually an executable. diff --git a/adapters/windows/runner_completion_test.go b/adapters/windows/runner_completion_test.go new file mode 100644 index 00000000..64211ad7 --- /dev/null +++ b/adapters/windows/runner_completion_test.go @@ -0,0 +1,138 @@ +package windows + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +const runnerHelperModeEnv = "TENEBRA_WINDOWS_RUNNER_HELPER_MODE" + +func TestMain(m *testing.M) { + switch os.Getenv(runnerHelperModeEnv) { + case "normal": + fmt.Fprintln(os.Stdout, "stdout-final") + fmt.Fprint(os.Stderr, "stderr-tail") + os.Exit(0) + case "orphan-parent": + cmd := exec.Command(os.Args[0]) + cmd.Env = withRunnerHelperMode(os.Environ(), "orphan-child") + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + fmt.Fprintln(os.Stderr, "start orphan:", err) + os.Exit(2) + } + fmt.Fprintln(os.Stdout, "orphan-ready") + time.Sleep(10 * time.Second) + os.Exit(0) + case "orphan-child": + time.Sleep(2500 * time.Millisecond) + os.Exit(0) + default: + os.Exit(m.Run()) + } +} + +func TestRunnerDoneIncludesFinalStdoutAndStderr(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "normal") + r := New() + r.binOverride = runnerHelperBinary(t) + + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + select { + case err := <-r.Done(): + if err != nil { + t.Fatalf("Done: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Done did not report helper exit") + } + + logs := strings.Join(r.Logs(), "\n") + for _, want := range []string{"stdout-final", "stderr-tail"} { + if !strings.Contains(logs, want) { + t.Fatalf("Logs() = %q, missing %q", logs, want) + } + } +} + +func TestRunnerStopBoundsOrphanedOutputPipeDrain(t *testing.T) { + t.Setenv(runnerHelperModeEnv, "orphan-parent") + r := New() + r.binOverride = runnerHelperBinary(t) + if err := r.Start(context.Background(), []byte(`{}`)); err != nil { + t.Fatalf("Start: %v", err) + } + waitForRunnerLog(t, r, "orphan-ready") + + started := time.Now() + stopped := make(chan error, 1) + go func() { stopped <- r.Stop() }() + select { + case err := <-stopped: + if err != nil { + t.Fatalf("Stop: %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("Stop took %v; orphaned output pipe drain was not bounded", elapsed) + } + case <-time.After(time.Second): + select { + case <-stopped: + case <-time.After(4 * time.Second): + } + t.Fatalf("Stop still blocked after %v on an orphaned output pipe", time.Since(started)) + } +} + +func runnerHelperBinary(t *testing.T) string { + t.Helper() + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + if runtime.GOOS != "windows" { + return exe + } + dir := t.TempDir() + dst := filepath.Join(dir, "runner-helper.exe") + if err := copyFile(dst, exe); err != nil { + t.Fatalf("copy helper binary: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "wintun.dll"), nil, 0o600); err != nil { + t.Fatalf("write test wintun.dll: %v", err) + } + return dst +} + +func waitForRunnerLog(t *testing.T, r *Runner, want string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(strings.Join(r.Logs(), "\n"), want) { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("runner log never contained %q; logs=%q", want, r.Logs()) +} + +func withRunnerHelperMode(env []string, mode string) []string { + prefix := runnerHelperModeEnv + "=" + out := make([]string, 0, len(env)+1) + for _, item := range env { + if !strings.HasPrefix(item, prefix) { + out = append(out, item) + } + } + return append(out, prefix+mode) +} diff --git a/core/control/daemon.go b/core/control/daemon.go index efa2cc5a..eb38bd11 100644 --- a/core/control/daemon.go +++ b/core/control/daemon.go @@ -64,8 +64,10 @@ type Runner interface { // did not take and the caller must fall back to a reconnect rather than // report a move that did not happen. Select(ctx context.Context, group, tag string) error - // Done delivers the process's exit: it sends the exit error (nil on a clean - // exit) once and is closed afterwards. Before any Start it must block. + // Done delivers the process's exit after stdout/stderr have been drained: it + // sends the exit error (nil on a clean exit) once and is closed afterwards. + // Thus Logs already contains the final diagnostic tail when Done is received. + // Before any Start it must block. Done() <-chan error // Logs returns a copy of the most recent sing-box output lines (newest last) — // the in-memory tail each runner keeps for diagnostics. It is safe to call at @@ -438,21 +440,22 @@ type Daemon struct { // pretending; tests inject a fake. probeRunner func() Runner // checkTargets are the destinations a node verdict is measured against, and - // checkBasePort where the per-node probe listeners start. Both are fields so a - // test can shrink the target list and move off the production ports. + // checkBasePort is the preferred start of the dynamically reserved listener + // block. Both are fields so a test can shrink the target list and choose a + // deterministic first candidate. checkTargets []string checkBasePort int // checkProbe runs one control request through a probe listener and reports how // far it got. Injectable so the ranking and reporting can be tested without a // network or a sing-box. - checkProbe func(ctx context.Context, port int, target string) (nodecheck.Stage, int64) + checkProbe func(ctx context.Context, binding singbox.ProbeBinding, target string) (nodecheck.Stage, int64) // checkBudget bounds a whole check run (see defaultCheckBudget). A field so a // test can shrink it to milliseconds instead of waiting one out. checkBudget time.Duration - // checkRunning is the single-flight guard for that run. The probe process - // binds a fixed range of loopback ports, so two overlapping runs would fight - // over them; it is an atomic rather than a mu-guarded flag because the check - // is served off the request loop and must not queue behind whatever holds mu. + // checkRunning is the single-flight guard for that run. Two overlapping runs + // would distort one another even when they receive separate free port blocks; + // it is an atomic rather than a mu-guarded flag because the check is served off + // the request loop and must not queue behind whatever holds mu. checkRunning atomic.Bool // localAddrs reports the machine's interface addresses, so a connect can pick diff --git a/core/control/nodecheck.go b/core/control/nodecheck.go index 3704820a..6b53895d 100644 --- a/core/control/nodecheck.go +++ b/core/control/nodecheck.go @@ -4,12 +4,16 @@ import ( "bufio" "context" "crypto/tls" + "encoding/base64" "encoding/json" + "errors" "fmt" + "io" "net" "net/http" "net/url" "strconv" + "strings" "sync" "time" @@ -33,11 +37,11 @@ var defaultCheckTargets = []string{ "https://api.anthropic.com/v1/messages", } -// defaultCheckBasePort is where the per-node probe listeners start. +// defaultCheckBasePort is the preferred start of the per-node probe block. // // High, unprivileged, and outside the ports this app already uses (the mixed -// inbound's 2081 and the clash API's 9090), so a check run cannot collide with -// the live tunnel it is running beside. +// inbound's 2081 and the clash API's 9090). The block is reserved before use +// and the allocator moves elsewhere when another local process already owns it. const defaultCheckBasePort = 24310 // checkFanout bounds how many nodes are probed at once. Each node costs a few @@ -54,6 +58,24 @@ const ( // checkListenerWait is how long to wait for the probe process to open its // loopback listeners before giving up on the run. checkListenerWait = 10 * time.Second + // probeListenerIOTimeout bounds one local SOCKS5 authentication exchange. + // Loopback either answers immediately or is not the listener we are waiting + // for; spending a node timeout here would only hide a local start failure. + probeListenerIOTimeout = 300 * time.Millisecond +) + +const ( + // Probe ports stay below the OS ephemeral range on the desktop platforms we + // ship. The preferred block is tried first; this range is the fallback when + // that block is occupied or lost in the narrow release-to-spawn race. + probeFallbackPortMin = 20000 + probeFallbackPortMax = 30000 + probePortSearchLimit = 2048 + probeStartAttempts = 3 + + probeFailureTailLines = 8 + probeFailureLineRunes = 320 + probeFailureMaxRunes = 3000 ) // defaultCheckBudget bounds a whole run — the wait for the probe's listeners and @@ -100,9 +122,9 @@ func (d *Daemon) handleCheckNodes(ctx context.Context, req Request) Response { return newError(req.ID, "check_nodes: probe runner not configured") } - // One run at a time for the whole daemon. The probe process binds a fixed - // range of loopback ports, so a second run started while the first still - // holds them would fail to bind and report every node dead — a measurement + // One run at a time for the whole daemon. Even with dynamically reserved + // loopback ports, a second probe process would compete for CPU and network and + // could distort both runs — a measurement // that lies is worse than one that is refused. The UI collapses its own // double-presses, but it is not the only caller: a session displaced // mid-check (the UI restarting) leaves its run unwinding while the new client @@ -112,8 +134,10 @@ func (d *Daemon) handleCheckNodes(ctx context.Context, req Request) Response { } defer d.checkRunning.Store(false) - // Everything below is bounded by one budget, and overrunning it truncates the - // run rather than failing it (see defaultCheckBudget). + // The request owns the process lifetime, while the shorter check budget owns + // only readiness and measurement. If the latter expires, exec.CommandContext + // must not kill the probe and race the intended successful partial result. + processParent := ctx ctx, cancel := context.WithTimeout(ctx, d.checkBudget) defer cancel() @@ -121,38 +145,91 @@ func (d *Daemon) handleCheckNodes(ctx context.Context, req Request) Response { for _, s := range p.Servers { nodes = append(nodes, s.Node) } - cfg, bindings, err := singbox.BuildProbe(nodes, d.checkBasePort) + // Render once at an arbitrary valid base to validate the nodes and learn how + // many listeners the usable subset needs. The actual config is rebuilt only + // after that many contiguous ports have been reserved successfully. + _, plannedBindings, err := singbox.BuildProbe(nodes, 1) if err != nil { return newError(req.ID, fmt.Sprintf("check_nodes: %v", err)) } - raw, err := json.Marshal(cfg) - if err != nil { - return newError(req.ID, fmt.Sprintf("check_nodes: encode probe config: %v", err)) - } d.emitLog(LogInfo, fmt.Sprintf("check_nodes: measuring %d node(s) of %q against %d target(s)", - len(bindings), p.Name, len(d.checkTargets))) + len(plannedBindings), p.Name, len(d.checkTargets))) + + var ( + results []nodecheck.NodeResult + tried []probePortSpan + ) + for attempt := 1; attempt <= probeStartAttempts; attempt++ { + reservation, reserveErr := reserveProbePortBlock(len(plannedBindings), d.checkBasePort, tried) + if reserveErr != nil { + msg := probeFailureMessage("port reservation", reserveErr, nil) + d.emitLog(LogWarn, msg) + return newError(req.ID, msg) + } + tried = append(tried, probePortSpan{first: reservation.base, last: reservation.base + len(plannedBindings) - 1}) - runner := d.probeRunner() - if err := runner.Start(ctx, raw); err != nil { - d.emitLog(LogWarn, fmt.Sprintf("check_nodes: the probe sing-box would not start: %v", err)) - return newError(req.ID, fmt.Sprintf("check_nodes: start probe: %v", err)) - } - // The probe process is ours alone and must not outlive the command, including - // when the caller cancels: a stranded sing-box holding loopback ports would - // make the next run fail to bind. - defer func() { _ = runner.Stop() }() + cfg, bindings, buildErr := singbox.BuildProbe(nodes, reservation.base) + if buildErr != nil { + reservation.release() + return newError(req.ID, fmt.Sprintf("check_nodes: %v", buildErr)) + } + raw, marshalErr := json.Marshal(cfg) + if marshalErr != nil { + reservation.release() + return newError(req.ID, fmt.Sprintf("check_nodes: encode probe config: %v", marshalErr)) + } - if !d.waitForProbeListeners(ctx, bindings) { - // This one is worth naming precisely: every node would otherwise score a - // failure it did not earn, and the report would blame the exits for a - // local process that never bound its ports. - d.emitLog(LogWarn, fmt.Sprintf("check_nodes: the probe's loopback listeners never came up within %s; no node was actually measured", - checkListenerWait)) - return newError(req.ID, "check_nodes: probe listeners never came up") - } + runner := d.probeRunner() + processCtx, cancelProcess := context.WithCancel(processParent) + stopProcess := func() { + cancelProcess() + _ = runner.Stop() + } + // Keep the whole contiguous block reserved while the config is rendered, + // then release it immediately before the process starts. There is no API + // for handing already-bound sockets to sing-box, so a tiny race remains; + // authenticated readiness plus the bounded retry below closes it honestly. + reservation.release() + if startErr := runner.Start(processCtx, raw); startErr != nil { + stopProcess() + msg := probeFailureMessage("startup", startErr, runner) + if isProbeBindCollision(startErr, runner.Logs()) && attempt < probeStartAttempts { + d.emitLog(LogWarn, fmt.Sprintf("%s; retrying on a new loopback block (%d/%d)", msg, attempt+1, probeStartAttempts)) + continue + } + d.emitLog(LogWarn, msg) + return newError(req.ID, msg) + } - results := d.probeBindings(ctx, p, bindings) + done := runner.Done() + retryable, readyErr := d.waitForProbeListeners(ctx, bindings, done) + if readyErr != nil { + stopProcess() + msg := probeFailureMessage("authenticated listener startup", readyErr, runner) + if (retryable || isProbeBindCollision(readyErr, runner.Logs())) && attempt < probeStartAttempts { + d.emitLog(LogWarn, fmt.Sprintf("%s; retrying on a new loopback block (%d/%d)", msg, attempt+1, probeStartAttempts)) + continue + } + d.emitLog(LogWarn, msg) + return newError(req.ID, msg) + } + + results, err = d.probeBindings(ctx, p, bindings, done) + if err != nil { + stopProcess() + msg := probeFailureMessage("measurement", err, runner) + d.emitLog(LogWarn, msg) + return newError(req.ID, msg) + } + stopProcess() + break + } + if results == nil { + msg := "check_nodes: local probe exhausted its startup attempts; no node was measured" + d.emitLog(LogWarn, msg) + return newError(req.ID, msg) + } d.logNodeCheck(results) lastGood := "" @@ -212,40 +289,256 @@ func (d *Daemon) logNodeCheck(results []nodecheck.NodeResult) { } } -// waitForProbeListeners blocks until every probe port accepts a connection, or -// the budget runs out. +// waitForProbeListeners blocks until every probe port completes this run's +// authenticated, no-egress SOCKS5 handshake, the process exits, or the budget +// runs out. // // Without it the first targets are measured against a process that has not // finished starting, and every node scores a failure it did not earn — the same // class of mistake that once labelled twelve working bypass strategies "did not // start". -func (d *Daemon) waitForProbeListeners(ctx context.Context, bindings []singbox.ProbeBinding) bool { +func (d *Daemon) waitForProbeListeners(ctx context.Context, bindings []singbox.ProbeBinding, done <-chan error) (bool, error) { deadline := time.Now().Add(checkListenerWait) + sawForeignListener := false for time.Now().Before(deadline) { - if ctx.Err() != nil { - return false + select { + case err, ok := <-done: + return isProbeBindCollision(err, nil), probeExitedError(err, ok) + case <-ctx.Done(): + return false, ctx.Err() + default: } all := true for _, b := range bindings { - c, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(b.Port)), 300*time.Millisecond) - if err != nil { + owned, listening := probeListenerOwnership(ctx, b) + if !owned { all = false + sawForeignListener = sawForeignListener || listening break } - _ = c.Close() } if all { - return true + // Do not let a process that died just after its final auth response be + // promoted to ready. A non-blocking exit read closes that last race. + select { + case err, ok := <-done: + return isProbeBindCollision(err, nil), probeExitedError(err, ok) + default: + return false, nil + } } select { + case err, ok := <-done: + return isProbeBindCollision(err, nil), probeExitedError(err, ok) case <-ctx.Done(): - return false + return false, ctx.Err() case <-time.After(200 * time.Millisecond): } } + return sawForeignListener, fmt.Errorf("probe listeners did not authenticate within %s", checkListenerWait) +} + +// probeListenerOwned authenticates to a mixed inbound over SOCKS5 and stops +// before issuing CONNECT. That proves the listener knows this run's random +// secret without sending a byte toward a node or any external destination. +func probeListenerOwned(ctx context.Context, binding singbox.ProbeBinding) bool { + owned, _ := probeListenerOwnership(ctx, binding) + return owned +} + +// probeListenerOwnership additionally reports whether something accepted TCP. +// A listener that answers but rejects the run's random auth is evidence that the +// release-to-spawn race was lost and a new port block should be tried. +func probeListenerOwnership(ctx context.Context, binding singbox.ProbeBinding) (owned, listening bool) { + if binding.Username == "" || binding.Password == "" || len(binding.Username) > 255 || len(binding.Password) > 255 { + return false, false + } + dialCtx, cancel := context.WithTimeout(ctx, probeListenerIOTimeout) + defer cancel() + conn, err := (&net.Dialer{}).DialContext(dialCtx, "tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(binding.Port))) + if err != nil { + return false, false + } + defer conn.Close() + listening = true + if deadline, ok := dialCtx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + + // Offer only username/password. A no-auth SOCKS server cannot select a + // different method from the offered set and therefore cannot look like ours. + if _, err := conn.Write([]byte{5, 1, 2}); err != nil { + return false, true + } + reply := make([]byte, 2) + if _, err := io.ReadFull(conn, reply); err != nil || reply[0] != 5 || reply[1] != 2 { + return false, true + } + auth := make([]byte, 0, 3+len(binding.Username)+len(binding.Password)) + auth = append(auth, 1, byte(len(binding.Username))) + auth = append(auth, binding.Username...) + auth = append(auth, byte(len(binding.Password))) + auth = append(auth, binding.Password...) + if _, err := conn.Write(auth); err != nil { + return false, true + } + if _, err := io.ReadFull(conn, reply); err != nil { + return false, true + } + return reply[0] == 1 && reply[1] == 0, true +} + +type probePortReservation struct { + base int + listeners []net.Listener +} + +func (r *probePortReservation) release() { + for _, l := range r.listeners { + _ = l.Close() + } + r.listeners = nil +} + +type probePortSpan struct { + first int + last int +} + +// reserveProbePortBlock proves that every port in one contiguous loopback block +// can be bound at the same time and holds the sockets until immediately before +// sing-box starts. A preferred block keeps normal runs stable; fallback search +// moves away from an occupied or previously raced block without trusting a +// connect-only availability check. +func reserveProbePortBlock(count, preferred int, tried []probePortSpan) (*probePortReservation, error) { + if count < 1 { + return nil, errors.New("probe needs at least one listener") + } + + candidates := make([]int, 0, probePortSearchLimit+1) + if preferred >= 1 && preferred+count-1 <= 65535 { + candidates = append(candidates, preferred) + } + for base := probeFallbackPortMin; base+count-1 <= probeFallbackPortMax && len(candidates) < probePortSearchLimit+1; base++ { + if base != preferred { + candidates = append(candidates, base) + } + } + + var lastErr error + for _, base := range candidates { + candidate := probePortSpan{first: base, last: base + count - 1} + if overlapsProbeSpan(candidate, tried) { + continue + } + listeners := make([]net.Listener, 0, count) + for port := candidate.first; port <= candidate.last; port++ { + l, err := net.Listen("tcp4", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + lastErr = err + for _, opened := range listeners { + _ = opened.Close() + } + listeners = nil + break + } + listeners = append(listeners, l) + } + if len(listeners) == count { + return &probePortReservation{base: base, listeners: listeners}, nil + } + } + if lastErr == nil { + lastErr = errors.New("no candidate block remained") + } + return nil, fmt.Errorf("reserve %d contiguous loopback port(s): %w", count, lastErr) +} + +func overlapsProbeSpan(candidate probePortSpan, prior []probePortSpan) bool { + for _, used := range prior { + if candidate.first <= used.last && used.first <= candidate.last { + return true + } + } return false } +func probeExitedError(err error, ok bool) error { + if !ok { + return errors.New("probe process exit channel closed without a status") + } + if err == nil { + return errors.New("probe process exited unexpectedly with a clean status") + } + return fmt.Errorf("probe process exited: %w", err) +} + +// isProbeBindCollision recognises the cross-platform diagnostics emitted when a +// port was claimed after reservation release. Only that local, transient start +// failure earns another process; invalid configs and missing binaries fail once +// with their real explanation. +func isProbeBindCollision(err error, logs []string) bool { + parts := make([]string, 0, len(logs)+1) + if err != nil { + parts = append(parts, err.Error()) + } + parts = append(parts, logs...) + text := strings.ToLower(strings.Join(parts, "\n")) + for _, marker := range []string{ + "eaddrinuse", + "wsaeaddrinuse", + "address already in use", + "address is already in use", + "only one usage of each socket address", + } { + if strings.Contains(text, marker) { + return true + } + } + return false +} + +// probeFailureMessage is the single wire/log representation of a local probe +// failure. It keeps only a bounded tail, flattens oversized/multiline entries, +// and then applies the daemon's established secret scrubber before anything can +// reach the UI or diagnostics ring. +func probeFailureMessage(phase string, cause error, runner Runner) string { + why := "unknown local failure" + if cause != nil { + why = cause.Error() + } + message := fmt.Sprintf("check_nodes: local probe failed during %s: %s", phase, why) + if runner != nil { + lines := runner.Logs() + if len(lines) > probeFailureTailLines { + lines = lines[len(lines)-probeFailureTailLines:] + } + clean := make([]string, 0, len(lines)) + for _, line := range lines { + line = strings.Join(strings.Fields(line), " ") + if line == "" { + continue + } + clean = append(clean, truncateRunes(line, probeFailureLineRunes)) + } + if len(clean) > 0 { + message += "; probe output: " + strings.Join(clean, " | ") + } + } + return truncateRunes(scrubSecrets(message), probeFailureMaxRunes) +} + +func truncateRunes(text string, max int) string { + runes := []rune(text) + if len(runes) <= max { + return text + } + if max < 2 { + return string(runes[:max]) + } + return string(runes[:max-1]) + "…" +} + // probeBindings measures every node, at most checkFanout at a time, and returns // one NodeResult per node in binding order. // @@ -253,9 +546,12 @@ func (d *Daemon) waitForProbeListeners(ctx context.Context, bindings []singbox.P // budget still reports the ones it never reached — with no targets, which both // Usable and Score already read as "not measured, not usable" — rather than // dropping them from the answer or naming them with an empty id. -func (d *Daemon) probeBindings(ctx context.Context, p profile.Profile, bindings []singbox.ProbeBinding) []nodecheck.NodeResult { +func (d *Daemon) probeBindings(ctx context.Context, p profile.Profile, bindings []singbox.ProbeBinding, done <-chan error) ([]nodecheck.NodeResult, error) { + measureCtx, cancel := context.WithCancel(ctx) + defer cancel() results := make([]nodecheck.NodeResult, len(bindings)) servers := make([]profile.Server, len(bindings)) + var exitErr error for i, b := range bindings { id := b.Tag if b.Index >= 0 && b.Index < len(p.Servers) { @@ -270,22 +566,52 @@ func (d *Daemon) probeBindings(ctx context.Context, p profile.Profile, bindings sem := make(chan struct{}, checkFanout) var wg sync.WaitGroup +launch: for i, b := range bindings { // Out of budget: the remaining nodes stay unmeasured rather than the run // carrying on past the deadline its caller was promised. - if ctx.Err() != nil { + if measureCtx.Err() != nil { break } - sem <- struct{}{} + select { + case err, ok := <-done: + exitErr = probeExitedError(err, ok) + cancel() + break launch + case <-measureCtx.Done(): + break launch + case sem <- struct{}{}: + } wg.Add(1) - go func(i, port int, srv profile.Server) { + go func(i int, binding singbox.ProbeBinding, srv profile.Server) { defer wg.Done() defer func() { <-sem }() - results[i].Targets = d.probeNode(ctx, port, srv) - }(i, b.Port, servers[i]) + results[i].Targets = d.probeNode(measureCtx, binding, srv) + }(i, b, servers[i]) + } + finished := make(chan struct{}) + go func() { + wg.Wait() + close(finished) + }() + if exitErr != nil { + <-finished + return nil, exitErr + } + select { + case err, ok := <-done: + cancel() + <-finished + return nil, probeExitedError(err, ok) + case <-finished: + // The process may have exited in the instant the last worker completed. + select { + case err, ok := <-done: + return nil, probeExitedError(err, ok) + default: + return results, nil + } } - wg.Wait() - return results } // probeNode measures one node: every target through its loopback proxy, and a @@ -297,7 +623,7 @@ func (d *Daemon) probeBindings(ctx context.Context, p profile.Profile, bindings // the state someone is in when they press this. Concurrency changes nothing // about the verdict — the ordering carried no information — while the load a // node sees, four cheap 204s at once, is less than opening one web page. -func (d *Daemon) probeNode(ctx context.Context, port int, srv profile.Server) []nodecheck.TargetResult { +func (d *Daemon) probeNode(ctx context.Context, binding singbox.ProbeBinding, srv profile.Server) []nodecheck.TargetResult { // Whether the node's own address answers at all decides which failure the // targets get reported as: unreachable address is a different problem for the // user (routing, firewall, dead host) than an address that answers and then @@ -315,7 +641,7 @@ func (d *Daemon) probeNode(ctx context.Context, port int, srv profile.Server) [] wg.Add(1) go func(i int, t string) { defer wg.Done() - stage, rtt := d.checkProbe(ctx, port, t) + stage, rtt := d.checkProbe(ctx, binding, t) probed[i] = nodecheck.TargetResult{Target: t, Stage: stage, RTTMs: rtt} measured[i] = true }(i, t) @@ -360,7 +686,7 @@ func (d *Daemon) probeNode(ctx context.Context, port int, srv profile.Server) [] // CONNECT means the tunnel came up and traffic did not survive it. Collapsed into // one "request failed" error, a black-hole node and an unreachable one look the // same, and the UI can only show a red dot instead of saying what broke. -func (d *Daemon) defaultCheckProbe(ctx context.Context, port int, target string) (nodecheck.Stage, int64) { +func (d *Daemon) defaultCheckProbe(ctx context.Context, binding singbox.ProbeBinding, target string) (nodecheck.Stage, int64) { u, err := url.Parse(target) if err != nil || u.Host == "" { return nodecheck.StageProbe, 0 @@ -372,7 +698,7 @@ func (d *Daemon) defaultCheckProbe(ctx context.Context, port int, target string) defer cancel() start := d.now() - conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(binding.Port))) if err != nil { // The listener is ours; failing to reach it is not the node's fault, but the // node cannot be credited either. @@ -383,7 +709,8 @@ func (d *Daemon) defaultCheckProbe(ctx context.Context, port int, target string) _ = conn.SetDeadline(dl) } - if _, err := fmt.Fprintf(conn, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", dest, dest); err != nil { + proxyAuth := base64.StdEncoding.EncodeToString([]byte(binding.Username + ":" + binding.Password)) + if _, err := fmt.Fprintf(conn, "CONNECT %s HTTP/1.1\r\nHost: %s\r\nProxy-Authorization: Basic %s\r\n\r\n", dest, dest, proxyAuth); err != nil { return nodecheck.StageHandshake, 0 } br := bufio.NewReader(conn) diff --git a/core/control/nodecheck_budget_test.go b/core/control/nodecheck_budget_test.go index e2c48a88..42792c24 100644 --- a/core/control/nodecheck_budget_test.go +++ b/core/control/nodecheck_budget_test.go @@ -9,6 +9,7 @@ import ( "github.com/Divaaaan/tenebra/core/model" "github.com/Divaaaan/tenebra/core/nodecheck" + "github.com/Divaaaan/tenebra/core/singbox" ) // manyNodes builds n distinguishable profile nodes. Their addresses are never @@ -48,7 +49,7 @@ func TestCheckNodesProbesOneNodesTargetsTogether(t *testing.T) { once sync.Once ) together := make(chan struct{}) - h.daemon.checkProbe = func(_ context.Context, _ int, _ string) (nodecheck.Stage, int64) { + h.daemon.checkProbe = func(_ context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { mu.Lock() inFlight++ if inFlight > peak { @@ -114,7 +115,7 @@ func TestCheckNodesAnswersWithinItsBudgetWithWhatItMeasured(t *testing.T) { } // A node that answers nothing: every request through it hangs until it is // abandoned. - h.daemon.checkProbe = func(ctx context.Context, _ int, _ string) (nodecheck.Stage, int64) { + h.daemon.checkProbe = func(ctx context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { select { case <-ctx.Done(): case <-time.After(perProbe): @@ -180,7 +181,7 @@ func TestCheckNodesDoesNotHoldTheRequestLoop(t *testing.T) { entered := make(chan struct{}) release := make(chan struct{}) var once sync.Once - h.daemon.checkProbe = func(ctx context.Context, _ int, _ string) (nodecheck.Stage, int64) { + h.daemon.checkProbe = func(ctx context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { once.Do(func() { close(entered) }) select { case <-release: @@ -207,11 +208,11 @@ func TestCheckNodesDoesNotHoldTheRequestLoop(t *testing.T) { } } -// TestCheckNodesRefusesASecondOverlappingRun: the probe process binds a fixed -// range of loopback ports, so a second run started while the first still holds -// them would fail to bind and report every node dead. Saying no is the honest -// answer — a measurement that lies is worse than one that is refused. It matters -// more now that a check no longer occupies the request loop it was serialised by. +// TestCheckNodesRefusesASecondOverlappingRun: even with separate reserved port +// blocks, two probe processes would compete for CPU and network and distort each +// other's measurements. Saying no is the honest answer — a measurement that +// lies is worse than one that is refused. It matters more now that a check no +// longer occupies the request loop it was serialised by. func TestCheckNodesRefusesASecondOverlappingRun(t *testing.T) { nodes := []model.Node{vlessNode("A", "a.example.11")} h, pid := newCheckHarness(t, nodes, 24730) @@ -219,7 +220,7 @@ func TestCheckNodesRefusesASecondOverlappingRun(t *testing.T) { entered := make(chan struct{}) release := make(chan struct{}) var once sync.Once - h.daemon.checkProbe = func(ctx context.Context, _ int, _ string) (nodecheck.Stage, int64) { + h.daemon.checkProbe = func(ctx context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { once.Do(func() { close(entered) }) select { case <-release: @@ -268,7 +269,7 @@ func TestCheckNodesKeepsTheStageAnAnsweringNodeEarned(t *testing.T) { // The node carries two destinations of three, while its own address answers no // dial (the harness fails every one) — the state a UDP-carried node is in // permanently. - h.daemon.checkProbe = func(_ context.Context, _ int, target string) (nodecheck.Stage, int64) { + h.daemon.checkProbe = func(_ context.Context, _ singbox.ProbeBinding, target string) (nodecheck.Stage, int64) { if target == blocked { return nodecheck.StageProbe, 0 } diff --git a/core/control/nodecheck_test.go b/core/control/nodecheck_test.go index f7a03a06..50f27648 100644 --- a/core/control/nodecheck_test.go +++ b/core/control/nodecheck_test.go @@ -1,37 +1,194 @@ package control import ( + "bufio" "context" + "encoding/base64" "encoding/json" "errors" + "fmt" + "io" "net" + "net/http" "strconv" + "strings" "sync" "testing" + "time" "github.com/Divaaaan/tenebra/core/model" "github.com/Divaaaan/tenebra/core/nodecheck" + "github.com/Divaaaan/tenebra/core/singbox" ) // checkHarness wires a daemon whose probe run needs neither a network nor a -// sing-box: the probe "process" is a fake runner, the listeners are real -// loopback sockets that accept and say nothing, and the per-target verdict comes -// from a table the test writes. +// sing-box: the probe "process" is a fake runner, the listeners perform only the +// authenticated SOCKS5 preflight (never CONNECT), and the per-target verdict +// comes from a table the test writes. type checkHarness struct { *harness - probe *fakeRunner - listener []net.Listener + probe *checkProbeRunner + preferredBase int } -// newCheckHarness prepares a daemon with profile nodes, a probe runner, and -// listeners opened on the ports BuildProbe will assign, so waitForProbeListeners -// is satisfied the way the real process would satisfy it. +// checkProbeRunner is a local, no-egress stand-in for sing-box. It binds the +// listeners from the rendered config and implements only the SOCKS5 +// authentication exchange: readiness can prove the process owns each port, but +// the fake never receives a CONNECT command and therefore never reaches a node. +type checkProbeRunner struct { + *fakeRunner + listenerMu sync.Mutex + listeners []net.Listener + bases []int + onStart func(context.Context, []byte, *checkProbeRunner) error + exitOnStartContext bool + startContextDone chan struct{} +} + +func newCheckProbeRunner() *checkProbeRunner { + return &checkProbeRunner{fakeRunner: newFakeRunner()} +} + +func (r *checkProbeRunner) Start(ctx context.Context, configJSON []byte) error { + if err := r.fakeRunner.Start(ctx, configJSON); err != nil { + return err + } + var err error + if r.onStart != nil { + err = r.onStart(ctx, configJSON, r) + } else { + err = r.serveConfig(configJSON) + } + if err != nil { + return err + } + if r.exitOnStartContext { + r.startContextDone = make(chan struct{}) + go func() { + <-ctx.Done() + r.exit(ctx.Err()) + close(r.startContextDone) + }() + } + return nil +} + +func (r *checkProbeRunner) serveConfig(configJSON []byte) error { + var cfg struct { + Inbounds []struct { + ListenPort int `json:"listen_port"` + Users []struct { + Username string `json:"username"` + Password string `json:"password"` + } `json:"users"` + } `json:"inbounds"` + } + if err := json.Unmarshal(configJSON, &cfg); err != nil { + return err + } + if len(cfg.Inbounds) == 0 { + return errors.New("fake probe: no inbounds") + } + + opened := make([]net.Listener, 0, len(cfg.Inbounds)) + for _, in := range cfg.Inbounds { + if len(in.Users) != 1 || in.Users[0].Username == "" || in.Users[0].Password == "" { + for _, prior := range opened { + _ = prior.Close() + } + return errors.New("fake probe: inbound has no authentication") + } + l, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(in.ListenPort))) + if err != nil { + for _, prior := range opened { + _ = prior.Close() + } + return err + } + opened = append(opened, l) + go serveProbeAuth(l, in.Users[0].Username, in.Users[0].Password) + } + + r.listenerMu.Lock() + r.listeners = append(r.listeners, opened...) + r.bases = append(r.bases, cfg.Inbounds[0].ListenPort) + r.listenerMu.Unlock() + return nil +} + +func (r *checkProbeRunner) Stop() error { + r.listenerMu.Lock() + listeners := r.listeners + r.listeners = nil + r.listenerMu.Unlock() + for _, l := range listeners { + _ = l.Close() + } + return r.fakeRunner.Stop() +} + +func (r *checkProbeRunner) lastBase() int { + r.listenerMu.Lock() + defer r.listenerMu.Unlock() + if len(r.bases) == 0 { + return 0 + } + return r.bases[len(r.bases)-1] +} + +func serveProbeAuth(l net.Listener, username, password string) { + for { + conn, err := l.Accept() + if err != nil { + return + } + go answerProbeAuth(conn, username, password) + } +} + +func answerProbeAuth(conn net.Conn, username, password string) { + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(time.Second)) + br := bufio.NewReader(conn) + hello := make([]byte, 3) + if _, err := io.ReadFull(br, hello); err != nil || hello[0] != 5 || hello[1] != 1 || hello[2] != 2 { + return + } + if _, err := conn.Write([]byte{5, 2}); err != nil { + return + } + head := make([]byte, 2) + if _, err := io.ReadFull(br, head); err != nil || head[0] != 1 { + return + } + ub := make([]byte, int(head[1])) + if _, err := io.ReadFull(br, ub); err != nil { + return + } + plen, err := br.ReadByte() + if err != nil { + return + } + pb := make([]byte, int(plen)) + if _, err := io.ReadFull(br, pb); err != nil { + return + } + status := byte(1) + if string(ub) == username && string(pb) == password { + status = 0 + } + _, _ = conn.Write([]byte{1, status}) +} + +// newCheckHarness prepares a daemon with profile nodes and a probe runner that +// opens the authenticated listeners from the config handed to Start, satisfying +// readiness the way the real process does without any external traffic. func newCheckHarness(t *testing.T, nodes []model.Node, basePort int) (*checkHarness, string) { t.Helper() h := newHarness(t) p := h.addProfile(nodes) - probe := newFakeRunner() + probe := newCheckProbeRunner() h.daemon.SetProbeRunner(func() Runner { return probe }) h.daemon.checkBasePort = basePort h.daemon.checkTargets = []string{"https://a.example/204", "https://b.example/204"} @@ -44,19 +201,8 @@ func newCheckHarness(t *testing.T, nodes []model.Node, basePort int) (*checkHarn return nil, errors.New("no such host") } - ch := &checkHarness{harness: h, probe: probe} - for i := range nodes { - l, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(basePort+i))) - if err != nil { - t.Fatalf("open fake probe listener: %v", err) - } - ch.listener = append(ch.listener, l) - } - t.Cleanup(func() { - for _, l := range ch.listener { - _ = l.Close() - } - }) + ch := &checkHarness{harness: h, probe: probe, preferredBase: basePort} + t.Cleanup(func() { _ = probe.Stop() }) return ch, p.ID } @@ -64,20 +210,368 @@ func newCheckHarness(t *testing.T, nodes []model.Node, basePort int) (*checkHarn // to the stage every one of its targets reports. func (c *checkHarness) verdicts(table map[int]nodecheck.Stage, rtt map[int]int64) { var mu sync.Mutex - c.daemon.checkProbe = func(_ context.Context, port int, _ string) (nodecheck.Stage, int64) { + c.daemon.checkProbe = func(_ context.Context, binding singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { mu.Lock() defer mu.Unlock() - st, ok := table[port] + logicalPort := c.preferredBase + binding.Port - c.probe.lastBase() + st, ok := table[logicalPort] if !ok { return nodecheck.StageProbe, 0 } if st == nodecheck.StageOK { - return st, rtt[port] + return st, rtt[logicalPort] } return st, 0 } } +// TestProbeListenerOwnershipRequiresMatchingAuthentication catches the exact +// false-ready path: accepting TCP is not ownership. An unrelated local SOCKS +// listener that selects no-auth must be rejected, while the per-run credentials +// rendered into our listener must complete only the auth exchange (no CONNECT). +func TestProbeListenerOwnershipRequiresMatchingAuthentication(t *testing.T) { + ours, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ours.Close() + go serveProbeAuth(ours, "probe-user", "probe-pass") + port := ours.Addr().(*net.TCPAddr).Port + if !probeListenerOwned(context.Background(), singbox.ProbeBinding{ + Port: port, Username: "probe-user", Password: "probe-pass", + }) { + t.Fatal("matching authenticated listener was not recognised") + } + if probeListenerOwned(context.Background(), singbox.ProbeBinding{ + Port: port, Username: "probe-user", Password: "wrong-pass", + }) { + t.Fatal("listener accepted with the wrong per-run password") + } + + foreign, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer foreign.Close() + go func() { + for { + conn, err := foreign.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + buf := make([]byte, 3) + _, _ = io.ReadFull(conn, buf) + _, _ = conn.Write([]byte{5, 0}) // unrelated no-auth SOCKS server + }() + } + }() + if probeListenerOwned(context.Background(), singbox.ProbeBinding{ + Port: foreign.Addr().(*net.TCPAddr).Port, Username: "probe-user", Password: "probe-pass", + }) { + t.Fatal("an unrelated listener was mistaken for the probe process") + } +} + +// TestDefaultCheckProbeAuthenticatesItsCONNECT protects the handoff from +// authenticated readiness to real measurement. Requiring credentials only in +// the sing-box config would make every subsequent CONNECT earn a 407 and label +// every healthy node dead. +func TestDefaultCheckProbeAuthenticatesItsCONNECT(t *testing.T) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + header := make(chan string, 1) + go func() { + conn, err := l.Accept() + if err != nil { + return + } + defer conn.Close() + req, err := http.ReadRequest(bufio.NewReader(conn)) + if err != nil { + header <- "read-error: " + err.Error() + return + } + header <- req.Header.Get("Proxy-Authorization") + _, _ = io.WriteString(conn, "HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n") + }() + + h := newHarness(t) + port := l.Addr().(*net.TCPAddr).Port + h.daemon.defaultCheckProbe(context.Background(), singbox.ProbeBinding{ + Port: port, Username: "probe-user", Password: "probe-pass", + }, "https://example.test/204") + want := "Basic " + base64.StdEncoding.EncodeToString([]byte("probe-user:probe-pass")) + if got := <-header; got != want { + t.Fatalf("Proxy-Authorization = %q, want %q", got, want) + } +} + +// TestCheckNodesMovesOffAnOccupiedPreferredPort catches the fixed-port incident: +// a local process already owning 24310 used to make sing-box fail after spawn, +// which was then reported as every remote node being unavailable. +func TestCheckNodesMovesOffAnOccupiedPreferredPort(t *testing.T) { + blocker, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer blocker.Close() + preferred := blocker.Addr().(*net.TCPAddr).Port - 1 + if preferred < 1 { + t.Skip("OS selected port 1; cannot place the blocker inside a two-port range") + } + + h, pid := newCheckHarness(t, []model.Node{ + vlessNode("A", "a.example.11"), + vlessNode("B", "b.example.22"), + }, preferred) + h.verdicts( + map[int]nodecheck.Stage{preferred: nodecheck.StageOK, preferred + 1: nodecheck.StageOK}, + map[int]int64{preferred: 20, preferred + 1: 25}, + ) + h.run(t, pid) + + if got := h.probe.lastBase(); got == preferred { + t.Fatalf("probe still used occupied preferred port %d", preferred) + } +} + +// TestCheckNodesRetriesAnAsyncBindCollision models the narrow race between +// releasing a verified-free block and sing-box binding it. The first process +// exits with the platform's bind diagnostic; the second attempt must use a new +// block and produce a measurement, not node-level failures. +func TestCheckNodesRetriesAnAsyncBindCollision(t *testing.T) { + const preferred = 24820 + h, pid := newCheckHarness(t, []model.Node{vlessNode("A", "a.example.11")}, preferred) + first := newCheckProbeRunner() + first.onStart = func(_ context.Context, _ []byte, r *checkProbeRunner) error { + r.setLogs("FATAL listen tcp 127.0.0.1: bind: address already in use") + go r.exit(errors.New("exit status 1")) + return nil + } + second := newCheckProbeRunner() + h.probe = second + h.verdicts(map[int]nodecheck.Stage{preferred: nodecheck.StageOK}, map[int]int64{preferred: 25}) + var factoryCalls int + h.daemon.SetProbeRunner(func() Runner { + factoryCalls++ + if factoryCalls == 1 { + return first + } + return second + }) + h.daemon.checkBudget = 2 * time.Second + + h.run(t, pid) + if factoryCalls != 2 { + t.Fatalf("probe runner factory called %d times, want 2", factoryCalls) + } + firstCfgs, secondCfgs := first.startCfgs(), second.startCfgs() + if len(firstCfgs) != 1 || len(secondCfgs) != 1 { + t.Fatalf("start configs: first=%d second=%d, want one each", len(firstCfgs), len(secondCfgs)) + } + if probeConfigBase(t, firstCfgs[0]) == probeConfigBase(t, secondCfgs[0]) { + t.Fatal("bind retry reused the collided port block") + } +} + +// TestCheckNodesReportsAndScrubsProbeExit ensures a local process death is never +// converted into per-node verdicts. The RPC error and daemon log retain the exit +// plus the useful tail, while UUID-shaped credentials are masked. +func TestCheckNodesReportsAndScrubsProbeExit(t *testing.T) { + const leaked = "11111111-2222-4333-8444-555555555555" + h, pid := newCheckHarness(t, []model.Node{vlessNode("A", "a.example.11")}, 24830) + var generated []string + h.probe.onStart = func(_ context.Context, raw []byte, r *checkProbeRunner) error { + var cfg struct { + Inbounds []struct { + Users []struct { + Username string `json:"username"` + Password string `json:"password"` + } `json:"users"` + } `json:"inbounds"` + } + if err := json.Unmarshal(raw, &cfg); err != nil { + return fmt.Errorf("decode generated auth: %w", err) + } + if len(cfg.Inbounds) == 0 || len(cfg.Inbounds[0].Users) != 1 { + return errors.New("generated config has no probe auth") + } + generated = []string{cfg.Inbounds[0].Users[0].Username, cfg.Inbounds[0].Users[0].Password} + r.setLogs( + "config loaded for user="+generated[0], + "FATAL password="+generated[1]+" credential="+leaked+" could not bind", + ) + go r.exit(errors.New("exit status 1")) + return nil + } + h.daemon.checkBudget = 2 * time.Second + probes := 0 + h.daemon.checkProbe = func(context.Context, singbox.ProbeBinding, string) (nodecheck.Stage, int64) { + probes++ + return nodecheck.StageOK, 1 + } + + start := time.Now() + resp := h.daemon.handleCheckNodes(context.Background(), Request{ID: 1, Cmd: CmdCheckNodes, Profile: pid}) + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("local probe exit took %v to report", elapsed) + } + if resp.Error == "" { + t.Fatal("probe exit returned a successful node report") + } + for _, want := range []string{"exit status 1", "could not bind", "***"} { + if !strings.Contains(resp.Error, want) { + t.Errorf("RPC error %q does not contain %q", resp.Error, want) + } + } + for _, secret := range append(generated, leaked) { + if secret != "" && strings.Contains(resp.Error, secret) { + t.Fatalf("RPC error leaked credential-like text: %s", resp.Error) + } + } + if probes != 0 { + t.Fatalf("measured %d node targets after local probe death", probes) + } + logs := h.daemon.logs.snapshot() + joined := "" + for _, entry := range logs { + joined += entry.Msg + "\n" + } + if !strings.Contains(joined, "exit status 1") || !strings.Contains(joined, "could not bind") { + t.Fatalf("daemon log lost the exit/tail: %s", joined) + } + for _, secret := range append(generated, leaked) { + if secret != "" && strings.Contains(joined, secret) { + t.Fatalf("daemon log leaked credential-like text: %s", joined) + } + } +} + +// TestCheckNodesAbortsWhenProbeDiesDuringMeasurement covers the second half of +// the Done contract: a process can die after readiness. In-flight target work +// must be cancelled and the whole command must fail locally rather than return a +// partially fabricated list of dead nodes. +func TestCheckNodesAbortsWhenProbeDiesDuringMeasurement(t *testing.T) { + h, pid := newCheckHarness(t, []model.Node{vlessNode("A", "a.example.11")}, 24840) + h.daemon.checkBudget = 2 * time.Second + entered := make(chan struct{}) + var once sync.Once + h.daemon.checkProbe = func(ctx context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { + once.Do(func() { close(entered) }) + <-ctx.Done() + return nodecheck.StageProbe, 0 + } + go func() { + <-entered + h.probe.setLogs("fatal: runtime crash") + h.probe.exit(errors.New("exit status 2")) + }() + + start := time.Now() + resp := h.daemon.handleCheckNodes(context.Background(), Request{ID: 1, Cmd: CmdCheckNodes, Profile: pid}) + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("mid-measurement probe exit took %v to report", elapsed) + } + if resp.Error == "" || !strings.Contains(resp.Error, "exit status 2") { + t.Fatalf("mid-measurement exit returned %q, want local process error", resp.Error) + } +} + +// TestCheckNodesBudgetDoesNotOwnProbeProcessLifetime catches a race between the +// intentional partial-result budget and exec.CommandContext. Expiring the +// measurement budget must stop scheduling/probing and return what completed; it +// must not kill sing-box underneath that collection and turn the same timeout +// into a local-process failure. The separately owned process context is still +// cancelled explicitly once the command has its result. +func TestCheckNodesBudgetDoesNotOwnProbeProcessLifetime(t *testing.T) { + const ( + preferred = 24850 + budget = 100 * time.Millisecond + ) + h, pid := newCheckHarness(t, manyNodes(12), preferred) + h.probe.exitOnStartContext = true + h.daemon.checkBudget = budget + h.daemon.checkTargets = []string{"https://a.example/204"} + h.daemon.checkProbe = func(ctx context.Context, _ singbox.ProbeBinding, _ string) (nodecheck.Stage, int64) { + <-ctx.Done() + // Make the process-context exit deterministic under the old wiring: its + // Done signal is queued before the measurement workers finish. + time.Sleep(50 * time.Millisecond) + return nodecheck.StageProbe, 0 + } + + resp := h.daemon.handleCheckNodes(context.Background(), Request{ID: 1, Cmd: CmdCheckNodes, Profile: pid}) + if resp.Error != "" { + t.Fatalf("ordinary measurement budget became a local probe failure: %s", resp.Error) + } + var out checkReply + if err := json.Unmarshal(resp.Data, &out); err != nil { + t.Fatalf("decode partial result: %v", err) + } + if len(out.Results) != 12 { + t.Fatalf("partial result contains %d nodes, want all 12 named", len(out.Results)) + } + unmeasured := 0 + for _, result := range out.Results { + if len(result.Targets) == 0 { + unmeasured++ + } + } + if unmeasured == 0 { + t.Fatal("budget expiry measured every node instead of returning a partial result") + } + select { + case <-h.probe.startContextDone: + case <-time.After(time.Second): + t.Fatal("probe process lifetime context was not cancelled after the partial result") + } +} + +func TestProbeBindCollisionClassifierIsSpecific(t *testing.T) { + tests := []struct { + name string + err error + logs []string + want bool + }{ + {name: "linux errno name", logs: []string{"listen tcp: EADDRINUSE"}, want: true}, + {name: "windows errno name", logs: []string{"listen tcp: WSAEADDRINUSE"}, want: true}, + {name: "unix message", err: errors.New("bind: address already in use"), want: true}, + {name: "windows message", logs: []string{"Only one usage of each socket address is normally permitted"}, want: true}, + {name: "generic bind failure", logs: []string{"failed to bind listener: permission denied"}}, + {name: "wrong local address", logs: []string{"listen tcp 127.0.0.1: cannot assign requested address"}}, + {name: "generic listen failure", logs: []string{"listen tcp 127.0.0.1: permission denied"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isProbeBindCollision(tt.err, tt.logs); got != tt.want { + t.Fatalf("isProbeBindCollision(%v, %q) = %t, want %t", tt.err, tt.logs, got, tt.want) + } + }) + } +} + +func probeConfigBase(t *testing.T, raw []byte) int { + t.Helper() + var cfg struct { + Inbounds []struct { + ListenPort int `json:"listen_port"` + } `json:"inbounds"` + } + if err := json.Unmarshal(raw, &cfg); err != nil { + t.Fatalf("decode probe config: %v", err) + } + if len(cfg.Inbounds) == 0 { + t.Fatal("probe config has no inbounds") + } + return cfg.Inbounds[0].ListenPort +} + type checkReply struct { Results []nodecheck.NodeResult `json:"results"` Best string `json:"best"` diff --git a/core/singbox/check_e2e_test.go b/core/singbox/check_e2e_test.go index 131a883b..37a1d5cc 100644 --- a/core/singbox/check_e2e_test.go +++ b/core/singbox/check_e2e_test.go @@ -42,6 +42,19 @@ func checkNodes() []model.Node { } } +func TestFindSingBoxHonorsOverride(t *testing.T) { + want := filepath.Join(t.TempDir(), "custom-sing-box") + if err := os.WriteFile(want, []byte("test binary placeholder"), 0o700); err != nil { + t.Fatalf("write override placeholder: %v", err) + } + t.Setenv("TENEBRA_SINGBOX", want) + + got, ruleSetDir, ok := findSingBox() + if !ok || got != want || ruleSetDir != filepath.Dir(want) { + t.Fatalf("findSingBox() = (%q, %q, %t), want (%q, %q, true)", got, ruleSetDir, ok, want, filepath.Dir(want)) + } +} + // findSingBox locates a real sing-box binary to validate generated configs // against. It walks up from the test's working directory to the bundled // resources (where fetch-resources drops sing-box next to the .srs files), then @@ -50,6 +63,11 @@ func checkNodes() []model.Node { // PATH), or ok=false when no binary is available — the caller then skips, so CI // without the (gitignored) binary is green rather than failing. func findSingBox() (bin, ruleSetDir string, ok bool) { + if override := os.Getenv("TENEBRA_SINGBOX"); override != "" { + if _, err := os.Stat(override); err == nil { + return override, filepath.Dir(override), true + } + } name := "sing-box" if runtime.GOOS == "windows" { name = "sing-box.exe" @@ -101,6 +119,22 @@ func singBoxCheck(t *testing.T, bin string, cfg map[string]any) { } } +// TestProbeConfigPassesSingBoxCheck keeps the authenticated mixed-inbound +// schema pinned to the bundled engine. The unit tests prove which credentials +// are rendered; the real checker proves sing-box accepts that shape before a +// release can turn every node check into a local config failure. +func TestProbeConfigPassesSingBoxCheck(t *testing.T) { + bin, _, ok := findSingBox() + if !ok { + t.Skip("sing-box binary not found (resources/ or bin/ or PATH); skipping real probe config check") + } + cfg, _, err := BuildProbe(checkNodes(), 24100) + if err != nil { + t.Fatalf("BuildProbe: %v", err) + } + singBoxCheck(t, bin, cfg) +} + // TestRulesConfigPassesSingBoxCheck feeds configs carrying the custom domain // rules and the RU presets through a real `sing-box check`. It runs the global // case (no rule-set file dependency) and, when the bundled .srs are present, the diff --git a/core/singbox/probe.go b/core/singbox/probe.go index b48db883..96030f35 100644 --- a/core/singbox/probe.go +++ b/core/singbox/probe.go @@ -1,6 +1,7 @@ package singbox import ( + "crypto/rand" "fmt" "github.com/Divaaaan/tenebra/core/model" @@ -16,6 +17,12 @@ type ProbeBinding struct { Name string // Port is the loopback port whose traffic is pinned to this node. Port int + // Username and Password are fresh for this probe process and are required by + // every mixed inbound it owns. The daemon authenticates without issuing a + // proxy request before trusting Port, so an unrelated listener cannot make a + // local start failure look like a dead remote node. + Username string `json:"-"` + Password string `json:"-"` // Index is the node's position in the slice handed to BuildProbe. It is the // only reliable way back to the caller's own identity for that node: the tag // is derived from the display name and de-duplicated, and names repeat across @@ -70,6 +77,14 @@ func BuildProbe(nodes []model.Node, basePort int) (map[string]any, []ProbeBindin if basePort+len(sel)-1 > 65535 { return nil, nil, fmt.Errorf("singbox: probe needs %d ports from %d, past 65535", len(sel), basePort) } + username, err := randomProbeCredential() + if err != nil { + return nil, nil, err + } + password, err := randomProbeCredential() + if err != nil { + return nil, nil, err + } bindings := make([]ProbeBinding, 0, len(sel)) inbounds := make([]map[string]any, 0, len(sel)) @@ -92,13 +107,20 @@ func BuildProbe(nodes []model.Node, basePort int) (map[string]any, []ProbeBindin // any host on the LAN relay through the user's nodes. "listen": mixedListen, "listen_port": port, + "users": []map[string]any{{ + "username": username, + "password": password, + }}, }) rules = append(rules, map[string]any{ "inbound": []string{inTag}, "action": "route", "outbound": nt.Tag, }) - bindings = append(bindings, ProbeBinding{Tag: nt.Tag, Name: name, Port: port, Index: nt.Index}) + bindings = append(bindings, ProbeBinding{ + Tag: nt.Tag, Name: name, Port: port, Index: nt.Index, + Username: username, Password: password, + }) } outbounds := make([]map[string]any, 0, len(outs)+1) @@ -125,3 +147,18 @@ func BuildProbe(nodes []model.Node, basePort int) (map[string]any, []ProbeBindin return cfg, bindings, nil } + +// randomProbeCredential returns a UUID-shaped 122-bit random value. UUID shape +// is deliberate as well as convenient: the daemon's established scrubSecrets +// policy already masks UUIDs in free-form process output, so a sing-box error +// that echoes the config cannot leak these short-lived credentials to the UI or +// support bundle. +func randomProbeCredential() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("singbox: generate probe credentials: %w", err) + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil +} diff --git a/core/singbox/probe_test.go b/core/singbox/probe_test.go index 6a3daaa4..324489bb 100644 --- a/core/singbox/probe_test.go +++ b/core/singbox/probe_test.go @@ -62,6 +62,53 @@ func TestBuildProbeGivesEachNodeItsOwnPort(t *testing.T) { } } +// TestBuildProbeRequiresFreshAuthentication catches the listener-ownership bug: +// a plain TCP connect could be accepted by an unrelated process on the expected +// port and the daemon would then attribute that process's behaviour to a VPN +// node. Every listener in one probe run must instead require the same fresh +// credentials, and the caller must receive those credentials so readiness can +// authenticate before any node traffic is measured. +func TestBuildProbeRequiresFreshAuthentication(t *testing.T) { + cfg1, bindings1, err := BuildProbe(probeNodes(), 24100) + if err != nil { + t.Fatalf("first BuildProbe: %v", err) + } + _, bindings2, err := BuildProbe(probeNodes(), 24200) + if err != nil { + t.Fatalf("second BuildProbe: %v", err) + } + if len(bindings1) == 0 || len(bindings2) == 0 { + t.Fatal("BuildProbe returned no bindings") + } + + user1, pass1 := bindings1[0].Username, bindings1[0].Password + if user1 == "" || pass1 == "" { + t.Fatal("first probe run has empty listener credentials") + } + for i, b := range bindings1 { + if b.Username != user1 || b.Password != pass1 { + t.Errorf("binding %d credentials differ within one run", i) + } + } + if bindings2[0].Username == user1 || bindings2[0].Password == pass1 { + t.Error("two probe runs reused listener credentials") + } + + ins, _ := cfg1["inbounds"].([]map[string]any) + if len(ins) != len(bindings1) { + t.Fatalf("got %d inbounds for %d bindings", len(ins), len(bindings1)) + } + for i, in := range ins { + users, _ := in["users"].([]map[string]any) + if len(users) != 1 { + t.Fatalf("inbound %d users = %#v, want one authenticated user", i, in["users"]) + } + if users[0]["username"] != user1 || users[0]["password"] != pass1 { + t.Errorf("inbound %d does not carry its binding credentials", i) + } + } +} + func TestBuildProbeRoutesEachListenerToItsOwnNode(t *testing.T) { cfg, bindings, err := BuildProbe(probeNodes(), 24100) if err != nil { From 17b77e3784569c03181ec508c06ef07297461d1b Mon Sep 17 00:00:00 2001 From: DivanMe <48186011+Divaaaan@users.noreply.github.com> Date: Wed, 16 Sep 2026 00:16:09 +0300 Subject: [PATCH 2/3] fix(ui): report node ping failures honestly --- ui-desktop/src/App.audit.test.tsx | 25 ++++- ui-desktop/src/App.protection.test.tsx | 9 +- ui-desktop/src/App.tsx | 29 +++--- ui-desktop/src/components/ServerList.test.tsx | 91 ++++++++++++++++++- ui-desktop/src/components/ServerList.tsx | 47 ++++++++-- ui-desktop/src/i18n/strings.ts | 16 ++++ ui-desktop/src/lib/useNodePings.test.ts | 72 +++++++++++++-- ui-desktop/src/lib/useNodePings.ts | 86 +++++++++++++----- ui-desktop/src/styles/servers.css | 8 +- 9 files changed, 327 insertions(+), 56 deletions(-) diff --git a/ui-desktop/src/App.audit.test.tsx b/ui-desktop/src/App.audit.test.tsx index 6480ed26..6f069585 100644 --- a/ui-desktop/src/App.audit.test.tsx +++ b/ui-desktop/src/App.audit.test.tsx @@ -1,4 +1,5 @@ import type { DeepLinkAction, PingResult, State } from "./api"; +import type { NodePingPhase } from "./lib/useNodePings"; import { createElement } from 'react'; import { act, cleanup, fireEvent, screen, waitFor, within } from '@testing-library/react'; import { afterEach, beforeEach, expect, it, vi } from 'vitest'; @@ -10,6 +11,7 @@ const m = vi.hoisted(() => ({ checkNodes: vi.fn(), importSubscription: vi.fn(), refreshProfiles: vi.fn(), updateAvailable: null as string | null, updateConfirm: false, confirmUpdate: vi.fn(), connect: vi.fn(), disconnect: vi.fn(), onDeepLink: vi.fn(), deep: null as ((e: DeepLinkAction) => void) | null, pings: new Map(), + pingPhase: "ready" as NodePingPhase, pingError: null as string | null, refreshPings: vi.fn(), profiles: [ { id: 'p1', name: 'Profile A', source: 'manual', nodes: [{id:'n1',name:'Node A',protocol:'vless',server:'198.51.100.10',port:443}], updatedAt:'2026-01-01T00:00:00Z' }, { id: 'p2', name: 'Profile B', source: 'manual', nodes: [{id:'n2',name:'Node B',protocol:'vless',server:'198.51.100.11',port:443}], updatedAt:'2026-01-01T00:00:00Z' }, @@ -27,7 +29,7 @@ vi.mock('./api/index.ts', () => ({ takeLaunchDeepLinks: vi.fn(async () => []), })); vi.mock('./lib/useNodePings.ts', () => ({ - useNodePings: () => ({results:m.pings,pinging:false,refresh:()=>{}}), + useNodePings: () => ({results:m.pings,phase:m.pingPhase,error:m.pingError,refresh:m.refreshPings}), })); vi.mock('./lib/useUpdateCheck.ts', () => ({ useUpdateCheck: () => ({available:m.updateAvailable,stalled:false,confirming:m.updateConfirm,installing:false,deferred:false,progress:null,install:vi.fn(),dismiss:vi.fn(),cancelInstall:vi.fn(),confirmInstall:m.confirmUpdate}), @@ -41,6 +43,7 @@ beforeEach(() => { m.importSubscription.mockResolvedValue({name:"Imported profile"}); m.refreshProfiles.mockResolvedValue(undefined); m.pings = new Map(); + m.pingPhase = "ready"; m.pingError = null; m.refreshPings.mockReset(); m.onDeepLink.mockImplementation(async (handler) => { m.deep = handler; return () => {}; }); m.connect.mockResolvedValue({state:'connecting'}); }); @@ -56,6 +59,26 @@ it('keeps failed ping unknown and permits a deliberate manual selection', async expect(document.querySelectorAll('.cur-meta .ping-scale-bar.on.good')).toHaveLength(0); }); +it('keeps a failed batch ping separate from the connect-time node check', async () => { + m.pings.set('n1',{node:'n1',ok:true,rttMs:17}); + m.pingPhase = 'failed'; + m.pingError = 'TCP prober unavailable'; + + renderWithProviders(createElement(App)); + + const status = await screen.findByText('Couldn’t check TCP'); + expect(status).toHaveAttribute('title','TCP prober unavailable'); + const row=screen.getByText('Node A',{selector:'.srv-node-code'}).closest('.srv-row')!; + expect(row).toHaveTextContent('stale'); + expect(row.querySelectorAll('.ping-scale-bar.on')).toHaveLength(0); + expect(screen.getByText('lowest ping')).toBeInTheDocument(); + expect(screen.queryByText(/lowest ping · now/)).not.toBeInTheDocument(); + expect(document.querySelector('.cur-rtt')).toBeNull(); + fireEvent.click(screen.getByRole('button',{name:'Retry TCP check'})); + expect(m.refreshPings).toHaveBeenCalledTimes(1); + expect(m.checkNodes).not.toHaveBeenCalled(); +}); + it.each([['0', false, null], ['0', true, 'service lost'], ['1', false, null], ['1', true, 'service lost']] as const)( 'blocks keyboard Connect as well as the button in mode %s with ready=%s error=%s', async (mode, ready, error) => { localStorage.setItem('tenebra.simpleMode', mode); diff --git a/ui-desktop/src/App.protection.test.tsx b/ui-desktop/src/App.protection.test.tsx index 317f7bf0..8027ba4e 100644 --- a/ui-desktop/src/App.protection.test.tsx +++ b/ui-desktop/src/App.protection.test.tsx @@ -17,7 +17,14 @@ vi.mock("./api", () => ({ onTrayConnect: vi.fn(async () => () => {}), onTrayShow: vi.fn(async () => () => {}), onDeepLink: vi.fn(async () => () => {}), takeLaunchDeepLinks: vi.fn(async () => []), })); -vi.mock("./lib/useNodePings", () => ({ useNodePings: () => ({ results: m.pings, pinging: false }) })); +vi.mock("./lib/useNodePings", () => ({ + useNodePings: () => ({ + results: m.pings, + phase: "ready", + error: null, + refresh: vi.fn(), + }), +})); vi.mock("./lib/useUpdateCheck", () => ({ useUpdateCheck: () => ({ available: null, stalled: false, confirming: false }) })); beforeEach(() => { localStorage.clear(); diff --git a/ui-desktop/src/App.tsx b/ui-desktop/src/App.tsx index 94aabe2f..2b9d4a09 100644 --- a/ui-desktop/src/App.tsx +++ b/ui-desktop/src/App.tsx @@ -291,6 +291,7 @@ export function App() { // Latency probes for the browsed profile, feeding the per-row ping + the // dead flag, and the live ping stat for the connected node. const pings = useNodePings(selectedProfileId); + const pingBatchReady = pings.phase === "ready"; // What actually survives each node, measured on demand — the connect button // runs it before choosing an exit (see handlePrimary). const nodeCheck = useNodeCheck(); @@ -317,28 +318,30 @@ export function App() { city: loc.label, region: loc.region, protocol: n.protocol, - rttMs: probe?.ok && !pings.stale ? probe.rttMs : null, - stale: !!probe && pings.stale, - dead: probe && !pings.stale ? !probe.ok : false, + rttMs: probe?.ok ? probe.rttMs : null, + stale: !!probe && !pingBatchReady, + dead: pingBatchReady && !!probe ? !probe.ok : false, insecure: n.insecure ?? false, }; }), - [nodes, pings.results, pings.stale], + [nodes, pings.results, pingBatchReady], ); // Lowest-ping live node, used as the auto target and the idle "current node". const bestNodeId = useMemo(() => { let best: string | null = null; let bestRtt = Infinity; - for (const n of nodes) { - const probe = pings.results.get(n.id); - if (probe?.ok && probe.rttMs < bestRtt) { - bestRtt = probe.rttMs; - best = n.id; + if (pingBatchReady) { + for (const n of nodes) { + const probe = pings.results.get(n.id); + if (probe?.ok && probe.rttMs < bestRtt) { + bestRtt = probe.rttMs; + best = n.id; + } } } return best ?? nodes[0]?.id ?? null; - }, [nodes, pings.results]); + }, [nodes, pings.results, pingBatchReady]); const targetNodeId = selectedNodeId || bestNodeId || ""; const displayedNode = connected @@ -347,7 +350,7 @@ export function App() { const liveNodeId = connected ? state.node : targetNodeId; const liveProbe = liveNodeId ? pings.results.get(liveNodeId) : undefined; - const livePing = liveProbe?.ok && !pings.stale ? liveProbe.rttMs : undefined; + const livePing = pingBatchReady && liveProbe?.ok ? liveProbe.rttMs : undefined; // Confirm the App-level actions the user takes (reaching connected, arming the // kill switch, changing routing) with a toast. The initial status load is @@ -878,7 +881,9 @@ export function App() { onQuery={setQuery} onSelectNode={handleSelectNode} onAddSubscription={() => setOverlay("profiles")} - pinging={pings.pinging} + pingPhase={pings.phase} + pingError={pings.error} + onRefreshPings={pings.refresh} disabled={selectionLocked} /> } diff --git a/ui-desktop/src/components/ServerList.test.tsx b/ui-desktop/src/components/ServerList.test.tsx index 0cf5c5df..d7ebfd3e 100644 --- a/ui-desktop/src/components/ServerList.test.tsx +++ b/ui-desktop/src/components/ServerList.test.tsx @@ -57,7 +57,9 @@ function baseProps(overrides: Partial[0]> = {}) { onQuery: vi.fn(), onSelectNode: vi.fn(), onAddSubscription: vi.fn(), - pinging: false, + pingPhase: "ready" as const, + pingError: null as string | null, + onRefreshPings: vi.fn(), ...overrides, }; } @@ -118,6 +120,80 @@ describe("ServerList", () => { expect(screen.getByRole("button", { name: /lowest ping · now fresh/i })).toBeInTheDocument(); }); + it("shows an in-progress TCP check without claiming zero reachable nodes", () => { + const rows = makeRows().map((row) => ({ + ...row, + rttMs: null, + dead: false, + })); + + renderWithProviders( + , + ); + + expect(screen.getByText("Checking TCP…")).toBeInTheDocument(); + expect(screen.queryByText(/0 TCP reachable/)).not.toBeInTheDocument(); + }); + + it("shows a failed TCP check and retries it from the header", async () => { + const onRefreshPings = vi.fn(); + const user = userEvent.setup(); + + const { rerender } = renderWithProviders( + , + ); + + const status = screen.getByText("Couldn’t check TCP"); + expect(status).toHaveAttribute( + "title", + "probe process failed on 127.0.0.1:24310", + ); + expect( + screen.queryByText("probe process failed on 127.0.0.1:24310"), + ).not.toBeInTheDocument(); + await user.click(screen.getByRole("button", { name: "Retry TCP check" })); + expect(onRefreshPings).toHaveBeenCalledTimes(1); + + rerender( + , + ); + expect(screen.getByRole("button", { name: "Retry TCP check" })).toBeDisabled(); + }); + + it("shows not checked while no profile batch has started", () => { + renderWithProviders(); + + expect(screen.getByText("Not checked")).toBeInTheDocument(); + expect(screen.queryByText(/TCP reachable/)).not.toBeInTheDocument(); + }); + + it("shows a truthful zero after a completed all-failed batch", () => { + const rows = makeRows().map((row) => ({ + ...row, + rttMs: null, + dead: true, + })); + + renderWithProviders( + , + ); + + expect(screen.getByText("0 TCP reachable")).toBeInTheDocument(); + }); + it("filters rows by region chip and back to all", async () => { const user = userEvent.setup(); // The region is controlled by the parent, so re-render with the value the @@ -368,6 +444,19 @@ describe("ServerList", () => { expect(deadRow.querySelectorAll(".ping-scale-bar")).toHaveLength(5); expect(deadRow.querySelectorAll(".ping-scale-bar.on")).toHaveLength(0); }); + + it("keeps a stale RTT visible as stale without lighting the meter", () => { + const rows = [{ ...makeRows()[0], rttMs: 27, stale: true }]; + + renderWithProviders( + , + ); + + const staleRow = screen.getByText("DE-FRA-01").closest('[role="button"]')!; + expect(staleRow).toHaveTextContent("stale"); + expect(staleRow.querySelectorAll(".ping-scale-bar")).toHaveLength(5); + expect(staleRow.querySelectorAll(".ping-scale-bar.on")).toHaveLength(0); + }); }); describe("focus-search event", () => { diff --git a/ui-desktop/src/components/ServerList.tsx b/ui-desktop/src/components/ServerList.tsx index ff631278..d76980bc 100644 --- a/ui-desktop/src/components/ServerList.tsx +++ b/ui-desktop/src/components/ServerList.tsx @@ -4,6 +4,7 @@ import type { NodeProtocol, Profile } from "../api"; import { useI18n } from "../i18n/I18nContext"; import type { Strings } from "../i18n/strings"; import { REGION_CHIPS, type Region } from "../lib/region"; +import type { NodePingPhase } from "../lib/useNodePings"; import { PingScale } from "./PingScale"; /** A node enriched with derived location and a latency probe, ready to render. */ @@ -38,7 +39,9 @@ interface ServerListProps { onQuery: (q: string) => void; onSelectNode: (id: string) => void; onAddSubscription: () => void; - pinging: boolean; + pingPhase: NodePingPhase; + pingError: string | null; + onRefreshPings: () => void; disabled?: boolean; /** * OPTIONAL — for the orchestrator to wire from App. True when the exit is @@ -105,7 +108,9 @@ export const ServerList = forwardRef( onQuery, onSelectNode, onAddSubscription, - pinging, + pingPhase, + pingError, + onRefreshPings, disabled = false, auto, onSelectAuto, @@ -156,6 +161,7 @@ export const ServerList = forwardRef( // as the core does when it picks. Null until something has a ping: shown as a // neutral row rather than a guessed name. const bestRow = useMemo(() => { + if (pingPhase !== "ready") return null; let best: ServerRow | null = null; let bestRtt = Infinity; for (const r of rows) { @@ -165,14 +171,23 @@ export const ServerList = forwardRef( } } return best; - }, [rows]); + }, [pingPhase, rows]); // AUTO is active whenever no node is pinned by hand. The `auto` prop is // authoritative once App wires it; until then "no active node" is the honest // stand-in (exact while idle; connected-auto needs the prop). const isAuto = auto ?? activeNodeId === null; - const online = rows.filter((r) => !r.dead && !r.stale && r.rttMs !== null).length; + const online = pingPhase === "ready" + ? rows.filter((r) => !r.dead && !r.stale && r.rttMs !== null).length + : 0; + const reachabilityLabel = pingPhase === "ready" + ? `${online} ${t.servers.online}` + : pingPhase === "checking" + ? t.servers.tcpChecking + : pingPhase === "failed" + ? t.servers.tcpFailed + : t.servers.tcpNotChecked; const insecureCount = rows.filter((r) => r.insecure).length; const insecureSummary = t.servers.insecureSummary .replace("{n}", String(insecureCount)) @@ -239,8 +254,24 @@ export const ServerList = forwardRef(

{t.servers.title} - {online} {t.servers.online} - {pinging && · …} + + + {reachabilityLabel} + + {pingPhase === "failed" && ( + + )}

@@ -352,7 +383,7 @@ export const ServerList = forwardRef( const active = !isAuto && s.id === activeNodeId; const pingCls = s.dead ? " dead" - : s.rttMs !== null && s.rttMs >= 120 + : !s.stale && s.rttMs !== null && s.rttMs >= 120 ? " hi" : ""; return ( @@ -392,7 +423,7 @@ export const ServerList = forwardRef( {s.city && {s.city}}
- {!s.dead && s.rttMs !== null ? ( + {!s.dead && !s.stale && s.rttMs !== null ? ( ) : (