Notify clients when devices are deleted (#1233)

* Recheck device lists when join/leave events come in

* Add PerformDeviceDeletion

* Notify clients when devices are deleted

* Unbreak things

* Remove debug logging

keyserver/api/api.go

@@ -19,9 +19,13 @@ import (
 	"encoding/json"
 	"strings"
 	"time"
+
+	userapi "github.com/matrix-org/dendrite/userapi/api"
 )
 
 type KeyInternalAPI interface {
+	// SetUserAPI assigns a user API to query when extracting device names.
+	SetUserAPI(i userapi.UserInternalAPI)
 	PerformUploadKeys(ctx context.Context, req *PerformUploadKeysRequest, res *PerformUploadKeysResponse)
 	// PerformClaimKeys claims one-time keys for use in pre-key messages
 	PerformClaimKeys(ctx context.Context, req *PerformClaimKeysRequest, res *PerformClaimKeysResponse)

keyserver/internal/internal.go

@@ -40,6 +40,10 @@ type KeyInternalAPI struct {
 	Producer   *producers.KeyChange
 }
 
+func (a *KeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI) {
+	a.UserAPI = i
+}
+
 func (a *KeyInternalAPI) QueryKeyChanges(ctx context.Context, req *api.QueryKeyChangesRequest, res *api.QueryKeyChangesResponse) {
 	if req.Partition < 0 {
 		req.Partition = a.Producer.DefaultPartition()
@@ -272,6 +276,10 @@ func (a *KeyInternalAPI) uploadDeviceKeys(ctx context.Context, req *api.PerformU
 	var keysToStore []api.DeviceKeys
 	// assert that the user ID / device ID are not lying for each key
 	for _, key := range req.DeviceKeys {
+		if len(key.KeyJSON) == 0 {
+			keysToStore = append(keysToStore, key)
+			continue // deleted keys don't need sanity checking
+		}
 		gotUserID := gjson.GetBytes(key.KeyJSON, "user_id").Str
 		gotDeviceID := gjson.GetBytes(key.KeyJSON, "device_id").Str
 		if gotUserID == key.UserID && gotDeviceID == key.DeviceID {
@@ -286,6 +294,7 @@ func (a *KeyInternalAPI) uploadDeviceKeys(ctx context.Context, req *api.PerformU
 			),
 		})
 	}
+
 	// get existing device keys so we can check for changes
 	existingKeys := make([]api.DeviceKeys, len(keysToStore))
 	for i := range keysToStore {
@@ -358,7 +367,9 @@ func (a *KeyInternalAPI) emitDeviceKeyChanges(existing, new []api.DeviceKeys) er
 	for _, newKey := range new {
 		exists := false
 		for _, existingKey := range existing {
-			if bytes.Equal(existingKey.KeyJSON, newKey.KeyJSON) {
+			// Do not treat the absence of keys as equal, or else we will not emit key changes
+			// when users delete devices which never had a key to begin with as both KeyJSONs are nil.
+			if bytes.Equal(existingKey.KeyJSON, newKey.KeyJSON) && len(existingKey.KeyJSON) > 0 {
 				exists = true
 				break
 			}

keyserver/inthttp/client.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal/httputil"
 	"github.com/matrix-org/dendrite/keyserver/api"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/opentracing/opentracing-go"
 )
 
@@ -52,6 +53,10 @@ type httpKeyInternalAPI struct {
 	httpClient *http.Client
 }
 
+func (h *httpKeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI) {
+	// no-op: doesn't need it
+}
+
 func (h *httpKeyInternalAPI) PerformClaimKeys(
 	ctx context.Context,
 	request *api.PerformClaimKeysRequest,

keyserver/keyserver.go

@@ -23,7 +23,6 @@ import (
 	"github.com/matrix-org/dendrite/keyserver/inthttp"
 	"github.com/matrix-org/dendrite/keyserver/producers"
 	"github.com/matrix-org/dendrite/keyserver/storage"
-	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
 )
@@ -37,7 +36,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) {
 // NewInternalAPI returns a concerete implementation of the internal API. Callers
 // can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
 func NewInternalAPI(
-	cfg *config.Dendrite, fedClient *gomatrixserverlib.FederationClient, userAPI userapi.UserInternalAPI, producer sarama.SyncProducer,
+	cfg *config.Dendrite, fedClient *gomatrixserverlib.FederationClient, producer sarama.SyncProducer,
 ) api.KeyInternalAPI {
 	db, err := storage.NewDatabase(
 		string(cfg.Database.E2EKey),
@@ -55,7 +54,6 @@ func NewInternalAPI(
 		DB:         db,
 		ThisServer: cfg.Matrix.ServerName,
 		FedClient:  fedClient,
-		UserAPI:    userAPI,
 		Producer:   keyChangeProducer,
 	}
 }

keyserver/producers/keychange.go

@@ -63,10 +63,11 @@ func (p *KeyChange) ProduceKeyChanges(keys []api.DeviceKeys) error {
 			return err
 		}
 		logrus.WithFields(logrus.Fields{
-			"user_id":   key.UserID,
-			"device_id": key.DeviceID,
-			"partition": partition,
-			"offset":    offset,
+			"user_id":       key.UserID,
+			"device_id":     key.DeviceID,
+			"partition":     partition,
+			"offset":        offset,
+			"len_key_bytes": len(key.KeyJSON),
 		}).Infof("Produced to key change topic '%s'", p.Topic)
 	}
 	return nil

keyserver/storage/interface.go

@@ -32,7 +32,8 @@ type Database interface {
 	// DeviceKeysJSON populates the KeyJSON for the given keys. If any proided `keys` have a `KeyJSON` already then it will be replaced.
 	DeviceKeysJSON(ctx context.Context, keys []api.DeviceKeys) error
 
-	// StoreDeviceKeys persists the given keys. Keys with the same user ID and device ID will be replaced.
+	// StoreDeviceKeys persists the given keys. Keys with the same user ID and device ID will be replaced. An empty KeyJSON removes the key
+	// for this (user, device).
 	// Returns an error if there was a problem storing the keys.
 	StoreDeviceKeys(ctx context.Context, keys []api.DeviceKeys) error