diff --git a/CLAUDE.md b/CLAUDE.md index 04e4921..dafb6e4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,9 +10,13 @@ maintains it. ## Invariants -- **FTW Cloud reads and never controls.** It sends a box only `challenge`, - `hello_ok`, `watch`, `resync` and `purged` (`internal/uplink/downlink.go`). - Commands go from the app to the box over Noise, never through here. +- **The readable path reads and never controls.** On `/v1/box` it sends a + box only `challenge`, `hello_ok`, `watch`, `resync` and `purged` + (`internal/uplink/downlink.go`). Remote commands are a separate, blind + path (`internal/relay`): the service forwards end-to-end Noise frames and + cannot open or originate one. ADR 0008's "commands stay on the Frankfurt + relay" is being revised; the ADR text is in `srcfl/ftw` and is not edited + here. - **One way to a home:** `store.For(principal)`. No listing, no second lookup, no principal for staff. A new route needs a row in `server.Routes` and must pass `TestNoRouteReadsAHome`. @@ -20,12 +24,15 @@ maintains it. never from a body field. Every later box message must verify with that key. - **No readings on disk.** Memory holds readings; SQLite holds only the home id, the box key, the signed roster, owners' suspensions, tombstones, APNs - registrations and the hashes of extension tokens. Nothing logs addresses. + registrations and the hashes of extension tokens. The command path's + dead-man file holds sealed push ciphertext, not readings. Nothing logs + addresses, rendezvous handles or frame bodies. - **A phone can only narrow.** On `/v1/app` it can say whether it is looking, and an owner can suspend a roster device. On `/v1/app/once` a watch can read, ask for its token and register for push, 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. + and only if the box's roster lists it. A command the phone sends is on + `/r/` or `/x/`, and that path does not admit it to a home. - **TLS ends in `ftwcloudd`.** The handshake is bound to the TLS session; a terminating proxy breaks it on purpose. A TCP load balancer is fine. - **Forward the box's bytes unchanged.** The service cannot sign for a box, diff --git a/README.md b/README.md index b72dad5..d46627a 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,14 @@ # FTW Cloud FTW Cloud keeps a signed, readable copy of each home that turns it on, and -serves it to that home's phones. It reads and never controls. The box opens -the connection, picks and signs every message, and can revoke at any time. +serves it to that home's phones. On that path it reads and never controls: +the box opens the connection, picks and signs every message, and can revoke +at any time. A second path on the same server carries remote commands. Those +frames are end-to-end Noise between the app and the box; this service +forwards the bytes and cannot open them, and it does not originate a +command. [ADR 0008](https://github.com/srcfl/ftw/blob/master/docs/adr/0008-ftw-cloud.md) +said commands stay on the Frankfurt relay. That stance is being revised. +The ADR text lives in `srcfl/ftw` and is not changed here. This is `ftwcloudd`, the service. The decision is [ADR 0008](https://github.com/srcfl/ftw/blob/master/docs/adr/0008-ftw-cloud.md) @@ -81,6 +87,20 @@ to it yet. token, valid for 7 days; a new one replaces the old. It stops reading when the roster drops the device, moves its epoch or runs out, an owner suspends the device, or the home is purged. +- **Remote commands.** The Frankfurt relay's protocol, on this server. + A box dials out to `/r///box`, an app to `…/app`, and the + box's exchange socket to `…/exchange`. A watch posts one command at + `/x//`. Binary frames pass through unchanged and stay in + the room named by the handle. The handle rotates every hour, spread over + five minutes, and `/x/` accepts the previous epoch during that window. + `/deadman` holds the box's sealed "I am gone" row and posts it when the + claim stays down past the deadline. The handle is the capability today's + clients already present, so an app, a watch or a box switches by changing + the origin and nothing else. A box may also send `X-FTW-Box-Key` and + `X-FTW-Box-Sig`; a proof that does not verify is refused, and a join with + neither header is still admitted, because current boxes send none. This + path is not `/v1/box`. That socket's downlink is still only `challenge`, + `hello_ok`, `watch`, `resync` and `purged`. Not built yet: @@ -88,6 +108,7 @@ Not built yet: |---|---|---| | C6 | Tiles | `internal/server` | | C7–C8 | Statistics | not started | +| — | Anonymous fleet ping (`POST /fleet`) | still only on `relay.ftw.energy`; Core's default URL is not this origin | ## How it holds a home @@ -96,6 +117,10 @@ box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) │ ◀── /v1/app/once ◀──Noise IK in one POST── watch │ ◀── /v1/now ◀──token──── widget └────▶ registry (SQLite: key, roster, tombstones, suspensions, APNs, token hashes) + +box ──wss, dials out──▶ /r/…/box ──┐ +app ──wss──────────▶ /r/…/app ─────┼──▶ relay (blind; frames not stored) ──▶ the other peer +watch ──POST──────▶ /x/… ─────────┘ ``` - **The home id is a hash of the box's key.** It comes from the key that @@ -107,13 +132,15 @@ box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) principal reads only `live`, `prices` and `hb` within that same scope. There is no listing and no principal for staff; `TestNoRouteReadsAHome` checks every route. -- **A phone or a watch can narrow and nothing else.** A phone can say - whether it is looking, and an owner can suspend a device in the roster. A - watch can read, ask for its token and register for push. Commands go to - the box over the relay, never through here. -- **The service sends five things.** `challenge`, `hello_ok`, `watch`, - `resync` and `purged` are the whole downlink, and none asks the box to act. - The box closes the uplink on anything else. +- **A phone or a watch can narrow and nothing else** on the readable path. + A phone can say whether it is looking, and an owner can suspend a device + in the roster. A watch can read, ask for its token and register for push. + A command is Noise between the app and the box. It crosses this process + on `/r/` and `/x/` and is not written into the home. +- **The service sends five things on `/v1/box`.** `challenge`, `hello_ok`, + `watch`, `resync` and `purged` are the whole downlink, and none asks the + box to act. The box closes the uplink on anything else. `ready` and `gone` + are words of the command path, and they are not sent on `/v1/box`. - **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. - **Root on the host can read memory.** The privacy notice says so. diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 2b47263..903529d 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -1,6 +1,7 @@ // Command ftwcloudd is FTW Cloud: it keeps a signed, readable copy of each -// home that turned it on, in memory, and serves it to that home's phones. It -// reads and never controls. See README.md. +// home that turned it on, in memory, and serves it to that home's phones. +// On that path it reads and never controls. On /r/, /x/ and /deadman it +// forwards command frames it cannot open. See README.md. // // ftwcloudd keygen -out service.key make the service key; prints its public half // ftwcloudd keygen -noise -out noise.key make the Noise key phones dial; prints its public half @@ -35,6 +36,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/push" "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/relay" "github.com/srcfl/ftw-cloud/internal/server" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" @@ -100,7 +102,15 @@ func serve(args []string) error { st.UseSuspensions(reg.Suspended) up := uplink.New(st, reg, key, log) tokens := token.New(st, reg, log) - srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Log: log} + // The command path is the blind relay. It shares this process and this + // listener, and nothing else: it cannot see the store. Dead-man rows + // sit beside the registry as sealed push ciphertext. + commands := relay.New(relay.Options{ + Log: log, + DeadmanPath: filepath.Join(*stateDir, "deadman.json"), + BoxKeys: reg, + }) + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Relay: commands, Log: log} // Alerts: the box's signed events to its phones through Apple, and // box.unreachable when a box is away. Widgets: a push with no data when @@ -179,6 +189,7 @@ func serve(args []string) error { defer stop() go srv.Upkeep(ctx, time.Minute) go pushSvc.Run(ctx) + go commands.Run(ctx) if *devPush != "" { if err := serveDevPush(ctx, *devPush, reg, pushSvc, sender, log); err != nil { return err diff --git a/deploy/README.md b/deploy/README.md index f5db043..7801e9e 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -42,7 +42,8 @@ Every AWS resource carries the tag `Project=ftw-cloud`. `/opt/ftw-cloud` is checked out at the commit that image was built from and supplies `compose.yml`. - The Docker volume `ftw-cloud_ftw-cloud-state` holds the SQLite registry - (home ids, box keys, signed rosters, tombstones) and the ACME cache, on the + (home ids, box keys, signed rosters, tombstones), the ACME cache, and + `deadman.json` (sealed push ciphertext for the command path), on the encrypted root disk. No readings. No backup. ## The service key @@ -196,6 +197,34 @@ keeps the service key for 30 days after `delete-secret`, so a mistaken teardown can still be undone, and then deletes it. Nothing else holds FTW Cloud data; the images on `ghcr.io` hold only code. +## Remote commands + +The same process answers the Frankfurt relay's paths: `/r/`, `/x/` and +`/deadman`. Frames are the bytes the app and the box already exchange, and +the service cannot read them. This file describes the host. It does not +change DNS, and it does not stop `relay.ftw.energy`. A client switches by +building against `cloud.ftw.energy`. Until the box, the app and the watch do +that, Frankfurt is still where commands go. + +Deploying the image that contains the path is a later step, by the owner. +Nothing here changes the running container, its command, its ports or its +secrets. + +Before Frankfurt can be turned off: + +- Box, webapp and watch builds must dial this origin. The paths and the + frames stay the same; the constants live in `srcfl/ftw` and `srcfl/ftw-webapp`. +- `POST /fleet` is not served here. Core's default fleet URL is + `https://relay.ftw.energy/fleet`, and it is not derived from the command + socket's origin. +- Dead-man rows already stored in Frankfurt are not copied. A box posts its + row again when it connects, so a box that is offline at the cutover has no + switch until it returns. Frankfurt's file stays until that has happened. +- A box-key proof is accepted and not required. Requiring it waits on boxes + sending `X-FTW-Box-Key` and `X-FTW-Box-Sig`. +- ADR 0008 in `srcfl/ftw` still says commands stay on the relay. That text + is the decision record to revise. + ## Not here yet The APNs key ([apns.md](apns.md)) and the Noise key phones dial diff --git a/deploy/compose.yml b/deploy/compose.yml index 5504f2a..d6f64c3 100644 --- a/deploy/compose.yml +++ b/deploy/compose.yml @@ -25,7 +25,8 @@ services: # /source publishes it, so anyone can compare it with what CI built. FTW_CLOUD_IMAGE: ${FTW_CLOUD_IMAGE} volumes: - # The registry and certificates. No readings are ever written here. + # The registry, certificates, and sealed dead-man rows (deadman.json). + # No readings are ever written here. - ftw-cloud-state:/state # Apple's push key, read from Parameter Store into tmpfs at boot by # deploy/apns.sh. Empty means no alerts, and the service runs. diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 056f8db..066eb7a 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -6,11 +6,13 @@ package registry import ( + "bytes" "database/sql" "errors" "fmt" "time" + "github.com/srcfl/ftw/go/protocol/cloudwire" _ "modernc.org/sqlite" ) @@ -108,6 +110,20 @@ func (r *Registry) Claim(home string, pk []byte, now time.Time) error { // ErrPurged is a home whose box asked for it to be deleted. var ErrPurged = errors.New("registry: home was purged") +// RecognizesBoxKey reports whether pk is the key a live home registered. +// A purged home is not recognised. The command path uses this only to +// check a box-key proof; it is not a way to read the home. +func (r *Registry) RecognizesBoxKey(pk []byte) bool { + if len(pk) != cloudwire.PublicKeyBytes { + return false + } + h, ok, err := r.Get(cloudwire.HomeID(pk)) + if err != nil || !ok { + return false + } + return bytes.Equal(h.PK, pk) +} + // Get returns one home. func (r *Registry) Get(home string) (Home, bool, error) { var h Home diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go index 6a9a685..5f2e0a9 100644 --- a/internal/registry/registry_test.go +++ b/internal/registry/registry_test.go @@ -1,10 +1,15 @@ package registry import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" "errors" "path/filepath" "testing" "time" + + "github.com/srcfl/ftw/go/protocol/cloudwire" ) func open(t *testing.T) *Registry { @@ -35,6 +40,38 @@ func TestAHomeIsClaimedOnceAndKeepsItsRoster(t *testing.T) { } } +func TestRecognizesBoxKeyIsTheRegisteredKeyOnly(t *testing.T) { + r := open(t) + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + pk, err := key.PublicKey.Bytes() + if err != nil { + t.Fatal(err) + } + if r.RecognizesBoxKey(pk) { + t.Fatal("an unregistered key was recognised") + } + if err := r.Claim(cloudwire.HomeID(pk), pk, time.UnixMilli(1_700_000_000_000)); err != nil { + t.Fatal(err) + } + if !r.RecognizesBoxKey(pk) { + t.Fatal("the registered key was not recognised") + } + other := append([]byte(nil), pk...) + other[len(other)-1] ^= 0xff + if r.RecognizesBoxKey(other) || r.RecognizesBoxKey(nil) || r.RecognizesBoxKey(pk[:8]) { + t.Fatal("a key that is not the home's was recognised") + } + if err := r.Purge(cloudwire.HomeID(pk), nil, time.UnixMilli(1_700_000_000_000)); err != nil { + t.Fatal(err) + } + if r.RecognizesBoxKey(pk) { + t.Fatal("a purged home's key was still recognised") + } +} + // A purged home leaves only a tombstone, and its box cannot claim it again. func TestAPurgeLeavesATombstone(t *testing.T) { r := open(t) diff --git a/internal/relay/boxkey.go b/internal/relay/boxkey.go new file mode 100644 index 0000000..f0522ce --- /dev/null +++ b/internal/relay/boxkey.go @@ -0,0 +1,96 @@ +package relay + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "math/big" + "net/http" + "strconv" +) + +// BoxKeys is the registry's answer to one question: does this P-256 key +// belong to a home that has turned FTW Cloud on? The command path never +// asks it for a reading, a roster or a home id. +type BoxKeys interface { + RecognizesBoxKey(pk []byte) bool +} + +// boxProofInput is what a box signs to prove its cloud key on a join. +// +// Existing clients send no proof. A join without one is admitted on the +// rendezvous handle, which is the capability those clients already present +// and what an endpoint-only switch requires. A join that does send one is +// admitted only when the signature verifies. The handle stays in the path +// either way: the proof does not name the room. +func boxProofInput(epoch int64, handle, role string) []byte { + return []byte("ftw.cloud.relay.v1\n" + strconv.FormatInt(epoch, 10) + "\n" + handle + "\n" + role) +} + +// proofResult is the outcome of looking at the two optional headers. +type proofResult int + +const ( + proofAbsent proofResult = iota + proofOK + proofBad +) + +// checkBoxProof verifies X-FTW-Box-Key and X-FTW-Box-Sig when the box sends +// them. App joins never carry a cloud key; the headers are ignored there. +func checkBoxProof(r *http.Request, keys BoxKeys, epoch int64, handle, role string) proofResult { + if role == roleApp { + return proofAbsent + } + keyHdr := r.Header.Get("X-FTW-Box-Key") + sigHdr := r.Header.Get("X-FTW-Box-Sig") + if keyHdr == "" && sigHdr == "" { + return proofAbsent + } + if keyHdr == "" || sigHdr == "" { + return proofBad + } + pk, err := base64.RawURLEncoding.DecodeString(keyHdr) + if err != nil { + return proofBad + } + sig, err := base64.RawURLEncoding.DecodeString(sigHdr) + if err != nil || len(sig) != 64 { + return proofBad + } + pub, err := ecdsa.ParseUncompressedPublicKey(elliptic.P256(), pk) + if err != nil { + return proofBad + } + sum := sha256.Sum256(boxProofInput(epoch, handle, role)) + if !ecdsa.Verify(pub, sum[:], new(big.Int).SetBytes(sig[:32]), new(big.Int).SetBytes(sig[32:])) { + return proofBad + } + if keys != nil && !keys.RecognizesBoxKey(pk) { + return proofBad + } + return proofOK +} + +// signBoxProof is the client half, used by tests and available to a later +// box build. It returns the two header values. +func signBoxProof(key *ecdsa.PrivateKey, epoch int64, handle, role string) (keyHdr, sigHdr string, err error) { + pub, err := key.PublicKey.Bytes() + if err != nil { + return "", "", err + } + sum := sha256.Sum256(boxProofInput(epoch, handle, role)) + r, s, err := ecdsa.Sign(rand.Reader, key, sum[:]) + if err != nil { + return "", "", err + } + sig := make([]byte, 64) + r.FillBytes(sig[:32]) + s.FillBytes(sig[32:]) + return base64.RawURLEncoding.EncodeToString(pub), base64.RawURLEncoding.EncodeToString(sig), nil +} + +// boxKeyReason is the close reason for a proof the service will not accept. +func boxKeyReason() string { return "box key" } diff --git a/internal/relay/deadman.go b/internal/relay/deadman.go new file mode 100644 index 0000000..59dd9bf --- /dev/null +++ b/internal/relay/deadman.go @@ -0,0 +1,374 @@ +package relay + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/base64" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "regexp" + "strings" + "sync" + "time" +) + +// The dead man's switch is the one thing this path keeps. A box can push +// every notification except the one that says it is gone, so it leaves a +// sealed row here: an opaque id, a push endpoint, a ciphertext encrypted at +// home, a deadline, and a delivery authorisation the box already signed. +// While a socket claims the id the switch is held. When the claim stays +// dropped past the deadline, the row is posted once. +// +// The ciphertext is RFC 8291. This file has no key for it and does not try +// to read it. The id is 128 bits the box derived from a secret that never +// arrives here. + +const ( + deadmanRefireFloor = 30 * time.Minute + deadmanMaxCT = 8192 + deadmanMaxAuth = 512 + deadmanMaxBody = 16 << 10 +) + +var deadmanID = regexp.MustCompile(`^[0-9a-f]{32}$`) + +// deadmanClaimID reads the one text frame the box may send on its command +// socket. Anything else on that socket is still a protocol error. +func deadmanClaimID(text string) (string, bool) { + const prefix = "deadman " + if len(text) != len(prefix)+32 || !strings.HasPrefix(text, prefix) { + return "", false + } + id := text[len(prefix):] + if !deadmanID.MatchString(id) { + return "", false + } + return id, true +} + +// PushPost delivers one fired row. Tests replace it. Production posts HTTPS +// and refuses a private or link-local address. +type PushPost func(ctx context.Context, endpoint string, body []byte, header http.Header) (status int, err error) + +type deadmanRow struct { + ID string `json:"id"` + Endpoint string `json:"endpoint"` + CT string `json:"ct"` + DeadlineS int `json:"deadline_s"` + Auth string `json:"auth,omitempty"` + LastFiredAtMs int64 `json:"lastFiredAtMs,omitempty"` +} + +type deadman struct { + mu sync.Mutex + path string + now func() time.Time + post PushPost + log func(status int, failed bool) + rows map[string]*deadmanRow + claims map[string]int + armed map[string]time.Time +} + +func newDeadman(path string, now func() time.Time, post PushPost, log func(int, bool)) *deadman { + if post == nil { + post = postPush + } + d := &deadman{ + path: path, + now: now, + post: post, + log: log, + rows: map[string]*deadmanRow{}, + claims: map[string]int{}, + armed: map[string]time.Time{}, + } + d.load() + // A row that outlived a restart has no claim. Arm it for a full + // deadline so a live box can reconnect first and a dead one is still + // spoken for. + at := now() + for _, row := range d.rows { + d.armed[row.ID] = at.Add(time.Duration(row.DeadlineS) * time.Second) + } + return d +} + +type deadmanIn struct { + ID string `json:"id"` + Endpoint string `json:"endpoint"` + CT string `json:"ct"` + Auth string `json:"auth"` + DeadlineS int `json:"deadline_s"` +} + +// putError validates a POST body the way the Frankfurt relay does. The +// sentence is the JSON error field. +func putError(b deadmanIn, ctLen int) string { + if !deadmanID.MatchString(b.ID) { + return "id must be 32 hex chars" + } + if !strings.HasPrefix(b.Endpoint, "https://") { + return "endpoint must be an https url" + } + if ctLen == 0 { + return "ct must be base64" + } + if ctLen > deadmanMaxCT { + return "ct too large" + } + if b.DeadlineS < 60 || b.DeadlineS > 86400 { + return "deadline_s must be 60..86400" + } + if len(b.Auth) > deadmanMaxAuth { + return "auth must be a short string" + } + return "" +} + +func (d *deadman) put(b deadmanIn) { + d.mu.Lock() + defer d.mu.Unlock() + row := &deadmanRow{ID: b.ID, Endpoint: b.Endpoint, CT: b.CT, DeadlineS: b.DeadlineS, Auth: b.Auth} + if prev := d.rows[b.ID]; prev != nil { + row.LastFiredAtMs = prev.LastFiredAtMs + } + d.rows[b.ID] = row + if d.claims[b.ID] > 0 { + delete(d.armed, b.ID) + } else { + d.armed[b.ID] = d.now().Add(time.Duration(b.DeadlineS) * time.Second) + } + d.save() +} + +func (d *deadman) remove(id string) { + d.mu.Lock() + defer d.mu.Unlock() + delete(d.rows, id) + delete(d.armed, id) + d.save() +} + +// claim marks id held by one more socket. An id with no row yet is +// remembered: the POST may still be in flight beside the socket. +func (d *deadman) claim(id string) { + d.mu.Lock() + defer d.mu.Unlock() + d.claims[id]++ + delete(d.armed, id) +} + +// release drops one socket's hold. The last one starts the countdown. +func (d *deadman) release(id string) { + d.mu.Lock() + defer d.mu.Unlock() + n := d.claims[id] - 1 + if n > 0 { + d.claims[id] = n + return + } + delete(d.claims, id) + if row := d.rows[id]; row != nil { + d.armed[id] = d.now().Add(time.Duration(row.DeadlineS) * time.Second) + } +} + +type deadmanView struct { + rows, claimed, armed int +} + +func (d *deadman) inspect() deadmanView { + d.mu.Lock() + defer d.mu.Unlock() + return deadmanView{len(d.rows), len(d.claims), len(d.armed)} +} + +// beat fires whatever is due. The post happens off this call so a slow push +// service cannot stall the heartbeat. +func (d *deadman) beat() { + d.mu.Lock() + now := d.now() + type job struct { + row deadmanRow + } + var jobs []job + for id, at := range d.armed { + if at.After(now) { + continue + } + delete(d.armed, id) + row := d.rows[id] + if row == nil { + continue + } + if row.LastFiredAtMs != 0 && now.Sub(time.UnixMilli(row.LastFiredAtMs)) < deadmanRefireFloor { + continue + } + row.LastFiredAtMs = now.UnixMilli() + jobs = append(jobs, job{row: *row}) + } + if len(jobs) > 0 { + d.save() + } + post := d.post + d.mu.Unlock() + for _, j := range jobs { + go d.fire(post, j.row) + } +} + +func (d *deadman) fire(post PushPost, row deadmanRow) { + body, err := base64.StdEncoding.DecodeString(row.CT) + if err != nil { + d.log(0, true) + return + } + h := make(http.Header) + h.Set("TTL", "86400") + h.Set("Content-Encoding", "aes128gcm") + h.Set("Content-Type", "application/octet-stream") + h.Set("Urgency", "high") + if row.Auth != "" { + h.Set("Authorization", row.Auth) + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + status, err := post(ctx, row.Endpoint, body, h) + d.log(status, err != nil) + if err == nil && (status == http.StatusNotFound || status == http.StatusGone) { + d.remove(row.ID) + } +} + +func (d *deadman) load() { + if d.path == "" { + return + } + raw, err := os.ReadFile(d.path) + if err != nil { + return + } + var rows []deadmanRow + if json.Unmarshal(raw, &rows) != nil { + return + } + for i := range rows { + if deadmanID.MatchString(rows[i].ID) { + d.rows[rows[i].ID] = &rows[i] + } + } +} + +func (d *deadman) save() { + if d.path == "" { + return + } + rows := make([]deadmanRow, 0, len(d.rows)) + for _, row := range d.rows { + rows = append(rows, *row) + } + raw, err := json.Marshal(rows) + if err != nil { + return + } + if err := os.MkdirAll(filepath.Dir(d.path), 0o755); err != nil { + return + } + tmp := d.path + ".tmp" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return + } + _ = os.Rename(tmp, d.path) +} + +// postPush is the production delivery. It refuses a URL that is not public +// HTTPS, including after a redirect, so a row cannot point this process at +// the metadata service or a host on the machine's own network. +func postPush(ctx context.Context, endpoint string, body []byte, header http.Header) (int, error) { + if err := allowPushURL(endpoint); err != nil { + return 0, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return 0, err + } + req.Header = header.Clone() + client := &http.Client{ + Timeout: 15 * time.Second, + Transport: &http.Transport{ + Proxy: nil, + TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}, + DialContext: dialPublic, + }, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 3 { + return errors.New("relay: too many redirects") + } + return allowPushURL(req.URL.String()) + }, + } + res, err := client.Do(req) + if err != nil { + return 0, err + } + defer res.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(res.Body, 1024)) + return res.StatusCode, nil +} + +func allowPushURL(raw string) error { + u, err := url.Parse(raw) + if err != nil || u.Scheme != "https" || u.Host == "" || u.User != nil { + return errors.New("relay: push url refused") + } + host := u.Hostname() + if ip := net.ParseIP(host); ip != nil { + if !publicIP(ip) { + return errors.New("relay: push url refused") + } + return nil + } + return nil +} + +func dialPublic(ctx context.Context, network, addr string) (net.Conn, error) { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + ips, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil { + return nil, err + } + if len(ips) == 0 { + return nil, errors.New("relay: push url refused") + } + for _, ip := range ips { + if !publicIP(ip.IP) { + return nil, errors.New("relay: push url refused") + } + } + var d net.Dialer + return d.DialContext(ctx, network, net.JoinHostPort(ips[0].IP.String(), port)) +} + +func publicIP(ip net.IP) bool { + if ip == nil || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || + ip.IsMulticast() || ip.IsUnspecified() { + return false + } + // Carrier-grade NAT. Go's IsPrivate does not cover it, and a push + // service is not there. + if v4 := ip.To4(); v4 != nil && v4[0] == 100 && v4[1]&0xc0 == 64 { + return false + } + return true +} diff --git a/internal/relay/deadman_test.go b/internal/relay/deadman_test.go new file mode 100644 index 0000000..4a14644 --- /dev/null +++ b/internal/relay/deadman_test.go @@ -0,0 +1,177 @@ +package relay + +import ( + "bytes" + "context" + "encoding/base64" + "net" + "net/http" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +type posted struct { + endpoint string + body []byte +} + +func TestDeadmanFiresTheSealedBytesOnce(t *testing.T) { + clk := &clock{t: testTime()} + var mu sync.Mutex + var got []posted + r := New(Options{ + Log: discardLog(), + Now: clk.Now, + Post: func(_ context.Context, endpoint string, body []byte, h http.Header) (int, error) { + if h.Get("Content-Encoding") != "aes128gcm" || h.Get("Authorization") != "vapid test" { + t.Errorf("headers %v", h) + } + mu.Lock() + got = append(got, posted{endpoint, append([]byte(nil), body...)}) + mu.Unlock() + return http.StatusCreated, nil + }, + }) + ct := []byte("sealed-farewell") + const id = "0123456789abcdef0123456789abcdef" + r.dead.put(deadmanIn{ + ID: id, Endpoint: "https://push.example/abc", + CT: base64.StdEncoding.EncodeToString(ct), Auth: "vapid test", DeadlineS: 60, + }) + r.dead.claim(id) + r.dead.release(id) + + clk.Set(testTime().Add(61 * time.Second)) + r.beat() + waitPosts(t, &mu, &got, 1) + if got[0].endpoint != "https://push.example/abc" || !bytes.Equal(got[0].body, ct) { + t.Fatalf("fired %+v", got[0]) + } + + // Claimed and released again inside the half-hour floor. One outage + // is one message. + r.dead.claim(id) + r.dead.release(id) + clk.Set(testTime().Add(61*time.Second + 61*time.Second)) + r.beat() + time.Sleep(30 * time.Millisecond) + mu.Lock() + n := len(got) + mu.Unlock() + if n != 1 { + t.Fatalf("fired %d times", n) + } +} + +func TestDeadmanHTTPAndARestart(t *testing.T) { + clk := &clock{t: testTime()} + var mu sync.Mutex + var got []posted + post := func(_ context.Context, endpoint string, body []byte, _ http.Header) (int, error) { + mu.Lock() + got = append(got, posted{endpoint, append([]byte(nil), body...)}) + mu.Unlock() + return http.StatusCreated, nil + } + path := filepath.Join(t.TempDir(), "deadman.json") + _, srv := startRelay(t, Options{Now: clk.Now, Post: post, DeadmanPath: path}) + ct := []byte("still-sealed") + const id = "abcdef0123456789abcdef0123456789" + raw := `{"id":"` + id + `","endpoint":"https://push.example/abc","ct":"` + + base64.StdEncoding.EncodeToString(ct) + `","auth":"vapid test","deadline_s":60}` + res, err := http.Post(srv.URL+"/deadman", "application/json", strings.NewReader(raw)) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusNoContent { + t.Fatalf("post %d", res.StatusCode) + } + res, err = http.Post(srv.URL+"/deadman", "application/json", strings.NewReader(`[]`)) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusBadRequest { + t.Fatalf("bad body %d", res.StatusCode) + } + + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + if err := box.WriteMessage(websocket.TextMessage, []byte("deadman "+id)); err != nil { + t.Fatal(err) + } + + // A restart forgets the claim and arms a fresh deadline, so a box that + // is in the middle of reconnecting is not declared gone. + r2 := New(Options{Log: discardLog(), Now: clk.Now, Post: post, DeadmanPath: path}) + r2.beat() + time.Sleep(20 * time.Millisecond) + mu.Lock() + n := len(got) + mu.Unlock() + if n != 0 { + t.Fatalf("a restart fired %d times", n) + } + clk.Set(testTime().Add(61 * time.Second)) + r2.beat() + waitPosts(t, &mu, &got, 1) + if !bytes.Equal(got[0].body, ct) { + t.Fatalf("restart fired %q", got[0].body) + } + + req, _ := http.NewRequest(http.MethodDelete, srv.URL+"/deadman/"+id, nil) + res, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusNoContent { + t.Fatalf("delete %d", res.StatusCode) + } +} + +func TestPushURLsStayOnThePublicInternet(t *testing.T) { + for _, raw := range []string{ + "http://push.example/x", + "https://127.0.0.1/x", + "https://10.1.2.3/x", + "https://169.254.169.254/latest", + "https://100.64.1.1/x", + "https://user:pass@push.example/x", + } { + if err := allowPushURL(raw); err == nil { + t.Fatalf("%s was allowed", raw) + } + } + if err := allowPushURL("https://push.example/token"); err != nil { + t.Fatal(err) + } + for _, s := range []string{"100.64.0.1", "192.168.1.1", "::1", "169.254.1.1"} { + if publicIP(net.ParseIP(s)) { + t.Fatalf("%s looked public", s) + } + } +} + +func waitPosts(t *testing.T, mu *sync.Mutex, got *[]posted, n int) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + mu.Lock() + have := len(*got) + mu.Unlock() + if have >= n { + return + } + time.Sleep(5 * time.Millisecond) + } + mu.Lock() + have := len(*got) + mu.Unlock() + t.Fatalf("got %d posts, want %d", have, n) +} diff --git a/internal/relay/limits.go b/internal/relay/limits.go new file mode 100644 index 0000000..a8917ee --- /dev/null +++ b/internal/relay/limits.go @@ -0,0 +1,87 @@ +package relay + +import ( + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/binary" + "time" +) + +// bucket is a per-socket allowance. It lives on the socket and dies with +// it, so a closed connection leaves no keyed record. +type bucket struct { + tokens float64 + capacity float64 + perSec float64 + last time.Time +} + +func newBucket(capacity, perSec float64, now time.Time) *bucket { + return &bucket{tokens: capacity, capacity: capacity, perSec: perSec, last: now} +} + +// take reports whether cost was affordable. A false means the caller is over +// and the connection should close: dropping one frame would leave a hole in +// a stream the session above would have to recover from anyway. +func (b *bucket) take(cost float64, now time.Time) bool { + if elapsed := now.Sub(b.last).Seconds(); elapsed > 0 { + b.tokens += elapsed * b.perSec + if b.tokens > b.capacity { + b.tokens = b.capacity + } + } + b.last = now + if b.tokens < cost { + return false + } + b.tokens -= cost + return true +} + +// attempts counts connection attempts per window without keeping addresses. +// +// The counts sit in a fixed array indexed by a keyed hash. The key is drawn +// at the start of each window and thrown away with the counts, so the array +// cannot be turned back into a list of who connected. Slots collide, and a +// heavy caller then shares a limit with whoever hashed beside them. +type attempts struct { + limit int + window time.Duration + counts []uint16 + key []byte + ends time.Time +} + +func newAttempts(limit int, window time.Duration, slots int, now time.Time) *attempts { + a := &attempts{limit: limit, window: window, counts: make([]uint16, slots)} + a.roll(now) + return a +} + +func (a *attempts) allow(addr string, now time.Time) bool { + if !now.Before(a.ends) { + a.roll(now) + } + slot := a.slot(addr) + if int(a.counts[slot]) >= a.limit { + return false + } + a.counts[slot]++ + return true +} + +func (a *attempts) roll(now time.Time) { + clear(a.counts) + key := make([]byte, 32) + _, _ = rand.Read(key) + a.key = key + a.ends = now.Add(a.window) +} + +func (a *attempts) slot(addr string) int { + mac := hmac.New(sha256.New, a.key) + _, _ = mac.Write([]byte(addr)) + sum := mac.Sum(nil) + return int(binary.BigEndian.Uint32(sum[:4]) % uint32(len(a.counts))) +} diff --git a/internal/relay/limits_test.go b/internal/relay/limits_test.go new file mode 100644 index 0000000..5d1867e --- /dev/null +++ b/internal/relay/limits_test.go @@ -0,0 +1,72 @@ +package relay + +import ( + "testing" + "time" +) + +func TestBucketRefusesPastItsCapacity(t *testing.T) { + b := newBucket(3, 1, time.Unix(0, 0)) + now := time.Unix(0, 0) + for i := 0; i < 3; i++ { + if !b.take(1, now) { + t.Fatal("capacity") + } + } + if b.take(1, now) { + t.Fatal("spent bucket allowed another") + } + if b.take(1, now.Add(50*time.Millisecond)) { + t.Fatal("a refill that has not arrived was spent") + } + if !b.take(1, now.Add(time.Second)) { + t.Fatal("a second did not refill one token") + } + if !b.take(3, now.Add(time.Hour)) { + t.Fatal("an idle hour did not refill to capacity") + } + if b.take(1, now.Add(time.Hour)) { + t.Fatal("an idle hour bought more than the capacity") + } +} + +func TestAttemptsForgetTheWindowAndKeepNoList(t *testing.T) { + a := newAttempts(3, time.Second, 1024, time.Unix(0, 0)) + now := time.Unix(0, 0) + for i := 0; i < 3; i++ { + if !a.allow("198.51.100.1", now) { + t.Fatal("within the window") + } + } + if a.allow("198.51.100.1", now) { + t.Fatal("over the window") + } + if !a.allow("198.51.100.1", now.Add(time.Second)) { + t.Fatal("the next window did not start clean") + } + + fresh := newAttempts(1, time.Second, 1024, now) + allowed := 0 + for i := 0; i < 100; i++ { + if fresh.allow("198.51.100."+itoa(i), now) { + allowed++ + } + } + if allowed < 85 { + t.Fatalf("distinct callers allowed %d, want most of them in their own slots", allowed) + } +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [4]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/internal/relay/protocol.go b/internal/relay/protocol.go new file mode 100644 index 0000000..bdfb7cc --- /dev/null +++ b/internal/relay/protocol.go @@ -0,0 +1,207 @@ +// The command path's wire contract, copied from srcfl/ftw-webapp +// relay/src/protocol.ts and relay/src/epoch.ts. +// +// The box, the webapp and a watch already speak this. They switch hosts by +// changing the origin and nothing else, so these numbers are not ours to +// improve. relay_test.go holds them against the values those clients +// duplicate. +package relay + +import ( + "net" + "net/http" + "strconv" + "strings" + "time" +) + +const ( + // CloseBadJoin is a malformed join, text on a binary channel, or an + // oversize message. + CloseBadJoin = 4400 + // CloseEpoch is a join that named the wrong hour. The reason is the + // right epoch, and nothing else. + CloseEpoch = 4409 + // CloseRotated is the hour turning. The reason is the new epoch. + CloseRotated = 4410 + // CloseBusy is a rate limit, a full room, or the process ceiling. + CloseBusy = 4429 + + // CtrlReady means the other side of the room is here. + CtrlReady = "ready" + // CtrlGone means it left. The socket stays up. + CtrlGone = "gone" + + // HandleChars is a rendezvous handle: 128 bits, lowercase hex. + HandleChars = 32 + // EpochMs is how long one handle lives. An hour. + EpochMs = 3_600_000 + + // ExchangeIDBytes is the id this service draws in front of a watch's + // request. The box copies it back in front of the answer. It names + // nothing outside this process. + ExchangeIDBytes = 16 +) + +const ( + roleBox = "box" + roleApp = "app" + roleExchange = "exchange" + + // rotateSpread is how long an hour's turn is spread over, so rooms do + // not all reconnect on the same second. + rotateSpread = 5 * time.Minute + + maxStreamsPerRoom = 4 + maxSockets = 4096 + maxSocketsPerAddr = 16 + maxFrameBytes = 64 << 10 + maxBufferedWrite = 10 * time.Second + heartbeatEvery = 15 * time.Second + frameCapacity = 240 + framesPerSecond = 60 + byteCapacity = 4 << 20 + bytesPerSecond = 512 << 10 + attemptLimit = 30 + attemptWindow = time.Minute + attemptSlots = 1024 + + // Exchange sizes, from docs/watch-exchange.md. Message 1 is 608 bytes + // and one transport message is 536. Nothing else is a request. + exchangeOpenBytes = 608 + exchangeRequestBytes = 536 + exchangeMaxReply = 4096 + exchangePerMinute = 30 + exchangeOpenPerRoom = 2 + exchangeOpenProcess = 1024 + exchangeTimeout = 10 * time.Second +) + +// CurrentEpoch is the hour the service is in. Peers guess from their own +// clocks and put the guess in the path; a wrong guess is answered with this +// number in the close reason. +func CurrentEpoch(now time.Time) int64 { + ms := now.UnixMilli() + // Floor, matching the app's Math.floor. Go divides toward zero, and a + // clock before 1970 would otherwise disagree with every client. + if ms < 0 { + return -((-ms + EpochMs - 1) / EpochMs) + } + return ms / EpochMs +} + +// RotationOffset is where in the spread this room's turn falls. It comes +// from the handle, so it needs no timer and two rooms do not share a turn +// by waiting for the same tick. +func RotationOffset(handle string, spread time.Duration) time.Duration { + if len(handle) < 4 || spread <= 0 { + return 0 + } + bucket, err := strconv.ParseUint(handle[:4], 16, 16) + if err != nil { + return 0 + } + return time.Duration(float64(spread) * (float64(bucket) / 0x10000)) +} + +type join struct { + epoch int64 + handle string + role string +} + +// parseJoin reads `/r///` and nothing else. +func parseJoin(path string) (join, bool) { + parts := splitPath(path) + if len(parts) != 4 || parts[0] != "r" { + return join{}, false + } + epoch, ok := parseEpoch(parts[1]) + if !ok || !handleOK(parts[2]) { + return join{}, false + } + switch parts[3] { + case roleBox, roleApp, roleExchange: + default: + return join{}, false + } + return join{epoch: epoch, handle: parts[2], role: parts[3]}, true +} + +type exchangeTarget struct { + epoch int64 + handle string +} + +// parseExchange reads `/x//` and nothing else. +func parseExchange(path string) (exchangeTarget, bool) { + parts := splitPath(path) + if len(parts) != 3 || parts[0] != "x" { + return exchangeTarget{}, false + } + epoch, ok := parseEpoch(parts[1]) + if !ok || !handleOK(parts[2]) { + return exchangeTarget{}, false + } + return exchangeTarget{epoch: epoch, handle: parts[2]}, true +} + +func splitPath(path string) []string { + raw := strings.Split(path, "/") + out := make([]string, 0, len(raw)) + for _, p := range raw { + if p != "" { + out = append(out, p) + } + } + return out +} + +func parseEpoch(s string) (int64, bool) { + // Reject a sign and leading junk. ParseInt accepts "+1" and the + // clients' own check does not treat that as an epoch. + if s == "" || s[0] < '0' || s[0] > '9' { + return 0, false + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil || n < 0 { + return 0, false + } + return n, true +} + +func handleOK(s string) bool { + if len(s) != HandleChars { + return false + } + for i := 0; i < len(s); i++ { + c := s[i] + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return false + } + } + return true +} + +// clientIP is the address a limit is counted against. The port is not part +// of it: every connection has a new port, and a limit that included it +// would never see the same caller twice. +// +// The forwarded header is ignored unless trust is set. ftwcloudd ends TLS +// itself, so nothing trusted sits in front and the header is a value the +// client chooses. trust exists so the test can show that choice. +func clientIP(r *http.Request, trust bool) string { + if trust { + if chain := r.Header.Get("X-Forwarded-For"); chain != "" { + parts := strings.Split(chain, ",") + if last := strings.TrimSpace(parts[len(parts)-1]); last != "" { + return last + } + } + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return r.RemoteAddr + } + return host +} diff --git a/internal/relay/relay.go b/internal/relay/relay.go new file mode 100644 index 0000000..f2214ae --- /dev/null +++ b/internal/relay/relay.go @@ -0,0 +1,847 @@ +// Package relay is the remote-command path: the blind relay, in this process. +// +// A box dials out and a phone, a browser or a watch meets it under a +// rendezvous handle that rotates every hour. Frames are end-to-end Noise +// between those peers. This package forwards the bytes and reads two things +// off them: their length, and, on the box's exchange socket, the 16-byte id +// it wrote there itself. It has no key that opens a command, and it does not +// import the Noise session or the readable store. +// +// The path is separate from /v1/box. That socket carries signed cloud +// envelopes and a closed downlink (challenge, hello_ok, watch, resync, +// purged); the box closes it on anything else. Command frames are not those +// envelopes, and ready/gone are not those five messages. Multiplexing the +// two would either make this service parse commands or make the box drop +// the uplink. Clients that already speak the Frankfurt relay switch by +// changing the origin alone: /r/// and +// POST /x// are the same paths, on this server. +package relay + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" +) + +// Options configures the path. Zero values are the Frankfurt limits. +type Options struct { + Log *slog.Logger + Now func() time.Time + // DeadmanPath is the sealed-row file. Empty keeps the rows in memory, + // which is what tests want. The file holds push ciphertext, not readings. + DeadmanPath string + // Post delivers a fired dead-man row. Nil posts HTTPS and refuses a + // private address. + Post PushPost + // BoxKeys, when set, is consulted for a join that presents a box-key + // proof. A proof from a key it does not recognise is refused. + BoxKeys BoxKeys + // RequireBoxKey refuses a box or exchange join that presents no proof. + // Leave it false until boxes send the proof: existing clients join with + // the handle alone, and requiring the proof before they send it would + // cut every house off. + RequireBoxKey bool +} + +// Relay is the command path's handler for /r/, /x/ and /deadman. +type Relay struct { + log *slog.Logger + now func() time.Time + keys BoxKeys + requireBoxKey bool + dead *deadman + + mu sync.Mutex + rooms map[string]*room + epoch int64 + rotateStarted time.Time + sockets int + perAddr map[string]int + frames int64 + nbytes int64 + heartbeats atomic.Int64 + openN int + joins *attempts + exchanges *attempts + closed bool +} + +type room struct { + uplink *peer + streams map[*peer]struct{} + exchange *peer + open map[string]*pending + rate *bucket + epoch int64 +} + +func (m *room) empty() bool { + return m.uplink == nil && m.exchange == nil && len(m.streams) == 0 +} + +func (m *room) peers() []*peer { + out := make([]*peer, 0, len(m.streams)+2) + if m.uplink != nil { + out = append(out, m.uplink) + } + for p := range m.streams { + out = append(out, p) + } + if m.exchange != nil { + out = append(out, m.exchange) + } + return out +} + +// peer is one socket. Its buckets die with it. +type peer struct { + conn *websocket.Conn + role string + frames *bucket + bytes *bucket + alive atomic.Bool + linked bool + addr string + claims map[string]struct{} + writeMu sync.Mutex + removed bool + closing atomic.Bool +} + +type pending struct { + w http.ResponseWriter + done chan struct{} + timer *time.Timer +} + +// New builds the handler. It does not start the heartbeat; call Run. +func New(opts Options) *Relay { + now := opts.Now + if now == nil { + now = time.Now + } + log := opts.Log + if log == nil { + log = slog.Default() + } + at := now() + r := &Relay{ + log: log, + now: now, + keys: opts.BoxKeys, + requireBoxKey: opts.RequireBoxKey, + rooms: map[string]*room{}, + epoch: CurrentEpoch(at), + perAddr: map[string]int{}, + joins: newAttempts(attemptLimit, attemptWindow, attemptSlots, at), + exchanges: newAttempts(exchangePerMinute, attemptWindow, attemptSlots, at), + } + r.dead = newDeadman(opts.DeadmanPath, now, opts.Post, func(status int, failed bool) { + // Status only. The endpoint is a push token, which is a household + // identifier, so it never reaches the log. + if failed { + log.Info("relay deadman", "result", "failed") + return + } + log.Info("relay deadman", "status", status) + }) + return r +} + +// Run drives the one timer: heartbeats, reaping and the hour's rotation. +// The cadence is fixed, so nothing the peers send can change it. +func (r *Relay) Run(ctx context.Context) { + t := time.NewTicker(heartbeatEvery) + defer t.Stop() + for { + select { + case <-ctx.Done(): + r.Close() + return + case <-t.C: + r.beat() + } + } +} + +// Close drops every socket and answers every watch that is still waiting. +func (r *Relay) Close() { + r.mu.Lock() + if r.closed { + r.mu.Unlock() + return + } + r.closed = true + var peers []*peer + var waiting []*pending + for _, m := range r.rooms { + peers = append(peers, m.peers()...) + for k, p := range m.open { + delete(m.open, k) + r.openN-- + waiting = append(waiting, p) + } + } + r.rooms = map[string]*room{} + r.mu.Unlock() + for _, p := range waiting { + writeStatus(p.w, http.StatusGatewayTimeout, "") + close(p.done) + if p.timer != nil { + p.timer.Stop() + } + } + for _, p := range peers { + _ = p.conn.Close() + } +} + +// ServeHTTP answers the command path. Anything else is a 404, which is what +// a request for a home would get here too: this handler has no way to read one. +func (r *Relay) ServeHTTP(w http.ResponseWriter, req *http.Request) { + switch { + case strings.HasPrefix(req.URL.Path, "/r/"): + r.serveJoin(w, req) + case strings.HasPrefix(req.URL.Path, "/x/"): + r.serveExchange(w, req) + case req.URL.Path == "/deadman" || strings.HasPrefix(req.URL.Path, "/deadman/"): + r.serveDeadman(w, req) + default: + http.NotFound(w, req) + } +} + +var upgrader = websocket.Upgrader{ + EnableCompression: false, // a compressed frame's size depends on its content + ReadBufferSize: 4096, + WriteBufferSize: 4096, + // The webapp is a browser on another origin. The handle is the + // capability; an Origin check would only refuse the clients this path + // exists for. Browsers still cannot read a watch's HTTP answer: /x/ + // sends no CORS header. + CheckOrigin: func(*http.Request) bool { return true }, +} + +func (r *Relay) serveJoin(w http.ResponseWriter, req *http.Request) { + if req.Header.Get("Upgrade") == "" { + writeStatus(w, http.StatusUpgradeRequired, "upgrade required\n") + return + } + now := r.now() + addr := clientIP(req, false) + + r.mu.Lock() + switch { + case r.closed || r.sockets >= maxSockets: + r.mu.Unlock() + r.rejectUpgrade(w, req, CloseBusy, "full") + return + case !r.joins.allow(addr, now): + r.mu.Unlock() + r.rejectUpgrade(w, req, CloseBusy, "slow down") + return + } + epoch := r.epoch + r.mu.Unlock() + + j, ok := parseJoin(req.URL.Path) + if !ok { + r.rejectUpgrade(w, req, CloseBadJoin, "join") + return + } + if j.epoch != epoch { + r.rejectUpgrade(w, req, CloseEpoch, strconv.FormatInt(epoch, 10)) + return + } + switch checkBoxProof(req, r.keys, j.epoch, j.handle, j.role) { + case proofBad: + r.rejectUpgrade(w, req, CloseBadJoin, boxKeyReason()) + return + case proofAbsent: + if r.requireBoxKey && j.role != roleApp { + r.rejectUpgrade(w, req, CloseBadJoin, boxKeyReason()) + return + } + } + + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + conn.SetReadLimit(maxFrameBytes) + p := &peer{ + conn: conn, + role: j.role, + frames: newBucket(frameCapacity, framesPerSecond, now), + bytes: newBucket(byteCapacity, bytesPerSecond, now), + addr: addr, + } + p.alive.Store(true) + conn.SetPongHandler(func(string) error { + p.alive.Store(true) + return nil + }) + + r.mu.Lock() + if r.closed || r.sockets >= maxSockets || r.perAddr[addr] >= maxSocketsPerAddr { + r.mu.Unlock() + closeConn(conn, CloseBusy, "full") + return + } + m := r.rooms[j.handle] + if m == nil { + m = &room{ + streams: map[*peer]struct{}{}, + open: map[string]*pending{}, + rate: newBucket(exchangePerMinute, float64(exchangePerMinute)/60, now), + epoch: j.epoch, + } + } + switch j.role { + case roleBox: + if m.uplink != nil { + r.mu.Unlock() + closeConn(conn, CloseBusy, "uplink") + return + } + m.uplink = p + case roleExchange: + if m.exchange != nil { + r.mu.Unlock() + closeConn(conn, CloseBusy, "exchange") + return + } + m.exchange = p + default: + if len(m.streams) >= maxStreamsPerRoom { + r.mu.Unlock() + closeConn(conn, CloseBusy, "streams") + return + } + m.streams[p] = struct{}{} + } + r.rooms[j.handle] = m + r.sockets++ + r.perAddr[addr]++ + notes := r.sync(m) + r.mu.Unlock() + r.flushText(notes) + r.readLoop(j.handle, m, p) +} + +// rejectUpgrade accepts the WebSocket and closes it with a relay code. +// Closing before the upgrade would turn every refusal into an HTTP status, +// which the box and the app do not read: they read the close code. +func (r *Relay) rejectUpgrade(w http.ResponseWriter, req *http.Request, code int, reason string) { + conn, err := upgrader.Upgrade(w, req, nil) + if err != nil { + return + } + closeConn(conn, code, reason) +} + +func (r *Relay) readLoop(handle string, m *room, p *peer) { + defer r.onClose(handle, m, p) + for { + kind, msg, err := p.conn.ReadMessage() + if err != nil { + return + } + if r.onMessage(m, p, kind, msg) { + return + } + } +} + +// onMessage reports whether the loop should stop. The bytes of a binary +// frame are forwarded unchanged; the only use of the slice is its length, +// except on the exchange socket where the first 16 bytes are the id this +// service chose. +func (r *Relay) onMessage(m *room, p *peer, kind int, msg []byte) bool { + if kind != websocket.BinaryMessage { + if id, ok := deadmanClaimID(string(msg)); ok && p.role == roleBox { + r.noteClaim(p, id) + return false + } + r.drop(p, CloseBadJoin, "binary only") + return true + } + if len(msg) > maxFrameBytes { + r.drop(p, CloseBadJoin, "oversize") + return true + } + now := r.now() + if !p.frames.take(1, now) || !p.bytes.take(float64(len(msg)), now) { + r.drop(p, CloseBusy, "rate") + return true + } + if p.role == roleExchange { + r.onAnswer(p, m, msg) + return false + } + + // Copy before the next read reuses the buffer. WriteMessage copies too, + // and the writes below finish before the next read; the copy is what + // keeps a later change from aliasing a frame. + frame := append([]byte(nil), msg...) + r.mu.Lock() + var dest []*peer + if p.role == roleBox { + for s := range m.streams { + dest = append(dest, s) + } + } else if m.uplink != nil { + dest = append(dest, m.uplink) + } + r.frames++ + r.nbytes += int64(len(frame)) + r.mu.Unlock() + for _, d := range dest { + if err := d.send(websocket.BinaryMessage, frame); err != nil { + r.drop(d, CloseBusy, "slow reader") + } + } + return false +} + +func (r *Relay) noteClaim(p *peer, id string) { + if p.claims == nil { + p.claims = map[string]struct{}{} + } + if _, ok := p.claims[id]; ok { + return + } + p.claims[id] = struct{}{} + r.dead.claim(id) +} + +func (r *Relay) onAnswer(p *peer, m *room, frame []byte) { + if len(frame) <= ExchangeIDBytes || len(frame) > ExchangeIDBytes+exchangeMaxReply { + r.drop(p, CloseBadJoin, "size") + return + } + key := hex.EncodeToString(frame[:ExchangeIDBytes]) + reply := append([]byte(nil), frame[ExchangeIDBytes:]...) + r.settle(m, key, http.StatusOK, reply) +} + +func (r *Relay) onClose(handle string, m *room, p *peer) { + r.mu.Lock() + if p.removed { + r.mu.Unlock() + return + } + p.removed = true + r.sockets-- + if p.addr != "" { + if n := r.perAddr[p.addr] - 1; n > 0 { + r.perAddr[p.addr] = n + } else { + delete(r.perAddr, p.addr) + } + } + var waiting []*pending + switch { + case m.uplink == p: + m.uplink = nil + case m.exchange == p: + m.exchange = nil + waiting = takeWaiting(r, m) + default: + delete(m.streams, p) + } + if r.rooms[handle] == m && m.empty() { + delete(r.rooms, handle) + } + notes := r.sync(m) + r.mu.Unlock() + for id := range p.claims { + r.dead.release(id) + } + r.finishWaiting(waiting, http.StatusGatewayTimeout) + r.flushText(notes) +} + +// sync queues ready/gone. The exchange socket is not a side of the room: +// it is never told, and it never tells. +func (r *Relay) sync(m *room) []textNote { + var notes []textNote + note := func(p *peer, present bool) { + if p == nil || p.linked == present || p.removed { + return + } + p.linked = present + word := CtrlGone + if present { + word = CtrlReady + } + notes = append(notes, textNote{p, word}) + } + if m.uplink != nil { + note(m.uplink, len(m.streams) > 0) + } + up := m.uplink != nil + for s := range m.streams { + note(s, up) + } + return notes +} + +type textNote struct { + p *peer + word string +} + +func (r *Relay) flushText(notes []textNote) { + for _, n := range notes { + if err := n.p.send(websocket.TextMessage, []byte(n.word)); err != nil { + r.drop(n.p, CloseBusy, "slow reader") + } + } +} + +// beat is the heartbeat, the reap and the rotation check. +func (r *Relay) beat() { + r.dead.beat() + now := r.now() + epoch := CurrentEpoch(now) + + r.mu.Lock() + var dead []*peer + var ping []*peer + for _, m := range r.rooms { + for _, p := range m.peers() { + if !p.alive.Load() { + dead = append(dead, p) + continue + } + p.alive.Store(false) + ping = append(ping, p) + } + } + if epoch != r.epoch { + r.epoch = epoch + r.rotateStarted = now + } + type rot struct { + peers []*peer + wait []*pending + } + var rotate []rot + if !r.rotateStarted.IsZero() { + elapsed := now.Sub(r.rotateStarted) + for handle, m := range r.rooms { + // A room that joined this hour already has this hour's handle. + // Evicting it would kick a healthy peer, and kick it again on + // every tick, which is also the vanish-and-return pattern + // rotation exists to avoid. + if m.epoch == r.epoch || elapsed < RotationOffset(handle, rotateSpread) { + continue + } + rotate = append(rotate, rot{peers: m.peers(), wait: takeWaiting(r, m)}) + delete(r.rooms, handle) + } + if elapsed >= rotateSpread { + r.rotateStarted = time.Time{} + } + } + epoch = r.epoch + rooms, sockets := len(r.rooms), r.sockets + frames, nbytes := r.frames, r.nbytes + r.mu.Unlock() + + for _, p := range dead { + _ = p.conn.Close() + } + for _, p := range ping { + deadline := time.Now().Add(maxBufferedWrite) + if p.conn.WriteControl(websocket.PingMessage, nil, deadline) != nil { + _ = p.conn.Close() + continue + } + r.heartbeats.Add(1) + } + reason := strconv.FormatInt(epoch, 10) + for _, item := range rotate { + for _, p := range item.peers { + closeConn(p.conn, CloseRotated, reason) + } + r.finishWaiting(item.wait, http.StatusGatewayTimeout) + } + r.log.Info("relay", "epoch", epoch, "rooms", rooms, "sockets", sockets, "frames", frames, "bytes", nbytes) +} + +// takeWaiting removes a room's open exchanges. The caller holds r.mu. +// Calling it twice returns an empty slice the second time. +func takeWaiting(r *Relay, m *room) []*pending { + if len(m.open) == 0 { + return nil + } + out := make([]*pending, 0, len(m.open)) + for k, p := range m.open { + delete(m.open, k) + r.openN-- + out = append(out, p) + } + return out +} + +func (r *Relay) finishWaiting(waiting []*pending, status int) { + for _, p := range waiting { + if p.timer != nil { + p.timer.Stop() + } + writeStatus(p.w, status, "") + close(p.done) + } +} + +func (r *Relay) drop(p *peer, code int, reason string) { + if !p.closing.CompareAndSwap(false, true) { + return + } + closeConn(p.conn, code, reason) +} + +func closeConn(conn *websocket.Conn, code int, reason string) { + _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, reason), time.Now().Add(time.Second)) + _ = conn.Close() +} + +func (p *peer) send(kind int, data []byte) error { + p.writeMu.Lock() + defer p.writeMu.Unlock() + if err := p.conn.SetWriteDeadline(time.Now().Add(maxBufferedWrite)); err != nil { + return err + } + return p.conn.WriteMessage(kind, data) +} + +// -------------------------------------------------------------------------- +// Watch exchange +// -------------------------------------------------------------------------- + +func (r *Relay) serveExchange(w http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + w.Header().Set("Allow", "POST") + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + now := r.now() + addr := clientIP(req, false) + r.mu.Lock() + allowed := !r.closed && r.exchanges.allow(addr, now) + epoch := r.epoch + rotating := !r.rotateStarted.IsZero() + r.mu.Unlock() + if !allowed { + writeStatus(w, http.StatusTooManyRequests, "") + return + } + target, ok := parseExchange(req.URL.Path) + media := strings.ToLower(strings.TrimSpace(strings.Split(req.Header.Get("Content-Type"), ";")[0])) + if !ok || media != "application/octet-stream" { + writeStatus(w, http.StatusBadRequest, "") + return + } + if !epochOK(target.epoch, epoch, rotating) { + writeStatus(w, http.StatusConflict, strconv.FormatInt(epoch, 10)) + return + } + if cl := req.Header.Get("Content-Length"); cl != "" { + n, err := strconv.Atoi(cl) + if err != nil || (n != exchangeOpenBytes && n != exchangeRequestBytes) { + writeStatus(w, http.StatusBadRequest, "") + return + } + } + body, err := readLimited(req.Body, exchangeOpenBytes) + if err != nil || (len(body) != exchangeOpenBytes && len(body) != exchangeRequestBytes) { + writeStatus(w, http.StatusBadRequest, "") + return + } + r.forward(target, body, w, req) +} + +func epochOK(got, current int64, rotating bool) bool { + if got == current { + return true + } + return rotating && got == current-1 +} + +func (r *Relay) forward(target exchangeTarget, body []byte, w http.ResponseWriter, req *http.Request) { + if req.Context().Err() != nil { + return + } + id := make([]byte, ExchangeIDBytes) + if _, err := rand.Read(id); err != nil { + writeStatus(w, http.StatusGatewayTimeout, "") + return + } + key := hex.EncodeToString(id) + + r.mu.Lock() + m := r.rooms[target.handle] + if m == nil || m.epoch != target.epoch || (m.uplink == nil && m.exchange == nil) { + r.mu.Unlock() + writeStatus(w, http.StatusNotFound, "") + return + } + if m.exchange == nil { + r.mu.Unlock() + writeStatus(w, http.StatusNotImplemented, "") + return + } + now := r.now() + if r.openN >= exchangeOpenProcess || len(m.open) >= exchangeOpenPerRoom || !m.rate.take(1, now) { + r.mu.Unlock() + writeStatus(w, http.StatusTooManyRequests, "") + return + } + box := m.exchange + pend := &pending{w: w, done: make(chan struct{})} + m.open[key] = pend + r.openN++ + // Armed under the lock, after the row exists. The callback takes the + // same lock, so it cannot win the race against the insert. + pend.timer = time.AfterFunc(exchangeTimeout, func() { + r.settle(m, key, http.StatusGatewayTimeout, nil) + }) + r.mu.Unlock() + frame := make([]byte, ExchangeIDBytes+len(body)) + copy(frame, id) + copy(frame[ExchangeIDBytes:], body) + if err := box.send(websocket.BinaryMessage, frame); err != nil { + r.settle(m, key, http.StatusGatewayTimeout, nil) + } + select { + case <-pend.done: + case <-req.Context().Done(): + r.settle(m, key, 0, nil) + <-pend.done + } +} + +func (r *Relay) settle(m *room, key string, status int, reply []byte) { + r.mu.Lock() + p := m.open[key] + if p == nil { + r.mu.Unlock() + return + } + delete(m.open, key) + r.openN-- + r.mu.Unlock() + if p.timer != nil { + p.timer.Stop() + } + if status == http.StatusOK { + p.w.Header().Set("Content-Type", "application/octet-stream") + p.w.Header().Set("Cache-Control", "no-store") + p.w.WriteHeader(http.StatusOK) + _, _ = p.w.Write(reply) + } else if status != 0 { + writeStatus(p.w, status, "") + } + close(p.done) +} + +// -------------------------------------------------------------------------- +// Dead man +// -------------------------------------------------------------------------- + +func (r *Relay) serveDeadman(w http.ResponseWriter, req *http.Request) { + path := req.URL.Path + if req.Method == http.MethodPost && (path == "/deadman" || path == "/deadman/") { + r.postDeadman(w, req) + return + } + if req.Method == http.MethodDelete { + id := strings.TrimPrefix(path, "/deadman/") + if deadmanID.MatchString(id) && "/deadman/"+id == path { + r.dead.remove(id) + w.WriteHeader(http.StatusNoContent) + return + } + } + http.NotFound(w, req) +} + +func (r *Relay) postDeadman(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") + raw, err := readLimited(req.Body, deadmanMaxBody) + if err != nil { + w.WriteHeader(http.StatusRequestEntityTooLarge) + _, _ = w.Write([]byte(`{"error":"too large"}`)) + return + } + var b deadmanIn + if json.Unmarshal(raw, &b) != nil { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"a JSON object"}`)) + return + } + ct, ctErr := base64.StdEncoding.DecodeString(b.CT) + ctLen := len(ct) + if ctErr != nil { + ctLen = 0 + if b.CT != "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"ct must be base64"}`)) + return + } + } + if fault := putError(b, ctLen); fault != "" { + code := http.StatusBadRequest + if fault == "ct too large" { + code = http.StatusRequestEntityTooLarge + } + w.WriteHeader(code) + raw, _ := json.Marshal(struct { + Error string `json:"error"` + }{fault}) + _, _ = w.Write(raw) + return + } + r.dead.put(b) + w.Header().Del("Content-Type") + w.WriteHeader(http.StatusNoContent) +} + +func readLimited(body io.Reader, max int) ([]byte, error) { + b, err := io.ReadAll(io.LimitReader(body, int64(max)+1)) + if err != nil { + return nil, err + } + if len(b) > max { + return nil, errors.New("too large") + } + return b, nil +} + +func writeStatus(w http.ResponseWriter, status int, body string) { + if w == nil { + return + } + w.Header().Set("Content-Type", "text/plain") + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(status) + if body != "" { + _, _ = io.WriteString(w, body) + } +} diff --git a/internal/relay/relay_test.go b/internal/relay/relay_test.go new file mode 100644 index 0000000..a4b19ed --- /dev/null +++ b/internal/relay/relay_test.go @@ -0,0 +1,711 @@ +package relay + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "encoding/base64" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +const ( + testEpoch = int64(500_000) + handleA = "ab0123456789abcdef0123456789abcd" + handleB = "cd0123456789abcdef0123456789abcd" +) + +func testTime() time.Time { return time.UnixMilli(testEpoch * EpochMs).UTC() } + +type clock struct { + mu sync.Mutex + t time.Time +} + +func (c *clock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *clock) Set(t time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = t +} + +func discardLog() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func startRelay(t *testing.T, opts Options) (*Relay, *httptest.Server) { + t.Helper() + if opts.Log == nil { + opts.Log = discardLog() + } + if opts.Now == nil { + opts.Now = testTime + } + r := New(opts) + srv := httptest.NewServer(r) + t.Cleanup(func() { + r.Close() + srv.Close() + }) + return r, srv +} + +func wsBase(httpURL string) string { return "ws" + strings.TrimPrefix(httpURL, "http") } + +func dial(t *testing.T, base string, epoch int64, handle, role string, hdr http.Header) *websocket.Conn { + t.Helper() + u := fmt.Sprintf("%s/r/%d/%s/%s", wsBase(base), epoch, handle, role) + c, _, err := (&websocket.Dialer{HandshakeTimeout: 2 * time.Second}).Dial(u, hdr) + if err != nil { + t.Fatalf("dial %s: %v", role, err) + } + t.Cleanup(func() { _ = c.Close() }) + return c +} + +func readMsg(t *testing.T, c *websocket.Conn) (int, []byte) { + t.Helper() + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + kind, msg, err := c.ReadMessage() + if err != nil { + t.Fatal(err) + } + return kind, msg +} + +func expectClose(t *testing.T, c *websocket.Conn, code int, reason string) { + t.Helper() + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, err := c.ReadMessage() + var ce *websocket.CloseError + if !errors.As(err, &ce) { + t.Fatalf("got %v, want close %d %q", err, code, reason) + } + if ce.Code != code || ce.Text != reason { + t.Fatalf("close %d %q, want %d %q", ce.Code, ce.Text, code, reason) + } +} + +func TestWireConstantsMatchTheClients(t *testing.T) { + if CloseBadJoin != 4400 || CloseEpoch != 4409 || CloseRotated != 4410 || CloseBusy != 4429 { + t.Fatal("close codes drifted from the box and the webapp") + } + if CtrlReady != "ready" || CtrlGone != "gone" { + t.Fatal("control words drifted") + } + if EpochMs != 3_600_000 || HandleChars != 32 || ExchangeIDBytes != 16 { + t.Fatal("epoch, handle or exchange id drifted") + } + if exchangeOpenBytes != 608 || exchangeRequestBytes != 536 || exchangeMaxReply != 4096 { + t.Fatal("exchange sizes drifted from the watch") + } + if CurrentEpoch(testTime()) != testEpoch { + t.Fatalf("epoch = %d", CurrentEpoch(testTime())) + } + // Floor, not truncation toward zero. The app uses Math.floor. + if CurrentEpoch(time.UnixMilli(-1)) != -1 { + t.Fatal("a clock before 1970 disagreed with the app") + } +} + +func TestFramesPassThroughUnchangedAndStayInTheRoom(t *testing.T) { + _, srv := startRelay(t, Options{}) + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + app1 := dial(t, srv.URL, testEpoch, handleA, roleApp, http.Header{"Origin": {"https://webapp.example"}}) + app2 := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + other := dial(t, srv.URL, testEpoch, handleB, roleBox, nil) + + for _, c := range []*websocket.Conn{box, app1, app2} { + kind, msg := readMsg(t, c) + if kind != websocket.TextMessage || string(msg) != CtrlReady { + t.Fatalf("ready: kind %d %q", kind, msg) + } + } + + payload := []byte("COMMAND-set_mode-watt-4242-\x00\xff") + if err := app1.WriteMessage(websocket.BinaryMessage, payload); err != nil { + t.Fatal(err) + } + kind, got := readMsg(t, box) + if kind != websocket.BinaryMessage || !bytes.Equal(got, payload) { + t.Fatalf("box got kind %d %q", kind, got) + } + + // A read that times out leaves a gorilla connection unusable, so the + // proof that a frame stayed in its room is the next frame each peer + // does see: app2's first binary is the box's reply, not app1's command, + // and the other room's box sees nothing before it is closed below. + back := []byte{0x01, 0x02, 0x03, 0x00} + if err := box.WriteMessage(websocket.BinaryMessage, back); err != nil { + t.Fatal(err) + } + if _, got = readMsg(t, app1); !bytes.Equal(got, back) { + t.Fatalf("app1 got %x", got) + } + if _, got = readMsg(t, app2); !bytes.Equal(got, back) { + t.Fatalf("app2 got %x, a stream saw another stream", got) + } + + // One stream leaving leaves the other side present. gone is the last one. + _ = app1.Close() + if err := app2.WriteMessage(websocket.BinaryMessage, []byte{9}); err != nil { + t.Fatal(err) + } + if _, got = readMsg(t, box); !bytes.Equal(got, []byte{9}) { + t.Fatalf("box after one stream left: %x", got) + } + _ = app2.Close() + kind, msg := readMsg(t, box) + if kind != websocket.TextMessage || string(msg) != CtrlGone { + t.Fatalf("gone: kind %d %q", kind, msg) + } + _ = other.SetReadDeadline(time.Now().Add(50 * time.Millisecond)) + if _, _, err := other.ReadMessage(); err == nil { + t.Fatal("another session received the frame") + } +} + +func TestTheIncumbentBoxKeepsTheRoom(t *testing.T) { + _, srv := startRelay(t, Options{}) + first := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + second := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + expectClose(t, second, CloseBusy, "uplink") + + app := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + if _, msg := readMsg(t, app); string(msg) != CtrlReady { + t.Fatalf("app: %q", msg) + } + if _, msg := readMsg(t, first); string(msg) != CtrlReady { + t.Fatalf("box: %q", msg) + } +} + +func TestJoinRefusalsUseTheRelayCloseCodes(t *testing.T) { + _, srv := startRelay(t, Options{}) + cases := []struct { + path string + code int + reason string + }{ + {"/r/nope/" + handleA + "/box", CloseBadJoin, "join"}, + {"/r/" + fmt.Sprint(testEpoch) + "/ABCDEF0123456789ABCDEF0123456789/box", CloseBadJoin, "join"}, + {"/r/" + fmt.Sprint(testEpoch) + "/" + handleA + "/phone", CloseBadJoin, "join"}, + {"/r/" + fmt.Sprint(testEpoch+3) + "/" + handleA + "/box", CloseEpoch, fmt.Sprint(testEpoch)}, + } + for _, c := range cases { + conn, _, err := (&websocket.Dialer{HandshakeTimeout: 2 * time.Second}).Dial(wsBase(srv.URL)+c.path, nil) + if err != nil { + t.Fatalf("%s: dial %v", c.path, err) + } + expectClose(t, conn, c.code, c.reason) + _ = conn.Close() + } +} + +func TestTextIsNotRoutedAndADeadmanClaimIsConsumed(t *testing.T) { + _, srv := startRelay(t, Options{}) + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + app := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + _, _ = readMsg(t, box) + _, _ = readMsg(t, app) + + id := "0123456789abcdef0123456789abcdef" + if err := box.WriteMessage(websocket.TextMessage, []byte("deadman "+id)); err != nil { + t.Fatal(err) + } + // The claim and the following frame share the box socket, so the frame + // cannot arrive ahead of a claim that was routed by mistake. + if err := box.WriteMessage(websocket.BinaryMessage, []byte{7}); err != nil { + t.Fatal(err) + } + kind, msg := readMsg(t, app) + if kind != websocket.BinaryMessage || !bytes.Equal(msg, []byte{7}) { + t.Fatalf("the claim was routed: kind %d %q", kind, msg) + } + if err := app.WriteMessage(websocket.TextMessage, []byte("hello")); err != nil { + t.Fatal(err) + } + expectClose(t, app, CloseBadJoin, "binary only") + // The box stays. A claim is not a reason to drop it, and neither is + // some other peer's mistake. + if err := box.WriteMessage(websocket.BinaryMessage, []byte{7}); err != nil { + t.Fatal(err) + } +} + +func TestRoomLimits(t *testing.T) { + _, srv := startRelay(t, Options{}) + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + for i := 0; i < maxStreamsPerRoom; i++ { + app := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + _, _ = readMsg(t, app) + } + _, _ = readMsg(t, box) + fifth := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + expectClose(t, fifth, CloseBusy, "streams") + + ex := dial(t, srv.URL, testEpoch, handleA, roleExchange, nil) + second := dial(t, srv.URL, testEpoch, handleA, roleExchange, nil) + expectClose(t, second, CloseBusy, "exchange") + _ = ex.Close() +} + +func TestAFrameAtTheSizeLimitPassesAndTheNextRateCloses(t *testing.T) { + _, srv := startRelay(t, Options{}) + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + app := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + _, _ = readMsg(t, box) + _, _ = readMsg(t, app) + + big := bytes.Repeat([]byte{0x11}, maxFrameBytes) + if err := app.WriteMessage(websocket.BinaryMessage, big); err != nil { + t.Fatal(err) + } + if _, got := readMsg(t, box); !bytes.Equal(got, big) { + t.Fatalf("got %d bytes", len(got)) + } + for i := 1; i < frameCapacity; i++ { + if err := app.WriteMessage(websocket.BinaryMessage, []byte{1}); err != nil { + t.Fatal(err) + } + } + // Drain so a slow reader is not why the next write fails. + for i := 1; i < frameCapacity; i++ { + _, _ = readMsg(t, box) + } + if err := app.WriteMessage(websocket.BinaryMessage, []byte{1}); err != nil { + t.Fatal(err) + } + expectClose(t, app, CloseBusy, "rate") +} + +func TestRotationIsSpreadAndANewRoomIsLeftAlone(t *testing.T) { + clk := &clock{t: testTime()} + r, srv := startRelay(t, Options{Now: clk.Now}) + low := "00000000000000000000000000000000" + high := "ffffffffffffffffffffffffffffffff" + early := dial(t, srv.URL, testEpoch, low, roleBox, nil) + late := dial(t, srv.URL, testEpoch, high, roleBox, nil) + lateApp := dial(t, srv.URL, testEpoch, high, roleApp, nil) + if _, msg := readMsg(t, late); string(msg) != CtrlReady { + t.Fatalf("late box: %q", msg) + } + if _, msg := readMsg(t, lateApp); string(msg) != CtrlReady { + t.Fatalf("late app: %q", msg) + } + + clk.Set(testTime().Add(time.Duration(EpochMs)*time.Millisecond + time.Millisecond)) + r.beat() + expectClose(t, early, CloseRotated, fmt.Sprint(testEpoch+1)) + + // The late room has not had its turn, so the peers already in it keep + // passing frames. A new join of the old epoch is still a wrong epoch: + // the handle is derived from the epoch, and a window there would park + // a peer in a room its box will never register. Frankfurt answers with + // the epoch to use. + if err := lateApp.WriteMessage(websocket.BinaryMessage, []byte{4}); err != nil { + t.Fatal(err) + } + if _, got := readMsg(t, late); !bytes.Equal(got, []byte{4}) { + t.Fatalf("late room dropped a frame before its turn: %x", got) + } + // Reading delivers the heartbeat's ping to both peers, so the next beat + // rotates them instead of reaping a missed pong. + if err := late.WriteMessage(websocket.BinaryMessage, []byte{4}); err != nil { + t.Fatal(err) + } + if _, got := readMsg(t, lateApp); !bytes.Equal(got, []byte{4}) { + t.Fatalf("late app: %x", got) + } + conn, _, err := (&websocket.Dialer{HandshakeTimeout: 2 * time.Second}).Dial( + fmt.Sprintf("%s/r/%d/%s/app", wsBase(srv.URL), testEpoch, high), nil) + if err != nil { + t.Fatal(err) + } + expectClose(t, conn, CloseEpoch, fmt.Sprint(testEpoch+1)) + _ = conn.Close() + + // A room that joined the new hour is not evicted when the window ends. + fresh := dial(t, srv.URL, testEpoch+1, handleA, roleBox, nil) + freshApp := dial(t, srv.URL, testEpoch+1, handleA, roleApp, nil) + _, _ = readMsg(t, fresh) + _, _ = readMsg(t, freshApp) + + clk.Set(testTime().Add(time.Duration(EpochMs)*time.Millisecond + rotateSpread)) + r.beat() + expectClose(t, late, CloseRotated, fmt.Sprint(testEpoch+1)) + // The uplink's close can say gone before the rotation close lands. + // Both words are the relay's; the close code is the one a client acts on. + _ = lateApp.SetReadDeadline(time.Now().Add(2 * time.Second)) + for { + _, msg, err := lateApp.ReadMessage() + var ce *websocket.CloseError + if errors.As(err, &ce) { + if ce.Code != CloseRotated || ce.Text != fmt.Sprint(testEpoch+1) { + t.Fatalf("close %d %q", ce.Code, ce.Text) + } + break + } + if err != nil { + t.Fatal(err) + } + if string(msg) != CtrlGone { + t.Fatalf("late app got %q", msg) + } + } + if err := freshApp.WriteMessage(websocket.BinaryMessage, []byte{5}); err != nil { + t.Fatal(err) + } + if _, got := readMsg(t, fresh); !bytes.Equal(got, []byte{5}) { + t.Fatal("a room on the new hour was rotated") + } +} + +func TestExchangeRoundTripIsBlind(t *testing.T) { + _, srv := startRelay(t, Options{}) + _ = dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + ex := dial(t, srv.URL, testEpoch, handleA, roleExchange, nil) + body := bytes.Repeat([]byte{0x5a}, exchangeRequestBytes) + copy(body, []byte("COMMAND-set_mode")) + reply := []byte("REPLY-not-parsed-\x00") + + type httpResult struct { + code int + body []byte + hdr http.Header + } + got := make(chan httpResult, 1) + go func() { + // Dial returns when the handshake completes, a moment before the + // socket is in the room. A 501 in that gap is the watch's "try + // again", not a missing box. + for i := 0; i < 50; i++ { + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/x/"+fmt.Sprint(testEpoch)+"/"+handleA, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/octet-stream") + res, err := http.DefaultClient.Do(req) + if err != nil { + got <- httpResult{code: -1} + return + } + b, _ := io.ReadAll(res.Body) + res.Body.Close() + if (res.StatusCode == http.StatusNotImplemented || res.StatusCode == http.StatusNotFound) && i+1 < 50 { + time.Sleep(2 * time.Millisecond) + continue + } + got <- httpResult{res.StatusCode, b, res.Header} + return + } + }() + + _, msg := readMsg(t, ex) + if len(msg) != ExchangeIDBytes+len(body) || !bytes.Equal(msg[ExchangeIDBytes:], body) { + t.Fatalf("box saw %d bytes", len(msg)) + } + out := append(append([]byte{}, msg[:ExchangeIDBytes]...), reply...) + if err := ex.WriteMessage(websocket.BinaryMessage, out); err != nil { + t.Fatal(err) + } + res := <-got + if res.code != http.StatusOK || !bytes.Equal(res.body, reply) { + t.Fatalf("watch got %d %q", res.code, res.body) + } + if res.hdr.Get("Access-Control-Allow-Origin") != "" { + t.Fatal("a web page could read the answer") + } + if res.hdr.Get("Content-Type") != "application/octet-stream" { + t.Fatalf("content-type %q", res.hdr.Get("Content-Type")) + } +} + +func TestExchangeRateRefills(t *testing.T) { + clk := &clock{t: testTime()} + _, srv := startRelay(t, Options{Now: clk.Now}) + _ = dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + ex := dial(t, srv.URL, testEpoch, handleA, roleExchange, nil) + go func() { + for { + _, msg, err := ex.ReadMessage() + if err != nil { + return + } + _ = ex.WriteMessage(websocket.BinaryMessage, append(append([]byte{}, msg[:ExchangeIDBytes]...), []byte{1}...)) + } + }() + body := bytes.Repeat([]byte{3}, exchangeRequestBytes) + post := func() int { + t.Helper() + res, err := http.Post(srv.URL+"/x/"+fmt.Sprint(testEpoch)+"/"+handleA, "application/octet-stream", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + _, _ = io.Copy(io.Discard, res.Body) + return res.StatusCode + } + // The socket may not be in the room for the first try. That 501 is not + // one of the thirty. + var primed int + for i := 0; i < 50 && primed != http.StatusOK; i++ { + primed = post() + } + if primed != http.StatusOK { + t.Fatalf("first exchange: %d", primed) + } + for i := 1; i < exchangePerMinute; i++ { + if code := post(); code != http.StatusOK { + t.Fatalf("exchange %d: %d", i, code) + } + } + if code := post(); code != http.StatusTooManyRequests { + t.Fatalf("over the minute: %d", code) + } + // The address window and the room bucket both last a minute. A bucket + // whose refill was computed with integer division (30/60 == 0) stays + // empty after the window and still answers 429. + clk.Set(testTime().Add(attemptWindow)) + if code := post(); code != http.StatusOK { + t.Fatalf("after refill: %d", code) + } +} + +func TestExchangeRefusals(t *testing.T) { + _, srv := startRelay(t, Options{}) + post := func(path string, n int) (int, string) { + t.Helper() + res, err := http.Post(srv.URL+path, "application/octet-stream", bytes.NewReader(bytes.Repeat([]byte{1}, n))) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + b, _ := io.ReadAll(res.Body) + return res.StatusCode, string(b) + } + if code, _ := post("/x/"+fmt.Sprint(testEpoch)+"/"+handleA, exchangeRequestBytes); code != http.StatusNotFound { + t.Fatalf("no box: %d", code) + } + _ = dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + // Dial returns at the handshake, a moment before the box is in the room. + var code int + for i := 0; i < 50; i++ { + code, _ = post("/x/"+fmt.Sprint(testEpoch)+"/"+handleA, exchangeOpenBytes) + if code != http.StatusNotFound { + break + } + time.Sleep(2 * time.Millisecond) + } + if code != http.StatusNotImplemented { + t.Fatalf("no exchange socket: %d", code) + } + if code, body := post("/x/"+fmt.Sprint(testEpoch+9)+"/"+handleA, exchangeRequestBytes); code != http.StatusConflict || body != fmt.Sprint(testEpoch) { + t.Fatalf("wrong epoch: %d %q", code, body) + } + res, err := http.Get(srv.URL + "/x/" + fmt.Sprint(testEpoch) + "/" + handleA) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + if res.StatusCode != http.StatusMethodNotAllowed || res.Header.Get("Access-Control-Allow-Origin") != "" { + t.Fatalf("GET %d", res.StatusCode) + } + if code, _ := post("/x/"+fmt.Sprint(testEpoch)+"/"+handleA, 10); code != http.StatusBadRequest { + t.Fatalf("bad size: %d", code) + } +} + +func TestExchangeAcceptsThePreviousEpochDuringRotation(t *testing.T) { + clk := &clock{t: testTime()} + r, srv := startRelay(t, Options{Now: clk.Now}) + high := "ffffffffffffffffffffffffffffffff" + _ = dial(t, srv.URL, testEpoch, high, roleBox, nil) + ex := dial(t, srv.URL, testEpoch, high, roleExchange, nil) + clk.Set(testTime().Add(time.Duration(EpochMs)*time.Millisecond + time.Millisecond)) + r.beat() + + body := bytes.Repeat([]byte{2}, exchangeRequestBytes) + got := make(chan int, 1) + go func() { + client := &http.Client{Timeout: 2 * time.Second} + for i := 0; i < 50; i++ { + res, err := client.Post(srv.URL+"/x/"+fmt.Sprint(testEpoch)+"/"+high, "application/octet-stream", bytes.NewReader(body)) + if err != nil { + got <- -1 + return + } + _, _ = io.Copy(io.Discard, res.Body) + res.Body.Close() + if (res.StatusCode == http.StatusNotImplemented || res.StatusCode == http.StatusNotFound) && i+1 < 50 { + time.Sleep(2 * time.Millisecond) + continue + } + got <- res.StatusCode + return + } + }() + _, msg := readMsg(t, ex) + if !bytes.Equal(msg[ExchangeIDBytes:], body) { + t.Fatal("previous epoch was not delivered") + } + _ = ex.WriteMessage(websocket.BinaryMessage, append(append([]byte{}, msg[:ExchangeIDBytes]...), []byte{9}...)) + if code := <-got; code != http.StatusOK { + t.Fatalf("previous epoch: %d", code) + } +} + +func TestBoxKeyProof(t *testing.T) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + pub, err := key.PublicKey.Bytes() + if err != nil { + t.Fatal(err) + } + _, srv := startRelay(t, Options{BoxKeys: acceptKey{pub}}) + + // No proof: the handle is enough, which is what today's clients send. + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + _ = box.Close() + + keyHdr, sigHdr, err := signBoxProof(key, testEpoch, handleB, roleBox) + if err != nil { + t.Fatal(err) + } + hdr := http.Header{"X-FTW-Box-Key": {keyHdr}, "X-FTW-Box-Sig": {sigHdr}} + ok := dial(t, srv.URL, testEpoch, handleB, roleBox, hdr) + _ = ok.Close() + + // A proof that does not verify is a bad join, even though a missing + // one would have been admitted. + bad := hdr.Clone() + raw, _ := base64.RawURLEncoding.DecodeString(sigHdr) + raw[0] ^= 0xff + bad.Set("X-FTW-Box-Sig", base64.RawURLEncoding.EncodeToString(raw)) + conn := dial(t, srv.URL, testEpoch, handleA, roleBox, bad) + expectClose(t, conn, CloseBadJoin, "box key") + + // The same proof does not admit the exchange socket: the role is signed. + conn = dial(t, srv.URL, testEpoch, handleA, roleExchange, hdr) + expectClose(t, conn, CloseBadJoin, "box key") + + // An unknown key is not a proof, signature or not. + other, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + k2, s2, err := signBoxProof(other, testEpoch, handleA, roleBox) + if err != nil { + t.Fatal(err) + } + conn = dial(t, srv.URL, testEpoch, handleA, roleBox, http.Header{"X-FTW-Box-Key": {k2}, "X-FTW-Box-Sig": {s2}}) + expectClose(t, conn, CloseBadJoin, "box key") + + _, strict := startRelay(t, Options{RequireBoxKey: true, BoxKeys: acceptKey{pub}}) + conn = dial(t, strict.URL, testEpoch, handleA, roleBox, nil) + expectClose(t, conn, CloseBadJoin, "box key") + // An app still joins with the handle. It does not hold the box's cloud key. + app := dial(t, strict.URL, testEpoch, handleA, roleApp, nil) + _ = app.Close() +} + +type acceptKey struct{ pk []byte } + +func (a acceptKey) RecognizesBoxKey(pk []byte) bool { return bytes.Equal(a.pk, pk) } + +func TestLogsDoNotContainACommandOrAPushToken(t *testing.T) { + var buf bytes.Buffer + log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + r, srv := startRelay(t, Options{Log: log}) + box := dial(t, srv.URL, testEpoch, handleA, roleBox, nil) + app := dial(t, srv.URL, testEpoch, handleA, roleApp, nil) + _, _ = readMsg(t, box) + _, _ = readMsg(t, app) + marker := "COMMAND-set_mode-watt-4242" + if err := app.WriteMessage(websocket.BinaryMessage, []byte(marker)); err != nil { + t.Fatal(err) + } + _, _ = readMsg(t, box) + + const token = "SECRET-PUSH-TOKEN" + res, err := http.Post(srv.URL+"/deadman", "application/json", strings.NewReader( + `{"id":"0123456789abcdef0123456789abcdef","endpoint":"https://push.example/`+token+`","ct":"aGVsbG8=","auth":"vapid `+token+`","deadline_s":600}`)) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != http.StatusNoContent { + t.Fatalf("deadman %d", res.StatusCode) + } + r.beat() + if bytes.Contains(buf.Bytes(), []byte(marker)) || bytes.Contains(buf.Bytes(), []byte(token)) || bytes.Contains(buf.Bytes(), []byte(handleA)) { + t.Fatalf("log retained a command, a token or a handle:\n%s", buf.String()) + } +} + +func TestClientIPIgnoresAHeaderTheClientWrote(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "/", nil) + req.RemoteAddr = "198.51.100.7:4242" + req.Header.Set("X-Forwarded-For", "1.1.1.1, 198.51.100.7") + if got := clientIP(req, false); got != "198.51.100.7" { + t.Fatalf("untrusted: %q", got) + } + if got := clientIP(req, true); got != "198.51.100.7" { + t.Fatalf("trusted tail: %q", got) + } + req.Header.Set("X-Forwarded-For", "9.9.9.9") + if got := clientIP(req, true); got != "9.9.9.9" { + t.Fatalf("trusted: %q", got) + } +} + +func TestCommandPathDoesNotImportAReader(t *testing.T) { + // Standard-library deps include the TLS and RNG ciphers. Those are not a + // reader of a command. What must stay out is this module's store and + // any Noise or NaCl package the relay could open a frame with. + out, err := exec.Command("go", "list", "-deps", "-f", "{{if not .Standard}}{{.ImportPath}}{{end}}", "github.com/srcfl/ftw-cloud/internal/relay").CombinedOutput() + if err != nil { + t.Fatal(err) + } + for _, p := range strings.Split(string(out), "\n") { + switch { + case strings.Contains(p, "internal/store"), + strings.Contains(p, "internal/uplink"), + strings.Contains(p, "internal/appsession"), + strings.Contains(p, "appwire"), + strings.Contains(p, "cloudwire"), + strings.Contains(p, "chacha"), + strings.Contains(p, "nacl"), + strings.Contains(p, "noise"): + t.Fatalf("the command path can read what it carries: %s", p) + } + } +} + +func TestRotationOffsetStaysInsideTheWindow(t *testing.T) { + seen := map[time.Duration]struct{}{} + for i := 0; i < 200; i++ { + h := fmt.Sprintf("%04x", i*300%0x10000) + "0000000000000000000000000000" + off := RotationOffset(h, rotateSpread) + if off < 0 || off >= rotateSpread { + t.Fatalf("offset %s", off) + } + seen[off] = struct{}{} + } + if len(seen) < 50 { + t.Fatalf("only %d distinct turns", len(seen)) + } +} diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index 965ee84..a5d9628 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -23,6 +23,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/push" "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/relay" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" "github.com/srcfl/ftw-cloud/internal/uplink" @@ -97,7 +98,8 @@ func newStack(t *testing.T) *stack { t.Cleanup(cancel) go pushSvc.Run(ctx) - s := &Server{Store: st, Registry: reg, Uplink: up, App: app, Tokens: tokens, Log: log} + s := &Server{Store: st, Registry: reg, Uplink: up, App: app, Tokens: tokens, + Relay: relay.New(relay.Options{Log: log, BoxKeys: reg}), Log: log} srv := httptest.NewUnstartedServer(s.Handler()) srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} srv.StartTLS() diff --git a/internal/server/server.go b/internal/server/server.go index b21a127..ec0e931 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -5,11 +5,15 @@ // /v1/app a phone's Noise session to the service (package appsession) // /v1/app/once a watch's one-shot read, Noise inside one POST (package appsession) // /v1/now a widget's token read (package token) +// /r/ the blind command path: box, app and exchange sockets (package relay) +// /x/ a watch's one command, posted, and answered by the box +// /deadman the box's sealed "I am gone" switch // /source the source of the running service, as AGPL-3.0 asks // /healthz liveness, and nothing about any home // // There is no route for staff. Reading a home needs its box's key, a phone in -// its roster, or a token bound to one. +// its roster, or a token bound to one. The command path does not read a home +// at all: it forwards Noise frames under a rendezvous handle. package server import ( @@ -22,6 +26,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/relay" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" "github.com/srcfl/ftw-cloud/internal/uplink" @@ -31,7 +36,7 @@ import ( const SourceArchive = "/usr/share/ftw-cloud/source.tar.gz" // Routes are every path the service answers. -var Routes = []string{"/v1/box", "/v1/app", "/v1/app/once", "/v1/now", "/source", "/healthz"} +var Routes = []string{"/v1/box", "/v1/app", "/v1/app/once", "/v1/now", "/r/", "/x/", "/deadman", "/source", "/healthz"} // devRoutes adds a local inspection route in binaries built with // -tags ftwclouddev. It is nil in every other build. @@ -48,7 +53,10 @@ type Server struct { // Tokens answers /v1/now. Tokens come only over the app session, so // without a Noise key every token read is refused. Tokens *token.Service - Log *slog.Logger + // Relay is the blind command path. Nil leaves /r/, /x/ and /deadman + // unmounted, which is only for a test that has not built one. + Relay *relay.Relay + Log *slog.Logger } // Handler is the whole HTTP surface. @@ -67,6 +75,12 @@ func (s *Server) Handler() http.Handler { } else { mux.HandleFunc("/v1/now", notYet) } + if s.Relay != nil { + mux.Handle("/r/", s.Relay) + mux.Handle("/x/", s.Relay) + mux.Handle("/deadman", s.Relay) + mux.Handle("/deadman/", s.Relay) + } mux.HandleFunc("/source", source) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store") diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 9d6e9d8..20d6662 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -15,6 +15,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/relay" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" "github.com/srcfl/ftw-cloud/internal/uplink" @@ -35,7 +36,7 @@ func newServer(t *testing.T) (*Server, string) { up := uplink.New(st, reg, key, log) noise, _ := appwire.GenerateKeyPair() s := &Server{Store: st, Registry: reg, Uplink: up, App: appsession.New(st, reg, up, noise, log), - Tokens: token.New(st, reg, log), Log: log} + Tokens: token.New(st, reg, log), Relay: relay.New(relay.Options{Log: log, BoxKeys: reg}), Log: log} // A home with readings in it, held the way a box's uplink holds them. const home = "canaryhomecanaryhomecanary" @@ -71,6 +72,22 @@ func TestNoRouteReadsAHome(t *testing.T) { } } +// A command posted at the exchange door is not a reading, and the reading +// the service holds is not an answer. +func TestACommandIsNotServedAsAReading(t *testing.T) { + s, _ := newServer(t) + body := bytes.Repeat([]byte{0x11}, 536) + copy(body, []byte("COMMAND-MARKER")) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodPost, "/x/1/0123456789abcdef0123456789abcdef", bytes.NewReader(body)) + r.Header.Set("Content-Type", "application/octet-stream") + r.RemoteAddr = "192.0.2.8:80" + s.Handler().ServeHTTP(w, r) + if bytes.Contains(w.Body.Bytes(), []byte("CANARY")) || bytes.Contains(w.Body.Bytes(), []byte("COMMAND-MARKER")) { + t.Fatalf("exchange body %q", w.Body.Bytes()) + } +} + func TestSourceSaysWhereTheSourceIs(t *testing.T) { s, _ := newServer(t) w := httptest.NewRecorder()