Merge branch 'master' into add-nats-support

federationsender/consumers/keychange.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/Shopify/sarama"
+	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/federationsender/queue"
 	"github.com/matrix-org/dendrite/federationsender/storage"
 	"github.com/matrix-org/dendrite/internal"
@@ -29,7 +30,7 @@ import (
 	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/setup/process"
 	"github.com/matrix-org/gomatrixserverlib"
-	log "github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus"
 )
 
 // KeyChangeConsumer consumes events that originate in key server.
@@ -81,10 +82,21 @@ func (t *KeyChangeConsumer) Start() error {
 func (t *KeyChangeConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 	var m api.DeviceMessage
 	if err := json.Unmarshal(msg.Value, &m); err != nil {
-		log.WithError(err).Errorf("failed to read device message from key change topic")
+		logrus.WithError(err).Errorf("failed to read device message from key change topic")
 		return nil
 	}
-	logger := log.WithField("user_id", m.UserID)
+	switch m.Type {
+	case api.TypeCrossSigningUpdate:
+		return t.onCrossSigningMessage(m)
+	case api.TypeDeviceKeyUpdate:
+		fallthrough
+	default:
+		return t.onDeviceKeyMessage(m)
+	}
+}
+
+func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) error {
+	logger := logrus.WithField("user_id", m.UserID)
 
 	// only send key change events which originated from us
 	_, originServerName, err := gomatrixserverlib.SplitID('@', m.UserID)
@@ -130,7 +142,51 @@ func (t *KeyChangeConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		return err
 	}
 
-	log.Infof("Sending device list update message to %q", destinations)
+	logrus.Infof("Sending device list update message to %q", destinations)
+	return t.queues.SendEDU(edu, t.serverName, destinations)
+}
+
+func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) error {
+	output := m.CrossSigningKeyUpdate
+	_, host, err := gomatrixserverlib.SplitID('@', output.UserID)
+	if err != nil {
+		logrus.WithError(err).Errorf("fedsender key change consumer: user ID parse failure")
+		return nil
+	}
+	if host != gomatrixserverlib.ServerName(t.serverName) {
+		// Ignore any messages that didn't originate locally, otherwise we'll
+		// end up parroting information we received from other servers.
+		return nil
+	}
+	logger := logrus.WithField("user_id", output.UserID)
+
+	var queryRes roomserverAPI.QueryRoomsForUserResponse
+	err = t.rsAPI.QueryRoomsForUser(context.Background(), &roomserverAPI.QueryRoomsForUserRequest{
+		UserID:         output.UserID,
+		WantMembership: "join",
+	}, &queryRes)
+	if err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined rooms for user")
+		return nil
+	}
+	// send this key change to all servers who share rooms with this user.
+	destinations, err := t.db.GetJoinedHostsForRooms(context.Background(), queryRes.RoomIDs)
+	if err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined hosts for rooms user is in")
+		return nil
+	}
+
+	// Pack the EDU and marshal it
+	edu := &gomatrixserverlib.EDU{
+		Type:   eduserverAPI.MSigningKeyUpdate,
+		Origin: string(t.serverName),
+	}
+	if edu.Content, err = json.Marshal(output); err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to marshal output, dropping")
+		return nil
+	}
+
+	logger.Infof("Sending cross-signing update message to %q", destinations)
 	return t.queues.SendEDU(edu, t.serverName, destinations)
 }