diff --git a/CLAUDE.md b/CLAUDE.md index 9f0971f..0634c78 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,7 +19,11 @@ maintains it. - **The home id comes from the key that signed hello** on this connection, 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 and tombstones. Nothing logs addresses. + id, the box key, the signed roster, owners' suspensions and tombstones. + Nothing logs addresses. +- **A phone can only narrow.** On `/v1/app` it can say whether it is looking, + and an owner can suspend a roster device. The service admits a phone by the + key its Noise handshake proved, and only if the box's roster lists it. - **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 44d20dc..f164e08 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ both in `srcfl/ftw`. Both are proposed. The service runs at `cloud.ftw.energy` ([deploy/README.md](deploy/README.md)), but no box sends to it yet. -## What works today (C2) +## What works today (C2, C3) - A box with FTW Cloud on keeps one WebSocket at `/v1/box`. The service checks that the box signed `hello` over this TLS session's exporter value, @@ -21,15 +21,26 @@ to it yet. - Each home's newest message of each class lives in memory, as the exact bytes the box sent. Readings are dropped an hour after the box goes quiet and are never written to disk. -- The registry (SQLite) holds the home id, the box key and the signed roster. - A box that turns FTW Cloud off gets its home deleted and a signed receipt. +- The registry (SQLite) holds the home id, the box key, the signed roster and + owners' suspensions. A box that turns FTW Cloud off gets its home deleted + and a signed receipt. - `/source` serves the source of the running build, as AGPL-3.0 asks. +- A phone reads its home at `/v1/app?home=`: Noise IK to the service's + Noise key, with the home id in the prologue. The service admits only a + phone whose key the box listed in its signed roster, and no owner has + suspended. Handshake message 2 carries the snapshot from memory; after it + each new box envelope the phone may read follows on the same socket, + unchanged. A roster change, a suspension, an expired roster or a purge + closes the sessions it touches at once. +- While a phone looks, the service asks the box for one-second updates and + asks again every 50 s; when nobody looks, it stops. +- Limits per home: 16 sessions, one per device, 256 envelopes queued per + phone, and a small message rate from each phone. Not built yet, with a marked place in the code: | Step | What | Where | |---|---|---| -| C3 | The app's Noise session, `/v1/app` | `internal/server`, `store.For` with a `Device` principal | | C4 | Alerts over APNs, `box.unreachable` | not started | | C5 | Widget token reads, `/v1/now`, tiles | `internal/server` | | C7–C8 | Statistics | not started | @@ -37,18 +48,21 @@ Not built yet, with a marked place in the code: ## How it holds a home ``` -box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) ◀── store.For(principal) - └────▶ registry (SQLite: key, roster, tombstones) +box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) ◀── store.For(principal) ◀── /v1/app ◀──Noise IK── phone + └────▶ registry (SQLite: key, roster, tombstones, suspensions) ``` - **The home id is a hash of the box's key.** It comes from the key that signed `hello`, never from a field anyone sends. A second box cannot claim a home that is not its own. - **One way in.** `store.For(principal)` is the only access to a home: a - `Box` principal may write its own home, and a `Device` principal (C3) may + `Box` principal may write its own home, and a `Device` principal may read the classes its role and the owner's grant both allow. There is no listing and no principal for staff; `TestNoRouteReadsAHome` checks every route. +- **A phone can narrow and nothing else.** It can say whether it is looking, + and an owner can suspend a device in the roster. Commands go from the app + 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. @@ -60,9 +74,10 @@ box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) ```bash go build -tags ftwclouddev -o ftwcloudd ./cmd/ftwcloudd -./ftwcloudd keygen -out service.key # prints the public key +./ftwcloudd keygen -out service.key # prints the public key, for Core +./ftwcloudd keygen -noise -out noise.key # prints the Noise public key, for the app ./ftwcloudd -listen 127.0.0.1:18443 -state ./state -service-key service.key \ - -tls-cert leaf.pem -tls-key leaf.key + -noise-key noise.key -tls-cert leaf.pem -tls-key leaf.key ``` Build Core from `srcfl/ftw` against it. A release build names no service, so @@ -84,6 +99,11 @@ curl -X POST http://127.0.0.1:8080/api/cloud/grant \ curl http://127.0.0.1:8080/api/cloud/preview ``` +Without `-noise-key` (or with the file missing) the service still serves +boxes, and `/v1/app` answers 501: phones read through the relay until a +Noise key is installed. A host that should serve phones needs the key as a +second secret beside the service key. + `-tags ftwclouddev` adds `/dev/held?home=` on loopback, which shows what the service holds so a test can compare it with the preview. Release builds do not have it. @@ -93,6 +113,9 @@ do not have it. ```bash go vet ./... && go test -race ./... go test -run XXX -fuzz FuzzBoxMessage -fuzztime 30s ./internal/uplink +go test -run XXX -fuzz FuzzPhoneMessage -fuzztime 30s ./internal/appsession +go test -run TestTheSnapshotArrivesFast -v ./internal/appsession # loopback p50/p95 +go test -run XXX -bench BenchmarkSnapshot ./internal/appsession bash scripts/check-imports.sh ``` diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index e758bd8..97847cd 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -2,12 +2,14 @@ // home that turned it on, in memory, and serves it to that home's phones. It // reads and never controls. See README.md. // -// ftwcloudd keygen -out service.key make the service key; prints its public half -// ftwcloudd [flags] serve +// 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 +// ftwcloudd [flags] serve package main import ( "context" + "crypto/ecdh" "crypto/ecdsa" "crypto/elliptic" "crypto/rand" @@ -28,10 +30,12 @@ import ( "syscall" "time" + "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/server" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/uplink" + "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" "golang.org/x/crypto/acme/autocert" ) @@ -56,6 +60,7 @@ func serve(args []string) error { domain := fs.String("domain", "cloud.ftw.energy", "name to get a certificate for") stateDir := fs.String("state", "/state", "directory for the registry and certificates") keyPath := fs.String("service-key", "/run/secrets/ftw-cloud-service.key", "service key, PEM") + noisePath := fs.String("noise-key", "/run/secrets/ftw-cloud-noise.key", "Noise key phones dial, PEM; without it /v1/app is off") certFile := fs.String("tls-cert", "", "certificate file, instead of ACME (local tests)") keyFile := fs.String("tls-key", "", "certificate key file, with -tls-cert") email := fs.String("acme-email", "", "contact for the certificate authority") @@ -67,6 +72,15 @@ func serve(args []string) error { if err != nil { return err } + // Phones read through /v1/app only with a Noise key. A host without one + // still serves boxes, and the app reads through the relay meanwhile. + noise, err := loadNoiseKey(*noisePath) + noiseOn := err == nil + if errors.Is(err, os.ErrNotExist) { + log.Warn("no Noise key: phones cannot read from FTW Cloud until one is installed", "path", *noisePath) + } else if err != nil { + return err + } reg, err := registry.Open(filepath.Join(*stateDir, "registry.db")) if err != nil { return err @@ -74,7 +88,13 @@ func serve(args []string) error { defer reg.Close() st := store.New(nil) - srv := &server.Server{Store: st, Registry: reg, Uplink: uplink.New(st, reg, key, log), Log: log} + st.UseSuspensions(reg.Suspended) + up := uplink.New(st, reg, key, log) + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Log: log} + if noiseOn { + srv.App = appsession.New(st, reg, up, noise, log) + up.OnConnect = srv.App.BoxConnected + } // TLS ends here and nowhere else: each box binds its handshake to this // TLS session, so a terminating proxy in front would break it. A plain @@ -119,7 +139,8 @@ func serve(args []string) error { }() pub, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) - log.Info("ftwcloudd listening", "addr", *listen, "service_key", base64.RawURLEncoding.EncodeToString(pub)) + log.Info("ftwcloudd listening", "addr", *listen, "service_key", base64.RawURLEncoding.EncodeToString(pub), + "noise_key", base64.RawURLEncoding.EncodeToString(noise.Public())) if err := httpSrv.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { return err } @@ -147,15 +168,53 @@ func loadServiceKey(path string) (*ecdsa.PrivateKey, error) { return key, nil } -// keygen writes a new service key and prints its public half, which Core -// carries in cloudlink.ServiceKeys. +// loadNoiseKey reads the X25519 key phones run Noise IK against, PKCS #8 PEM. +func loadNoiseKey(path string) (appwire.KeyPair, error) { + raw, err := os.ReadFile(path) + if err != nil { + return appwire.KeyPair{}, fmt.Errorf("noise key: %w", err) + } + block, _ := pem.Decode(raw) + if block == nil || block.Type != "PRIVATE KEY" { + return appwire.KeyPair{}, fmt.Errorf("noise key: %s is not a PKCS #8 private key", path) + } + parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes) + if err != nil { + return appwire.KeyPair{}, fmt.Errorf("noise key: %w", err) + } + x, ok := parsed.(*ecdh.PrivateKey) + if !ok || x.Curve() != ecdh.X25519() { + return appwire.KeyPair{}, fmt.Errorf("noise key: %s is not an X25519 key", path) + } + return appwire.KeyPairFromSecret(x.Bytes()) +} + +// keygen writes a new key and prints its public half. The service key's goes +// into Core's cloudlink.ServiceKeys; with -noise, the Noise key's goes into +// the app. func keygen(args []string) error { fs := flag.NewFlagSet("keygen", flag.ExitOnError) out := fs.String("out", "", "file to write the key to") + noise := fs.Bool("noise", false, "make the X25519 key phones dial instead of the P-256 service key") _ = fs.Parse(args) if *out == "" { return errors.New("keygen: -out is required") } + if *noise { + x, err := ecdh.X25519().GenerateKey(rand.Reader) + if err != nil { + return err + } + der, err := x509.MarshalPKCS8PrivateKey(x) + if err != nil { + return err + } + if err := writePEM(*out, "PRIVATE KEY", der); err != nil { + return err + } + fmt.Println(base64.RawURLEncoding.EncodeToString(x.PublicKey().Bytes())) + return nil + } key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) if err != nil { return err @@ -164,15 +223,7 @@ func keygen(args []string) error { if err != nil { return err } - f, err := os.OpenFile(*out, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) - if err != nil { - return err - } - if err := pem.Encode(f, &pem.Block{Type: "EC PRIVATE KEY", Bytes: der}); err != nil { - _ = f.Close() - return err - } - if err := f.Close(); err != nil { + if err := writePEM(*out, "EC PRIVATE KEY", der); err != nil { return err } pub, err := cloudwire.PublicKeyBytesOf(&key.PublicKey) @@ -182,3 +233,17 @@ func keygen(args []string) error { fmt.Println(base64.RawURLEncoding.EncodeToString(pub)) return nil } + +// writePEM writes a new key file that only its owner can read, and never over +// an existing one. +func writePEM(path, kind string, der []byte) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return err + } + if err := pem.Encode(f, &pem.Block{Type: kind, Bytes: der}); err != nil { + _ = f.Close() + return err + } + return f.Close() +} diff --git a/go.mod b/go.mod index 2c89de0..384d6c3 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.26.0 require ( github.com/fxamacker/cbor/v2 v2.9.4 github.com/gorilla/websocket v1.5.3 - github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805 + github.com/srcfl/ftw/go/protocol v0.0.0-20261013055540-c6301a0bb1f7 golang.org/x/crypto v0.57.0 golang.org/x/sys v0.48.0 modernc.org/sqlite v1.59.0 diff --git a/go.sum b/go.sum index 70f63f7..6483a7f 100644 --- a/go.sum +++ b/go.sum @@ -16,8 +16,8 @@ github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOF github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= -github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805 h1:DmeVlx+LhctSDD4sQS1ltcN8QD2EdaKiHtqU1Qn3K1U= -github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261013055540-c6301a0bb1f7 h1:+2OArNncGqfjaoBXbaZ2FJXi5S9t6iw7sh6ndAVGmeE= +github.com/srcfl/ftw/go/protocol v0.0.0-20261013055540-c6301a0bb1f7/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= diff --git a/internal/appsession/appsession.go b/internal/appsession/appsession.go new file mode 100644 index 0000000..d9b119a --- /dev/null +++ b/internal/appsession/appsession.go @@ -0,0 +1,468 @@ +// Package appsession serves /v1/app, where a phone reads its home. +// docs/protocol-cloud.md in srcfl/ftw is the wire format ("App session"); +// go/protocol/cloudwire holds its bytes. +// +// The phone runs Noise IK as initiator with the service's Noise key as the +// responder's static key and the home id in the prologue. The service admits +// the phone only through store.For with the key the handshake authenticated, +// so a phone reaches only a home whose box listed it in a signed roster. +// Message 2 carries the snapshot; after it every envelope the box sends that +// the device may read goes out unchanged, one frame per Noise message. +// +// A phone may ask for faster updates, and an owner may suspend a device. That +// is all it can do here: commands go to the box over the relay, never +// through FTW Cloud. +package appsession + +import ( + "crypto/tls" + "errors" + "log/slog" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +const ( + // maxPhoneMessage bounds one message from a phone. Message 1 with now is + // about 120 bytes and nothing a phone sends later is larger. + maxPhoneMessage = 4 << 10 + // handshakeTimeout bounds message 1. + handshakeTimeout = 10 * time.Second + // The same ping, read and write deadlines as the uplink. + pingEvery = 15 * time.Second + readTimeout = 45 * time.Second + writeTimeout = 10 * time.Second + // A phone sends a message when it looks away or back, or when an owner + // suspends a device. Past this burst at this rate the session closes. + rateBurst = 20 + ratePerSec = 2 + // maxNoiseMessage is the largest Noise message, handshake or transport. + maxNoiseMessage = 0xffff + // WatchRenew is how often the service asks the box again for one-second + // updates while a phone looks. The box's watch lasts 60 s, so the ask + // comes before it lapses. + WatchRenew = 50 * time.Second + // roleOwner is the registry's owner role, as the roster names it. + // go/protocol does not generate role names yet. + roleOwner = "owner" +) + +// Watcher asks a home's box for one-second updates. The uplink is one. +type Watcher interface { + Watch(home string) +} + +// Server is the /v1/app handler. +type Server struct { + store *store.Store + registry *registry.Registry + watcher Watcher + key appwire.KeyPair + log *slog.Logger + now func() time.Time + // renewEvery is WatchRenew; tests shorten it. + renewEvery time.Duration + + mu sync.Mutex + looking map[string]*look +} + +// look is one home's phones that can see it, and the loop that keeps the +// box's watch in force while there is at least one. +type look struct { + n int + stop chan struct{} +} + +// New builds the handler. key is the service's Noise static key; the app +// carries its public half. +func New(st *store.Store, reg *registry.Registry, w Watcher, key appwire.KeyPair, log *slog.Logger) *Server { + if log == nil { + log = slog.Default() + } + return &Server{store: st, registry: reg, watcher: w, key: key, log: log, now: time.Now, + renewEvery: WatchRenew, looking: map[string]*look{}} +} + +var upgrader = websocket.Upgrader{ + EnableCompression: false, + // The app sends no Origin. A browser always does, and has no business + // here. + CheckOrigin: func(r *http.Request) bool { return r.Header.Get("Origin") == "" }, +} + +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.TLS == nil || r.TLS.Version != tls.VersionTLS13 { + http.Error(w, "TLS 1.3 only", http.StatusUpgradeRequired) + return + } + home := r.URL.Query().Get(cloudwire.AppHomeParam) + if !cloudwire.ValidHome(home) { + http.Error(w, "name a home", http.StatusBadRequest) + return + } + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + conn.SetReadLimit(maxPhoneMessage) + code, err := s.serve(conn, home) + if err != nil { + s.log.Debug("app session closed", "err", err) + } + if code != 0 { + _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, ""), time.Now().Add(time.Second)) + } +} + +var errProtocol = errors.New("appsession: the phone broke the protocol") + +// serve runs one session and returns the close code to end it with. +func (s *Server) serve(conn *websocket.Conn, home string) (int, error) { + _ = conn.SetReadDeadline(time.Now().Add(handshakeTimeout)) + kind, msg1, err := conn.ReadMessage() + if err != nil { + return 0, err + } + if kind != websocket.BinaryMessage { + return cloudwire.CloseProtocol, errProtocol + } + hs, err := appwire.NewResponder(appwire.HandshakeOptions{StaticKey: s.key, Prologue: cloudwire.AppPrologue(home)}) + if err != nil { + return websocket.CloseInternalServerErr, err + } + payload, err := hs.ReadMessage(msg1) + if err != nil { + // Another service key, or another home in the prologue than in the + // query. Neither says anything about the home. + return cloudwire.CloseDenied, err + } + hz, err := parseNow(payload) + if err != nil { + return cloudwire.CloseProtocol, err + } + + dev, err := s.store.For(store.Principal{Kind: store.Device, Home: home, DeviceKey: hs.RemoteStatic()}) + if err != nil { + return cloudwire.CloseDenied, err + } + snap, sess, err := dev.Open() + switch { + case errors.Is(err, store.ErrBusy): + return cloudwire.CloseBusy, err + case err != nil: + return cloudwire.CloseDenied, err + } + defer sess.Close() + + msg2Payload, rest, err := snapshotPayload(snap) + if err != nil { + return websocket.CloseInternalServerErr, err + } + msg2, err := hs.WriteMessage(msg2Payload) + if err != nil { + return websocket.CloseInternalServerErr, err + } + c := &phoneConn{s: s, conn: conn, dev: dev, home: home, tokens: rateBurst, refilled: s.now()} + if err := c.write(msg2); err != nil { + return 0, err + } + keys, err := hs.Split() + if err != nil { + return websocket.CloseInternalServerErr, err + } + c.tr = appwire.NewTransport(keys) + for _, env := range rest { + if err := c.forward(env); err != nil { + return 0, err + } + } + + c.setLooking(hz >= cloudwire.HzLooking) + defer c.stopLooking() + return c.run(sess) +} + +// parseNow reads message 1's payload. +func parseNow(payload []byte) (int, error) { + env, err := cloudwire.DecodeEnvelope(payload) + if err != nil || env.T != cloudwire.MsgNow { + return 0, errProtocol + } + var now cloudwire.Now + if err := cloudwire.Unmarshal(env.B, &now); err != nil { + return 0, errProtocol + } + return now.Hz, nil +} + +// snapshotPayload is message 2. Envelopes that do not fit in one Noise +// message come back as rest, to follow as frames. +func snapshotPayload(snap store.Snapshot) ([]byte, [][]byte, error) { + envs := snap.Envelopes + var rest [][]byte + for { + payload, err := cloudwire.SnapshotPayload(envs, snap.LastSeenMs, snap.Online, len(rest)) + if err != nil { + return nil, nil, err + } + if len(payload)+appwire.Message2Overhead <= maxNoiseMessage || len(envs) == 0 { + return payload, rest, nil + } + rest = append([][]byte{envs[len(envs)-1]}, rest...) + envs = envs[:len(envs)-1] + } +} + +// setLooking keeps the box's watch in force while at least one phone of the +// home can see it. +func (s *Server) setLooking(home string, delta int) { + s.mu.Lock() + defer s.mu.Unlock() + l := s.looking[home] + if l == nil { + l = &look{} + s.looking[home] = l + } + l.n += delta + switch { + case l.n > 0 && l.stop == nil: + l.stop = make(chan struct{}) + go s.renew(home, l.stop) + case l.n <= 0: + if l.stop != nil { + close(l.stop) + } + delete(s.looking, home) + } +} + +func (s *Server) renew(home string, stop <-chan struct{}) { + s.watcher.Watch(home) + t := time.NewTicker(s.renewEvery) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + s.watcher.Watch(home) + } + } +} + +// BoxConnected asks a box that just connected for one-second updates when a +// phone is already looking. The uplink calls it after hello_ok. +func (s *Server) BoxConnected(home string) { + s.mu.Lock() + looking := s.looking[home] != nil + s.mu.Unlock() + if looking { + s.watcher.Watch(home) + } +} + +// phoneConn is one phone's session after the handshake. +type phoneConn struct { + s *Server + conn *websocket.Conn + tr *appwire.Transport + dev *store.Handle + home string + + // looking is touched by the read loop and by the session's end. + lookMu sync.Mutex + looking bool + ended bool + + tokens float64 + refilled time.Time +} + +func (c *phoneConn) setLooking(on bool) { + c.lookMu.Lock() + defer c.lookMu.Unlock() + if c.ended || on == c.looking { + return + } + c.looking = on + delta := -1 + if on { + delta = 1 + } + c.s.setLooking(c.home, delta) +} + +// stopLooking ends this phone's part in the watch for good, whatever the +// read loop still delivers. +func (c *phoneConn) stopLooking() { + c.setLooking(false) + c.lookMu.Lock() + c.ended = true + c.lookMu.Unlock() +} + +// write sends one WebSocket message. Only the session's own goroutine +// writes; pings go through WriteControl, which may run beside it. +func (c *phoneConn) write(msg []byte) error { + if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + return err + } + return c.conn.WriteMessage(websocket.BinaryMessage, msg) +} + +// forward sends one box envelope as the box sent it. An envelope too large +// for a Noise message cannot reach a phone; it is dropped, and the phone sees +// the gap through the next hb. +func (c *phoneConn) forward(env []byte) error { + frame, err := cloudwire.EncodeFrame(env) + if err != nil || len(frame)+appwire.TransportOverhead > maxNoiseMessage { + return nil + } + msg, err := c.tr.Encrypt(frame) + if err != nil { + return err + } + return c.write(msg) +} + +// run forwards until the store ends the session, the phone leaves or breaks +// the protocol. +func (c *phoneConn) run(sess *store.Session) (int, error) { + type result struct { + code int + err error + } + read := make(chan result, 1) + go func() { + code, err := c.readLoop() + read <- result{code, err} + }() + ping := time.NewTicker(pingEvery) + defer ping.Stop() + for { + select { + case env := <-sess.C(): + if err := c.forward(env); err != nil { + return 0, err + } + case <-sess.Done(): + return sess.Code(), nil + case r := <-read: + // The read loop's own work (a suspension) may have ended this + // session; the store's reason wins over a plain close. + select { + case <-sess.Done(): + return sess.Code(), r.err + default: + } + return r.code, r.err + case <-ping.C: + if err := c.conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeTimeout)); err != nil { + return 0, err + } + } + } +} + +func (c *phoneConn) readLoop() (int, error) { + _ = c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + c.conn.SetPongHandler(func(string) error { + return c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + }) + for { + kind, msg, err := c.conn.ReadMessage() + if err != nil { + return 0, err + } + _ = c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + if !c.allow() { + return cloudwire.CloseProtocol, errProtocol + } + if kind != websocket.BinaryMessage { + return cloudwire.CloseProtocol, errProtocol + } + frame, err := c.tr.Decrypt(msg) + if err != nil { + return cloudwire.CloseProtocol, err + } + if code, err := c.apply(frame); err != nil { + return code, err + } + } +} + +// apply acts on one message from the phone. Types it does not know are +// ignored, so a newer app works with an older service. +func (c *phoneConn) apply(frame []byte) (int, error) { + payload, err := cloudwire.DecodeFrame(frame) + if err != nil { + return cloudwire.CloseProtocol, err + } + env, err := cloudwire.DecodeEnvelope(payload) + if err != nil { + return cloudwire.CloseProtocol, err + } + switch env.T { + case cloudwire.MsgWatch: + var w cloudwire.Watch + if err := cloudwire.Unmarshal(env.B, &w); err != nil { + return cloudwire.CloseProtocol, err + } + c.setLooking(w.Hz >= cloudwire.HzLooking) + case cloudwire.MsgSuspend, cloudwire.MsgUnsuspend: + var sus cloudwire.Suspend + if err := cloudwire.Unmarshal(env.B, &sus); err != nil { + return cloudwire.CloseProtocol, err + } + return 0, c.suspend(sus.DeviceID, env.T == cloudwire.MsgSuspend) + } + return 0, nil +} + +// suspend applies an owner's suspension or lifts one. It only narrows what +// the roster allows: the device must be in the roster, and lifting a +// suspension gives back no more than the roster says. +func (c *phoneConn) suspend(deviceID string, on bool) error { + if c.dev.Device().Role != roleOwner { + return nil + } + r, _ := c.dev.Roster() + if r == nil { + return nil + } + for _, d := range r.Devices { + if d.ID != deviceID { + continue + } + if !on { + return c.s.registry.Unsuspend(c.home, d.ID) + } + if err := c.s.registry.Suspend(c.home, d.ID, d.Epoch, c.s.now()); err != nil { + return err + } + c.dev.Recheck() + return nil + } + return nil +} + +// allow is a token bucket over the phone's messages. +func (c *phoneConn) allow() bool { + now := c.s.now() + c.tokens = min(rateBurst, c.tokens+now.Sub(c.refilled).Seconds()*ratePerSec) + c.refilled = now + if c.tokens < 1 { + return false + } + c.tokens-- + return true +} diff --git a/internal/appsession/appsession_test.go b/internal/appsession/appsession_test.go new file mode 100644 index 0000000..408da70 --- /dev/null +++ b/internal/appsession/appsession_test.go @@ -0,0 +1,354 @@ +package appsession + +import ( + "bytes" + "crypto/ecdsa" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// The snapshot carries the box's own bytes in rising seq, and every later +// envelope follows on the same connection, unchanged. +func TestSnapshotThenStreamCarryTheBoxsBytes(t *testing.T) { + h := newHarness(t) + full := h.live(true) + plan := h.plan() + delta := h.live(false) + hb := h.hb() + + p := h.mustDial(dialOpts{key: h.owner, hz: 1}) + want := [][]byte{full, plan, delta, hb} + if len(p.snap.Envelopes) != len(want) || !p.snap.Online || p.snap.More != 0 { + t.Fatalf("snapshot: %d envelopes, online %v, more %d", len(p.snap.Envelopes), p.snap.Online, p.snap.More) + } + pub := &h.boxKey.PublicKey + for i, e := range p.snap.Envelopes { + if !bytes.Equal(e, want[i]) { + t.Fatalf("snapshot entry %d is not the box's bytes", i) + } + verify(t, pub, h.home, e) + } + + next := h.live(false) + got, code := p.next(2 * time.Second) + if code != 0 || !bytes.Equal(got, next) { + t.Fatalf("the stream carried %x (code %d), not the box's envelope", got, code) + } + verify(t, pub, h.home, got) +} + +func verify(t *testing.T, pub *ecdsa.PublicKey, home string, raw []byte) { + t.Helper() + env, err := cloudwire.DecodeEnvelope(raw) + if err != nil { + t.Fatal(err) + } + if err := cloudwire.Verify(pub, home, env); err != nil { + t.Fatalf("%s does not verify with the box key: %v", env.T, err) + } +} + +// A viewer reads its effective scope, on the snapshot and on the stream. +func TestAViewerReadsOnlyItsScope(t *testing.T) { + h := newHarness(t) + full := h.live(true) + h.plan() + p := h.mustDial(dialOpts{key: h.viewer, hz: 1}) + if len(p.snap.Envelopes) != 1 || !bytes.Equal(p.snap.Envelopes[0], full) { + t.Fatalf("viewer snapshot: %d envelopes", len(p.snap.Envelopes)) + } + h.plan() + hb := h.hb() + got, _ := p.next(2 * time.Second) + if !bytes.Equal(got, hb) { + t.Fatal("the viewer was sent a plan, or not the hb") + } +} + +// Every refusal closes before message 2 with the one code that says nothing +// about which check failed. +func TestRefusalsSayNothing(t *testing.T) { + h := newHarness(t) + h.live(true) + stranger, _ := appwire.GenerateKeyPair() + otherService, _ := appwire.GenerateKeyPair() + otherHome := "aaaaaaaaaaaaaaaaaaaaaaaaaa" + for name, o := range map[string]dialOpts{ + "a key the roster lacks": {key: stranger}, + "another service key": {key: h.owner, service: otherService.Public()}, + "another home in prologue": {key: h.owner, prologue: otherHome}, + "a home the service lacks": {key: h.owner, home: otherHome}, + } { + _, code, err := h.dial(o) + if err != nil || code != cloudwire.CloseDenied { + t.Errorf("%s: code %d, %v", name, code, err) + } + } +} + +// The doc's C3 proof: a revoke cuts the stream within a second. +func TestARosterWithoutTheDeviceClosesItsSessionWithinASecond(t *testing.T) { + h := newHarness(t) + h.live(true) + viewer := h.mustDial(dialOpts{key: h.viewer, hz: 1}) + owner := h.mustDial(dialOpts{key: h.owner, hz: 1}) + + start := time.Now() + h.setRoster(h.devices(1, 1)[:1], protocol.ScopeLiveRead, protocol.ScopePlanRead) + if code := viewer.closed(time.Second); code != cloudwire.CloseDenied { + t.Fatalf("revoked viewer closed with %d", code) + } + if took := time.Since(start); took > time.Second { + t.Fatalf("revoke took %v", took) + } + // The owner's access did not change, so its stream goes on. + next := h.live(false) + if got, code := owner.next(time.Second); code != 0 || !bytes.Equal(got, next) { + t.Fatalf("the owner's stream ended with the viewer's: code %d", code) + } + if _, code, _ := h.dial(dialOpts{key: h.viewer}); code != cloudwire.CloseDenied { + t.Fatalf("a revoked device reconnected: code %d", code) + } +} + +// Narrower scopes or a new epoch end the session; the phone comes back and +// learns what it may read now. +func TestAChangedAccessEndsTheSession(t *testing.T) { + h := newHarness(t) + h.live(true) + owner := h.mustDial(dialOpts{key: h.owner}) + h.setRoster(h.devices(1, 1), protocol.ScopeLiveRead) // plan dropped from the grant + if code := owner.closed(time.Second); code != cloudwire.CloseChanged { + t.Fatalf("narrowed grant closed with %d", code) + } + viewer := h.mustDial(dialOpts{key: h.viewer}) + h.setRoster(h.devices(1, 2), protocol.ScopeLiveRead) // viewer paired again + if code := viewer.closed(time.Second); code != cloudwire.CloseChanged { + t.Fatalf("new epoch closed with %d", code) + } + h.mustDial(dialOpts{key: h.viewer}) +} + +func TestTurningFTWCloudOffClosesEverySession(t *testing.T) { + h := newHarness(t) + h.live(true) + p := h.mustDial(dialOpts{key: h.owner}) + h.box.Delete() + if code := p.closed(time.Second); code != cloudwire.CloseDenied { + t.Fatalf("purge closed with %d", code) + } +} + +// An owner's suspension only narrows: it closes the device's session and +// keeps it out, a viewer cannot suspend anyone, a device outside the roster +// cannot be named, and lifting it gives back what the roster allows. +func TestSuspensionOnlyNarrows(t *testing.T) { + h := newHarness(t) + h.live(true) + owner := h.mustDial(dialOpts{key: h.owner}) + viewer := h.mustDial(dialOpts{key: h.viewer}) + + viewer.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "owner"}) + owner.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "nobody"}) + next := h.live(false) + if got, code := owner.next(time.Second); code != 0 || !bytes.Equal(got, next) { + t.Fatalf("a viewer suspended the owner: code %d", code) + } + if h.reg.Suspended(h.home, "nobody", 0) { + t.Fatal("a device outside the roster was suspended") + } + + owner.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "viewer"}) + if code := viewer.closed(time.Second); code != cloudwire.CloseDenied { + t.Fatalf("suspended viewer closed with %d", code) + } + if _, code, _ := h.dial(dialOpts{key: h.viewer}); code != cloudwire.CloseDenied { + t.Fatalf("a suspended device reconnected: code %d", code) + } + + owner.send(cloudwire.MsgUnsuspend, cloudwire.Suspend{DeviceID: "viewer"}) + waitFor(t, "the suspension to lift", func() bool { return !h.reg.Suspended(h.home, "viewer", 1) }) + h.mustDial(dialOpts{key: h.viewer}) + + // Paired again on the box: a new epoch ends a suspension. + owner.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "viewer"}) + waitFor(t, "the suspension", func() bool { return h.reg.Suspended(h.home, "viewer", 1) }) + h.setRoster(h.devices(1, 2), protocol.ScopeLiveRead, protocol.ScopePlanRead) + h.mustDial(dialOpts{key: h.viewer}) +} + +// While a phone looks, the box is asked for one-second updates, and asked +// again before its watch lapses. When nobody looks, the asking stops. +func TestTheBoxIsWatchedWhileAPhoneLooks(t *testing.T) { + h := newHarness(t) + h.app.renewEvery = 40 * time.Millisecond + h.live(true) + + away := h.mustDial(dialOpts{key: h.viewer, hz: 0}) + time.Sleep(60 * time.Millisecond) + if n := h.watch.count(); n != 0 { + t.Fatalf("a phone that is not looking caused %d watches", n) + } + looking := h.mustDial(dialOpts{key: h.owner, hz: 1}) + waitFor(t, "renewals", func() bool { return h.watch.count() >= 3 }) + + // Two phones looking, one looks away: still watched. + away.send(cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}) + looking.send(cloudwire.MsgWatch, cloudwire.Watch{Hz: 0}) + n := h.watch.count() + waitFor(t, "renewals with one phone left", func() bool { return h.watch.count() >= n+2 }) + + away.send(cloudwire.MsgWatch, cloudwire.Watch{Hz: 0}) + waitFor(t, "the watch to stop", func() bool { + h.app.mu.Lock() + defer h.app.mu.Unlock() + return len(h.app.looking) == 0 + }) + n = h.watch.count() + time.Sleep(120 * time.Millisecond) + if h.watch.count() != n { + t.Fatal("the service kept asking with nobody looking") + } + + // A closed session stops its share too. + away.send(cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}) + waitFor(t, "a watch", func() bool { return h.watch.count() > n }) + away.conn.Close() + waitFor(t, "the watch to stop after close", func() bool { + h.app.mu.Lock() + defer h.app.mu.Unlock() + return len(h.app.looking) == 0 + }) +} + +func TestABoxThatComesBackIsAskedAtOnce(t *testing.T) { + h := newHarness(t) + h.app.renewEvery = time.Hour + h.live(true) + h.mustDial(dialOpts{key: h.owner, hz: 1}) + waitFor(t, "the first watch", func() bool { return h.watch.count() == 1 }) + h.app.BoxConnected(h.home) + h.app.BoxConnected("aaaaaaaaaaaaaaaaaaaaaaaaaa") + if n := h.watch.count(); n != 2 { + t.Fatalf("watches after the box came back: %d", n) + } +} + +func TestANewSessionFromTheSameDeviceReplacesTheOld(t *testing.T) { + h := newHarness(t) + h.live(true) + first := h.mustDial(dialOpts{key: h.owner}) + h.mustDial(dialOpts{key: h.owner}) + if code := first.closed(time.Second); code != cloudwire.CloseReplaced { + t.Fatalf("the older session closed with %d", code) + } +} + +func TestAHomeHoldsABoundedNumberOfSessions(t *testing.T) { + h := newHarness(t) + h.live(true) + var devices []cloudwire.Device + var keys []appwire.KeyPair + for i := 0; i <= store.MaxSessionsPerHome; i++ { + k, _ := appwire.GenerateKeyPair() + keys = append(keys, k) + devices = append(devices, cloudwire.Device{ID: string(rune('a' + i)), Key: k.Public(), Role: "viewer", + Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1}) + } + h.setRoster(devices, protocol.ScopeLiveRead) + for _, k := range keys[:store.MaxSessionsPerHome] { + h.mustDial(dialOpts{key: k}) + } + if _, code, _ := h.dial(dialOpts{key: keys[store.MaxSessionsPerHome]}); code != cloudwire.CloseBusy { + t.Fatalf("one session over the limit: code %d", code) + } +} + +// A snapshot larger than one Noise message spills its newest envelopes into +// frames right after message 2, still in rising seq. +func TestALargeSnapshotSpillsIntoFrames(t *testing.T) { + h := newHarness(t) + h.live(true) + big := func(class string) []byte { + return h.put(class, func(hdr cloudwire.Header) any { + return struct { + cloudwire.Header + Pad []byte `cbor:"pad"` + }{hdr, bytes.Repeat([]byte{7}, 30_000)} + }, false) + } + plan := big(cloudwire.MsgPlan) + prices := big(cloudwire.MsgPrices) + lp := big(cloudwire.MsgLP) + p := h.mustDial(dialOpts{key: h.owner}) + if p.snap.More == 0 || len(p.snap.Envelopes)+p.snap.More != 4 { + t.Fatalf("snapshot held %d, more %d", len(p.snap.Envelopes), p.snap.More) + } + var rest [][]byte + for range p.snap.More { + env, code := p.next(time.Second) + if code != 0 { + t.Fatalf("closed with %d", code) + } + rest = append(rest, env) + } + var all [][]byte + for _, e := range p.snap.Envelopes { + all = append(all, e) + } + all = append(all, rest...) + for i, want := range [][]byte{plan, prices, lp} { + if !bytes.Equal(all[i+1], want) { + t.Fatalf("entry %d out of order", i+1) + } + } +} + +func TestAPhoneThatFloodsIsClosed(t *testing.T) { + h := newHarness(t) + h.live(true) + p := h.mustDial(dialOpts{key: h.owner}) + for range rateBurst + 5 { + env, _ := cloudwire.Unsigned("chatter", struct{}{}) + frame, _ := cloudwire.EncodeFrame(env) + msg, _ := p.tr.Encrypt(frame) + if p.conn.WriteMessage(websocket.BinaryMessage, msg) != nil { + break + } + } + if code := p.closed(2 * time.Second); code != cloudwire.CloseProtocol { + t.Fatalf("a flood closed with %d", code) + } +} + +func TestATamperedMessageClosesTheSession(t *testing.T) { + h := newHarness(t) + h.live(true) + p := h.mustDial(dialOpts{key: h.owner}) + env, _ := cloudwire.Unsigned(cloudwire.MsgWatch, cloudwire.Watch{Hz: 0}) + frame, _ := cloudwire.EncodeFrame(env) + msg, _ := p.tr.Encrypt(frame) + msg[len(msg)-1] ^= 1 + _ = p.conn.WriteMessage(websocket.BinaryMessage, msg) + if code := p.closed(2 * time.Second); code != cloudwire.CloseProtocol { + t.Fatalf("a tampered message closed with %d", code) + } +} + +func TestPlainHTTPAndABadHomeAreRefused(t *testing.T) { + h := newHarness(t) + resp, err := h.srv.Client().Get(h.srv.URL + "/v1/app?home=nothome") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != 400 { + t.Fatalf("a bad home: %d", resp.StatusCode) + } +} diff --git a/internal/appsession/fuzz_test.go b/internal/appsession/fuzz_test.go new file mode 100644 index 0000000..67cd253 --- /dev/null +++ b/internal/appsession/fuzz_test.go @@ -0,0 +1,78 @@ +package appsession + +import ( + "bytes" + "testing" + + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// Whatever a phone sends after the handshake, the service neither panics nor +// does more than the protocol allows: a phone can change whether it looks, +// and an owner can suspend a device in the roster. Nothing else changes. +func FuzzPhoneMessage(f *testing.F) { + h := newHarness(f) + h.live(true) + for _, seed := range [][]byte{ + mustFrame(cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}), + mustFrame(cloudwire.MsgWatch, cloudwire.Watch{Hz: 0}), + mustFrame(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "viewer"}), + mustFrame(cloudwire.MsgUnsuspend, cloudwire.Suspend{DeviceID: "viewer"}), + mustFrame("token.get", struct{}{}), + {1, 1, 0, 0, 0, 1, 0xa0}, + {1, 0, 0, 0, 0, 0}, + {}, + } { + f.Add(seed) + } + f.Fuzz(func(t *testing.T, frame []byte) { + for _, key := range []appwire.KeyPair{h.viewer, h.owner} { + dev, err := h.store.For(storePrincipal(h, key)) + if err != nil { + t.Fatal(err) + } + c := &phoneConn{s: h.app, dev: dev, home: h.home, tokens: rateBurst, refilled: h.app.now()} + code, err := c.apply(frame) + if err != nil && code != cloudwire.CloseProtocol && code != 0 { + t.Fatalf("closed with %d", code) + } + c.stopLooking() + // Only the owner may suspend, and only the viewer exists to be + // suspended besides the owner itself. + if bytes.Equal(key.Public(), h.viewer.Public()) && h.reg.Suspended(h.home, "owner", 1) { + t.Fatal("a viewer suspended the owner") + } + _ = h.reg.Unsuspend(h.home, "viewer") + _ = h.reg.Unsuspend(h.home, "owner") + } + h.app.mu.Lock() + left := len(h.app.looking) + h.app.mu.Unlock() + if left != 0 { + t.Fatal("a session that ended is still counted as looking") + } + }) +} + +// Message 1's payload: anything at all is refused or read, never a panic. +func FuzzNow(f *testing.F) { + now, _ := cloudwire.Unsigned(cloudwire.MsgNow, cloudwire.Now{Hz: 1}) + f.Add(now) + f.Add([]byte{0xa0}) + f.Fuzz(func(t *testing.T, payload []byte) { + _, _ = parseNow(payload) + }) +} + +func mustFrame(t string, body any) []byte { + env, err := cloudwire.Unsigned(t, body) + if err != nil { + panic(err) + } + frame, err := cloudwire.EncodeFrame(env) + if err != nil { + panic(err) + } + return frame +} diff --git a/internal/appsession/harness_test.go b/internal/appsession/harness_test.go new file mode 100644 index 0000000..7386374 --- /dev/null +++ b/internal/appsession/harness_test.go @@ -0,0 +1,310 @@ +package appsession + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// A home with a box key, a roster with an owner and a viewer, and a TLS +// server answering /v1/app the way ftwcloudd does. +type harness struct { + t testing.TB + store *store.Store + reg *registry.Registry + app *Server + srv *httptest.Server + watch *watchCount + noise appwire.KeyPair + boxKey *ecdsa.PrivateKey + home string + box *store.Handle + owner appwire.KeyPair + viewer appwire.KeyPair + seq uint64 + rosterN uint64 + boot []byte +} + +type watchCount struct { + mu sync.Mutex + homes []string +} + +func (w *watchCount) Watch(home string) { + w.mu.Lock() + defer w.mu.Unlock() + w.homes = append(w.homes, home) +} + +func (w *watchCount) count() int { + w.mu.Lock() + defer w.mu.Unlock() + return len(w.homes) +} + +func newHarness(t testing.TB) *harness { + t.Helper() + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(nil) + st.UseSuspensions(reg.Suspended) + noise, _ := appwire.GenerateKeyPair() + w := &watchCount{} + app := New(st, reg, w, noise, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv := httptest.NewUnstartedServer(app) + srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + srv.StartTLS() + t.Cleanup(srv.Close) + + boxKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + pk, _ := cloudwire.PublicKeyBytesOf(&boxKey.PublicKey) + home := cloudwire.HomeID(pk) + box, err := st.For(store.Principal{Kind: store.Box, Home: home}) + if err != nil { + t.Fatal(err) + } + box.Connected() + if err := reg.Claim(home, pk, time.Now()); err != nil { + t.Fatal(err) + } + owner, _ := appwire.GenerateKeyPair() + viewer, _ := appwire.GenerateKeyPair() + h := &harness{t: t, store: st, reg: reg, app: app, srv: srv, watch: w, noise: noise, boxKey: boxKey, + home: home, box: box, owner: owner, viewer: viewer, boot: bytes.Repeat([]byte{0xb0}, cloudwire.BootBytes)} + h.setRoster(h.devices(1, 1), protocol.ScopeLiveRead, protocol.ScopePlanRead) + return h +} + +func (h *harness) devices(ownerEpoch, viewerEpoch uint64) []cloudwire.Device { + return []cloudwire.Device{ + {ID: "owner", Key: h.owner.Public(), Role: "owner", Scopes: protocol.CloudScopes, Epoch: ownerEpoch}, + {ID: "viewer", Key: h.viewer.Public(), Role: "viewer", Scopes: []string{protocol.ScopeLiveRead}, Epoch: viewerEpoch}, + } +} + +func (h *harness) setRoster(devices []cloudwire.Device, grant ...string) { + h.rosterN++ + r := &cloudwire.Roster{RosterSeq: h.rosterN, TTLS: 86_400, Devices: devices, + Grant: cloudwire.Grant{ID: make([]byte, 16), Scopes: grant, Purposes: []string{protocol.CloudPurposeServe}, Controller: "sourceful"}} + if !h.box.SetRoster(r) { + h.t.Fatal("roster refused") + } + epochs := map[string]uint64{} + for _, d := range devices { + epochs[d.ID] = d.Epoch + } + if err := h.reg.PruneSuspensions(h.home, epochs); err != nil { + h.t.Fatal(err) + } +} + +// put signs a body as the box and holds it, as the uplink does. +func (h *harness) put(t string, body func(cloudwire.Header) any, table bool) []byte { + h.t.Helper() + h.seq++ + hdr := cloudwire.Header{Seq: h.seq, Boot: h.boot, UptimeMs: int64(60_000 + h.seq*100)} + env, err := cloudwire.Seal(h.boxKey, h.home, t, body(hdr)) + if err != nil { + h.t.Fatal(err) + } + raw, err := env.Encode() + if err != nil { + h.t.Fatal(err) + } + if err := h.box.Put(t, raw, h.seq, table); err != nil { + h.t.Fatal(err) + } + return raw +} + +func (h *harness) live(table bool) []byte { + return h.put(cloudwire.MsgLive, func(hdr cloudwire.Header) any { + l := cloudwire.Live{Header: hdr, Fields: map[string]int64{"2": 1800}, SrcRev: 1, Ages: map[string]int64{"m": 400}} + if table { + l.Sources = map[string]cloudwire.Source{"m": {Kind: "driver", State: protocol.SourceLive, StaleAfterMs: 10_000}} + l.FieldSources = map[string][]string{"2": {"m"}} + } + return l + }, table) +} + +func (h *harness) plan() []byte { + return h.put(cloudwire.MsgPlan, func(hdr cloudwire.Header) any { + return struct { + cloudwire.Header + DecisionID string `cbor:"decisionId"` + }{hdr, "d1"} + }, false) +} + +func (h *harness) hb() []byte { + return h.put(cloudwire.MsgHB, func(hdr cloudwire.Header) any { + return cloudwire.HB{Header: hdr, Clock: cloudwire.ClockNTP, SrcRev: 1, LiveSeq: h.seq - 1} + }, false) +} + +// phone is a Go client of /v1/app. +type phone struct { + t testing.TB + conn *websocket.Conn + tr *appwire.Transport + snap cloudwire.Snapshot +} + +type dialOpts struct { + key appwire.KeyPair + service []byte // the service key the phone pins; nil uses the harness's + home string // query; "" uses the harness's + prologue string // home in the prologue; "" uses query + hz int +} + +// dial runs the handshake. It returns the close code when the service refuses +// before message 2. +func (h *harness) dial(o dialOpts) (*phone, int, error) { + h.t.Helper() + if o.service == nil { + o.service = h.noise.Public() + } + if o.home == "" { + o.home = h.home + } + if o.prologue == "" { + o.prologue = o.home + } + d := websocket.Dialer{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, MinVersion: tls.VersionTLS13}} //nolint:gosec // the test's own server + url := strings.Replace(h.srv.URL, "https://", "wss://", 1) + "/v1/app?home=" + o.home + conn, _, err := d.Dial(url, nil) + if err != nil { + return nil, 0, err + } + init, err := appwire.NewInitiator(appwire.InitiatorOptions{StaticKey: o.key, RemoteStatic: o.service, Prologue: cloudwire.AppPrologue(o.prologue)}) + if err != nil { + h.t.Fatal(err) + } + now, _ := cloudwire.Unsigned(cloudwire.MsgNow, cloudwire.Now{Hz: o.hz}) + msg1, _ := init.WriteMessage(now) + if err := conn.WriteMessage(websocket.BinaryMessage, msg1); err != nil { + return nil, 0, err + } + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, msg2, err := conn.ReadMessage() + if err != nil { + var ce *websocket.CloseError + if errors.As(err, &ce) { + conn.Close() + return nil, ce.Code, nil + } + return nil, 0, err + } + payload, err := init.ReadMessage(msg2) + if err != nil { + h.t.Fatalf("message 2: %v", err) + } + snap, err := cloudwire.ParseSnapshot(payload) + if err != nil { + h.t.Fatalf("snapshot: %v", err) + } + keys, _ := init.Split() + p := &phone{t: h.t, conn: conn, tr: appwire.NewTransport(keys), snap: snap} + h.t.Cleanup(func() { conn.Close() }) + return p, 0, nil +} + +func (h *harness) mustDial(o dialOpts) *phone { + h.t.Helper() + p, code, err := h.dial(o) + if err != nil || code != 0 { + h.t.Fatalf("dial: code %d, %v", code, err) + } + return p +} + +// next reads one forwarded envelope, or the close code. +func (p *phone) next(within time.Duration) ([]byte, int) { + p.t.Helper() + _ = p.conn.SetReadDeadline(time.Now().Add(within)) + _, msg, err := p.conn.ReadMessage() + if err != nil { + var ce *websocket.CloseError + if errors.As(err, &ce) { + return nil, ce.Code + } + p.t.Fatalf("read: %v", err) + } + frame, err := p.tr.Decrypt(msg) + if err != nil { + p.t.Fatalf("decrypt: %v", err) + } + env, err := cloudwire.DecodeFrame(frame) + if err != nil { + p.t.Fatalf("frame: %v", err) + } + return env, 0 +} + +// closed waits for the service to end the session and returns its code. +func (p *phone) closed(within time.Duration) int { + p.t.Helper() + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if _, code := p.next(time.Until(deadline)); code != 0 { + return code + } + } + p.t.Fatal("the session stayed open") + return 0 +} + +func (p *phone) send(t string, body any) { + p.t.Helper() + env, _ := cloudwire.Unsigned(t, body) + frame, _ := cloudwire.EncodeFrame(env) + msg, err := p.tr.Encrypt(frame) + if err != nil { + p.t.Fatal(err) + } + if err := p.conn.WriteMessage(websocket.BinaryMessage, msg); err != nil { + p.t.Fatal(err) + } +} + +func waitFor(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for !ok() { + if time.Now().After(deadline) { + t.Fatalf("waited 3 s for %s", what) + } + time.Sleep(5 * time.Millisecond) + } +} + +var _ http.Handler = (*Server)(nil) + +func storePrincipal(h *harness, key appwire.KeyPair) store.Principal { + return store.Principal{Kind: store.Device, Home: h.home, DeviceKey: key.Public()} +} diff --git a/internal/appsession/perf_test.go b/internal/appsession/perf_test.go new file mode 100644 index 0000000..e203758 --- /dev/null +++ b/internal/appsession/perf_test.go @@ -0,0 +1,111 @@ +package appsession + +import ( + "slices" + "testing" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// A home the size of a real one: the source table, a live, an lp with one +// car, a day and a half of quarter-hour prices and an hb. +func (h *harness) fillLikeAHouse() { + srcs := map[string]cloudwire.Source{} + ages := map[string]int64{} + for _, id := range []string{"ACfHt4DWL7g", "f-oHMDqV-mA", "DUYWWQmn-Xg", "j3HZORNCWHs"} { + srcs[id] = cloudwire.Source{Kind: "driver", State: protocol.SourceLive, LastOkMs: 59_000, StaleAfterMs: 10_000} + ages[id] = 900 + } + fields := map[string]int64{"1": 0, "2": 1840, "3": -2310, "4": 950, "5": 612, "6": 1100, "10": 2100} + h.put(cloudwire.MsgLive, func(hdr cloudwire.Header) any { + return cloudwire.Live{Header: hdr, Fields: fields, Sources: srcs, SrcRev: 1, Ages: ages, + FieldSources: map[string][]string{"2": {"ACfHt4DWL7g"}, "3": {"f-oHMDqV-mA"}, "10": {"DUYWWQmn-Xg"}}} + }, true) + h.put(cloudwire.MsgLive, func(hdr cloudwire.Header) any { + return cloudwire.Live{Header: hdr, Fields: fields, SrcRev: 1, Ages: ages} + }, false) + v := func(x int64) *int64 { return &x } + h.put(cloudwire.MsgLP, func(hdr cloudwire.Header) any { + return cloudwire.LP{Header: hdr, ControlRev: 812, Loadpoints: []cloudwire.Loadpoint{{ + ID: "j3HZORNCWHs", Src: "DUYWWQmn-Xg", Plugged: true, SocPermille: v(430), SocSource: "inferred", + TargetPermille: v(800), ReadyByMs: v(1_791_867_600_000), PowerW: 2100, SessionWh: 3400, Session: "s1", + CommandedW: v(2070), CommandedReason: "plan", NextStartMs: v(1_791_850_000_000)}}, + Feedback: []cloudwire.Feedback{{Kind: "ev", Src: "DUYWWQmn-Xg", Status: "following", Reason: "plan", + Severity: "ok", Evidence: "measured", Fresh: true, ActualW: v(2100), AgeMs: v(900)}}} + }, false) + var spot, total []*int64 + for i := range 144 { + spot, total = append(spot, v(int64(40+i%30))), append(total, v(int64(110+i%30))) + } + h.put(cloudwire.MsgPrices, func(hdr cloudwire.Header) any { + return cloudwire.Prices{Header: hdr, Zone: "SE3", Currency: "SEK", T0Ms: 1_791_820_800_000, SlotMs: 900_000, + SpotMinor: spot, TotalMinor: total} + }, false) + h.put(cloudwire.MsgHB, func(hdr cloudwire.Header) any { + return cloudwire.HB{Header: hdr, Clock: cloudwire.ClockNTP, SrcRev: 1, LiveSeq: 2, LpSeq: 3, Ages: ages} + }, false) +} + +// The doc's speed claim for the second layer: the cloud's copy, from memory, +// in well under 100 ms. Here the whole of it on loopback: TCP, TLS 1.3, the +// WebSocket upgrade, Noise IK and the snapshot, twenty times. +func TestTheSnapshotArrivesFast(t *testing.T) { + h := newHarness(t) + h.fillLikeAHouse() + var took []time.Duration + var size int + for range 20 { + start := time.Now() + p := h.mustDial(dialOpts{key: h.owner, hz: 1}) + took = append(took, time.Since(start)) + size = len(p.snap.Envelopes) + p.conn.Close() + } + slices.Sort(took) + p50, p95 := took[len(took)/2], took[len(took)*95/100] + t.Logf("handshake and snapshot of %d envelopes on loopback: p50 %v, p95 %v, max %v", size, p50, p95, took[len(took)-1]) + if p95 > 100*time.Millisecond { + t.Fatalf("p95 %v is over the 100 ms the snapshot has", p95) + } +} + +// What the service spends per phone between message 1 and message 2: the +// store's snapshot, the payload and the Noise work, with no network. +func BenchmarkSnapshot(b *testing.B) { + h := newHarness(b) + h.fillLikeAHouse() + now, _ := cloudwire.Unsigned(cloudwire.MsgNow, cloudwire.Now{Hz: 1}) + prologue := cloudwire.AppPrologue(h.home) + b.ReportAllocs() + var bytesOut int + for b.Loop() { + init, _ := appwire.NewInitiator(appwire.InitiatorOptions{StaticKey: h.owner, RemoteStatic: h.noise.Public(), Prologue: prologue}) + msg1, _ := init.WriteMessage(now) + resp, _ := appwire.NewResponder(appwire.HandshakeOptions{StaticKey: h.noise, Prologue: prologue}) + if _, err := resp.ReadMessage(msg1); err != nil { + b.Fatal(err) + } + dev, err := h.store.For(storePrincipal(h, h.owner)) + if err != nil { + b.Fatal(err) + } + snap, sess, err := dev.Open() + if err != nil { + b.Fatal(err) + } + payload, _, err := snapshotPayload(snap) + if err != nil { + b.Fatal(err) + } + msg2, err := resp.WriteMessage(payload) + if err != nil { + b.Fatal(err) + } + sess.Close() + bytesOut = len(msg2) + } + b.ReportMetric(float64(bytesOut), "msg2-bytes") +} diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 6808993..0cfaca4 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -63,6 +63,13 @@ CREATE TABLE IF NOT EXISTS tombstones ( home TEXT PRIMARY KEY, grant_id BLOB, deleted_ms INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS suspensions ( + home TEXT NOT NULL, + device_id TEXT NOT NULL, + epoch INTEGER NOT NULL, + since_ms INTEGER NOT NULL, + PRIMARY KEY (home, device_id) );`) if err != nil { _ = db.Close() @@ -130,6 +137,9 @@ func (r *Registry) Purge(home string, grantID []byte, now time.Time) error { if _, err := tx.Exec(`DELETE FROM homes WHERE home = ?`, home); err != nil { return err } + if _, err := tx.Exec(`DELETE FROM suspensions WHERE home = ?`, home); err != nil { + return err + } if _, err := tx.Exec(`INSERT INTO tombstones (home, grant_id, deleted_ms) VALUES (?, ?, ?) ON CONFLICT (home) DO NOTHING`, home, grantID, now.UnixMilli()); err != nil { return err @@ -182,7 +192,66 @@ func (r *Registry) Expire(now time.Time) ([]string, error) { if _, err := r.db.Exec(`DELETE FROM homes WHERE home = ?`, h); err != nil { return nil, err } + if _, err := r.db.Exec(`DELETE FROM suspensions WHERE home = ?`, h); err != nil { + return nil, err + } } _, err = r.db.Exec(`DELETE FROM tombstones WHERE deleted_ms < ?`, now.Add(-TombstoneKept).UnixMilli()) return gone, err } + +// Suspend records an owner's suspension of one roster device at its current +// epoch. It only narrows access: the device must already be in the roster, +// which the caller checks, and the suspension ends when the roster drops the +// device or its epoch changes. +func (r *Registry) Suspend(home, deviceID string, epoch uint64, now time.Time) error { + _, err := r.db.Exec(`INSERT INTO suspensions (home, device_id, epoch, since_ms) VALUES (?, ?, ?, ?) + ON CONFLICT (home, device_id) DO UPDATE SET epoch = excluded.epoch, since_ms = excluded.since_ms`, + home, deviceID, int64(epoch), now.UnixMilli()) + return err +} + +// Unsuspend lifts a suspension. +func (r *Registry) Unsuspend(home, deviceID string) error { + _, err := r.db.Exec(`DELETE FROM suspensions WHERE home = ? AND device_id = ?`, home, deviceID) + return err +} + +// Suspended reports whether a device is suspended at this epoch. A read that +// fails counts as suspended: the safe answer narrows. +func (r *Registry) Suspended(home, deviceID string, epoch uint64) bool { + var held int64 + err := r.db.QueryRow(`SELECT epoch FROM suspensions WHERE home = ? AND device_id = ?`, home, deviceID).Scan(&held) + if errors.Is(err, sql.ErrNoRows) { + return false + } + return err != nil || uint64(held) == epoch +} + +// PruneSuspensions drops the suspensions a new roster ended: devices it no +// longer lists, and devices whose epoch moved. +func (r *Registry) PruneSuspensions(home string, epochs map[string]uint64) error { + rows, err := r.db.Query(`SELECT device_id, epoch FROM suspensions WHERE home = ?`, home) + if err != nil { + return err + } + var gone []string + for rows.Next() { + var id string + var epoch int64 + if err := rows.Scan(&id, &epoch); err != nil { + _ = rows.Close() + return err + } + if e, ok := epochs[id]; !ok || e != uint64(epoch) { + gone = append(gone, id) + } + } + _ = rows.Close() + for _, id := range gone { + if err := r.Unsuspend(home, id); err != nil { + return err + } + } + return nil +} diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go index b878673..6a9a685 100644 --- a/internal/registry/registry_test.go +++ b/internal/registry/registry_test.go @@ -70,3 +70,31 @@ func TestHomesWithoutARosterExpire(t *testing.T) { t.Fatalf("after 90 days: %v", gone) } } + +// A suspension holds at the epoch it was made at, ends when the roster drops +// the device or pairs it again, and goes with the home. +func TestASuspensionHoldsUntilTheRosterMovesPastIt(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + if err := r.Suspend("h1", "phone", 2, now); err != nil { + t.Fatal(err) + } + if !r.Suspended("h1", "phone", 2) || r.Suspended("h1", "phone", 3) || r.Suspended("h2", "phone", 2) { + t.Fatal("a suspension reached past its device and epoch") + } + if err := r.PruneSuspensions("h1", map[string]uint64{"phone": 2}); err != nil || !r.Suspended("h1", "phone", 2) { + t.Fatal("an unchanged roster lifted a suspension") + } + if err := r.PruneSuspensions("h1", map[string]uint64{"phone": 3}); err != nil || r.Suspended("h1", "phone", 3) { + t.Fatal("a new epoch kept the suspension") + } + _ = r.Suspend("h1", "phone", 3, now) + if err := r.PruneSuspensions("h1", map[string]uint64{}); err != nil || r.Suspended("h1", "phone", 3) { + t.Fatal("a device the roster dropped stayed suspended") + } + _ = r.Suspend("h1", "phone", 3, now) + if err := r.Purge("h1", nil, now); err != nil || r.Suspended("h1", "phone", 3) { + t.Fatal("a purge left the suspension") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index eb4a0c0..da5d10c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -2,7 +2,7 @@ // route is listed in Routes, and TestNoRouteReadsAHome holds the list. // // /v1/box a box's uplink (package uplink) -// /v1/app the app's Noise session to the service: C3, not built yet +// /v1/app a phone's Noise session to the service (package appsession) // /v1/now an extension's token read: C5, not built yet // /source the source of the running service, as AGPL-3.0 asks // /healthz liveness, and nothing about any home @@ -19,6 +19,7 @@ import ( "runtime/debug" "time" + "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/uplink" @@ -39,18 +40,20 @@ type Server struct { Store *store.Store Registry *registry.Registry Uplink *uplink.Uplink - Log *slog.Logger + // App is nil on a host without a Noise key; /v1/app then says so. + App *appsession.Server + Log *slog.Logger } // Handler is the whole HTTP surface. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.Handle("/v1/box", s.Uplink) - // C3: the app opens a WebSocket here and runs Noise IK to the service's - // static key, with prologue "ftw.cloud.v1:" and the home id. The - // service admits a phone only through store.For with a Device - // principal, and forwards the box's envelopes unchanged. - mux.HandleFunc("/v1/app", notYet) + if s.App != nil { + mux.Handle("/v1/app", s.App) + } else { + mux.HandleFunc("/v1/app", notYet) + } // C5: widgets read the newest live, prices and hb here with a 7-day // token bound to one roster entry. mux.HandleFunc("/v1/now", notYet) diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 1322d9c..d1e0b4a 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -13,9 +13,11 @@ import ( "slices" "testing" + "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/uplink" + "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" ) @@ -29,13 +31,16 @@ func newServer(t *testing.T) (*Server, string) { t.Cleanup(func() { _ = reg.Close() }) st := store.New(nil) log := slog.New(slog.NewTextHandler(io.Discard, nil)) - s := &Server{Store: st, Registry: reg, Uplink: uplink.New(st, reg, key, log), Log: log} + 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), Log: log} // A home with readings in it, held the way a box's uplink holds them. - box, _ := st.For(store.Principal{Kind: store.Box, Home: "canaryhome"}) + const home = "canaryhomecanaryhomecanary" + box, _ := st.For(store.Principal{Kind: store.Box, Home: home}) box.Connected() _ = box.Put(cloudwire.MsgLive, []byte("CANARY-READING"), 1, true) - return s, "canaryhome" + return s, home } // No route reads a home. Every path answers without the canary, whatever the @@ -73,11 +78,25 @@ func TestSourceSaysWhereTheSourceIs(t *testing.T) { } } -func TestTheUplinkRefusesPlainHTTP(t *testing.T) { - s, _ := newServer(t) +func TestTheSocketsRefusePlainHTTP(t *testing.T) { + s, home := newServer(t) + for _, p := range []string{"/v1/box", "/v1/app?home=" + home} { + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, p, nil)) + if w.Code != http.StatusUpgradeRequired { + t.Fatalf("plain HTTP to %s: %d", p, w.Code) + } + } +} + +// A host without a Noise key serves boxes and says /v1/app is not there yet; +// phones read through the relay meanwhile. +func TestWithoutANoiseKeyTheAppEndpointIsOff(t *testing.T) { + s, home := newServer(t) + s.App = nil w := httptest.NewRecorder() - s.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/box", nil)) - if w.Code != http.StatusUpgradeRequired { - t.Fatalf("plain HTTP to /v1/box: %d", w.Code) + s.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/app?home="+home, nil)) + if w.Code != http.StatusNotImplemented { + t.Fatalf("/v1/app without a Noise key: %d", w.Code) } } diff --git a/internal/store/session.go b/internal/store/session.go new file mode 100644 index 0000000..7398408 --- /dev/null +++ b/internal/store/session.go @@ -0,0 +1,121 @@ +package store + +import ( + "bytes" + + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// Limits on the phones reading one home. +const ( + // MaxSessionsPerHome bounds the phones one home can hold open at once. + // A household has a few phones; past this the home is busy, as it is at + // the relay. + MaxSessionsPerHome = 16 + // SessionQueue is how many envelopes may wait for one slow phone. A box + // sends a few a second while watched, so a phone this far behind has a + // dead socket. It is closed and gets a fresh snapshot when it comes + // back, rather than a stream with a hole in it. + SessionQueue = 256 + + // WebSocket close codes for the two endings that are not the protocol's + // own: the phone left, or it fell too far behind. + closeNormal = 1000 + closeSlow = 1013 +) + +// Session is one phone's live view of its home: every envelope the box sends +// after the snapshot that the device may read, in the order the box sent +// them. The store ends it when the device's access changes. +type Session struct { + x *Handle + c chan []byte + // done is closed when the store ends the session; code says why, as a + // close code from cloudwire. + done chan struct{} + code int + ended bool +} + +// Open returns the device's snapshot and a session that receives every later +// envelope it may read. Both are taken under one lock, so nothing the box +// sends falls between them. A second session from the same device replaces +// the first. +func (x *Handle) Open() (Snapshot, *Session, error) { + if x.kind != Device { + return Snapshot{}, nil, ErrDenied + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + // The access checked in For may have changed since: check it again + // under the lock that roster changes take. + if _, _, ok := x.s.access(x.id, x.h, x.device.Key); !ok { + return Snapshot{}, nil, ErrDenied + } + for sess := range x.h.sessions { + if bytes.Equal(sess.x.device.Key, x.device.Key) { + sess.endLocked(cloudwire.CloseReplaced) + } + } + if len(x.h.sessions) >= MaxSessionsPerHome { + return Snapshot{}, nil, ErrBusy + } + sess := &Session{x: x, c: make(chan []byte, SessionQueue), done: make(chan struct{})} + x.h.sessions[sess] = struct{}{} + return x.snapshotLocked(), sess, nil +} + +// C carries the envelopes to forward. +func (s *Session) C() <-chan []byte { return s.c } + +// Done is closed when the store ends the session. +func (s *Session) Done() <-chan struct{} { return s.done } + +// Code is why the store ended the session, once Done is closed. +func (s *Session) Code() int { + s.x.h.mu.Lock() + defer s.x.h.mu.Unlock() + return s.code +} + +// Close is the phone leaving. It is safe to call more than once. +func (s *Session) Close() { + s.x.h.mu.Lock() + defer s.x.h.mu.Unlock() + s.endLocked(closeNormal) +} + +// deliver queues one envelope without ever blocking the box. The caller holds +// the home's lock. +func (s *Session) deliver(env []byte) { + if s.ended { + return + } + select { + case s.c <- env: + default: + s.endLocked(closeSlow) + } +} + +// endLocked ends the session once. The caller holds the home's lock. +func (s *Session) endLocked(code int) { + if s.ended { + return + } + s.ended, s.code = true, code + delete(s.x.h.sessions, s) + close(s.done) +} + +// Device is the roster entry this device matched. +func (x *Handle) Device() cloudwire.Device { return x.device } + +// Scopes is the device's effective scope. +func (x *Handle) Scopes() []string { return x.scopes } + +// Home is the home this handle reaches. +func (x *Handle) Home() string { return x.id } + +// MayRead reports whether the device may read a class. +func (x *Handle) MayRead(class string) bool { return x.mayRead(class) } diff --git a/internal/store/store.go b/internal/store/store.go index 0713484..1d46fff 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -44,11 +44,17 @@ type Principal struct { // ErrDenied is every refusal. It does not say whether the home exists. var ErrDenied = errors.New("store: denied") +// ErrBusy is a home with as many phone sessions open as it may have. +var ErrBusy = errors.New("store: busy") + // Store is every home the service holds. type Store struct { mu sync.Mutex homes map[string]*home now func() time.Time + // suspended reports an owner's suspension of a roster device. Nil + // suspends nobody. + suspended func(home, deviceID string, epoch uint64) bool } // New starts an empty store. now is for tests; nil uses the clock. @@ -59,6 +65,12 @@ func New(now func() time.Time) *Store { return &Store{homes: map[string]*home{}, now: now} } +// UseSuspensions sets where owners' suspensions are kept. Call it before +// serving. +func (s *Store) UseSuspensions(suspended func(home, deviceID string, epoch uint64) bool) { + s.suspended = suspended +} + type home struct { mu sync.Mutex // latest is the newest envelope of each class, exactly as the box sent @@ -70,6 +82,8 @@ type home struct { online bool conn uint64 lastSeen time.Time + // sessions are the phones reading this home now. + sessions map[*Session]struct{} } // Held is one envelope as received. @@ -85,19 +99,20 @@ type Handle struct { id string h *home kind Kind - // scopes is a device's effective scope: its roster scopes intersected - // with the grant's. + // device is the roster entry a device principal matched, and scopes its + // effective scope: its roster scopes intersected with the grant's. + device cloudwire.Device scopes []string } // For returns the principal's access to its home, or ErrDenied. A box may // create its home; a device reaches only a home whose current roster lists -// its key. +// its key, and only while no owner has suspended it. func (s *Store) For(p Principal) (*Handle, error) { s.mu.Lock() h := s.homes[p.Home] if h == nil && p.Kind == Box && p.Home != "" { - h = &home{latest: map[string]Held{}} + h = &home{latest: map[string]Held{}, sessions: map[*Session]struct{}{}} s.homes[p.Home] = h } s.mu.Unlock() @@ -110,23 +125,38 @@ func (s *Store) For(p Principal) (*Handle, error) { case Device: h.mu.Lock() defer h.mu.Unlock() - r := h.roster - if r == nil || s.now().After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { + dev, scopes, ok := s.access(p.Home, h, p.DeviceKey) + if !ok { return nil, ErrDenied } - for _, d := range r.Devices { - if bytes.Equal(d.Key, p.DeviceKey) && len(p.DeviceKey) == 32 { - var scopes []string - for _, sc := range d.Scopes { - if slices.Contains(r.Grant.Scopes, sc) { - scopes = append(scopes, sc) - } - } - return &Handle{s: s, id: p.Home, h: h, kind: Device, scopes: scopes}, nil + return &Handle{s: s, id: p.Home, h: h, kind: Device, device: dev, scopes: scopes}, nil + } + return nil, ErrDenied +} + +// access is the one admission rule for a device: a roster in force that +// lists its key, and no suspension. The caller holds h.mu. +func (s *Store) access(id string, h *home, key []byte) (cloudwire.Device, []string, bool) { + r := h.roster + if r == nil || len(key) != 32 || s.now().After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { + return cloudwire.Device{}, nil, false + } + for _, d := range r.Devices { + if !bytes.Equal(d.Key, key) { + continue + } + if s.suspended != nil && s.suspended(id, d.ID, d.Epoch) { + return cloudwire.Device{}, nil, false + } + var scopes []string + for _, sc := range d.Scopes { + if slices.Contains(r.Grant.Scopes, sc) { + scopes = append(scopes, sc) } } + return d, scopes, true } - return nil, ErrDenied + return cloudwire.Device{}, nil, false } // -------------------------------------------------------------------------- @@ -167,6 +197,11 @@ func (x *Handle) Put(t string, env []byte, seq uint64, carriesTable bool) error x.h.full = &held } x.h.lastSeen = x.s.now() + for sess := range x.h.sessions { + if sess.x.mayRead(t) { + sess.deliver(env) + } + } return nil } @@ -192,9 +227,32 @@ func (x *Handle) SetRoster(r *cloudwire.Roster) bool { return false } x.h.roster, x.h.rosterAt = r, x.s.now() + x.recheckLocked() return true } +// Recheck ends every session whose access is no longer what it was opened +// with, as after an owner suspends a device. The phone reconnects and learns +// its new access, or is refused. +func (x *Handle) Recheck() { + x.h.mu.Lock() + defer x.h.mu.Unlock() + x.recheckLocked() +} + +func (x *Handle) recheckLocked() { + for sess := range x.h.sessions { + dev, scopes, ok := x.s.access(x.id, x.h, sess.x.device.Key) + switch { + case !ok: + sess.endLocked(cloudwire.CloseDenied) + case dev.ID != sess.x.device.ID || dev.Epoch != sess.x.device.Epoch || dev.Role != sess.x.device.Role || + !slices.Equal(scopes, sess.x.scopes): + sess.endLocked(cloudwire.CloseChanged) + } + } +} + // Delete forgets the home. Only its box may. func (x *Handle) Delete() { if x.kind != Box { @@ -205,6 +263,9 @@ func (x *Handle) Delete() { x.s.mu.Unlock() x.h.mu.Lock() x.h.latest, x.h.full, x.h.roster = map[string]Held{}, nil, nil + for sess := range x.h.sessions { + sess.endLocked(cloudwire.CloseDenied) + } x.h.mu.Unlock() } @@ -228,9 +289,13 @@ func (x *Handle) Snapshot() (Snapshot, error) { } x.h.mu.Lock() defer x.h.mu.Unlock() + return x.snapshotLocked(), nil +} + +func (x *Handle) snapshotLocked() Snapshot { out := Snapshot{LastSeenMs: x.h.lastSeen.UnixMilli(), Online: x.h.online} if !x.h.online && x.s.now().Sub(x.h.lastSeen) > ReadingsKeptFor { - return out, nil + return out } var held []Held for t, e := range x.h.latest { @@ -245,7 +310,7 @@ func (x *Handle) Snapshot() (Snapshot, error) { for _, e := range held { out.Envelopes = append(out.Envelopes, e.Envelope) } - return out, nil + return out } func (x *Handle) mayRead(class string) bool { @@ -272,6 +337,13 @@ func (s *Store) Sweep() { if !h.online && now.Sub(h.lastSeen) > ReadingsKeptFor { h.latest, h.full = map[string]Held{}, nil } + // A roster that ran out closes the home's sessions: the box has not + // confirmed who may read for a day. + if r := h.roster; r != nil && now.After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { + for sess := range h.sessions { + sess.endLocked(cloudwire.CloseDenied) + } + } h.mu.Unlock() } } @@ -280,6 +352,14 @@ func (s *Store) Sweep() { // roster. It is called by upkeep, never by a request. func (s *Store) Forget(id string) { s.mu.Lock() - defer s.mu.Unlock() + h := s.homes[id] delete(s.homes, id) + s.mu.Unlock() + if h != nil { + h.mu.Lock() + for sess := range h.sessions { + sess.endLocked(cloudwire.CloseDenied) + } + h.mu.Unlock() + } } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index d0c1078..162077a 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -165,3 +165,98 @@ func TestDeleteForgetsTheHome(t *testing.T) { t.Fatal("a deleted home still answers") } } + +// A session's snapshot and stream join without a gap: what the box sends +// after Open reaches the session, and nothing before it does twice. +func TestASessionMissesNothingAfterItsSnapshot(t *testing.T) { + s, box, _ := filled(t) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: viewerKey}) + snap, sess, err := dev.Open() + if err != nil { + t.Fatal(err) + } + defer sess.Close() + if len(snap.Envelopes) != 3 { + t.Fatalf("snapshot held %d", len(snap.Envelopes)) + } + _ = box.Put(cloudwire.MsgPlan, []byte("plan-2"), 8, false) // outside the viewer's scope + _ = box.Put(cloudwire.MsgLive, []byte("live-3"), 9, false) + select { + case got := <-sess.C(): + if string(got) != "live-3" { + t.Fatalf("the session got %q", got) + } + default: + t.Fatal("nothing reached the session") + } + select { + case got := <-sess.C(): + t.Fatalf("the session also got %q", got) + default: + } +} + +// A phone too far behind is ended, never allowed to stall the box. +func TestASlowSessionIsEndedNotWaitedFor(t *testing.T) { + s, box, _ := filled(t) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + _, sess, _ := dev.Open() + for i := range SessionQueue + 1 { + _ = box.Put(cloudwire.MsgHB, []byte("hb"), uint64(100+i), false) + } + select { + case <-sess.Done(): + default: + t.Fatal("a full queue left the session open") + } + if sess.Code() != closeSlow { + t.Fatalf("ended with %d", sess.Code()) + } +} + +func TestARosterThatRunsOutEndsSessions(t *testing.T) { + s, _, c := filled(t) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + _, sess, _ := dev.Open() + c.t = c.t.Add(24*time.Hour + time.Second) + s.Sweep() + select { + case <-sess.Done(): + default: + t.Fatal("an expired roster left a session open") + } + if sess.Code() != cloudwire.CloseDenied { + t.Fatalf("ended with %d", sess.Code()) + } +} + +func TestForgettingAHomeEndsItsSessions(t *testing.T) { + s, _, _ := filled(t) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + _, sess, _ := dev.Open() + s.Forget("h1") + select { + case <-sess.Done(): + default: + t.Fatal("a forgotten home kept a session") + } +} + +// A suspended device is refused, and only at the epoch it was suspended at. +func TestSuspensionsAreConsultedOnEveryAdmission(t *testing.T) { + s, _, _ := filled(t) + s.UseSuspensions(func(home, id string, epoch uint64) bool { return home == "h1" && id == "viewer" && epoch == 0 }) + if _, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: viewerKey}); !errors.Is(err, ErrDenied) { + t.Fatal("a suspended device was admitted") + } + if _, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}); err != nil { + t.Fatal("the owner was refused with the viewer") + } +} + +func TestABoxHandleCannotOpenASession(t *testing.T) { + _, box, _ := filled(t) + if _, _, err := box.Open(); !errors.Is(err, ErrDenied) { + t.Fatal("the box's handle opened a session") + } +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index da1e4f6..e550361 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -62,6 +62,10 @@ type Uplink struct { // Dropped counts box messages of a type this service does not know. A // newer box works with an older service by this rule. Dropped atomic.Int64 + + // OnConnect runs after a box's hello_ok, so the app sessions can ask a + // box that just came back for one-second updates. Set before serving. + OnConnect func(home string) } // New builds the handler. key is the service key that signs hello_ok and @@ -107,7 +111,7 @@ func (u *Uplink) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // Watch asks a home's box for one-second updates for a minute. The app -// session (C3) calls it while a phone is looking. +// sessions call it while a phone is looking. func (u *Uplink) Watch(home string) { u.mu.Lock() c := u.conns[home] @@ -202,6 +206,9 @@ func (c *boxConn) run() error { if err := c.send(helloOK{BoxNonce: hello.BoxNonce, EKM: c.ekm, Have: map[string]string{}, RosterSeq: c.rosterSeq()}); err != nil { return err } + if !c.purged && c.u.OnConnect != nil { + go c.u.OnConnect(c.home) + } done := make(chan struct{}) defer close(done) @@ -385,7 +392,16 @@ func (c *boxConn) roster(env cloudwire.Envelope) error { if err != nil { return err } - return c.u.registry.SetRoster(c.home, r.RosterSeq, r.Grant.ID, raw, c.u.now()) + if err := c.u.registry.SetRoster(c.home, r.RosterSeq, r.Grant.ID, raw, c.u.now()); err != nil { + return err + } + // A device the roster dropped, or whose epoch moved, is no longer + // suspended: it is gone, or paired again. + epochs := make(map[string]uint64, len(r.Devices)) + for _, d := range r.Devices { + epochs[d.ID] = d.Epoch + } + return c.u.registry.PruneSuspensions(c.home, epochs) } // purge deletes the home and sends the signed receipt. For a home already