From 0fef724c65ecc954e3c543acbe0f33f2ddaf6352 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 16:32:14 +0200 Subject: [PATCH 1/2] feat(once): carry a watch's exchange to its box as x The relay that carried a watch's commands stopped on 17 October 2026. A watch now sends each exchange message in a one-shot request as x {d}. The service sends x {dev, id, d} down the box's uplink, with the device id its own handshake matched and a number of its own, and answers with the box's d, unread and unchanged. - Moves the go/protocol pin to srcfl/ftw 657f84ae, which adds x. - /v1/app/once admits x by the roster entry's tun, so commands work with the readable copy off. The other types now need a non-empty effective scope, as protocol-cloud.md step 3 says; before, a tun device with the copy off got an empty snapshot. - No box, a box without x in hello.caps, 8 exchanges open at it, or 5 s without an answer: an empty d, never an error. The one-shot limits hold for the whole wait. - The uplink routes the box's x by id and dev. One nobody waits for is dropped, counted and charged to the box's message rate. When the box's connection ends, every waiting exchange gets an empty d at once. - Tests replay cloud-once-x-vectors.json through the real handler and uplink, and cover admission, no answer, routing and an end-to-end round trip. Pairs with srcfl/ftw 657f84ae (go/protocol and the golden vectors). Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 24 ++- README.md | 33 ++-- cmd/ftwcloudd/main.go | 1 + go.mod | 2 +- go.sum | 4 +- internal/appsession/appsession.go | 11 ++ internal/appsession/exchange_test.go | 159 ++++++++++++++++ internal/appsession/fuzz_test.go | 4 +- internal/appsession/once.go | 42 ++++- internal/appsession/once_vectors_test.go | 114 +++++++++++ internal/server/e2e_test.go | 57 +++++- internal/uplink/downlink.go | 8 +- internal/uplink/exchange_test.go | 229 +++++++++++++++++++++++ internal/uplink/uplink.go | 142 +++++++++++++- internal/uplink/uplink_test.go | 9 +- 15 files changed, 785 insertions(+), 54 deletions(-) create mode 100644 internal/appsession/exchange_test.go create mode 100644 internal/uplink/exchange_test.go diff --git a/CLAUDE.md b/CLAUDE.md index f558a40..a2eb604 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,12 +13,14 @@ maintains it. - **FTW Cloud forwards ciphertext it cannot read. It acts on a home only as an agent its owner approved, within limits the box enforces.** It sends a - box only `challenge`, `hello_ok`, `watch`, `resync`, `purged` and `tun` - (`internal/uplink/downlink.go`). Commands go from the app to the box inside - the phone's own Noise session, which rides here as `tun` under the device - id the service's own handshake proved, never one a phone names. An agent's - requests ride the agent's own Noise session the same way, and the box - admits it only while its signed roster lists the agent's key. + box only `challenge`, `hello_ok`, `watch`, `resync`, `purged`, `tun` and + `x` (`internal/uplink/downlink.go`). Commands go from the app to the box + inside the phone's own Noise session, which rides here as `tun` under the + device id the service's own handshake proved, never one a phone names. A + watch's commands go inside its exchange with the box, which rides as `x` + the same way. An agent's requests ride the agent's own Noise session the + same way, and the box admits it only while its signed roster lists the + agent's key. - **Agent keys live only in `ftwkeysd`** (`internal/keys`, `cmd/ftwkeysd`). It answers on a local unix socket, seals keys at rest under KMS, unseals a key only to open one session, and logs no key. `ftwcloudd` never holds an @@ -29,9 +31,10 @@ maintains it. once at start: it puts back the rosters the registry holds, each checked against its home's key, and reads nothing out. - **The home id comes from the key that signed hello** on this connection, - never from a body field. Every later box message but `tun` must verify - with that key. `tun` is unsigned: its `d` is a phone's ciphertext, which - the service routes by `dev` and `sid` and never reads. + never from a body field. Every later box message but `tun` and `x` must + verify with that key. Both are unsigned: a `tun`'s `d` is a phone's + ciphertext, which the service routes by `dev` and `sid` and never reads, + and an `x`'s a watch's, routed by `dev` and `id`. - **Live readings stay in memory.** The versioned project statistics scope covers rounded closed-hour energy and configured capacities on disk for 48 hours. Only group cells pass `/v1/stats`; no staff route reads @@ -46,7 +49,8 @@ maintains it. - **A phone can only narrow, or talk to its own box.** On `/v1/app` it can say whether it is looking, an owner can suspend a roster device, and a device the roster lists with `tun` can send its box session as `tun`. On - `/v1/app/once` a watch can read, ask for its token and register for push, + `/v1/app/once` a watch can read, ask for its token, register for push and, + if the roster lists it with `tun`, send its exchange with the box as `x`, and nothing else. The service admits a phone or a watch by the key its Noise handshake proved, and only if the box's roster lists it, or once for a pairing code the roster lists by its hash. diff --git a/README.md b/README.md index 102eea0..290886d 100644 --- a/README.md +++ b/README.md @@ -79,11 +79,12 @@ to it yet. - **One-shot read (ADR 0009).** watchOS gives an app HTTPS and nothing lower, so a watch reads at `POST /v1/app/once?home=`: the body is Noise IK message 1 with the prologue `ftw.cloud.once.v1:` and the home id, - and it carries one of `read {have, hz}`, `token.get`, `push.register` or - `push.unregister`. The service admits the watch exactly as `/v1/app` - admits a phone: its key in the box's roster, a roster that has not run - out, no suspension. The box lists a watch with role `watch` and the - scopes it derives; the service checks no role. A refusal is + and it carries one of `read {have, hz}`, `token.get`, `push.register`, + `push.unregister` or `x {d}`. The service admits the watch exactly as + `/v1/app` admits a phone: its key in the box's roster, a roster that has + not run out, no suspension. `x` needs the entry's `tun`; the rest need a + scope the readable copy grants. The box lists a watch with role `watch` + and the scopes it derives; the service checks no role. A refusal is 403 with an empty body. The answer is message 2 with the snapshot, less the envelopes `have` says the watch holds, and the envelopes that did not fit after it, each after its length. The box's bytes go unchanged, so the @@ -91,6 +92,15 @@ to it yet. Noise state after the answer. Limits: one read at a time per device and 30 a minute, 16 at once per home, 64 at once in the process, message 1 at most 4 KiB, the answer within 10 s. +- **A watch's commands ride `x`.** Its `d` is one message of the watch's + own short Noise session with the box (`ftw.exchange.v1`), which the + service cannot read. The service sends `x {dev, id, d}` down `/v1/box`, + with the device id its own handshake matched and a number of its own, and + answers with the `d` of the box's `x` under the same two, unchanged. It + answers an empty `d` at once when the box is not connected, did not list + `x` in `hello.caps` or has 8 exchanges open, and after 5 s without an + answer. `d` is at most 2,048 bytes, and the one-shot limits hold for the + whole wait. The service drops and counts an answer nobody waits for. - Limits per home: 16 sessions, one per device, 256 envelopes queued per phone, and a small message rate from each phone. - **QUIC.** UDP 443 carries what TCP 443 does, with the same certificate @@ -333,12 +343,13 @@ box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) - **A phone or a watch can narrow, and talk to its own box.** A phone can say whether it is looking, an owner can suspend a device in the roster, and a phone sends its box session as `tun`, which the service cannot read. A - watch can read, ask for its token and register for push. -- **The service sends six things.** `challenge`, `hello_ok`, `watch`, - `resync`, `purged` and `tun` are the whole downlink. The first five ask - the box for nothing; `tun` carries a phone's ciphertext, which the box - judges as it judges the same session over the relay. The box closes the - uplink on anything else. + watch can read, ask for its token, register for push and send its + exchange with the box as `x`, which the service cannot read either. +- **The service sends seven things.** `challenge`, `hello_ok`, `watch`, + `resync`, `purged`, `tun` and `x` are the whole downlink. The first five + ask the box for nothing; `tun` carries a phone's ciphertext and `x` a + watch's, which the box judges as it judges the same session over the + relay. The box closes the uplink on anything else. - **No TLS terminator in front.** Box and service bind the handshake to the TLS session, so `ftwcloudd` ends TLS itself. A TCP load balancer is fine, and so is a UDP one that forwards packets; nothing may end QUIC. diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 4867ad8..46a2b01 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -230,6 +230,7 @@ func serve(args []string) error { srv.App.Push = pushSvc srv.App.Tokens = tokens srv.App.Tunnel = up + srv.App.Exchange = up srv.App.Shards = shards } app := srv.App diff --git a/go.mod b/go.mod index f974bc9..89a60c1 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/fxamacker/cbor/v2 v2.9.4 github.com/gorilla/websocket v1.5.3 github.com/quic-go/quic-go v0.63.0 - github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104 + github.com/srcfl/ftw/go/protocol v0.0.0-20261023141450-657f84aef07f golang.org/x/crypto v0.57.0 golang.org/x/net v0.58.0 golang.org/x/sys v0.48.0 diff --git a/go.sum b/go.sum index 528c4e3..2261a5b 100644 --- a/go.sum +++ b/go.sum @@ -52,8 +52,8 @@ github.com/quic-go/quic-go v0.63.0 h1:LIFGHI4PFUhhw2dDD1ARHdCff143ffMHwZtbnbuJ78 github.com/quic-go/quic-go v0.63.0/go.mod h1:RAro2j2yN9a9EiPACLHT9IB2NXCvGQmmo/alT0yYI0w= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= -github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104 h1:nwfkV8mMubBLB+cx7mxFYrVT6Asgl8ctXWfjXci8MZU= -github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023141450-657f84aef07f h1:DoDixJsYu8Kau1ZTMIsN9L1WAhYDIQAUX3SQh8WGDYs= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023141450-657f84aef07f/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= diff --git a/internal/appsession/appsession.go b/internal/appsession/appsession.go index 4b94a4a..9a8a6ea 100644 --- a/internal/appsession/appsession.go +++ b/internal/appsession/appsession.go @@ -17,6 +17,7 @@ package appsession import ( + "context" "crypto/tls" "errors" "log/slog" @@ -83,6 +84,13 @@ type Tunneler interface { Tun(home string, t cloudwire.Tun) bool } +// Exchanger sends a watch's exchange message to its home's box and returns +// the box's answer. The uplink is one. dev comes from store.For, never from +// the watch. Nil or empty is no answer. +type Exchanger interface { + Exchange(ctx context.Context, home, dev string, d []byte) []byte +} + // What the app sessions count, summed over all homes. var ( appSockets = metrics.Sockets("app") @@ -124,6 +132,9 @@ type Server struct { // Tunnel carries the phones' sessions to their boxes. Nil answers every // tun with its end. Set before serving. Tunnel Tunneler + // Exchange carries the watches' exchanges to their boxes. Nil answers + // every x with no answer. Set before serving. + Exchange Exchanger // Shards says which homes this instance holds. Nil holds every home. // Set before serving. Shards *shard.Directory diff --git a/internal/appsession/exchange_test.go b/internal/appsession/exchange_test.go new file mode 100644 index 0000000..023841a --- /dev/null +++ b/internal/appsession/exchange_test.go @@ -0,0 +1,159 @@ +package appsession + +import ( + "context" + "net/http" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// exchanger is the uplink as a one-shot request sees it: what reached the +// box, and the box's answer. hold, when set, keeps each exchange open until +// it closes. +type exchanger struct { + mu sync.Mutex + got []cloudwire.X + answer []byte + hold chan struct{} +} + +func (e *exchanger) Exchange(_ context.Context, _ string, dev string, d []byte) []byte { + e.mu.Lock() + e.got = append(e.got, cloudwire.X{Dev: dev, D: d}) + hold := e.hold + e.mu.Unlock() + if hold != nil { + <-hold + } + return e.answer +} + +func (e *exchanger) sent() []cloudwire.X { + e.mu.Lock() + defer e.mu.Unlock() + return e.got +} + +// xOf reads an exchange's message 2. +func xOf(t *testing.T, payload []byte) []byte { + t.Helper() + env, err := cloudwire.DecodeEnvelope(payload) + var x cloudwire.AppX + if err != nil || env.T != cloudwire.MsgX || len(env.Sig) != 0 || cloudwire.Unmarshal(env.B, &x) != nil { + t.Fatalf("message 2 is %+v: %v", env, err) + } + return x.D +} + +// A watch's exchange is admitted by its roster entry's tun, so commands work +// with the readable copy off; everything else it may do needs the copy. The +// device that reaches the box is the one the handshake proved, whatever the +// body says. +func TestOnceAdmitsAnExchangeByTun(t *testing.T) { + watch, _ := appwire.GenerateKeyPair() + stranger, _ := appwire.GenerateKeyPair() + for _, c := range []struct { + name string + key appwire.KeyPair + tun, copyOn, susp bool + wantX, wantRead int + }{ + {"tun, the copy off", watch, true, false, false, http.StatusOK, http.StatusForbidden}, + {"tun, the copy on", watch, true, true, false, http.StatusOK, http.StatusOK}, + {"no tun, the copy on", watch, false, true, false, http.StatusForbidden, http.StatusOK}, + {"no tun, the copy off", watch, false, false, false, http.StatusForbidden, http.StatusForbidden}, + {"suspended", watch, true, true, true, http.StatusForbidden, http.StatusForbidden}, + {"a key the roster lacks", stranger, true, true, false, http.StatusForbidden, http.StatusForbidden}, + } { + t.Run(c.name, func(t *testing.T) { + h := newHarness(t) + e := &exchanger{answer: []byte("the box's answer")} + h.app.Exchange = e + grant := cloudwire.Grant{ID: make([]byte, 16), Controller: cloudwire.ControllerSourceful} + if c.copyOn { + grant.Scopes, grant.Purposes = []string{protocol.ScopeLiveRead}, []string{protocol.CloudPurposeServe} + } + h.rosterN++ + h.box.SetRoster(&cloudwire.Roster{RosterSeq: h.rosterN, TTLS: 86_400, Grant: grant, + Devices: append(h.devices(1, 1), cloudwire.Device{ID: "watch", Key: watch.Public(), Role: "watch", + Tun: c.tun, Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1})}) + if c.susp { + _ = h.reg.Suspend(h.home, "watch", 1, time.Now()) + } + h.live(true) + // A dev and an id in the body are not the watch's to give. + a := h.once(onceOpts{key: c.key, t: cloudwire.MsgX, + body: map[string]any{"d": []byte("message 1"), "dev": "owner", "id": 7}}) + if a.status != c.wantX { + t.Fatalf("x: %d, want %d", a.status, c.wantX) + } + if sent := e.sent(); a.status == http.StatusOK { + if len(sent) != 1 || sent[0].Dev != "watch" || string(sent[0].D) != "message 1" { + t.Fatalf("the box got %+v", sent) + } + if d := xOf(t, a.payload); string(d) != "the box's answer" { + t.Fatalf("the watch got %q", d) + } + } else if len(sent) != 0 { + t.Fatalf("a refused exchange reached the box: %+v", sent) + } + if a := h.once(onceOpts{key: c.key}); a.status != c.wantRead { + t.Fatalf("read: %d, want %d", a.status, c.wantRead) + } + }) + } +} + +// d longer than XDataMax is refused before it reaches the box; no box, or +// none that answers, is an empty d and not an error. +func TestAnExchangesSizeAndNoAnswer(t *testing.T) { + h := newHarness(t) + watch, _ := appwire.GenerateKeyPair() + h.setRoster(append(h.devices(1, 1), cloudwire.Device{ID: "watch", Key: watch.Public(), Role: "watch", Tun: true, + Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1}), protocol.ScopeLiveRead) + if a := h.once(onceOpts{key: watch, t: cloudwire.MsgX, body: cloudwire.AppX{D: make([]byte, cloudwire.XDataMax)}}); a.status != http.StatusOK || len(xOf(t, a.payload)) != 0 { + t.Fatalf("without an exchanger: %d", a.status) + } + e := &exchanger{} + h.app.Exchange = e + if a := h.once(onceOpts{key: watch, t: cloudwire.MsgX, body: cloudwire.AppX{D: make([]byte, cloudwire.XDataMax+1)}}); a.status != http.StatusBadRequest { + t.Fatalf("a d over XDataMax: %d", a.status) + } + if a := h.once(onceOpts{key: watch, t: cloudwire.MsgX, body: cloudwire.AppX{D: []byte{1}}}); a.status != http.StatusOK || len(xOf(t, a.payload)) != 0 { + t.Fatalf("no answer: %d", a.status) + } + if len(e.sent()) != 1 { + t.Fatalf("the box got %d", len(e.sent())) + } +} + +// One request at a time per device holds an exchange for its whole wait. +func TestAnExchangeHoldsItsDevicesTurn(t *testing.T) { + h := newHarness(t) + watch, _ := appwire.GenerateKeyPair() + h.setRoster(append(h.devices(1, 1), cloudwire.Device{ID: "watch", Key: watch.Public(), Role: "watch", Tun: true, + Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1}), protocol.ScopeLiveRead) + e := &exchanger{answer: []byte{2}, hold: make(chan struct{})} + h.app.Exchange = e + done := make(chan onceAnswer, 1) + go func() { done <- h.once(onceOpts{key: watch, t: cloudwire.MsgX, body: cloudwire.AppX{D: []byte{1}}}) }() + waitFor(t, "the exchange to wait", func() bool { return len(e.sent()) == 1 }) + if a := h.once(onceOpts{key: watch, t: cloudwire.MsgX, body: cloudwire.AppX{D: []byte{1}}}); a.status != http.StatusTooManyRequests { + t.Fatalf("a second exchange while one waits: %d", a.status) + } + if a := h.once(onceOpts{key: watch}); a.status != http.StatusTooManyRequests { + t.Fatalf("a read while an exchange waits: %d", a.status) + } + if a := h.once(onceOpts{key: h.owner}); a.status != http.StatusOK { + t.Fatalf("another device: %d", a.status) + } + close(e.hold) + if a := <-done; a.status != http.StatusOK || string(xOf(t, a.payload)) != "\x02" { + t.Fatalf("the exchange ended %d", a.status) + } +} diff --git a/internal/appsession/fuzz_test.go b/internal/appsession/fuzz_test.go index 5ce1be7..d7622d0 100644 --- a/internal/appsession/fuzz_test.go +++ b/internal/appsession/fuzz_test.go @@ -2,6 +2,7 @@ package appsession import ( "bytes" + "context" "testing" "github.com/srcfl/ftw/go/protocol/appwire" @@ -119,6 +120,7 @@ func FuzzOnceMessage(f *testing.F) { {cloudwire.MsgPushRegister, cloudwire.PushRegister{APNsToken: []byte{1}, Environment: "sandbox", Kind: "alert", App: "watch"}}, {cloudwire.MsgPushUnregister, cloudwire.PushUnregister{APNsToken: []byte{1}}}, {cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "owner"}}, + {cloudwire.MsgX, cloudwire.AppX{D: []byte{1}}}, } { payload, _ := cloudwire.Unsigned(seed.t, seed.body) f.Add(payload) @@ -137,7 +139,7 @@ func FuzzOnceMessage(f *testing.F) { if err != nil { return // larger than one Noise message: never reaches the service } - body, status := h.app.answerOnce(h.home, msg1) + body, status := h.app.answerOnce(context.Background(), h.home, msg1) if !allowed[status] { t.Fatalf("status %d", status) } diff --git a/internal/appsession/once.go b/internal/appsession/once.go index 20723b5..5866200 100644 --- a/internal/appsession/once.go +++ b/internal/appsession/once.go @@ -1,6 +1,7 @@ package appsession import ( + "context" "crypto/tls" "errors" "io" @@ -22,12 +23,13 @@ import ( // // The body is Noise IK message 1 to the service's Noise key, with // cloudwire.OncePrologue(home) as the prologue. It carries one unsigned -// envelope: read, token.get, push.register or push.unregister. The service -// admits the device as /v1/app admits a phone, through store.For with the -// key the handshake proved, and answers with message 2 and, for a read, the -// envelopes that did not fit, each after its length. It keeps no session and -// no Noise state after the answer; only the counts that bound how often a -// device, a home and the process may read. +// envelope: read, token.get, push.register, push.unregister or x, a watch's +// exchange message, which goes to the box and back unread ("Exchange"). The +// service admits the device as /v1/app admits a phone, through store.For +// with the key the handshake proved, and answers with message 2 and, for a +// read, the envelopes that did not fit, each after its length. It keeps no +// session and no Noise state after the answer; only the counts that bound +// how often a device, a home and the process may read. const ( // MaxOnceInFlight bounds the reads the process answers at once, counted @@ -166,7 +168,7 @@ func (s *Server) ServeOnce(w http.ResponseWriter, r *http.Request) { } defer s.once.leaveProcess() start := time.Now() - body, status := s.answerOnce(home, msg1) + body, status := s.answerOnce(r.Context(), home, msg1) onceHandshake.Since(start) if status != http.StatusOK { s.log.Debug("one-shot read refused", "status", status) @@ -179,7 +181,7 @@ func (s *Server) ServeOnce(w http.ResponseWriter, r *http.Request) { // answerOnce runs the handshake and returns the answer body, or the status // that refuses it. -func (s *Server) answerOnce(home string, msg1 []byte) ([]byte, int) { +func (s *Server) answerOnce(ctx context.Context, home string, msg1 []byte) ([]byte, int) { hs, err := appwire.NewResponder(appwire.HandshakeOptions{StaticKey: s.key, Prologue: cloudwire.OncePrologue(home), Ephemeral: s.onceEphemeral}) if err != nil { @@ -199,13 +201,18 @@ func (s *Server) answerOnce(home string, msg1 []byte) ([]byte, int) { if err != nil { return nil, http.StatusForbidden } + // x needs the roster's tun, as a phone's tunnel does, so commands work + // with the copy off. The rest needs the copy: an effective scope. + if (env.T == cloudwire.MsgX && !dev.Device().Tun) || (env.T != cloudwire.MsgX && len(dev.Scopes()) == 0) { + return nil, http.StatusForbidden + } release, ok := s.once.enter(home, hs.RemoteStatic(), s.now()) if !ok { return nil, http.StatusTooManyRequests } defer release() - reply, rest, status := s.onceReply(home, dev, env) + reply, rest, status := s.onceReply(ctx, home, dev, env) if status != http.StatusOK { return nil, status } @@ -242,7 +249,7 @@ func (s *Server) answerOnce(home string, msg1 []byte) ([]byte, int) { // onceReply is message 2's payload and, for a read, the envelopes to follow // it. dev is the roster entry store.For matched, never anything the message // says. -func (s *Server) onceReply(home string, dev *store.Handle, env cloudwire.Envelope) ([]byte, [][]byte, int) { +func (s *Server) onceReply(ctx context.Context, home string, dev *store.Handle, env cloudwire.Envelope) ([]byte, [][]byte, int) { switch env.T { case cloudwire.MsgRead: var rd cloudwire.Read @@ -308,6 +315,21 @@ func (s *Server) onceReply(home string, dev *store.Handle, env cloudwire.Envelop return nil, nil, http.StatusInternalServerError } return nil, nil, http.StatusOK + case cloudwire.MsgX: + var x cloudwire.AppX + if cloudwire.Unmarshal(env.B, &x) != nil || len(x.D) > cloudwire.XDataMax { + return nil, nil, http.StatusBadRequest + } + // No box, or no answer from it, is an empty d, never an error. + var d []byte + if s.Exchange != nil { + d = s.Exchange.Exchange(ctx, home, dev.Device().ID, x.D) + } + payload, err := cloudwire.Unsigned(cloudwire.MsgX, cloudwire.AppX{D: d}) + if err != nil { + return nil, nil, http.StatusInternalServerError + } + return payload, nil, http.StatusOK } return nil, nil, http.StatusBadRequest } diff --git a/internal/appsession/once_vectors_test.go b/internal/appsession/once_vectors_test.go index 0be93d8..1793770 100644 --- a/internal/appsession/once_vectors_test.go +++ b/internal/appsession/once_vectors_test.go @@ -2,6 +2,9 @@ package appsession import ( "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" "crypto/tls" "encoding/hex" "encoding/json" @@ -16,8 +19,10 @@ import ( "testing" "time" + "github.com/gorilla/websocket" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw-cloud/internal/uplink" "github.com/srcfl/ftw/go/protocol" "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" @@ -153,3 +158,112 @@ func TestOnceAnswersTheGoldenVectors(t *testing.T) { t.Fatalf("the vectors' have covers %v; this test assumes the full live and the lp", once.Held) } } + +// The service carries srcfl/ftw's cloud-once-x-vectors.json byte for byte: +// the watch's request, with the service's keys fixed as the vectors fix them, +// sends the box exactly the vectors' down frame on the real uplink, and the +// box's up frame draws exactly the answer body the watch replays. The copy is +// off: the watch's tun admits it. +func TestOnceCarriesTheExchangeVectors(t *testing.T) { + var v struct { + ServiceStatic, ServiceEphemeral, WatchPublic string + Home, Request, Down, Up, Answer string + } + readVectors(t, "cloud-once-x-vectors.json", &v) + var signed struct { + BoxKey struct{ D, PK, Home string } `json:"boxKey"` + } + readVectors(t, "cloud-vectors.json", &signed) + if signed.BoxKey.Home != v.Home { + t.Fatalf("the vectors name two homes, %s and %s", signed.BoxKey.Home, v.Home) + } + boxKey, err := ecdsa.ParseRawPrivateKey(elliptic.P256(), unhex(t, signed.BoxKey.D)) + if err != nil { + t.Fatal(err) + } + key, _ := appwire.KeyPairFromSecret(unhex(t, v.ServiceStatic)) + eph, _ := appwire.KeyPairFromSecret(unhex(t, v.ServiceEphemeral)) + serviceKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(nil) + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + up := uplink.New(st, reg, serviceKey, log) + app := New(st, reg, up, key, log) + app.onceEphemeral, app.Exchange = &eph, up + box, _ := st.For(store.Principal{Kind: store.Box, Home: v.Home}) + watch := unhex(t, v.WatchPublic) + box.SetRoster(&cloudwire.Roster{RosterSeq: 1, TTLS: 86_400, + Devices: []cloudwire.Device{{ID: cloudwire.DeviceID(watch), Key: watch, Role: "watch", Tun: true, Epoch: 1}}, + Grant: cloudwire.Grant{ID: make([]byte, 16), Controller: cloudwire.ControllerSourceful}}) + mux := http.NewServeMux() + mux.Handle("/v1/box", up) + mux.HandleFunc(cloudwire.OncePath, app.ServeOnce) + srv := httptest.NewUnstartedServer(mux) + srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + srv.StartTLS() + t.Cleanup(srv.Close) + + // The box: hello with x in its caps, then the one exchange. + d := websocket.Dialer{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, MinVersion: tls.VersionTLS13}} //nolint:gosec // the test's own server + conn, _, err := d.Dial(strings.Replace(srv.URL, "https://", "wss://", 1)+"/v1/box", nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + cs := conn.NetConn().(*tls.Conn).ConnectionState() + ekm, _ := cs.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) + read := func() cloudwire.Envelope { + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatal(err) + } + payload, _ := cloudwire.DecodeFrame(msg) + env, _ := cloudwire.DecodeEnvelope(payload) + return env + } + var ch cloudwire.Challenge + if cloudwire.Unmarshal(read().B, &ch) != nil { + t.Fatal("no challenge") + } + hello, _ := cloudwire.Seal(boxKey, v.Home, cloudwire.MsgHello, cloudwire.Hello{ + Header: cloudwire.Header{Seq: 1, Boot: bytes.Repeat([]byte{1}, cloudwire.BootBytes)}, V: cloudwire.Version, + PK: unhex(t, signed.BoxKey.PK), Nonce: ch.Nonce, EKM: ekm, BoxNonce: bytes.Repeat([]byte{4}, cloudwire.NonceBytes), + Build: "vectors", Clock: cloudwire.ClockNTP, Caps: []string{cloudwire.CapX}}) + raw, _ := hello.Encode() + frame, _ := cloudwire.EncodeFrame(raw) + if err := conn.WriteMessage(websocket.BinaryMessage, frame); err != nil { + t.Fatal(err) + } + if env := read(); env.T != cloudwire.MsgHelloOK { + t.Fatalf("got %q", env.T) + } + waitFor(t, "the box to be the home's", func() bool { return up.Boxes() == 1 }) + heard := make(chan []byte, 1) + go func() { + _, msg, err := conn.ReadMessage() + if err == nil && bytes.Equal(msg, unhex(t, v.Down)) { + _ = conn.WriteMessage(websocket.BinaryMessage, unhex(t, v.Up)) + } + heard <- msg + }() + + req, _ := http.NewRequest(http.MethodPost, srv.URL+cloudwire.OncePath+"?"+cloudwire.AppHomeParam+"="+v.Home, + bytes.NewReader(unhex(t, v.Request))) + req.Header.Set("Content-Type", "application/octet-stream") + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if msg := <-heard; !bytes.Equal(msg, unhex(t, v.Down)) { + t.Fatalf("the box got another frame than the vectors' down:\n got %x\nwant %s", msg, v.Down) + } + if resp.StatusCode != http.StatusOK || !bytes.Equal(body, unhex(t, v.Answer)) { + t.Fatalf("%d, not the vectors' answer:\n got %x\nwant %s", resp.StatusCode, body, v.Answer) + } +} diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index 07dabb0..1a29db7 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -90,7 +90,7 @@ func newStack(t *testing.T) *stack { pushSvc.WatchApp = "energy.ftw.app.watchkitapp" tokens := token.New(st, reg, log) app := appsession.New(st, reg, up, noise, log) - app.Push, app.Tokens, app.Tunnel = pushSvc, tokens, up + app.Push, app.Tokens, app.Tunnel, app.Exchange = pushSvc, tokens, up, up up.OnHeld = pushSvc.Held up.OnRoster = func(home string, r *cloudwire.Roster) { pushSvc.Roster(home, r) @@ -123,12 +123,14 @@ type testBox struct { // liveSeq is the seq of the newest live sent, which hb names. liveSeq uint64 conn wire.Conn - // heard is what the service told the box after hello_ok. + // heard is what the service told the box after hello_ok. An x it + // answers at once, with "answer to " before the watch's d. heard chan string start time.Time + mu sync.Mutex } -func (s *stack) connectBox() *testBox { +func (s *stack) connectBox(caps ...string) *testBox { s.t.Helper() key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) pk, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) @@ -152,7 +154,7 @@ func (s *stack) connectBox() *testBox { } boxNonce := bytes.Repeat([]byte{4}, cloudwire.NonceBytes) b.send(cloudwire.MsgHello, cloudwire.Hello{Header: b.header(), V: cloudwire.Version, PK: pk, Nonce: ch.Nonce, EKM: ekm, - BoxNonce: boxNonce, Build: "e2e", Clock: cloudwire.ClockNTP, RosterSeq: 1, GrantID: bytes.Repeat([]byte{5}, 16)}) + BoxNonce: boxNonce, Build: "e2e", Clock: cloudwire.ClockNTP, RosterSeq: 1, GrantID: bytes.Repeat([]byte{5}, 16), Caps: caps}) env = b.read() if env.T != cloudwire.MsgHelloOK || cloudwire.Verify(&s.serviceKey.PublicKey, b.home, env) != nil { s.t.Fatal("hello_ok did not check out") @@ -166,6 +168,16 @@ func (s *stack) connectBox() *testBox { } payload, _ := cloudwire.DecodeFrame(msg) e, _ := cloudwire.DecodeEnvelope(payload) + var x cloudwire.X + if e.T == cloudwire.MsgX && cloudwire.Unmarshal(e.B, &x) == nil { + x.D = append([]byte("answer to "), x.D...) + body, _ := cloudwire.Marshal(x) + raw, _ := cloudwire.Envelope{T: cloudwire.MsgX, B: body}.Encode() + frame, _ := cloudwire.EncodeFrame(raw) + b.mu.Lock() + _ = conn.WriteMessage(websocket.BinaryMessage, frame) + b.mu.Unlock() + } b.heard <- e.T } }() @@ -204,6 +216,8 @@ func (b *testBox) send(t string, body any) []byte { } raw, _ := env.Encode() frame, _ := cloudwire.EncodeFrame(raw) + b.mu.Lock() + defer b.mu.Unlock() if err := b.conn.WriteMessage(websocket.BinaryMessage, frame); err != nil { b.t.Fatal(err) } @@ -529,6 +543,41 @@ func (s *stack) now(tok string) (int, token.Now) { return resp.StatusCode, n } +// A watch whose roster entry has tun sends its exchange through the real +// uplink to its box and back; with nothing the copy lets it read, it reads +// nothing. When the box goes, the answer is empty and comes at once. +func TestAWatchExchangesWithItsBoxEndToEnd(t *testing.T) { + s := newStack(t) + box := s.connectBox(cloudwire.CapX) + watchKey, _ := appwire.GenerateKeyPair() + box.roster(1, cloudwire.Device{ID: "watch", Key: watchKey.Public(), Role: "watch", Tun: true, Epoch: 1}) + waitFor(t, "the roster", func() bool { + _, err := s.st.For(store.Principal{Kind: store.Device, Home: box.home, DeviceKey: watchKey.Public()}) + return err == nil + }) + watch := &watchClient{t: t, http: s.srv.Client(), base: s.srv.URL, key: watchKey, service: s.noise.Public(), home: box.home} + answer := func() (int, []byte) { + status, reply, _ := watch.call(cloudwire.MsgX, cloudwire.AppX{D: []byte("message 1")}) + var x cloudwire.AppX + if env, err := cloudwire.DecodeEnvelope(reply); status == http.StatusOK && (err != nil || cloudwire.Unmarshal(env.B, &x) != nil) { + t.Fatalf("message 2: %v", err) + } + return status, x.D + } + if status, d := answer(); status != http.StatusOK || string(d) != "answer to message 1" { + t.Fatalf("%d %q", status, d) + } + if status, _, _ := watch.call(cloudwire.MsgRead, cloudwire.Read{}); status != http.StatusForbidden { + t.Fatalf("a read with no scope: %d", status) + } + _ = box.conn.Close() + waitFor(t, "the box to go", func() bool { return s.up.Boxes() == 0 }) + start := time.Now() + if status, d := answer(); status != http.StatusOK || len(d) != 0 || time.Since(start) > time.Second { + t.Fatalf("%d %q after %v", status, d, time.Since(start)) + } +} + // waitHeld waits until the service holds n envelopes a device may read. func waitHeld(t *testing.T, st *store.Store, home string, key appwire.KeyPair, n int) { t.Helper() diff --git a/internal/uplink/downlink.go b/internal/uplink/downlink.go index 69d32af..81e263f 100644 --- a/internal/uplink/downlink.go +++ b/internal/uplink/downlink.go @@ -7,14 +7,15 @@ import ( ) // down is a message the service may send a box. The method is unexported, so -// only the six types below implement it and a seventh cannot be sent by +// only the seven types below implement it and an eighth cannot be sent by // accident. TestDownlinkIsTheProtocolsClosedSet holds the list to // cloudwire.Downlink, and the box closes the uplink on anything else. // // Five ask the box for nothing: watch asks it to send sooner, resync to send // again, and purged confirms a deletion the box asked for. tun carries a // phone's own Noise session, which the service cannot read or forge; the box -// judges it as it judges the same session over the relay. +// judges it as it judges the same session over the relay. exchange carries a +// watch's exchange message, which the box judges the same way. type down interface { down() (t string, body any, signed bool) } @@ -26,6 +27,7 @@ type ( resync cloudwire.Resync purged cloudwire.Purged tun cloudwire.Tun + exchange cloudwire.X ) func (m challenge) down() (string, any, bool) { @@ -37,6 +39,8 @@ func (m resync) down() (string, any, bool) { return cloudwire.MsgResync, cloudw func (m purged) down() (string, any, bool) { return cloudwire.MsgPurged, cloudwire.Purged(m), true } func (m tun) down() (string, any, bool) { return cloudwire.MsgTun, cloudwire.Tun(m), false } +func (m exchange) down() (string, any, bool) { return cloudwire.MsgX, cloudwire.X(m), false } + // encodeDown frames one downlink message. hello_ok and purged are signed with // the service key over the box's home id; the rest are not. func (u *Uplink) encodeDown(home string, m down) ([]byte, error) { diff --git a/internal/uplink/exchange_test.go b/internal/uplink/exchange_test.go new file mode 100644 index 0000000..0871efa --- /dev/null +++ b/internal/uplink/exchange_test.go @@ -0,0 +1,229 @@ +package uplink + +import ( + "bytes" + "context" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// xConnect connects a box that lists x in hello.caps, or none when caps is +// empty, and waits until its connection is the home's. +func xConnect(t *testing.T, s *service, caps ...string) *box { + t.Helper() + b := newBox(t) + b.caps = caps + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + wait(t, "the box is the home's", func() bool { + s.up.mu.Lock() + defer s.up.mu.Unlock() + return s.up.conns[b.home] != nil + }) + return b +} + +// ask runs one exchange in the background; its answer comes on the channel. +func ask(s *service, home, dev string, d []byte) <-chan []byte { + out := make(chan []byte, 1) + go func() { out <- s.up.Exchange(context.Background(), home, dev, d) }() + return out +} + +// answered waits for an exchange's answer and how long it took from start. +func answered(t *testing.T, got <-chan []byte, start time.Time) ([]byte, time.Duration) { + t.Helper() + select { + case a := <-got: + return a, time.Since(start) + case <-time.After(10 * time.Second): + t.Fatal("the exchange never ended") + return nil, 0 + } +} + +// readX reads the next x the service sent the box. +func (b *box) readX() cloudwire.X { + b.t.Helper() + env := b.read() + var x cloudwire.X + if env.T != cloudwire.MsgX || len(env.Sig) != 0 || cloudwire.Unmarshal(env.B, &x) != nil { + b.t.Fatalf("the box got %q", env.T) + } + return x +} + +// sendX sends the box's x, unsigned, as the box does. +func (b *box) sendX(x any) { + body, _ := cloudwire.Marshal(x) + raw, _ := cloudwire.Envelope{T: cloudwire.MsgX, B: body}.Encode() + frame, _ := cloudwire.EncodeFrame(raw) + _ = b.conn.WriteMessage(websocket.BinaryMessage, frame) +} + +// An exchange goes down under the device the service names and the service's +// own number, and the box's answer under the same comes back unchanged. +func TestAnExchangeReachesItsBoxAndComesBack(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + start := time.Now() + got := ask(s, b.home, "watch", []byte("message 1")) + x := b.readX() + if x.Dev != "watch" || x.ID == 0 || string(x.D) != "message 1" { + t.Fatalf("the box got %+v", x) + } + b.sendX(cloudwire.X{Dev: x.Dev, ID: x.ID, D: []byte("message 2")}) + if a, _ := answered(t, got, start); string(a) != "message 2" { + t.Fatalf("the watch got %q", a) + } + // An empty answer is the box's silence, and goes back empty. + got = ask(s, b.home, "watch", []byte("spent")) + x = b.readX() + b.sendX(cloudwire.X{Dev: x.Dev, ID: x.ID}) + if a, _ := answered(t, got, start); len(a) != 0 { + t.Fatalf("the watch got %q", a) + } +} + +// Two watches of one home with exchanges open each get their own answer, +// whatever order the box answers in, and an answer under another device is +// nobody's. +func TestEachWatchGetsItsOwnAnswer(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + start := time.Now() + one := ask(s, b.home, "one", []byte("from one")) + x1 := b.readX() + two := ask(s, b.home, "two", []byte("from two")) + x2 := b.readX() + if x1.ID == x2.ID { + t.Fatalf("two exchanges under one number %d", x1.ID) + } + before := dropXUnknown.Load() + b.sendX(cloudwire.X{Dev: "two", ID: x1.ID, D: []byte("stolen")}) + b.sendX(cloudwire.X{Dev: "two", ID: x2.ID, D: []byte("for two")}) + b.sendX(cloudwire.X{Dev: "one", ID: x1.ID, D: []byte("for one")}) + if a, _ := answered(t, two, start); string(a) != "for two" { + t.Fatalf("two got %q", a) + } + if a, _ := answered(t, one, start); string(a) != "for one" { + t.Fatalf("one got %q", a) + } + if n := dropXUnknown.Load() - before; n != 1 { + t.Fatalf("dropped %d", n) + } +} + +// An answer nobody waits for is dropped and counted, and the box stays. +func TestAnAnswerNobodyWaitsForIsDropped(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + before := dropXUnknown.Load() + b.sendX(cloudwire.X{Dev: "watch", ID: 99, D: []byte{1}}) + wait(t, "the drop", func() bool { return dropXUnknown.Load() == before+1 }) + start := time.Now() + got := ask(s, b.home, "watch", []byte{2}) + x := b.readX() + b.sendX(cloudwire.X{Dev: x.Dev, ID: x.ID, D: []byte{3}}) + if a, _ := answered(t, got, start); !bytes.Equal(a, []byte{3}) { + t.Fatalf("after the drop the watch got %v", a) + } +} + +// No answer is an empty one: at once with no box, a box that does not take +// x or XInFlight open, and after the wait for a box that keeps silent. +func TestNoAnswerIsEmpty(t *testing.T) { + t.Run("no box", func(t *testing.T) { + s := newService(t) + start := time.Now() + a, took := answered(t, ask(s, newBox(t).home, "watch", []byte{1}), start) + if a != nil || took > 500*time.Millisecond { + t.Fatalf("%v after %v", a, took) + } + }) + t.Run("a box without x", func(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapTun) + start := time.Now() + a, took := answered(t, ask(s, b.home, "watch", []byte{1}), start) + if a != nil || took > 500*time.Millisecond { + t.Fatalf("%v after %v", a, took) + } + _ = b.conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + if _, _, err := b.conn.ReadMessage(); err == nil || !strings.Contains(err.Error(), "timeout") { + t.Fatalf("a box without x heard something: %v", err) + } + }) + t.Run("a silent box", func(t *testing.T) { + s := newService(t) + s.up.xWithin = 300 * time.Millisecond + b := xConnect(t, s, cloudwire.CapX) + start := time.Now() + got := ask(s, b.home, "watch", []byte{1}) + b.readX() + a, took := answered(t, got, start) + if a != nil || took < 300*time.Millisecond || took > 2*time.Second { + t.Fatalf("%v after %v", a, took) + } + }) + t.Run("past XInFlight", func(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + var open []<-chan []byte + for range cloudwire.XInFlight { + open = append(open, ask(s, b.home, "watch", []byte{1})) + b.readX() + } + before := dropXBusy.Load() + start := time.Now() + a, took := answered(t, ask(s, b.home, "watch", []byte{1}), start) + if a != nil || took > 500*time.Millisecond || dropXBusy.Load() != before+1 { + t.Fatalf("%v after %v", a, took) + } + _ = b.conn.Close() + for _, got := range open { + answered(t, got, start) + } + }) +} + +// When the box's connection ends, every exchange waiting on it is answered, +// empty, at once rather than after the wait. +func TestTheEndOfAnUplinkAnswersEveryExchange(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + one, two := ask(s, b.home, "one", []byte{1}), ask(s, b.home, "two", []byte{2}) + b.readX() + b.readX() + start := time.Now() + _ = b.conn.Close() + for _, got := range []<-chan []byte{one, two} { + if a, took := answered(t, got, start); a != nil || took > time.Second { + t.Fatalf("%v after %v", a, took) + } + } +} + +// An x the service cannot route closes the uplink, as a tun does. +func TestAMalformedXCloses(t *testing.T) { + for name, send := range map[string]func(b *box){ + "signed": func(b *box) { b.send(cloudwire.MsgX, cloudwire.X{Dev: "watch", ID: 1, D: []byte{1}}) }, + "no device": func(b *box) { b.sendX(cloudwire.X{ID: 1, D: []byte{1}}) }, + "d too long": func(b *box) { b.sendX(cloudwire.X{Dev: "watch", ID: 1, D: make([]byte, cloudwire.XDataMax+1)}) }, + "not an x": func(b *box) { b.sendX(map[string]any{"dev": 7, "id": "one"}) }, + } { + t.Run(name, func(t *testing.T) { + s := newService(t) + b := xConnect(t, s, cloudwire.CapX) + send(b) + if !b.closed() { + t.Fatal("the box stayed") + } + }) + } +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index 6e63fc1..522d512 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -4,14 +4,15 @@ // // The service learns a home only from its box: the home id is a hash of the // key that signed hello on this TLS session, never a field anyone sends, and -// every message after it but tun must carry that key's signature. tun is a -// phone's own Noise session with the box, which the service routes by device -// and never reads. What the service may say back is the closed set in -// downlink.go. +// every message after it but tun and x must carry that key's signature. tun is +// a phone's own Noise session with the box, and x a watch's exchange with it, +// which the service routes by device and never reads. What the service may +// say back is the closed set in downlink.go. package uplink import ( "bytes" + "context" "crypto/ecdsa" "crypto/rand" "crypto/tls" @@ -51,7 +52,7 @@ const ( writeTimeout = 10 * time.Second // A box sends a few messages a second at most. Past the burst at this // rate the connection closes; one home cannot occupy the service. tun - // is counted apart. + // is counted apart, and so is an x that answers an exchange. rateBurst = 60 ratePerSec = 10 // tun per home and direction, the limits the relay puts on a box's @@ -92,6 +93,8 @@ var ( dropTunDown = metrics.Drop("tun_rate_down") dropUnknown = metrics.Drop("box_unknown_type") dropConnect = metrics.Drop("box_connect_rate") + dropXBusy = metrics.Drop("x_in_flight") + dropXUnknown = metrics.Drop("x_unknown") boxesFollow = metrics.Default.Gauge("ftwcloud_boxes_following_shards", "Connected boxes that list shard in hello.caps.") ) @@ -109,6 +112,9 @@ type Uplink struct { key *ecdsa.PrivateKey log *slog.Logger now func() time.Time + // xWithin is how long Exchange waits for the box's answer: + // cloudwire.XAnswerWithin. Tests shorten it. + xWithin time.Duration mu sync.Mutex conns map[string]*boxConn @@ -155,8 +161,8 @@ func New(st *store.Store, reg *registry.Registry, key *ecdsa.PrivateKey, log *sl if log == nil { log = slog.Default() } - return &Uplink{store: st, registry: reg, key: key, log: log, now: time.Now, conns: map[string]*boxConn{}, - limits: map[string]*limits{}} + return &Uplink{store: st, registry: reg, key: key, log: log, now: time.Now, xWithin: cloudwire.XAnswerWithin, + conns: map[string]*boxConn{}, limits: map[string]*limits{}} } // limits are one home's rate limits. They outlive the connection, so a box @@ -354,6 +360,38 @@ func (u *Uplink) Tun(home string, t cloudwire.Tun) bool { return true } +// Exchange sends a watch's exchange message d to its home's box as x and +// waits for the box's answer. dev is the device the one-shot request's own +// handshake matched, never one the watch named. Nil is no answer: no box is +// connected, it does not take x, it has cloudwire.XInFlight open, or it did +// not answer within cloudwire.XAnswerWithin. +func (u *Uplink) Exchange(ctx context.Context, home, dev string, d []byte) []byte { + u.mu.Lock() + c := u.conns[home] + u.mu.Unlock() + if c == nil || !c.x { + return nil + } + id, answer, ok := c.wait(dev) + if !ok { + dropXBusy.Inc() + return nil + } + defer c.unwait(id) + if c.send(exchange{Dev: dev, ID: id, D: d}) != nil { + return nil + } + t := time.NewTimer(u.xWithin) + defer t.Stop() + select { + case a := <-answer: + return a + case <-t.C: + case <-ctx.Done(): + } + return nil +} + // Restore holds each home's last signed roster again after a restart, read // back from the registry, so its phones are admitted before its box has // reconnected. A roster that does not verify with the home's key is left @@ -437,9 +475,16 @@ type boxConn struct { pub *ecdsa.PublicKey handle *store.Handle purged bool - // tun is whether the box listed tun in hello.caps. An older box closes - // the uplink on a type it does not know, so it is sent none. - tun bool + // tun and x are whether the box listed each in hello.caps. An older box + // closes the uplink on a type it does not know, so it is sent none. + tun, x bool + // xs are the exchanges waiting for the box's answer, by the number + // xNext gave each: at most cloudwire.XInFlight. xEnded is set when the + // connection has ended and each was answered with nothing. + xMu sync.Mutex + xs map[uint32]waiter + xNext uint32 + xEnded bool // bye is the reason the box gave for leaving, if it gave one. bye string // sock closes the socket with a code; started is when it was @@ -460,6 +505,50 @@ type boxConn struct { statsAllowed bool // current connection accepted the full statistics roster } +// waiter is one exchange waiting for the box: the device it is for, and +// where the box's answer goes. +type waiter struct { + dev string + answer chan []byte +} + +// wait opens one exchange for dev and returns its number and where its +// answer comes. False when cloudwire.XInFlight are open or the connection +// has ended. +func (c *boxConn) wait(dev string) (uint32, <-chan []byte, bool) { + c.xMu.Lock() + defer c.xMu.Unlock() + if c.xEnded || len(c.xs) >= cloudwire.XInFlight { + return 0, nil, false + } + if c.xs == nil { + c.xs = map[uint32]waiter{} + } + c.xNext++ + w := waiter{dev: dev, answer: make(chan []byte, 1)} + c.xs[c.xNext] = w + return c.xNext, w.answer, true +} + +// unwait closes one exchange, answered or not. +func (c *boxConn) unwait(id uint32) { + c.xMu.Lock() + defer c.xMu.Unlock() + delete(c.xs, id) +} + +// endX answers each exchange still waiting with nothing, at once: the box +// that would answer is gone. +func (c *boxConn) endX() { + c.xMu.Lock() + defer c.xMu.Unlock() + c.xEnded = true + for id, w := range c.xs { + close(w.answer) + delete(c.xs, id) + } +} + // bucket is a token bucket. type bucket struct { tokens, burst, rate float64 @@ -589,6 +678,7 @@ func (c *boxConn) run() error { if !c.purged { gen := c.handle.Connected() defer c.handle.Disconnected(gen) + defer c.endX() c.u.mu.Lock() if old := c.u.conns[c.home]; old != nil { // One uplink per home: the newer one wins. @@ -646,6 +736,12 @@ func (c *boxConn) run() error { } continue } + if env.T == cloudwire.MsgX && !c.purged { + if err := c.answer(env); err != nil { + return err + } + continue + } if !c.allow() { return fmt.Errorf("%w: faster than %d messages a second", errProtocol, ratePerSec) } @@ -674,6 +770,7 @@ func (c *boxConn) hello(env cloudwire.Envelope, nonce []byte) (cloudwire.Hello, return h, fmt.Errorf("%w: hello is not bound to this challenge and TLS session", errProtocol) } c.home, c.pub, c.tun = home, pub, slices.Contains(h.Caps, cloudwire.CapTun) + c.x = slices.Contains(h.Caps, cloudwire.CapX) return h, nil } @@ -708,6 +805,31 @@ func (c *boxConn) fromBox(env cloudwire.Envelope, bulk bool) error { return nil } +// answer hands the box's x to the exchange waiting under its number. Like +// tun it is unsigned and never read. One nobody waits for, late or under +// another device, is dropped and counted; past the box's message rate, the +// uplink closes, since the box answers only what it was asked. +func (c *boxConn) answer(env cloudwire.Envelope) error { + var a cloudwire.X + if len(env.Sig) != 0 || cloudwire.Unmarshal(env.B, &a) != nil || len(a.D) > cloudwire.XDataMax || a.Dev == "" { + return fmt.Errorf("%w: an x the service cannot route", errProtocol) + } + c.xMu.Lock() + w, ok := c.xs[a.ID] + if ok = ok && w.dev == a.Dev; ok { + delete(c.xs, a.ID) + w.answer <- a.D + } + c.xMu.Unlock() + if !ok { + dropXUnknown.Inc() + if !c.allow() { + return fmt.Errorf("%w: faster than %d messages a second", errProtocol, ratePerSec) + } + } + return nil +} + func (c *boxConn) rosterSeq() uint64 { var seq uint64 if c.handle != nil { diff --git a/internal/uplink/uplink_test.go b/internal/uplink/uplink_test.go index 438a763..87d5a18 100644 --- a/internal/uplink/uplink_test.go +++ b/internal/uplink/uplink_test.go @@ -58,6 +58,8 @@ type box struct { boot []byte seq uint64 conn *websocket.Conn + // caps are what hello lists. + caps []string } func newBox(t *testing.T) *box { @@ -98,7 +100,7 @@ func (b *box) connect(s *service, ekmOverride []byte) (cloudwire.HelloOK, error) boxNonce := bytes.Repeat([]byte{4}, 32) b.send(cloudwire.MsgHello, cloudwire.Hello{Header: b.header(), V: 1, PK: b.pk, Nonce: ch.Nonce, EKM: ekm, BoxNonce: boxNonce, Build: "test", Clock: cloudwire.ClockNTP, RosterSeq: 1, - GrantID: bytes.Repeat([]byte{5}, 16)}) + GrantID: bytes.Repeat([]byte{5}, 16), Caps: b.caps}) _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second)) _, msg, err := conn.ReadMessage() @@ -334,12 +336,13 @@ func TestASecondUplinkReplacesTheFirst(t *testing.T) { } } -// The service can say five things to a box, and they are the protocol's five. +// The service can say seven things to a box, and they are the protocol's +// seven. func TestDownlinkIsTheProtocolsClosedSet(t *testing.T) { s := newService(t) all := []down{challenge{V: 1, Nonce: make([]byte, 32)}, helloOK{Have: map[string]string{}}, watch{Hz: 1}, resync{What: []string{cloudwire.ResyncFull}}, purged{What: cloudwire.PurgedHome}, - tun{Dev: "q5Lw0Zb1", Sid: 1, D: []byte{1}}} + tun{Dev: "q5Lw0Zb1", Sid: 1, D: []byte{1}}, exchange{Dev: "q5Lw0Zb1", ID: 1, D: make([]byte, cloudwire.XDataMax)}} var names []string for _, m := range all { frame, err := s.up.encodeDown("home", m) From efee312e254f2bab6663512c2c08917517533a80 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 18:02:49 +0200 Subject: [PATCH 2/2] fix(once): x is a watch's alone A roster entry with tun but another role, a phone's, has its own tunnel and no business on the watch's door. Review of #49 (N1). Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 3 ++- README.md | 7 ++++--- internal/appsession/exchange_test.go | 17 ++++++++++------- internal/appsession/once.go | 12 +++++++++--- 4 files changed, 25 insertions(+), 14 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a2eb604..04b8019 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -50,7 +50,8 @@ maintains it. say whether it is looking, an owner can suspend a roster device, and a device the roster lists with `tun` can send its box session as `tun`. On `/v1/app/once` a watch can read, ask for its token, register for push and, - if the roster lists it with `tun`, send its exchange with the box as `x`, + if the roster lists it as a watch with `tun`, send its exchange with the + box as `x`, and nothing else. The service admits a phone or a watch by the key its Noise handshake proved, and only if the box's roster lists it, or once for a pairing code the roster lists by its hash. diff --git a/README.md b/README.md index 290886d..06d6ed5 100644 --- a/README.md +++ b/README.md @@ -82,9 +82,10 @@ to it yet. and it carries one of `read {have, hz}`, `token.get`, `push.register`, `push.unregister` or `x {d}`. The service admits the watch exactly as `/v1/app` admits a phone: its key in the box's roster, a roster that has - not run out, no suspension. `x` needs the entry's `tun`; the rest need a - scope the readable copy grants. The box lists a watch with role `watch` - and the scopes it derives; the service checks no role. A refusal is + not run out, no suspension. `x` needs the entry's `tun` and role `watch`; + the rest need a scope the readable copy grants. The box lists a watch + with role `watch` and the scopes it derives; the service checks no other + role. A refusal is 403 with an empty body. The answer is message 2 with the snapshot, less the envelopes `have` says the watch holds, and the envelopes that did not fit after it, each after its length. The box's bytes go unchanged, so the diff --git a/internal/appsession/exchange_test.go b/internal/appsession/exchange_test.go index 023841a..59604f8 100644 --- a/internal/appsession/exchange_test.go +++ b/internal/appsession/exchange_test.go @@ -60,15 +60,18 @@ func TestOnceAdmitsAnExchangeByTun(t *testing.T) { for _, c := range []struct { name string key appwire.KeyPair + role string tun, copyOn, susp bool wantX, wantRead int }{ - {"tun, the copy off", watch, true, false, false, http.StatusOK, http.StatusForbidden}, - {"tun, the copy on", watch, true, true, false, http.StatusOK, http.StatusOK}, - {"no tun, the copy on", watch, false, true, false, http.StatusForbidden, http.StatusOK}, - {"no tun, the copy off", watch, false, false, false, http.StatusForbidden, http.StatusForbidden}, - {"suspended", watch, true, true, true, http.StatusForbidden, http.StatusForbidden}, - {"a key the roster lacks", stranger, true, true, false, http.StatusForbidden, http.StatusForbidden}, + {"tun, the copy off", watch, "watch", true, false, false, http.StatusOK, http.StatusForbidden}, + {"tun, the copy on", watch, "watch", true, true, false, http.StatusOK, http.StatusOK}, + {"no tun, the copy on", watch, "watch", false, true, false, http.StatusForbidden, http.StatusOK}, + {"no tun, the copy off", watch, "watch", false, false, false, http.StatusForbidden, http.StatusForbidden}, + {"suspended", watch, "watch", true, true, true, http.StatusForbidden, http.StatusForbidden}, + {"a key the roster lacks", stranger, "watch", true, true, false, http.StatusForbidden, http.StatusForbidden}, + // x is a watch's: a phone with tun has its own tunnel. + {"a phone with tun", watch, "owner", true, true, false, http.StatusForbidden, http.StatusOK}, } { t.Run(c.name, func(t *testing.T) { h := newHarness(t) @@ -80,7 +83,7 @@ func TestOnceAdmitsAnExchangeByTun(t *testing.T) { } h.rosterN++ h.box.SetRoster(&cloudwire.Roster{RosterSeq: h.rosterN, TTLS: 86_400, Grant: grant, - Devices: append(h.devices(1, 1), cloudwire.Device{ID: "watch", Key: watch.Public(), Role: "watch", + Devices: append(h.devices(1, 1), cloudwire.Device{ID: "watch", Key: watch.Public(), Role: c.role, Tun: c.tun, Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1})}) if c.susp { _ = h.reg.Suspend(h.home, "watch", 1, time.Now()) diff --git a/internal/appsession/once.go b/internal/appsession/once.go index 5866200..278e3cd 100644 --- a/internal/appsession/once.go +++ b/internal/appsession/once.go @@ -12,6 +12,7 @@ import ( "time" "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" ) @@ -201,9 +202,14 @@ func (s *Server) answerOnce(ctx context.Context, home string, msg1 []byte) ([]by if err != nil { return nil, http.StatusForbidden } - // x needs the roster's tun, as a phone's tunnel does, so commands work - // with the copy off. The rest needs the copy: an effective scope. - if (env.T == cloudwire.MsgX && !dev.Device().Tun) || (env.T != cloudwire.MsgX && len(dev.Scopes()) == 0) { + // x is a watch's, and needs the roster's tun, as a phone's tunnel does, + // so commands work with the copy off. The rest needs the copy: an + // effective scope. + if env.T == cloudwire.MsgX { + if d := dev.Device(); !d.Tun || d.Role != protocol.RoleWatch { + return nil, http.StatusForbidden + } + } else if len(dev.Scopes()) == 0 { return nil, http.StatusForbidden } release, ok := s.once.enter(home, hs.RemoteStatic(), s.now())