From 5024dd1104e72ecd2676cddf9c4572d399aba95e Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 30 Oct 2026 10:30:06 -0400 Subject: [PATCH 1/3] fix: reach a paired computer through the relay sync uses codeaf serve and chat --at fall back to the relay sync resolves, the relay client keeps the relay address's path, the carrier rides a WebSocket (with a fallback to the raw upgrade for older relays), the hosted relay gains the machine doors, and upstream failures become plain sentences. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/chatv3_at.go | 11 +- cmd/codeaf/chatv3_at_test.go | 44 +++- cmd/codeaf/pairroute.go | 10 + docs/remote-access-testing.md | 2 +- .../chat/reaching-this-machine-without-ssh.md | 57 +++-- internal/pair/dial.go | 7 +- internal/pair/edge_serve_test.go | 112 ++++++++++ internal/pair/errors.go | 6 + internal/pair/serve.go | 19 +- internal/pair/service.go | 30 +-- internal/relay/client.go | 66 ++++-- internal/relay/hosted_test.go | 86 ++++++++ internal/relay/mux.go | 24 +-- internal/relay/server.go | 55 +++-- internal/relay/transport.go | 103 ++++++++++ internal/relay/transport_test.go | 126 ++++++++++++ relay/hosted/src/engine/carrier.js | 72 +++++++ relay/hosted/src/engine/door.js | 29 +++ relay/hosted/src/engine/machine.js | 194 ++++++++++++++++++ relay/hosted/src/engine/names.js | 53 +++++ relay/hosted/src/front.js | 2 + relay/hosted/src/worker.js | 1 + relay/hosted/test/config.test.js | 4 +- relay/hosted/test/engine.test.js | 49 +++++ relay/hosted/wrangler.toml | 8 + 25 files changed, 1090 insertions(+), 80 deletions(-) create mode 100644 internal/relay/hosted_test.go create mode 100644 internal/relay/transport.go create mode 100644 internal/relay/transport_test.go create mode 100644 relay/hosted/src/engine/carrier.js create mode 100644 relay/hosted/src/engine/door.js create mode 100644 relay/hosted/src/engine/machine.js create mode 100644 relay/hosted/src/engine/names.js create mode 100644 relay/hosted/test/engine.test.js diff --git a/cmd/codeaf/chatv3_at.go b/cmd/codeaf/chatv3_at.go index 8dc3daa123..3a1fe34ca4 100644 --- a/cmd/codeaf/chatv3_at.go +++ b/cmd/codeaf/chatv3_at.go @@ -151,7 +151,8 @@ func openChatV3At(launch atLaunch) error { Label: pair.ThisMachineLabel(), // EVERY LINE OF THE PAIRING GOES TO STDERR, so that `--once` piped into // a file is the reply and nothing else. - Say: func(line string) { fmt.Fprintln(os.Stderr, line) }, + Say: func(line string) { fmt.Fprintln(os.Stderr, line) }, + Sync: syncRelayURL(), } if stdinIsTerminal(os.Stdin) { reach.AskCode = askPairingCode @@ -261,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) + relayAddress := flags.String("via", "", "the sync address to be reachable through; empty reads "+pair.RelayEnv+", then the address this machine syncs through") renamedFlag(flags, "relay", "via") if err := parseCommandFlags(flags, args); err != nil { return err @@ -272,11 +273,11 @@ func runServe(args []string) error { service := strings.TrimSpace(*relayAddress) if service == "" { - service = pair.Relay() + service = pair.Relay(syncRelayURL()) } if service == "" { - // THE HONEST STATE, AND THE ONE MOST PEOPLE WILL MEET. The relay is a - // service and this build does not assume one exists. + // Only a machine whose person turned sync off and named no relay + // gets here; every other machine has sync's address. return errors.New("no sync address is set up on this machine, so there is nowhere to be reachable from — set " + pair.RelayEnv + " to your sync address, or let people in over ssh with `codeaf chat --host` from their side") } diff --git a/cmd/codeaf/chatv3_at_test.go b/cmd/codeaf/chatv3_at_test.go index 68837d13dd..9d275095a9 100644 --- a/cmd/codeaf/chatv3_at_test.go +++ b/cmd/codeaf/chatv3_at_test.go @@ -3,6 +3,10 @@ package main import ( "strings" "testing" + + "github.com/Agent-Field/codeaf/internal/home" + "github.com/Agent-Field/codeaf/internal/pair" + "github.com/Agent-Field/codeaf/internal/syncsetup" ) // The target is read the way the ssh door reads its own, because it is the same @@ -58,12 +62,13 @@ func TestTheFlagsThatBuildASessionCannotTravelOverAt(t *testing.T) { } // WITH NO RELAY SET UP, BOTH DOORS SAY SO IN ONE SENTENCE AND NAME THE ROAD -// THAT DOES WORK. The relay is a service, this build does not assume one is -// running, and a person who meets this must not be left guessing whether their -// network is broken. +// THAT DOES WORK. Sync falls back to the hosted relay, so the only machine +// with no relay is one whose person turned sync off, and they must not be +// left guessing whether their network is broken. func TestWithNoRelayBothDoorsSayWhatIsWrongAndWhatToDoInstead(t *testing.T) { t.Setenv("CODEAF_HOME", t.TempDir()) t.Setenv("CODEAF_RELAY", "") + t.Setenv(syncsetup.URLVar, "off") err := openChatV3At(atLaunch{target: "otter-lamp-42"}) if err == nil { @@ -91,6 +96,39 @@ func TestWithNoRelayBothDoorsSayWhatIsWrongAndWhatToDoInstead(t *testing.T) { } } +// SERVE, --at, PAIRING AND SYNC AGREE ABOUT THE RELAY. A machine whose chats +// sync is reachable through that same relay with nothing else set up, whether +// sync's address came from the variable, a saved pairing or the hosted default. +func TestServeAtPairingAndSyncResolveOneRelay(t *testing.T) { + cases := []struct{ name, variable, saved, want string }{ + {"the hosted default", "", "", syncsetup.HostedRelayURL}, + {"the relay a pairing saved", "", "https://saved.example/fabric", "https://saved.example/fabric"}, + {"the variable", "https://variable.example", "https://saved.example/fabric", "https://variable.example"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + t.Setenv(pair.RelayEnv, "") + t.Setenv(syncsetup.URLVar, c.variable) + if c.saved != "" { + if err := syncsetup.SaveRelayURL(home.Dir(), c.saved); err != nil { + t.Fatal(err) + } + } + mailbox, err := pairMailbox("") + if err != nil { + t.Fatal(err) + } + sync := syncsetup.Resolve(home.Dir()).URL + serve := pair.Relay(syncRelayURL()) + at := pair.RelayFor(pair.Known{}, false, syncRelayURL()) + if sync != c.want || mailbox.URL != c.want || serve != c.want || at != c.want { + t.Fatalf("sync %q, pairing %q, serve %q, --at %q; want all %q", sync, mailbox.URL, serve, at, c.want) + } + }) + } +} + // `codeaf devices` on a machine nothing has paired with is one sentence and an // invitation, not a heading over an empty table. func TestDevicesOnAFreshMachineSaysNothingIsPaired(t *testing.T) { diff --git a/cmd/codeaf/pairroute.go b/cmd/codeaf/pairroute.go index be3d23a4f4..22df5b1bf9 100644 --- a/cmd/codeaf/pairroute.go +++ b/cmd/codeaf/pairroute.go @@ -24,6 +24,16 @@ import ( // wait and this client never cuts a healthy one short. const pairPollTimeout = pairbox.MaxWait + 10*time.Second +// syncRelayURL is the relay this computer syncs through, empty when sync is +// off. `codeaf serve` and `--at` fall back to it, so a machine whose chats sync +// is reachable through the same relay with nothing else set up. +func syncRelayURL() string { + if relay := syncsetup.Resolve(home.Dir()); relay.On() { + return relay.URL + } + return "" +} + // pairMailbox is the relay a pairing runs through: the --relay flag, else the // relay this computer syncs through. A relay named by flag is the one the other // device is told to pass, so it is `Shown`; the default needs no telling. diff --git a/docs/remote-access-testing.md b/docs/remote-access-testing.md index 30ed897545..5c10946dd5 100644 --- a/docs/remote-access-testing.md +++ b/docs/remote-access-testing.md @@ -375,7 +375,7 @@ clear refusal. Run it anyway; the refusals are the deliverable. ```sh codeaf devices # who may reach this machine, and its name -codeaf serve # be reachable (needs CODEAF_RELAY set) +codeaf serve # be reachable (through the relay sync uses, or CODEAF_RELAY) codeaf chat --at otter-lamp-42 # reach a machine by its name ``` diff --git a/internal/manual/chat/reaching-this-machine-without-ssh.md b/internal/manual/chat/reaching-this-machine-without-ssh.md index 7ca5f9f885..8bb0abfffe 100644 --- a/internal/manual/chat/reaching-this-machine-without-ssh.md +++ b/internal/manual/chat/reaching-this-machine-without-ssh.md @@ -34,12 +34,10 @@ laptop$ codeaf chat --at otter-lamp-42 router setting. That machine dials *out* to a relay and stays connected, which is why firewalls and NAT stop mattering. -**This needs a relay service, and there is not one running yet.** Until one is switched -on, `--at` prints: - -``` -no service address is set up on this machine, so --at has nowhere to look for otter-lamp-42 — set CODEAF_RELAY to a service address, or reach that machine with --host over ssh -``` +**There is nothing to set up.** Both machines go through the relay their chats already +sync through: the hosted relay, unless sync was pointed somewhere else with +`CODEAF_SYNC_URL` or by pairing. `CODEAF_RELAY`, or `--via` on `codeaf serve`, picks a +different relay for this door alone, such as one you run yourself. `--host` works today and is not going anywhere. If you can already reach the machine that way, use that; the two doors are compared at the bottom of this page. @@ -65,19 +63,39 @@ 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. `--via` beats the `CODEAF_RELAY` -environment variable. +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. Running it twice on one machine is refused, because there is only one of it: ``` -this machine is already connected to the service as otter-lamp-42 — there is only one of it, so close the other `codeaf serve` +this machine is already connected to sync as otter-lamp-42 — there is only one of it, so close the other `codeaf serve` +``` + +## codeaf serve when the network drops or the relay says no + +If the network drops while it runs, it says once: + +``` +lost sync — trying again ``` -With no relay set up at all it says: +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. + +A relay that answers but cannot connect machines, such as an older sync service, stops +it with: ``` -no service address is set up on this machine, so there is nowhere to be reachable from — set CODEAF_RELAY to your service address, or let people in over ssh with `codeaf chat --host` from their side +the service at https://service.example.com syncs chats but does not connect machines — run your own with `codeaf relay` and set CODEAF_RELAY to it, or reach that machine with --host over ssh +``` + +A relay that cannot be reached when it starts stops it with the sentence in *When --at +will not open*. With sync turned off (`CODEAF_SYNC_URL=off`) and no `CODEAF_RELAY`, there +is no relay at all: + +``` +no sync address is set up on this machine, so there is nowhere to be reachable from — set CODEAF_RELAY to your sync address, or let people in over ssh with `codeaf chat --host` from their side ``` Ctrl+c ends it, and the machine gives its name back on the way out: @@ -351,9 +369,9 @@ that still has it in the list, which is what revoking removes. ## When --at will not open -There are four different reasons, and each says which one it is, in one sentence. +There are five different reasons, and each says which one it is, in one sentence. -**No relay is set up on this machine:** +**No relay is set up on this machine** (sync is off and `CODEAF_RELAY` is empty): ``` no service address is set up on this machine, so --at has nowhere to look for otter-lamp-42 — set CODEAF_RELAY to a service address, or reach that machine with --host over ssh @@ -365,6 +383,12 @@ no service address is set up on this machine, so --at has nowhere to look for ot the service at https://service.example.com cannot be reached from here — check this machine's network, or reach that machine with --host over ssh ``` +**The relay answers but cannot connect machines:** + +``` +the service at https://service.example.com syncs chats but does not connect machines — run your own with `codeaf relay` and set CODEAF_RELAY to it, or reach that machine with --host over ssh +``` + **The relay is fine and that machine is not connected to it:** ``` @@ -464,15 +488,18 @@ sitting at. They differ only in how the two halves reach each other. | | `--host` | `--at` | | --- | --- | --- | -| Needs | `ssh ` already works | a relay service, and one pairing | +| Needs | `ssh ` already works | one pairing; the relay is the one sync uses | | Set up on the far machine | nothing | `codeaf serve` running | | Works through NAT and firewalls | only if ssh does | yes — that machine dials out | | Anything in the middle | nothing of ours | a relay that cannot read the conversation | -| Available today | yes | not until a relay is running | +| Available today | yes | yes | **Prefer `--host` when ssh works.** It has nothing between the two machines, nothing to set up, and no service to depend on. `--at` is for the machine ssh cannot reach. +Both machines must use the same relay. If one syncs through a relay of your own and the +other through the hosted one, set `CODEAF_RELAY` to the same address on both. + Both open the same surface with the same limits — what does and does not work over a connection is a property of the connection, not of which of these two doors opened it. diff --git a/internal/pair/dial.go b/internal/pair/dial.go index f7ad76784e..304b897dd8 100644 --- a/internal/pair/dial.go +++ b/internal/pair/dial.go @@ -72,6 +72,9 @@ type Reach struct { AskCode func(name string) (string, error) // Say is where the pairing's own lines go. Nil says nothing. Say func(string) + // Sync is the relay this device syncs through, used when nothing more + // particular names one; see [Relay]. + Sync string // Now is the clock, swapped by tests. Now func() time.Time } @@ -102,7 +105,7 @@ func (r Reach) Open(ctx context.Context) (*Tunnel, error) { if err != nil { return nil, err } - service := RelayFor(known, found) + service := RelayFor(known, found, r.Sync) if strings.TrimSpace(service) == "" { return nil, NoRelay(r.Name) } @@ -204,6 +207,8 @@ func (r Reach) dial(ctx context.Context, service string) (io.ReadWriteCloser, er return nil, NotConnected(r.Name) case errors.Is(err, relay.ErrUnreachable): return nil, Unreachable(service) + case errors.Is(err, relay.ErrNoDoor): + return nil, NoDoor(service) case errors.Is(err, relay.ErrTooMany): return nil, Busy() default: diff --git a/internal/pair/edge_serve_test.go b/internal/pair/edge_serve_test.go index 088eea0be0..a7e93f9796 100644 --- a/internal/pair/edge_serve_test.go +++ b/internal/pair/edge_serve_test.go @@ -8,12 +8,18 @@ package pair import ( "context" "fmt" + "net" + "net/http" + "net/http/httptest" "path/filepath" "regexp" "strings" "sync" + "sync/atomic" "testing" "time" + + "github.com/Agent-Field/codeaf/internal/relay" ) // screenLog is everything the machine has said, in order. @@ -267,3 +273,109 @@ 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. +func TestServeSaysOnceThatItLostSyncAndComesBack(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + var down atomic.Bool + door := &relay.Server{} + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if down.Load() { + http.Error(w, "error code: 502", http.StatusBadGateway) + return + } + door.ServeHTTP(w, r) + })) + wire := &cuttable{Listener: server.Listener} + server.Listener = wire + server.Start() + t.Cleanup(server.Close) + screen := &screenLog{} + host := &Host{Service: server.URL, Device: aDevice(t), Devices: BookAt(filepath.Join(t.TempDir(), "devices.json")), Desk: &Desk{}, Say: screen.say} + 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 }) + + down.Store(true) + 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 }) + lost := 0 + for _, line := range lines { + if line == "lost sync — trying again" { + lost++ + } + } + if lost != 1 { + t.Fatalf("the machine said it lost sync %d times over one outage; want once: %q", lost, lines) + } +} + +// 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. +func TestServeWaitsOutItsOwnStaleName(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + door := &relay.Server{} + server := httptest.NewUnstartedServer(door) + wire := &cuttable{Listener: server.Listener} + server.Listener = wire + server.Start() + t.Cleanup(server.Close) + screen := &screenLog{} + device := aDevice(t) + host := &Host{Service: server.URL, Device: device, Devices: BookAt(filepath.Join(t.TempDir(), "devices.json")), Desk: &Desk{}, Say: screen.say} + ctx, stop := context.WithCancel(context.Background()) + ended := make(chan struct{}) + var runErr error + go func() { runErr = 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 }) + + wire.cut() + stale, err := relay.Register(context.Background(), server.URL, device.Private()) + if err != nil { + t.Fatal(err) + } + time.Sleep(3 * time.Second) + _ = stale.Close() + select { + case <-ended: + 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 }) +} + +// cuttable is a listener whose connections a test can drop, hijacked ones +// included, the way a network that goes away drops them. +type cuttable struct { + net.Listener + mu sync.Mutex + conns []net.Conn +} + +func (c *cuttable) Accept() (net.Conn, error) { + conn, err := c.Listener.Accept() + if err == nil { + c.mu.Lock() + c.conns = append(c.conns, conn) + c.mu.Unlock() + } + return conn, err +} + +func (c *cuttable) cut() { + c.mu.Lock() + defer c.mu.Unlock() + for _, conn := range c.conns { + _ = conn.Close() + } + c.conns = nil +} diff --git a/internal/pair/errors.go b/internal/pair/errors.go index 7c901c9091..8256611345 100644 --- a/internal/pair/errors.go +++ b/internal/pair/errors.go @@ -45,6 +45,12 @@ func Unreachable(service string) error { return fmt.Errorf("the service at %s cannot be reached from here — check this machine's network, or reach that machine with --host over ssh", service) } +// NoDoor is a service that answers but cannot connect machines: one that only +// syncs chats, or one from before machines could register with it. +func NoDoor(service string) error { + return fmt.Errorf("the service at %s syncs chats but does not connect machines — run your own with `codeaf relay` and set %s to it, or reach that machine with --host over ssh", service, RelayEnv) +} + // NotConnected is the third: the relay is fine and that machine is not there. func NotConnected(name string) error { return fmt.Errorf("%s is not connected to the service right now — run `codeaf serve` on that machine", name) diff --git a/internal/pair/serve.go b/internal/pair/serve.go index 581dfd4fae..e13fad3c21 100644 --- a/internal/pair/serve.go +++ b/internal/pair/serve.go @@ -81,10 +81,20 @@ func (h *Host) Run(ctx context.Context) error { } wait := time.Second announced := false + // lost is an outage already told: one line per outage, not one per retry. + lost := false for { registration, err := relay.Register(ctx, h.Service, h.Device.Private()) switch { case err == nil: + 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 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 @@ -95,7 +105,13 @@ func (h *Host) Run(ctx context.Context) error { if !announced { return Unreachable(h.Service) } - h.say("lost sync — trying again") + if !lost { + h.say("lost sync — trying again") + } + lost = true + case errors.Is(err, relay.ErrNoDoor): + // Trying again cannot add a door the service does not have. + return NoDoor(h.Service) default: if !announced { return err @@ -105,6 +121,7 @@ func (h *Host) Run(ctx context.Context) error { if err == nil { wait = time.Second + lost = false if !announced { announced = true } diff --git a/internal/pair/service.go b/internal/pair/service.go index a1801ea64c..98d7f40004 100644 --- a/internal/pair/service.go +++ b/internal/pair/service.go @@ -2,12 +2,12 @@ package pair // Which relay this machine uses. // -// THERE IS NO BUILT-IN DEFAULT, AND THAT IS DELIBERATE. The relay service is -// not deployed; a default pointing at an address that does not answer would -// turn "nothing is set up" — which has a next step — into "the network is -// broken", which does not. So an unset relay is its own state with its own -// sentence ([NoRelay]), and the day the service goes live is the day a default -// belongs in this file and not before. +// THE RELAY A MACHINE IS REACHED THROUGH IS THE ONE IT SYNCS THROUGH unless +// somebody said otherwise. Sync already found an address — the person's +// variable, the relay a pairing saved, or the hosted default — and a second +// answer here would be a machine whose chats sync while `codeaf serve` says +// there is nowhere to be reachable from. So the caller hands in sync's answer +// and it is the last word, after the overrides this package has always read. // // A MACHINE REMEMBERS THE RELAY IT WAS PAIRED THROUGH, so a device that has // used two of them reaches each machine through the right one without anybody @@ -30,27 +30,27 @@ const RelayEnv = "CODEAF_RELAY" // a shell. func RelayFile() string { return filepath.Join(Dir(), "relay") } -// Relay is the address this machine uses when nothing else says otherwise. +// Relay is the address this machine uses: the override, the saved file, then +// sync, the address this machine syncs through (empty when sync is off). // Empty means none is set up. -func Relay() string { +func Relay(sync string) string { if set := strings.TrimSpace(env.Get(RelayEnv)); set != "" { return set } - raw, err := os.ReadFile(RelayFile()) - if err != nil { - return "" + if raw, err := os.ReadFile(RelayFile()); err == nil && strings.TrimSpace(string(raw)) != "" { + return strings.TrimSpace(string(raw)) } - return strings.TrimSpace(string(raw)) + return strings.TrimSpace(sync) } // RelayFor is the address to reach one machine through: the override, then the -// relay that machine was paired through, then the configured one. -func RelayFor(known Known, found bool) string { +// relay that machine was paired through, then [Relay]. +func RelayFor(known Known, found bool, sync string) string { if set := strings.TrimSpace(env.Get(RelayEnv)); set != "" { return set } if found && strings.TrimSpace(known.Service) != "" { return strings.TrimSpace(known.Service) } - return Relay() + return Relay(sync) } diff --git a/internal/relay/client.go b/internal/relay/client.go index dcb5a4334a..2482db04fd 100644 --- a/internal/relay/client.go +++ b/internal/relay/client.go @@ -26,6 +26,8 @@ import ( "net/url" "strings" "time" + + "github.com/Agent-Field/codeaf/internal/wireauth" ) // dialWithin is replaceable by the dead-relay test because WSL2 mirrored @@ -45,6 +47,9 @@ var ( ErrNameTaken = errors.New("relay: that name is already connected") // ErrTooMany is a rate limit, either door. ErrTooMany = errors.New("relay: too many connections from here") + // ErrNoDoor is a relay that answers but offers no machine door: a sync + // service that only stores, or one from before machines could register. + ErrNoDoor = errors.New("relay: that service does not offer machine connections") ) // Registration is a machine's held connection to the relay and the surfaces @@ -93,6 +98,9 @@ func Register(ctx context.Context, service string, private *ecdh.PrivateKey) (*R carrier.onFrame = func() { _ = conn.SetReadDeadline(time.Now().Add(IdleAfter)) } _ = conn.SetReadDeadline(time.Now().Add(IdleAfter)) go carrier.pump() + if alive, ok := conn.(pinger); ok { + go proveAlive(carrier, alive) + } return &Registration{name: name, carrier: carrier, conn: conn}, nil } @@ -139,12 +147,33 @@ func Dial(ctx context.Context, service, name string) (io.ReadWriteCloser, error) } // upgrade dials the service and takes the connection over. +// +// THE CARRIER RIDES A WEBSOCKET FIRST, because a WebSocket is the one upgrade +// every front end passes: the hosted relay is a Cloudflare Worker, which +// answers no other. A relay from before that door answers a WebSocket with +// 426, and this end then takes the raw upgrade it always spoke, so a +// self-hosted relay keeps working without being upgraded. func upgrade(ctx context.Context, service, path string, header http.Header) (net.Conn, error) { address, err := url.Parse(strings.TrimSpace(service)) if err != nil || address.Host == "" || (address.Scheme != "http" && address.Scheme != "https") { 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. + door := *address + door.Path = wireauth.Endpoint(address.Path, path) + door.RawPath, door.RawQuery, door.Fragment = "", "", "" + + conn, err := upgradeWebSocket(ctx, &door, path, header) + if !errors.Is(err, errRawOnly) { + return conn, err + } + return upgradeRaw(ctx, &door, 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) { dialer := &net.Dialer{Timeout: dialWithin} conn, err := dialer.DialContext(ctx, "tcp", hostPort(address)) if err != nil { @@ -161,7 +190,7 @@ func upgrade(ctx context.Context, service, path string, header http.Header) (net request := &http.Request{ Method: http.MethodGet, - URL: &url.URL{Path: path}, + URL: &url.URL{Path: address.Path}, Host: address.Host, Header: header.Clone(), Proto: "HTTP/1.1", ProtoMajor: 1, ProtoMinor: 1, @@ -183,17 +212,7 @@ func upgrade(ctx context.Context, service, path string, header http.Header) (net if response.StatusCode != http.StatusSwitchingProtocols { said := strings.TrimSpace(readBounded(response.Body)) _ = conn.Close() - switch response.StatusCode { - case http.StatusNotFound: - return nil, ErrNoMachine - case http.StatusTooManyRequests: - return nil, ErrTooMany - default: - if said == "" { - said = response.Status - } - return nil, errors.New(said) - } + return nil, refusal(address.Host, path, response.StatusCode, said) } _ = conn.SetDeadline(time.Time{}) // A 101 has no body, so there should be nothing left in the reader — but @@ -205,6 +224,29 @@ func upgrade(ctx context.Context, service, path string, header http.Header) (net return conn, nil } +// 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 { + 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: + return ErrNoMachine + case status == http.StatusTooManyRequests: + 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 with %d", host, status) + default: + return errors.New(said) + } +} + // joinedConn is a connection whose first bytes were already read into a buffer. type joinedConn struct { net.Conn diff --git a/internal/relay/hosted_test.go b/internal/relay/hosted_test.go new file mode 100644 index 0000000000..5ffdfda348 --- /dev/null +++ b/internal/relay/hosted_test.go @@ -0,0 +1,86 @@ +package relay + +import ( + "bytes" + "context" + "crypto/rand" + "errors" + "io" + "os" + "testing" + "time" +) + +// hostedRelayVar names a hosted relay to hold these tests to: `wrangler dev` of +// relay/hosted, or a deployment. Unset, they skip. +const hostedRelayVar = "CODEAF_RELAY_HOSTED_URL" + +func hostedRelay(t *testing.T) string { + t.Helper() + address := os.Getenv(hostedRelayVar) + if address == "" { + t.Skip(hostedRelayVar + " is not set") + } + return address +} + +// The hosted relay carries a machine the way `codeaf relay` does: a round trip, +// a payload bigger than one frame, and several surfaces on one carrier. +func TestAHostedRelayCarriesAMachine(t *testing.T) { + address := hostedRelay(t) + reachAndEcho(t, address) + + registration, err := Register(context.Background(), address, aKey(t)) + if err != nil { + t.Fatal(err) + } + defer func() { _ = registration.Close() }() + big := make([]byte, 3*MaxPayload+17) + _, _ = rand.Read(big) + for i := 0; i < 3; i++ { + surface, err := Dial(context.Background(), address, registration.Name()) + if err != nil { + t.Fatal(err) + } + engine, err := registration.Accept() + if err != nil { + t.Fatal(err) + } + go func() { _, _ = io.Copy(engine, engine) }() + go func() { _, _ = surface.Write(big) }() + got := make([]byte, len(big)) + if _, err := io.ReadFull(surface, got); err != nil || !bytes.Equal(got, big) { + t.Fatalf("surface %d: echo of %d bytes came back wrong: %v", i, len(big), err) + } + _ = surface.Close() + } +} + +// Rule three holds on the hosted relay: a live registration is not taken over, +// and once it ends the name is free again at once. +func TestAHostedRelayKeepsANameForItsLiveMachine(t *testing.T) { + address := hostedRelay(t) + key := aKey(t) + first, err := Register(context.Background(), address, key) + if err != nil { + t.Fatal(err) + } + if _, err := Register(context.Background(), address, key); !errors.Is(err, ErrNameTaken) { + t.Fatalf("a second registration under a live name gave %v; want ErrNameTaken", err) + } + _ = first.Close() + var again *Registration + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(200 * time.Millisecond) { + if again, err = Register(context.Background(), address, key); err == nil { + break + } + } + if err != nil { + t.Fatalf("the name was not free after its machine left: %v", err) + } + _ = again.Close() + + if _, err := Dial(context.Background(), address, NameFor(aKey(t).PublicKey().Bytes())); !errors.Is(err, ErrNoMachine) { + t.Fatalf("dialling a name nothing holds gave %v; want ErrNoMachine", err) + } +} diff --git a/internal/relay/mux.go b/internal/relay/mux.go index 716c7c939c..e71f2f6b77 100644 --- a/internal/relay/mux.go +++ b/internal/relay/mux.go @@ -46,24 +46,20 @@ const frameHeader = 1 + 8 + 4 // surface's read and a surface's write agree about what happened. var errCarrierGone = errors.New("the connection to the relay ended") -// writeFrame puts one frame on a carrier. Callers hold the carrier's write -// mutex; the mutex is not taken here because a frame is sometimes written as -// part of a longer critical section. +// writeFrame puts one frame on a carrier, in ONE write, so a carrier that is a +// WebSocket carries each frame as one message (see transport.go). Callers hold +// the carrier's write mutex; the mutex is not taken here because a frame is +// sometimes written as part of a longer critical section. func writeFrame(w io.Writer, op byte, stream uint64, payload []byte) error { if len(payload) > MaxPayload { return fmt.Errorf("relay: frame of %d bytes is over the %d-byte limit", len(payload), MaxPayload) } - var header [frameHeader]byte - header[0] = op - binary.BigEndian.PutUint64(header[1:9], stream) - binary.BigEndian.PutUint32(header[9:13], uint32(len(payload))) - if _, err := w.Write(header[:]); err != nil { - return err - } - if len(payload) == 0 { - return nil - } - _, err := w.Write(payload) + frame := make([]byte, frameHeader+len(payload)) + frame[0] = op + binary.BigEndian.PutUint64(frame[1:9], stream) + binary.BigEndian.PutUint32(frame[9:13], uint32(len(payload))) + copy(frame[frameHeader:], payload) + _, err := w.Write(frame) return err } diff --git a/internal/relay/server.go b/internal/relay/server.go index 7c7f4c5f86..da4a560a20 100644 --- a/internal/relay/server.go +++ b/internal/relay/server.go @@ -18,6 +18,7 @@ package relay // test asserts the payloads never appear in it. import ( + "context" "errors" "io" "net" @@ -25,6 +26,8 @@ import ( "strings" "sync" "time" + + "github.com/coder/websocket" ) // Ledger is everything the relay knows about one connected machine. IT IS THE @@ -118,7 +121,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { // serveEngine takes a machine's registration and then holds its carrier. func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { - if !strings.EqualFold(r.Header.Get("Upgrade"), Protocol) { + if !upgradable(r) { http.Error(w, "this door speaks "+Protocol, http.StatusUpgradeRequired) return } @@ -142,15 +145,11 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { return } - conn, err := hijack(w) + conn, err := takeOver(w, r) if err != nil { - http.Error(w, "this server cannot hold a connection open", http.StatusInternalServerError) return } defer func() { _ = conn.Close() }() - if err := writeUpgraded(conn); err != nil { - return - } // RULE TWO: possession is proved. A machine that cannot answer the // challenge never reaches the table. @@ -208,7 +207,7 @@ func (s *Server) serveEngine(w http.ResponseWriter, r *http.Request) { // serveDial matches a surface to a machine and then gets out of the way. func (s *Server) serveDial(w http.ResponseWriter, r *http.Request, name string) { - if !strings.EqualFold(r.Header.Get("Upgrade"), Protocol) { + if !upgradable(r) { http.Error(w, "this door speaks "+Protocol, http.StatusUpgradeRequired) return } @@ -240,15 +239,11 @@ func (s *Server) serveDial(w http.ResponseWriter, r *http.Request, name string) } defer func() { _ = stream.Close() }() - conn, err := hijack(w) + conn, err := takeOver(w, r) if err != nil { - http.Error(w, "this server cannot hold a connection open", http.StatusInternalServerError) return } defer func() { _ = conn.Close() }() - if err := writeUpgraded(conn); err != nil { - return - } s.note(name, "a surface arrived") defer s.note(name, "a surface left") @@ -279,6 +274,42 @@ func (r *registration) add(up, down int64) { r.mu.Unlock() } +// upgradable is a request for either carrier: the raw upgrade, or a WebSocket +// offering this protocol. +func upgradable(r *http.Request) bool { + if strings.EqualFold(r.Header.Get("Upgrade"), Protocol) { + return true + } + return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && strings.Contains(r.Header.Get("Sec-WebSocket-Protocol"), Protocol) +} + +// 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) { + if strings.EqualFold(r.Header.Get("Upgrade"), "websocket") { + socket, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + Subprotocols: []string{Protocol}, + CompressionMode: websocket.CompressionDisabled, + }) + if err != nil { + return nil, err + } + socket.SetReadLimit(messageLimit) + return websocket.NetConn(context.Background(), socket, websocket.MessageBinary), nil + } + conn, err := hijack(w) + if err != nil { + http.Error(w, "this server cannot hold a connection open", http.StatusInternalServerError) + return nil, err + } + if err := writeUpgraded(conn); err != nil { + _ = conn.Close() + return nil, err + } + return conn, nil +} + // hijack takes the raw connection out from under the HTTP server. func hijack(w http.ResponseWriter) (net.Conn, error) { hijacker, ok := w.(http.Hijacker) diff --git a/internal/relay/transport.go b/internal/relay/transport.go new file mode 100644 index 0000000000..26436225eb --- /dev/null +++ b/internal/relay/transport.go @@ -0,0 +1,103 @@ +package relay + +// The WebSocket carrier. It carries exactly the bytes the raw upgrade carries — +// the registration proof, then the frames of mux.go — so nothing above the +// connection knows which one it got. Each frame is one message, because +// [writeFrame] writes a frame in one call; that lets a relay that sees +// messages and not a byte stream, such as the hosted one, read a frame without +// buffering half of one across a wake-up. + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "strings" + "time" + + "github.com/coder/websocket" +) + +// errRawOnly is a relay that answered a WebSocket with 426: one from before +// this door, which only speaks the raw upgrade. +var errRawOnly = errors.New("relay: that relay only speaks the raw upgrade") + +// messageLimit is the largest message either end accepts: one full frame. +const messageLimit = frameHeader + MaxPayload + +// wsConn is a WebSocket presented as the byte stream the carrier reads, with +// the protocol-level ping a carrier uses to keep a quiet connection proven. +type wsConn struct { + net.Conn + socket *websocket.Conn +} + +// ping is answered by the far end's WebSocket layer, never by the relay's own +// code, so a hosted relay that is asleep answers it without waking. +func (w *wsConn) ping(ctx context.Context) error { return w.socket.Ping(ctx) } + +// pinger is a carrier connection that can prove itself alive from this end. +type pinger interface { + ping(ctx context.Context) error +} + +func upgradeWebSocket(ctx context.Context, address *url.URL, path string, header http.Header) (net.Conn, error) { + dialer := &net.Dialer{Timeout: dialWithin} + client := &http.Client{Transport: &http.Transport{DialContext: dialer.DialContext, ForceAttemptHTTP2: false}} + handshake, cancel := context.WithTimeout(ctx, HandshakeWithin) + defer cancel() + socket, response, err := websocket.Dial(handshake, address.String(), &websocket.DialOptions{ + HTTPClient: client, + HTTPHeader: header.Clone(), + Subprotocols: []string{Protocol}, + CompressionMode: websocket.CompressionDisabled, // the payloads are ciphertext + }) + if err != nil { + if response == nil { + return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) + } + 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, refusal(address.Host, path, response.StatusCode, said) + } + socket.SetReadLimit(messageLimit) + // 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 +} + +// proveAlive pings a carrier from this end every [PingEvery] and counts an +// answer as an arrival. A relay that never sends its own pings — the hosted one +// stays asleep rather than wake to send them — is then still a carrier whose +// silence is tested. +func proveAlive(c *carrier, p pinger) { + ticker := time.NewTicker(PingEvery) + defer ticker.Stop() + for { + select { + case <-c.done: + return + case <-ticker.C: + ctx, cancel := context.WithTimeout(context.Background(), HandshakeWithin) + err := p.ping(ctx) + cancel() + if err != nil { + c.shutdown(errCarrierGone) + return + } + if c.onFrame != nil { + c.onFrame() + } + } + } +} diff --git a/internal/relay/transport_test.go b/internal/relay/transport_test.go new file mode 100644 index 0000000000..a39ecce4bf --- /dev/null +++ b/internal/relay/transport_test.go @@ -0,0 +1,126 @@ +package relay + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// mounted serves a relay beneath base, the way the hosted relay lives at +// https://host/fabric, and refuses every path outside it as that relay does. +func mounted(t *testing.T, base string, door http.Handler) string { + t.Helper() + mux := http.NewServeMux() + mux.Handle(base+"/", http.StripPrefix(base, door)) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + t.Cleanup(server.CloseClientConnections) + return server.URL + base +} + +// reachAndEcho registers a machine at address, dials it, and checks one round +// trip, which is the whole of what a person needs from the relay. +func reachAndEcho(t *testing.T, address string) { + t.Helper() + registration, err := Register(context.Background(), address, aKey(t)) + if err != nil { + t.Fatalf("Register(%s): %v", address, err) + } + defer func() { _ = registration.Close() }() + surface, err := Dial(context.Background(), address, registration.Name()) + if err != nil { + t.Fatalf("Dial(%s): %v", address, err) + } + defer func() { _ = surface.Close() }() + engine, err := registration.Accept() + if err != nil { + t.Fatal(err) + } + defer func() { _ = engine.Close() }() + if _, err := surface.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + if got := readN(t, engine, 4); got != "ping" { + t.Fatalf("the machine read %q", got) + } + if _, err := engine.Write([]byte("pong")); err != nil { + t.Fatal(err) + } + if got := readN(t, surface, 4); got != "pong" { + t.Fatalf("the surface read %q", got) + } +} + +// A relay address keeps its path, as every other client's does: a relay at +// https://host/fabric is called at /fabric/v1/engine and /fabric/v1/dial/. +func TestARelayAddressKeepsItsPath(t *testing.T) { + for _, base := range []string{"/fabric", "/a/relay"} { + for _, slash := range []string{"", "/"} { + reachAndEcho(t, mounted(t, base, &Server{})+slash) + } + } + _, root := live(t) + reachAndEcho(t, root) + reachAndEcho(t, root+"/") +} + +// rawOnly is a relay from before the WebSocket door: it answers a WebSocket +// upgrade with 426, as every earlier `codeaf relay` does. +func rawOnly(door http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.EqualFold(r.Header.Get("Upgrade"), "websocket") { + http.Error(w, "this door speaks "+Protocol, http.StatusUpgradeRequired) + return + } + door.ServeHTTP(w, r) + }) +} + +// wsOnly is a front end that passes WebSocket upgrades and nothing else, which +// is what a Cloudflare Worker is. +func wsOnly(door http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") { + http.Error(w, "error code: 522", 522) + return + } + door.ServeHTTP(w, r) + }) +} + +// The carrier rides a WebSocket, which is the one upgrade every front end +// passes, and still reaches a relay that only speaks the raw upgrade. +func TestTheCarrierCrossesAWebSocketOnlyFrontAndAnOlderRelay(t *testing.T) { + t.Run("websocket only", func(t *testing.T) { + reachAndEcho(t, mounted(t, "/fabric", wsOnly(&Server{}))) + }) + t.Run("raw only", func(t *testing.T) { + reachAndEcho(t, mounted(t, "/fabric", rawOnly(&Server{}))) + }) +} + +// An upstream proxy's body is not the relay's sentence. A relay that is down +// is unreachable; one that has no machine door says so; neither is printed as +// "error code: 522". +func TestUpstreamFailuresBecomeFacts(t *testing.T) { + down := mounted(t, "/fabric", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "error code: 522", 522) + })) + _, err := Register(context.Background(), down, aKey(t)) + if !errors.Is(err, ErrUnreachable) || strings.Contains(err.Error(), "error code") { + t.Fatalf("a relay answering 522: %v; want ErrUnreachable without the proxy's body", err) + } + + noDoor := mounted(t, "/fabric", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"err":"not_found"}`)) + })) + _, err = Register(context.Background(), noDoor, aKey(t)) + if !errors.Is(err, ErrNoDoor) || strings.Contains(err.Error(), "not_found") { + t.Fatalf("a relay with no machine door: %v; want ErrNoDoor", err) + } +} diff --git a/relay/hosted/src/engine/carrier.js b/relay/hosted/src/engine/carrier.js new file mode 100644 index 0000000000..21d5610446 --- /dev/null +++ b/relay/hosted/src/engine/carrier.js @@ -0,0 +1,72 @@ +// The carrier's bytes, shared by the machine object and its tests: frames (mux.go), the registration verdict and the +// possession proof (handshake.go). Nothing here reads a payload. + +/** PROTOCOL is the upgrade token both ends spell; an on-the-wire identifier, not product prose. */ +export const PROTOCOL = 'aforge-relay/1'; // legacy-name +/** The headers of a registration: the name claimed and the machine's public key, base64 raw-url. */ +export const HEADER_NAME = 'aforge-name'; // legacy-name: persisted on the wire. +export const HEADER_KEY = 'aforge-key'; // legacy-name: persisted on the wire. + +const LABEL = new TextEncoder().encode('aforge relay registration v1'); // legacy-name + +// The tuning of relay.go, in the same names. +export const MAX_PAYLOAD = 1 << 17; +export const MAX_STREAMS = 16; +const FRAME_HEADER = 13; + +export const OP = { open: 1, data: 2, close: 3, ping: 4, pong: 5 }; +const VERDICT_READY = 1; +const VERDICT_REFUSED = 2; + + +/** frame is one carrier frame: an op byte, an 8-byte stream id, a 4-byte length, the payload. */ +export function frame(op, stream, payload = new Uint8Array(0)) { + const out = new Uint8Array(FRAME_HEADER + payload.length); + const view = new DataView(out.buffer); + out[0] = op; + view.setBigUint64(1, BigInt(stream)); + view.setUint32(9, payload.length); + out.set(payload, FRAME_HEADER); + return out; +} + +/** readFrame parses one message as one frame, or answers null for a message that is not exactly one. */ +export function readFrame(bytes) { + if (bytes.length < FRAME_HEADER) return null; + const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength); + const length = view.getUint32(9); + if (length > MAX_PAYLOAD || bytes.length !== FRAME_HEADER + length) return null; + return { op: bytes[0], stream: view.getBigUint64(1).toString(), payload: bytes.subarray(FRAME_HEADER) }; +} + +/** verdict is the relay's last word on a registration: ready, or refused with a sentence. */ +export function verdict(reason = '') { + if (!reason) return new Uint8Array([VERDICT_READY, 0]); + const said = new TextEncoder().encode(reason).subarray(0, 255); + const out = new Uint8Array(2 + said.length); + out[0] = VERDICT_REFUSED; + out[1] = said.length; + out.set(said, 2); + return out; +} + +/** registrationProof is the MAC the machine must answer a challenge with (handshake.go's registrationProof). */ +export async function registrationProof(shared, nonce, name, publicKey) { + const ikm = await crypto.subtle.importKey('raw', shared, 'HKDF', false, ['deriveBits']); + const key = await crypto.subtle.deriveBits({ name: 'HKDF', hash: 'SHA-256', salt: nonce, info: LABEL }, ikm, 256); + const mac = await crypto.subtle.importKey('raw', key, { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']); + const body = new Uint8Array([...new TextEncoder().encode(name), ...publicKey]); + return new Uint8Array(await crypto.subtle.sign('HMAC', mac, body)); +} + +/** challenge mints the relay's half of the proof: the 64 bytes it sends, and the proof it then expects. */ +export async function challenge(name, publicKey) { + const ephemeral = await crypto.subtle.generateKey({ name: 'X25519' }, true, ['deriveBits']); + const offered = await crypto.subtle.importKey('raw', publicKey, { name: 'X25519' }, false, []); + const shared = new Uint8Array(await crypto.subtle.deriveBits({ name: 'X25519', public: offered }, ephemeral.privateKey, 256)); + const nonce = crypto.getRandomValues(new Uint8Array(32)); + const sent = new Uint8Array(64); + sent.set(new Uint8Array(await crypto.subtle.exportKey('raw', ephemeral.publicKey))); + sent.set(nonce, 32); + return { sent, want: await registrationProof(shared, nonce, name, publicKey) }; +} diff --git a/relay/hosted/src/engine/door.js b/relay/hosted/src/engine/door.js new file mode 100644 index 0000000000..01ae2c2c18 --- /dev/null +++ b/relay/hosted/src/engine/door.js @@ -0,0 +1,29 @@ +// The front door's half of the machine doors: refuse what needs no object to refuse, then hand the upgrade to the +// one object that holds that machine name (machine.js). +import { validName } from './names.js'; +import { HEADER_NAME, PROTOCOL } from './carrier.js'; + +export const ENGINE_PATH = '/v1/engine'; +export const DIAL_PREFIX = '/v1/dial/'; + +/** isMachineDoor is a path this module serves. */ +export const isMachineDoor = (pathname) => pathname === ENGINE_PATH || pathname.startsWith(DIAL_PREFIX); + +const say = (status, sentence) => new Response(sentence, { status, headers: { 'content-type': 'text/plain; charset=utf-8' } }); + +/** + * 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. + */ +export async function serveMachineDoor(request, env) { + const offered = (request.headers.get('sec-websocket-protocol') ?? '').split(',').map((p) => p.trim()); + if ((request.headers.get('upgrade') ?? '').toLowerCase() !== 'websocket' || !offered.includes(PROTOCOL)) { + return say(426, `this door speaks ${PROTOCOL} over a WebSocket`); + } + const { pathname } = new URL(request.url); + const name = pathname === ENGINE_PATH ? request.headers.get(HEADER_NAME) ?? '' : decodeURIComponent(pathname.slice(DIAL_PREFIX.length)); + if (!validName(name)) return say(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. + return new Response(answer.body, { status: answer.status, headers: answer.headers, webSocket: answer.webSocket }); +} diff --git a/relay/hosted/src/engine/machine.js b/relay/hosted/src/engine/machine.js new file mode 100644 index 0000000000..4430b4d451 --- /dev/null +++ b/relay/hosted/src/engine/machine.js @@ -0,0 +1,194 @@ +// 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, and the object sets no timer. +import { DurableObject } from 'cloudflare:workers'; +import { nameFor } from './names.js'; +import { MAX_PAYLOAD, MAX_STREAMS, OP, PROTOCOL, HEADER_NAME, HEADER_KEY, challenge, frame, 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)); +const refuse = (status, sentence) => new Response(sentence, { status, headers: { 'content-type': 'text/plain; charset=utf-8' } }); + +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(); + } + + async fetch(request) { + const path = new URL(request.url).pathname; + return path === '/v1/engine' ? this.#register(request) : this.#dial(); + } + + /** 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(503, `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) surface.close(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'); + } + const older = this.#engine(ws); + 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()); + } + + /** + * 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 {} + } + + async webSocketError(ws) { + this.#gone(ws); + } + + /** 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'); + } + } + + #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/src/engine/names.js b/relay/hosted/src/engine/names.js new file mode 100644 index 0000000000..863a7c4760 --- /dev/null +++ b/relay/hosted/src/engine/names.js @@ -0,0 +1,53 @@ +// Machine names, ported from internal/relay/name.go: a name is derived from the machine's own key, so the +// relay can refuse a registration under a name that key does not derive. test/engine.test.js holds the port +// to vectors the Go side computes. + +// NAME_SALT is an on-the-wire identifier, not product prose. +const NAME_SALT = new TextEncoder().encode('aforge relay name\0'); // legacy-name + +const WORDS = [ + 'acorn', 'alder', 'amber', 'anchor', 'apple', 'arbor', 'arrow', 'ash', + 'aspen', 'atlas', 'aurora', 'autumn', 'azure', 'badger', 'bamboo', 'banjo', + 'basil', 'beacon', 'beetle', 'birch', 'bishop', 'bison', 'blossom', 'bluff', + 'bobcat', 'bonfire', 'boulder', 'bramble', 'brass', 'breeze', 'bridge', 'bronze', + 'brook', 'buffalo', 'burrow', 'cabin', 'cactus', 'camel', 'canary', 'candle', + 'canoe', 'canyon', 'cardinal', 'carrot', 'cascade', 'cedar', 'cello', 'chalk', + 'cherry', 'chestnut', 'chimney', 'cinder', 'citrus', 'clover', 'cobalt', 'comet', + 'compass', 'copper', 'coral', 'cottage', 'cougar', 'crane', 'crater', 'cricket', + 'crimson', 'crocus', 'cypress', 'daffodil', 'dahlia', 'daisy', 'dapple', 'dawn', + 'delta', 'denim', 'dingo', 'dolphin', 'domino', 'donkey', 'dragon', 'driftwood', + 'dune', 'eagle', 'ember', 'emerald', 'falcon', 'fennel', 'fern', 'ferry', + 'fiddle', 'finch', 'flint', 'flute', 'forest', 'fossil', 'fountain', 'foxglove', + 'gadget', 'gallery', 'garnet', 'gazelle', 'geode', 'geyser', 'ginger', 'glacier', + 'glider', 'granite', 'grotto', 'guitar', 'gulch', 'gypsum', 'harbor', 'harvest', + 'hazel', 'heather', 'hedge', 'heron', 'hickory', 'hollow', 'honey', 'hornet', + 'ibis', 'iceberg', 'indigo', 'iris', 'island', 'ivory', 'jackal', 'jasmine', + 'jasper', 'jetty', 'jigsaw', 'juniper', 'kayak', 'kelp', 'kestrel', 'kettle', + 'kingfisher', 'kite', 'koala', 'lagoon', 'lamp', 'lantern', 'larch', 'lattice', + 'lavender', 'ledger', 'lemon', 'lichen', 'lighthouse', 'lilac', 'linen', 'lobster', + 'lotus', 'lumber', 'lupine', 'lynx', 'magnet', 'magnolia', 'mahogany', 'mallard', + 'mango', 'maple', 'marble', 'marigold', 'marlin', 'marsh', 'meadow', 'mesa', + 'meteor', 'mica', 'midnight', 'mimosa', 'mineral', 'mint', 'mirror', 'mist', + 'mongoose', 'moonlight', 'moraine', 'mosaic', 'moss', 'mulberry', 'mustang', 'nectar', + 'nettle', 'nimbus', 'nomad', 'nutmeg', 'oasis', 'obsidian', 'ocelot', 'octopus', + 'olive', 'onyx', 'opal', 'orchard', 'orchid', 'oriole', 'osprey', 'otter', + 'paddle', 'pampas', 'pansy', 'papaya', 'parsley', 'pebble', 'pelican', 'pepper', + 'periwinkle', 'petal', 'pewter', 'phoenix', 'piano', 'pigeon', 'pilot', 'pine', + 'pistachio', 'plateau', 'plover', 'plum', 'pollen', 'pomelo', 'poppy', 'porcelain', + 'prairie', 'puffin', 'pumice', 'quail', 'quartz', 'quiver', 'rabbit', 'raccoon', + 'radish', 'rafter', 'rapids', 'raven', 'redwood', 'reef', 'rhubarb', 'ribbon', + 'river', 'robin', 'rosemary', 'rudder', 'saffron', 'sage', 'salmon', 'sandal', + 'sapphire', 'sardine', 'satchel', 'scarlet', 'seagull', 'sequoia', 'shale', 'willow', +]; + +/** nameFor is the name a machine's 32-byte X25519 public key derives. */ +export async function nameFor(publicKey) { + const salted = new Uint8Array(NAME_SALT.length + publicKey.length); + salted.set(NAME_SALT); + salted.set(publicKey, NAME_SALT.length); + const sum = new Uint8Array(await crypto.subtle.digest('SHA-256', salted)); + 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); diff --git a/relay/hosted/src/front.js b/relay/hosted/src/front.js index 1dad964b44..9e7ba58ee5 100644 --- a/relay/hosted/src/front.js +++ b/relay/hosted/src/front.js @@ -8,6 +8,7 @@ import { checkDeclared,readBody } from './body.js'; import { serveLink } from './link/front.js'; import { isPagePath, servePage } from './page.js'; import { servePair } from './pair/front.js'; +import { isMachineDoor, serveMachineDoor } from './engine/door.js'; import { matchRoute } from './routes.js'; import { identityOf } from './verify.js'; import { notFound, respond } from './wire.js'; @@ -36,6 +37,7 @@ export function dispatch(outer, env) { const { pathname } = new URL(request.url); if (pathname.startsWith('/v1/link/')) return serveLink(request, env); if (pathname.startsWith('/v1/pair')) return servePair(request, env); + if (isMachineDoor(pathname)) return serveMachineDoor(request, env); if (pathname.startsWith('/v1/') || pathname.startsWith('/v2/cluster/')) return forward(request, env); throw notFound(); } diff --git a/relay/hosted/src/worker.js b/relay/hosted/src/worker.js index 1b6dc3bfcc..2cfc7fc8f2 100644 --- a/relay/hosted/src/worker.js +++ b/relay/hosted/src/worker.js @@ -12,6 +12,7 @@ export { LinkGate } from './link/gate.js'; export { LinkRequest } from './link/request.js'; export { PairGate } from './pair/gate.js'; export { Mailbox } from './pair/mailbox.js'; +export { MachineDO } from './engine/machine.js'; export default { fetch: (request, env) => respond(() => dispatch(request, env)), diff --git a/relay/hosted/test/config.test.js b/relay/hosted/test/config.test.js index 2cc437d7b8..29baad4c70 100644 --- a/relay/hosted/test/config.test.js +++ b/relay/hosted/test/config.test.js @@ -12,8 +12,10 @@ test('R30 the base config uses the Raft entry and an append-only v3-raft-members assert.match(text, /name = "RAFT_MEMBERS"\s+class_name = "RaftMemberDO"/); assert.match(text, /\[build\]\s+command = "sh \.\.\/\.\.\/scripts\/raft-hosted-build\.sh"/); const tags = [...text.matchAll(/^tag = "([^"]+)"/gm)].map((m) => m[1]); - assert.deepEqual(tags, ['v1', 'v2', 'v3-raft-members']); + assert.deepEqual(tags, ['v1', 'v2', 'v3-raft-members', 'v4-machines']); assert.match(text, /tag = "v3-raft-members"\s+new_sqlite_classes = \["RaftMemberDO"\]/); assert.doesNotMatch(text, /deleted_classes|renamed_classes/); 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"\]/); }); diff --git a/relay/hosted/test/engine.test.js b/relay/hosted/test/engine.test.js new file mode 100644 index 0000000000..d51a07516f --- /dev/null +++ b/relay/hosted/test/engine.test.js @@ -0,0 +1,49 @@ +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'; + +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); + +// 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 () => { + for (const [key, name] of [ + ['000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f', 'lantern-amber-86'], + ['0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20', 'jasmine-dragon-58'], + ['7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e', 'quartz-donkey-28'], + ['ff000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e', 'lavender-rhubarb-13'], + ]) { + 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); +}); + +// Computed by internal/relay's registrationProof over the same shared secret, nonce, name and key. +test('the possession proof is the one the Go side computes', async () => { + const key = run(200); + const proof = await registrationProof(run(0), run(100), 'bluff-meadow-35', key); + assert.equal(toHex(proof), '858616a2eaed21cdcae7e1c815c02ec366f7cc550f63ccf189881aa641feff82'); +}); + +test('a frame is one message and a message that is not exactly one frame is refused', () => { + const sent = frame(2, '7', new Uint8Array([1, 2, 3])); + assert.deepEqual([...sent.subarray(0, 13)], [2, 0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 3]); + const got = readFrame(sent); + assert.equal(got.op, 2); + assert.equal(got.stream, '7'); + assert.deepEqual([...got.payload], [1, 2, 3]); + assert.equal(readFrame(sent.subarray(0, 15)), null); + assert.equal(readFrame(new Uint8Array([...sent, 9])), null); + const huge = new Uint8Array(13); + new DataView(huge.buffer).setUint32(9, MAX_PAYLOAD + 1); + assert.equal(readFrame(huge), null); +}); + +test('a verdict is ready, or refused with its sentence', () => { + assert.deepEqual([...verdict()], [1, 0]); + const said = verdict('no'); + assert.deepEqual([...said], [2, 2, 110, 111]); +}); diff --git a/relay/hosted/wrangler.toml b/relay/hosted/wrangler.toml index c6be7524bf..ce65d46f67 100644 --- a/relay/hosted/wrangler.toml +++ b/relay/hosted/wrangler.toml @@ -33,6 +33,10 @@ class_name = "LinkGate" name = "LINK_REQUEST" class_name = "LinkRequest" +[[durable_objects.bindings]] +name = "MACHINE" +class_name = "MachineDO" + [[migrations]] tag = "v1" new_sqlite_classes = ["IdentityDO", "PairGate", "Mailbox", "NewcomerGate"] @@ -55,3 +59,7 @@ class_name = "RaftMemberDO" [[migrations]] tag = "v3-raft-members" new_sqlite_classes = ["RaftMemberDO"] + +[[migrations]] +tag = "v4-machines" +new_sqlite_classes = ["MachineDO"] From 00eda897ae0f8ca106bc4bdc61572fb5a41a840d Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 30 Oct 2026 10:42:55 -0400 Subject: [PATCH 2/3] fix: address review of the machine doors A full machine answers 429 on both relays, a registration being decided is not taken over, a closed stream is not closed twice, a running serve waits out a redeploying relay, frames reuse one buffer, the door paths are one constant, per-address limits match codeaf relay, and the hosted README lists the binding, the migration and the doors. Co-Authored-By: Claude Opus 5.5 --- .../2073-reach-machines-through-sync-relay.md | 15 +++++ internal/pair/serve.go | 7 +++ internal/pair/service.go | 2 +- internal/relay/client.go | 4 +- internal/relay/hosted_test.go | 5 ++ internal/relay/mux.go | 29 +++++----- internal/relay/server.go | 18 +++++- internal/relay/transport.go | 3 + internal/relay/transport_test.go | 26 +++++++++ relay/hosted/README.md | 25 ++++++++- relay/hosted/src/engine/carrier.js | 9 +++ relay/hosted/src/engine/door.js | 13 ++--- relay/hosted/src/engine/machine.js | 56 +++++++++++++++---- 13 files changed, 170 insertions(+), 42 deletions(-) create mode 100644 docs/changes/unreleased/2073-reach-machines-through-sync-relay.md diff --git a/docs/changes/unreleased/2073-reach-machines-through-sync-relay.md b/docs/changes/unreleased/2073-reach-machines-through-sync-relay.md new file mode 100644 index 0000000000..fff6bd5962 --- /dev/null +++ b/docs/changes/unreleased/2073-reach-machines-through-sync-relay.md @@ -0,0 +1,15 @@ +--- +kind: fixed +title: codeaf serve and chat --at work through the relay your chats already sync through +pr: 2073 +surface: [remote, docs] +invalidates: + - "codeaf serve and chat --at needed CODEAF_RELAY or a saved relay file; with neither they said no sync address was set up. They now fall back to the relay sync resolves (CODEAF_SYNC_URL, the relay a pairing saved, then the hosted relay)." + - "The relay client dropped the path of a relay address, so https://host/fabric was called at https://host/v1/engine. It now calls https://host/fabric/v1/engine and /fabric/v1/dial/, as every other client does." + - "The machine carrier was only the raw aforge-relay/1 HTTP upgrade. It now rides a WebSocket offering that protocol and falls back to the raw upgrade when a relay answers 426; codeaf relay accepts both." + - "The hosted relay had no machine doors and answered /v1/engine with 404. It now serves /v1/engine and /v1/dial/ through one Durable Object per machine name, which needs the v4-machines migration and the MACHINE binding in a deployment's config." + - "codeaf serve said lost sync on every retry and quit when the relay still held its own name from a dropped connection. It now says it once per outage and waits the stale name out." +--- + +A relay that is down, or one that cannot connect machines, is now one sentence naming the relay and what +to try, never an upstream body such as `error code: 522`. diff --git a/internal/pair/serve.go b/internal/pair/serve.go index e13fad3c21..88aaa6998f 100644 --- a/internal/pair/serve.go +++ b/internal/pair/serve.go @@ -109,6 +109,13 @@ func (h *Host) Run(ctx context.Context) error { h.say("lost sync — trying again") } lost = true + 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 case errors.Is(err, relay.ErrNoDoor): // Trying again cannot add a door the service does not have. return NoDoor(h.Service) diff --git a/internal/pair/service.go b/internal/pair/service.go index 98d7f40004..f0baa641c2 100644 --- a/internal/pair/service.go +++ b/internal/pair/service.go @@ -2,7 +2,7 @@ package pair // Which relay this machine uses. // -// THE RELAY A MACHINE IS REACHED THROUGH IS THE ONE IT SYNCS THROUGH unless +// The relay a machine is reached through is the one it syncs through, unless // somebody said otherwise. Sync already found an address — the person's // variable, the relay a pairing saved, or the hosted default — and a second // answer here would be a machine whose chats sync while `codeaf serve` says diff --git a/internal/relay/client.go b/internal/relay/client.go index 2482db04fd..687a6464c3 100644 --- a/internal/relay/client.go +++ b/internal/relay/client.go @@ -148,7 +148,7 @@ func Dial(ctx context.Context, service, name string) (io.ReadWriteCloser, error) // upgrade dials the service and takes the connection over. // -// THE CARRIER RIDES A WEBSOCKET FIRST, because a WebSocket is the one upgrade +// The carrier rides a WebSocket first, because a WebSocket is the one upgrade // every front end passes: the hosted relay is a Cloudflare Worker, which // answers no other. A relay from before that door answers a WebSocket with // 426, and this end then takes the raw upgrade it always spoke, so a @@ -241,7 +241,7 @@ func refusal(host, path string, status int, said string) error { case status >= 500: return fmt.Errorf("%w: %s answered %d", ErrUnreachable, host, status) case said == "": - return fmt.Errorf("the relay at %s refused with %d", host, status) + return fmt.Errorf("the relay at %s refused this connection (%d) — check the address it was given", host, status) default: return errors.New(said) } diff --git a/internal/relay/hosted_test.go b/internal/relay/hosted_test.go index 5ffdfda348..65c76b1da2 100644 --- a/internal/relay/hosted_test.go +++ b/internal/relay/hosted_test.go @@ -84,3 +84,8 @@ func TestAHostedRelayKeepsANameForItsLiveMachine(t *testing.T) { t.Fatalf("dialling a name nothing holds gave %v; want ErrNoMachine", err) } } + +// The stream ceiling reads the same on the hosted relay: busy, not unreachable. +func TestAHostedRelayTurnsAFullMachineAwayAsBusy(t *testing.T) { + fullMachineIsBusy(t, hostedRelay(t)) +} diff --git a/internal/relay/mux.go b/internal/relay/mux.go index e71f2f6b77..1268676de5 100644 --- a/internal/relay/mux.go +++ b/internal/relay/mux.go @@ -46,21 +46,21 @@ const frameHeader = 1 + 8 + 4 // surface's read and a surface's write agree about what happened. var errCarrierGone = errors.New("the connection to the relay ended") -// writeFrame puts one frame on a carrier, in ONE write, so a carrier that is a -// WebSocket carries each frame as one message (see transport.go). Callers hold -// the carrier's write mutex; the mutex is not taken here because a frame is -// sometimes written as part of a longer critical section. -func writeFrame(w io.Writer, op byte, stream uint64, payload []byte) error { +// writeFrame puts one frame on a carrier in one write, so a carrier that is a +// WebSocket carries each frame as one message (see transport.go). The frame is +// assembled in scratch, which is handed back for the next frame so a carrier +// does not allocate one per write. Callers hold the carrier's write mutex, +// which also guards scratch. +func writeFrame(w io.Writer, scratch []byte, op byte, stream uint64, payload []byte) ([]byte, error) { if len(payload) > MaxPayload { - return fmt.Errorf("relay: frame of %d bytes is over the %d-byte limit", len(payload), MaxPayload) + return scratch, fmt.Errorf("relay: frame of %d bytes is over the %d-byte limit", len(payload), MaxPayload) } - frame := make([]byte, frameHeader+len(payload)) - frame[0] = op - binary.BigEndian.PutUint64(frame[1:9], stream) - binary.BigEndian.PutUint32(frame[9:13], uint32(len(payload))) - copy(frame[frameHeader:], payload) + frame := append(scratch[:0], op) + frame = binary.BigEndian.AppendUint64(frame, stream) + frame = binary.BigEndian.AppendUint32(frame, uint32(len(payload))) + frame = append(frame, payload...) _, err := w.Write(frame) - return err + return frame, err } // readFrame takes one frame off a carrier. The payload is freshly allocated @@ -199,6 +199,7 @@ type carrier struct { conn io.ReadWriteCloser writeMu sync.Mutex + scratch []byte // the frame being written, reused under writeMu mu sync.Mutex streams map[uint64]*muxStream @@ -243,7 +244,9 @@ func (c *carrier) send(op byte, stream uint64, payload []byte) error { if gone { return errCarrierGone } - if err := writeFrame(c.conn, op, stream, payload); err != nil { + scratch, err := writeFrame(c.conn, c.scratch, op, stream, payload) + c.scratch = scratch + if err != nil { return err } return nil diff --git a/internal/relay/server.go b/internal/relay/server.go index da4a560a20..ffad1c0052 100644 --- a/internal/relay/server.go +++ b/internal/relay/server.go @@ -234,7 +234,9 @@ func (s *Server) serveDial(w http.ResponseWriter, r *http.Request, name string) stream, err := reg.carrier.open() if err != nil { - http.Error(w, err.Error(), http.StatusServiceUnavailable) + // 429, as a rate limit is: the machine is full now, and a surface + // that tries again in a minute may find room. + http.Error(w, err.Error(), http.StatusTooManyRequests) return } defer func() { _ = stream.Close() }() @@ -280,14 +282,24 @@ func upgradable(r *http.Request) bool { if strings.EqualFold(r.Header.Get("Upgrade"), Protocol) { return true } - return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && strings.Contains(r.Header.Get("Sec-WebSocket-Protocol"), Protocol) + if !wantsWebSocket(r) { + return false + } + for _, offered := range strings.Split(r.Header.Get("Sec-WebSocket-Protocol"), ",") { + if strings.TrimSpace(offered) == Protocol { + return true + } + } + return false } +func wantsWebSocket(r *http.Request) bool { return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") } + // 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) { - if strings.EqualFold(r.Header.Get("Upgrade"), "websocket") { + if wantsWebSocket(r) { socket, err := websocket.Accept(w, r, &websocket.AcceptOptions{ Subprotocols: []string{Protocol}, CompressionMode: websocket.CompressionDisabled, diff --git a/internal/relay/transport.go b/internal/relay/transport.go index 26436225eb..d0f69d5280 100644 --- a/internal/relay/transport.go +++ b/internal/relay/transport.go @@ -55,6 +55,9 @@ func upgradeWebSocket(ctx context.Context, address *url.URL, path string, header CompressionMode: websocket.CompressionDisabled, // the payloads are ciphertext }) if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } if response == nil { return nil, fmt.Errorf("%w: %s", ErrUnreachable, address.Host) } diff --git a/internal/relay/transport_test.go b/internal/relay/transport_test.go index a39ecce4bf..02a87c4aa2 100644 --- a/internal/relay/transport_test.go +++ b/internal/relay/transport_test.go @@ -124,3 +124,29 @@ func TestUpstreamFailuresBecomeFacts(t *testing.T) { t.Fatalf("a relay with no machine door: %v; want ErrNoDoor", err) } } + +// A machine at its stream ceiling is busy, not unreachable: the relay answers +// 429 and the surface hears that it is being turned away for now. +func TestAFullMachineIsBusyNotUnreachable(t *testing.T) { + _, address := live(t) + fullMachineIsBusy(t, address) +} + +func fullMachineIsBusy(t *testing.T, address string) { + t.Helper() + registration, err := Register(context.Background(), address, aKey(t)) + if err != nil { + t.Fatal(err) + } + defer func() { _ = registration.Close() }() + for i := 0; i < MaxStreams; i++ { + surface, err := Dial(context.Background(), address, registration.Name()) + if err != nil { + t.Fatalf("surface %d: %v", i, err) + } + defer func() { _ = surface.Close() }() + } + if _, err := Dial(context.Background(), address, registration.Name()); !errors.Is(err, ErrTooMany) { + t.Fatalf("a dial past %d streams gave %v; want ErrTooMany", MaxStreams, err) + } +} diff --git a/relay/hosted/README.md b/relay/hosted/README.md index 622df60831..b0401bd9a4 100644 --- a/relay/hosted/README.md +++ b/relay/hosted/README.md @@ -157,14 +157,15 @@ canonical names, so the stricter side is the safe one. | `MAILBOX` | Durable Object `Mailbox` | one pairing, one per nameplate | | `LINK_GATE` | Durable Object `LinkGate` (SQLite) | link-request codes, per-IP counters, one object | | `LINK_REQUEST` | Durable Object `LinkRequest` | one pending request, one per code | +| `MACHINE` | Durable Object `MachineDO` (SQLite) | one machine name's `codeaf serve` socket and the surfaces dialled to it | | `LEASE_POLICY` | var | `amended` (default: 90 s, a publish renews) or `stage1` (30 s leases) | | `CAF_LIMITS` | var, optional | JSON that overrides the numbers below | | `BASE_PATH` | var, optional | a path prefix every route is served beneath (for example, `/fabric`); empty, the default, serves from the root. A request outside the prefix is `404 not_found` | | `TRUST_PROXY` | var, optional | `1` takes the caller's network from the first `X-Forwarded-For` address (the self-hosted relay's `--trust-proxy`); for test rigs only, never production, where `CF-Connecting-IP` is the one truth | `wrangler.toml` declares all of them and the migration `v1` that creates the four classes as -SQLite-backed; migration `v2` adds the link classes and `v3-raft-members` adds the three-anchor -namespace. A deployment's own configuration must carry the `RAFT_MEMBERS` binding, the +SQLite-backed; migration `v2` adds the link classes, `v3-raft-members` adds the three-anchor +namespace and `v4-machines` adds `MachineDO`. A deployment's own configuration must carry the `RAFT_MEMBERS` binding, the `v3-raft-members` migration, the `src/raft-worker.js` entry and the build command that `wrangler.toml` declares; the first authenticated use of an existing identity then upgrades it in place, and that upgrade cannot be rolled back to the single-object writer. @@ -176,6 +177,24 @@ Running an older relay after the upgrade is not supported. Restore current code with the same complete storage to resume; the original directory and encrypted frames remain retained. +## Machine doors + +`codeaf serve` registers at `/v1/engine` and `codeaf chat --at ` dials at `/v1/dial/`, the +two doors of the self-hosted `codeaf relay` (`internal/relay`). A Worker answers only WebSocket +upgrades, so both doors take a WebSocket offering the `aforge-relay/1` subprotocol and answer anything +else with `426`, which is what tells a client to use the WebSocket. Over it run the same bytes as +over the self-hosted relay's raw upgrade: the registration proof, then the carrier's frames, one frame +to a message (`src/engine/carrier.js`, a port of `handshake.go` and `mux.go`). + +Each machine name has one `MachineDO` (`src/engine/machine.js`). It checks that the name is the one +the machine's key derives and that the machine holds that key, then holds the machine's socket and +splices each dialled socket onto it as one stream. It reads no payload. Every socket hibernates, and +the routing lives in socket tags and attachments, so a quiet machine costs nothing: the machine +proves the socket alive with WebSocket pings, which the platform answers without waking the object. +A second registration under a live name is refused; one under a name whose socket stopped answering +(a pong is asked for and not given within 5 s) replaces it, which is how a machine whose network +dropped comes back before the platform has closed its old socket. + ## Base path A deployment can share a hostname by using a path prefix, such as @@ -229,6 +248,8 @@ npx wrangler deploy --config wrangler.custom.toml | link requests: creates an hour, pending at once, per IP | 10, 3 | `429 rate_limited`, `Retry-After` | | link requests: reads a minute, misses a minute, open polls, per IP | 60, 20, 4 | `429 rate_limited` | | link requests: live in all | 2,000 | `503 full` | +| machine doors: surfaces on one machine at once (`MAX_STREAMS`) | 16 | `429` | +| machine doors: dials and registrations a minute, per network and machine name (in memory) | 30, 10 | `429` | Stored quotas (bytes, objects, frames a day), the stats counts and the per-IP counters live in the identity object's SQLite and survive eviction. A frame the store already holds costs no quota, so a client that resends is never refused diff --git a/relay/hosted/src/engine/carrier.js b/relay/hosted/src/engine/carrier.js index 21d5610446..50fd0c90eb 100644 --- a/relay/hosted/src/engine/carrier.js +++ b/relay/hosted/src/engine/carrier.js @@ -12,6 +12,15 @@ const LABEL = new TextEncoder().encode('aforge relay registration v1'); // legac // The tuning of relay.go, in the same names. export const MAX_PAYLOAD = 1 << 17; export const MAX_STREAMS = 16; +export const DIALS_PER_MINUTE = 30; +export const REGISTRATIONS_PER_MINUTE = 10; + +/** The two doors, beneath the deployment's base path. */ +export const ENGINE_PATH = '/v1/engine'; +export const DIAL_PREFIX = '/v1/dial/'; + +/** plain is a refusal in the relay's own sentence, which the Go client passes on as it is. */ +export const plain = (status, sentence) => new Response(sentence, { status, headers: { 'content-type': 'text/plain; charset=utf-8' } }); const FRAME_HEADER = 13; export const OP = { open: 1, data: 2, close: 3, ping: 4, pong: 5 }; diff --git a/relay/hosted/src/engine/door.js b/relay/hosted/src/engine/door.js index 01ae2c2c18..26e5899fa4 100644 --- a/relay/hosted/src/engine/door.js +++ b/relay/hosted/src/engine/door.js @@ -1,16 +1,11 @@ // The front door's half of the machine doors: refuse what needs no object to refuse, then hand the upgrade to the // one object that holds that machine name (machine.js). import { validName } from './names.js'; -import { HEADER_NAME, PROTOCOL } from './carrier.js'; - -export const ENGINE_PATH = '/v1/engine'; -export const DIAL_PREFIX = '/v1/dial/'; +import { DIAL_PREFIX, ENGINE_PATH, HEADER_NAME, PROTOCOL, plain } from './carrier.js'; /** isMachineDoor is a path this module serves. */ export const isMachineDoor = (pathname) => pathname === ENGINE_PATH || pathname.startsWith(DIAL_PREFIX); -const say = (status, sentence) => new Response(sentence, { status, headers: { 'content-type': 'text/plain; charset=utf-8' } }); - /** * 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. @@ -18,11 +13,11 @@ const say = (status, sentence) => new Response(sentence, { status, headers: { 'c export async function serveMachineDoor(request, env) { const offered = (request.headers.get('sec-websocket-protocol') ?? '').split(',').map((p) => p.trim()); if ((request.headers.get('upgrade') ?? '').toLowerCase() !== 'websocket' || !offered.includes(PROTOCOL)) { - return say(426, `this door speaks ${PROTOCOL} over a WebSocket`); + 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) ?? '' : decodeURIComponent(pathname.slice(DIAL_PREFIX.length)); - if (!validName(name)) return say(400, 'that is not a machine name'); + const name = pathname === ENGINE_PATH ? 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. return new Response(answer.body, { status: answer.status, headers: answer.headers, webSocket: answer.webSocket }); diff --git a/relay/hosted/src/engine/machine.js b/relay/hosted/src/engine/machine.js index 4430b4d451..419d13be04 100644 --- a/relay/hosted/src/engine/machine.js +++ b/relay/hosted/src/engine/machine.js @@ -7,10 +7,18 @@ // 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, and the object sets no timer. +// 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. import { DurableObject } from 'cloudflare:workers'; import { nameFor } from './names.js'; -import { MAX_PAYLOAD, MAX_STREAMS, OP, PROTOCOL, HEADER_NAME, HEADER_KEY, challenge, frame, readFrame, verdict } from './carrier.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; @@ -22,18 +30,35 @@ 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)); -const refuse = (status, sentence) => new Response(sentence, { status, headers: { 'content-type': 'text/plain; charset=utf-8' } }); 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 path = new URL(request.url).pathname; - return path === '/v1/engine' ? this.#register(request) : this.#dial(); + 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. */ @@ -66,7 +91,7 @@ export class MachineDO extends DurableObject { 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(503, `that machine already has ${MAX_STREAMS} connections open, which is as many as the relay carries at once`); + 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)); @@ -106,7 +131,7 @@ export class MachineDO extends DurableObject { 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) surface.close(1000, 'closed'); + 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. */ @@ -114,13 +139,20 @@ export class MachineDO extends DurableObject { 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 (older && (await this.#answers(older))) { - return this.#refuseRegistration(ws, 'that machine is already connected to the relay under this name'); + 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; } - 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()); } /** From 37965cfbc94c2b9ec78e6d91d9ea34bc638db172 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 30 Oct 2026 10:43:02 -0400 Subject: [PATCH 3/3] style: gofmt relay server Co-Authored-By: Claude Opus 5.5 --- internal/relay/server.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/internal/relay/server.go b/internal/relay/server.go index ffad1c0052..fb3fb185b9 100644 --- a/internal/relay/server.go +++ b/internal/relay/server.go @@ -293,7 +293,9 @@ func upgradable(r *http.Request) bool { return false } -func wantsWebSocket(r *http.Request) bool { return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") } +func wantsWebSocket(r *http.Request) bool { + return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") +} // 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