From 98f2f09bb46f8bd126214f7874065d6b311bdeba Mon Sep 17 00:00:00 2001 From: Kegsay Date: Thu, 23 Jul 2020 16:41:36 +0100 Subject: [PATCH] keyserver: produce key change events (#1218) * Produce kafka events when keys are added * Consume key changes in syncapi with TODO markers for handling them and catching up * unbreak tests * Linting --- build/gobind/monolith.go | 2 +- cmd/dendrite-demo-libp2p/main.go | 2 +- cmd/dendrite-demo-yggdrasil/main.go | 2 +- cmd/dendrite-key-server/main.go | 2 +- cmd/dendrite-monolith-server/main.go | 2 +- cmd/dendritejs/main.go | 2 +- internal/config/config.go | 4 ++ internal/config/config_test.go | 2 + keyserver/internal/internal.go | 25 +++++++- keyserver/keyserver.go | 11 +++- keyserver/producers/keychange.go | 57 +++++++++++++++++ syncapi/consumers/keychange.go | 95 ++++++++++++++++++++++++++++ 12 files changed, 196 insertions(+), 10 deletions(-) create mode 100644 keyserver/producers/keychange.go create mode 100644 syncapi/consumers/keychange.go diff --git a/build/gobind/monolith.go b/build/gobind/monolith.go index 7da53084..8e976632 100644 --- a/build/gobind/monolith.go +++ b/build/gobind/monolith.go @@ -155,7 +155,7 @@ func (m *DendriteMonolith) Start() { RoomserverAPI: rsAPI, UserAPI: userAPI, StateAPI: stateAPI, - KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI), + KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI, base.KafkaProducer), ExtPublicRoomsProvider: yggrooms.NewYggdrasilRoomProvider( ygg, fsAPI, federation, ), diff --git a/cmd/dendrite-demo-libp2p/main.go b/cmd/dendrite-demo-libp2p/main.go index c9430543..ed731018 100644 --- a/cmd/dendrite-demo-libp2p/main.go +++ b/cmd/dendrite-demo-libp2p/main.go @@ -186,7 +186,7 @@ func main() { ServerKeyAPI: serverKeyAPI, StateAPI: stateAPI, UserAPI: userAPI, - KeyAPI: keyserver.NewInternalAPI(base.Base.Cfg, federation, userAPI), + KeyAPI: keyserver.NewInternalAPI(base.Base.Cfg, federation, userAPI, base.Base.KafkaProducer), ExtPublicRoomsProvider: provider, } monolith.AddAllPublicRoutes(base.Base.PublicAPIMux) diff --git a/cmd/dendrite-demo-yggdrasil/main.go b/cmd/dendrite-demo-yggdrasil/main.go index 8666e8f5..53cf8ffe 100644 --- a/cmd/dendrite-demo-yggdrasil/main.go +++ b/cmd/dendrite-demo-yggdrasil/main.go @@ -141,7 +141,7 @@ func main() { RoomserverAPI: rsAPI, UserAPI: userAPI, StateAPI: stateAPI, - KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI), + KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI, base.KafkaProducer), //ServerKeyAPI: serverKeyAPI, ExtPublicRoomsProvider: yggrooms.NewYggdrasilRoomProvider( ygg, fsAPI, federation, diff --git a/cmd/dendrite-key-server/main.go b/cmd/dendrite-key-server/main.go index 1aafa144..d58d475a 100644 --- a/cmd/dendrite-key-server/main.go +++ b/cmd/dendrite-key-server/main.go @@ -24,7 +24,7 @@ func main() { base := setup.NewBaseDendrite(cfg, "KeyServer", true) defer base.Close() // nolint: errcheck - intAPI := keyserver.NewInternalAPI(base.Cfg, base.CreateFederationClient(), base.UserAPIClient()) + intAPI := keyserver.NewInternalAPI(base.Cfg, base.CreateFederationClient(), base.UserAPIClient(), base.KafkaProducer) keyserver.AddInternalRoutes(base.InternalAPIMux, intAPI) diff --git a/cmd/dendrite-monolith-server/main.go b/cmd/dendrite-monolith-server/main.go index 80a45c99..b312579c 100644 --- a/cmd/dendrite-monolith-server/main.go +++ b/cmd/dendrite-monolith-server/main.go @@ -119,7 +119,7 @@ func main() { rsImpl.SetFederationSenderAPI(fsAPI) stateAPI := currentstateserver.NewInternalAPI(base.Cfg, base.KafkaConsumer) - keyAPI := keyserver.NewInternalAPI(base.Cfg, federation, userAPI) + keyAPI := keyserver.NewInternalAPI(base.Cfg, federation, userAPI, base.KafkaProducer) monolith := setup.Monolith{ Config: base.Cfg, diff --git a/cmd/dendritejs/main.go b/cmd/dendritejs/main.go index 0bb2dbe9..70e5279c 100644 --- a/cmd/dendritejs/main.go +++ b/cmd/dendritejs/main.go @@ -233,7 +233,7 @@ func main() { RoomserverAPI: rsAPI, StateAPI: stateAPI, UserAPI: userAPI, - KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI), + KeyAPI: keyserver.NewInternalAPI(base.Cfg, federation, userAPI, base.KafkaProducer), //ServerKeyAPI: serverKeyAPI, ExtPublicRoomsProvider: p2pPublicRoomProvider, } diff --git a/internal/config/config.go b/internal/config/config.go index 6a29f52b..657e804f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -159,6 +159,8 @@ type Dendrite struct { OutputTypingEvent Topic `yaml:"output_typing_event"` // Topic for eduserver/api.OutputSendToDeviceEvent events. OutputSendToDeviceEvent Topic `yaml:"output_send_to_device_event"` + // Topic for keyserver when new device keys are added. + OutputKeyChangeEvent Topic `yaml:"output_key_change_event"` } } `yaml:"kafka"` @@ -603,6 +605,8 @@ func (config *Dendrite) checkKafka(configErrs *configErrors, monolithic bool) { checkNotEmpty(configErrs, "kafka.topics.output_room_event", string(config.Kafka.Topics.OutputRoomEvent)) checkNotEmpty(configErrs, "kafka.topics.output_client_data", string(config.Kafka.Topics.OutputClientData)) checkNotEmpty(configErrs, "kafka.topics.output_typing_event", string(config.Kafka.Topics.OutputTypingEvent)) + checkNotEmpty(configErrs, "kafka.topics.output_send_to_device_event", string(config.Kafka.Topics.OutputSendToDeviceEvent)) + checkNotEmpty(configErrs, "kafka.topics.output_key_change_event", string(config.Kafka.Topics.OutputKeyChangeEvent)) } // checkDatabase verifies the parameters database.* are valid. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 6b819872..758d7552 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -46,6 +46,8 @@ kafka: output_room_event: output.room output_client_data: output.client output_typing_event: output.typing + output_send_to_device_event: output.std + output_key_change_event: output.key_change user_updates: output.user database: media_api: "postgresql:///media_api" diff --git a/keyserver/internal/internal.go b/keyserver/internal/internal.go index 174a72dc..d3a6d4ba 100644 --- a/keyserver/internal/internal.go +++ b/keyserver/internal/internal.go @@ -23,6 +23,7 @@ import ( "time" "github.com/matrix-org/dendrite/keyserver/api" + "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" @@ -36,6 +37,7 @@ type KeyInternalAPI struct { ThisServer gomatrixserverlib.ServerName FedClient *gomatrixserverlib.FederationClient UserAPI userapi.UserInternalAPI + Producer *producers.KeyChange } func (a *KeyInternalAPI) PerformUploadKeys(ctx context.Context, req *api.PerformUploadKeysRequest, res *api.PerformUploadKeysResponse) { @@ -290,7 +292,10 @@ func (a *KeyInternalAPI) uploadDeviceKeys(ctx context.Context, req *api.PerformU } return } - a.emitDeviceKeyChanges(existingKeys, keysToStore) + err := a.emitDeviceKeyChanges(existingKeys, keysToStore) + if err != nil { + util.GetLogger(ctx).Errorf("Failed to emitDeviceKeyChanges: %s", err) + } } func (a *KeyInternalAPI) uploadOneTimeKeys(ctx context.Context, req *api.PerformUploadKeysRequest, res *api.PerformUploadKeysResponse) { @@ -332,6 +337,20 @@ func (a *KeyInternalAPI) uploadOneTimeKeys(ctx context.Context, req *api.Perform } -func (a *KeyInternalAPI) emitDeviceKeyChanges(existing, new []api.DeviceKeys) { - // TODO +func (a *KeyInternalAPI) emitDeviceKeyChanges(existing, new []api.DeviceKeys) error { + // find keys in new that are not in existing + var keysAdded []api.DeviceKeys + for _, newKey := range new { + exists := false + for _, existingKey := range existing { + if bytes.Equal(existingKey.KeyJSON, newKey.KeyJSON) { + exists = true + break + } + } + if !exists { + keysAdded = append(keysAdded, newKey) + } + } + return a.Producer.ProduceKeyChanges(keysAdded) } diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go index 2e1ddb6c..47c6a8c3 100644 --- a/keyserver/keyserver.go +++ b/keyserver/keyserver.go @@ -15,11 +15,13 @@ package keyserver import ( + "github.com/Shopify/sarama" "github.com/gorilla/mux" "github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/keyserver/api" "github.com/matrix-org/dendrite/keyserver/internal" "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" @@ -34,7 +36,9 @@ 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) api.KeyInternalAPI { +func NewInternalAPI( + cfg *config.Dendrite, fedClient *gomatrixserverlib.FederationClient, userAPI userapi.UserInternalAPI, producer sarama.SyncProducer, +) api.KeyInternalAPI { db, err := storage.NewDatabase( string(cfg.Database.E2EKey), cfg.DbProperties(), @@ -42,10 +46,15 @@ func NewInternalAPI(cfg *config.Dendrite, fedClient *gomatrixserverlib.Federatio if err != nil { logrus.WithError(err).Panicf("failed to connect to key server database") } + keyChangeProducer := &producers.KeyChange{ + Topic: string(cfg.Kafka.Topics.OutputKeyChangeEvent), + Producer: producer, + } return &internal.KeyInternalAPI{ DB: db, ThisServer: cfg.Matrix.ServerName, FedClient: fedClient, UserAPI: userAPI, + Producer: keyChangeProducer, } } diff --git a/keyserver/producers/keychange.go b/keyserver/producers/keychange.go new file mode 100644 index 00000000..6683a936 --- /dev/null +++ b/keyserver/producers/keychange.go @@ -0,0 +1,57 @@ +// Copyright 2020 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package producers + +import ( + "encoding/json" + + "github.com/Shopify/sarama" + "github.com/matrix-org/dendrite/keyserver/api" + "github.com/sirupsen/logrus" +) + +// KeyChange produces key change events for the sync API and federation sender to consume +type KeyChange struct { + Topic string + Producer sarama.SyncProducer +} + +// ProduceKeyChanges creates new change events for each key +func (p *KeyChange) ProduceKeyChanges(keys []api.DeviceKeys) error { + for _, key := range keys { + var m sarama.ProducerMessage + + value, err := json.Marshal(key) + if err != nil { + return err + } + + m.Topic = string(p.Topic) + m.Key = sarama.StringEncoder(key.UserID) + m.Value = sarama.ByteEncoder(value) + + partition, offset, err := p.Producer.SendMessage(&m) + if err != nil { + return err + } + logrus.WithFields(logrus.Fields{ + "user_id": key.UserID, + "device_id": key.DeviceID, + "partition": partition, + "offset": offset, + }).Infof("Produced to key change topic '%s'", p.Topic) + } + return nil +} diff --git a/syncapi/consumers/keychange.go b/syncapi/consumers/keychange.go new file mode 100644 index 00000000..23961452 --- /dev/null +++ b/syncapi/consumers/keychange.go @@ -0,0 +1,95 @@ +// Copyright 2020 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package consumers + +import ( + "context" + "encoding/json" + + "github.com/Shopify/sarama" + currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api" + "github.com/matrix-org/dendrite/internal" + "github.com/matrix-org/dendrite/internal/config" + "github.com/matrix-org/dendrite/keyserver/api" + "github.com/matrix-org/dendrite/syncapi/storage" + "github.com/matrix-org/gomatrixserverlib" + log "github.com/sirupsen/logrus" +) + +// OutputKeyChangeEventConsumer consumes events that originated in the key server. +type OutputKeyChangeEventConsumer struct { + keyChangeConsumer *internal.ContinualConsumer + db storage.Database + serverName gomatrixserverlib.ServerName // our server name + currentStateAPI currentstateAPI.CurrentStateInternalAPI + // keyAPI api.KeyInternalAPI +} + +// NewOutputKeyChangeEventConsumer creates a new OutputKeyChangeEventConsumer. +// Call Start() to begin consuming from the key server. +func NewOutputKeyChangeEventConsumer( + cfg *config.Dendrite, + kafkaConsumer sarama.Consumer, + currentStateAPI currentstateAPI.CurrentStateInternalAPI, + store storage.Database, +) *OutputKeyChangeEventConsumer { + + consumer := internal.ContinualConsumer{ + Topic: string(cfg.Kafka.Topics.OutputKeyChangeEvent), + Consumer: kafkaConsumer, + PartitionStore: store, + } + + s := &OutputKeyChangeEventConsumer{ + keyChangeConsumer: &consumer, + db: store, + serverName: cfg.Matrix.ServerName, + currentStateAPI: currentStateAPI, + } + + consumer.ProcessMessage = s.onMessage + + return s +} + +// Start consuming from the key server +func (s *OutputKeyChangeEventConsumer) Start() error { + return s.keyChangeConsumer.Start() +} + +func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { + var output api.DeviceKeys + if err := json.Unmarshal(msg.Value, &output); err != nil { + // If the message was invalid, log it and move on to the next message in the stream + log.WithError(err).Error("syncapi: failed to unmarshal key change event from key server") + return err + } + // work out who we need to notify about the new key + var queryRes currentstateAPI.QuerySharedUsersResponse + err := s.currentStateAPI.QuerySharedUsers(context.Background(), ¤tstateAPI.QuerySharedUsersRequest{}, &queryRes) + if err != nil { + log.WithError(err).Error("syncapi: failed to QuerySharedUsers for key change event from key server") + return err + } + // TODO: notify users by waking up streams + return nil +} + +// Catchup returns a list of user IDs of users who have changed their device keys between the partition|offset given and now. +// Returns the new offset for this partition. +func (s *OutputKeyChangeEventConsumer) Catchup(parition int32, offset int64) (userIDs []string, newOffset int, err error) { + //return s.keyAPI.QueryKeyChangeCatchup(ctx, partition, offset) + return +}