diff --git a/CLAUDE.md b/CLAUDE.md index 0634c78..ebc8527 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,8 +19,8 @@ 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, owners' suspensions and tombstones. - Nothing logs addresses. + id, the box key, the signed roster, owners' suspensions, tombstones, APNs + registrations and the hashes of extension tokens. 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. diff --git a/README.md b/README.md index 46a327f..9da9a92 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, C3, C4) +## What works today - 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,8 +21,9 @@ 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, the signed roster and - owners' suspensions. A box that turns FTW Cloud off gets its home deleted +- The registry (SQLite) holds the home id, the box key, the signed roster, + owners' suspensions, APNs registrations and the hashes of extension + tokens. 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, and names its commit and image digest. @@ -48,27 +49,36 @@ to it yet. device, and go when the roster drops them or Apple says they are gone. Apple's key comes from Parameter Store ([deploy/apns.md](deploy/apns.md)). +- **Widget token.** A device asks for its extension token over its app + session with `token.get`. A widget reads `GET /v1/now` with it: the newest + `live`, `prices` and `hb` as the box signed them, and nothing else. The + service keeps the token's SHA-256, never the token. A device holds one + 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. -Not built yet, with a marked place in the code: +Not built yet: | Step | What | Where | |---|---|---| -| C5 | Widget token reads, `/v1/now`, tiles | `internal/server` | +| C6 | Tiles, widget push | `internal/server`, `internal/push` | | C7–C8 | Statistics | not started | ## How it holds a home ``` 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) + │ ◀── /v1/now ◀──token──── widget + └────▶ registry (SQLite: key, roster, tombstones, suspensions, APNs, token hashes) ``` - **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 may - read the classes its role and the owner's grant both allow. There is no + `Box` principal may write its own home, a `Device` principal may + read the classes its role and the owner's grant both allow, and a `Token` + 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 can narrow and nothing else.** It can say whether it is looking, diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index f0d89f7..edb78eb 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -37,6 +37,7 @@ import ( "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/token" "github.com/srcfl/ftw-cloud/internal/uplink" "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" @@ -97,7 +98,8 @@ func serve(args []string) error { st := store.New(nil) st.UseSuspensions(reg.Suspended) up := uplink.New(st, reg, key, log) - srv := &server.Server{Store: st, Registry: reg, Uplink: up, Log: log} + tokens := token.New(st, reg, log) + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Log: log} // Alerts: the box's signed events to its phones through Apple, and // box.unreachable when a box is away. Without a key the service keeps @@ -117,7 +119,10 @@ func serve(args []string) error { } pushSvc := push.New(reg, sender, log) up.OnEvent = pushSvc.Event - up.OnRoster = pushSvc.Roster + up.OnRoster = func(home string, r *cloudwire.Roster) { + pushSvc.Roster(home, r) + tokens.Roster(home, r) + } up.OnGone = pushSvc.BoxGone // Phones register their tokens over their app session, so without a @@ -125,6 +130,7 @@ func serve(args []string) error { if noiseOn { srv.App = appsession.New(st, reg, up, noise, log) srv.App.Push = pushSvc + srv.App.Tokens = tokens } app := srv.App up.OnConnect = func(home string) { diff --git a/deploy/README.md b/deploy/README.md index 9d43c01..f5db043 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -153,7 +153,8 @@ curl -sI https://cloud.ftw.energy/source `ok` means DNS, the certificate and the service all work. `/source` returns the source archive, and its headers name the commit (`X-Source-Revision`) and the image digest (`X-Source-Image`), which should match what CI built. -`/v1/app` and `/v1/now` answer 501 until C3 and C5. +`/v1/app` answers 501 on a host without the Noise key, and `/v1/now` answers +401 to a request without a valid token. ## Cost @@ -198,6 +199,7 @@ Cloud data; the images on `ghcr.io` hold only code. ## Not here yet The APNs key ([apns.md](apns.md)) and the Noise key phones dial -([noise.md](noise.md)) come from Parameter Store. Widget tokens (C5) and -statistics (C7–C8) add secrets, egress and storage when they land. The erase test, canary homes +([noise.md](noise.md)) come from Parameter Store. Widget tokens need no +secret: the registry holds their hashes. Statistics (C7–C8) add secrets, +egress and storage when they land. The erase test, canary homes and the external heartbeat that pages Sourceful belong with the first homes. diff --git a/go.mod b/go.mod index f4567f9..e48952a 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-20261013083029-b1b34697d7ab + github.com/srcfl/ftw/go/protocol v0.0.0-20261014110822-7c72a5b33aa0 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 d680714..be93d77 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-20261013083029-b1b34697d7ab h1:InL7LLEvfapkns81aQJCIQq0PUEPQ8JRzMtSEaMqWNo= -github.com/srcfl/ftw/go/protocol v0.0.0-20261013083029-b1b34697d7ab/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261014110822-7c72a5b33aa0 h1:4p9TFWdkIk5y2fK3/9FOeUV9AukTjIBd7jQZDLG8Qx8= +github.com/srcfl/ftw/go/protocol v0.0.0-20261014110822-7c72a5b33aa0/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 index 8835240..ac554c7 100644 --- a/internal/appsession/appsession.go +++ b/internal/appsession/appsession.go @@ -9,9 +9,9 @@ // 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. +// A phone may ask for faster updates, ask for its extension token, register +// for alerts, 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 ( @@ -74,6 +74,9 @@ type Server struct { // Push keeps the phones' APNs tokens. Nil ignores push.register and // push.unregister. Set before serving. Push Pusher + // Tokens issues extension tokens. Nil ignores token.get. Set before + // serving. + Tokens Tokens mu sync.Mutex looking map[string]*look @@ -86,6 +89,13 @@ type Pusher interface { Unregister(home string, dev cloudwire.Device, r cloudwire.PushUnregister) error } +// Tokens gives a device its extension token. token.Service is one. dev +// comes from store.For, never from the phone's message. +type Tokens interface { + Issue(home string, dev cloudwire.Device) (cloudwire.Token, error) + Drop(home, deviceID string) error +} + // 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 { @@ -183,7 +193,8 @@ func (s *Server) serve(conn *websocket.Conn, home string) (int, error) { if err != nil { return websocket.CloseInternalServerErr, err } - c := &phoneConn{s: s, conn: conn, dev: dev, home: home, tokens: rateBurst, refilled: s.now()} + c := &phoneConn{s: s, conn: conn, dev: dev, home: home, tokens: rateBurst, refilled: s.now(), + replies: make(chan []byte, 1), quit: make(chan struct{})} if err := c.write(msg2); err != nil { return 0, err } @@ -297,6 +308,12 @@ type phoneConn struct { tokens float64 refilled time.Time + + // replies carries the service's answers from the read loop to the + // session's goroutine, the only one that writes; quit closes when that + // goroutine is done. + replies chan []byte + quit chan struct{} } func (c *phoneConn) setLooking(on bool) { @@ -349,6 +366,7 @@ func (c *phoneConn) forward(env []byte) error { // run forwards until the store ends the session, the phone leaves or breaks // the protocol. func (c *phoneConn) run(sess *store.Session) (int, error) { + defer close(c.quit) type result struct { code int err error @@ -366,6 +384,10 @@ func (c *phoneConn) run(sess *store.Session) (int, error) { if err := c.forward(env); err != nil { return 0, err } + case env := <-c.replies: + if err := c.forward(env); err != nil { + return 0, err + } case <-sess.Done(): return sess.Code(), nil case r := <-read: @@ -452,10 +474,35 @@ func (c *phoneConn) apply(frame []byte) (int, error) { if c.s.Push != nil { return 0, pushed(c.s.Push.Unregister(c.home, c.dev.Device(), r)) } + case cloudwire.MsgTokenGet: + if c.s.Tokens == nil { + return 0, nil + } + tok, err := c.s.Tokens.Issue(c.home, c.dev.Device()) + if err != nil { + return 0, err + } + return 0, c.say(cloudwire.MsgToken, tok) } return 0, nil } +// say hands an answer to the session's goroutine to send. +func (c *phoneConn) say(t string, body any) error { + env, err := cloudwire.Unsigned(t, body) + if err != nil { + return err + } + select { + case c.replies <- env: + return nil + case <-c.quit: + return errClosed + } +} + +var errClosed = errors.New("appsession: the session ended") + // pushed lets the session go on after a registration the device may not // make, such as a viewer's: it gets no alerts, and nothing else changes. func pushed(err error) error { @@ -486,7 +533,13 @@ func (c *phoneConn) suspend(deviceID string, on bool) error { if err := c.s.registry.Suspend(c.home, d.ID, d.Epoch, c.s.now()); err != nil { return err } + // The suspended device's sessions end first, whatever happens to + // its token. Its token dies; lifting the suspension does not bring + // it back, the app asks for a new one. c.dev.Recheck() + if c.s.Tokens != nil { + return c.s.Tokens.Drop(c.home, d.ID) + } return nil } return nil diff --git a/internal/appsession/token_test.go b/internal/appsession/token_test.go new file mode 100644 index 0000000..9f0c6d7 --- /dev/null +++ b/internal/appsession/token_test.go @@ -0,0 +1,149 @@ +package appsession + +import ( + "bytes" + "crypto/tls" + "errors" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/srcfl/ftw-cloud/internal/token" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// tokens gives the harness a token service and /v1/now behind TLS, and +// returns a reader that answers with the status a token gets there. +func (h *harness) tokens(t *testing.T) func(tok string) int { + t.Helper() + svc := token.New(h.store, h.reg, nil) + h.app.Tokens = svc + srv := httptest.NewUnstartedServer(svc) + srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + srv.StartTLS() + t.Cleanup(srv.Close) + return func(tok string) int { + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/v1/now", nil) + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + return resp.StatusCode + } +} + +// getToken asks over the session and reads the answer, skipping any box +// envelope that arrives first. +func (p *phone) getToken() cloudwire.Token { + p.t.Helper() + p.send(cloudwire.MsgTokenGet, struct{}{}) + for { + raw, code := p.next(2 * time.Second) + if code != 0 { + p.t.Fatalf("closed with %d", code) + } + env, err := cloudwire.DecodeEnvelope(raw) + if err != nil { + p.t.Fatal(err) + } + if env.T != cloudwire.MsgToken { + continue + } + if len(env.Sig) != 0 { + p.t.Fatal("the service signed its token answer") + } + var tok cloudwire.Token + if err := cloudwire.Unmarshal(env.B, &tok); err != nil { + p.t.Fatal(err) + } + return tok + } +} + +// A device asks for its token over its own session. It reads /v1/now, the +// next one replaces it, and the session goes on. +func TestADeviceGetsItsTokenOverItsSession(t *testing.T) { + h := newHarness(t) + read := h.tokens(t) + h.live(true) + viewer := h.mustDial(dialOpts{key: h.viewer}) + first := viewer.getToken() + if first.ExpiresInS != int64(token.Life/time.Second) { + t.Fatalf("expires in %d s", first.ExpiresInS) + } + if code := read(first.Token); code != http.StatusOK { + t.Fatalf("the token read %d", code) + } + second := viewer.getToken() + if code := read(first.Token); code != http.StatusUnauthorized { + t.Fatalf("the replaced token read %d", code) + } + if code := read(second.Token); code != http.StatusOK { + t.Fatalf("the new token read %d", code) + } + next := h.live(false) + if got, code := viewer.next(time.Second); code != 0 || !bytes.Equal(got, next) { + t.Fatalf("the session did not go on: code %d", code) + } +} + +// An owner's suspension kills the device's token; lifting it does not bring +// the token back. +func TestASuspensionKillsTheToken(t *testing.T) { + h := newHarness(t) + read := h.tokens(t) + h.live(true) + owner := h.mustDial(dialOpts{key: h.owner}) + viewer := h.mustDial(dialOpts{key: h.viewer}) + tok := viewer.getToken() + owner.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "viewer"}) + if code := viewer.closed(2 * time.Second); code != cloudwire.CloseDenied { + t.Fatalf("the suspended viewer closed with %d", code) + } + if code := read(tok.Token); code != http.StatusUnauthorized { + t.Fatalf("a suspended device's token read %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) }) + if code := read(tok.Token); code != http.StatusUnauthorized { + t.Fatalf("lifting the suspension brought the token back: %d", code) + } +} + +// Without a token service, token.get is ignored and the session goes on. +func TestWithoutTokensTokenGetIsIgnored(t *testing.T) { + h := newHarness(t) + h.live(true) + owner := h.mustDial(dialOpts{key: h.owner}) + owner.send(cloudwire.MsgTokenGet, struct{}{}) + next := h.live(false) + if got, code := owner.next(time.Second); code != 0 || !bytes.Equal(got, next) { + t.Fatalf("the session did not go on: code %d", code) + } +} + +// The suspended device's session ends even when its token cannot be dropped. +func TestASuspensionEndsTheSessionWhateverTheToken(t *testing.T) { + h := newHarness(t) + h.app.Tokens = failingTokens{} + h.live(true) + owner := h.mustDial(dialOpts{key: h.owner}) + viewer := h.mustDial(dialOpts{key: h.viewer}) + owner.send(cloudwire.MsgSuspend, cloudwire.Suspend{DeviceID: "viewer"}) + if code := viewer.closed(2 * time.Second); code != cloudwire.CloseDenied { + t.Fatalf("the suspended viewer closed with %d", code) + } +} + +type failingTokens struct{} + +func (failingTokens) Issue(string, cloudwire.Device) (cloudwire.Token, error) { + return cloudwire.Token{}, errors.New("no tokens") +} + +func (failingTokens) Drop(string, string) error { return errors.New("the registry is down") } diff --git a/internal/registry/registry.go b/internal/registry/registry.go index a2f1291..056f8db 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -1,6 +1,7 @@ // Package registry is the service's only disk state: which homes exist, the -// box key each was claimed with, and the roster the box last signed. It holds -// no readings. Deleted rows are overwritten (secure_delete), so a purge does +// box key each was claimed with, the roster the box last signed, owners' +// suspensions, APNs registrations and the hashes of extension tokens. It +// holds no readings. Deleted rows are overwritten (secure_delete), so a purge does // not leave a home's roster in free pages. package registry @@ -79,6 +80,10 @@ CREATE TABLE IF NOT EXISTS suspensions ( _ = db.Close() return nil, fmt.Errorf("registry: schema: %w", err) } + if err := createTokenTable(db); err != nil { + _ = db.Close() + return nil, fmt.Errorf("registry: schema: %w", err) + } return &Registry{db: db}, nil } diff --git a/internal/registry/token.go b/internal/registry/token.go new file mode 100644 index 0000000..e0a7312 --- /dev/null +++ b/internal/registry/token.go @@ -0,0 +1,112 @@ +package registry + +import ( + "database/sql" + "errors" + "time" +) + +// Extension tokens. Each row binds the SHA-256 of one token to one roster +// entry (device id and epoch) of one home; the token itself is never stored. +// A device holds one token: a new one replaces the old. A row goes when its +// home goes (the foreign key cascades on purge and expiry), when a roster +// drops the device or moves its epoch, when an owner suspends the device, +// and when it expires. + +// TokenRow is one token's binding. +type TokenRow struct { + Home string + DeviceID string + Epoch uint64 + Expires time.Time +} + +func createTokenTable(db *sql.DB) error { + _, err := db.Exec(` +CREATE TABLE IF NOT EXISTS tokens ( + home TEXT NOT NULL REFERENCES homes(home) ON DELETE CASCADE, + device_id TEXT NOT NULL, + epoch INTEGER NOT NULL, + hash BLOB NOT NULL UNIQUE, + expires_ms INTEGER NOT NULL, + PRIMARY KEY (home, device_id) +);`) + return err +} + +// TokenIssue binds a token's hash to a roster entry, in place of the +// device's old token. +func (r *Registry) TokenIssue(row TokenRow, hash []byte) error { + _, err := r.db.Exec(`INSERT INTO tokens (home, device_id, epoch, hash, expires_ms) VALUES (?, ?, ?, ?, ?) + ON CONFLICT (home, device_id) DO UPDATE SET epoch = excluded.epoch, hash = excluded.hash, + expires_ms = excluded.expires_ms`, + row.Home, row.DeviceID, int64(row.Epoch), hash, row.Expires.UnixMilli()) + return err +} + +// TokenLookup returns the binding of a token's hash, if it has not expired. +func (r *Registry) TokenLookup(hash []byte, now time.Time) (TokenRow, bool, error) { + var row TokenRow + var epoch, ms int64 + err := r.db.QueryRow(`SELECT home, device_id, epoch, expires_ms FROM tokens WHERE hash = ? AND expires_ms > ?`, + hash, now.UnixMilli()).Scan(&row.Home, &row.DeviceID, &epoch, &ms) + if errors.Is(err, sql.ErrNoRows) { + return TokenRow{}, false, nil + } + if err != nil { + return TokenRow{}, false, err + } + row.Epoch, row.Expires = uint64(epoch), time.UnixMilli(ms) + return row, true, nil +} + +// TokenDrop forgets a device's token, as when an owner suspends it. +func (r *Registry) TokenDrop(home, deviceID string) error { + _, err := r.db.Exec(`DELETE FROM tokens WHERE home = ? AND device_id = ?`, home, deviceID) + return err +} + +// TokenDropHash forgets one token by its hash. +func (r *Registry) TokenDropHash(hash []byte) error { + _, err := r.db.Exec(`DELETE FROM tokens WHERE hash = ?`, hash) + return err +} + +// TokenPrune keeps only the tokens whose device keep allows at its epoch, as +// a new roster says. +func (r *Registry) TokenPrune(home string, keep func(deviceID string, epoch uint64) bool) error { + rows, err := r.db.Query(`SELECT device_id, epoch FROM tokens WHERE home = ?`, home) + if err != nil { + return err + } + type entry struct { + id string + epoch int64 + } + var gone []entry + for rows.Next() { + var e entry + if err := rows.Scan(&e.id, &e.epoch); err != nil { + _ = rows.Close() + return err + } + if !keep(e.id, uint64(e.epoch)) { + gone = append(gone, e) + } + } + _ = rows.Close() + // The epoch is part of the match, so a token the device got at its new + // epoch meanwhile stays. + for _, e := range gone { + if _, err := r.db.Exec(`DELETE FROM tokens WHERE home = ? AND device_id = ? AND epoch = ?`, home, e.id, e.epoch); err != nil { + return err + } + } + return nil +} + +// TokenExpire forgets the tokens that have expired. +func (r *Registry) TokenExpire(now time.Time) error { + _, err := r.db.Exec(`DELETE FROM tokens WHERE expires_ms <= ?`, now.UnixMilli()) + return err +} diff --git a/internal/registry/token_test.go b/internal/registry/token_test.go new file mode 100644 index 0000000..0508c21 --- /dev/null +++ b/internal/registry/token_test.go @@ -0,0 +1,71 @@ +package registry + +import ( + "testing" + "time" +) + +// A device holds one token; a new one replaces the old, and only an unexpired +// hash is found. +func TestADeviceHoldsOneToken(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + row := TokenRow{Home: "h1", DeviceID: "phone", Epoch: 2, Expires: now.Add(time.Hour)} + if err := r.TokenIssue(row, []byte("first")); err != nil { + t.Fatal(err) + } + if err := r.TokenIssue(row, []byte("second")); err != nil { + t.Fatal(err) + } + if _, ok, _ := r.TokenLookup([]byte("first"), now); ok { + t.Fatal("the replaced token still reads") + } + got, ok, err := r.TokenLookup([]byte("second"), now) + if err != nil || !ok || got.Home != "h1" || got.DeviceID != "phone" || got.Epoch != 2 { + t.Fatalf("lookup: %+v %v %v", got, ok, err) + } + if _, ok, _ := r.TokenLookup([]byte("second"), now.Add(time.Hour)); ok { + t.Fatal("an expired token was found") + } + if err := r.TokenExpire(now.Add(time.Hour)); err != nil { + t.Fatal(err) + } + if _, ok, _ := r.TokenLookup([]byte("second"), now); ok { + t.Fatal("upkeep left an expired token") + } +} + +// A roster that drops a device or moves its epoch drops its token, a +// suspension drops it, and a purge takes every token of the home. +func TestTokensGoWithTheirRosterEntry(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + issue := func(id string, epoch uint64) { + if err := r.TokenIssue(TokenRow{Home: "h1", DeviceID: id, Epoch: epoch, Expires: now.Add(time.Hour)}, []byte(id)); err != nil { + t.Fatal(err) + } + } + found := func(id string) bool { + _, ok, _ := r.TokenLookup([]byte(id), now) + return ok + } + issue("kept", 1) + issue("moved", 1) + issue("dropped", 1) + issue("suspended", 1) + keep := map[string]uint64{"kept": 1, "moved": 2, "suspended": 1} + if err := r.TokenPrune("h1", func(id string, epoch uint64) bool { e, ok := keep[id]; return ok && e == epoch }); err != nil { + t.Fatal(err) + } + if !found("kept") || found("moved") || found("dropped") { + t.Fatal("the prune kept the wrong tokens") + } + if err := r.TokenDrop("h1", "suspended"); err != nil || found("suspended") { + t.Fatal("a dropped token still reads") + } + if err := r.Purge("h1", nil, now); err != nil || found("kept") { + t.Fatal("a purge left a token") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 24a8230..61cf12f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -3,12 +3,12 @@ // // /v1/box a box's uplink (package uplink) // /v1/app a phone's Noise session to the service (package appsession) -// /v1/now an extension's token read: C5, not built yet +// /v1/now a widget's token read (package token) // /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 or a -// phone in its roster. +// 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. package server import ( @@ -22,6 +22,7 @@ import ( "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/token" "github.com/srcfl/ftw-cloud/internal/uplink" ) @@ -42,7 +43,10 @@ type Server struct { Uplink *uplink.Uplink // App is nil on a host without a Noise key; /v1/app then says so. App *appsession.Server - Log *slog.Logger + // 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 } // Handler is the whole HTTP surface. @@ -54,9 +58,11 @@ func (s *Server) Handler() http.Handler { } 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) + if s.Tokens != nil { + mux.Handle("/v1/now", s.Tokens) + } else { + mux.HandleFunc("/v1/now", notYet) + } mux.HandleFunc("/source", source) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store") @@ -130,7 +136,7 @@ func revision() string { // Upkeep applies the retention rules until ctx ends: readings go an hour // after the box does, homes without a roster go after 24 hours or 90 days, -// and old tombstones go. +// expired tokens go, and old tombstones go. func (s *Server) Upkeep(ctx context.Context, every time.Duration) { t := time.NewTicker(every) defer t.Stop() @@ -140,6 +146,9 @@ func (s *Server) Upkeep(ctx context.Context, every time.Duration) { return case <-t.C: s.Store.Sweep() + if err := s.Registry.TokenExpire(time.Now()); err != nil { + s.Log.Warn("token upkeep failed", "err", err) + } gone, err := s.Registry.Expire(time.Now()) if err != nil { s.Log.Warn("registry upkeep failed", "err", err) diff --git a/internal/server/server_test.go b/internal/server/server_test.go index a1c13f6..f51b6b0 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -16,6 +16,7 @@ import ( "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/token" "github.com/srcfl/ftw-cloud/internal/uplink" "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" @@ -33,7 +34,8 @@ func newServer(t *testing.T) (*Server, string) { log := slog.New(slog.NewTextHandler(io.Discard, nil)) 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} + s := &Server{Store: st, Registry: reg, Uplink: up, App: appsession.New(st, reg, up, noise, log), + Tokens: token.New(st, reg, log), Log: log} // A home with readings in it, held the way a box's uplink holds them. const home = "canaryhomecanaryhomecanary" diff --git a/internal/store/session.go b/internal/store/session.go index 7398408..8b8f657 100644 --- a/internal/store/session.go +++ b/internal/store/session.go @@ -49,7 +49,7 @@ func (x *Handle) Open() (Snapshot, *Session, error) { 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 { + if _, _, ok := x.s.access(x.id, x.h, byKey(x.device.Key)); !ok { return Snapshot{}, nil, ErrDenied } for sess := range x.h.sessions { diff --git a/internal/store/store.go b/internal/store/store.go index 1d46fff..8e42d49 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -2,9 +2,10 @@ // disk: a restart empties it, and boxes fill it again when they reconnect. // // The only way to a home is For, with a principal that proves whose home it -// is: the box, by the signed hello on its own connection, or a phone in the -// home's roster, by its Noise handshake. There is no listing, no lookup by -// anything else, and no principal for staff. One bug in a second access path +// is: the box, by the signed hello on its own connection; a phone in the +// home's roster, by its Noise handshake; or an extension token the registry +// bound to a roster entry. There is no listing, no lookup by anything else, +// and no principal for staff. One bug in a second access path // would expose every home at once, so there is no second access path. package store @@ -31,6 +32,9 @@ const ( Box Kind = iota + 1 // Device is a phone in the home's roster, proven by its Noise static key. Device + // Token is a widget or an extension, proven by a token the registry + // bound to one roster entry. It reads only Now. + Token ) // Principal is who is asking for a home. @@ -39,6 +43,10 @@ type Principal struct { Home string // DeviceKey is the phone's X25519 static key, for Device. DeviceKey []byte + // DeviceID and Epoch are the roster entry a token is bound to, for + // Token. + DeviceID string + Epoch uint64 } // ErrDenied is every refusal. It does not say whether the home exists. @@ -107,7 +115,8 @@ type Handle struct { // 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, and only while no owner has suspended it. +// its key, and only while no owner has suspended it. A token reaches it by +// the same rule, through the roster entry it is bound to at that epoch. func (s *Store) For(p Principal) (*Handle, error) { s.mu.Lock() h := s.homes[p.Home] @@ -125,24 +134,39 @@ func (s *Store) For(p Principal) (*Handle, error) { case Device: h.mu.Lock() defer h.mu.Unlock() - dev, scopes, ok := s.access(p.Home, h, p.DeviceKey) + dev, scopes, ok := s.access(p.Home, h, byKey(p.DeviceKey)) if !ok { return nil, ErrDenied } return &Handle{s: s, id: p.Home, h: h, kind: Device, device: dev, scopes: scopes}, nil + case Token: + h.mu.Lock() + defer h.mu.Unlock() + dev, scopes, ok := s.access(p.Home, h, func(d cloudwire.Device) bool { + return p.DeviceID != "" && d.ID == p.DeviceID && d.Epoch == p.Epoch + }) + if !ok { + return nil, ErrDenied + } + return &Handle{s: s, id: p.Home, h: h, kind: Token, device: dev, scopes: scopes}, nil } return nil, ErrDenied } +// byKey matches the roster entry with a device's Noise static key. +func byKey(key []byte) func(cloudwire.Device) bool { + return func(d cloudwire.Device) bool { return len(key) == 32 && bytes.Equal(d.Key, key) } +} + // 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) { +// lists it, and no suspension. The caller holds h.mu. +func (s *Store) access(id string, h *home, match func(cloudwire.Device) bool) (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)) { + if r == nil || 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) { + if !match(d) { continue } if s.suspended != nil && s.suspended(id, d.ID, d.Epoch) { @@ -242,7 +266,7 @@ func (x *Handle) Recheck() { func (x *Handle) recheckLocked() { for sess := range x.h.sessions { - dev, scopes, ok := x.s.access(x.id, x.h, sess.x.device.Key) + dev, scopes, ok := x.s.access(x.id, x.h, byKey(sess.x.device.Key)) switch { case !ok: sess.endLocked(cloudwire.CloseDenied) @@ -292,14 +316,19 @@ func (x *Handle) Snapshot() (Snapshot, error) { return x.snapshotLocked(), nil } -func (x *Handle) snapshotLocked() Snapshot { +func (x *Handle) snapshotLocked() Snapshot { return x.viewLocked(nil) } + +// viewLocked is the newest envelope of each class the device may read, of +// only those classes when only is not nil, and the newest live that carried +// sources, in rising seq. The caller holds the home's lock. +func (x *Handle) viewLocked(only []string) 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 } var held []Held for t, e := range x.h.latest { - if x.mayRead(t) { + if x.mayRead(t) && (only == nil || slices.Contains(only, t)) { held = append(held, e) } } @@ -313,6 +342,22 @@ func (x *Handle) snapshotLocked() Snapshot { return out } +// NowClasses are what a token reads: the newest live, prices and hb, and the +// newest live that carried sources. Never lp, plan, event, tiles or health. +var NowClasses = []string{cloudwire.MsgLive, cloudwire.MsgPrices, cloudwire.MsgHB} + +// Now is what a widget reads with its token: the newest envelope of each of +// NowClasses the device may read, and the newest live that carried sources, +// in rising seq. After an hour without the box, no readings at all. +func (x *Handle) Now() (Snapshot, error) { + if x.kind != Token { + return Snapshot{}, ErrDenied + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + return x.viewLocked(NowClasses), nil +} + func (x *Handle) mayRead(class string) bool { scope, forwarded := cloudwire.ClassScope[class] return forwarded && (scope == "" || slices.Contains(x.scopes, scope)) diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 162077a..95b5bcb 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -3,6 +3,7 @@ package store import ( "bytes" "errors" + "slices" "testing" "time" @@ -260,3 +261,87 @@ func TestABoxHandleCannotOpenASession(t *testing.T) { t.Fatal("the box's handle opened a session") } } + +// A token reaches its home through the roster entry it is bound to, at that +// epoch, and reads only Now: the newest live, prices and hb, and the newest +// live that carried sources. +func TestATokenReadsOnlyNow(t *testing.T) { + s, box, _ := filled(t) + _ = box.Put(cloudwire.MsgPrices, []byte("prices"), 8, false) + _ = box.Put(cloudwire.MsgLP, []byte("lp"), 9, false) + _ = box.Put(cloudwire.MsgHealth, []byte("health"), 10, false) + _ = box.Put(cloudwire.MsgTiles, []byte("tiles"), 11, false) + box.SetRoster(roster(2, protocol.CloudScopes...)) + x, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "owner"}) + if err != nil { + t.Fatal(err) + } + snap, err := x.Now() + if err != nil { + t.Fatal(err) + } + var got []string + for _, e := range snap.Envelopes { + got = append(got, string(e)) + } + if want := []string{"live-full", "live-delta", "hb", "prices"}; !slices.Equal(got, want) { + t.Fatalf("a token read %v, want %v", got, want) + } + if _, err := x.Snapshot(); !errors.Is(err, ErrDenied) { + t.Fatal("a token read the whole snapshot") + } + if _, _, err := x.Open(); !errors.Is(err, ErrDenied) { + t.Fatal("a token opened a session") + } + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + if _, err := dev.Now(); !errors.Is(err, ErrDenied) { + t.Fatal("a device key read Now as a token") + } +} + +// A token is held to the same rule as a device: a roster in force that lists +// its entry at its epoch, and no suspension. +func TestATokenIsAdmittedAsItsDeviceIs(t *testing.T) { + s, box, c := filled(t) + for name, p := range map[string]Principal{ + "no device": {Kind: Token, Home: "h1"}, + "another device": {Kind: Token, Home: "h1", DeviceID: "stranger"}, + "another epoch": {Kind: Token, Home: "h1", DeviceID: "owner", Epoch: 1}, + "another home": {Kind: Token, Home: "h2", DeviceID: "owner"}, + } { + if _, err := s.For(p); !errors.Is(err, ErrDenied) { + t.Errorf("%s: %v", name, err) + } + } + s.UseSuspensions(func(home, id string, epoch uint64) bool { return id == "viewer" }) + if _, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "viewer"}); !errors.Is(err, ErrDenied) { + t.Fatal("a suspended device's token was admitted") + } + c.t = c.t.Add(24*time.Hour + time.Second) + if _, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "owner"}); !errors.Is(err, ErrDenied) { + t.Fatal("a token read past an expired roster") + } + box.SetRoster(roster(2, protocol.ScopeLiveRead)) + if _, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "owner"}); err != nil { + t.Fatal("a renewed roster still refused the token") + } + box.Delete() + if _, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "owner"}); !errors.Is(err, ErrDenied) { + t.Fatal("a token reached a deleted home") + } +} + +// A token whose device may not read live gets the hb alone. +func TestATokenReadsWithinItsDevicesScope(t *testing.T) { + s, box, _ := filled(t) + _ = box.Put(cloudwire.MsgPrices, []byte("prices"), 8, false) + box.SetRoster(roster(2, protocol.ScopePlanRead)) + x, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "viewer"}) + if err != nil { + t.Fatal(err) + } + snap, _ := x.Now() + if len(snap.Envelopes) != 1 || string(snap.Envelopes[0]) != "hb" { + t.Fatalf("a token without live read %q", snap.Envelopes) + } +} diff --git a/internal/token/token.go b/internal/token/token.go new file mode 100644 index 0000000..a27a684 --- /dev/null +++ b/internal/token/token.go @@ -0,0 +1,170 @@ +// Package token gives each roster device one extension token and serves +// GET /v1/now with it. docs/protocol-cloud.md in srcfl/ftw, "Extension +// token", is the wire format. +// +// A token is 32 random bytes, base64url without padding. The registry keeps +// its SHA-256 and the roster entry it is bound to, never the token. A read +// reaches the home only through store.For with that roster entry, so a token +// stops working the moment the roster drops the device, moves its epoch or +// runs out, an owner suspends the device, or the grant ends. It reads the +// newest live, prices and hb, as the box signed them, and nothing else. It +// is not a device key and cannot command. +package token + +import ( + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "errors" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +const ( + // Life is how long a token reads. The app asks for a new one at each + // foreground and background refresh. + Life = 7 * 24 * time.Hour + // tokenBytes is the token's length before base64url. + tokenBytes = 32 +) + +// Now is the body of GET /v1/now: the box's envelopes as it signed them, in +// rising seq, and when the service last heard from the box. lastSeenMs is +// the service's own word and is never shown as a reading. +type Now struct { + Envelopes []cbor.RawMessage `cbor:"envelopes"` + LastSeenMs int64 `cbor:"lastSeenMs"` +} + +// Service issues tokens and answers /v1/now. +type Service struct { + store *store.Store + reg *registry.Registry + log *slog.Logger + now func() time.Time +} + +// New builds the service. +func New(st *store.Store, reg *registry.Registry, log *slog.Logger) *Service { + if log == nil { + log = slog.Default() + } + return &Service{store: st, reg: reg, log: log, now: time.Now} +} + +// Issue makes a new token for a roster entry, in place of the device's old +// one. dev comes from store.For, never from the device's message. +// +// The device's access is checked again once the token is stored: a +// suspension or a roster that landed since store.For admitted the device +// either finds the row and drops it, or is seen here and the row goes. +func (s *Service) Issue(home string, dev cloudwire.Device) (cloudwire.Token, error) { + raw := make([]byte, tokenBytes) + if _, err := rand.Read(raw); err != nil { + return cloudwire.Token{}, err + } + hash := sha256.Sum256(raw) + row := registry.TokenRow{Home: home, DeviceID: dev.ID, Epoch: dev.Epoch, Expires: s.now().Add(Life)} + if err := s.reg.TokenIssue(row, hash[:]); err != nil { + return cloudwire.Token{}, err + } + if _, err := s.store.For(store.Principal{Kind: store.Token, Home: home, DeviceID: dev.ID, Epoch: dev.Epoch}); err != nil { + if err := s.reg.TokenDropHash(hash[:]); err != nil { + return cloudwire.Token{}, err + } + return cloudwire.Token{}, store.ErrDenied + } + return cloudwire.Token{Token: base64.RawURLEncoding.EncodeToString(raw), ExpiresInS: int64(Life / time.Second)}, nil +} + +// Drop forgets a device's token, as when an owner suspends it. +func (s *Service) Drop(home, deviceID string) error { + return s.reg.TokenDrop(home, deviceID) +} + +// Roster drops the tokens of devices the box's roster no longer lists at +// their epoch. It runs on the box's goroutine and never blocks for long. +func (s *Service) Roster(home string, r *cloudwire.Roster) { + if r == nil { + return + } + epochs := make(map[string]uint64, len(r.Devices)) + for _, d := range r.Devices { + epochs[d.ID] = d.Epoch + } + if err := s.reg.TokenPrune(home, func(id string, epoch uint64) bool { + e, ok := epochs[id] + return ok && e == epoch + }); err != nil { + s.log.Warn("could not drop the tokens a roster dropped", "err", err) + } +} + +// ServeHTTP is GET /v1/now. Every refusal is 401 with an empty body and says +// nothing about which check failed. +func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if r.TLS == nil || r.TLS.Version != tls.VersionTLS13 { + http.Error(w, "TLS 1.3 only", http.StatusUpgradeRequired) + return + } + if r.Method != http.MethodGet { + w.Header().Set("Allow", http.MethodGet) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + body, err := s.read(r.Header.Get("Authorization")) + switch { + case errors.Is(err, store.ErrDenied): + w.Header().Set("WWW-Authenticate", "Bearer") + w.WriteHeader(http.StatusUnauthorized) + return + case err != nil: + s.log.Warn("a token read failed", "err", err) + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/cbor") + _, _ = w.Write(body) +} + +// read checks a bearer token and returns the body it may read. +func (s *Service) read(auth string) ([]byte, error) { + scheme, value, ok := strings.Cut(auth, " ") + if !ok || !strings.EqualFold(scheme, "Bearer") { + return nil, store.ErrDenied + } + raw, err := base64.RawURLEncoding.DecodeString(strings.TrimSpace(value)) + if err != nil || len(raw) != tokenBytes { + return nil, store.ErrDenied + } + hash := sha256.Sum256(raw) + row, ok, err := s.reg.TokenLookup(hash[:], s.now()) + if err != nil { + return nil, err + } + if !ok { + return nil, store.ErrDenied + } + x, err := s.store.For(store.Principal{Kind: store.Token, Home: row.Home, DeviceID: row.DeviceID, Epoch: row.Epoch}) + if err != nil { + return nil, store.ErrDenied + } + snap, err := x.Now() + if err != nil { + return nil, store.ErrDenied + } + out := Now{Envelopes: make([]cbor.RawMessage, len(snap.Envelopes)), LastSeenMs: snap.LastSeenMs} + for i, e := range snap.Envelopes { + out.Envelopes[i] = e + } + return cloudwire.Marshal(out) +} diff --git a/internal/token/token_test.go b/internal/token/token_test.go new file mode 100644 index 0000000..1b4bb12 --- /dev/null +++ b/internal/token/token_test.go @@ -0,0 +1,372 @@ +package token + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "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/cloudwire" +) + +type clock struct{ t time.Time } + +func (c *clock) now() time.Time { return c.t } + +// A home with a signed copy of every class, an owner and a viewer in its +// roster, and /v1/now behind TLS 1.3. +type harness struct { + t *testing.T + clock *clock + store *store.Store + reg *registry.Registry + svc *Service + srv *httptest.Server + boxKey *ecdsa.PrivateKey + home string + box *store.Handle + seq uint64 + held map[string][]byte + rsq uint64 +} + +func newHarness(t *testing.T) *harness { + t.Helper() + c := &clock{t: time.Now()} + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(c.now) + st.UseSuspensions(reg.Suspended) + svc := New(st, reg, slog.New(slog.NewTextHandler(io.Discard, nil))) + svc.now = c.now + srv := httptest.NewUnstartedServer(svc) + 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) + if err := reg.Claim(home, pk, c.t); err != nil { + t.Fatal(err) + } + box, _ := st.For(store.Principal{Kind: store.Box, Home: home}) + box.Connected() + h := &harness{t: t, clock: c, store: st, reg: reg, svc: svc, srv: srv, boxKey: boxKey, home: home, box: box, + held: map[string][]byte{}} + h.roster(1, protocol.CloudScopes...) + return h +} + +func (h *harness) devices(ownerEpoch uint64) []cloudwire.Device { + return []cloudwire.Device{ + {ID: "owner", Key: bytes.Repeat([]byte{1}, 32), Role: "owner", Scopes: protocol.CloudScopes, Epoch: ownerEpoch}, + {ID: "viewer", Key: bytes.Repeat([]byte{2}, 32), Role: "viewer", Scopes: []string{protocol.ScopeLiveRead}}, + } +} + +func (h *harness) roster(ownerEpoch uint64, grant ...string) *cloudwire.Roster { + return h.setRoster(h.devices(ownerEpoch), grant...) +} + +// setRoster holds a roster the way the uplink does, and lets the service drop +// the tokens it no longer allows. +func (h *harness) setRoster(devices []cloudwire.Device, grant ...string) *cloudwire.Roster { + h.rsq++ + r := &cloudwire.Roster{RosterSeq: h.rsq, 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") + } + h.svc.Roster(h.home, r) + return r +} + +// put signs a body of class t as the box and holds it; name keys it for the +// test. +func (h *harness) put(name, t string, body any, table bool) { + h.t.Helper() + h.seq++ + hdr := cloudwire.Header{Seq: h.seq, Boot: bytes.Repeat([]byte{0xb0}, cloudwire.BootBytes), UptimeMs: int64(h.seq * 100)} + env, err := cloudwire.Seal(h.boxKey, h.home, t, struct { + cloudwire.Header + Body any `cbor:"x"` + }{hdr, body}) + if err != nil { + h.t.Fatal(err) + } + raw, _ := env.Encode() + if err := h.box.Put(t, raw, h.seq, table); err != nil { + h.t.Fatal(err) + } + h.held[name] = raw +} + +// fill holds one envelope of every class, in this order. +func (h *harness) fill() { + h.put("live full", cloudwire.MsgLive, "full", true) + h.put("prices", cloudwire.MsgPrices, "p", false) + h.put("lp", cloudwire.MsgLP, "lp", false) + h.put("plan", cloudwire.MsgPlan, "plan", false) + h.put("event", cloudwire.MsgEvent, "e", false) + h.put("tiles", cloudwire.MsgTiles, "t", false) + h.put("health", cloudwire.MsgHealth, "h", false) + h.put("live delta", cloudwire.MsgLive, "delta", false) + h.put("hb", cloudwire.MsgHB, "hb", false) +} + +func (h *harness) issue(id string, epoch uint64) string { + h.t.Helper() + tok, err := h.svc.Issue(h.home, cloudwire.Device{ID: id, Epoch: epoch}) + if err != nil { + h.t.Fatal(err) + } + if tok.ExpiresInS != 7*24*3600 || len(tok.Token) != 43 { + h.t.Fatalf("token %q expires in %d s", tok.Token, tok.ExpiresInS) + } + return tok.Token +} + +// get reads /v1/now with an Authorization header as given. +func (h *harness) get(auth string) (int, []byte) { + h.t.Helper() + req, _ := http.NewRequest(http.MethodGet, h.srv.URL+"/v1/now", nil) + if auth != "" { + req.Header.Set("Authorization", auth) + } + resp, err := h.srv.Client().Do(req) + if err != nil { + h.t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.Header.Get("Cache-Control") != "no-store" { + h.t.Fatal("a token read may be cached") + } + return resp.StatusCode, body +} + +func (h *harness) now(token string) Now { + h.t.Helper() + code, body := h.get("Bearer " + token) + if code != http.StatusOK { + h.t.Fatalf("token read: %d", code) + } + var n Now + if err := cloudwire.Unmarshal(body, &n); err != nil { + h.t.Fatal(err) + } + return n +} + +// The protocol's promise: a token reads live, prices and hb as the box +// signed them, in rising seq, and never lp, plan, event, tiles or health. +func TestATokenReadsLivePricesAndHBAndNothingElse(t *testing.T) { + h := newHarness(t) + h.fill() + n := h.now(h.issue("owner", 1)) + want := []string{"live full", "prices", "live delta", "hb"} + if len(n.Envelopes) != len(want) { + t.Fatalf("a token read %d envelopes, want %v", len(n.Envelopes), want) + } + for i, name := range want { + if !bytes.Equal(n.Envelopes[i], h.held[name]) { + t.Fatalf("envelope %d is not the box's %s", i, name) + } + env, err := cloudwire.DecodeEnvelope(n.Envelopes[i]) + if err != nil { + t.Fatal(err) + } + if err := cloudwire.Verify(&h.boxKey.PublicKey, h.home, env); err != nil { + t.Fatalf("%s does not verify with the box key: %v", name, err) + } + } + if n.LastSeenMs != h.clock.t.UnixMilli() { + t.Fatalf("lastSeenMs %d", n.LastSeenMs) + } +} + +// Every refusal is the same 401 with an empty body. +func TestRefusalsSayNothing(t *testing.T) { + h := newHarness(t) + h.fill() + good := h.issue("owner", 1) + stranger := "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + for name, auth := range map[string]string{ + "no header": "", + "another scheme": "Basic " + good, + "not base64url": "Bearer !!!", + "too short": "Bearer " + good[:20], + "a token unknown": "Bearer " + stranger, + } { + code, body := h.get(auth) + if code != http.StatusUnauthorized || len(body) != 0 { + t.Errorf("%s: %d %q", name, code, body) + } + } + if code, _ := h.get("bearer " + good); code != http.StatusOK { + t.Fatalf("the scheme is case-insensitive: %d", code) + } +} + +// A token stops reading when its roster entry goes, moves its epoch, is +// suspended, when the roster runs out, when the token expires, when a new +// token replaces it, and when the home is purged. +func TestATokenDiesWithItsAccess(t *testing.T) { + refused := func(t *testing.T, h *harness, tok string) { + t.Helper() + if code, body := h.get("Bearer " + tok); code != http.StatusUnauthorized || len(body) != 0 { + t.Fatalf("still reads: %d", code) + } + } + t.Run("roster drops the device", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("viewer", 0) + h.setRoster(h.devices(1)[:1], protocol.CloudScopes...) + refused(t, h, tok) + if _, ok, _ := h.reg.TokenLookup(hashOf(t, tok), h.clock.t); ok { + t.Fatal("the roster left the token's row") + } + }) + t.Run("new epoch", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + h.roster(2, protocol.CloudScopes...) + refused(t, h, tok) + }) + t.Run("suspended", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + _ = h.reg.Suspend(h.home, "owner", 1, h.clock.t) + refused(t, h, tok) + }) + t.Run("roster runs out", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + h.clock.t = h.clock.t.Add(24*time.Hour + time.Second) + refused(t, h, tok) + }) + t.Run("expires", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + h.clock.t = h.clock.t.Add(Life - time.Second) + h.roster(1, protocol.CloudScopes...) + if code, _ := h.get("Bearer " + tok); code != http.StatusOK { + t.Fatalf("a token a second from expiry: %d", code) + } + h.clock.t = h.clock.t.Add(time.Second) + refused(t, h, tok) + }) + t.Run("replaced", func(t *testing.T) { + h := newHarness(t) + old := h.issue("owner", 1) + fresh := h.issue("owner", 1) + refused(t, h, old) + if code, _ := h.get("Bearer " + fresh); code != http.StatusOK { + t.Fatalf("the new token: %d", code) + } + }) + t.Run("purged", func(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + _ = h.reg.Purge(h.home, nil, h.clock.t) + h.box.Delete() + refused(t, h, tok) + }) +} + +// Readings go an hour after the box does; the token then reads only when +// the box was last seen. +func TestATokenReadsNoReadingsAnHourAfterTheBox(t *testing.T) { + h := newHarness(t) + h.fill() + tok := h.issue("owner", 1) + gen := h.box.Connected() + h.box.Disconnected(gen) + h.clock.t = h.clock.t.Add(store.ReadingsKeptFor + time.Second) + n := h.now(tok) + if len(n.Envelopes) != 0 || n.LastSeenMs == 0 { + t.Fatalf("after an hour: %d envelopes, last seen %d", len(n.Envelopes), n.LastSeenMs) + } +} + +func TestOnlyGETOverTLS13(t *testing.T) { + h := newHarness(t) + tok := h.issue("owner", 1) + req, _ := http.NewRequest(http.MethodPost, h.srv.URL+"/v1/now", nil) + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := h.srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Fatalf("POST: %d", resp.StatusCode) + } + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/v1/now", nil) + r.Header.Set("Authorization", "Bearer "+tok) + h.svc.ServeHTTP(w, r) + if w.Code != http.StatusUpgradeRequired || bytes.Contains(w.Body.Bytes(), []byte("envelopes")) { + t.Fatalf("plain HTTP: %d", w.Code) + } +} + +func hashOf(t *testing.T, tok string) []byte { + t.Helper() + raw, err := base64.RawURLEncoding.DecodeString(tok) + if err != nil { + t.Fatal(err) + } + sum := sha256.Sum256(raw) + return sum[:] +} + +// A token asked for by a device that a suspension or roster has just shut +// out is not kept: the access is checked again once the row is stored. +func TestATokenIsNotIssuedPastItsDevicesAccess(t *testing.T) { + h := newHarness(t) + _ = h.reg.Suspend(h.home, "owner", 1, h.clock.t) + if _, err := h.svc.Issue(h.home, cloudwire.Device{ID: "owner", Epoch: 1}); err != store.ErrDenied { + t.Fatalf("a suspended device got a token: %v", err) + } + if _, err := h.svc.Issue(h.home, cloudwire.Device{ID: "owner", Epoch: 2}); err != store.ErrDenied { + t.Fatalf("a device at an epoch the roster does not list got a token: %v", err) + } + var n int + if err := h.reg.TokenPrune(h.home, func(string, uint64) bool { n++; return true }); err != nil || n != 0 { + t.Fatalf("refused tokens left %d rows", n) + } +} + +// A roster that moves a device's epoch drops the token of the old epoch and +// keeps one the device already got at the new epoch. +func TestAPruneKeepsATokenOfTheNewEpoch(t *testing.T) { + h := newHarness(t) + old := h.issue("owner", 1) + r := h.roster(2, protocol.CloudScopes...) + fresh := h.issue("owner", 2) + h.svc.Roster(h.home, r) + if code, _ := h.get("Bearer " + old); code != http.StatusUnauthorized { + t.Fatalf("the old epoch's token read %d", code) + } + if code, _ := h.get("Bearer " + fresh); code != http.StatusOK { + t.Fatalf("the new epoch's token read %d", code) + } +}