Add a SigningKeyUpdate producer
This adds a new stream for signing key updates, this should ensure we
don't lose any updates over federation.
MODIFY
federationapi/federationapi.go
federationapi/federationapi.go
@@ -18,6 +18,8 @@ import (
"time"
"github.com/gorilla/mux"
+ "github.com/sirupsen/logrus"
+
"github.com/matrix-org/dendrite/federationapi/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/consumers"
@@ -33,10 +35,10 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
userapi "github.com/matrix-org/dendrite/userapi/api"
- "github.com/sirupsen/logrus"
- "github.com/matrix-org/dendrite/federationapi/routing"
"github.com/matrix-org/gomatrixserverlib"
+
+ "github.com/matrix-org/dendrite/federationapi/routing"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@@ -66,6 +68,7 @@ func AddPublicRoutes(
TopicTypingEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
TopicPresenceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
TopicDeviceListUpdate: cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
+ TopicSigningKeyUpdate: cfg.Matrix.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
ServerName: cfg.Matrix.ServerName,
UserAPI: userAPI,
}

MODIFY
federationapi/producers/syncapi.go
federationapi/producers/syncapi.go
@@ -21,12 +21,13 @@ import (
"strconv"
"time"
- "github.com/matrix-org/dendrite/setup/jetstream"
- "github.com/matrix-org/dendrite/syncapi/types"
- userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
log "github.com/sirupsen/logrus"
+
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/syncapi/types"
+ userapi "github.com/matrix-org/dendrite/userapi/api"
)
// SyncAPIProducer produces events for the sync API server to consume
@@ -36,6 +37,7 @@ type SyncAPIProducer struct {
TopicTypingEvent string
TopicPresenceEvent string
TopicDeviceListUpdate string
+ TopicSigningKeyUpdate string
JetStream nats.JetStreamContext
ServerName gomatrixserverlib.ServerName
UserAPI userapi.UserInternalAPI
@@ -178,3 +180,15 @@ func (p *SyncAPIProducer) SendDeviceListUpdate(
_, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
return err
}
+
+func (p *SyncAPIProducer) SendSigningKeyUpdate(
+ ctx context.Context, data gomatrixserverlib.RawJSON, origin gomatrixserverlib.ServerName,
+) (err error) {
+ m := nats.NewMsg(p.TopicSigningKeyUpdate)
+ m.Header.Set("origin", string(origin))
+ m.Data = data
+
+ log.Debugf("Sending signing key update")
+ _, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
+ return err
+}

MODIFY
federationapi/routing/send.go
federationapi/routing/send.go
@@ -22,6 +22,11 @@ import (
"sync"
"time"
+ "github.com/matrix-org/gomatrixserverlib"
+ "github.com/matrix-org/util"
+ "github.com/prometheus/client_golang/prometheus"
+ "github.com/sirupsen/logrus"
+
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/producers"
@@ -31,10 +36,6 @@ import (
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
syncTypes "github.com/matrix-org/dendrite/syncapi/types"
- "github.com/matrix-org/gomatrixserverlib"
- "github.com/matrix-org/util"
- "github.com/prometheus/client_golang/prometheus"
- "github.com/sirupsen/logrus"
)
const (
@@ -391,7 +392,7 @@ func (t *txnReq) processEDUs(ctx context.Context) {
}
}
case types.MSigningKeyUpdate:
- if err := t.processSigningKeyUpdate(ctx, e); err != nil {
+ if err := t.producer.SendSigningKeyUpdate(ctx, e.Content, t.Origin); err != nil {
logrus.WithError(err).Errorf("Failed to process signing key update")
}
case gomatrixserverlib.MPresence:
@@ -431,42 +432,6 @@ func (t *txnReq) processPresence(ctx context.Context, e gomatrixserverlib.EDU) e
return nil
}
-func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error {
- var updatePayload keyapi.CrossSigningKeyUpdate
- if err := json.Unmarshal(e.Content, &updatePayload); err != nil {
- util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
- "user_id": updatePayload.UserID,
- }).Debug("Failed to unmarshal signing key update")
- return err
- }
- if _, serverName, err := gomatrixserverlib.SplitID('@', updatePayload.UserID); err != nil {
- return nil
- } else if serverName == t.ourServerName {
- return nil
- } else if serverName != t.Origin {
- return nil
- }
- keys := gomatrixserverlib.CrossSigningKeys{}
- if updatePayload.MasterKey != nil {
- keys.MasterKey = *updatePayload.MasterKey
- }
- if updatePayload.SelfSigningKey != nil {
- keys.SelfSigningKey = *updatePayload.SelfSigningKey
- }
- uploadReq := &keyapi.PerformUploadDeviceKeysRequest{
- CrossSigningKeys: keys,
- UserID: updatePayload.UserID,
- }
- uploadRes := &keyapi.PerformUploadDeviceKeysResponse{}
- if err := t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes); err != nil {
- return err
- }
- if uploadRes.Error != nil {
- return uploadRes.Error
- }
- return nil
-}
-
// processReceiptEvent sends receipt events to JetStream
func (t *txnReq) processReceiptEvent(ctx context.Context,
userID, roomID, receiptType string,

ADD
keyserver/consumers/signingkeyupdate.go
+// Copyright 2022 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/matrix-org/gomatrixserverlib"
+ "github.com/nats-io/nats.go"
+ "github.com/sirupsen/logrus"
+
+ keyapi "github.com/matrix-org/dendrite/keyserver/api"
+ "github.com/matrix-org/dendrite/keyserver/internal"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
+)
+
+// SigningKeyUpdateConsumer consumes signing key updates that came in over federation.
+type SigningKeyUpdateConsumer struct {
+ ctx context.Context
+ jetstream nats.JetStreamContext
+ durable string
+ topic string
+ keyAPI *internal.KeyInternalAPI
+ cfg *config.KeyServer
+}
+
+// NewSigningKeyUpdateConsumer creates a new SigningKeyUpdateConsumer. Call Start() to begin consuming from key servers.
+func NewSigningKeyUpdateConsumer(
+ process *process.ProcessContext,
+ cfg *config.KeyServer,
+ js nats.JetStreamContext,
+ keyAPI *internal.KeyInternalAPI,
+) *SigningKeyUpdateConsumer {
+ return &SigningKeyUpdateConsumer{
+ ctx: process.Context(),
+ jetstream: js,
+ durable: cfg.Matrix.JetStream.Prefixed("KeyServerSigningKeyConsumer"),
+ topic: cfg.Matrix.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
+ keyAPI: keyAPI,
+ cfg: cfg,
+ }
+}
+
+// Start consuming from key servers
+func (t *SigningKeyUpdateConsumer) Start() error {
+ return jetstream.JetStreamConsumer(
+ t.ctx, t.jetstream, t.topic, t.durable, 1,
+ t.onMessage, nats.DeliverAll(), nats.ManualAck(),
+ )
+}
+
+// onMessage is called in response to a message received on the
+// signing key update events topic from the key server.
+func (t *SigningKeyUpdateConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) bool {
+ msg := msgs[0] // Guaranteed to exist if onMessage is called
+ var updatePayload keyapi.CrossSigningKeyUpdate
+ if err := json.Unmarshal(msg.Data, &updatePayload); err != nil {
+ logrus.WithError(err).Errorf("Failed to read from signing key update input topic")
+ return true
+ }
+ origin := gomatrixserverlib.ServerName(msg.Header.Get("origin"))
+ if _, serverName, err := gomatrixserverlib.SplitID('@', updatePayload.UserID); err != nil {
+ logrus.WithError(err).Error("failed to split user id")
+ return true
+ } else if serverName == t.cfg.Matrix.ServerName {
+ logrus.Warn("dropping device key update from ourself")
+ return true
+ } else if serverName != origin {
+ logrus.Warnf("dropping device key update, %s != %s", serverName, origin)
+ return true
+ }
+
+ keys := gomatrixserverlib.CrossSigningKeys{}
+ if updatePayload.MasterKey != nil {
+ keys.MasterKey = *updatePayload.MasterKey
+ }
+ if updatePayload.SelfSigningKey != nil {
+ keys.SelfSigningKey = *updatePayload.SelfSigningKey
+ }
+ uploadReq := &keyapi.PerformUploadDeviceKeysRequest{
+ CrossSigningKeys: keys,
+ UserID: updatePayload.UserID,
+ }
+ uploadRes := &keyapi.PerformUploadDeviceKeysResponse{}
+ if err := t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes); err != nil {
+ logrus.WithError(err).Error("failed to upload device keys")
+ return false
+ }
+ if uploadRes.Error != nil {
+ logrus.WithError(uploadRes.Error).Error("failed to upload device keys")
+ return true
+ }
+
+ return true
+}

