Cross-signing fixes, notifications via sync, federation (#1974)

* Initial work on signing key update EDUs

* Fix build

* Produce/consume EDUs

* Producer logging

* Only produce key change notifications for local users

* Better naming

* Try to notify sync

* Enable feature

* Use key change topic

* Don't bother verifying signatures, validate key lengths if we can, notifier fixes

* Copyright notices

* Remove tests from whitelist until matrix-org/sytest#1117

* Some review comment fixes

* Update to matrix-org/gomatrixserverlib@f9416ac

* Remove unneeded parameter

clientapi/jsonerror/jsonerror.go

@@ -131,6 +131,12 @@ func InvalidSignature(msg string) *MatrixError {
 	return &MatrixError{"M_INVALID_SIGNATURE", msg}
 }
 
+// InvalidParam is an error that is returned when a parameter was invalid,
+// traditionally with cross-signing.
+func InvalidParam(msg string) *MatrixError {
+	return &MatrixError{"M_INVALID_PARAM", msg}
+}
+
 // MissingParam is an error that is returned when a parameter was incorrect,
 // traditionally with cross-signing.
 func MissingParam(msg string) *MatrixError {

clientapi/routing/key_crosssigning.go

@@ -73,6 +73,11 @@ func UploadCrossSigningDeviceKeys(
 				Code: http.StatusBadRequest,
 				JSON: jsonerror.MissingParam(err.Error()),
 			}
+		case err.IsInvalidParam:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.InvalidParam(err.Error()),
+			}
 		default:
 			return util.JSONResponse{
 				Code: http.StatusBadRequest,
@@ -110,6 +115,11 @@ func UploadCrossSigningDeviceSignatures(req *http.Request, keyserverAPI api.KeyI
 				Code: http.StatusBadRequest,
 				JSON: jsonerror.MissingParam(err.Error()),
 			}
+		case err.IsInvalidParam:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.InvalidParam(err.Error()),
+			}
 		default:
 			return util.JSONResponse{
 				Code: http.StatusBadRequest,

clientapi/routing/routing.go

@@ -65,7 +65,7 @@ func Setup(
 	userInteractiveAuth := auth.NewUserInteractive(accountDB.GetAccountByPassword, cfg)
 
 	unstableFeatures := map[string]bool{
-		//"org.matrix.e2e_cross_signing": true,
+		"org.matrix.e2e_cross_signing": true,
 	}
 	for _, msc := range cfg.MSCs.MSCs {
 		unstableFeatures["org.matrix."+msc] = true

eduserver/api/input.go

@@ -75,6 +75,12 @@ type InputReceiptEventRequest struct {
 // InputReceiptEventResponse is a response to InputReceiptEventRequest
 type InputReceiptEventResponse struct{}
 
+type InputCrossSigningKeyUpdateRequest struct {
+	CrossSigningKeyUpdate `json:"signing_keys"`
+}
+
+type InputCrossSigningKeyUpdateResponse struct{}
+
 // EDUServerInputAPI is used to write events to the typing server.
 type EDUServerInputAPI interface {
 	InputTypingEvent(
@@ -94,4 +100,10 @@ type EDUServerInputAPI interface {
 		request *InputReceiptEventRequest,
 		response *InputReceiptEventResponse,
 	) error
+
+	InputCrossSigningKeyUpdate(
+		ctx context.Context,
+		request *InputCrossSigningKeyUpdateRequest,
+		response *InputCrossSigningKeyUpdateResponse,
+	) error
 }

eduserver/api/output.go

@@ -33,14 +33,6 @@ type OutputTypingEvent struct {
 	ExpireTime *time.Time
 }
 
-// TypingEvent represents a matrix edu event of type 'm.typing'.
-type TypingEvent struct {
-	Type   string `json:"type"`
-	RoomID string `json:"room_id"`
-	UserID string `json:"user_id"`
-	Typing bool   `json:"typing"`
-}
-
 // OutputSendToDeviceEvent is an entry in the send-to-device output kafka log.
 // This contains the full event content, along with the user ID and device ID
 // to which it is destined.
@@ -50,14 +42,6 @@ type OutputSendToDeviceEvent struct {
 	gomatrixserverlib.SendToDeviceEvent
 }
 
-type ReceiptEvent struct {
-	UserID    string                      `json:"user_id"`
-	RoomID    string                      `json:"room_id"`
-	EventID   string                      `json:"event_id"`
-	Type      string                      `json:"type"`
-	Timestamp gomatrixserverlib.Timestamp `json:"timestamp"`
-}
-
 // OutputReceiptEvent is an entry in the receipt output kafka log
 type OutputReceiptEvent struct {
 	UserID    string                      `json:"user_id"`
@@ -67,21 +51,7 @@ type OutputReceiptEvent struct {
 	Timestamp gomatrixserverlib.Timestamp `json:"timestamp"`
 }
 
-// Helper structs for receipts json creation
-type ReceiptMRead struct {
-	User map[string]ReceiptTS `json:"m.read"`
-}
-
-type ReceiptTS struct {
-	TS gomatrixserverlib.Timestamp `json:"ts"`
-}
-
-// FederationSender output
-type FederationReceiptMRead struct {
-	User map[string]FederationReceiptData `json:"m.read"`
-}
-
-type FederationReceiptData struct {
-	Data     ReceiptTS `json:"data"`
-	EventIDs []string  `json:"event_ids"`
+// OutputCrossSigningKeyUpdate is an entry in the signing key update output kafka log
+type OutputCrossSigningKeyUpdate struct {
+	CrossSigningKeyUpdate `json:"signing_keys"`
 }

eduserver/api/types.go

@@ -0,0 +1,59 @@
+// Copyright 2021 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 api
+
+import "github.com/matrix-org/gomatrixserverlib"
+
+const (
+	MSigningKeyUpdate = "m.signing_key_update"
+)
+
+type TypingEvent struct {
+	Type   string `json:"type"`
+	RoomID string `json:"room_id"`
+	UserID string `json:"user_id"`
+	Typing bool   `json:"typing"`
+}
+
+type ReceiptEvent struct {
+	UserID    string                      `json:"user_id"`
+	RoomID    string                      `json:"room_id"`
+	EventID   string                      `json:"event_id"`
+	Type      string                      `json:"type"`
+	Timestamp gomatrixserverlib.Timestamp `json:"timestamp"`
+}
+
+type FederationReceiptMRead struct {
+	User map[string]FederationReceiptData `json:"m.read"`
+}
+
+type FederationReceiptData struct {
+	Data     ReceiptTS `json:"data"`
+	EventIDs []string  `json:"event_ids"`
+}
+
+type ReceiptMRead struct {
+	User map[string]ReceiptTS `json:"m.read"`
+}
+
+type ReceiptTS struct {
+	TS gomatrixserverlib.Timestamp `json:"ts"`
+}
+
+type CrossSigningKeyUpdate struct {
+	MasterKey      *gomatrixserverlib.CrossSigningKey `json:"master_key,omitempty"`
+	SelfSigningKey *gomatrixserverlib.CrossSigningKey `json:"self_signing_key,omitempty"`
+	UserID         string                             `json:"user_id"`
+}

eduserver/eduserver.go

@@ -52,6 +52,7 @@ func NewInternalAPI(
 		OutputTypingEventTopic:       cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent),
 		OutputSendToDeviceEventTopic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent),
 		OutputReceiptEventTopic:      cfg.Matrix.Kafka.TopicFor(config.TopicOutputReceiptEvent),
+		OutputKeyChangeEventTopic:    cfg.Matrix.Kafka.TopicFor(config.TopicOutputKeyChangeEvent),
 		ServerName:                   cfg.Matrix.ServerName,
 	}
 }

eduserver/input/input.go

@@ -24,6 +24,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/eduserver/cache"
+	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
@@ -39,6 +40,8 @@ type EDUServerInputAPI struct {
 	OutputSendToDeviceEventTopic string
 	// The kafka topic to output new receipt events to
 	OutputReceiptEventTopic string
+	// The kafka topic to output new key change events to
+	OutputKeyChangeEventTopic string
 	// kafka producer
 	Producer sarama.SyncProducer
 	// Internal user query API
@@ -77,6 +80,36 @@ func (t *EDUServerInputAPI) InputSendToDeviceEvent(
 	return t.sendToDeviceEvent(ise)
 }
 
+// InputCrossSigningKeyUpdate implements api.EDUServerInputAPI
+func (t *EDUServerInputAPI) InputCrossSigningKeyUpdate(
+	ctx context.Context,
+	request *api.InputCrossSigningKeyUpdateRequest,
+	response *api.InputCrossSigningKeyUpdateResponse,
+) error {
+	eventJSON, err := json.Marshal(&keyapi.DeviceMessage{
+		Type: keyapi.TypeCrossSigningUpdate,
+		OutputCrossSigningKeyUpdate: &api.OutputCrossSigningKeyUpdate{
+			CrossSigningKeyUpdate: request.CrossSigningKeyUpdate,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	logrus.WithFields(logrus.Fields{
+		"user_id": request.UserID,
+	}).Infof("Producing to topic '%s'", t.OutputKeyChangeEventTopic)
+
+	m := &sarama.ProducerMessage{
+		Topic: string(t.OutputKeyChangeEventTopic),
+		Key:   sarama.StringEncoder(request.UserID),
+		Value: sarama.ByteEncoder(eventJSON),
+	}
+
+	_, _, err = t.Producer.SendMessage(m)
+	return err
+}
+
 func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error {
 	ev := &api.TypingEvent{
 		Type:   gomatrixserverlib.MTyping,

eduserver/inthttp/client.go

@@ -12,9 +12,10 @@ import (
 
 // HTTP paths for the internal HTTP APIs
 const (
-	EDUServerInputTypingEventPath       = "/eduserver/input"
-	EDUServerInputSendToDeviceEventPath = "/eduserver/sendToDevice"
-	EDUServerInputReceiptEventPath      = "/eduserver/receipt"
+	EDUServerInputTypingEventPath           = "/eduserver/input"
+	EDUServerInputSendToDeviceEventPath     = "/eduserver/sendToDevice"
+	EDUServerInputReceiptEventPath          = "/eduserver/receipt"
+	EDUServerInputCrossSigningKeyUpdatePath = "/eduserver/crossSigningKeyUpdate"
 )
 
 // NewEDUServerClient creates a EDUServerInputAPI implemented by talking to a HTTP POST API.
@@ -68,3 +69,16 @@ func (h *httpEDUServerInputAPI) InputReceiptEvent(
 	apiURL := h.eduServerURL + EDUServerInputReceiptEventPath
 	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
 }
+
+// InputCrossSigningKeyUpdate implements EDUServerInputAPI
+func (h *httpEDUServerInputAPI) InputCrossSigningKeyUpdate(
+	ctx context.Context,
+	request *api.InputCrossSigningKeyUpdateRequest,
+	response *api.InputCrossSigningKeyUpdateResponse,
+) error {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "InputCrossSigningKeyUpdate")
+	defer span.Finish()
+
+	apiURL := h.eduServerURL + EDUServerInputCrossSigningKeyUpdatePath
+	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+}

eduserver/inthttp/server.go

@@ -51,4 +51,17 @@ func AddRoutes(t api.EDUServerInputAPI, internalAPIMux *mux.Router) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(EDUServerInputCrossSigningKeyUpdatePath,
+		httputil.MakeInternalAPI("inputCrossSigningKeyUpdate", func(req *http.Request) util.JSONResponse {
+			var request api.InputCrossSigningKeyUpdateRequest
+			var response api.InputCrossSigningKeyUpdateResponse
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			if err := t.InputCrossSigningKeyUpdate(req.Context(), &request, &response); err != nil {
+				return util.ErrorResponse(err)
+			}
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 }

federationapi/routing/send.go

@@ -502,6 +502,22 @@ func (t *txnReq) processEDUs(ctx context.Context) {
 					}
 				}
 			}
+		case eduserverAPI.MSigningKeyUpdate:
+			var updatePayload eduserverAPI.CrossSigningKeyUpdate
+			if err := json.Unmarshal(e.Content, &updatePayload); err != nil {
+				util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
+					"user_id": updatePayload.UserID,
+				}).Error("Failed to send signing key update to edu server")
+				continue
+			}
+			inputReq := &eduserverAPI.InputCrossSigningKeyUpdateRequest{
+				CrossSigningKeyUpdate: updatePayload,
+			}
+			inputRes := &eduserverAPI.InputCrossSigningKeyUpdateResponse{}
+			if err := t.eduAPI.InputCrossSigningKeyUpdate(ctx, inputReq, inputRes); err != nil {
+				util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal cross-signing update")
+				continue
+			}
 		default:
 			util.GetLogger(ctx).WithField("type", e.Type).Debug("Unhandled EDU")
 		}

federationapi/routing/send_test.go

@@ -84,6 +84,14 @@ func (o *testEDUProducer) InputReceiptEvent(
 	return nil
 }
 
+func (o *testEDUProducer) InputCrossSigningKeyUpdate(
+	ctx context.Context,
+	request *eduAPI.InputCrossSigningKeyUpdateRequest,
+	response *eduAPI.InputCrossSigningKeyUpdateResponse,
+) error {
+	return nil
+}
+
 type testRoomserverAPI struct {
 	api.RoomserverInternalAPITrace
 	inputRoomEvents            []api.InputRoomEvent