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
This commit is contained in:
Kegsay 2020-07-23 16:41:36 +01:00 committed by GitHub
parent 7b862384a7
commit 98f2f09bb4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 196 additions and 10 deletions

View File

@ -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,
),

View File

@ -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)

View File

@ -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,

View File

@ -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)

View File

@ -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,

View File

@ -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,
}

View File

@ -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.

View File

@ -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"

View File

@ -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)
}

View File

@ -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,
}
}

View File

@ -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
}

View File

@ -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(), &currentstateAPI.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
}