diff --git a/docs/e2ee.md b/docs/e2ee.md index e68c176..ca44781 100644 --- a/docs/e2ee.md +++ b/docs/e2ee.md @@ -27,6 +27,20 @@ servers' users too. | `keyserver_backup_versions`, `keyserver_backup_keys` | server-side key backup | | `keyserver_sequences` | the device list stream's counter | +Migration `0009_remote_keys.sql` adds the cache of other servers' users' +keys (see [The remote key cache](#the-remote-key-cache)): + +| Table | What | +|---|---| +| `keyserver_remote_users` | one row per cached user: the `stream_id` the copy is at, and whether it is `stale` | +| `keyserver_remote_device_keys` | their devices' keys and display names, as their server sent them | +| `keyserver_remote_cross_signing_keys` | their master and self-signing keys | + +They are tables of their own rather than rows in the local ones: every +write to a local user's keys moves the device list stream and goes out over +federation, and keeping the copies apart means no path that announces a +local change can announce a copy. + Keys are stored in canonical JSON. That is what lets "the same key uploaded again" be a byte comparison: a client that re-sends a one-time key it already uploaded, encoded differently, is not told it conflicts. @@ -40,7 +54,7 @@ exactly what its owner uploaded. | Endpoint | Notes | |---|---| | `POST /keys/upload` | device keys must be for the caller's own device; one-time and fallback keys (`fallback_keys`, or MSC2732's unstable name); answers the one-time key counts, `signed_curve25519` always included | -| `POST /keys/query` | local users from the store, remote users asked of their servers in parallel; `failures` for a server that did not answer | +| `POST /keys/query` | local users from the store; remote users who share a room with somebody here from the cache, fetched on a miss; anybody else asked of their server; all servers in parallel, `failures` for one that did not answer | | `POST /keys/claim` | a one-time key if there is one, the fallback key otherwise; remote users claimed from their servers | | `GET /keys/changes` | whose device lists changed between two sync tokens | | `POST /keys/device_signing/upload` | cross-signing keys; asks for the password, except where the spec says not to | @@ -147,23 +161,71 @@ uploads one. | EDU | Out | In | |---|---|---| -| `m.device_list_update` | one per device change, to every server sharing a room with the user, with the keys or `deleted: true`, a `stream_id` and the `prev_id` before it | moves the remote user's position in the device list stream | -| `m.signing_key_update` | when a user's master or self-signing key changes, or the master key gains a device's signature | likewise | +| `m.device_list_update` | one per device change, to every server sharing a room with the user, with the keys or `deleted: true`, a `stream_id` and the `prev_id` before it | applied to the cached copy when its `prev_id` includes the copy's `stream_id`, otherwise the copy is marked stale and fetched again; moves the remote user's position in the device list stream | +| `m.signing_key_update` | when a user's master or self-signing key changes, or the master key gains a device's signature | replaces the cached master and self-signing keys; moves the position likewise | Each device of a multi-device deletion gets a stream ID of its own: remote servers treat a stream ID they have seen as a duplicate, so three deletions sharing one would reach them as one deletion and two repeats. -Warp does **not cache other servers' keys**. A query about a remote user is -answered by asking their server, so an incoming update has nothing to -invalidate; what it does is tell local clients, who do cache, by moving the -user in the device list stream. That only happens when a local user shares -a room with them: an update about a stranger concerns nobody here, and -recording it would let any server grow the table with users of its own -invention. +An incoming update about a remote user always tells local clients, who +keep their own copies, by moving the user in the device list stream — and +only does anything when a local user shares a room with them: an update +about a stranger concerns nobody here, and recording it would let any +server grow the tables with users of its own invention. A repeat of an +update already applied changes nothing, not even the stream. Device display names are not sent over federation, in answers or in EDUs. -Local users see them in `unsigned.device_display_name`. +Local users see them in `unsigned.device_display_name`: their own +devices', and those of remote devices whose server did send one. + +### The remote key cache + +A client asks for the keys of everybody in a room before it sends an +encrypted message, and again whenever a sync says one of them moved. +Without a cache each of those is a federation round trip in front of the +message. So the keys of remote users who **share a room with a local +user** are kept (`internal/keyserver/remotekeys.go`): + +- **On a miss**, `/keys/query` fetches the user's whole device list with + `GET /user/devices/{userId}` (which also gives the `stream_id` the list + is at), caches it and serves it. A device whose keys name another user + or device is left out, and so is a cross-signing key that is not that + user's key for its purpose. The next query does not leave the server. +- **An `m.device_list_update`** whose `prev_id` includes the cached + `stream_id` is applied: the device's keys replaced, or deleted. One + whose `stream_id` is at or before the copy's is a repeat and ignored. + Anything else (a gap, an update with no keys for a device it says + changed, keys that are not that device's) marks the copy **stale**, and + it is fetched again in the background, not inside the transaction that + brought the update, which would have the sending server wait on a + request back to itself. +- **A stale copy is never served.** A query that finds one does what it + does on a miss and asks synchronously: a device list known to be missing + a change would have the sender skip the device just added, and the + message would be undecryptable on it. If the user's server is down, the + query reports it in `failures`. +- **An `m.signing_key_update`** replaces the cached master and + self-signing keys with the ones it carries; one it leaves out is + dropped. +- **A copy goes** once no local user shares a room with the user. The + keyserver listens on the roomserver's emit stage: a remote user's leave + or ban has their copy checked, a local user's has every copy checked, in + the background. A query or update that finds a cached user has become a + stranger drops the copy on the spot. Strangers are never cached; a query + about one is passed to their server as before. + +The cache and the updates race in one place: a snapshot fetched just +before an update arrives. An update about a user with nothing cached +leaves a stale marker at its `stream_id`, an update to a stale copy raises +its `stream_id`, and a snapshot older than the `stream_id` held is not +stored. So a snapshot that predates an update is never cached as current. + +Local moves that Warp does not send as an `m.device_list_update` (a device +rename, a cross-signing change, a signature) still move the user's +`stream_id`, so a Warp server caching another Warp server's user sees the +next update as a gap and fetches again. That costs one request and is +always correct. With federation off, remote users in a query or claim are reported in `failures` straight away; nothing is sent. @@ -187,8 +249,6 @@ not exist, `404`. A version's algorithm cannot be changed. | Thing | Why | |---|---| -| A cache of remote users' keys | Every query asks their server. Correct, and slower for a client in rooms with many remote users; a cache invalidated by the EDUs above is the obvious next step. | -| Resync on a gap in a remote server's `prev_id` | Follows from the above: with nothing cached, a missed update costs nothing. | | Dehydrated devices (MSC3814) | — | | Claiming several keys per device (MSC3983), appservice key claims and queries (MSC3983/MSC3984) | application services are Phase 9 | | Purging a deleted device's to-device queue | the messages stay queued; a new device later logged in under the same ID is handed them, encrypted to the old device's keys and so unreadable to it | @@ -199,6 +259,8 @@ not exist, `404`. A version's algorithm cannot be changed. | Where | What | |---|---| | `internal/storage/keys_test.go` | the key store against every driver: the device list stream, claims, fallback keys, signatures, the backup replacement rule | +| `internal/storage/remotekeys_test.go` | the remote key cache against every driver: snapshots, updates that follow on, repeats, gaps, stale markers, older snapshots refused, dropping | | `internal/keyserver/keyserver_test.go` | the checks, with real ed25519 signatures: uploads, conflicts, cross-signing, the three signature kinds and their visibility, deletions, remote updates, backups | -| `internal/setup/phase7_test.go` | through the whole HTTP stack: upload, sync, query, claim, `keys/changes`, cross-signing with and without UIA, backup, device deletion, leaving a room; and two federated servers exchanging keys and device list updates | +| `internal/keyserver/remotekeys_test.go` | the cache: a second query not leaving the server, updates applied, a gap fetched again in the background, a stale copy not served, signing key updates, strangers dropped | +| `internal/setup/phase7_test.go` | through the whole HTTP stack: upload, sync, query, claim, `keys/changes`, cross-signing with and without UIA, backup, device deletion, leaving a room; and two federated servers exchanging keys and device list updates, with a second query and an update served from B's cache without a request to A | | Complement | `TestUploadKey*`, `TestKeyClaimOrdering`, `TestKeyChanges*`, `TestDeviceListUpdates*`, `TestE2EKeyBackup*`, `TestFallbackKeys*` and `TestCrossSigning*` are in the target sets | diff --git a/server/internal/federationapi/keys.go b/server/internal/federationapi/keys.go index 2865e16..3749bd4 100644 --- a/server/internal/federationapi/keys.go +++ b/server/internal/federationapi/keys.go @@ -20,7 +20,8 @@ type KeyService interface { QueryLocalKeys(ctx context.Context, req federation.UserKeysQueryRequest) (federation.UserKeysQueryResponse, error) ClaimLocalKeys(ctx context.Context, want map[string]map[string]string) (map[string]map[string]map[string]json.RawMessage, error) UserDevices(ctx context.Context, userID string) (federation.UserDevicesResponse, error) - RemoteKeysChanged(ctx context.Context, userID string) error + RemoteDeviceListUpdate(ctx context.Context, update federation.DeviceListUpdateContent) error + RemoteSigningKeyUpdate(ctx context.Context, update federation.SigningKeyUpdateContent) error } // registerKeyRoutes adds the key endpoints, if there is a keyserver. @@ -77,20 +78,33 @@ func (s *Server) handleUserDevices(w http.ResponseWriter, r *http.Request, _ str return nil } -// receiveKeyEDU takes in an m.device_list_update or m.signing_key_update. -// Either says only that a user of the sending server changed their keys; -// the keyserver passes that on to the local clients that care. +// receiveKeyEDU takes in an m.device_list_update or m.signing_key_update: +// a user of the sending server changed their keys. The keyserver brings +// its cached copy of them up to date and tells the local clients that +// care. An update about a user of another server is dropped — a server +// speaks for its own users only. func (s *Server) receiveKeyEDU(ctx context.Context, origin string, edu federation.EDU) { if s.keyService == nil { return } - var c struct { - UserID string `json:"user_id"` - } - if json.Unmarshal(edu.Content, &c) != nil || !sameServer(c.UserID, origin) { + var err error + switch edu.Type { + case federation.EDUDeviceListUp: + var c federation.DeviceListUpdateContent + if json.Unmarshal(edu.Content, &c) != nil || !sameServer(c.UserID, origin) { + return + } + err = s.keyService.RemoteDeviceListUpdate(ctx, c) + case federation.EDUSigningKeys: + var c federation.SigningKeyUpdateContent + if json.Unmarshal(edu.Content, &c) != nil || !sameServer(c.UserID, origin) { + return + } + err = s.keyService.RemoteSigningKeyUpdate(ctx, c) + default: return } - if err := s.keyService.RemoteKeysChanged(ctx, c.UserID); err != nil { + if err != nil { s.log.Debug("ignored a key update EDU", "origin", origin, "type", edu.Type, "error", err) } } diff --git a/server/internal/keyserver/devicelists.go b/server/internal/keyserver/devicelists.go index 580295f..68533f5 100644 --- a/server/internal/keyserver/devicelists.go +++ b/server/internal/keyserver/devicelists.go @@ -131,32 +131,3 @@ func (k *API) announceSigningKeys(ctx context.Context, userID string) { } k.outbound.QueueEDU(ctx, servers, federation.EDU{Type: federation.EDUSigningKeys, Content: encoded}) } - -// RemoteKeysChanged takes in an m.device_list_update or -// m.signing_key_update from another server: one of its users' keys -// changed. -// -// Warp does not cache other servers' keys — a query for them is answered -// by asking their server — so there is nothing to invalidate. What there -// is to do is tell local clients, which do cache them, by moving the -// user's position in the device list stream. That only happens when a -// local user shares a room with them: an update about a stranger concerns -// nobody here, and recording it would let any server grow the table with -// users of its own invention. -func (k *API) RemoteKeysChanged(ctx context.Context, userID string) error { - if k.isLocal(userID) || k.rooms == nil { - return nil - } - users, err := k.rooms.UsersSharingRooms(ctx, userID) - if err != nil { - return err - } - if !slices.ContainsFunc(users, k.isLocal) { - return nil - } - if _, err := k.store.MarkDeviceListsChanged(ctx, []string{userID}); err != nil { - return err - } - k.changed(ctx, userID) - return nil -} diff --git a/server/internal/keyserver/export_test.go b/server/internal/keyserver/export_test.go new file mode 100644 index 0000000..71a050c --- /dev/null +++ b/server/internal/keyserver/export_test.go @@ -0,0 +1,5 @@ +package keyserver + +// WaitIdle waits for the background work running now — fetching stale +// copies of remote keys again, dropping strangers' — to finish. +func (k *API) WaitIdle() { k.background.wait() } diff --git a/server/internal/keyserver/keyserver.go b/server/internal/keyserver/keyserver.go index 0c0afa8..1e8f78e 100644 --- a/server/internal/keyserver/keyserver.go +++ b/server/internal/keyserver/keyserver.go @@ -97,6 +97,7 @@ type Devices interface { type Remote interface { QueryUserKeys(ctx context.Context, destination string, req federation.UserKeysQueryRequest) (federation.UserKeysQueryResponse, error) ClaimUserKeys(ctx context.Context, destination string, req federation.UserKeysClaimRequest) (federation.UserKeysClaimResponse, error) + UserDevices(ctx context.Context, destination, userID string) (federation.UserDevicesResponse, error) } // Outbound queues EDUs to other servers. *federationapi.Queue satisfies it. @@ -115,6 +116,13 @@ type API struct { remote Remote outbound Outbound log *slog.Logger + + // The background work of the remote key cache: fetching stale copies + // again, and dropping the copies of users who no longer share a room + // with anybody here. + background *background + resyncs *workQueue + sweeps *workQueue } // Options configure the component. @@ -159,7 +167,7 @@ func New(opts Options) (*API, error) { if opts.Logger == nil { opts.Logger = slog.Default() } - return &API{ + k := &API{ serverName: opts.ServerName, store: opts.Store, rooms: opts.Rooms, @@ -168,7 +176,11 @@ func New(opts Options) (*API, error) { remote: opts.Remote, outbound: opts.Outbound, log: opts.Logger, - }, nil + background: newBackground(), + } + k.resyncs = newWorkQueue(k, k.resync) + k.sweeps = newWorkQueue(k, k.sweep) + return k, nil } // SetOutbound connects the federation queue, which is built after the @@ -181,6 +193,15 @@ func (k *API) isLocal(userID string) bool { return err == nil && server == k.serverName } +// serverOf returns the server of a user, or "" for a malformed user ID. +func serverOf(userID string) string { + server, err := id.ServerNameOf(userID) + if err != nil { + return "" + } + return server +} + // localpartOf returns the localpart of a local user. func (k *API) localpartOf(userID string) (string, bool) { localpart, server, err := id.SplitUserID(userID) diff --git a/server/internal/keyserver/keyserver_test.go b/server/internal/keyserver/keyserver_test.go index d89a2dd..222a51c 100644 --- a/server/internal/keyserver/keyserver_test.go +++ b/server/internal/keyserver/keyserver_test.go @@ -24,10 +24,22 @@ const ( // --- fakes -------------------------------------------------------------------- -type fakeRooms struct{ sharing map[string][]string } +type fakeRooms struct { + mu sync.Mutex + sharing map[string][]string +} -func (f fakeRooms) UsersSharingRooms(_ context.Context, userID string) ([]string, error) { - return f.sharing[userID], nil +func (f *fakeRooms) UsersSharingRooms(_ context.Context, userID string) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.sharing[userID]), nil +} + +// set replaces who shares a room with userID. +func (f *fakeRooms) set(userID string, users ...string) { + f.mu.Lock() + defer f.mu.Unlock() + f.sharing[userID] = users } type fakeNotifier struct { @@ -86,6 +98,7 @@ func (o *fakeOutbound) ofType(t string) []sentEDU { type harness struct { keys *keyserver.API store storage.KeyStore + rooms *fakeRooms notifier *fakeNotifier outbound *fakeOutbound } @@ -106,15 +119,16 @@ func newHarness(t *testing.T, remote keyserver.Remote) harness { } h := harness{store: db.Keys(), notifier: &fakeNotifier{}, outbound: &fakeOutbound{}} + h.rooms = &fakeRooms{sharing: map[string][]string{ + alice: {alice, bob, carol}, + bob: {alice, bob}, + carol: {alice, carol}, + }} h.keys, err = keyserver.New(keyserver.Options{ ServerName: "warp.example", Store: db.Keys(), - Rooms: fakeRooms{sharing: map[string][]string{ - alice: {alice, bob, carol}, - bob: {alice, bob}, - carol: {alice, carol}, - }}, - Notifier: h.notifier, + Rooms: h.rooms, + Notifier: h.notifier, Devices: fakeDevices{byLocalpart: map[string][]storage.Device{ "alice": {{Localpart: "alice", ID: "D1", DisplayName: "Alice's phone"}, {Localpart: "alice", ID: "D2"}}, "bob": {{Localpart: "bob", ID: "B1"}}, @@ -125,6 +139,7 @@ func newHarness(t *testing.T, remote keyserver.Remote) harness { if err != nil { t.Fatalf("keyserver.New: %v", err) } + t.Cleanup(h.keys.Close) return h } @@ -351,7 +366,7 @@ type fakeRemote struct{} func (fakeRemote) QueryUserKeys(_ context.Context, _ string, req federation.UserKeysQueryRequest) (federation.UserKeysQueryResponse, error) { return federation.UserKeysQueryResponse{ DeviceKeys: map[string]map[string]json.RawMessage{ - carol: {"C1": json.RawMessage(`{"device_id":"C1"}`)}, + dave: {"V1": json.RawMessage(`{"device_id":"V1"}`)}, "@mallory:other.test": {"M1": json.RawMessage(`{}`)}, }, }, nil @@ -361,13 +376,20 @@ func (fakeRemote) ClaimUserKeys(context.Context, string, federation.UserKeysClai return federation.UserKeysClaimResponse{}, errors.New("unreachable") } +func (fakeRemote) UserDevices(context.Context, string, string) (federation.UserDevicesResponse, error) { + return federation.UserDevicesResponse{}, errors.New("unreachable") +} + +// dave is a remote user nobody here shares a room with. +const dave = "@dave:remote.example" + func TestQueryAsksOtherServers(t *testing.T) { h := newHarness(t, fakeRemote{}) - resp, err := h.keys.QueryKeys(t.Context(), alice, keyserver.QueryRequest{DeviceKeys: map[string][]string{carol: nil}}) + resp, err := h.keys.QueryKeys(t.Context(), alice, keyserver.QueryRequest{DeviceKeys: map[string][]string{dave: nil}}) if err != nil { t.Fatal(err) } - if _, ok := resp.DeviceKeys[carol]["C1"]; !ok { + if _, ok := resp.DeviceKeys[dave]["V1"]; !ok { t.Errorf("remote keys = %v", resp.DeviceKeys) } if _, ok := resp.DeviceKeys["@mallory:other.test"]; ok { @@ -556,14 +578,18 @@ func TestRemoteKeyChangesOnlyConcernSharedUsers(t *testing.T) { ctx := t.Context() h := newHarness(t, nil) - if err := h.keys.RemoteKeysChanged(ctx, "@stranger:remote.example"); err != nil { + stranger := federation.DeviceListUpdateContent{UserID: dave, DeviceID: "V1", StreamID: 1} + if err := h.keys.RemoteDeviceListUpdate(ctx, stranger); err != nil { t.Fatal(err) } - if pos, _ := h.store.DeviceListPosition(ctx, "@stranger:remote.example"); pos != 0 { + if pos, _ := h.store.DeviceListPosition(ctx, dave); pos != 0 { t.Error("an update about a user nobody here shares a room with was recorded") } + if _, ok, _ := h.store.RemoteKeys(ctx, dave); ok { + t.Error("an update about a stranger left something in the cache") + } - if err := h.keys.RemoteKeysChanged(ctx, carol); err != nil { + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{UserID: carol, DeviceID: "C1", StreamID: 1}); err != nil { t.Fatal(err) } if pos, _ := h.store.DeviceListPosition(ctx, carol); pos == 0 { diff --git a/server/internal/keyserver/query.go b/server/internal/keyserver/query.go index 7a52d0b..7ea0225 100644 --- a/server/internal/keyserver/query.go +++ b/server/internal/keyserver/query.go @@ -174,7 +174,8 @@ func (k *API) crossSigningKeys(ctx context.Context, userID, requester string) (m // --- /keys/query ---------------------------------------------------------------- // QueryKeys answers POST /keys/query for requester: local users from the -// store, remote ones by asking their servers, all at once. +// store, remote ones from the cache or by asking their servers, all at +// once. func (k *API) QueryKeys(ctx context.Context, requester string, req QueryRequest) (QueryResponse, error) { resp := QueryResponse{ DeviceKeys: map[string]map[string]json.RawMessage{}, @@ -217,9 +218,14 @@ func (k *API) QueryKeys(ctx context.Context, requester string, req QueryRequest) return resp, nil } -// queryRemote asks each server for its users' keys in parallel and merges -// the answers in. A server that does not answer in time is a failure, not -// an error: the keys of everybody else are still worth returning. +// queryRemote answers for other servers' users: from the cache for those +// who share a room with somebody here, by asking their server for the +// rest. A copy that is missing or stale is fetched whole and cached, so +// the next query for the same user does not leave this server. +// +// The servers are asked in parallel. One that does not answer in time is a +// failure, not an error: the keys of everybody else are still worth +// returning. func (k *API) queryRemote(ctx context.Context, requester string, byServer map[string]map[string][]string, timeout time.Duration, resp *QueryResponse) { if len(byServer) == 0 { return @@ -231,33 +237,77 @@ func (k *API) queryRemote(ctx context.Context, requester string, byServer map[st mu sync.Mutex wg sync.WaitGroup ) + fail := func(server string, err error) { + mu.Lock() + defer mu.Unlock() + resp.Failures[server] = remoteFailure(err) + } + for server, users := range byServer { + passOn := map[string][]string{} + var fetch []string + for _, userID := range sortedKeys(users) { + if !k.cacheable(ctx, userID) { + // A user nobody here shares a room with any more may still + // have a copy from when somebody did. + k.forget(ctx, userID) + passOn[userID] = users[userID] + continue + } + if keys, ok := k.freshCopy(ctx, userID); ok { + mu.Lock() + k.serveRemote(ctx, requester, keys, users[userID], resp) + mu.Unlock() + continue + } + fetch = append(fetch, userID) + } + + if len(passOn) == 0 && len(fetch) == 0 { + continue + } if k.remote == nil { - resp.Failures[server] = remoteFailure(errors.New("federation is not available on this server")) + fail(server, errors.New("federation is not available on this server")) + continue + } + + for _, userID := range fetch { + wg.Go(func() { + keys, err := k.fetchRemoteKeys(ctx, server, userID) + if err != nil { + fail(server, err) + return + } + mu.Lock() + defer mu.Unlock() + k.serveRemote(ctx, requester, keys, users[userID], resp) + }) + } + if len(passOn) == 0 { continue } wg.Go(func() { - answer, err := k.remote.QueryUserKeys(ctx, server, federation.UserKeysQueryRequest{DeviceKeys: users}) + answer, err := k.remote.QueryUserKeys(ctx, server, federation.UserKeysQueryRequest{DeviceKeys: passOn}) + if err != nil { + fail(server, err) + return + } mu.Lock() defer mu.Unlock() - if err != nil { - resp.Failures[server] = remoteFailure(err) - return - } for userID, devices := range answer.DeviceKeys { // A server answers for its own users and nobody else's. - if _, asked := users[userID]; asked && devices != nil { + if _, asked := passOn[userID]; asked && devices != nil { resp.DeviceKeys[userID] = devices } } for userID, master := range answer.MasterKeys { - if _, asked := users[userID]; asked { + if _, asked := passOn[userID]; asked { resp.MasterKeys[userID] = k.withRequesterSignature(ctx, userID, requester, master) } } for userID, selfSigning := range answer.SelfSigningKeys { - if _, asked := users[userID]; asked { + if _, asked := passOn[userID]; asked { resp.SelfSigningKeys[userID] = selfSigning } } diff --git a/server/internal/keyserver/remotekeys.go b/server/internal/keyserver/remotekeys.go new file mode 100644 index 0000000..7f5327e --- /dev/null +++ b/server/internal/keyserver/remotekeys.go @@ -0,0 +1,419 @@ +package keyserver + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sync" + + "github.com/dariuszjaniak/warp/server/internal/federation" + "github.com/dariuszjaniak/warp/server/internal/storage" +) + +// The cache of other servers' users' keys. +// +// A local client asks for the keys of everybody in a room before it sends +// an encrypted message, and again whenever a sync says one of them moved. +// Without a cache each of those is a federation round trip, in front of +// the message, to a server that may be slow or down. With one, a remote +// user's keys are fetched once — the whole device list, through GET +// /user/devices/{userId}, which also says where in their server's device +// list stream that list stands — and then kept current by the updates +// their server pushes: an m.device_list_update per device change, naming +// the stream position it follows on from, and an m.signing_key_update per +// change of their cross-signing keys. +// +// Only users who share a room with somebody here are cached, because only +// for them is there anybody to send updates here: a server tells the +// servers its user shares rooms with, and nobody else. A copy of a +// stranger's keys would go stale without anybody knowing. So a query +// about a stranger is passed on to their server as before, and a cached +// user who stops sharing a room with anybody here is dropped. +// +// An update that does not follow on from the copy held — one was lost, or +// arrived out of order — marks the copy stale, and it is fetched again in +// the background. Until it is, a query does what it does on a miss: asks +// their server, synchronously. Serving a copy known to be missing a change +// would hand a sender a device list without the device just added, and +// the message would go out undecryptable on it. + +// cacheable reports whether a remote user's keys are kept here: whether a +// local user shares a room with them. An error reading the rooms is "no", +// which costs a federation round trip and nothing else. +func (k *API) cacheable(ctx context.Context, userID string) bool { + if k.rooms == nil || k.isLocal(userID) { + return false + } + shares, err := k.sharesWithLocal(ctx, userID) + if err != nil { + k.log.Warn("could not read who shares a room with a remote user", "user_id", userID, "error", err) + return false + } + return shares +} + +// sharesWithLocal reports whether any local user shares a room with +// userID. +func (k *API) sharesWithLocal(ctx context.Context, userID string) (bool, error) { + if k.rooms == nil { + return false, nil + } + users, err := k.rooms.UsersSharingRooms(ctx, userID) + if err != nil { + return false, err + } + return slices.ContainsFunc(users, k.isLocal), nil +} + +// forget drops what is cached for users. A failure only leaves a copy +// behind that the next sweep takes. +func (k *API) forget(ctx context.Context, userIDs ...string) { + if err := k.store.DeleteRemoteKeys(ctx, userIDs); err != nil { + k.log.Warn("could not drop cached keys", "user_ids", userIDs, "error", err) + } +} + +// freshCopy reads the cached keys of a remote user, if there is a copy +// that can be served. +func (k *API) freshCopy(ctx context.Context, userID string) (storage.RemoteKeys, bool) { + keys, ok, err := k.store.RemoteKeys(ctx, userID) + if err != nil { + k.log.Warn("could not read cached keys", "user_id", userID, "error", err) + return storage.RemoteKeys{}, false + } + return keys, ok && !keys.Stale +} + +// fetchRemoteKeys asks a user's server for their whole device list and +// cross-signing keys, and caches the answer. +// +// What the server sent is checked as far as the server can check it: a +// device whose keys name another user or device is left out, and so is a +// cross-signing key that is not that user's key for that purpose. The +// signatures are the client's to check — a server that verified them could +// still not tell a device its user added from one their server invented. +func (k *API) fetchRemoteKeys(ctx context.Context, server, userID string) (storage.RemoteKeys, error) { + if k.remote == nil { + return storage.RemoteKeys{}, errors.New("federation is not available on this server") + } + answer, err := k.remote.UserDevices(ctx, server, userID) + if err != nil { + return storage.RemoteKeys{}, err + } + if answer.UserID != userID { + return storage.RemoteKeys{}, fmt.Errorf("%s answered for %q when asked about %q", server, answer.UserID, userID) + } + + keys := storage.RemoteKeys{UserID: userID, StreamID: answer.StreamID, CrossSigning: map[string][]byte{}} + for _, d := range answer.Devices { + if !deviceKeysOf(d.Keys, userID, d.DeviceID) { + continue + } + keys.Devices = append(keys.Devices, storage.RemoteDevice{ + DeviceID: d.DeviceID, DisplayName: d.DeviceDisplayName, JSON: d.Keys, + }) + } + setCrossSigning(keys.CrossSigning, userID, storage.CrossSigningMaster, answer.MasterKey) + setCrossSigning(keys.CrossSigning, userID, storage.CrossSigningSelfSigning, answer.SelfSigningKey) + + // Still a stranger's keys if the last shared room went while the + // request was out; they are served this once and not kept. + if !k.cacheable(ctx, userID) { + return keys, nil + } + if _, err := k.store.SetRemoteKeys(ctx, keys); err != nil { + k.log.Warn("could not cache keys", "user_id", userID, "error", err) + } + return keys, nil +} + +// deviceKeysOf reports whether a device_keys object is the one of userID's +// device deviceID. +func deviceKeysOf(raw json.RawMessage, userID, deviceID string) bool { + var keys deviceKeysObject + return json.Unmarshal(raw, &keys) == nil && keys.UserID == userID && keys.DeviceID == deviceID && deviceID != "" +} + +// setCrossSigning adds a remote user's cross-signing key to keys when it +// is one. +func setCrossSigning(keys map[string][]byte, userID, purpose string, raw json.RawMessage) { + if len(raw) == 0 || string(raw) == "null" { + return + } + if encoded, _, err := checkCrossSigningKey(userID, purpose, raw); err == nil { + keys[purpose] = encoded + } +} + +// serveRemote writes a cached copy into a query response, limited to the +// devices asked for. +func (k *API) serveRemote(ctx context.Context, requester string, keys storage.RemoteKeys, deviceIDs []string, resp *QueryResponse) { + devices := map[string]json.RawMessage{} + for _, d := range keys.Devices { + if len(deviceIDs) > 0 && !slices.Contains(deviceIDs, d.DeviceID) { + continue + } + raw, err := withSignatures(d.JSON, nil, d.DisplayName) + if err != nil { + raw = d.JSON + } + devices[d.DeviceID] = raw + } + resp.DeviceKeys[keys.UserID] = devices + if master, ok := keys.CrossSigning[storage.CrossSigningMaster]; ok { + resp.MasterKeys[keys.UserID] = k.withRequesterSignature(ctx, keys.UserID, requester, master) + } + if selfSigning, ok := keys.CrossSigning[storage.CrossSigningSelfSigning]; ok { + resp.SelfSigningKeys[keys.UserID] = selfSigning + } +} + +// --- updates from other servers ------------------------------------------------- + +// RemoteDeviceListUpdate takes in an m.device_list_update from another +// server: one of its users' devices was added, changed or deleted. +// +// The update is applied to the cached copy when it follows on from it, +// and otherwise marks the copy stale and has it fetched again. Either way +// the user's position in the device list stream moves, which is what +// tells local clients — who keep their own copies — to ask again. +// +// Nothing happens unless a local user shares a room with the user: an +// update about a stranger concerns nobody here, and recording it would let +// any server grow the tables with users of its own invention. A cached +// user who has become a stranger is dropped. +func (k *API) RemoteDeviceListUpdate(ctx context.Context, u federation.DeviceListUpdateContent) error { + if k.isLocal(u.UserID) || k.rooms == nil { + return nil + } + shares, err := k.sharesWithLocal(ctx, u.UserID) + if err != nil { + return err + } + if !shares { + k.forget(ctx, u.UserID) + return nil + } + + update := storage.RemoteDeviceUpdate{ + UserID: u.UserID, DeviceID: u.DeviceID, DisplayName: u.DeviceDisplayName, + Deleted: u.Deleted, StreamID: u.StreamID, PrevIDs: u.PrevID, + } + // Keys that are not the device's own are as good as none: the update + // cannot be applied, and the copy is fetched again. + if !u.Deleted && deviceKeysOf(u.Keys, u.UserID, u.DeviceID) { + update.JSON = u.Keys + } + result, err := k.store.UpdateRemoteDevice(ctx, update) + if err != nil { + return err + } + switch result { + case storage.RemoteUpdateDuplicate: + return nil + case storage.RemoteUpdateStale: + k.resyncs.add(u.UserID) + } + return k.remoteChanged(ctx, u.UserID) +} + +// RemoteSigningKeyUpdate takes in an m.signing_key_update: one of another +// server's users replaced their master or self-signing key. The update +// carries the keys as they now stand, and replaces the cached ones; a key +// it leaves out no longer exists. +func (k *API) RemoteSigningKeyUpdate(ctx context.Context, u federation.SigningKeyUpdateContent) error { + if k.isLocal(u.UserID) || k.rooms == nil { + return nil + } + shares, err := k.sharesWithLocal(ctx, u.UserID) + if err != nil { + return err + } + if !shares { + k.forget(ctx, u.UserID) + return nil + } + + keys := map[string][]byte{} + setCrossSigning(keys, u.UserID, storage.CrossSigningMaster, u.MasterKey) + setCrossSigning(keys, u.UserID, storage.CrossSigningSelfSigning, u.SelfSigningKey) + if _, err := k.store.SetRemoteCrossSigningKeys(ctx, u.UserID, keys); err != nil { + return err + } + return k.remoteChanged(ctx, u.UserID) +} + +// remoteChanged moves a remote user's position in the device list stream +// and wakes the local syncs that share a room with them. +func (k *API) remoteChanged(ctx context.Context, userID string) error { + if _, err := k.store.MarkDeviceListsChanged(ctx, []string{userID}); err != nil { + return err + } + k.changed(ctx, userID) + return nil +} + +// resync fetches a stale copy again. It runs in the background: the +// update that found the copy stale arrived in a transaction from the +// user's server, which should not wait on a request back to itself. +func (k *API) resync(ctx context.Context, userID string) { + if !k.cacheable(ctx, userID) { + k.forget(ctx, userID) + return + } + server := serverOf(userID) + if server == "" { + return + } + ctx, cancel := context.WithTimeout(ctx, defaultRemoteTimeout) + defer cancel() + if _, err := k.fetchRemoteKeys(ctx, server, userID); err != nil { + // The copy stays stale, and the next query asks again. + k.log.Debug("could not fetch stale keys again", "user_id", userID, "error", err) + } +} + +// --- dropping strangers --------------------------------------------------------- + +// MembershipLeft is told that a user's membership of a room ended: they +// left, were kicked or were banned. If the user is remote, they may no +// longer share a room with anybody here, and their cached keys go; if the +// user is local, any of the remote users they shared that room with may +// have stopped sharing one with anybody, and every cached user is +// checked. +// +// It does not block. The check reads room memberships, and is called from +// the roomserver's emit stage with the room's lock held. +func (k *API) MembershipLeft(userID string) { + if k.isLocal(userID) { + k.sweeps.add(sweepEveryone) + return + } + k.sweeps.add(userID) +} + +// sweepEveryone is the sweep queue's key for "check every cached user". +const sweepEveryone = "" + +// sweep drops the cached keys of users nobody here shares a room with any +// more: one user, or with sweepEveryone all of them. +func (k *API) sweep(ctx context.Context, userID string) { + users := []string{userID} + if userID == sweepEveryone { + var err error + if users, err = k.store.RemoteKeyUsers(ctx); err != nil { + k.log.Warn("could not list the users with cached keys", "error", err) + return + } + } + var strangers []string + for _, u := range users { + shares, err := k.sharesWithLocal(ctx, u) + if err != nil { + k.log.Warn("could not read who shares a room with a remote user", "user_id", u, "error", err) + continue + } + if !shares { + strangers = append(strangers, u) + } + } + if len(strangers) > 0 { + k.forget(ctx, strangers...) + } +} + +// --- background work ------------------------------------------------------------ + +// workQueue runs a job per key in one background goroutine, one key at a +// time, and folds a key added while it is already waiting into the wait. +// A burst of updates about one user is one resync, and a burst of leaves +// one sweep. +type workQueue struct { + k *API + run func(ctx context.Context, key string) + + mu sync.Mutex + pending []string + running bool +} + +func newWorkQueue(k *API, run func(context.Context, string)) *workQueue { + return &workQueue{k: k, run: run} +} + +// add queues a key, and starts the queue's goroutine if it is not running. +// After Close it does nothing. +func (q *workQueue) add(key string) { + q.mu.Lock() + defer q.mu.Unlock() + if !slices.Contains(q.pending, key) { + q.pending = append(q.pending, key) + } + if q.running || !q.k.background.start() { + return + } + q.running = true + go func() { + defer q.k.background.done() + for { + q.mu.Lock() + if len(q.pending) == 0 { + q.running = false + q.mu.Unlock() + return + } + key := q.pending[0] + q.pending = q.pending[1:] + q.mu.Unlock() + + q.run(q.k.background.ctx, key) + } + }() +} + +// background is the lifetime of the keyserver's goroutines. +type background struct { + ctx context.Context + cancel context.CancelFunc + + mu sync.Mutex + closed bool + wg sync.WaitGroup +} + +func newBackground() *background { + ctx, cancel := context.WithCancel(context.Background()) + return &background{ctx: ctx, cancel: cancel} +} + +// start counts a goroutine about to start, unless the keyserver is closed. +func (b *background) start() bool { + b.mu.Lock() + defer b.mu.Unlock() + if b.closed { + return false + } + b.wg.Add(1) + return true +} + +func (b *background) done() { b.wg.Done() } + +// wait waits for the goroutines running now to finish. +func (b *background) wait() { b.wg.Wait() } + +// close stops the goroutines and waits for them. +func (b *background) close() { + b.mu.Lock() + b.closed = true + b.mu.Unlock() + b.cancel() + b.wg.Wait() +} + +// Close stops the keyserver's background work — fetching stale copies of +// remote keys again, and dropping those of users nobody here shares a room +// with — and waits for it to stop. +func (k *API) Close() { k.background.close() } diff --git a/server/internal/keyserver/remotekeys_test.go b/server/internal/keyserver/remotekeys_test.go new file mode 100644 index 0000000..cf821cb --- /dev/null +++ b/server/internal/keyserver/remotekeys_test.go @@ -0,0 +1,309 @@ +package keyserver_test + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + + "github.com/dariuszjaniak/warp/server/internal/federation" + "github.com/dariuszjaniak/warp/server/internal/keyserver" +) + +// snapshotRemote is remote.example with Carol's device list in it, and a +// count of how often it was asked. +type snapshotRemote struct { + mu sync.Mutex + snapshot federation.UserDevicesResponse + fetches int + queries int + down bool +} + +func (r *snapshotRemote) QueryUserKeys(context.Context, string, federation.UserKeysQueryRequest) (federation.UserKeysQueryResponse, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.queries++ + return federation.UserKeysQueryResponse{}, nil +} + +func (r *snapshotRemote) ClaimUserKeys(context.Context, string, federation.UserKeysClaimRequest) (federation.UserKeysClaimResponse, error) { + return federation.UserKeysClaimResponse{}, errors.New("unreachable") +} + +func (r *snapshotRemote) UserDevices(_ context.Context, _, userID string) (federation.UserDevicesResponse, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.fetches++ + if r.down { + return federation.UserDevicesResponse{}, &federation.HTTPError{Status: 502} + } + out := r.snapshot + out.UserID = userID + return out, nil +} + +func (r *snapshotRemote) set(stream int64, devices ...federation.UserDevice) { + r.mu.Lock() + defer r.mu.Unlock() + r.snapshot.StreamID = stream + r.snapshot.Devices = devices +} + +func (r *snapshotRemote) counts() (fetches, queries int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.fetches, r.queries +} + +func queryCarol(t *testing.T, h harness) keyserver.QueryResponse { + t.Helper() + resp, err := h.keys.QueryKeys(t.Context(), alice, keyserver.QueryRequest{DeviceKeys: map[string][]string{carol: nil}}) + if err != nil { + t.Fatal(err) + } + return resp +} + +func carolDevice(t *testing.T, deviceID string) federation.UserDevice { + t.Helper() + _, keys := device(t, carol, deviceID) + return federation.UserDevice{DeviceID: deviceID, Keys: keys, DeviceDisplayName: "Carol's " + deviceID} +} + +func TestRemoteKeysAreCached(t *testing.T) { + remote := &snapshotRemote{} + master := newKey(t) + remote.snapshot.MasterKey = signed(t, carol, master, crossSigningKey(carol, "master", master)) + remote.set(5, carolDevice(t, "C1"), + // A device whose keys name somebody else is not Carol's. + federation.UserDevice{DeviceID: "FAKE", Keys: json.RawMessage(`{"user_id":"@mallory:remote.example","device_id":"FAKE"}`)}) + h := newHarness(t, remote) + + first := queryCarol(t, h) + second := queryCarol(t, h) + if fetches, queries := remote.counts(); fetches != 1 || queries != 0 { + t.Errorf("two queries made %d fetches and %d key queries, want 1 and 0", fetches, queries) + } + for i, resp := range []keyserver.QueryResponse{first, second} { + raw, ok := resp.DeviceKeys[carol]["C1"] + if !ok { + t.Fatalf("query %d: device keys = %v", i, resp.DeviceKeys) + } + var keys struct { + Unsigned map[string]string `json:"unsigned"` + } + _ = json.Unmarshal(raw, &keys) + if keys.Unsigned["device_display_name"] != "Carol's C1" { + t.Errorf("query %d: C1 = %s, want its display name", i, raw) + } + if _, ok := resp.DeviceKeys[carol]["FAKE"]; ok { + t.Errorf("query %d: a device with another user's keys was served", i) + } + if resp.MasterKeys[carol] == nil { + t.Errorf("query %d: master keys = %v", i, resp.MasterKeys) + } + } + + // Asking for one device is answered from the same copy. + resp, _ := h.keys.QueryKeys(t.Context(), alice, keyserver.QueryRequest{DeviceKeys: map[string][]string{carol: {"NOPE"}}}) + if len(resp.DeviceKeys[carol]) != 0 { + t.Errorf("a query for a device Carol does not have = %v", resp.DeviceKeys[carol]) + } + + // A stranger is not cached: every query for them is a key query. + for range 2 { + _, _ = h.keys.QueryKeys(t.Context(), alice, keyserver.QueryRequest{DeviceKeys: map[string][]string{dave: nil}}) + } + if fetches, queries := remote.counts(); fetches != 1 || queries != 2 { + t.Errorf("after two queries for a stranger: %d fetches, %d key queries; want 1 and 2", fetches, queries) + } + if _, ok, _ := h.store.RemoteKeys(t.Context(), dave); ok { + t.Error("a stranger's keys were cached") + } +} + +func TestRemoteDeviceListUpdatesKeepTheCacheCurrent(t *testing.T) { + ctx := t.Context() + remote := &snapshotRemote{} + remote.set(5, carolDevice(t, "C1")) + h := newHarness(t, remote) + queryCarol(t, h) + + // An update that follows on from the copy is applied to it. + c2 := carolDevice(t, "C2") + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{ + UserID: carol, DeviceID: "C2", StreamID: 6, PrevID: []int64{5}, Keys: c2.Keys, + }); err != nil { + t.Fatal(err) + } + if resp := queryCarol(t, h); resp.DeviceKeys[carol]["C2"] == nil || resp.DeviceKeys[carol]["C1"] == nil { + t.Errorf("after the update: %v, want C1 and C2", resp.DeviceKeys[carol]) + } + if fetches, _ := remote.counts(); fetches != 1 { + t.Errorf("an update that followed on caused %d fetches, want 1", fetches) + } + pos, _ := h.store.DeviceListPosition(ctx, carol) + if pos == 0 { + t.Error("the update did not move Carol's device list") + } + + // The same update again changes nothing, not even the stream. + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{ + UserID: carol, DeviceID: "C2", StreamID: 6, PrevID: []int64{5}, Keys: c2.Keys, + }); err != nil { + t.Fatal(err) + } + if again, _ := h.store.DeviceListPosition(ctx, carol); again != pos { + t.Errorf("a repeated update moved the device list from %d to %d", pos, again) + } + + // A deletion that follows on. + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{ + UserID: carol, DeviceID: "C1", StreamID: 7, PrevID: []int64{6}, Deleted: true, + }); err != nil { + t.Fatal(err) + } + if resp := queryCarol(t, h); resp.DeviceKeys[carol]["C1"] != nil { + t.Errorf("a deleted device is still served: %v", resp.DeviceKeys[carol]) + } + + // An update after a gap marks the copy stale, and it is fetched again + // in the background. + remote.set(10, carolDevice(t, "C3")) + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{ + UserID: carol, DeviceID: "C3", StreamID: 10, PrevID: []int64{9}, Keys: json.RawMessage(`{}`), + }); err != nil { + t.Fatal(err) + } + h.keys.WaitIdle() + if fetches, _ := remote.counts(); fetches != 2 { + t.Errorf("after a gap there were %d fetches, want 2", fetches) + } + cached, ok, _ := h.store.RemoteKeys(ctx, carol) + if !ok || cached.Stale || cached.StreamID != 10 || len(cached.Devices) != 1 || cached.Devices[0].DeviceID != "C3" { + t.Errorf("after the resync the cache holds %+v", cached) + } + resp := queryCarol(t, h) + if resp.DeviceKeys[carol]["C3"] == nil || len(resp.DeviceKeys[carol]) != 1 { + t.Errorf("after the resync: %v, want C3 alone", resp.DeviceKeys[carol]) + } + if fetches, _ := remote.counts(); fetches != 2 { + t.Errorf("a query after the resync fetched again: %d fetches", fetches) + } +} + +func TestAStaleCopyIsNotServed(t *testing.T) { + ctx := t.Context() + remote := &snapshotRemote{} + remote.set(5, carolDevice(t, "C1")) + h := newHarness(t, remote) + queryCarol(t, h) + + // With Carol's server down, the background fetch fails and the copy + // stays stale; a query then asks again rather than serve it. + remote.mu.Lock() + remote.down = true + remote.mu.Unlock() + if err := h.keys.RemoteDeviceListUpdate(ctx, federation.DeviceListUpdateContent{ + UserID: carol, DeviceID: "C2", StreamID: 8, PrevID: []int64{7}, + }); err != nil { + t.Fatal(err) + } + h.keys.WaitIdle() + resp := queryCarol(t, h) + if failure, ok := resp.Failures["remote.example"].(map[string]any); !ok || failure["status"] != 502 { + t.Errorf("failures = %v, want a 502 for remote.example", resp.Failures) + } + if fetches, _ := remote.counts(); fetches != 3 { + t.Errorf("%d fetches, want the first, the background one and the query's", fetches) + } +} + +func TestRemoteSigningKeyUpdatesReplaceTheCrossSigningKeys(t *testing.T) { + ctx := t.Context() + remote := &snapshotRemote{} + oldMaster, newMaster, selfSigning := newKey(t), newKey(t), newKey(t) + remote.snapshot.MasterKey = signed(t, carol, oldMaster, crossSigningKey(carol, "master", oldMaster)) + remote.snapshot.SelfSigningKey = signed(t, carol, oldMaster, crossSigningKey(carol, "self_signing", selfSigning)) + remote.set(1, carolDevice(t, "C1")) + h := newHarness(t, remote) + queryCarol(t, h) + + if err := h.keys.RemoteSigningKeyUpdate(ctx, federation.SigningKeyUpdateContent{ + UserID: carol, + MasterKey: signed(t, carol, newMaster, crossSigningKey(carol, "master", newMaster)), + }); err != nil { + t.Fatal(err) + } + resp := queryCarol(t, h) + var master struct { + Keys map[string]string `json:"keys"` + } + _ = json.Unmarshal(resp.MasterKeys[carol], &master) + if master.Keys["ed25519:"+newMaster.PublicBase64()] == "" { + t.Errorf("master key after the update = %s, want the new one", resp.MasterKeys[carol]) + } + if resp.SelfSigningKeys[carol] != nil { + t.Errorf("a self-signing key the update left out is still served: %s", resp.SelfSigningKeys[carol]) + } + if fetches, _ := remote.counts(); fetches != 1 { + t.Errorf("a signing key update caused %d fetches, want 1", fetches) + } + if !h.notifier.wasWoken(alice) { + t.Error("alice was not told Carol's keys changed") + } +} + +func TestCachedKeysOfStrangersAreDropped(t *testing.T) { + ctx := t.Context() + remote := &snapshotRemote{} + remote.set(1, carolDevice(t, "C1")) + h := newHarness(t, remote) + queryCarol(t, h) + + // Alice leaving a room is not enough while Carol still shares another + // one with her. + h.keys.MembershipLeft(alice) + h.keys.WaitIdle() + if _, ok, _ := h.store.RemoteKeys(ctx, carol); !ok { + t.Fatal("the cache was dropped while Carol still shares a room with alice") + } + + h.rooms.set(carol, carol) + h.keys.MembershipLeft(carol) + h.keys.WaitIdle() + if _, ok, _ := h.store.RemoteKeys(ctx, carol); ok { + t.Error("Carol's keys are still cached after she stopped sharing a room with anybody here") + } + if users, _ := h.store.RemoteKeyUsers(ctx); len(users) != 0 { + t.Errorf("users with cached keys = %v", users) + } +} + +func TestCachedKeysAreDroppedWhenALocalUserLeaves(t *testing.T) { + ctx := t.Context() + remote := &snapshotRemote{} + remote.set(1, carolDevice(t, "C1")) + h := newHarness(t, remote) + queryCarol(t, h) + + // Alice left the room she shared with Carol: everybody cached is + // checked, since it is not known who else was in it. + h.rooms.set(carol, carol) + h.keys.MembershipLeft(alice) + h.keys.WaitIdle() + if _, ok, _ := h.store.RemoteKeys(ctx, carol); ok { + t.Error("Carol's keys are still cached after the last local user left") + } + + // And a query for her now passes through. + queryCarol(t, h) + if fetches, queries := remote.counts(); fetches != 1 || queries != 1 { + t.Errorf("%d fetches and %d key queries, want 1 and 1", fetches, queries) + } + if _, ok, _ := h.store.RemoteKeys(ctx, carol); ok { + t.Error("a stranger's keys were cached") + } +} diff --git a/server/internal/setup/federation.go b/server/internal/setup/federation.go index 1226f26..d009e12 100644 --- a/server/internal/setup/federation.go +++ b/server/internal/setup/federation.go @@ -9,6 +9,8 @@ import ( "sync" "github.com/dariuszjaniak/warp/server/internal/federationapi" + "github.com/dariuszjaniak/warp/server/internal/keyserver" + "github.com/dariuszjaniak/warp/server/internal/matrix/event" "github.com/dariuszjaniak/warp/server/internal/roomserver" "github.com/dariuszjaniak/warp/server/internal/storage" ) @@ -39,6 +41,27 @@ func (f *fanout) EventPersisted(ctx context.Context, n roomserver.Notification) } } +// keyMemberships is the keyserver's listener on the emit stage: a +// membership ending is when a remote user may stop sharing a room with +// anybody here, and the keyserver drops its cached copy of their keys once +// they have. MembershipLeft does not block, as the emit stage requires. +type keyMemberships struct{ keys *keyserver.API } + +// EventPersisted implements roomserver.Notifier. +func (l keyMemberships) EventPersisted(_ context.Context, n roomserver.Notification) { + if n.Event == nil || n.Event.Type() != event.TypeMember || !n.StateChanged { + return + } + userID, ok := n.Event.StateKey() + if !ok { + return + } + switch membership, _ := n.Event.Membership(); membership { + case event.MembershipLeave, event.MembershipBan: + l.keys.MembershipLeft(userID) + } +} + // federationStatus adapts the module to the admin API. type federationStatus struct { module *federationapi.Module diff --git a/server/internal/setup/phase7_test.go b/server/internal/setup/phase7_test.go index 58800a9..1fe9cae 100644 --- a/server/internal/setup/phase7_test.go +++ b/server/internal/setup/phase7_test.go @@ -8,6 +8,8 @@ import ( "net/http" "net/http/httptest" "net/url" + "strings" + "sync/atomic" "testing" "github.com/dariuszjaniak/warp/server/internal/matrix/signing" @@ -280,6 +282,19 @@ func TestKeysOverFederation(t *testing.T) { a, b := nodes["hs-a.example"].srv, nodes["hs-b.example"].srv const alice, bob = "@alice:hs-a.example", "@bob:hs-b.example" + // Count the key requests that reach A, which is how the test tells a + // query B answered from its cache from one it passed on. + var keyRequests atomic.Int64 + inner := *nodes["hs-a.example"].handler.Load() + counting := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.URL.Path, "/_matrix/federation/v1/user/devices/") || + r.URL.Path == "/_matrix/federation/v1/user/keys/query" { + keyRequests.Add(1) + } + inner.ServeHTTP(w, r) + })) + nodes["hs-a.example"].handler.Store(&counting) + aliceToken := registerThrough(t, a, "alice") bobToken := registerThrough(t, b, "bob") aliceDevice := deviceIDOf(t, a, aliceToken) @@ -297,12 +312,43 @@ func TestKeysOverFederation(t *testing.T) { return contains(got.DeviceLists.Changed, alice) }) - // Bob's query and claim are answered by Alice's server. - query := mustOK(t, post(t, b, "/_matrix/client/v3/keys/query", map[string]any{ - "device_keys": map[string]any{alice: []string{}}, - }, bobToken), "remote keys/query") - if devices, _ := query["device_keys"].(map[string]any)[alice].(map[string]any); devices[aliceDevice] == nil { - t.Fatalf("remote keys/query = %v", query) + // Bob's first query is answered by Alice's server, and cached: the + // second does not leave B. + queryAlice := func(what string) map[string]any { + t.Helper() + query := mustOK(t, post(t, b, "/_matrix/client/v3/keys/query", map[string]any{ + "device_keys": map[string]any{alice: []string{}}, + }, bobToken), what) + devices, _ := query["device_keys"].(map[string]any)[alice].(map[string]any) + return devices + } + if devices := queryAlice("remote keys/query"); devices[aliceDevice] == nil { + t.Fatalf("remote keys/query = %v", devices) + } + if n := keyRequests.Load(); n != 1 { + t.Errorf("the first query made %d key requests to A, want 1", n) + } + if devices := queryAlice("second remote keys/query"); devices[aliceDevice] == nil { + t.Fatalf("second remote keys/query = %v", devices) + } + if n := keyRequests.Load(); n != 1 { + t.Errorf("the second query reached A: %d key requests, want 1", n) + } + + // A new device of Alice's reaches B as an m.device_list_update that + // follows on from the cached copy, which B applies: Bob sees the new + // device, and A is still not asked. + login := mustOK(t, post(t, a, "/_matrix/client/v3/login", map[string]any{ + "type": "m.login.password", "identifier": map[string]string{"type": "m.id.user", "user": "alice"}, + "password": "sUp3rs3kr1t", "device_id": "PHONE", + }, ""), "login") + uploadDeviceKeys(t, a, login["access_token"].(string), alice, "PHONE", 1) + eventually(t, "alice's new device in bob's query", func() bool { + devices := queryAlice("keys/query after the update") + return devices["PHONE"] != nil && devices[aliceDevice] != nil + }) + if n := keyRequests.Load(); n != 1 { + t.Errorf("the update was not applied from the EDU: %d key requests to A, want 1", n) } claim := mustOK(t, post(t, b, "/_matrix/client/v3/keys/claim", map[string]any{ "one_time_keys": map[string]any{alice: map[string]string{aliceDevice: "signed_curve25519"}}, @@ -320,9 +366,15 @@ func TestKeysOverFederation(t *testing.T) { if err := a.Settings().SetFederationEnabled(t.Context(), false); err != nil { t.Fatal(err) } - query = mustOK(t, post(t, b, "/_matrix/client/v3/keys/query", map[string]any{ - "device_keys": map[string]any{alice: []string{}}, "timeout": 5000, - }, bobToken), "keys/query with A's federation off") + // The cached copy is still B's to serve. + if devices := queryAlice("keys/query with A's federation off"); devices["PHONE"] == nil { + t.Errorf("with A's federation off, B no longer serves its cached copy: %v", devices) + } + // A user B has nothing cached for is asked of A, and A is reported as + // a failure instead of hanging. + query := mustOK(t, post(t, b, "/_matrix/client/v3/keys/query", map[string]any{ + "device_keys": map[string]any{"@stranger:hs-a.example": []string{}}, "timeout": 5000, + }, bobToken), "keys/query for a stranger with A's federation off") if failures, _ := query["failures"].(map[string]any); failures["hs-a.example"] == nil { t.Errorf("failures = %v, want hs-a.example", query["failures"]) } diff --git a/server/internal/setup/server.go b/server/internal/setup/server.go index b24a1d2..2d9a29f 100644 --- a/server/internal/setup/server.go +++ b/server/internal/setup/server.go @@ -295,6 +295,7 @@ func New(ctx context.Context, cfg *config.Config, opts Options) (*Server, error) return nil, err } s.users.SetDeviceObserver(s.keys) + emit.add(keyMemberships{keys: s.keys}) s.sync, err = syncapi.New(syncapi.Options{ ServerName: cfg.Global.ServerName, @@ -576,6 +577,11 @@ func (s *Server) Close() error { if s.fedModule != nil { s.fedModule.Stop() } + if s.keys != nil { + // Background fetches of remote keys write to the database, which + // is about to close. + s.keys.Close() + } if s.sync != nil { // The typing tracker holds a timer per typing user; a server // shutting down should not wait for them. diff --git a/server/internal/storage/keys.go b/server/internal/storage/keys.go index 6588742..c50370e 100644 --- a/server/internal/storage/keys.go +++ b/server/internal/storage/keys.go @@ -119,6 +119,77 @@ type DeviceListChange struct { StreamPos int64 } +// RemoteDevice is one device of a remote user, as their server described +// it. +type RemoteDevice struct { + DeviceID string + DisplayName string + + // JSON is the device_keys object as the user's server sent it. + JSON []byte +} + +// RemoteKeys is the cached copy of a remote user's keys. +type RemoteKeys struct { + UserID string + + // StreamID is the position in the user's server's device list stream + // the copy is at: the stream_id of the snapshot it was fetched as, or + // of the last update applied to it since. + StreamID int64 + + // Stale says an update was missed, or did not make sense, and the copy + // is not to be served until it has been fetched again. A stale copy + // may have no devices at all: an update about a user with nothing + // cached leaves a stale marker, so that a snapshot fetched before the + // update happened is not taken for a current one. + Stale bool + + Devices []RemoteDevice + + // CrossSigning is the user's master and self-signing keys by purpose. + CrossSigning map[string][]byte +} + +// RemoteDeviceUpdate is one m.device_list_update, as the store applies it. +type RemoteDeviceUpdate struct { + UserID string + DeviceID string + DisplayName string + + // JSON is the device's new keys, or nil for a device that was deleted + // or that the update carried no keys for. + JSON []byte + Deleted bool + + StreamID int64 + PrevIDs []int64 +} + +// RemoteUpdateResult is what applying a RemoteDeviceUpdate did. +type RemoteUpdateResult int + +const ( + // RemoteUpdateApplied is an update that followed on from the copy + // held, and is now part of it. + RemoteUpdateApplied RemoteUpdateResult = iota + 1 + + // RemoteUpdateDuplicate is an update at or before the position the + // copy is already at: a retransmission, changing nothing. + RemoteUpdateDuplicate + + // RemoteUpdateStale is an update to a copy that is stale: one that did + // not follow on from the copy held, which is now marked so, or any + // update to a copy already marked. Either way the copy has to be + // fetched again. + RemoteUpdateStale + + // RemoteUpdateNotCached is an update about a user with nothing cached + // at all. A stale marker is written at the update's position, so that + // a snapshot fetched before it is not stored as current. + RemoteUpdateNotCached +) + // BackupVersion is one version of a user's server-side key backup. type BackupVersion struct { UserID string @@ -248,6 +319,33 @@ type KeyStore interface { // stream, or 0 when their list has never moved. DeviceListPosition(ctx context.Context, userID string) (int64, error) + // RemoteKeys reads the cached copy of a remote user's keys. The + // boolean is false when nothing at all is held for them. + RemoteKeys(ctx context.Context, userID string) (RemoteKeys, bool, error) + + // SetRemoteKeys replaces the cached copy of a remote user's keys with + // a snapshot fetched from their server, and clears the stale mark. A + // snapshot older than the position already held — fetched while an + // update was arriving — is not stored, and the boolean is false. + SetRemoteKeys(ctx context.Context, keys RemoteKeys) (bool, error) + + // UpdateRemoteDevice applies an m.device_list_update to a cached copy: + // only when one of its PrevIDs is the position held, otherwise the + // copy is marked stale. The position held never moves backwards. + UpdateRemoteDevice(ctx context.Context, update RemoteDeviceUpdate) (RemoteUpdateResult, error) + + // SetRemoteCrossSigningKeys replaces a cached user's master and + // self-signing keys; a purpose missing from keys is deleted. The + // boolean is false, and nothing is written, when the user has no + // fresh copy cached. + SetRemoteCrossSigningKeys(ctx context.Context, userID string, keys map[string][]byte) (bool, error) + + // RemoteKeyUsers lists every user something is cached for. + RemoteKeyUsers(ctx context.Context) ([]string, error) + + // DeleteRemoteKeys drops everything cached for the given users. + DeleteRemoteKeys(ctx context.Context, userIDs []string) error + // CreateBackup creates a new key backup version, which becomes the // user's current one, and returns its version. CreateBackup(ctx context.Context, userID, algorithm string, authData []byte) (string, error) diff --git a/server/internal/storage/keys_test.go b/server/internal/storage/keys_test.go index 2cf42ab..bc810d1 100644 --- a/server/internal/storage/keys_test.go +++ b/server/internal/storage/keys_test.go @@ -27,7 +27,8 @@ func migratedKeysDB(t *testing.T, driver string) storage.Database { defer func() { _ = pool.Close() }() if _, err := pool.ExecContext(t.Context(), `TRUNCATE keyserver_device_keys, keyserver_one_time_keys, keyserver_fallback_keys, keyserver_cross_signing_keys, keyserver_signatures, - keyserver_device_lists, keyserver_backup_versions, keyserver_backup_keys`); err != nil { + keyserver_device_lists, keyserver_backup_versions, keyserver_backup_keys, + keyserver_remote_users, keyserver_remote_device_keys, keyserver_remote_cross_signing_keys`); err != nil { t.Fatalf("reset the keyserver tables: %v", err) } } diff --git a/server/internal/storage/pocketbase/pocketbase.go b/server/internal/storage/pocketbase/pocketbase.go index 541fee2..4f3b2d5 100644 --- a/server/internal/storage/pocketbase/pocketbase.go +++ b/server/internal/storage/pocketbase/pocketbase.go @@ -581,3 +581,27 @@ func (notImplementedKeys) BackupKeys(context.Context, string, string, string, st func (notImplementedKeys) DeleteBackupKeys(context.Context, string, string, string, string) (storage.BackupVersion, error) { return storage.BackupVersion{}, storage.ErrNotImplemented } + +func (notImplementedKeys) RemoteKeys(context.Context, string) (storage.RemoteKeys, bool, error) { + return storage.RemoteKeys{}, false, storage.ErrNotImplemented +} + +func (notImplementedKeys) SetRemoteKeys(context.Context, storage.RemoteKeys) (bool, error) { + return false, storage.ErrNotImplemented +} + +func (notImplementedKeys) UpdateRemoteDevice(context.Context, storage.RemoteDeviceUpdate) (storage.RemoteUpdateResult, error) { + return 0, storage.ErrNotImplemented +} + +func (notImplementedKeys) SetRemoteCrossSigningKeys(context.Context, string, map[string][]byte) (bool, error) { + return false, storage.ErrNotImplemented +} + +func (notImplementedKeys) RemoteKeyUsers(context.Context) ([]string, error) { + return nil, storage.ErrNotImplemented +} + +func (notImplementedKeys) DeleteRemoteKeys(context.Context, []string) error { + return storage.ErrNotImplemented +} diff --git a/server/internal/storage/postgres/migrations/0009_remote_keys.sql b/server/internal/storage/postgres/migrations/0009_remote_keys.sql new file mode 100644 index 0000000..cd6af2c --- /dev/null +++ b/server/internal/storage/postgres/migrations/0009_remote_keys.sql @@ -0,0 +1,48 @@ +-- The cache of other servers' users' keys. +-- +-- A local client about to send an encrypted message asks for the keys of +-- everyone in the room, and asks again whenever a sync says their device +-- list moved. Answering that for a remote user by asking their server +-- every time puts a federation round trip in front of every message, and +-- turns a busy room into a steady stream of identical queries. So the +-- keys of remote users who share a room with somebody here are kept, and +-- kept current by the m.device_list_update and m.signing_key_update EDUs +-- their server sends whenever they change. +-- +-- These are tables of their own rather than rows in the local key tables. +-- A local user's keys are the source of truth, and every write to them +-- moves the device list stream and goes out over federation; a cached +-- copy of somebody else's is neither, and keeping them apart means no +-- code path that announces a local change can ever announce a copy. + +-- One row per cached user: the marker that the user's device list is held +-- here at all, and the position in their server's device list stream the +-- copy is at. An update names the position before it (prev_id), and is +-- applied only when that is the position held; anything else means an +-- update was missed, and the copy is marked stale until it is fetched +-- again whole. +CREATE TABLE IF NOT EXISTS keyserver_remote_users ( + user_id TEXT NOT NULL PRIMARY KEY, + stream_id BIGINT NOT NULL, + stale BOOLEAN NOT NULL DEFAULT FALSE +); + +-- The identity keys of a cached user's devices, as their server sent +-- them, and the display name it gave each. +CREATE TABLE IF NOT EXISTS keyserver_remote_device_keys ( + user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + display_name TEXT NOT NULL DEFAULT '', + key_json TEXT NOT NULL, + PRIMARY KEY (user_id, device_id) +); + +-- A cached user's master and self-signing keys. The user-signing key is +-- never sent over federation, so there is no row for one. +CREATE TABLE IF NOT EXISTS keyserver_remote_cross_signing_keys ( + user_id TEXT NOT NULL, + -- master or self_signing. + purpose TEXT NOT NULL, + key_json TEXT NOT NULL, + PRIMARY KEY (user_id, purpose) +); diff --git a/server/internal/storage/remotekeys_test.go b/server/internal/storage/remotekeys_test.go new file mode 100644 index 0000000..31d84a4 --- /dev/null +++ b/server/internal/storage/remotekeys_test.go @@ -0,0 +1,185 @@ +package storage_test + +import ( + "slices" + "testing" + + "github.com/dariuszjaniak/warp/server/internal/storage" +) + +const carol = "@carol:remote.example" + +func TestRemoteKeyCache(t *testing.T) { + for _, driver := range testDrivers { + t.Run(driver, func(t *testing.T) { + ctx := t.Context() + db := migratedKeysDB(t, driver) + keys := db.Keys() + + if _, ok, err := keys.RemoteKeys(ctx, carol); ok || err != nil { + t.Fatalf("RemoteKeys before anything was cached = %v, %v", ok, err) + } + + snapshot := storage.RemoteKeys{ + UserID: carol, StreamID: 5, + Devices: []storage.RemoteDevice{ + {DeviceID: "C2", JSON: []byte(`{"d":2}`)}, + {DeviceID: "C1", DisplayName: "phone", JSON: []byte(`{"d":1}`)}, + }, + CrossSigning: map[string][]byte{ + storage.CrossSigningMaster: []byte(`{"m":1}`), + storage.CrossSigningSelfSigning: []byte(`{"s":1}`), + }, + } + if stored, err := keys.SetRemoteKeys(ctx, snapshot); !stored || err != nil { + t.Fatalf("SetRemoteKeys = %v, %v", stored, err) + } + got, ok, err := keys.RemoteKeys(ctx, carol) + if err != nil || !ok { + t.Fatalf("RemoteKeys = %v, %v", ok, err) + } + if got.StreamID != 5 || got.Stale || len(got.Devices) != 2 || + got.Devices[0].DeviceID != "C1" || got.Devices[0].DisplayName != "phone" || string(got.Devices[1].JSON) != `{"d":2}` || + string(got.CrossSigning[storage.CrossSigningMaster]) != `{"m":1}` || len(got.CrossSigning) != 2 { + t.Errorf("RemoteKeys = %+v", got) + } + + // Caching moves nothing in the device list stream: that is the + // keyserver's call, per update. + if pos, _ := keys.DeviceListPosition(ctx, carol); pos != 0 { + t.Errorf("caching moved the device list to %d", pos) + } + + apply := func(u storage.RemoteDeviceUpdate) storage.RemoteUpdateResult { + t.Helper() + u.UserID = carol + result, err := keys.UpdateRemoteDevice(ctx, u) + if err != nil { + t.Fatalf("UpdateRemoteDevice(%+v): %v", u, err) + } + return result + } + + // Following on, a repeat, and a deletion. + if r := apply(storage.RemoteDeviceUpdate{DeviceID: "C3", JSON: []byte(`{"d":3}`), StreamID: 6, PrevIDs: []int64{5}}); r != storage.RemoteUpdateApplied { + t.Errorf("an update that follows on = %v", r) + } + if r := apply(storage.RemoteDeviceUpdate{DeviceID: "C3", JSON: []byte(`{"d":3}`), StreamID: 6, PrevIDs: []int64{5}}); r != storage.RemoteUpdateDuplicate { + t.Errorf("the same update again = %v", r) + } + if r := apply(storage.RemoteDeviceUpdate{DeviceID: "C2", Deleted: true, StreamID: 7, PrevIDs: []int64{4, 6}}); r != storage.RemoteUpdateApplied { + t.Errorf("a deletion naming the position held among others = %v", r) + } + got, _, _ = keys.RemoteKeys(ctx, carol) + var ids []string + for _, d := range got.Devices { + ids = append(ids, d.DeviceID) + } + if got.StreamID != 7 || !slices.Equal(ids, []string{"C1", "C3"}) { + t.Errorf("after the updates: position %d, devices %v", got.StreamID, ids) + } + + // A gap marks the copy stale at the newest position heard of. + if r := apply(storage.RemoteDeviceUpdate{DeviceID: "C4", JSON: []byte(`{}`), StreamID: 12, PrevIDs: []int64{11}}); r != storage.RemoteUpdateStale { + t.Errorf("an update after a gap = %v", r) + } + got, _, _ = keys.RemoteKeys(ctx, carol) + if !got.Stale || got.StreamID != 12 || len(got.Devices) != 2 { + t.Errorf("after the gap: %+v", got) + } + if r := apply(storage.RemoteDeviceUpdate{DeviceID: "C4", JSON: []byte(`{}`), StreamID: 13, PrevIDs: []int64{12}}); r != storage.RemoteUpdateStale { + t.Errorf("an update to a stale copy = %v", r) + } + if stored, _ := keys.SetRemoteCrossSigningKeys(ctx, carol, map[string][]byte{storage.CrossSigningMaster: []byte(`{}`)}); stored { + t.Error("cross-signing keys were written into a stale copy") + } + + // A snapshot older than the position held is not stored; one at + // or past it replaces the copy whole. + if stored, _ := keys.SetRemoteKeys(ctx, storage.RemoteKeys{UserID: carol, StreamID: 10}); stored { + t.Error("a snapshot older than the copy was stored") + } + if stored, _ := keys.SetRemoteKeys(ctx, storage.RemoteKeys{ + UserID: carol, StreamID: 13, Devices: []storage.RemoteDevice{{DeviceID: "C4", JSON: []byte(`{"d":4}`)}}, + }); !stored { + t.Error("a current snapshot was not stored") + } + got, _, _ = keys.RemoteKeys(ctx, carol) + if got.Stale || got.StreamID != 13 || len(got.Devices) != 1 || len(got.CrossSigning) != 0 { + t.Errorf("after the resync: %+v", got) + } + + // Cross-signing keys are replaced as a set. + if stored, err := keys.SetRemoteCrossSigningKeys(ctx, carol, map[string][]byte{ + storage.CrossSigningMaster: []byte(`{"m":2}`), storage.CrossSigningSelfSigning: nil, + }); !stored || err != nil { + t.Fatalf("SetRemoteCrossSigningKeys = %v, %v", stored, err) + } + got, _, _ = keys.RemoteKeys(ctx, carol) + if string(got.CrossSigning[storage.CrossSigningMaster]) != `{"m":2}` || len(got.CrossSigning) != 1 { + t.Errorf("cross-signing keys = %v", got.CrossSigning) + } + + // An update about a user with nothing cached leaves a stale + // marker, and a snapshot from before the update loses to it. + const dave = "@dave:remote.example" + if r, _ := keys.UpdateRemoteDevice(ctx, storage.RemoteDeviceUpdate{UserID: dave, DeviceID: "V1", StreamID: 3}); r != storage.RemoteUpdateNotCached { + t.Errorf("an update about an uncached user = %v", r) + } + marker, ok, _ := keys.RemoteKeys(ctx, dave) + if !ok || !marker.Stale || marker.StreamID != 3 { + t.Errorf("the marker = %+v, %v", marker, ok) + } + if stored, _ := keys.SetRemoteKeys(ctx, storage.RemoteKeys{UserID: dave, StreamID: 2}); stored { + t.Error("a snapshot from before the update was stored") + } + if stored, _ := keys.SetRemoteCrossSigningKeys(ctx, "@nobody:remote.example", map[string][]byte{}); stored { + t.Error("cross-signing keys were cached for a user with no copy") + } + + users, err := keys.RemoteKeyUsers(ctx) + if err != nil || !slices.Equal(users, []string{carol, dave}) { + t.Errorf("RemoteKeyUsers = %v, %v", users, err) + } + if err := keys.DeleteRemoteKeys(ctx, []string{carol, dave}); err != nil { + t.Fatalf("DeleteRemoteKeys: %v", err) + } + if _, ok, _ := keys.RemoteKeys(ctx, carol); ok { + t.Error("a dropped copy is still there") + } + if users, _ := keys.RemoteKeyUsers(ctx); len(users) != 0 { + t.Errorf("RemoteKeyUsers after dropping everybody = %v", users) + } + }) + } +} + +func TestRemoteKeyCacheFirstUpdate(t *testing.T) { + for _, driver := range testDrivers { + t.Run(driver, func(t *testing.T) { + ctx := t.Context() + keys := migratedKeysDB(t, driver).Keys() + + // A user whose server had never moved their list is cached at + // position 0, and the first update, with no prev_id, follows on. + if _, err := keys.SetRemoteKeys(ctx, storage.RemoteKeys{UserID: carol}); err != nil { + t.Fatal(err) + } + r, err := keys.UpdateRemoteDevice(ctx, storage.RemoteDeviceUpdate{UserID: carol, DeviceID: "C1", JSON: []byte(`{}`), StreamID: 1}) + if err != nil || r != storage.RemoteUpdateApplied { + t.Errorf("the first update = %v, %v", r, err) + } + // A later one with no prev_id does not. + r, _ = keys.UpdateRemoteDevice(ctx, storage.RemoteDeviceUpdate{UserID: carol, DeviceID: "C1", JSON: []byte(`{}`), StreamID: 2}) + if r != storage.RemoteUpdateStale { + t.Errorf("a later update with no prev_id = %v", r) + } + // Nor does one that says a device changed without its keys. + _, _ = keys.SetRemoteKeys(ctx, storage.RemoteKeys{UserID: carol, StreamID: 2}) + r, _ = keys.UpdateRemoteDevice(ctx, storage.RemoteDeviceUpdate{UserID: carol, DeviceID: "C1", StreamID: 3, PrevIDs: []int64{2}}) + if r != storage.RemoteUpdateStale { + t.Errorf("an update with no keys = %v", r) + } + }) + } +} diff --git a/server/internal/storage/sqlite/migrations/0009_remote_keys.sql b/server/internal/storage/sqlite/migrations/0009_remote_keys.sql new file mode 100644 index 0000000..980efde --- /dev/null +++ b/server/internal/storage/sqlite/migrations/0009_remote_keys.sql @@ -0,0 +1,48 @@ +-- The cache of other servers' users' keys. +-- +-- A local client about to send an encrypted message asks for the keys of +-- everyone in the room, and asks again whenever a sync says their device +-- list moved. Answering that for a remote user by asking their server +-- every time puts a federation round trip in front of every message, and +-- turns a busy room into a steady stream of identical queries. So the +-- keys of remote users who share a room with somebody here are kept, and +-- kept current by the m.device_list_update and m.signing_key_update EDUs +-- their server sends whenever they change. +-- +-- These are tables of their own rather than rows in the local key tables. +-- A local user's keys are the source of truth, and every write to them +-- moves the device list stream and goes out over federation; a cached +-- copy of somebody else's is neither, and keeping them apart means no +-- code path that announces a local change can ever announce a copy. + +-- One row per cached user: the marker that the user's device list is held +-- here at all, and the position in their server's device list stream the +-- copy is at. An update names the position before it (prev_id), and is +-- applied only when that is the position held; anything else means an +-- update was missed, and the copy is marked stale until it is fetched +-- again whole. +CREATE TABLE IF NOT EXISTS keyserver_remote_users ( + user_id TEXT NOT NULL PRIMARY KEY, + stream_id BIGINT NOT NULL, + stale BOOLEAN NOT NULL DEFAULT 0 +); + +-- The identity keys of a cached user's devices, as their server sent +-- them, and the display name it gave each. +CREATE TABLE IF NOT EXISTS keyserver_remote_device_keys ( + user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + display_name TEXT NOT NULL DEFAULT '', + key_json TEXT NOT NULL, + PRIMARY KEY (user_id, device_id) +); + +-- A cached user's master and self-signing keys. The user-signing key is +-- never sent over federation, so there is no row for one. +CREATE TABLE IF NOT EXISTS keyserver_remote_cross_signing_keys ( + user_id TEXT NOT NULL, + -- master or self_signing. + purpose TEXT NOT NULL, + key_json TEXT NOT NULL, + PRIMARY KEY (user_id, purpose) +); diff --git a/server/internal/storage/sqlstore/remotekeys.go b/server/internal/storage/sqlstore/remotekeys.go new file mode 100644 index 0000000..de40ae2 --- /dev/null +++ b/server/internal/storage/sqlstore/remotekeys.go @@ -0,0 +1,315 @@ +package sqlstore + +import ( + "context" + "database/sql" + "errors" + "fmt" + "slices" + + "github.com/dariuszjaniak/warp/server/internal/storage" +) + +// The cache of remote users' keys: storage.KeyStore's RemoteKeys and the +// methods after it. +// +// None of these writes touch the device list stream. A remote user's +// position in it is moved by MarkDeviceListsChanged, which the keyserver +// calls for every update that concerns a local client whether or not the +// cache could take the update in: the stream says "ask again", and the +// cache is only where the answer comes from. + +const ( + selectRemoteUserSQL = ` + SELECT stream_id, stale FROM keyserver_remote_users WHERE user_id = ?` + + upsertRemoteUserSQL = ` + INSERT INTO keyserver_remote_users (user_id, stream_id, stale) VALUES (?, ?, ?) + ON CONFLICT (user_id) DO UPDATE SET stream_id = excluded.stream_id, stale = excluded.stale` + + selectRemoteDevicesSQL = ` + SELECT device_id, display_name, key_json FROM keyserver_remote_device_keys + WHERE user_id = ? ORDER BY device_id` + + upsertRemoteDeviceSQL = ` + INSERT INTO keyserver_remote_device_keys (user_id, device_id, display_name, key_json) + VALUES (?, ?, ?, ?) + ON CONFLICT (user_id, device_id) DO UPDATE SET + display_name = excluded.display_name, key_json = excluded.key_json` + + deleteRemoteDeviceSQL = ` + DELETE FROM keyserver_remote_device_keys WHERE user_id = ? AND device_id = ?` + + selectRemoteCrossSigningSQL = ` + SELECT purpose, key_json FROM keyserver_remote_cross_signing_keys WHERE user_id = ?` + + insertRemoteCrossSigningSQL = ` + INSERT INTO keyserver_remote_cross_signing_keys (user_id, purpose, key_json) VALUES (?, ?, ?)` + + deleteRemoteCrossSigningSQL = ` + DELETE FROM keyserver_remote_cross_signing_keys WHERE user_id = ?` + + selectRemoteUsersSQL = ` + SELECT user_id FROM keyserver_remote_users ORDER BY user_id` +) + +// remoteKeyTables is every table a cached user has rows in. +var remoteKeyTables = []string{ + "keyserver_remote_users", "keyserver_remote_device_keys", "keyserver_remote_cross_signing_keys", +} + +// remoteMarker reads a cached user's marker row. +func (s *Keys) remoteMarker(ctx context.Context, db rowQuerier, userID string) (streamID int64, stale, ok bool, err error) { + err = db.QueryRowContext(ctx, s.q(selectRemoteUserSQL), userID).Scan(&streamID, &stale) + switch { + case errors.Is(err, sql.ErrNoRows): + return 0, false, false, nil + case err != nil: + return 0, false, false, fmt.Errorf("read the cached keys of %q: %w", userID, err) + } + return streamID, stale, true, nil +} + +// RemoteKeys implements storage.KeyStore. +func (s *Keys) RemoteKeys(ctx context.Context, userID string) (storage.RemoteKeys, bool, error) { + out := storage.RemoteKeys{UserID: userID, CrossSigning: map[string][]byte{}} + + // One transaction, so that the marker and the keys are one snapshot: + // a copy read halfway through its replacement would be served as + // fresh with half of another one's devices. + var found bool + err := s.tx(ctx, func(tx *sql.Tx) error { + var err error + out.StreamID, out.Stale, found, err = s.remoteMarker(ctx, tx, userID) + if err != nil || !found { + return err + } + if out.Devices, err = s.remoteDevices(ctx, tx, userID); err != nil { + return err + } + out.CrossSigning, err = s.remoteCrossSigning(ctx, tx, userID) + return err + }) + if err != nil || !found { + return storage.RemoteKeys{}, false, err + } + return out, true, nil +} + +func (s *Keys) remoteDevices(ctx context.Context, tx *sql.Tx, userID string) ([]storage.RemoteDevice, error) { + rows, err := tx.QueryContext(ctx, s.q(selectRemoteDevicesSQL), userID) + if err != nil { + return nil, fmt.Errorf("read the cached devices of %q: %w", userID, err) + } + defer func() { _ = rows.Close() }() + + var out []storage.RemoteDevice + for rows.Next() { + var d storage.RemoteDevice + var keyJSON string + if err := rows.Scan(&d.DeviceID, &d.DisplayName, &keyJSON); err != nil { + return nil, fmt.Errorf("read the cached devices of %q: %w", userID, err) + } + d.JSON = []byte(keyJSON) + out = append(out, d) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("read the cached devices of %q: %w", userID, err) + } + return out, nil +} + +func (s *Keys) remoteCrossSigning(ctx context.Context, tx *sql.Tx, userID string) (map[string][]byte, error) { + rows, err := tx.QueryContext(ctx, s.q(selectRemoteCrossSigningSQL), userID) + if err != nil { + return nil, fmt.Errorf("read the cached cross-signing keys of %q: %w", userID, err) + } + defer func() { _ = rows.Close() }() + + out := map[string][]byte{} + for rows.Next() { + var purpose, keyJSON string + if err := rows.Scan(&purpose, &keyJSON); err != nil { + return nil, fmt.Errorf("read the cached cross-signing keys of %q: %w", userID, err) + } + out[purpose] = []byte(keyJSON) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("read the cached cross-signing keys of %q: %w", userID, err) + } + return out, nil +} + +// SetRemoteKeys implements storage.KeyStore. +func (s *Keys) SetRemoteKeys(ctx context.Context, keys storage.RemoteKeys) (bool, error) { + stored := false + err := s.tx(ctx, func(tx *sql.Tx) error { + held, _, ok, err := s.remoteMarker(ctx, tx, keys.UserID) + if err != nil { + return err + } + // A snapshot older than what is held was fetched while an update + // was on its way in. Storing it as fresh would lose that update + // until the next one showed the gap; leaving the copy as it is — + // stale, as the update left it — has the next query fetch again. + if ok && held > keys.StreamID { + return nil + } + + if err := s.deleteRemote(ctx, tx, keys.UserID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, s.q(upsertRemoteUserSQL), keys.UserID, keys.StreamID, false); err != nil { + return fmt.Errorf("cache the keys of %q: %w", keys.UserID, err) + } + for _, d := range keys.Devices { + if _, err := tx.ExecContext(ctx, s.q(upsertRemoteDeviceSQL), keys.UserID, d.DeviceID, d.DisplayName, string(d.JSON)); err != nil { + return fmt.Errorf("cache the keys of %s/%s: %w", keys.UserID, d.DeviceID, err) + } + } + if err := s.insertRemoteCrossSigning(ctx, tx, keys.UserID, keys.CrossSigning); err != nil { + return err + } + stored = true + return nil + }) + return stored, err +} + +// UpdateRemoteDevice implements storage.KeyStore. +func (s *Keys) UpdateRemoteDevice(ctx context.Context, u storage.RemoteDeviceUpdate) (storage.RemoteUpdateResult, error) { + var result storage.RemoteUpdateResult + err := s.tx(ctx, func(tx *sql.Tx) error { + held, stale, ok, err := s.remoteMarker(ctx, tx, u.UserID) + if err != nil { + return err + } + + // A stale copy keeps the highest position it has heard of, so that + // a snapshot fetched before this update cannot be stored over it. + markStale := func(r storage.RemoteUpdateResult) error { + result = r + if _, err := tx.ExecContext(ctx, s.q(upsertRemoteUserSQL), u.UserID, max(held, u.StreamID), true); err != nil { + return fmt.Errorf("mark the cached keys of %q stale: %w", u.UserID, err) + } + return nil + } + + switch { + case !ok: + return markStale(storage.RemoteUpdateNotCached) + case stale: + return markStale(storage.RemoteUpdateStale) + case u.StreamID <= held: + result = storage.RemoteUpdateDuplicate + return nil + case !followsOn(held, u.PrevIDs): + return markStale(storage.RemoteUpdateStale) + case !u.Deleted && u.JSON == nil: + // An update that says a device changed without saying how + // cannot be applied; the copy has to be fetched again. + return markStale(storage.RemoteUpdateStale) + } + + if u.Deleted { + _, err = tx.ExecContext(ctx, s.q(deleteRemoteDeviceSQL), u.UserID, u.DeviceID) + } else { + _, err = tx.ExecContext(ctx, s.q(upsertRemoteDeviceSQL), u.UserID, u.DeviceID, u.DisplayName, string(u.JSON)) + } + if err != nil { + return fmt.Errorf("apply an update to the cached keys of %s/%s: %w", u.UserID, u.DeviceID, err) + } + if _, err := tx.ExecContext(ctx, s.q(upsertRemoteUserSQL), u.UserID, u.StreamID, false); err != nil { + return fmt.Errorf("move the cached keys of %q: %w", u.UserID, err) + } + result = storage.RemoteUpdateApplied + return nil + }) + return result, err +} + +// followsOn reports whether an update with the given prev_id comes right +// after the position held. An update with no prev_id is the first its +// server ever sent for the user, which only follows on from nothing. +func followsOn(held int64, prevIDs []int64) bool { + if len(prevIDs) == 0 { + return held == 0 + } + return slices.Contains(prevIDs, held) +} + +// SetRemoteCrossSigningKeys implements storage.KeyStore. +func (s *Keys) SetRemoteCrossSigningKeys(ctx context.Context, userID string, keys map[string][]byte) (bool, error) { + stored := false + err := s.tx(ctx, func(tx *sql.Tx) error { + _, stale, ok, err := s.remoteMarker(ctx, tx, userID) + if err != nil || !ok || stale { + return err + } + if _, err := tx.ExecContext(ctx, s.q(deleteRemoteCrossSigningSQL), userID); err != nil { + return fmt.Errorf("replace the cached cross-signing keys of %q: %w", userID, err) + } + if err := s.insertRemoteCrossSigning(ctx, tx, userID, keys); err != nil { + return err + } + stored = true + return nil + }) + return stored, err +} + +func (s *Keys) insertRemoteCrossSigning(ctx context.Context, tx *sql.Tx, userID string, keys map[string][]byte) error { + for _, purpose := range sortedKeys(keys) { + if keys[purpose] == nil { + continue + } + if _, err := tx.ExecContext(ctx, s.q(insertRemoteCrossSigningSQL), userID, purpose, string(keys[purpose])); err != nil { + return fmt.Errorf("cache the %s key of %q: %w", purpose, userID, err) + } + } + return nil +} + +// RemoteKeyUsers implements storage.KeyStore. +func (s *Keys) RemoteKeyUsers(ctx context.Context) ([]string, error) { + rows, err := s.db.QueryContext(ctx, s.q(selectRemoteUsersSQL)) + if err != nil { + return nil, fmt.Errorf("read the users with cached keys: %w", err) + } + defer func() { _ = rows.Close() }() + + var out []string + for rows.Next() { + var userID string + if err := rows.Scan(&userID); err != nil { + return nil, fmt.Errorf("read the users with cached keys: %w", err) + } + out = append(out, userID) + } + return out, rows.Err() +} + +// DeleteRemoteKeys implements storage.KeyStore. +func (s *Keys) DeleteRemoteKeys(ctx context.Context, userIDs []string) error { + if len(userIDs) == 0 { + return nil + } + return s.tx(ctx, func(tx *sql.Tx) error { + for _, userID := range dedupe(userIDs) { + if err := s.deleteRemote(ctx, tx, userID); err != nil { + return err + } + } + return nil + }) +} + +// deleteRemote drops every row cached for a user. +func (s *Keys) deleteRemote(ctx context.Context, tx *sql.Tx, userID string) error { + for _, table := range remoteKeyTables { + if _, err := tx.ExecContext(ctx, s.q(`DELETE FROM `+table+` WHERE user_id = ?`), userID); err != nil { + return fmt.Errorf("delete from %s for %q: %w", table, userID, err) + } + } + return nil +}