MODIFY
keyserver/keyserver.go
keyserver/keyserver.go
@@ -16,6 +16,8 @@ package keyserver
import (
"github.com/gorilla/mux"
+ "github.com/sirupsen/logrus"
+
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/consumers"
@@ -26,7 +28,6 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
- "github.com/sirupsen/logrus"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@@ -72,5 +73,12 @@ func NewInternalAPI(
logrus.WithError(err).Panic("failed to start device list consumer")
}
+ sigConsumer := consumers.NewSigningKeyUpdateConsumer(
+ base.ProcessContext, cfg, js, ap,
+ )
+ if err := sigConsumer.Start(); err != nil {
+ logrus.WithError(err).Panic("failed to start signing key consumer")
+ }
+
return ap
}

MODIFY
setup/jetstream/streams.go
setup/jetstream/streams.go
@@ -17,6 +17,7 @@ const (
var (
InputRoomEvent = "InputRoomEvent"
InputDeviceListUpdate = "InputDeviceListUpdate"
+ InputSigningKeyUpdate = "InputSigningKeyUpdate"
OutputRoomEvent = "OutputRoomEvent"
OutputSendToDeviceEvent = "OutputSendToDeviceEvent"
OutputKeyChangeEvent = "OutputKeyChangeEvent"
@@ -51,6 +52,11 @@ var streams = []*nats.StreamConfig{
Retention: nats.InterestPolicy,
Storage: nats.FileStorage,
},
+ {
+ Name: InputSigningKeyUpdate,
+ Retention: nats.InterestPolicy,
+ Storage: nats.FileStorage,
+ },
{
Name: OutputRoomEvent,
Retention: nats.InterestPolicy,

Add a SigningKeyUpdate producer (#2697)

This adds a new stream for signing key updates, this should ensure we
don't lose any updates over federation.