diff --git a/cmd/codeaf/chatv3_at.go b/cmd/codeaf/chatv3_at.go index 3a1fe34ca4..8a4f04a476 100644 --- a/cmd/codeaf/chatv3_at.go +++ b/cmd/codeaf/chatv3_at.go @@ -262,7 +262,7 @@ func askPairingCode(name string) (string, error) { func runServe(args []string) error { flags := commandFlags("serve") workspace := flags.String("workspace", "", "directory a connection works in when it does not name one; empty is the directory this command was run in") - relayAddress := flags.String("via", "", "the sync address to be reachable through; empty reads "+pair.RelayEnv+", then the address this machine syncs through") + relayAddress := flags.String("via", "", "the sync address to be reachable through; empty reads "+pair.RelayEnv+", ~/.codeaf/v3/remote/relay, then the address this machine syncs through") renamedFlag(flags, "relay", "via") if err := parseCommandFlags(flags, args); err != nil { return err diff --git a/cmd/codeaf/chatv3_at_test.go b/cmd/codeaf/chatv3_at_test.go index 9d275095a9..2e94b34071 100644 --- a/cmd/codeaf/chatv3_at_test.go +++ b/cmd/codeaf/chatv3_at_test.go @@ -1,6 +1,9 @@ package main import ( + "bytes" + "os" + "path/filepath" "strings" "testing" @@ -146,3 +149,49 @@ func TestDevicesOnAFreshMachineSaysNothingIsPaired(t *testing.T) { t.Fatal("an unknown word was accepted") } } + +// Contract item 7 (Finding 10): `codeaf serve --help` names the full relay +// precedence — CODEAF_RELAY, ~/.codeaf/v3/remote/relay, then sync — matching +// what pair.Relay and pair.RelayFor actually resolve. +func TestServeHelpAndRelayPrecedenceIncludeSavedRemoteRelayFile(t *testing.T) { + var buf bytes.Buffer + previous := usageOut + usageOut = &buf + t.Cleanup(func() { usageOut = previous }) + + if err := runServe([]string{"--help"}); err != exitHelped { + t.Fatalf("runServe(--help) returned %v, want exitHelped", err) + } + help := strings.Join(strings.Fields(buf.String()), " ") + envPos := strings.Index(help, pair.RelayEnv) + filePos := strings.Index(help, "~/.codeaf/v3/remote/relay") + syncPos := strings.Index(help, "syncs through") + if envPos < 0 || filePos < 0 || syncPos < 0 || !(envPos < filePos && filePos < syncPos) { + t.Fatalf("serve --help did not list %s, ~/.codeaf/v3/remote/relay, then sync in order: %q", pair.RelayEnv, help) + } + + t.Setenv("CODEAF_HOME", t.TempDir()) + t.Setenv(pair.RelayEnv, "") + t.Setenv(syncsetup.URLVar, "https://sync.example") + if err := os.MkdirAll(filepath.Dir(pair.RelayFile()), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(pair.RelayFile(), []byte("https://file.example/fabric\n"), 0o600); err != nil { + t.Fatal(err) + } + + if got := pair.Relay(syncRelayURL()); got != "https://file.example/fabric" { + t.Fatalf("serve precedence with v3/remote/relay gave %q, want file override", got) + } + if got := pair.RelayFor(pair.Known{}, false, syncRelayURL()); got != "https://file.example/fabric" { + t.Fatalf("unpaired --at precedence with v3/remote/relay gave %q, want file override", got) + } + paired := pair.Known{Service: "https://paired.example/fabric"} + if got := pair.RelayFor(paired, true, syncRelayURL()); got != "https://paired.example/fabric" { + t.Fatalf("paired --at precedence gave %q, want paired machine service over file", got) + } + t.Setenv(pair.RelayEnv, "https://env.example/fabric") + if got := pair.RelayFor(paired, true, syncRelayURL()); got != "https://env.example/fabric" { + t.Fatalf("CODEAF_RELAY precedence gave %q, want env override", got) + } +} diff --git a/docs/changes/unreleased/2076-relay-machine-door-hardening.md b/docs/changes/unreleased/2076-relay-machine-door-hardening.md new file mode 100644 index 0000000000..4fcfc847f2 --- /dev/null +++ b/docs/changes/unreleased/2076-relay-machine-door-hardening.md @@ -0,0 +1,25 @@ +--- +kind: fixed +title: a stranger can no longer hold, squat or flood a machine reached through the relay +pr: 2076 +surface: [remote] +invalidates: + - >- + A relay stream had no deadline, so the serving machine's 30-second + handshake limit never applied: silent dials could hold all 16 of a + machine's slots, and one byte could burn its pairing code and freeze + pairing. Streams now end at the handshake deadline. + - >- + The hosted relay kept its rate limits per machine name. It now limits each + caller network at the front door (30 dials and 10 registrations a minute) + and refuses names codeaf cannot produce before any object is created. + - >- + A registration with a different key could replace a live machine that + stalled for five seconds. Only the same key may replace a silent one. + - >- + codeaf serve printed "lost sync — trying again" and then nothing when it + recovered. It now answers with "back — reachable as again". +--- + +Found by a review of #2073 after its machine doors went live on the hosted +relay, and checked by hand against the Worker run locally. diff --git a/internal/manual/chat/reaching-this-machine-without-ssh.md b/internal/manual/chat/reaching-this-machine-without-ssh.md index 7ac4ea529c..423adf9173 100644 --- a/internal/manual/chat/reaching-this-machine-without-ssh.md +++ b/internal/manual/chat/reaching-this-machine-without-ssh.md @@ -63,8 +63,11 @@ digits, the same every time, on any relay. You cannot pick a nicer one, and nobo can register under yours while your machine is connected. `--workspace` is the directory a connection works in when it does not name one; empty -means the directory you ran the command in. The relay is, in order: `--via`, the -`CODEAF_RELAY` environment variable, then the relay this machine syncs through. +means the directory you ran the command in. For `codeaf serve`, the relay is, in order: +`--via`, the `CODEAF_RELAY` environment variable, `~/.codeaf/v3/remote/relay`, then the +relay this machine syncs through. For `codeaf chat --at`, it is `CODEAF_RELAY`, the relay +that machine was paired through, `~/.codeaf/v3/remote/relay`, then the relay this machine +syncs through. Running it twice on one machine is refused, because there is only one of it: @@ -80,8 +83,22 @@ If the network drops while it runs, it says once: lost sync — trying again ``` -and keeps trying, waiting a little longer each time up to a minute, until it prints its -name and code again. It does not need restarting, and devices reach it as soon as it has. +and keeps trying, waiting a little longer each time up to a minute, resetting that wait +only after a connection has held for at least a minute. When it reconnects, it says once: + +``` +back — reachable as otter-lamp-42 again +``` + +It prints its name and code again only if the name or code changed. It does not need +restarting, and devices reach it as soon as it is back. + +If the relay turns registrations away — it allows 10 registrations a minute from one +address — `codeaf serve` backs off the same way and says once: + +``` +the service is turning registrations away right now — waiting before trying again +``` A relay that answers but cannot connect machines, such as an older sync service, stops it with: @@ -410,7 +427,9 @@ Two more you may meet. A machine that does not answer the handshake at all: `otter-lamp-42 did not answer this device's handshake, so nothing was sent — if that machine was rebuilt it has a new key and this device has to pair with it again`. And a relay that is turning connections away: `the service is turning connections away right now — try again in a minute` — it allows -30 connections a minute from one address. +30 connections a minute from one address. If a proxy or wrong path at the relay address +refuses the upgrade, it names the full address it tried: +`the relay at https://service.example.com/wrong-prefix refused this connection (404) — check the address it was given`. **A pairing that says the code was wrong when you are sure it was right** has one other cause, and it is the machine's own disk. The machine writes a device into its list diff --git a/internal/pair/edge_serve_test.go b/internal/pair/edge_serve_test.go index a7e93f9796..7c0009de94 100644 --- a/internal/pair/edge_serve_test.go +++ b/internal/pair/edge_serve_test.go @@ -8,6 +8,7 @@ package pair import ( "context" "fmt" + "io" "net" "net/http" "net/http/httptest" @@ -276,7 +277,8 @@ func TestServeApproveGetsTheDeviceLabel(t *testing.T) { // A DROPPED NETWORK IS ONE LINE, NOT ONE LINE A RETRY. While the relay is gone // the machine says once that it lost sync and is trying again, keeps trying -// without being restarted, and shows its name again when the relay is back. +// without being restarted, and is reachable again when the relay is back +// without reprinting a banner whose name and code have not changed. func TestServeSaysOnceThatItLostSyncAndComesBack(t *testing.T) { t.Setenv("CODEAF_HOME", t.TempDir()) var down atomic.Bool @@ -304,25 +306,57 @@ func TestServeSaysOnceThatItLostSyncAndComesBack(t *testing.T) { wire.cut() time.Sleep(4 * time.Second) // the first retries come at one and two seconds down.Store(false) - lines := screen.until(t, "its name again", func(l []string) bool { return len(codes(l)) == 2 }) + waitForCondition(t, "the machine to register again", func() bool { + return len(door.Ledgers()) == 1 + }) + lines := screen.all() lost := 0 for _, line := range lines { - if line == "lost sync — trying again" { + if line == LostSyncLine { lost++ } } if lost != 1 { t.Fatalf("the machine said it lost sync %d times over one outage; want once: %q", lost, lines) } + if got := len(codes(lines)); got != 1 { + t.Fatalf("the unchanged banner was printed %d times across one brief outage; want 1: %q", got, lines) + } + // The lost line is answered by exactly one back line, so the screen of a + // machine that recovered never ends on "lost sync". + screen.until(t, "the back line", func(l []string) bool { + for _, line := range l { + if line == BackLine(host.Device.Name()) { + return true + } + } + return false + }) + back := 0 + for _, line := range screen.all() { + if line == BackLine(host.Device.Name()) { + back++ + } + } + if back != 1 { + t.Fatalf("the machine said it was back %d times after one outage; want once: %q", back, screen.all()) + } } -// A RELAY THAT STILL HOLDS THE OLD CONNECTION IS WAITED OUT. After a network -// drop the relay may keep this machine's name for the dead connection until it -// times out; a machine that was serving keeps trying instead of quitting as if -// a second `codeaf serve` had taken its name. +// A RELAY THAT STILL HOLDS A LIVE SOCKET FOR THIS NAME IS WAITED OUT. While +// another live carrier holds this machine's name after a drop, a machine that +// was already serving keeps trying instead of quitting, and comes back as soon +// as that carrier ends. func TestServeWaitsOutItsOwnStaleName(t *testing.T) { t.Setenv("CODEAF_HOME", t.TempDir()) - door := &relay.Server{} + var connected atomic.Int32 + door := &relay.Server{ + Note: func(_, what string) { + if what == "connected" { + connected.Add(1) + } + }, + } server := httptest.NewUnstartedServer(door) wire := &cuttable{Listener: server.Listener} server.Listener = wire @@ -350,7 +384,343 @@ func TestServeWaitsOutItsOwnStaleName(t *testing.T) { t.Fatalf("serve quit while its own old name was still held: %v", runErr) default: } - screen.until(t, "its name again", func(l []string) bool { return len(codes(l)) == 2 }) + waitForCondition(t, "the machine to retake its name", func() bool { + return connected.Load() >= 3 && len(door.Ledgers()) == 1 + }) +} + +// Contract item 1: SILENT DIALS END AT THE HANDSHAKE DEADLINE. Sixteen silent +// dials fill a served machine's stream ceiling only until the handshake +// deadline fires; once it does, the serving machine closes all sixteen and a +// real `--at` dial succeeds. +func TestSilentDialsCannotHoldAMachine(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + door, address := liveRelay(t) + t.Setenv(RelayEnv, address) + + screen := &screenLog{} + devices := BookAt(filepath.Join(t.TempDir(), "devices.json")) + machines := BookAt(filepath.Join(t.TempDir(), "machines.json")) + host := &Host{ + Service: address, + Device: aDevice(t), + Devices: devices, + Desk: &Desk{}, + Approve: func(string, string) bool { return true }, + Say: screen.say, + handshake: 60 * time.Millisecond, + Open: func(pipe io.ReadWriteCloser) { + defer func() { _ = pipe.Close() }() + _, _ = io.Copy(pipe, pipe) + }, + } + ctx, stop := context.WithCancel(context.Background()) + ended := make(chan struct{}) + go func() { _ = host.Run(ctx); close(ended) }() + t.Cleanup(func() { stop(); <-ended }) + first := codes(screen.until(t, "its first code", func(l []string) bool { return len(codes(l)) == 1 }))[0] + + // Pair one device first so the post-flood check exercises an ordinary + // paired `--at` dial. + reach := Reach{ + Name: host.Device.Name(), + Device: aDevice(t), + Machines: machines, + Label: "laptop", + AskCode: func(string) (string, error) { return first, nil }, + } + warm, err := reach.Open(context.Background()) + if err != nil { + t.Fatalf("initial pairing failed: %v", err) + } + _ = warm.Close() + + // Open 16 silent streams (relay.MaxStreams) and send zero bytes on each. + silent := make([]io.ReadWriteCloser, 0, relay.MaxStreams) + t.Cleanup(func() { + for _, s := range silent { + _ = s.Close() + } + }) + for i := 0; i < relay.MaxStreams; i++ { + s, err := relay.Dial(context.Background(), address, host.Device.Name()) + if err != nil { + t.Fatalf("silent dial %d: %v", i, err) + } + silent = append(silent, s) + } + + // Without waiting thirty real seconds, the short handshake deadline closes + // every silent stream and frees the machine's slots on the relay. + waitForCondition(t, "all 16 silent streams to close at the handshake deadline", func() bool { + ledgers := door.Ledgers() + return len(ledgers) == 1 && ledgers[0].Streams == 0 + }) + + tunnel, err := reach.Open(context.Background()) + if err != nil { + t.Fatalf("real --at dial after 16 silent dials failed: %v", err) + } + defer func() { _ = tunnel.Close() }() + const want = "still reachable" + if _, err := tunnel.Write([]byte(want)); err != nil { + t.Fatal(err) + } + buf := make([]byte, len(want)) + if _, err := io.ReadFull(tunnel, buf); err != nil { + t.Fatal(err) + } + if string(buf) != want { + t.Fatalf("echoed %q, want %q", string(buf), want) + } +} + +// Contract item 2: A DIAL THAT ONLY SAYS "PAIR" CANNOT FREEZE PAIRING. Sending +// the single `P` intent byte and then nothing spends the live code before the +// exchange (preserving the one-attempt rule), hits the handshake deadline, +// retires that code, and prints a fresh code that pairs a real device. +func TestHalfStartedPairIntentRotatesTheCode(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + _, address := liveRelay(t) + t.Setenv(RelayEnv, address) + + screen := &screenLog{} + devices := BookAt(filepath.Join(t.TempDir(), "devices.json")) + machines := BookAt(filepath.Join(t.TempDir(), "machines.json")) + host := &Host{ + Service: address, + Device: aDevice(t), + Devices: devices, + Desk: &Desk{}, + Approve: func(string, string) bool { return true }, + Say: screen.say, + handshake: 60 * time.Millisecond, + Open: func(pipe io.ReadWriteCloser) { + defer func() { _ = pipe.Close() }() + _, _ = io.Copy(pipe, pipe) + }, + } + ctx, stop := context.WithCancel(context.Background()) + ended := make(chan struct{}) + go func() { _ = host.Run(ctx); close(ended) }() + t.Cleanup(func() { stop(); <-ended }) + first := codes(screen.until(t, "its first code", func(l []string) bool { return len(codes(l)) == 1 }))[0] + + half, err := relay.Dial(context.Background(), address, host.Device.Name()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = half.Close() }) + if _, err := half.Write([]byte{intentPair}); err != nil { + t.Fatal(err) + } + + lines := screen.until(t, "the next code after a stalled P intent", func(l []string) bool { + return len(codes(l)) >= 2 + }) + if firstWith(lines, burnLine) < 0 { + t.Fatalf("the machine never reported the burned attempt: %q", lines) + } + second := codes(lines)[1] + if second == first { + t.Fatalf("the rotated code %q matched the spent code", second) + } + + reach := Reach{ + Name: host.Device.Name(), + Device: aDevice(t), + Machines: machines, + Label: "laptop", + AskCode: func(string) (string, error) { return second, nil }, + } + tunnel, err := reach.Open(context.Background()) + if err != nil { + t.Fatalf("pairing with the rotated code failed: %v", err) + } + _ = tunnel.Close() + if list, _ := devices.Devices(); len(list) != 1 { + t.Fatalf("the machine admitted %v, want 1 paired device", list) + } +} + +// Contract item 5: A DROPPED REGISTRATION BACKS OFF, PRINTS AN UNCHANGED BANNER +// ONCE, AND PHRASES RATE LIMITS FOR A PERSON. +func TestDroppedRegistrationBacksOffAndPrintsBannerOnce(t *testing.T) { + t.Run("ten immediate drops grow the wait each time up to the cap and print the banner once", func(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + var clockMu sync.Mutex + now := time.Now() + clock := func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return now + } + advance := func(d time.Duration) { + clockMu.Lock() + now = now.Add(d) + clockMu.Unlock() + } + + connected := make(chan struct{}, 32) + door := &relay.Server{ + Now: clock, + Note: func(_, what string) { + if what == "connected" { + connected <- struct{}{} + } + }, + } + server := httptest.NewUnstartedServer(door) + wire := &cuttable{Listener: server.Listener} + server.Listener = wire + server.Start() + t.Cleanup(server.Close) + + slept := make(chan time.Duration, 32) + screen := &screenLog{} + host := &Host{ + Service: server.URL, + Device: aDevice(t), + Devices: BookAt(filepath.Join(t.TempDir(), "devices.json")), + Desk: &Desk{Now: clock}, + Say: screen.say, + Now: clock, + sleep: func(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + advance(d) + slept <- d + return nil + }, + } + ctx, stop := context.WithCancel(context.Background()) + ended := make(chan struct{}) + go func() { _ = host.Run(ctx); close(ended) }() + t.Cleanup(func() { stop(); <-ended }) + + screen.until(t, "its first code", func(l []string) bool { return len(codes(l)) == 1 }) + awaitStep(t, connected, "initial registration") + + var waits []time.Duration + for i := 0; i < 10; i++ { + wire.cut() + waits = append(waits, awaitStep(t, slept, fmt.Sprintf("backoff %d", i+1))) + awaitStep(t, connected, fmt.Sprintf("reconnect %d", i+1)) + } + + wantWaits := []time.Duration{ + 1 * time.Second, + 2 * time.Second, + 4 * time.Second, + 8 * time.Second, + 16 * time.Second, + 32 * time.Second, + time.Minute, + time.Minute, + time.Minute, + time.Minute, + } + for i, want := range wantWaits { + if waits[i] != want { + t.Fatalf("backoff %d was %v, want %v (all: %v)", i+1, waits[i], want, waits) + } + } + if got := len(codes(screen.all())); got != 1 { + t.Fatalf("banner printed %d times across 10 immediate drops; want 1: %q", got, screen.all()) + } + + // Confirm Host.Run has entered hold (and recorded `since := h.now()`) + // before advancing the clock, by sending an unknown intent byte that + // Host.answer immediately closes without touching the code or screen. + probe, err := relay.Dial(context.Background(), server.URL, host.Device.Name()) + if err != nil { + t.Fatal(err) + } + _, _ = probe.Write([]byte{'?'}) + _, _ = io.Copy(io.Discard, probe) + _ = probe.Close() + + // Once a registration stays up for at least one minute, a later drop + // resets the backoff to 1s. + advance(steadyAfter + time.Second) + wire.cut() + if reset := awaitStep(t, slept, "backoff after steady registration"); reset != time.Second { + t.Fatalf("backoff after holding >= %v was %v, want 1s", steadyAfter, reset) + } + }) + + t.Run("ErrTooMany gets a person-facing sentence and backs off without leaking the sentinel", func(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + throttled := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "too many registrations from here — wait a minute", http.StatusTooManyRequests) + })) + t.Cleanup(throttled.Close) + + ctx, stop := context.WithCancel(context.Background()) + defer stop() + var waits []time.Duration + screen := &screenLog{} + host := &Host{ + Service: throttled.URL, + Device: aDevice(t), + Devices: BookAt(filepath.Join(t.TempDir(), "devices.json")), + Desk: &Desk{}, + Say: screen.say, + sleep: func(ctx context.Context, d time.Duration) error { + waits = append(waits, d) + if len(waits) == 3 { + stop() + return context.Canceled + } + return nil + }, + } + if err := host.Run(ctx); err != nil { + t.Fatalf("Host.Run under 429 returned %v, want nil on context cancel", err) + } + lines := screen.all() + if len(lines) != 1 || lines[0] != TooManyRegistrationsLine { + t.Fatalf("screen under 429 said %q; want [%q] once", lines, TooManyRegistrationsLine) + } + for _, line := range lines { + if strings.Contains(line, "relay:") { + t.Fatalf("raw package sentinel leaked to screen: %q", line) + } + } + want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} + for i := range want { + if waits[i] != want[i] { + t.Fatalf("429 backoff %d was %v, want %v (all: %v)", i+1, waits[i], want[i], waits) + } + } + }) +} + +func waitForCondition(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if ok() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +func awaitStep[T any](t *testing.T, ch <-chan T, what string) T { + t.Helper() + select { + case v := <-ch: + return v + case <-time.After(5 * time.Second): + var zero T + t.Fatalf("timed out waiting for %s", what) + return zero + } } // cuttable is a listener whose connections a test can drop, hijacked ones diff --git a/internal/pair/manual_test.go b/internal/pair/manual_test.go index 094eb42a3b..757817be78 100644 --- a/internal/pair/manual_test.go +++ b/internal/pair/manual_test.go @@ -34,6 +34,7 @@ func TestEverySentenceThisPackageShowsIsInTheManual(t *testing.T) { // The four ways `--at` fails, each naming its own cause. NoRelay("otter-lamp-42").Error(), Unreachable("https://service.example.com").Error(), + NoDoor("https://service.example.com").Error(), NotConnected("otter-lamp-42").Error(), NotPaired("otter-lamp-42").Error(), // And the ones a person meets less often. @@ -43,6 +44,9 @@ func TestEverySentenceThisPackageShowsIsInTheManual(t *testing.T) { Busy().Error(), Stopped(), RevokedLine("laptop"), + LostSyncLine, + BackLine("otter-lamp-42"), + TooManyRegistrationsLine, // The join words on the machine being opened, and what its own screen // says about the attempts it turned away. WaitingLine("otter-lamp-42", "amber fox dune"), @@ -117,11 +121,13 @@ func TestTheManualQuotesTheLimitsThisPackageActuallyApplies(t *testing.T) { t.Fatal("the page reaching-this-machine-without-ssh is not in the corpus") } for _, quoted := range []string{ - "10 minutes", // CodeValidFor - "1 attempt", // CodeAttempts - "up to 16", // relay.MaxStreams - "30 connections a minute", // relay.DialsPerMinute + "10 minutes", // CodeValidFor + "1 attempt", // CodeAttempts + "up to 16", // relay.MaxStreams + "30 connections a minute", // relay.DialsPerMinute + "10 registrations a minute", // relay.RegistrationsPerMinute "~/.codeaf/v3/remote/device.key", + "~/.codeaf/v3/remote/relay", } { if !strings.Contains(page, quoted) { t.Errorf("the page does not quote %q", quoted) @@ -133,6 +139,9 @@ func TestTheManualQuotesTheLimitsThisPackageActuallyApplies(t *testing.T) { if relay.DialsPerMinute != 30 { t.Fatalf("relay.DialsPerMinute is %d and the page says 30", relay.DialsPerMinute) } + if relay.RegistrationsPerMinute != 10 { + t.Fatalf("relay.RegistrationsPerMinute is %d and the page says 10", relay.RegistrationsPerMinute) + } if int(CodeValidFor.Minutes()) != 10 { t.Fatalf("CodeValidFor is %v and the page says 10 minutes", CodeValidFor) } diff --git a/internal/pair/serve.go b/internal/pair/serve.go index 88aaa6998f..56fed9b108 100644 --- a/internal/pair/serve.go +++ b/internal/pair/serve.go @@ -52,8 +52,35 @@ type Host struct { Open func(io.ReadWriteCloser) // Now is the clock, swapped by tests. Now func() time.Time + + // handshake overrides [HandshakeWithin] in tests so silent-stream + // deadlines can be checked without waiting thirty real seconds. + handshake time.Duration + // sleep overrides the retry wait in tests so backoff schedules can be + // checked with an injected clock. + sleep func(context.Context, time.Duration) error + + bannerMu sync.Mutex + shownBanner string } +// steadyAfter is how long a registration must stay up before a later drop is +// counted as a fresh outage rather than the same flapping connection. +const steadyAfter = time.Minute + +// TooManyRegistrationsLine is what `codeaf serve` prints when the relay turns +// registrations away at its per-minute ceiling. +const TooManyRegistrationsLine = "the service is turning registrations away right now — waiting before trying again" + +// LostSyncLine is what `codeaf serve` prints, once per outage, when its +// connection to the relay is lost and it starts trying again. +const LostSyncLine = "lost sync — trying again" + +// BackLine is what `codeaf serve` prints when it is reachable again after +// saying [LostSyncLine]. It names the machine because that is the fact a +// person checking the screen wants: devices can reach it under this name. +func BackLine(name string) string { return "back — reachable as " + name + " again" } + func (h *Host) now() time.Time { if h.Now != nil { return h.Now() @@ -61,12 +88,48 @@ func (h *Host) now() time.Time { return time.Now() } +func (h *Host) handshakeWithin() time.Duration { + if h.handshake > 0 { + return h.handshake + } + return HandshakeWithin +} + +func (h *Host) wait(ctx context.Context, d time.Duration) error { + if h.sleep != nil { + return h.sleep(ctx, d) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} + func (h *Host) say(line string) { if h.Say != nil { h.Say(line) } } +func (h *Host) banner(name string, code *Code, codeErr error) string { + if codeErr == nil { + return strings.TrimRight(Lines(name, code), "\n") + } + return " this machine is reachable as " + name +} + +func (h *Host) sayBanner(line string, force bool) { + h.bannerMu.Lock() + same := line == h.shownBanner + h.shownBanner = line + h.bannerMu.Unlock() + if force || !same { + h.say(line) + } +} + // Run holds the registration until the context is cancelled, redialling when // the connection to the relay is lost. // @@ -79,22 +142,36 @@ func (h *Host) Run(ctx context.Context) error { if strings.TrimSpace(h.Service) == "" { return errors.New("no sync address is set up on this machine, so there is nowhere to be reachable from — set " + RelayEnv + " to your sync address") } + if h.Desk != nil && h.Desk.Now == nil && h.Now != nil { + h.Desk.Now = h.Now + } wait := time.Second announced := false - // lost is an outage already told: one line per outage, not one per retry. - lost := false + // told and throttled are each said once per spell, not once per retry. + // told is set when the person has been told sync was lost and not yet + // that it came back, so the two lines always arrive as a pair. + told := false + throttled := false + tellLost := func() { + if !told { + h.say(LostSyncLine) + } + told = true + } for { registration, err := relay.Register(ctx, h.Service, h.Device.Private()) switch { case err == nil: + case errors.Is(err, relay.ErrTooMany): + if !throttled { + h.say(TooManyRegistrationsLine) + } + throttled = true case errors.Is(err, relay.ErrNameTaken) && announced: // This machine held the name a moment ago, so what holds it now is // its own dead connection, which the relay forgets when it times // out. That is the outage still going on, and is waited out. - if !lost { - h.say("lost sync — trying again") - } - lost = true + tellLost() case errors.Is(err, relay.ErrNameTaken): // RULE THREE OF THE NAME POLICY, MET FROM THIS SIDE. Something is // already connected under this machine's name, which in practice @@ -105,17 +182,11 @@ func (h *Host) Run(ctx context.Context) error { if !announced { return Unreachable(h.Service) } - if !lost { - h.say("lost sync — trying again") - } - lost = true + tellLost() case errors.Is(err, relay.ErrNoDoor) && announced: // The door was there a moment ago, so this is a relay being // redeployed, and is waited out like any other outage. - if !lost { - h.say("lost sync — trying again") - } - lost = true + tellLost() case errors.Is(err, relay.ErrNoDoor): // Trying again cannot add a door the service does not have. return NoDoor(h.Service) @@ -127,26 +198,35 @@ func (h *Host) Run(ctx context.Context) error { } if err == nil { - wait = time.Second - lost = false - if !announced { - announced = true - } - if code, codeErr := h.Desk.Offer(); codeErr == nil { - h.say(strings.TrimRight(Lines(registration.Name(), code), "\n")) - } else { - h.say(" this machine is reachable as " + registration.Name()) + announced = true + // A LOST LINE IS ANSWERED BY A BACK LINE. The banner is reprinted only + // when the name or code changed, so without this the last thing on + // the screen of a machine that recovered would still say it was lost. + if told { + h.say(BackLine(registration.Name())) + told = false } + code, codeErr := h.Desk.Offer() + // Reprints happen only when the name or pairing code actually + // changed, so a brief network blip neither spams the screen nor + // disguises a flapping carrier as a fresh start. + h.sayBanner(h.banner(registration.Name(), code, codeErr), false) + since := h.now() h.hold(ctx, registration) + // Backoff resets only after the carrier stayed up for a meaningful + // stretch; an upgrade that drops right after Register keeps + // backing off instead of hammering the relay every second. + if h.now().Sub(since) >= steadyAfter { + wait = time.Second + throttled = false + } } - select { - case <-ctx.Done(): + if err := h.wait(ctx, wait); err != nil { return nil - case <-time.After(wait): } - if wait < time.Minute { - wait *= 2 + if wait *= 2; wait > time.Minute { + wait = time.Minute } } } @@ -159,9 +239,14 @@ func (h *Host) Run(ctx context.Context) error { // fails; waiting for them is what makes "Run returned" mean that nothing of // this machine is still writing its book or speaking to its screen. func (h *Host) hold(ctx context.Context, registration *relay.Registration) { + done := make(chan struct{}) + defer close(done) go func() { - <-ctx.Done() - _ = registration.Close() + select { + case <-ctx.Done(): + _ = registration.Close() + case <-done: + } }() var workers sync.WaitGroup defer workers.Wait() @@ -182,7 +267,7 @@ func (h *Host) hold(ctx context.Context, registration *relay.Registration) { // answer takes one arriving stream through whichever of the two exchanges it // asked for. func (h *Host) answer(stream io.ReadWriteCloser) { - withDeadline(stream, h.now().Add(HandshakeWithin)) + withDeadline(stream, time.Now().Add(h.handshakeWithin())) var intent [1]byte if _, err := io.ReadFull(stream, intent[:]); err != nil { _ = stream.Close() @@ -220,7 +305,7 @@ func (h *Host) pair(stream io.ReadWriteCloser) { // one: a person who watched a code burn must not have to ask for another. h.Desk.Retire() if next, err := h.Desk.Offer(); err == nil { - h.say(strings.TrimRight(Lines(h.Device.Name(), next), "\n")) + h.sayBanner(h.banner(h.Device.Name(), next, nil), true) } } @@ -229,8 +314,8 @@ func (h *Host) pair(stream io.ReadWriteCloser) { // nobody is sitting at never admits anything. func (h *Host) approver(stream io.ReadWriteCloser) func(label, words string) error { return func(label, words string) error { - withDeadline(stream, h.now().Add(ConfirmWithin+HandshakeWithin)) - defer withDeadline(stream, h.now().Add(HandshakeWithin)) + withDeadline(stream, time.Now().Add(ConfirmWithin+h.handshakeWithin())) + defer withDeadline(stream, time.Now().Add(h.handshakeWithin())) if h.Approve == nil || !h.Approve(label, words) { return ErrRefused } diff --git a/internal/pair/tunnel.go b/internal/pair/tunnel.go index 30d1ad7258..311cd7f43c 100644 --- a/internal/pair/tunnel.go +++ b/internal/pair/tunnel.go @@ -207,9 +207,9 @@ func acceptAsMachine(conn io.ReadWriteCloser, me Device, allow func(key []byte, return &Tunnel{conn: conn, peer: deviceKey, send: second, recv: first}, deviceKey, nil } -// withDeadline puts a bound on a handshake when the thing underneath can carry -// one. A relay stream cannot — it is a queue, not a socket — so this is a -// courtesy on the transports that can and a silence on the ones that cannot. +// withDeadline puts a bound on a handshake when the stream underneath carries +// deadlines, as both TCP sockets and relay mux streams do, so an unpaired +// caller that opens a stream and says nothing cannot hold a slot forever. type deadliner interface{ SetDeadline(time.Time) error } func withDeadline(conn io.ReadWriteCloser, when time.Time) { diff --git a/internal/relay/client.go b/internal/relay/client.go index 687a6464c3..e82eb701b4 100644 --- a/internal/relay/client.go +++ b/internal/relay/client.go @@ -159,31 +159,38 @@ func upgrade(ctx context.Context, service, path string, header http.Header) (net return nil, fmt.Errorf("%q is not a relay address — it looks like https://relay.example.com", service) } // The relay address may carry a path of its own (https://host/fabric), and - // the door is beneath it, as every other client's route is. + // the door is beneath it, as every other client's route is. Refusals name + // the full base address with that path intact so a mistyped prefix is + // visible in the error. + base := (&url.URL{ + Scheme: address.Scheme, + Host: address.Host, + Path: strings.TrimRight(address.Path, "/"), + }).String() door := *address door.Path = wireauth.Endpoint(address.Path, path) door.RawPath, door.RawQuery, door.Fragment = "", "", "" - conn, err := upgradeWebSocket(ctx, &door, path, header) + conn, err := upgradeWebSocket(ctx, &door, base, path, header) if !errors.Is(err, errRawOnly) { return conn, err } - return upgradeRaw(ctx, &door, path, header) + return upgradeRaw(ctx, &door, base, path, header) } // upgradeRaw is the original carrier: an HTTP upgrade to [Protocol] on one TCP // connection. -func upgradeRaw(ctx context.Context, address *url.URL, path string, header http.Header) (net.Conn, error) { +func upgradeRaw(ctx context.Context, address *url.URL, base, path string, header http.Header) (net.Conn, error) { dialer := &net.Dialer{Timeout: dialWithin} conn, err := dialer.DialContext(ctx, "tcp", hostPort(address)) if err != nil { - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } if address.Scheme == "https" { secure := tls.Client(conn, &tls.Config{ServerName: address.Hostname()}) if err := secure.HandshakeContext(ctx); err != nil { _ = conn.Close() - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } conn = secure } @@ -201,18 +208,18 @@ func upgradeRaw(ctx context.Context, address *url.URL, path string, header http. _ = conn.SetDeadline(time.Now().Add(HandshakeWithin)) if err := request.Write(conn); err != nil { _ = conn.Close() - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } reader := bufio.NewReader(conn) response, err := http.ReadResponse(reader, request) if err != nil { _ = conn.Close() - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } if response.StatusCode != http.StatusSwitchingProtocols { said := strings.TrimSpace(readBounded(response.Body)) _ = conn.Close() - return nil, refusal(address.Host, path, response.StatusCode, said) + return nil, refusal(base, path, response.StatusCode, said) } _ = conn.SetDeadline(time.Time{}) // A 101 has no body, so there should be nothing left in the reader — but @@ -225,26 +232,45 @@ func upgradeRaw(ctx context.Context, address *url.URL, path string, header http. } // refusal turns a relay's non-101 answer into the fact a caller above can act -// on. ONLY A RELAY'S OWN 4xx SENTENCE IS PASSED ON: a 5xx is the relay or the -// proxy in front of it being down, and that proxy's body ("error code: 522") -// is not a sentence anybody should be shown. -func refusal(host, path string, status int, said string) error { +// on. ONLY A RELAY'S OWN SENTENCE IS PASSED ON: an upstream proxy, WAF or +// web server answering with HTML or its own 4xx/5xx body becomes a one-sentence +// fact naming the relay address with its base path, never the raw body. +func refusal(base, path string, status int, said string) error { switch { - case status == http.StatusNotFound && (path == EnginePath || strings.HasPrefix(said, "{")): - // The relay answers, but not at a machine door: an engine path is - // never "no machine", and a JSON body is the hosted wire's not_found. - return fmt.Errorf("%w: %s", ErrNoDoor, host) - case status == http.StatusNotFound: + case status == http.StatusNotFound && path != EnginePath && said == "no machine is connected under that name": return ErrNoMachine + case status == http.StatusNotFound && (strings.HasPrefix(said, "{") || said == "not a relay door"): + // The relay answers, but not at a machine door: the hosted wire's JSON + // not_found, or a self-hosted relay hit outside its two paths. + return fmt.Errorf("%w: %s", ErrNoDoor, base) case status == http.StatusTooManyRequests: return ErrTooMany + case status == http.StatusServiceUnavailable && strings.Contains(said, "connections open, which is as many as the relay carries at once"): + // Older self-hosted relays answered 503 when a machine reached its + // stream ceiling; that machine is busy, not unreachable. + return ErrTooMany case status >= 500: - return fmt.Errorf("%w: %s answered %d", ErrUnreachable, host, status) - case said == "": - return fmt.Errorf("the relay at %s refused this connection (%d) — check the address it was given", host, status) - default: + return fmt.Errorf("%w: %s answered %d", ErrUnreachable, base, status) + case relaySentence(said): return errors.New(said) + default: + return fmt.Errorf("the relay at %s refused this connection (%d) — check the address it was given", base, status) + } +} + +// relaySentence reports whether said is one of the plain-text validation +// sentences our own relay doors write before upgrading. Anything else — HTML +// from a proxy, a captive portal, or an upstream web server — is withheld. +func relaySentence(said string) bool { + switch said { + case "that is not a machine name", + "that name does not belong to that key", + "that key is not readable", + "that key is not a usable machine key": + return true } + return strings.HasPrefix(said, "a machine key is 32 bytes and that one is ") || + strings.HasPrefix(said, "this door speaks ") } // joinedConn is a connection whose first bytes were already read into a buffer. diff --git a/internal/relay/mux.go b/internal/relay/mux.go index 1268676de5..daba8b3d0f 100644 --- a/internal/relay/mux.go +++ b/internal/relay/mux.go @@ -24,6 +24,7 @@ import ( "errors" "fmt" "io" + "os" "sync" "time" ) @@ -92,7 +93,9 @@ func readFrame(r io.Reader) (op byte, stream uint64, payload []byte, err error) const streamDepth = 8 // muxStream is one surface's byte stream, presented as the [io.ReadWriteCloser] -// everything above this package deals in. +// everything above this package deals in. It carries real read and write +// deadlines so a handshake above this package can evict a silent caller instead +// of letting it hold one of a machine's [MaxStreams] slots forever. type muxStream struct { id uint64 carrier *carrier @@ -100,6 +103,9 @@ type muxStream struct { incoming chan []byte rest []byte + readDeadline streamDeadline + writeDeadline streamDeadline + // closed is THIS END letting go, and incoming being closed is the FAR END // letting go. THE TWO ARE DELIBERATELY NOT THE SAME SIGNAL, and conflating // them was a real bug: a reader selecting on both picks between two ready @@ -115,14 +121,40 @@ type muxStream struct { func newMuxStream(id uint64, c *carrier) *muxStream { return &muxStream{ - id: id, - carrier: c, - incoming: make(chan []byte, streamDepth), - closed: make(chan struct{}), + id: id, + carrier: c, + incoming: make(chan []byte, streamDepth), + readDeadline: newStreamDeadline(), + writeDeadline: newStreamDeadline(), + closed: make(chan struct{}), } } +func (s *muxStream) SetDeadline(t time.Time) error { + s.readDeadline.set(t) + s.writeDeadline.set(t) + return nil +} + +func (s *muxStream) SetReadDeadline(t time.Time) error { + s.readDeadline.set(t) + return nil +} + +func (s *muxStream) SetWriteDeadline(t time.Time) error { + s.writeDeadline.set(t) + return nil +} + +func (s *muxStream) stopDeadlines() { + s.readDeadline.set(time.Time{}) + s.writeDeadline.set(time.Time{}) +} + func (s *muxStream) Read(b []byte) (int, error) { + if isClosedChan(s.readDeadline.wait()) { + return 0, os.ErrDeadlineExceeded + } for len(s.rest) == 0 { select { case chunk, ok := <-s.incoming: @@ -135,6 +167,8 @@ func (s *muxStream) Read(b []byte) (int, error) { s.rest = chunk case <-s.closed: return 0, io.EOF + case <-s.readDeadline.wait(): + return 0, os.ErrDeadlineExceeded } } n := copy(b, s.rest) @@ -148,10 +182,20 @@ func (s *muxStream) Write(b []byte) (int, error) { return 0, io.ErrClosedPipe default: } + if isClosedChan(s.writeDeadline.wait()) { + return 0, os.ErrDeadlineExceeded + } // A write is cut into carrier-sized frames, because a caller above knows // nothing about this framing and must not have to. written := 0 for len(b) > 0 { + select { + case <-s.closed: + return written, io.ErrClosedPipe + case <-s.writeDeadline.wait(): + return written, os.ErrDeadlineExceeded + default: + } chunk := b if len(chunk) > MaxPayload { chunk = chunk[:MaxPayload] @@ -168,12 +212,72 @@ func (s *muxStream) Write(b []byte) (int, error) { func (s *muxStream) Close() error { s.closeOnce.Do(func() { close(s.closed) + s.stopDeadlines() _ = s.carrier.send(opClose, s.id, nil) s.carrier.forget(s.id) }) return nil } +// streamDeadline signals when a stream's read or write deadline has passed, in +// the shape of net.Pipe's pipeDeadline so blocked callers wake at once and +// cleared deadlines hold no timer. +type streamDeadline struct { + mu sync.Mutex + timer *time.Timer + cancel chan struct{} +} + +func newStreamDeadline() streamDeadline { + return streamDeadline{cancel: make(chan struct{})} +} + +func (d *streamDeadline) set(t time.Time) { + d.mu.Lock() + defer d.mu.Unlock() + + if d.timer != nil && !d.timer.Stop() { + <-d.cancel + } + d.timer = nil + + closed := isClosedChan(d.cancel) + if t.IsZero() { + if closed { + d.cancel = make(chan struct{}) + } + return + } + if dur := time.Until(t); dur > 0 { + if closed { + d.cancel = make(chan struct{}) + } + cancel := d.cancel + d.timer = time.AfterFunc(dur, func() { + close(cancel) + }) + return + } + if !closed { + close(d.cancel) + } +} + +func (d *streamDeadline) wait() chan struct{} { + d.mu.Lock() + defer d.mu.Unlock() + return d.cancel +} + +func isClosedChan(c <-chan struct{}) bool { + select { + case <-c: + return true + default: + return false + } +} + // deliver hands a frame to this stream's reader, or drops it on the floor once // the stream is closed — which is not a loss, because a closed stream has no // reader and the peer will be told. @@ -203,6 +307,7 @@ type carrier struct { mu sync.Mutex streams map[uint64]*muxStream + probes map[chan struct{}]struct{} nextID uint64 gone bool @@ -328,12 +433,53 @@ func (c *carrier) pump() { case opPing: _ = c.send(opPong, id, nil) case opPong: - // Nothing to do: the frame arriving is the whole of the answer, - // and [IdleAfter] is measured against ANY byte on the carrier. + c.mu.Lock() + for ch := range c.probes { + close(ch) + delete(c.probes, ch) + } + c.mu.Unlock() } } } +// probe tests whether an older carrier under the same key is still answering, +// waiting at most within for its pong. A live machine answers in one round +// trip; one whose network vanished says nothing and lets its successor in. +func (c *carrier) probe(within time.Duration) bool { + ch := make(chan struct{}) + c.mu.Lock() + if c.gone { + c.mu.Unlock() + return false + } + if c.probes == nil { + c.probes = map[chan struct{}]struct{}{} + } + c.probes[ch] = struct{}{} + c.mu.Unlock() + + defer func() { + c.mu.Lock() + delete(c.probes, ch) + c.mu.Unlock() + }() + + if err := c.send(opPing, 0, nil); err != nil { + return false + } + timer := time.NewTimer(within) + defer timer.Stop() + select { + case <-ch: + return true + case <-c.done: + return false + case <-timer.C: + return false + } +} + // shutdown ends the carrier and everything on it, once. func (c *carrier) shutdown(cause error) { c.closeOnce.Do(func() { @@ -355,7 +501,10 @@ func (c *carrier) shutdown(cause error) { // without the race. close(c.done) for _, stream := range streams { - stream.closeOnce.Do(func() { close(stream.closed) }) + stream.closeOnce.Do(func() { + close(stream.closed) + stream.stopDeadlines() + }) } _ = c.conn.Close() }) diff --git a/internal/relay/relay.go b/internal/relay/relay.go index e41ae3ced7..902b7f1511 100644 --- a/internal/relay/relay.go +++ b/internal/relay/relay.go @@ -70,18 +70,22 @@ const ( // The tuning, all in one place so that a relay operator changes a number here // and every sentence in this package that quotes it changes with it. ONE SOURCE -// OF TRUTH: nothing below re-states a limit as a literal. +// OF TRUTH: nothing below re-states a limit as a literal, and the hosted Worker +// mirrors these constants in relay/hosted/src/engine/carrier.js (and its +// front-door rate-limit bindings in relay/hosted/wrangler.toml). const ( - // MaxPayload is the largest body one framed message may carry. It is a - // little over 64 KiB because the tunnel above frames its ciphertext at - // Noise's own 65535-byte ceiling and adds a small header; a limit under - // that would cut a legal message in half. + // MaxPayload is the largest body one framed message may carry (MAX_PAYLOAD + // in relay/hosted/src/engine/carrier.js). It is a little over 64 KiB + // because the tunnel above frames its ciphertext at Noise's own 65535-byte + // ceiling and adds a small header; a limit under that would cut a legal + // message in half. MaxPayload = 1 << 17 - // MaxStreams is how many surfaces may be attached to one machine at once. - // It is a limit on the relay's memory and on nothing else — the engine - // machine has its own opinion about how many surfaces a session will have, - // and this is only the number the pipe will carry. + // MaxStreams is how many surfaces may be attached to one machine at once + // (MAX_STREAMS in relay/hosted/src/engine/carrier.js). It is a limit on + // the relay's memory and on nothing else — the engine machine has its own + // opinion about how many surfaces a session will have, and this is only + // the number the pipe will carry. MaxStreams = 16 // PingEvery is how often the relay pokes an idle carrier, and IdleAfter is @@ -91,15 +95,25 @@ const ( PingEvery = 45 * time.Second IdleAfter = 2 * PingEvery - // HandshakeWithin bounds the registration exchange. A connection that has - // claimed a name but not yet proved it holds the key is a connection that - // could hold a name for free, so it is given one round trip's worth of time - // and no more. + // HandshakeWithin bounds the registration exchange (HANDSHAKE_MS in + // relay/hosted/src/engine/carrier.js). A connection that has claimed a name + // but not yet proved it holds the key is a connection that could hold a + // name for free, so it is given one round trip's worth of time and no more. HandshakeWithin = 20 * time.Second - // DialsPerMinute and RegistrationsPerMinute are the per-address ceilings. - // They are deliberately generous for a person and stingy for a script: a - // person dials once and stays, and reconnects a handful of times a day. + // ProbeWithin is how long a registration waits for an older carrier under + // the same key to answer a ping before replacing it (PROBE_MS in + // relay/hosted/src/engine/carrier.js). A machine whose network dropped can + // come back before IdleAfter runs out; a live carrier answers in one round + // trip and keeps its name. + ProbeWithin = 5 * time.Second + + // DialsPerMinute and RegistrationsPerMinute are the per-address ceilings + // (DIALS_PER_MINUTE and REGISTRATIONS_PER_MINUTE in + // relay/hosted/src/engine/carrier.js, and MACHINE_DIAL_LIMIT / + // MACHINE_REGISTER_LIMIT in relay/hosted/wrangler.toml). They are + // deliberately generous for a person and stingy for a script: a person + // dials once and stays, and reconnects a handful of times a day. DialsPerMinute = 30 RegistrationsPerMinute = 10 ) diff --git a/internal/relay/server.go b/internal/relay/server.go index fb3fb185b9..002a727604 100644 --- a/internal/relay/server.go +++ b/internal/relay/server.go @@ -19,6 +19,7 @@ package relay import ( "context" + "crypto/subtle" "errors" "io" "net" @@ -57,10 +58,15 @@ type Server struct { // no payload in this package to hand it. Note func(name, what string) - mu sync.Mutex - live map[string]*registration - dials map[string]*bucket - regs map[string]*bucket + // probeWithin overrides [ProbeWithin] in tests so a silent older carrier + // can be replaced without waiting five real seconds. + probeWithin time.Duration + + mu sync.Mutex + live map[string]*registration + claiming map[string]bool + dials map[string]*bucket + regs map[string]*bucket } // registration is one machine's held carrier. @@ -81,6 +87,13 @@ func (s *Server) now() time.Time { return time.Now() } +func (s *Server) probeTimeout() time.Duration { + if s.probeWithin > 0 { + return s.probeWithin + } + return ProbeWithin +} + func (s *Server) note(name, what string) { if s.Note != nil { s.Note(name, what) @@ -145,7 +158,7 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { return } - conn, err := takeOver(w, r) + conn, err := takeOver(w, r, messageLimit) if err != nil { return } @@ -153,7 +166,7 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { // RULE TWO: possession is proved. A machine that cannot answer the // challenge never reaches the table. - _ = conn.SetDeadline(s.now().Add(HandshakeWithin)) + _ = conn.SetDeadline(time.Now().Add(HandshakeWithin)) if err := challengeMachine(conn, name, key); err != nil { _ = writeVerdict(conn, err.Error()) s.note(name, "registration refused") @@ -164,18 +177,36 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { carrier := newCarrier(conn, false) reg := ®istration{name: name, key: key, carrier: carrier, since: s.now()} - // RULE THREE: a live registration is never taken over. Whoever is - // connected under this name stays connected under it. + // RULE THREE: a live registration is never taken over. A different key is + // refused outright even if the current socket is stalled; the SAME key may + // probe the older socket and replace it only when it has gone silent. s.mu.Lock() if s.live == nil { s.live = map[string]*registration{} } - if _, taken := s.live[name]; taken { + if s.claiming == nil { + s.claiming = map[string]bool{} + } + older := s.live[name] + if s.claiming[name] || (older != nil && subtle.ConstantTimeCompare(older.key, key) != 1) { s.mu.Unlock() _ = writeVerdict(conn, "that machine is already connected to the relay under this name") s.note(name, "registration refused, name already connected") return } + if older != nil { + s.claiming[name] = true + s.mu.Unlock() + alive := older.carrier.probe(s.probeTimeout()) + s.mu.Lock() + delete(s.claiming, name) + if alive { + s.mu.Unlock() + _ = writeVerdict(conn, "that machine is already connected to the relay under this name") + s.note(name, "registration refused, name already connected") + return + } + } s.live[name] = reg s.mu.Unlock() @@ -192,6 +223,9 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { if err := writeVerdict(conn, ""); err != nil { return } + if older != nil { + older.carrier.shutdown(nil) + } s.note(name, "connected") // THE CARRIER IS ONLY ALIVE WHILE IT IS SAYING SOMETHING. The relay pings a @@ -199,8 +233,8 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { // pushed out by an ARRIVING frame, so a connection that has stopped // answering — a NAT that forgot the mapping, a machine that was unplugged — // ends at [IdleAfter] instead of being held for ever. - carrier.onFrame = func() { _ = conn.SetReadDeadline(s.now().Add(IdleAfter)) } - _ = conn.SetReadDeadline(s.now().Add(IdleAfter)) + carrier.onFrame = func() { _ = conn.SetReadDeadline(time.Now().Add(IdleAfter)) } + _ = conn.SetReadDeadline(time.Now().Add(IdleAfter)) go carrier.keepAlive() carrier.pump() } @@ -241,7 +275,7 @@ func (s *Server) serveDial(w http.ResponseWriter, r *http.Request, name string) } defer func() { _ = stream.Close() }() - conn, err := takeOver(w, r) + conn, err := takeOver(w, r, MaxPayload) if err != nil { return } @@ -300,7 +334,7 @@ func wantsWebSocket(r *http.Request) bool { // takeOver answers the upgrade and hands back the connection the carrier // reads, whichever of the two the client asked for. On failure the answer has // already been written. -func takeOver(w http.ResponseWriter, r *http.Request) (net.Conn, error) { +func takeOver(w http.ResponseWriter, r *http.Request, maxWrite int) (net.Conn, error) { if wantsWebSocket(r) { socket, err := websocket.Accept(w, r, &websocket.AcceptOptions{ Subprotocols: []string{Protocol}, @@ -309,8 +343,11 @@ func takeOver(w http.ResponseWriter, r *http.Request) (net.Conn, error) { if err != nil { return nil, err } + // websocket.NetConn clears the read limit on entry, so SetReadLimit has + // to come after it or an oversized frame is buffered without bound. + nc := websocket.NetConn(context.Background(), socket, websocket.MessageBinary) socket.SetReadLimit(messageLimit) - return websocket.NetConn(context.Background(), socket, websocket.MessageBinary), nil + return &wsConn{Conn: nc, socket: socket, maxWrite: maxWrite}, nil } conn, err := hijack(w) if err != nil { diff --git a/internal/relay/server_test.go b/internal/relay/server_test.go index 042e63cdd6..061133a711 100644 --- a/internal/relay/server_test.go +++ b/internal/relay/server_test.go @@ -1,13 +1,17 @@ package relay import ( + "bytes" "context" "crypto/ecdh" "crypto/rand" + "encoding/binary" "errors" "fmt" "io" + "net" "net/http/httptest" + "os" "strings" "sync" "testing" @@ -161,23 +165,175 @@ func TestAKeySomebodyOnlyCopiedCannotRegister(t *testing.T) { } } -// RULE THREE: a live registration is never taken over. -func TestALiveNameIsNotHandedToASecondMachine(t *testing.T) { +// RULE THREE: a live registration is never taken over, and only the SAME key +// may replace a registration whose socket went silent. +func TestOnlyTheSameKeyMayReplaceASilentRegistration(t *testing.T) { + owner, squatter := collidingKeys(t) + name := NameFor(owner.PublicKey().Bytes()) + if got := NameFor(squatter.PublicKey().Bytes()); got != name { + t.Fatalf("colliding keys derived %q and %q", name, got) + } + + // registerSilent completes the registration proof by hand and then never + // runs carrier.pump(), so the socket holds the name without answering opPing. + registerSilent := func(t *testing.T, address string, key *ecdh.PrivateKey) net.Conn { + t.Helper() + conn, err := upgrade(context.Background(), address, EnginePath, header(name, EncodeKey(key.PublicKey().Bytes()))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + if err := proveToRelay(conn, name, key); err != nil { + t.Fatal(err) + } + return conn + } + + t.Run("different key with silent old is refused", func(t *testing.T) { + service, address := live(t) + service.probeWithin = 30 * time.Millisecond + _ = registerSilent(t, address, owner) + + impostor, err := Register(context.Background(), address, squatter) + if err == nil { + _ = impostor.Close() + t.Fatal("a colliding key took over a silent registration") + } + if !errors.Is(err, ErrNameTaken) { + t.Fatalf("colliding key got %v; want ErrNameTaken", err) + } + if ledgers := service.Ledgers(); len(ledgers) != 1 || ledgers[0].Name != name { + t.Fatalf("silent owner lost its registration: %+v", ledgers) + } + }) + + t.Run("same key with live old is refused", func(t *testing.T) { + service, address := live(t) + service.probeWithin = 30 * time.Millisecond + first, err := Register(context.Background(), address, owner) + if err != nil { + t.Fatal(err) + } + defer func() { _ = first.Close() }() + + second, err := Register(context.Background(), address, owner) + if err == nil { + _ = second.Close() + t.Fatal("a second registration replaced a live machine") + } + if !errors.Is(err, ErrNameTaken) { + t.Fatalf("live replacement got %v; want ErrNameTaken", err) + } + }) + + t.Run("same key with silent old replaces it", func(t *testing.T) { + service, address := live(t) + service.probeWithin = 30 * time.Millisecond + _ = registerSilent(t, address, owner) + + replacement, err := Register(context.Background(), address, owner) + if err != nil { + t.Fatalf("same key could not replace its own silent socket: %v", err) + } + defer func() { _ = replacement.Close() }() + + surface, err := Dial(context.Background(), address, name) + if err != nil { + t.Fatal(err) + } + defer func() { _ = surface.Close() }() + engine, err := replacement.Accept() + if err != nil { + t.Fatal(err) + } + defer func() { _ = engine.Close() }() + if _, err := surface.Write([]byte("ok")); err != nil { + t.Fatal(err) + } + if got := readN(t, engine, 2); got != "ok" { + t.Fatalf("replacement read %q", got) + } + }) +} + +// muxStream carries real read and write deadlines so handshakes above this +// package can time out silent peers and clear the deadline afterward. +func TestMuxStreamEnforcesAndClearsDeadlines(t *testing.T) { _, address := live(t) - key := aKey(t) - first, err := Register(context.Background(), address, key) + registration, err := Register(context.Background(), address, aKey(t)) if err != nil { t.Fatal(err) } - defer func() { _ = first.Close() }() + defer func() { _ = registration.Close() }() - second, err := Register(context.Background(), address, key) - if err == nil { - _ = second.Close() - t.Fatal("a second machine took a name that was already connected") + surface, err := Dial(context.Background(), address, registration.Name()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = surface.Close() }() + + engine, err := registration.Accept() + if err != nil { + t.Fatal(err) + } + defer func() { _ = engine.Close() }() + + deadliner, ok := engine.(interface { + SetDeadline(time.Time) error + SetReadDeadline(time.Time) error + SetWriteDeadline(time.Time) error + }) + if !ok { + t.Fatalf("%T does not implement stream deadlines", engine) + } + + if err := deadliner.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil { + t.Fatal(err) + } + var one [1]byte + if _, err := engine.Read(one[:]); !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("silent read returned %v; want os.ErrDeadlineExceeded", err) + } + + if err := deadliner.SetWriteDeadline(time.Now().Add(-time.Millisecond)); err != nil { + t.Fatal(err) + } + if _, err := engine.Write([]byte("x")); !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("expired write returned %v; want os.ErrDeadlineExceeded", err) + } + + if err := deadliner.SetDeadline(time.Time{}); err != nil { + t.Fatal(err) + } + if _, err := surface.Write([]byte("hi")); err != nil { + t.Fatal(err) } - if !errors.Is(err, ErrNameTaken) { - t.Fatalf("the refusal was %v, which a caller cannot tell apart from any other", err) + if got := readN(t, engine, 2); got != "hi" { + t.Fatalf("after clearing deadline, read %q", got) + } + if _, err := engine.Write([]byte("ok")); err != nil { + t.Fatal(err) + } + if got := readN(t, surface, 2); got != "ok" { + t.Fatalf("after clearing deadline, surface read %q", got) + } +} + +func collidingKeys(t *testing.T) (*ecdh.PrivateKey, *ecdh.PrivateKey) { + t.Helper() + seen := map[string]*ecdh.PrivateKey{} + var seed [32]byte + for i := uint64(1); ; i++ { + binary.LittleEndian.PutUint64(seed[:8], i) + key, err := ecdh.X25519().NewPrivateKey(seed[:]) + if err != nil { + t.Fatal(err) + } + name := NameFor(key.PublicKey().Bytes()) + if first, ok := seen[name]; ok && !bytes.Equal(first.PublicKey().Bytes(), key.PublicKey().Bytes()) { + return first, key + } + seen[name] = key } } diff --git a/internal/relay/transport.go b/internal/relay/transport.go index d0f69d5280..bf54ca8e74 100644 --- a/internal/relay/transport.go +++ b/internal/relay/transport.go @@ -18,6 +18,7 @@ import ( "time" "github.com/coder/websocket" + "golang.org/x/net/http/httpproxy" ) // errRawOnly is a relay that answered a WebSocket with 426: one from before @@ -31,7 +32,32 @@ const messageLimit = frameHeader + MaxPayload // the protocol-level ping a carrier uses to keep a quiet connection proven. type wsConn struct { net.Conn - socket *websocket.Conn + socket *websocket.Conn + maxWrite int +} + +// Write cuts surface writes larger than [MaxPayload] into relay-sized WebSocket +// messages so a caller above [Dial] cannot trip a front end's per-message cap +// or the peer's [messageLimit]. Carrier frames on [EnginePath] set maxWrite to +// [messageLimit] so each frame stays one message. +func (w *wsConn) Write(b []byte) (int, error) { + if w.maxWrite <= 0 || len(b) <= w.maxWrite { + return w.Conn.Write(b) + } + written := 0 + for len(b) > 0 { + chunk := b + if len(chunk) > w.maxWrite { + chunk = chunk[:w.maxWrite] + } + n, err := w.Conn.Write(chunk) + written += n + if err != nil { + return written, err + } + b = b[n:] + } + return written, nil } // ping is answered by the far end's WebSocket layer, never by the relay's own @@ -43,9 +69,31 @@ type pinger interface { ping(ctx context.Context) error } -func upgradeWebSocket(ctx context.Context, address *url.URL, path string, header http.Header) (net.Conn, error) { +// proxyFromEnvironment resolves HTTP_PROXY, HTTPS_PROXY and NO_PROXY with the +// same rules [http.ProxyFromEnvironment] uses, reading the current environment +// so process-lifetime changes and tests are honoured. +func proxyFromEnvironment(req *http.Request) (*url.URL, error) { + return httpproxy.FromEnvironment().ProxyFunc()(req.URL) +} + +func upgradeWebSocket(ctx context.Context, address *url.URL, base, path string, header http.Header) (net.Conn, error) { dialer := &net.Dialer{Timeout: dialWithin} - client := &http.Client{Transport: &http.Transport{DialContext: dialer.DialContext, ForceAttemptHTTP2: false}} + transport := &http.Transport{ + Proxy: proxyFromEnvironment, + DialContext: dialer.DialContext, + ForceAttemptHTTP2: false, + DisableKeepAlives: true, + } + upgraded := false + defer func() { + // Any non-101 response (including every 426 fallback to an older + // self-hosted relay) must release its socket and transport goroutines + // before returning rather than leaving idle state behind. + if !upgraded { + transport.CloseIdleConnections() + } + }() + client := &http.Client{Transport: transport} handshake, cancel := context.WithTimeout(ctx, HandshakeWithin) defer cancel() socket, response, err := websocket.Dial(handshake, address.String(), &websocket.DialOptions{ @@ -55,28 +103,44 @@ func upgradeWebSocket(ctx context.Context, address *url.URL, path string, header CompressionMode: websocket.CompressionDisabled, // the payloads are ciphertext }) if err != nil { + // Always drain and close a non-101 body first — including on 426 + // fallbacks — so net/http closes the underlying TCP socket instead of + // holding an active response open. + said := "" + if response != nil && response.Body != nil { + said = strings.TrimSpace(readBounded(response.Body)) + } if ctx.Err() != nil { return nil, ctx.Err() } if response == nil { - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } if response.StatusCode == http.StatusUpgradeRequired { return nil, errRawOnly } - said := "" - if response.Body != nil { - said = strings.TrimSpace(readBounded(response.Body)) - } if response.StatusCode == http.StatusSwitchingProtocols { - return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + return nil, fmt.Errorf("%w: %s", ErrUnreachable, base) } - return nil, refusal(address.Host, path, response.StatusCode, said) + return nil, refusal(base, path, response.StatusCode, said) } - socket.SetReadLimit(messageLimit) + // coder/websocket permits a server that omitted Sec-WebSocket-Protocol; a + // generic WebSocket endpoint is not a relay door and must fail closed here. + if socket.Subprotocol() != Protocol { + _ = socket.CloseNow() + return nil, fmt.Errorf("%w: %s", ErrNoDoor, base) + } + upgraded = true // The connection outlives the handshake's context: it ends when the - // carrier closes it, not when the dial's caller returns. - return &wsConn{Conn: websocket.NetConn(context.Background(), socket, websocket.MessageBinary), socket: socket}, nil + // carrier closes it, not when the dial's caller returns. SetReadLimit must + // follow websocket.NetConn because NetConn resets the limit to -1 on entry. + nc := websocket.NetConn(context.Background(), socket, websocket.MessageBinary) + socket.SetReadLimit(messageLimit) + maxWrite := MaxPayload + if path == EnginePath { + maxWrite = messageLimit + } + return &wsConn{Conn: nc, socket: socket, maxWrite: maxWrite}, nil } // proveAlive pings a carrier from this end every [PingEvery] and counts an diff --git a/internal/relay/transport_test.go b/internal/relay/transport_test.go index 02a87c4aa2..18519bd957 100644 --- a/internal/relay/transport_test.go +++ b/internal/relay/transport_test.go @@ -1,12 +1,22 @@ package relay import ( + "bytes" "context" + "crypto/rand" "errors" + "io" + "net" "net/http" "net/http/httptest" + "runtime" "strings" + "sync" + "sync/atomic" "testing" + "time" + + "github.com/coder/websocket" ) // mounted serves a relay beneath base, the way the hosted relay lives at @@ -150,3 +160,311 @@ func fullMachineIsBusy(t *testing.T, address string) { t.Fatalf("a dial past %d streams gave %v; want ErrTooMany", MaxStreams, err) } } + +// Contract item 6 (part 1): WebSocket upgrades honour HTTP_PROXY and NO_PROXY. +func TestWebSocketUpgradeHonoursProxyEnvironment(t *testing.T) { + upstream := httptest.NewServer(http.StripPrefix("/fabric", wsOnly(&Server{}))) + t.Cleanup(upstream.Close) + t.Cleanup(upstream.CloseClientConnections) + upstreamHost := strings.TrimPrefix(upstream.URL, "http://") + + var proxied atomic.Int32 + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + proxied.Add(1) + hijacker, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "no hijack", http.StatusInternalServerError) + return + } + backend, err := net.Dial("tcp", upstreamHost) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + clientConn, buffered, err := hijacker.Hijack() + if err != nil { + _ = backend.Close() + return + } + out := r.Clone(r.Context()) + out.RequestURI = "" + out.URL.Scheme = "" + out.URL.Host = "" + if err := out.Write(backend); err != nil { + _ = backend.Close() + _ = clientConn.Close() + return + } + go func() { + if buffered.Reader.Buffered() > 0 { + _, _ = io.CopyN(backend, buffered, int64(buffered.Reader.Buffered())) + } + _, _ = io.Copy(backend, clientConn) + _ = backend.Close() + }() + go func() { + _, _ = io.Copy(clientConn, backend) + _ = clientConn.Close() + }() + })) + t.Cleanup(proxy.Close) + t.Cleanup(proxy.CloseClientConnections) + + t.Setenv("HTTP_PROXY", proxy.URL) + t.Setenv("NO_PROXY", "") + // Non-loopback host name so httpproxy does not apply its loopback bypass. + reachAndEcho(t, "http://relay.example.test/fabric") + if got := proxied.Load(); got < 2 { + t.Fatalf("proxy saw %d requests; want both Register and Dial routed through HTTP_PROXY", got) + } + + // NO_PROXY bypasses the proxy for the named host. + before := proxied.Load() + t.Setenv("NO_PROXY", "unproxied.invalid") + previous := dialWithin + dialWithin = 50 * time.Millisecond + t.Cleanup(func() { dialWithin = previous }) + _, _ = Dial(context.Background(), "http://unproxied.invalid/fabric", "otter-lamp-42") + if proxied.Load() != before { + t.Fatalf("NO_PROXY host still hit HTTP_PROXY (%d -> %d)", before, proxied.Load()) + } +} + +// trackedListener counts every accepted TCP socket — hijacked or plain HTTP — +// until its Close method runs, because http.Server.ConnState stops reporting +// once a connection transitions to StateHijacked. +type trackedListener struct { + net.Listener + open atomic.Int32 +} + +type trackedServerConn struct { + net.Conn + once sync.Once + open *atomic.Int32 +} + +func (l *trackedListener) Accept() (net.Conn, error) { + c, err := l.Listener.Accept() + if err != nil { + return nil, err + } + l.open.Add(1) + return &trackedServerConn{Conn: c, open: &l.open}, nil +} + +func (c *trackedServerConn) Close() error { + err := c.Conn.Close() + c.once.Do(func() { c.open.Add(-1) }) + return err +} + +// Contract item 6 (part 2): 20 attempts over the 426 fallback leave no open +// connections on the server and leak no transport goroutines on the client. +func TestFallback426DoesNotGrowOpenConnectionsOverTwentyAttempts(t *testing.T) { + service := &Server{} + server := httptest.NewUnstartedServer(rawOnly(service)) + tracked := &trackedListener{Listener: server.Listener} + server.Listener = tracked + server.Start() + t.Cleanup(server.Close) + t.Cleanup(server.CloseClientConnections) + + countOpen := func() int { + return int(tracked.open.Load()) + } + + registration, err := Register(context.Background(), server.URL, aKey(t)) + if err != nil { + t.Fatal(err) + } + defer func() { _ = registration.Close() }() + + // Warm one dial so baseline goroutines and the one held registration settle. + warm, err := Dial(context.Background(), server.URL, registration.Name()) + if err != nil { + t.Fatal(err) + } + accepted, err := registration.Accept() + if err != nil { + t.Fatal(err) + } + _ = warm.Close() + _ = accepted.Close() + baselineGoroutines := runtime.NumGoroutine() + + const attempts = 20 + for i := 0; i < attempts; i++ { + surface, err := Dial(context.Background(), server.URL, registration.Name()) + if err != nil { + t.Fatalf("attempt %d: %v", i+1, err) + } + engine, err := registration.Accept() + if err != nil { + t.Fatalf("attempt %d accept: %v", i+1, err) + } + _ = surface.Close() + _ = engine.Close() + } + + // Only the single held registration connection may remain open once the 20 + // 426-fallback dials have closed. + deadline := time.Now().Add(2 * time.Second) + for countOpen() > 1 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := countOpen(); got > 1 { + t.Fatalf("after %d 426 fallbacks, %d server connections remained open; want at most 1 (the held registration)", attempts, got) + } + if grew := runtime.NumGoroutine() - baselineGoroutines; grew > 6 { + t.Fatalf("goroutines grew by %d across %d 426 fallbacks (baseline %d)", grew, attempts, baselineGoroutines) + } +} + +// Contract item 7 (Findings 7 & 8): upstream HTML/non-JSON bodies never leak +// into errors, wrong-prefix 404s are not misreported as ErrNoMachine, the base +// path is preserved in refusal messages, and an older relay's 503 stream +// ceiling maps to ErrTooMany. +func TestUpstreamHTMLAndLegacy503BecomeCleanFactsWithBasePath(t *testing.T) { + t.Run("upstream 403 HTML becomes one sentence with base path and no HTML", func(t *testing.T) { + waf := mounted(t, "/wrong-prefix", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.WriteHeader(http.StatusForbidden) + _, _ = w.Write([]byte("

403 Forbidden

\nAccess denied")) + })) + _, err := Dial(context.Background(), waf, "otter-lamp-42") + if err == nil { + t.Fatal("403 succeeded") + } + said := err.Error() + if strings.Contains(said, "404 Not Found")) + })) + _, err := Dial(context.Background(), nginx, "otter-lamp-42") + if err == nil || errors.Is(err, ErrNoMachine) { + t.Fatalf("HTML 404 at bad base path returned %v; must not report ErrNoMachine", err) + } + if strings.Contains(err.Error(), " pathname === ENGINE_PATH || pathname.startsWith(DIAL_PREFIX); +/** + * fallbackBuckets bounds per-caller cost in Worker memory if a custom deployment forgot the [[ratelimits]] bindings in + * wrangler.toml. When the bindings are present, Cloudflare enforces the rate across isolates at the edge. + */ +const fallbackBuckets = new Map(); + +async function allowCaller(request, env, register) { + const ip = ipOf(request, env); + const limiter = register ? env.MACHINE_REGISTER_LIMIT : env.MACHINE_DIAL_LIMIT; + if (limiter && typeof limiter.limit === 'function') { + const { success } = await limiter.limit({ key: ip }); + return Boolean(success); + } + const perMinute = register ? REGISTRATIONS_PER_MINUTE : DIALS_PER_MINUTE; + const minute = Math.floor(Date.now() / 60000); + for (const [key, bucket] of fallbackBuckets) if (bucket.minute !== minute) fallbackBuckets.delete(key); + const key = (register ? 'register ' : 'dial ') + ip; + const bucket = fallbackBuckets.get(key) ?? { minute, count: 0 }; + bucket.count += 1; + fallbackBuckets.set(key, bucket); + return bucket.count <= perMinute; +} + /** * serveMachineDoor routes a registration or a dial. Only a WebSocket offering the carrier's protocol is taken: a - * Worker cannot answer codeaf's raw upgrade, and the 426 is what tells a client to use the WebSocket. + * Worker cannot answer codeaf's raw upgrade, and the 426 is what tells a client to use the WebSocket. Rate limits and + * name validity are enforced here before env.MACHINE is touched so one caller cannot enumerate names or instantiate + * arbitrary Durable Objects. */ export async function serveMachineDoor(request, env) { const offered = (request.headers.get('sec-websocket-protocol') ?? '').split(',').map((p) => p.trim()); @@ -16,7 +45,11 @@ export async function serveMachineDoor(request, env) { return plain(426, `this door speaks ${PROTOCOL} over a WebSocket`); } const { pathname } = new URL(request.url); - const name = pathname === ENGINE_PATH ? request.headers.get(HEADER_NAME) ?? '' : pathname.slice(DIAL_PREFIX.length); + const register = pathname === ENGINE_PATH; + if (!(await allowCaller(request, env, register))) { + return plain(429, register ? TOO_MANY_REGISTRATIONS : TOO_MANY_DIALS); + } + const name = register ? request.headers.get(HEADER_NAME) ?? '' : pathname.slice(DIAL_PREFIX.length); if (!validName(name)) return plain(400, 'that is not a machine name'); const answer = await env.MACHINE.get(env.MACHINE.idFromName(name)).fetch(request); // An object's answer has immutable headers, and the Worker stamps Codeaf-Now on every answer. diff --git a/relay/hosted/src/engine/machine.js b/relay/hosted/src/engine/machine.js index 419d13be04..0279ed9799 100644 --- a/relay/hosted/src/engine/machine.js +++ b/relay/hosted/src/engine/machine.js @@ -1,226 +1,27 @@ -// The machine doors: `codeaf serve` registers at /v1/engine and a surface dials it at /v1/dial/. A port of -// internal/relay's server (server.go, handshake.go, mux.go) onto one Durable Object per machine name. -// -// THE RELAY READS NO PAYLOAD. What crosses here is the registration proof and the carrier's frames; a frame's -// payload is the tunnel's ciphertext, moved between two sockets and never decoded. -// -// EVERY SOCKET HIBERNATES. Each frame is one WebSocket message (the Go client writes a frame in one write), so the -// object keeps no half-read bytes across a wake-up, and the routing lives in socket tags and attachments, which -// survive hibernation. A quiet machine therefore costs nothing: its liveness is proved by WebSocket pings from the -// machine, which the platform answers without waking the object. The one timer is a registration's bounded wait for -// an older socket's pong, while the object is awake anyway. -// -// The per-address ceilings of relay.go are counted in memory, as the Worker's other per-minute rates are: an eviction -// can only forgive a burst. One object serves one machine name, so they bound what one address can do to one machine. +// Durable Object wrapper around MachineRoom (room.js). Kept thin so Node's test runner can exercise MachineRoom +// without importing `cloudflare:workers`. import { DurableObject } from 'cloudflare:workers'; -import { nameFor } from './names.js'; -import { ipOf } from '../wire.js'; -import { - DIALS_PER_MINUTE, ENGINE_PATH, HEADER_KEY, HEADER_NAME, MAX_PAYLOAD, MAX_STREAMS, OP, PROTOCOL, REGISTRATIONS_PER_MINUTE, - challenge, frame, plain as refuse, readFrame, verdict, -} from './carrier.js'; - -/** How long a registration waits for an older socket under the same name to prove it is still there. */ -const PROBE_MS = 5000; - -const ENGINE = 'engine'; -const DIAL = 'dial'; - -const sameBytes = (a, b) => a.length === b.length && a.every((x, i) => x === b[i]); -const b64 = (bytes) => btoa(String.fromCharCode(...bytes)); -const unb64 = (text) => Uint8Array.from(atob(text), (c) => c.charCodeAt(0)); -const bytesOf = (message) => (typeof message === 'string' ? new TextEncoder().encode(message) : new Uint8Array(message)); +import { MachineRoom } from './room.js'; export class MachineDO extends DurableObject { constructor(ctx, env) { super(ctx, env); - /** Registrations waiting for an older socket's pong, which is in-memory because the object is awake while they wait. */ - this.probes = new Set(); - /** claiming is a registration between its proof and its verdict; a second one meanwhile is refused. */ - this.claiming = false; - /** buckets are this minute's counts per caller network and door. */ - this.buckets = new Map(); - } - - async fetch(request) { - const register = new URL(request.url).pathname === ENGINE_PATH; - if (!this.#allow(ipOf(request, this.env), register ? 'register' : 'dial', register ? REGISTRATIONS_PER_MINUTE : DIALS_PER_MINUTE)) { - return refuse(429, register ? 'too many registrations from here — wait a minute' : 'too many connections from here — wait a minute'); - } - return register ? this.#register(request) : this.#dial(); - } - - /** allow spends one of a caller's per-minute allowance at a door. */ - #allow(ip, door, perMinute) { - const minute = Math.floor(Date.now() / 60000); - for (const [key, bucket] of this.buckets) if (bucket.minute !== minute) this.buckets.delete(key); - const key = door + ' ' + ip; - const bucket = this.buckets.get(key) ?? { minute, count: 0 }; - bucket.count += 1; - this.buckets.set(key, bucket); - return bucket.count <= perMinute; - } - - /** register accepts a machine's socket and sends the challenge; the proof arrives as its first message. */ - async #register(request) { - const name = request.headers.get(HEADER_NAME) ?? ''; - let key; - try { - key = unb64((request.headers.get(HEADER_KEY) ?? '').replace(/-/g, '+').replace(/_/g, '/')); - } catch { - return refuse(400, 'that key is not readable'); - } - if (key.length !== 32) return refuse(400, `a machine key is 32 bytes and that one is ${key.length}`); - // RULE ONE: the name must be the one this key derives, recomputed rather than trusted. - if ((await nameFor(key)) !== name) return refuse(400, 'that name does not belong to that key'); - let sent, want; - try { - ({ sent, want } = await challenge(name, key)); - } catch { - return refuse(400, 'that key is not a usable machine key'); - } - const { 0: client, 1: server } = new WebSocketPair(); - this.ctx.acceptWebSocket(server, [ENGINE]); - server.serializeAttachment({ role: 'pending', want: b64(want), name }); - server.send(sent); - return new Response(null, { status: 101, webSocket: client, headers: { 'sec-websocket-protocol': PROTOCOL } }); - } - - /** dial matches a surface to the machine and opens a stream on its carrier. */ - async #dial() { - const engine = this.#engine(); - if (!engine) return refuse(404, 'no machine is connected under that name'); - if (this.ctx.getWebSockets(DIAL).length >= MAX_STREAMS) { - return refuse(429, `that machine already has ${MAX_STREAMS} connections open, which is as many as the relay carries at once`); - } - const id = String((this.ctx.storage.kv.get('next') ?? 0) + 1); - this.ctx.storage.kv.put('next', Number(id)); - const { 0: client, 1: server } = new WebSocketPair(); - this.ctx.acceptWebSocket(server, [DIAL, 's:' + id]); - server.serializeAttachment({ role: 'surface', id }); - engine.send(frame(OP.open, id)); - return new Response(null, { status: 101, webSocket: client, headers: { 'sec-websocket-protocol': PROTOCOL } }); - } - - /** engine is the registered machine's socket, or null. */ - #engine(except) { - return this.ctx.getWebSockets(ENGINE).find((ws) => ws !== except && ws.deserializeAttachment()?.role === 'engine') ?? null; - } - - async webSocketMessage(ws, message) { - const seen = ws.deserializeAttachment(); - const bytes = bytesOf(message); - if (seen.role === 'surface') return this.#fromSurface(seen.id, bytes); - if (seen.role === 'pending') return this.#proof(ws, seen, bytes); - if (seen.role === 'engine') return this.#fromMachine(ws, bytes); - } - - /** fromSurface carries a surface's bytes to the machine, cut into carrier-sized frames. */ - #fromSurface(id, bytes) { - const engine = this.#engine(); - if (!engine) return this.#closeStream(id, 1011, 'the machine left'); - for (let at = 0; at < bytes.length; at += MAX_PAYLOAD) engine.send(frame(OP.data, id, bytes.subarray(at, at + MAX_PAYLOAD))); - } - - /** fromMachine routes one carrier frame from the machine. */ - #fromMachine(ws, bytes) { - const got = readFrame(bytes); - if (!got) return this.#endEngine(ws, 1002, 'that frame is not one carrier frame'); - if (got.op === OP.pong) return this.probes.forEach((probe) => probe(ws)); - if (got.op === OP.ping) return ws.send(frame(OP.pong, got.stream)); - const surface = this.ctx.getWebSockets('s:' + got.stream)[0]; - if (!surface) return; - if (got.op === OP.data) surface.send(got.payload); - if (got.op === OP.close) this.#closeStream(got.stream, 1000, 'closed'); - } - - /** proof checks the machine's answer and, rule three, never lets a live registration be taken over. */ - async #proof(ws, seen, bytes) { - if (!sameBytes(bytes, unb64(seen.want))) { - return this.#refuseRegistration(ws, 'that machine did not prove it holds the key its name is derived from'); - } - // RULE THREE: a live registration is never taken over, and neither is one being decided. - const older = this.#engine(ws); - if (this.claiming) return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); - this.claiming = true; - try { - if (older && (await this.#answers(older))) { - return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); - } - if (older) this.#endEngine(older, 1001, 'replaced by a newer registration after it stopped answering'); - ws.serializeAttachment({ role: 'engine', name: seen.name }); - ws.send(verdict()); - } finally { - this.claiming = false; - } - } - - /** - * answers pings an older socket under this name and reports whether it pongs. A machine whose network dropped can - * leave a socket the platform has not yet closed; a registration that holds the key replaces only a silent one. - */ - #answers(older) { - return new Promise((resolve) => { - const probe = (from) => from === older && done(true); - const done = (alive) => { - this.probes.delete(probe); - clearTimeout(timer); - resolve(alive); - }; - const timer = setTimeout(() => done(false), PROBE_MS); - this.probes.add(probe); - try { - older.send(frame(OP.ping, 0)); - } catch { - done(false); - } - }); - } - - #refuseRegistration(ws, reason) { - try { - ws.send(verdict(reason)); - ws.close(1008, 'registration refused'); - } catch {} - } - - async webSocketClose(ws, code) { - this.#gone(ws); - try { - ws.close(code === 1005 || code === 1006 ? 1000 : code, 'closed'); - } catch {} + this.room = new MachineRoom(ctx, env); } - async webSocketError(ws) { - this.#gone(ws); + fetch(request) { + return this.room.fetch(request); } - /** gone tells the other end of whatever the socket that ended was carrying. */ - #gone(ws) { - const seen = ws.deserializeAttachment(); - if (seen?.role === 'surface') { - try { - this.#engine()?.send(frame(OP.close, seen.id)); - } catch {} - } else if (seen?.role === 'engine') { - ws.serializeAttachment({ ...seen, role: 'ended' }); - for (const surface of this.ctx.getWebSockets(DIAL)) this.#closeStream(surface.deserializeAttachment().id, 1001, 'the machine left'); - } + webSocketMessage(ws, message) { + return this.room.webSocketMessage(ws, message); } - #endEngine(ws, code, reason) { - this.#gone(ws); - try { - ws.close(code, reason); - } catch {} + webSocketClose(ws, code) { + return this.room.webSocketClose(ws, code); } - #closeStream(id, code, reason) { - for (const surface of this.ctx.getWebSockets('s:' + id)) { - surface.serializeAttachment({ role: 'ended', id }); - try { - surface.close(code, reason); - } catch {} - } + webSocketError(ws) { + return this.room.webSocketError(ws); } } diff --git a/relay/hosted/src/engine/names.js b/relay/hosted/src/engine/names.js index 863a7c4760..5eff35bd58 100644 --- a/relay/hosted/src/engine/names.js +++ b/relay/hosted/src/engine/names.js @@ -49,5 +49,15 @@ export async function nameFor(publicKey) { return `${WORDS[sum[0]]}-${WORDS[sum[1]]}-${10 + (sum[2] % 90)}`; } -/** validName is the shape of a machine name: two lowercase words of up to 12 letters and two digits. */ -export const validName = (name) => typeof name === 'string' && /^[a-z]{1,12}-[a-z]{1,12}-[0-9]{2}$/.test(name); +const WORD_SET = new Set(WORDS); + +/** + * validName reports whether name is one nameFor can produce: two words from WORDS and a number 10–99. Unlike the Go + * relay's in-memory table lookup, each distinct name on Workers addresses a Durable Object, so anything outside the + * real name space is refused at the front door before idFromName can create an object for it. + */ +export function validName(name) { + if (typeof name !== 'string') return false; + const parts = name.split('-'); + return parts.length === 3 && WORD_SET.has(parts[0]) && WORD_SET.has(parts[1]) && /^[1-9][0-9]$/.test(parts[2]); +} diff --git a/relay/hosted/src/engine/room.js b/relay/hosted/src/engine/room.js new file mode 100644 index 0000000000..e3c3a38612 --- /dev/null +++ b/relay/hosted/src/engine/room.js @@ -0,0 +1,326 @@ +// One machine name's room: `codeaf serve` registers at /v1/engine and a surface dials it at /v1/dial/. Kept free +// of the `cloudflare:workers` base class so Node unit tests can exercise the real socket state machine directly, while +// MachineDO in machine.js wraps it as a Durable Object. +// +// THE RELAY READS NO PAYLOAD. What crosses here is the registration proof and the carrier's frames; a frame's +// payload is the tunnel's ciphertext, moved between two sockets and never decoded. +// +// EVERY SOCKET HIBERNATES. Each frame is one WebSocket message (the Go client writes a frame in one write), so the +// object keeps no half-read bytes across a wake-up, and the routing lives in socket tags and attachments, which +// survive hibernation. A quiet machine therefore costs nothing: its liveness is proved by WebSocket pings from the +// machine, which the platform answers without waking the object. The one timer is a registration's bounded wait for +// an older socket's pong under the same key, while the object is awake anyway. +import { nameFor } from './names.js'; +import { ipOf } from '../wire.js'; +import { + DIALS_PER_MINUTE, ENGINE_PATH, HANDSHAKE_MS, HEADER_KEY, HEADER_NAME, MAX_PAYLOAD, MAX_STREAMS, OP, PROBE_MS, + PROTOCOL, REGISTRATIONS_PER_MINUTE, TOO_MANY_DIALS, TOO_MANY_REGISTRATIONS, + challenge, frame, plain as refuse, readFrame, verdict, +} from './carrier.js'; + +const ENGINE = 'engine'; +const DIAL = 'dial'; + +function sameBytes(a, b) { + if (a.length !== b.length) return false; + let diff = 0; + for (let i = 0; i < a.length; i += 1) diff |= a[i] ^ b[i]; + return diff === 0; +} + +const b64 = (bytes) => btoa(String.fromCharCode(...bytes)); +const unb64 = (text) => Uint8Array.from(atob(text), (c) => c.charCodeAt(0)); +const bytesOf = (message) => (typeof message === 'string' ? new TextEncoder().encode(message) : new Uint8Array(message)); + +export class MachineRoom { + constructor(ctx, env, options = {}) { + this.ctx = ctx; + this.env = env; + this.probeMs = options.probeMs ?? PROBE_MS; + this.handshakeMs = options.handshakeMs ?? HANDSHAKE_MS; + this.clock = options.clock ?? Date.now; + /** Registrations waiting for an older socket's pong, kept in memory because the object is awake while they wait. */ + this.probes = new Set(); + /** claiming is a registration between its proof and its verdict; a second one meanwhile is refused. */ + this.claiming = false; + /** buckets are this minute's counts per caller network and door, kept as a second line behind door.js. */ + this.buckets = new Map(); + } + + async fetch(request) { + this.#sweepPending(); + const register = new URL(request.url).pathname === ENGINE_PATH; + if (!this.#allow(ipOf(request, this.env), register ? 'register' : 'dial', register ? REGISTRATIONS_PER_MINUTE : DIALS_PER_MINUTE)) { + return refuse(429, register ? TOO_MANY_REGISTRATIONS : TOO_MANY_DIALS); + } + return register ? this.#register(request) : this.#dial(); + } + + /** allow spends one of a caller's per-minute allowance at this machine's door. */ + #allow(ip, door, perMinute) { + const minute = Math.floor(this.clock() / 60000); + for (const [key, bucket] of this.buckets) if (bucket.minute !== minute) this.buckets.delete(key); + const key = door + ' ' + ip; + const bucket = this.buckets.get(key) ?? { minute, count: 0 }; + bucket.count += 1; + this.buckets.set(key, bucket); + return bucket.count <= perMinute; + } + + /** + * sweepPending evicts /v1/engine upgrades that never sent their possession proof within HANDSHAKE_MS. Running on + * arrivals instead of a resting timer keeps idle registered machines hibernating at zero duration cost. + */ + #sweepPending() { + const now = this.clock(); + for (const ws of this.ctx.getWebSockets(ENGINE)) { + const seen = ws.deserializeAttachment(); + if (seen?.role === 'pending' && typeof seen.at === 'number' && now - seen.at >= this.handshakeMs) { + ws.serializeAttachment({ ...seen, role: 'ended' }); + try { + ws.close(1008, 'registration timed out'); + } catch {} + } + } + } + + /** register accepts a machine's socket and sends the challenge; the proof arrives as its first message. */ + async #register(request) { + const name = request.headers.get(HEADER_NAME) ?? ''; + let key; + try { + key = unb64((request.headers.get(HEADER_KEY) ?? '').replace(/-/g, '+').replace(/_/g, '/')); + } catch { + return refuse(400, 'that key is not readable'); + } + if (key.length !== 32) return refuse(400, `a machine key is 32 bytes and that one is ${key.length}`); + // RULE ONE: the name must be the one this key derives, recomputed rather than trusted. + if ((await nameFor(key)) !== name) return refuse(400, 'that name does not belong to that key'); + let sent, want; + try { + ({ sent, want } = await challenge(name, key)); + } catch { + return refuse(400, 'that key is not a usable machine key'); + } + const { 0: client, 1: server } = new WebSocketPair(); + this.ctx.acceptWebSocket(server, [ENGINE]); + server.serializeAttachment({ role: 'pending', want: b64(want), key: b64(key), name, at: this.clock() }); + try { + server.send(sent); + } catch { + server.serializeAttachment({ role: 'ended', name }); + try { + server.close(1011, 'closed'); + } catch {} + return refuse(503, 'the registration closed before it began'); + } + return new Response(null, { status: 101, webSocket: client, headers: { 'sec-websocket-protocol': PROTOCOL } }); + } + + /** dial matches a surface to the machine and opens a stream on its carrier. */ + async #dial() { + const engine = this.#engine(); + if (!engine) return refuse(404, 'no machine is connected under that name'); + const active = this.ctx.getWebSockets(DIAL).filter((ws) => ws.deserializeAttachment()?.role === 'surface'); + if (active.length >= MAX_STREAMS) { + return refuse(429, `that machine already has ${MAX_STREAMS} connections open, which is as many as the relay carries at once`); + } + const id = String((this.ctx.storage.kv.get('next') ?? 0) + 1); + this.ctx.storage.kv.put('next', Number(id)); + // Tell the machine before accepting the surface socket into ctx: if engine has already begun closing, throwing + // here must not leave an orphaned DIAL socket occupying one of MAX_STREAMS forever. + try { + engine.send(frame(OP.open, id)); + } catch { + this.#endEngine(engine, 1011, 'the machine left'); + return refuse(404, 'no machine is connected under that name'); + } + const { 0: client, 1: server } = new WebSocketPair(); + this.ctx.acceptWebSocket(server, [DIAL, 's:' + id]); + server.serializeAttachment({ role: 'surface', id }); + return new Response(null, { status: 101, webSocket: client, headers: { 'sec-websocket-protocol': PROTOCOL } }); + } + + /** engine is the registered machine's socket, or null. */ + #engine(except) { + return this.ctx.getWebSockets(ENGINE).find((ws) => ws !== except && ws.deserializeAttachment()?.role === 'engine') ?? null; + } + + async webSocketMessage(ws, message) { + const seen = ws.deserializeAttachment(); + if (!seen || seen.role === 'ended') return; + const bytes = bytesOf(message); + if (seen.role === 'surface') return this.#fromSurface(seen.id, bytes); + if (seen.role === 'pending') return this.#proof(ws, seen, bytes); + if (seen.role === 'engine') return this.#fromMachine(ws, bytes); + } + + /** fromSurface carries a surface's bytes to the machine, cut into carrier-sized frames. */ + #fromSurface(id, bytes) { + const engine = this.#engine(); + if (!engine) return this.#closeStream(id, 1011, 'the machine left'); + try { + for (let at = 0; at < bytes.length; at += MAX_PAYLOAD) { + engine.send(frame(OP.data, id, bytes.subarray(at, at + MAX_PAYLOAD))); + } + } catch { + this.#endEngine(engine, 1011, 'the machine left'); + } + } + + /** fromMachine routes one carrier frame from the machine. */ + #fromMachine(ws, bytes) { + const got = readFrame(bytes); + if (!got) return this.#endEngine(ws, 1002, 'that frame is not one carrier frame'); + if (got.op === OP.pong) return this.probes.forEach((probe) => probe(ws)); + if (got.op === OP.ping) { + try { + ws.send(frame(OP.pong, got.stream)); + } catch { + this.#endEngine(ws, 1011, 'closed'); + } + return; + } + const surface = this.ctx.getWebSockets('s:' + got.stream).find((s) => s.deserializeAttachment()?.role === 'surface'); + if (!surface) return; + if (got.op === OP.data) { + try { + surface.send(got.payload); + } catch { + this.#gone(surface); + try { + surface.close(1011, 'closed'); + } catch {} + } + } + if (got.op === OP.close) this.#closeStream(got.stream, 1000, 'closed'); + } + + /** + * proof checks the machine's answer and, rule three, never lets a live registration be taken over. Only the SAME + * public key may probe and replace an older socket that went silent; a colliding key is refused whether or not the + * current socket answers within PROBE_MS. + */ + async #proof(ws, seen, bytes) { + this.#sweepPending(); + if (ws.deserializeAttachment()?.role !== 'pending') return; + let want; + try { + want = unb64(seen.want); + } catch { + return this.#refuseRegistration(ws, 'that machine did not prove it holds the key its name is derived from'); + } + if (!sameBytes(bytes, want)) { + return this.#refuseRegistration(ws, 'that machine did not prove it holds the key its name is derived from'); + } + const older = this.#engine(ws); + if (this.claiming) { + return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); + } + if (older) { + const olderKey = older.deserializeAttachment()?.key; + if (!olderKey || !seen.key || !sameBytes(unb64(olderKey), unb64(seen.key))) { + return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); + } + } + this.claiming = true; + try { + if (older && (await this.#answers(older))) { + return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); + } + // If the new registrant disconnected while we waited on older's pong, neither evict older nor promote ws. + if (ws.deserializeAttachment()?.role !== 'pending') return; + try { + ws.send(verdict()); + } catch { + ws.serializeAttachment({ ...seen, role: 'ended' }); + try { + ws.close(1011, 'closed'); + } catch {} + return; + } + if (older) this.#endEngine(older, 1001, 'replaced by a newer registration after it stopped answering'); + ws.serializeAttachment({ role: 'engine', name: seen.name, key: seen.key }); + } finally { + this.claiming = false; + } + } + + /** + * answers pings an older socket under the same key and reports whether it pongs. A machine whose network dropped can + * leave a socket the platform has not yet closed; a registration that holds the same key replaces only a silent one. + */ + #answers(older) { + return new Promise((resolve) => { + const probe = (from) => from === older && done(true); + const done = (alive) => { + this.probes.delete(probe); + clearTimeout(timer); + resolve(alive); + }; + const timer = setTimeout(() => done(false), this.probeMs); + this.probes.add(probe); + try { + older.send(frame(OP.ping, 0)); + } catch { + done(false); + } + }); + } + + #refuseRegistration(ws, reason) { + const seen = ws.deserializeAttachment(); + if (seen) ws.serializeAttachment({ ...seen, role: 'ended' }); + try { + ws.send(verdict(reason)); + ws.close(1008, 'registration refused'); + } catch {} + } + + async webSocketClose(ws, code) { + this.#gone(ws); + try { + ws.close(code === 1005 || code === 1006 || code === 1015 ? 1000 : code, 'closed'); + } catch {} + } + + async webSocketError(ws) { + this.#gone(ws); + try { + ws.close(1011, 'closed'); + } catch {} + } + + /** gone tells the other end of whatever the socket that ended was carrying, once. */ + #gone(ws) { + const seen = ws.deserializeAttachment(); + if (!seen || seen.role === 'ended') return; + ws.serializeAttachment({ ...seen, role: 'ended' }); + if (seen.role === 'surface') { + try { + this.#engine()?.send(frame(OP.close, seen.id)); + } catch {} + } else if (seen.role === 'engine') { + for (const surface of this.ctx.getWebSockets(DIAL)) { + const id = surface.deserializeAttachment()?.id; + if (id) this.#closeStream(id, 1001, 'the machine left'); + } + } + } + + #endEngine(ws, code, reason) { + this.#gone(ws); + try { + ws.close(code, reason); + } catch {} + } + + #closeStream(id, code, reason) { + for (const surface of this.ctx.getWebSockets('s:' + id)) { + surface.serializeAttachment({ role: 'ended', id }); + try { + surface.close(code, reason); + } catch {} + } + } +} diff --git a/relay/hosted/test/config.test.js b/relay/hosted/test/config.test.js index 29baad4c70..11453ad189 100644 --- a/relay/hosted/test/config.test.js +++ b/relay/hosted/test/config.test.js @@ -18,4 +18,6 @@ test('R30 the base config uses the Raft entry and an append-only v3-raft-members assert.match(text, /name = "IDENTITY"\s+class_name = "IdentityDO"/); assert.match(text, /name = "MACHINE"\s+class_name = "MachineDO"/); assert.match(text, /tag = "v4-machines"\s+new_sqlite_classes = \["MachineDO"\]/); + assert.match(text, /name = "MACHINE_DIAL_LIMIT"[\s\S]*?limit = 30,\s*period = 60/); + assert.match(text, /name = "MACHINE_REGISTER_LIMIT"[\s\S]*?limit = 10,\s*period = 60/); }); diff --git a/relay/hosted/test/engine.test.js b/relay/hosted/test/engine.test.js index d51a07516f..7ac5a3b7f5 100644 --- a/relay/hosted/test/engine.test.js +++ b/relay/hosted/test/engine.test.js @@ -1,14 +1,206 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { nameFor, validName } from '../src/engine/names.js'; -import { MAX_PAYLOAD, frame, readFrame, registrationProof, verdict } from '../src/engine/carrier.js'; +import { + DIALS_PER_MINUTE, ENGINE_PATH, HEADER_KEY, HEADER_NAME, MAX_PAYLOAD, MAX_STREAMS, OP, PROTOCOL, + REGISTRATIONS_PER_MINUTE, TOO_MANY_DIALS, TOO_MANY_REGISTRATIONS, + frame, readFrame, registrationProof, verdict, +} from '../src/engine/carrier.js'; +import { serveMachineDoor } from '../src/engine/door.js'; +import { MachineRoom } from '../src/engine/room.js'; const hex = (text) => Uint8Array.from(text.match(/../g), (b) => parseInt(b, 16)); const toHex = (bytes) => [...bytes].map((b) => b.toString(16).padStart(2, '0')).join(''); const run = (from, n = 32) => Uint8Array.from({ length: n }, (_, i) => (from + i) & 0xff); +const b64url = (bytes) => Buffer.from(bytes).toString('base64url'); + +// PKCS#8 header for a raw 32-byte X25519 private seed (RFC 8410). +const PKCS8_X25519 = hex('302e020100300506032b656e04220420'); + +async function keyFromSeed(seed32) { + const pkcs8 = new Uint8Array(PKCS8_X25519.length + 32); + pkcs8.set(PKCS8_X25519); + pkcs8.set(seed32, PKCS8_X25519.length); + const privateKey = await crypto.subtle.importKey('pkcs8', pkcs8, { name: 'X25519' }, true, ['deriveBits']); + // JWK export of an X25519 private key carries the derived raw public key in `x`. + const jwk = await crypto.subtle.exportKey('jwk', privateKey); + const publicKey = new Uint8Array(Buffer.from(jwk.x, 'base64url')); + return { privateKey, publicKey, name: await nameFor(publicKey) }; +} + +let cachedCollision = null; + +// collidingKeys finds two distinct X25519 keypairs whose public keys derive the exact same machine name. +async function collidingKeys() { + if (cachedCollision) return cachedCollision; + const seen = new Map(); + for (let i = 1; ; i += 1) { + const seed = new Uint8Array(32); + new DataView(seed.buffer).setUint32(0, i, true); + const kp = await keyFromSeed(seed); + const other = seen.get(kp.name); + if (other && toHex(other.publicKey) !== toHex(kp.publicKey)) { + cachedCollision = [other, kp]; + return cachedCollision; + } + seen.set(kp.name, kp); + } +} + +async function answerChallenge(challengeMessage, name, kp) { + const relayPub = await crypto.subtle.importKey('raw', challengeMessage.subarray(0, 32), { name: 'X25519' }, false, []); + const shared = new Uint8Array(await crypto.subtle.deriveBits({ name: 'X25519', public: relayPub }, kp.privateKey, 256)); + const nonce = challengeMessage.subarray(32, 64); + return registrationProof(shared, nonce, name, kp.publicKey); +} + +// makeRoom harnesses MachineRoom with a real in-memory hibernatable-WebSocket context in Node. +function makeRoom(options = {}) { + const realPair = globalThis.WebSocketPair; + const realResponse = globalThis.Response; + const sockets = []; + const kv = new Map(); + const clock = { now: 1_000_000 }; + + function makeSocket(peerRef) { + const ws = { + tags: [], + attachment: null, + sent: [], + closed: null, + accepted: false, + sendFail: false, + peer: peerRef, + serializeAttachment(value) { + this.attachment = structuredClone(value); + }, + deserializeAttachment() { + return this.attachment ? structuredClone(this.attachment) : null; + }, + send(data) { + if (this.sendFail || this.closed) throw new Error('WebSocket is closing'); + const copy = typeof data === 'string' ? data : new Uint8Array(data); + this.sent.push(copy); + if (this.peer && this.peer.onmessage) this.peer.onmessage(copy); + }, + close(code = 1000, reason = '') { + this.closed = [code, reason]; + }, + }; + return ws; + } + + globalThis.WebSocketPair = class { + constructor() { + const client = { sent: [], onmessage: null }; + const server = makeSocket(client); + client.server = server; + Object.assign(this, { 0: client, 1: server }); + } + }; + + globalThis.Response = class extends realResponse { + constructor(body, init = {}) { + const { status = 200, webSocket, ...rest } = init; + super(status === 101 ? null : body, { ...rest, status: status === 101 ? 200 : status }); + this.realStatus = status; + this.webSocket = webSocket; + } + get status() { + return this.realStatus; + } + }; + + const ctx = { + storage: { kv: { get: (k) => kv.get(k), put: (k, v) => kv.set(k, v) } }, + acceptWebSocket(ws, tags = []) { + ws.accepted = true; + ws.tags = [...tags]; + sockets.push(ws); + }, + getWebSockets(tag) { + // keepClosing simulates Cloudflare keeping a socket in getWebSockets while a silent peer has not echoed Close. + return sockets.filter((ws) => (!ws.closed || ws.keepClosing) && (tag === undefined || ws.tags.includes(tag))); + }, + }; + + const env = options.env ?? {}; + const room = new MachineRoom(ctx, env, { + probeMs: options.probeMs ?? 25, + handshakeMs: options.handshakeMs ?? 20_000, + clock: () => clock.now, + }); + + const restore = () => { + globalThis.WebSocketPair = realPair; + globalThis.Response = realResponse; + }; + + async function register(kp, { answerPings = true, ip = '198.51.100.1' } = {}) { + const req = new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': ip, + [HEADER_NAME]: kp.name, + [HEADER_KEY]: b64url(kp.publicKey), + }, + }); + const res = await room.fetch(req); + assert.equal(res.status, 101); + const server = res.webSocket.server; + assert.equal(server.sent.length, 1); + const proof = await answerChallenge(server.sent[0], kp.name, kp); + res.webSocket.onmessage = (msg) => { + if (!answerPings) return; + const parsed = readFrame(msg); + if (parsed && parsed.op === OP.ping) { + void room.webSocketMessage(server, frame(OP.pong, parsed.stream)); + } + }; + await room.webSocketMessage(server, proof); + const last = server.sent.at(-1); + return { res, client: res.webSocket, server, verdict: last }; + } + + async function dial(name, ip = '198.51.100.2') { + const req = new Request('https://relay.example/v1/dial/' + name, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': ip, + }, + }); + return room.fetch(req); + } + + return { room, ctx, sockets, clock, register, dial, restore }; +} + +async function withRoom(options, fn) { + const harness = makeRoom(options); + try { + return await fn(harness); + } finally { + harness.restore(); + } +} + +// stubLimiter mimics Cloudflare's [[ratelimits]] binding (`env..limit({ key })`). +function stubLimiter(limit) { + const counts = new Map(); + return { + counts, + async limit({ key }) { + const next = (counts.get(key) ?? 0) + 1; + counts.set(key, next); + return { success: next <= limit }; + }, + }; +} // Computed by internal/relay's NameFor; the port must derive the same name from the same key. -test('a machine name is the one the Go side derives from the same key', async () => { +test('a machine name is the one the Go side derives from the same key, and validName accepts only real names', async () => { for (const [key, name] of [ ['000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f', 'lantern-amber-86'], ['0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20', 'jasmine-dragon-58'], @@ -18,7 +210,22 @@ test('a machine name is the one the Go side derives from the same key', async () assert.equal(await nameFor(hex(key)), name); assert.ok(validName(name)); } - for (const bad of ['otter-lamp-4', 'Otter-lamp-42', 'otter-lamp', 'otter-lamp-42-x', 'otterlampotter-x-42']) assert.ok(!validName(bad), bad); + assert.ok(validName('otter-lamp-10')); + assert.ok(validName('willow-acorn-99')); + for (const bad of [ + 'otter-lamp-4', + 'Otter-lamp-42', + 'otter-lamp', + 'otter-lamp-42-x', + 'otterlampotter-x-42', + 'a-a-42', + 'notaword-lamp-42', + 'otter-notaword-42', + 'otter-lamp-00', + 'otter-lamp-09', + ]) { + assert.ok(!validName(bad), bad); + } }); // Computed by internal/relay's registrationProof over the same shared secret, nonce, name and key. @@ -47,3 +254,242 @@ test('a verdict is ready, or refused with its sentence', () => { const said = verdict('no'); assert.deepEqual([...said], [2, 2, 110, 111]); }); + +// Contract item 4 (part 1): invalid names are refused in the Worker front before any Durable Object is touched. +test('the Worker front refuses invalid machine names before touching a Durable Object', async () => { + let touched = 0; + const env = { + MACHINE_DIAL_LIMIT: stubLimiter(DIALS_PER_MINUTE), + MACHINE_REGISTER_LIMIT: stubLimiter(REGISTRATIONS_PER_MINUTE), + MACHINE: { + idFromName() { + touched += 1; + throw new Error('MACHINE.idFromName must not be called for an invalid name'); + }, + get() { + touched += 1; + throw new Error('MACHINE.get must not be called for an invalid name'); + }, + }, + }; + + for (const bad of ['a-a-42', 'otter-lamp-00', 'otter-lamp-09', 'notaword-lamp-42', 'otter-lamp-999']) { + const dialRes = await serveMachineDoor( + new Request('https://relay.example/v1/dial/' + bad, { + headers: { upgrade: 'websocket', 'sec-websocket-protocol': PROTOCOL, 'cf-connecting-ip': '203.0.113.7' }, + }), + env, + ); + assert.equal(dialRes.status, 400); + assert.equal(await dialRes.text(), 'that is not a machine name'); + } + + const regRes = await serveMachineDoor( + new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': '203.0.113.7', + [HEADER_NAME]: 'a-a-42', + }, + }), + env, + ); + assert.equal(regRes.status, 400); + assert.equal(await regRes.text(), 'that is not a machine name'); + assert.equal(touched, 0); +}); + +// Contract item 4 (part 2): rate limits follow the caller network across distinct names before touching env.MACHINE. +test('the 31st dial and 11th registration in a minute from one network across distinct names are refused with 429', async () => { + const dialLimiter = stubLimiter(DIALS_PER_MINUTE); + const regLimiter = stubLimiter(REGISTRATIONS_PER_MINUTE); + const reached = []; + const env = { + MACHINE_DIAL_LIMIT: dialLimiter, + MACHINE_REGISTER_LIMIT: regLimiter, + MACHINE: { + idFromName: (name) => name, + get: (id) => ({ + async fetch() { + reached.push(id); + return new Response('no machine is connected under that name', { status: 404 }); + }, + }), + }, + }; + + for (let i = 0; i < DIALS_PER_MINUTE; i += 1) { + const name = `acorn-alder-${10 + i}`; + const res = await serveMachineDoor( + new Request('https://relay.example/v1/dial/' + name, { + // Distinct hosts inside one IPv6 /64 share one caller network bucket, just as ipOf enforces elsewhere. + headers: { upgrade: 'websocket', 'sec-websocket-protocol': PROTOCOL, 'cf-connecting-ip': `2001:db8:1:2::${i + 1}` }, + }), + env, + ); + assert.equal(res.status, 404); + } + assert.equal(reached.length, DIALS_PER_MINUTE); + + const thirtyFirst = await serveMachineDoor( + new Request('https://relay.example/v1/dial/acorn-alder-50', { + headers: { upgrade: 'websocket', 'sec-websocket-protocol': PROTOCOL, 'cf-connecting-ip': '2001:db8:1:2::99' }, + }), + env, + ); + assert.equal(thirtyFirst.status, 429); + assert.equal(await thirtyFirst.text(), TOO_MANY_DIALS); + assert.equal(reached.length, DIALS_PER_MINUTE, 'the 31st dial must not wake a Durable Object'); + + for (let i = 0; i < REGISTRATIONS_PER_MINUTE; i += 1) { + const name = `amber-anchor-${10 + i}`; + const res = await serveMachineDoor( + new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': '198.51.100.44', + [HEADER_NAME]: name, + }, + }), + env, + ); + assert.equal(res.status, 404); + } + const eleventh = await serveMachineDoor( + new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': '198.51.100.44', + [HEADER_NAME]: 'amber-anchor-30', + }, + }), + env, + ); + assert.equal(eleventh.status, 429); + assert.equal(await eleventh.text(), TOO_MANY_REGISTRATIONS); + assert.equal(reached.length, DIALS_PER_MINUTE + REGISTRATIONS_PER_MINUTE, 'the 11th registration must not wake a Durable Object'); +}); + +// Contract item 3: key-checked replacement on MachineRoom (different key + silent old -> refused; +// same key + live old -> refused; same key + silent old -> replaces). +test('only the same public key may replace a silent registration', async () => { + const [owner, squatter] = await collidingKeys(); + assert.equal(owner.name, squatter.name); + assert.notEqual(toHex(owner.publicKey), toHex(squatter.publicKey)); + + // 1. Different key + silent old -> refused without evicting the owner. + await withRoom({ probeMs: 20 }, async (h) => { + const first = await h.register(owner, { answerPings: false }); + assert.deepEqual([...first.verdict], [1, 0]); + + const impostor = await h.register(squatter, { answerPings: true, ip: '198.51.100.9' }); + assert.equal(impostor.verdict[0], 2, 'a colliding key must be refused even when the owner is silent'); + assert.match(new TextDecoder().decode(impostor.verdict.subarray(2)), /already connected/); + assert.equal(first.server.closed, null, 'the silent owner must not be evicted by a different key'); + }); + + // 2. Same key + live old -> refused. + await withRoom({ probeMs: 20 }, async (h) => { + const first = await h.register(owner, { answerPings: true }); + assert.deepEqual([...first.verdict], [1, 0]); + + const second = await h.register(owner, { answerPings: true }); + assert.equal(second.verdict[0], 2, 'a live registration must not be replaced even by the same key'); + assert.equal(first.server.closed, null); + }); + + // 3. Same key + silent old -> replaces and routes new dials to the new socket. + await withRoom({ probeMs: 20 }, async (h) => { + const first = await h.register(owner, { answerPings: false }); + assert.deepEqual([...first.verdict], [1, 0]); + + const second = await h.register(owner, { answerPings: true }); + assert.deepEqual([...second.verdict], [1, 0], 'the same key must replace its own silent socket'); + assert.equal(first.server.closed?.[0], 1001); + + const surface = await h.dial(owner.name); + assert.equal(surface.status, 101); + const opened = readFrame(second.server.sent.at(-1)); + assert.equal(opened?.op, OP.open); + }); +}); + +// Contract item 7 (Finding 6 & Finding 1 socket hygiene): throwing engine.send in #dial leaks no stream slot, +// closing/errored surface sockets free MAX_STREAMS immediately, dropped pending registrants do not evict older, +// and abandoned pending upgrades are swept past HANDSHAKE_MS. +test('MachineRoom cleans up throwing sends, closing surfaces, dropped challengers and stale pending sockets', async () => { + const [owner] = await collidingKeys(); + + // 1. engine.send throwing during #dial does not leak a DIAL socket or return 500. + await withRoom({ probeMs: 20 }, async (h) => { + const first = await h.register(owner); + first.server.sendFail = true; + const failedDial = await h.dial(owner.name); + assert.equal(failedDial.status, 404); + assert.equal(h.ctx.getWebSockets('dial').length, 0, 'failed #dial must not accept an orphaned surface socket'); + }); + + // 2. 16 surface sockets that error/close while their peer withholds Close echo still free MAX_STREAMS at once. + await withRoom({ probeMs: 20 }, async (h) => { + await h.register(owner); + const surfaces = []; + for (let i = 0; i < MAX_STREAMS; i += 1) { + const res = await h.dial(owner.name, `198.51.100.${10 + i}`); + assert.equal(res.status, 101); + res.webSocket.server.keepClosing = true; + surfaces.push(res.webSocket.server); + } + const full = await h.dial(owner.name, '198.51.100.99'); + assert.equal(full.status, 429); + + await h.room.webSocketError(surfaces[0]); + assert.equal(surfaces[0].deserializeAttachment()?.role, 'ended'); + const reopened = await h.dial(owner.name, '198.51.100.100'); + assert.equal(reopened.status, 101, 'an errored surface socket must release its MAX_STREAMS slot immediately'); + }); + + // 3. A new same-key registrant that drops during the PROBE_MS wait does not evict older. + await withRoom({ probeMs: 30 }, async (h) => { + const first = await h.register(owner, { answerPings: false }); + const req = new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': '198.51.100.2', + [HEADER_NAME]: owner.name, + [HEADER_KEY]: b64url(owner.publicKey), + }, + }); + const res = await h.room.fetch(req); + const challenger = res.webSocket.server; + const proof = await answerChallenge(challenger.sent[0], owner.name, owner); + const pendingProof = h.room.webSocketMessage(challenger, proof); + await h.room.webSocketClose(challenger, 1006); + await pendingProof; + assert.equal(first.server.closed, null, 'a challenger that disconnected mid-probe must not evict the existing socket'); + assert.equal(challenger.deserializeAttachment()?.role, 'ended'); + }); + + // 4. Abandoned pending /v1/engine sockets are swept once HANDSHAKE_MS elapses. + await withRoom({ handshakeMs: 500 }, async (h) => { + const req = new Request('https://relay.example' + ENGINE_PATH, { + headers: { + upgrade: 'websocket', + 'sec-websocket-protocol': PROTOCOL, + 'cf-connecting-ip': '198.51.100.3', + [HEADER_NAME]: owner.name, + [HEADER_KEY]: b64url(owner.publicKey), + }, + }); + const res = await h.room.fetch(req); + const pending = res.webSocket.server; + assert.equal(pending.deserializeAttachment()?.role, 'pending'); + h.clock.now += 501; + await h.dial(owner.name); + assert.equal(pending.closed?.[0], 1008); + assert.equal(pending.deserializeAttachment()?.role, 'ended'); + }); +}); diff --git a/relay/hosted/wrangler.toml b/relay/hosted/wrangler.toml index ce65d46f67..8c9c4c086c 100644 --- a/relay/hosted/wrangler.toml +++ b/relay/hosted/wrangler.toml @@ -63,3 +63,16 @@ new_sqlite_classes = ["RaftMemberDO"] [[migrations]] tag = "v4-machines" new_sqlite_classes = ["MachineDO"] + +# Front-door per-network rate limits for /v1/dial/ and /v1/engine, +# matching DialsPerMinute (30) and RegistrationsPerMinute (10) in +# internal/relay/relay.go before any MachineDO wakes. +[[ratelimits]] +name = "MACHINE_DIAL_LIMIT" +namespace_id = "1001" +simple = { limit = 30, period = 60 } + +[[ratelimits]] +name = "MACHINE_REGISTER_LIMIT" +namespace_id = "1002" +simple = { limit = 10, period = 60 }