feat(once): carry a watch's exchange to its box as x

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,9 @@ 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 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.

README.md

@@ -79,18 +79,29 @@ 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=<id>`: 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` 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
   watch grades their freshness itself. The service keeps no session and no
   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
@@ -355,12 +366,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.

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

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-20261023161258-bac597f77f0a
+	github.com/srcfl/ftw/go/protocol v0.0.0-20261023165843-21e3d16946f1
 	golang.org/x/crypto v0.57.0
 	golang.org/x/net v0.58.0
 	golang.org/x/sys v0.48.0

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-20261023161258-bac597f77f0a h1:1Bxa1t8Kum4ECjMCogXw87M4RyfMJFJT7vWzCXLyw08=
-github.com/srcfl/ftw/go/protocol v0.0.0-20261023161258-bac597f77f0a/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU=
+github.com/srcfl/ftw/go/protocol v0.0.0-20261023165843-21e3d16946f1 h1:OCneVMFORT+ejZeiWAQ+66tHIN0IMXX7MtLtNbF6m1U=
+github.com/srcfl/ftw/go/protocol v0.0.0-20261023165843-21e3d16946f1/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=

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

internal/appsession/exchange_test.go

@@ -0,0 +1,162 @@
+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
+		role              string
+		tun, copyOn, susp bool
+		wantX, wantRead   int
+	}{
+		{"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)
+			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: c.role,
+					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)
+	}
+}

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)
 		}