From b56aa0d07fa9f794990b9847da40ddf0b2de0e6d Mon Sep 17 00:00:00 2001 From: Moses Narrow <36607567+0pcom@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:03:19 -0500 Subject: [PATCH] feat(interstitial): stream live route-setup progress via chunked encoding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The route interstitial can now hold the browser connection open and stream live progress via HTTP chunked transfer-encoding, reloading into the live content once the route is up — instead of a one-shot page + blind meta-refresh. New pkg/proxyinterstitial/stream.go: StreamConn (chunked net.Conn for the SOCKS path) + DriveStream + StreamOpen/StreamStep/StreamClose (shared progressive-HTML rendering). A caller-supplied Probe drives a real route-setup attempt; each attempt's real StatusLine is flushed as a progress line, and on success a final chunk reloads the page. HTTP/1.0 clients fall back to the existing one-shot meta-refresh page. Wiring (only real signals; no fabricated per-hop events): - browse-origin (meshInterstitialRT) streams the SAME detached round-trip it already runs — the one surface with a genuine in-flight attempt to observe — reloading on success; ReverseProxy FlushInterval<0 flushes each fragment. - dmsgweb / skynetweb SOCKS plaintext path drive a fresh dial via a re-dial Probe (dmsgRedialProbe / skynetRedialProbe) and stream each attempt. Seam reported (not built here): the granular "2-hop via -> noise handshake -> route group up" narrative needs a router-side progress callback (DialOptions has no observer; the mux harness only synthesizes coarse events around its own DialRoutes). pkg/skysocks ServeSOCKS5 is left one-shot (its mint point tears the session down with nothing to observe). Details in docs/proxy-status-and-interstitial.md. Tests: streaming success/hard-error/HTTP1.0-fallback/escaping for the primitive; updated the cold meshInterstitialRT test + a streamed-success test. --- docs/proxy-status-and-interstitial.md | 44 +++++ pkg/dmsgweb/runtime.go | 60 ++++++- pkg/proxyinterstitial/interstitial.go | 4 + pkg/proxyinterstitial/stream.go | 250 ++++++++++++++++++++++++++ pkg/proxyinterstitial/stream_test.go | 108 +++++++++++ pkg/skynetweb/runtime.go | 55 +++++- pkg/visor/meshproxy.go | 96 ++++++++-- pkg/visor/meshproxy_test.go | 41 ++++- 8 files changed, 640 insertions(+), 18 deletions(-) create mode 100644 pkg/proxyinterstitial/stream.go create mode 100644 pkg/proxyinterstitial/stream_test.go diff --git a/docs/proxy-status-and-interstitial.md b/docs/proxy-status-and-interstitial.md index 6b9513cd2c..bce15c6fc6 100644 --- a/docs/proxy-status-and-interstitial.md +++ b/docs/proxy-status-and-interstitial.md @@ -34,6 +34,50 @@ anyway) cleanly falls through to the real error instead of a per-request log spam. The page itself gained a footer with a deep-link to the surface's status host (below). +## 1b. Streaming interstitial (live route-setup progress) + +The interstitial can hold the browser connection OPEN and stream **live** route- +setup progress via HTTP chunked transfer-encoding, instead of the one-shot +`Content-Length` page + client meta-refresh. `pkg/proxyinterstitial/stream.go` +(`StreamConn` / `DriveStream` / `StreamOpen`+`StreamStep`+`StreamClose`) renders a +progressive-HTML shell and flushes a line per real attempt, then — once the route +is up — flushes a final chunk that reloads the page into the now-live content. + +**What signal is real (the seam).** The interstitial is minted *after* a dial has +already failed transiently; there is no in-flight route-setup to subscribe to at +that point, and `pkg/router` exposes **no** per-hop / noise-handshake progress +hook — `DialOptions` has no observer, and even the mux-telemetry harness only +synthesizes coarse `RouteEstablished`/failed events around its own `DialRoutes` +call. So the granular `finding route → 2-hop via → noise handshake → route +group up` narrative is **not observable today**. Two honest sources are used +instead: + +- **browse-origin** (`meshInterstitialRT`): it already runs the real cold-route + round-trip in the background; the stream renders that actual in-flight + attempt's outcome (`streamingInterstitialResponse`, flushed via the reverse + proxy's `FlushInterval < 0`). +- **SOCKS resolving proxies** (`dmsgweb` / `skynetweb`): the mint point has no + live attempt, so the streamer DRIVES a fresh one via a `Probe` (`dmsgRedialProbe` + / `skynetRedialProbe`) and streams each real attempt's `StatusLine` + the + success. Coarse, but real. + +The **missing seam** to get the granular lines is a router-side progress callback +— e.g. `DialOptions.OnProgress func(phase RouteSetupPhase)` invoked by +`DialRoutes` / the cascade setup path as hops resolve and the noise handshake +completes. That lives in `pkg/router` (owned separately) and is intentionally +**not** added here; this PR streams only the coarse signal actually available and +does not fabricate per-hop events. `pkg/skysocks`'s `ServeSOCKS5` mint point is +also left one-shot: it tears its session down for reconnect, with no attempt to +drive/observe from that function. + +**Fallback.** `StreamConn` reads the request first and serves the existing +one-shot meta-refresh page to an HTTP/1.0 client; a nil probe / absent event +source likewise degrades to the static page. + +**Liveness tie-in.** The same probe *is* the "is-my-connection-up" signal; a +future `status.*` mux view can share it as a WS/WT liveness indicator (extension +point, not built here). + ## 2. Per-proxy status hosts Each proxy serves a read-only diagnostic page at a reserved, well-known host diff --git a/pkg/dmsgweb/runtime.go b/pkg/dmsgweb/runtime.go index 98260f5e81..7c5847d220 100644 --- a/pkg/dmsgweb/runtime.go +++ b/pkg/dmsgweb/runtime.go @@ -355,9 +355,20 @@ func serveSOCKS5Direct(ctx context.Context, log *logging.Logger, dmsgC *dmsg.Cli } switch { case proxyinterstitial.ShouldServe(origPort): + // Stream REAL route-setup progress: hold the browser open with a + // chunked response and drive a fresh dial via a probe, flushing a + // line per real attempt, then reload into live content once the + // route is warm. Falls back to the one-shot page for an HTTP/1.0 + // client (handled inside StreamConn). Wrapped in tcpAddrConn: the + // stream conn is a net.Pipe (pipeAddr), and go-socks5 asserts + // *net.TCPAddr on the returned conn when building its BND reply. log.WithField("host", target).WithField("err", dialErr). - Debug("SOCKS5 → serving branded route interstitial") - conn, dialErr = proxyinterstitial.Conn(target, proxyinterstitial.StatusLine(dialErr), "dmsg", false), nil + Debug("SOCKS5 → serving streaming route interstitial") + conn, dialErr = &tcpAddrConn{Conn: proxyinterstitial.StreamConn(context.Background(), proxyinterstitial.StreamConfig{ + Target: target, + Mechanism: "dmsg", + Probe: dmsgRedialProbe(dmsgC, upstream, cfg, origHost, addr, origPort), + })}, nil case cfg.TLSMITM && isTLSPort(origPort, cfg.TLSPort) && skynetca.Permits(cfg.LeafMinter, origHost): // HTTPS request whose route is still warming: terminate the // browser's TLS locally with a per-host leaf and serve the @@ -586,6 +597,51 @@ func (f *upstreamForwarder) dial(network, addr string) (net.Conn, error) { return conn, err } +// dmsgRedialProbe builds a proxyinterstitial.Probe that re-attempts the dial the +// streaming interstitial stands in for, reporting the route ready (nil) once it +// succeeds. It is a READINESS probe: for a .dmsg destination it re-dials the +// destination directly (the common case) — a coarse but real signal, since +// pkg/router/dmsg expose no per-hop setup event to observe (see +// pkg/proxyinterstitial/stream.go). A non-.dmsg target re-attempts the +// upstream/direct forward. On success the opened stream is closed immediately; +// the browser's reload then rides the now-warm session/route. +func dmsgRedialProbe(dmsgC *dmsg.Client, upstream *upstreamForwarder, cfg Config, origHost, addr, origPort string) proxyinterstitial.Probe { + return func(ctx context.Context) error { + hostOnly := origHost + if i := strings.IndexByte(hostOnly, ':'); i >= 0 { + hostOnly = hostOnly[:i] + } + if strings.HasSuffix(hostOnly, cfg.DomainSuffix) { + hp := origHost + if origPort != "" && origPort != "80" { + hp = origHost + ":" + origPort + } + _, _, dest, port, perr := skynetweb.ParseResolverHost(hp, cfg.DomainSuffix, cfg.Aliases) + if perr != nil { + return perr + } + c, e := dmsgC.Dial(ctx, dmsg.Addr{PK: dest, Port: port}) + if e == nil { + _ = c.Close() //nolint:errcheck + } + return e + } + var ( + c net.Conn + e error + ) + if upstream != nil { + c, e = upstream.dial("tcp", addr) + } else { + c, e = net.Dial("tcp", addr) + } + if e == nil { + _ = c.Close() //nolint:errcheck + } + return e + } +} + // Context keys used by the SOCKS5 resolver ↔ Dial callback handshake. type ( dmsgResolverPortKey_t struct{} // set when .dmsg matched → presence is the signal; value unused diff --git a/pkg/proxyinterstitial/interstitial.go b/pkg/proxyinterstitial/interstitial.go index 2cb39a30b9..7a46e73beb 100644 --- a/pkg/proxyinterstitial/interstitial.go +++ b/pkg/proxyinterstitial/interstitial.go @@ -183,6 +183,10 @@ const css = `:root{--bg:#0b0d17;--fg:#c7cbe6;--muted:#7a80a8;--accent:#7c83ff;-- `#mesh-host{display:block;margin-top:14px;font:12px/1.4 ui-monospace,SFMono-Regular,monospace;color:var(--muted);opacity:.75;word-break:break-all}` + `.foot{margin-top:16px;padding-top:12px;border-top:1px solid var(--line);font-size:11.5px;color:var(--muted)}` + `.foot a{color:var(--accent);text-decoration:none}.foot a:hover{text-decoration:underline}` + + // Streaming (chunked) variant: live per-attempt progress lines. Reuses the + // .steps list; adds a completed (◐→●), errored (✗) and "route up" style. + `.steps li.err{color:var(--err)}.steps li.err::before{content:"✗";color:var(--err);animation:none}` + + `.ready{color:var(--accent);margin:12px 0 0;font-size:13px;font-weight:600}` + `.err #mesh-title{color:var(--err)}.err .sp{display:none}` // httpResponse wraps the HTML in a minimal HTTP/1.1 response. Connection:close diff --git a/pkg/proxyinterstitial/stream.go b/pkg/proxyinterstitial/stream.go new file mode 100644 index 0000000000..47807bd2cf --- /dev/null +++ b/pkg/proxyinterstitial/stream.go @@ -0,0 +1,250 @@ +// Package proxyinterstitial pkg/proxyinterstitial/stream.go c4-app-web +// +// Streaming (HTTP chunked-transfer) variant of the route interstitial. Instead +// of the one-shot Content-Length page + client-side meta-refresh (interstitial.go), +// this holds the browser connection OPEN and flushes a progressive-HTML chunk per +// REAL route-setup attempt as it happens, then — once the route is up — flushes a +// final chunk that reloads the page into the now-live content. +// +// What "real" means here, precisely. The interstitial is minted AFTER a dial has +// already failed transiently; there is no in-flight route-setup to subscribe to +// at that point, and pkg/router exposes no per-hop / noise-handshake progress +// hook (DialOptions has no observer; even the mux-telemetry harness only +// synthesizes coarse "established/failed" events around its own DialRoutes call). +// So the streamer DRIVES a fresh attempt via a caller-supplied Probe and streams +// the real outcome of each attempt (StatusLine of the actual error, then success). +// That is the honest, coarse signal available today; the granular +// "2-hop via → noise handshake → route group up" lines the design sketches +// require a NEW router-side progress callback — see the package doc / the +// docs/proxy-status-and-interstitial.md "streaming seam" note. This file does not +// invent those events. +// +// Graceful fallback: an HTTP/1.0 client (no reliable chunked/progressive render) +// is served the existing one-shot page instead. +package proxyinterstitial + +import ( + "bufio" + "context" + "fmt" + "html" + "io" + "net" + "net/http" + "strings" + "time" +) + +// Probe attempts to establish/verify the route to the interstitial's target. +// It returns nil when the route is ready (the browser should reload into live +// content), or an error while still warming. The streamer calls it on a loop +// (StreamConfig.Interval) until it succeeds, a non-transient error occurs, or +// StreamConfig.Deadline elapses. A successful Probe should leave the route warm +// (e.g. in the resolver's route pool) so the browser's reload hits real content. +type Probe func(ctx context.Context) error + +// StreamConfig configures a streaming interstitial. +type StreamConfig struct { + Target string // host shown on the page (HTML-escaped by the renderer) + Mechanism string // "skysocks" / "dmsg" / "skynet" — brands the copy + Probe Probe // drives + observes the real route-setup attempt + Interval time.Duration + Deadline time.Duration +} + +func (c StreamConfig) withDefaults() StreamConfig { + if c.Interval <= 0 { + c.Interval = 1 * time.Second + } + if c.Deadline <= 0 { + c.Deadline = 30 * time.Second + } + return c +} + +// StreamConn returns a net.Conn that serves a chunked, progressively-rendered +// interstitial driven by cfg.Probe, suitable for handing to go-socks5 as the +// tunnel conn (same injection model as Conn). Writes (the browser's request) are +// read to completion first; an HTTP/1.0 request falls back to the one-shot page. +func StreamConn(ctx context.Context, cfg StreamConfig) net.Conn { + cfg = cfg.withDefaults() + srvConn, cliConn := net.Pipe() + go func() { + defer srvConn.Close() //nolint:errcheck + br := bufio.NewReader(srvConn) + req, err := http.ReadRequest(br) + if err != nil { + return + } + // HTTP/1.0 clients don't reliably render a streamed/chunked body — serve + // the one-shot page (still branded + auto-refreshing) instead. + if req.ProtoMajor == 1 && req.ProtoMinor == 0 { + _, _ = srvConn.Write(httpResponse(cfg.Target, "", cfg.Mechanism, false)) //nolint:errcheck + return + } + head := "HTTP/1.1 200 OK\r\n" + + "Content-Type: text/html; charset=utf-8\r\n" + + "Transfer-Encoding: chunked\r\n" + + "Cache-Control: no-store, must-revalidate\r\n" + + "Connection: close\r\n\r\n" + if _, err := io.WriteString(srvConn, head); err != nil { + return + } + cw := &chunkWriter{w: srvConn} + DriveStream(ctx, cw, cfg) + _ = cw.End() //nolint:errcheck + }() + return cliConn +} + +// DriveStream renders the streaming interstitial to w: the opening shell, one +// progress line per real Probe attempt, and a closing chunk (reload-into-content +// on success, or an error + manual-retry on a hard failure / deadline). Exported +// so a caller that already streams over an http.ResponseWriter / io.Pipe (e.g. +// the browse-origin reverse proxy) can reuse the exact same rendering without the +// chunked-conn framing. w is written as raw HTML fragments; the HTTP layer +// (chunkWriter for the SOCKS path, the net/http server for the proxy path) frames +// and flushes them. +func DriveStream(ctx context.Context, w io.Writer, cfg StreamConfig) { + cfg = cfg.withDefaults() + emit := func(s string) bool { _, err := io.WriteString(w, s); return err == nil } + if f, ok := w.(flusher); ok { + orig := emit + emit = func(s string) bool { + if !orig(s) { + return false + } + f.Flush() + return true + } + } + + if !emit(StreamOpen(cfg.Target, cfg.Mechanism)) { + return + } + emit(StreamStep("Building a route over "+mechanismLabel(cfg.Mechanism)+"…", "active")) + + if cfg.Probe == nil { + emit(StreamClose(false, "no route-setup probe available")) + return + } + + deadline := time.Now().Add(cfg.Deadline) + ticker := time.NewTicker(cfg.Interval) + defer ticker.Stop() + for { + err := cfg.Probe(ctx) + if err == nil { + emit(StreamStep("Route group up", "done")) + emit(StreamClose(true, "")) + return + } + if !IsTransient(err) { + emit(StreamStep(StatusLine(err), "err")) + emit(StreamClose(false, err.Error())) + return + } + if time.Now().After(deadline) { + emit(StreamStep(StatusLine(err), "err")) + emit(StreamClose(false, "route still warming — try again")) + return + } + emit(StreamStep(StatusLine(err), "active")) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// flusher is net/http's http.Flusher without importing it here (the SOCKS path's +// chunkWriter also implements it), so DriveStream flushes each fragment on either +// transport. +type flusher interface{ Flush() } + +// StreamOpen is the opening HTML of a streaming interstitial: the branded card +// header and the (open) live progress list. Subsequent StreamStep fragments +// append
  • rows to it; StreamClose finishes the list and the document. +func StreamOpen(target, _ string) string { + title := "Connecting over skywire…" + heading := "Building a route over the mesh…" + msg := "Establishing a private route to this site. Live progress below." + return `` + + `` + + `` + title + `` + + `
    ` + + brandMark + + `
    ` + + `

    ` + html.EscapeString(heading) + `

    ` + + `

    ` + html.EscapeString(msg) + `

    ` + + `` + html.EscapeString(strings.TrimSpace(target)) + `` + + `
      ` + + // Enough leading padding so browsers begin incremental render immediately + // rather than buffering a minimum before first paint. + "" +} + +// StreamStep is one live progress line. state is "active" (in progress), +// "done" (completed) or "err" (failed). text is HTML-escaped. +func StreamStep(text, state string) string { + cls := "active" + switch state { + case "done": + cls = "done" + case "err": + cls = "err" + } + return `
    • ` + html.EscapeString(text) + `
      • ` +} + +// StreamClose finishes the document. On ok it appends a "route up" line and a +// script that reloads the SAME URL — which now streams real content because the +// successful probe left the route warm. On failure it shows detail + a manual +// retry, and does NOT auto-reload. +func StreamClose(ok bool, detail string) string { + var b strings.Builder + b.WriteString(`
    `) + if ok { + b.WriteString(`

    Route up — loading the page…

    `) + // replace() so the interstitial isn't left in history; the reload + // re-requests through the proxy and hits the now-warm route. + b.WriteString(``) + } else { + if d := strings.TrimSpace(detail); d != "" { + b.WriteString(`

    ` + html.EscapeString(d) + `

    `) + } + b.WriteString(``) + } + b.WriteString(`
    `) + return b.String() +} + +// chunkWriter frames each Write as a single HTTP/1.1 chunk and flushes it (a +// net.Pipe Write already delivers synchronously, so Flush is a no-op but lets +// chunkWriter satisfy the flusher interface DriveStream keys on). +type chunkWriter struct{ w io.Writer } + +func (c *chunkWriter) Write(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + if _, err := fmt.Fprintf(c.w, "%x\r\n", len(p)); err != nil { + return 0, err + } + if _, err := c.w.Write(p); err != nil { + return 0, err + } + if _, err := io.WriteString(c.w, "\r\n"); err != nil { + return 0, err + } + return len(p), nil +} + +func (c *chunkWriter) Flush() {} + +// End writes the terminating zero-length chunk. +func (c *chunkWriter) End() error { + _, err := io.WriteString(c.w, "0\r\n\r\n") + return err +} diff --git a/pkg/proxyinterstitial/stream_test.go b/pkg/proxyinterstitial/stream_test.go new file mode 100644 index 0000000000..b56a20d26a --- /dev/null +++ b/pkg/proxyinterstitial/stream_test.go @@ -0,0 +1,108 @@ +package proxyinterstitial + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" +) + +// readStreamBody drives one request/response over the streaming conn and returns +// the fully de-chunked body. +func readStreamBody(t *testing.T, c net.Conn, reqLine string) string { + t.Helper() + go func() { _, _ = c.Write([]byte(reqLine)) }() + resp, err := http.ReadResponse(bufio.NewReader(c), nil) + if err != nil { + t.Fatalf("ReadResponse: %v", err) + } + defer resp.Body.Close() //nolint:errcheck + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + return string(body) +} + +func TestStreamConnReloadsOnReady(t *testing.T) { + var n int32 + cfg := StreamConfig{ + Target: "host.skynet", Mechanism: "skynet", + Interval: 2 * time.Millisecond, Deadline: 2 * time.Second, + Probe: func(context.Context) error { + if atomic.AddInt32(&n, 1) < 3 { // fail twice, then succeed + return errors.New("no route to host") + } + return nil + }, + } + c := StreamConn(context.Background(), cfg) + body := readStreamBody(t, c, "GET / HTTP/1.1\r\nHost: host.skynet\r\n\r\n") + + for _, want := range []string{"skywire", "host.skynet", "Building a route", "Route group up", "location.replace(location.href)"} { + if !strings.Contains(body, want) { + t.Errorf("streamed body missing %q\n---\n%s", want, body) + } + } + if strings.Contains(body, "Retry") { + t.Error("success stream must not offer a manual retry") + } + if got := atomic.LoadInt32(&n); got < 3 { + t.Errorf("probe called %d times, expected >=3", got) + } +} + +func TestStreamConnHardError(t *testing.T) { + cfg := StreamConfig{ + Target: "host.dmsg", Mechanism: "dmsg", + Interval: 2 * time.Millisecond, Deadline: 2 * time.Second, + Probe: func(context.Context) error { + return errors.New("permanently broken widget") // not transient + }, + } + c := StreamConn(context.Background(), cfg) + body := readStreamBody(t, c, "GET / HTTP/1.1\r\nHost: host.dmsg\r\n\r\n") + + if !strings.Contains(body, "Retry") { + t.Errorf("hard-error stream should offer a manual retry\n%s", body) + } + if strings.Contains(body, "location.replace") { + t.Error("hard-error stream must not auto-reload") + } +} + +func TestStreamConnHTTP10Fallback(t *testing.T) { + cfg := StreamConfig{ + Target: "host.skynet", Mechanism: "skynet", + Probe: func(context.Context) error { return nil }, + } + c := StreamConn(context.Background(), cfg) + body := readStreamBody(t, c, "GET / HTTP/1.0\r\nHost: host.skynet\r\n\r\n") + + // The one-shot page (auto-refresh), not the streamed one. + if !strings.Contains(body, `http-equiv="refresh"`) { + t.Errorf("HTTP/1.0 client should get the one-shot meta-refresh page\n%s", body) + } + if strings.Contains(body, "location.replace") { + t.Error("fallback page should not carry the stream reload script") + } +} + +func TestStreamConnEscapesTarget(t *testing.T) { + cfg := StreamConfig{ + Target: "", Mechanism: "skynet", + Interval: time.Millisecond, Deadline: time.Second, + Probe: func(context.Context) error { return nil }, + } + c := StreamConn(context.Background(), cfg) + body := readStreamBody(t, c, "GET / HTTP/1.1\r\nHost: x\r\n\r\n") + if strings.Contains(body, "