Cross-signing groundwork

clientapi/auth/password.go

@@ -52,7 +52,7 @@ func (t *LoginTypePassword) Login(ctx context.Context, req interface{}) (*Login,
 	if username == "" {
 		return nil, &util.JSONResponse{
 			Code: http.StatusUnauthorized,
-			JSON: jsonerror.BadJSON("'user' must be supplied."),
+			JSON: jsonerror.BadJSON("A username must be supplied."),
 		}
 	}
 	localpart, err := userutil.ParseUsernameParam(username, &t.Config.Matrix.ServerName)
@@ -68,7 +68,7 @@ func (t *LoginTypePassword) Login(ctx context.Context, req interface{}) (*Login,
 		// but that would leak the existence of the user.
 		return nil, &util.JSONResponse{
 			Code: http.StatusForbidden,
-			JSON: jsonerror.Forbidden("username or password was incorrect, or the account does not exist"),
+			JSON: jsonerror.Forbidden("The username or password was incorrect or the account does not exist."),
 		}
 	}
 	return &r.Login, nil

clientapi/auth/user_interactive.go

@@ -220,7 +220,7 @@ func (u *UserInteractive) Verify(ctx context.Context, bodyBytes []byte, device *
 	if !ok {
 		return nil, &util.JSONResponse{
 			Code: http.StatusBadRequest,
-			JSON: jsonerror.BadJSON("unknown auth.type: " + authType),
+			JSON: jsonerror.BadJSON("Unknown auth.type: " + authType),
 		}
 	}
 
@@ -231,7 +231,7 @@ func (u *UserInteractive) Verify(ctx context.Context, bodyBytes []byte, device *
 		if !u.IsSingleStageFlow(authType) {
 			return nil, &util.JSONResponse{
 				Code: http.StatusBadRequest,
-				JSON: jsonerror.Unknown("missing or unknown auth.session"),
+				JSON: jsonerror.Unknown("The auth.session is missing or unknown."),
 			}
 		}
 	}

clientapi/jsonerror/jsonerror.go

@@ -125,6 +125,18 @@ func GuestAccessForbidden(msg string) *MatrixError {
 	return &MatrixError{"M_GUEST_ACCESS_FORBIDDEN", msg}
 }
 
+// InvalidSignature is an error which is returned when the client tries
+// to upload invalid signatures.
+func InvalidSignature(msg string) *MatrixError {
+	return &MatrixError{"M_INVALID_SIGNATURE", msg}
+}
+
+// MissingParam is an error that is returned when a parameter was incorrect,
+// traditionally with cross-signing.
+func MissingParam(msg string) *MatrixError {
+	return &MatrixError{"M_MISSING_PARAM", msg}
+}
+
 type IncompatibleRoomVersionError struct {
 	RoomVersion string `json:"room_version"`
 	Error       string `json:"error"`

clientapi/routing/key_crosssigning.go

@@ -0,0 +1,138 @@
+// 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 routing
+
+import (
+	"net/http"
+
+	"github.com/matrix-org/dendrite/clientapi/auth"
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/clientapi/httputil"
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/setup/config"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/dendrite/userapi/storage/accounts"
+	"github.com/matrix-org/util"
+)
+
+type crossSigningRequest struct {
+	api.PerformUploadDeviceKeysRequest
+	Auth newPasswordAuth `json:"auth"`
+}
+
+func UploadCrossSigningDeviceKeys(
+	req *http.Request, keyserverAPI api.KeyInternalAPI, device *userapi.Device,
+	accountDB accounts.Database, cfg *config.ClientAPI,
+) util.JSONResponse {
+	uploadReq := &crossSigningRequest{}
+	uploadRes := &api.PerformUploadDeviceKeysResponse{}
+
+	resErr := httputil.UnmarshalJSONRequest(req, &uploadReq)
+	if resErr != nil {
+		return *resErr
+	}
+	sessionID := uploadReq.Auth.Session
+	if sessionID == "" {
+		sessionID = util.RandomString(sessionIDLength)
+	}
+	if uploadReq.Auth.Type != authtypes.LoginTypePassword {
+		return util.JSONResponse{
+			Code: http.StatusUnauthorized,
+			JSON: newUserInteractiveResponse(
+				sessionID,
+				[]authtypes.Flow{
+					{
+						Stages: []authtypes.LoginType{authtypes.LoginTypePassword},
+					},
+				},
+				nil,
+			),
+		}
+	}
+	typePassword := auth.LoginTypePassword{
+		GetAccountByPassword: accountDB.GetAccountByPassword,
+		Config:               cfg,
+	}
+	if _, authErr := typePassword.Login(req.Context(), &uploadReq.Auth.PasswordRequest); authErr != nil {
+		return *authErr
+	}
+	AddCompletedSessionStage(sessionID, authtypes.LoginTypePassword)
+
+	uploadReq.UserID = device.UserID
+	keyserverAPI.PerformUploadDeviceKeys(req.Context(), &uploadReq.PerformUploadDeviceKeysRequest, uploadRes)
+
+	if err := uploadRes.Error; err != nil {
+		switch {
+		case err.IsInvalidSignature:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.InvalidSignature(err.Error()),
+			}
+		case err.IsMissingParam:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.MissingParam(err.Error()),
+			}
+		default:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.Unknown(err.Error()),
+			}
+		}
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: struct{}{},
+	}
+}
+
+func UploadCrossSigningDeviceSignatures(req *http.Request, keyserverAPI api.KeyInternalAPI, device *userapi.Device) util.JSONResponse {
+	uploadReq := &api.PerformUploadDeviceSignaturesRequest{}
+	uploadRes := &api.PerformUploadDeviceSignaturesResponse{}
+
+	if err := httputil.UnmarshalJSONRequest(req, &uploadReq.Signatures); err != nil {
+		return *err
+	}
+
+	uploadReq.UserID = device.UserID
+	keyserverAPI.PerformUploadDeviceSignatures(req.Context(), uploadReq, uploadRes)
+
+	if err := uploadRes.Error; err != nil {
+		switch {
+		case err.IsInvalidSignature:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.InvalidSignature(err.Error()),
+			}
+		case err.IsMissingParam:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.MissingParam(err.Error()),
+			}
+		default:
+			return util.JSONResponse{
+				Code: http.StatusBadRequest,
+				JSON: jsonerror.Unknown(err.Error()),
+			}
+		}
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: struct{}{},
+	}
+}

clientapi/routing/keys.go

@@ -100,7 +100,7 @@ func (r *queryKeysRequest) GetTimeout() time.Duration {
 	return time.Duration(r.Timeout) * time.Millisecond
 }
 
-func QueryKeys(req *http.Request, keyAPI api.KeyInternalAPI) util.JSONResponse {
+func QueryKeys(req *http.Request, keyAPI api.KeyInternalAPI, device *userapi.Device) util.JSONResponse {
 	var r queryKeysRequest
 	resErr := httputil.UnmarshalJSONRequest(req, &r)
 	if resErr != nil {
@@ -108,6 +108,7 @@ func QueryKeys(req *http.Request, keyAPI api.KeyInternalAPI) util.JSONResponse {
 	}
 	queryRes := api.QueryKeysResponse{}
 	keyAPI.QueryKeys(req.Context(), &api.QueryKeysRequest{
+		UserID:        device.UserID,
 		UserToDevices: r.DeviceKeys,
 		Timeout:       r.GetTimeout(),
 		// TODO: Token?
@@ -116,6 +117,9 @@ func QueryKeys(req *http.Request, keyAPI api.KeyInternalAPI) util.JSONResponse {
 		Code: 200,
 		JSON: map[string]interface{}{
 			"device_keys":       queryRes.DeviceKeys,
+			"master_keys":       queryRes.MasterKeys,
+			"self_signing_keys": queryRes.SelfSigningKeys,
+			"user_signing_keys": queryRes.UserSigningKeys,
 			"failures":          queryRes.Failures,
 		},
 	}

clientapi/routing/routing.go

@@ -64,7 +64,9 @@ func Setup(
 	rateLimits := newRateLimits(&cfg.RateLimiting)
 	userInteractiveAuth := auth.NewUserInteractive(accountDB.GetAccountByPassword, cfg)
 
-	unstableFeatures := make(map[string]bool)
+	unstableFeatures := map[string]bool{
+		"org.matrix.e2e_cross_signing": true,
+	}
 	for _, msc := range cfg.MSCs.MSCs {
 		unstableFeatures["org.matrix."+msc] = true
 	}
@@ -1066,6 +1068,22 @@ func Setup(
 
 	// Deleting E2E Backup Keys
 
+	// Cross-signing device keys
+
+	postDeviceSigningKeys := httputil.MakeAuthAPI("post_device_signing_keys", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+		return UploadCrossSigningDeviceKeys(req, keyAPI, device, accountDB, cfg)
+	})
+
+	postDeviceSigningSignatures := httputil.MakeAuthAPI("post_device_signing_signatures", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+		return UploadCrossSigningDeviceSignatures(req, keyAPI, device)
+	})
+
+	r0mux.Handle("/keys/device_signing/upload", postDeviceSigningKeys).Methods(http.MethodPost, http.MethodOptions)
+	r0mux.Handle("/keys/signatures/upload", postDeviceSigningSignatures).Methods(http.MethodPost, http.MethodOptions)
+
+	unstableMux.Handle("/keys/device_signing/upload", postDeviceSigningKeys).Methods(http.MethodPost, http.MethodOptions)
+	unstableMux.Handle("/keys/signatures/upload", postDeviceSigningSignatures).Methods(http.MethodPost, http.MethodOptions)
+
 	// Supplying a device ID is deprecated.
 	r0mux.Handle("/keys/upload/{deviceID}",
 		httputil.MakeAuthAPI("keys_upload", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
@@ -1079,7 +1097,7 @@ func Setup(
 	).Methods(http.MethodPost, http.MethodOptions)
 	r0mux.Handle("/keys/query",
 		httputil.MakeAuthAPI("keys_query", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
-			return QueryKeys(req, keyAPI)
+			return QueryKeys(req, keyAPI, device)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 	r0mux.Handle("/keys/claim",

cmd/dendrite-demo-libp2p/main.go

@@ -147,7 +147,7 @@ func main() {
 
 	accountDB := base.Base.CreateAccountsDB()
 	federation := createFederationClient(base)
-	keyAPI := keyserver.NewInternalAPI(&base.Base.Cfg.KeyServer, federation)
+	keyAPI := keyserver.NewInternalAPI(&base.Base, &base.Base.Cfg.KeyServer, federation)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

cmd/dendrite-demo-pinecone/main.go

@@ -179,7 +179,7 @@ func main() {
 		base, federation, rsAPI, keyRing, true,
 	)
 
-	keyAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, fsAPI)
+	keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

cmd/dendrite-demo-yggdrasil/main.go

@@ -102,7 +102,7 @@ func main() {
 	serverKeyAPI := &signing.YggdrasilKeys{}
 	keyRing := serverKeyAPI.KeyRing()
 
-	keyAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, federation)
+	keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

cmd/dendrite-monolith-server/main.go

@@ -111,7 +111,7 @@ func main() {
 	// This is different to rsAPI which can be the http client which doesn't need this dependency
 	rsImpl.SetFederationSenderAPI(fsAPI)
 
-	keyAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, fsAPI)
+	keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, cfg.Derived.ApplicationServices, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

cmd/dendrite-polylith-multi/personalities/keyserver.go

@@ -22,7 +22,7 @@ import (
 
 func KeyServer(base *setup.BaseDendrite, cfg *config.Dendrite) {
 	fsAPI := base.FederationSenderHTTPClient()
-	intAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, fsAPI)
+	intAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
 	intAPI.SetUserAPI(base.UserAPIClient())
 
 	keyserver.AddInternalRoutes(base.InternalAPIMux, intAPI)

cmd/dendritejs-pinecone/main.go

@@ -184,7 +184,7 @@ func startup() {
 
 	accountDB := base.CreateAccountsDB()
 	federation := conn.CreateFederationClient(base, pSessions)
-	keyAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, federation)
+	keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

cmd/dendritejs/main.go

@@ -192,7 +192,7 @@ func main() {
 
 	accountDB := base.CreateAccountsDB()
 	federation := createFederationClient(cfg, node)
-	keyAPI := keyserver.NewInternalAPI(&base.Cfg.KeyServer, federation)
+	keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
 	userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
 	keyAPI.SetUserAPI(userAPI)
 

go.mod

@@ -31,7 +31,7 @@ require (
 	github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20210709140738-b0d1ba599a6d
 	github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20210722110442-5061d6986876
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20210802103449-167a408ac8c9
 	github.com/matrix-org/naffka v0.0.0-20210623111924-14ff508b58e0
 	github.com/matrix-org/pinecone v0.0.0-20210623102758-74f885644c1b
 	github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4

go.sum

@@ -1029,6 +1029,10 @@ github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5d
 github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20210722110442-5061d6986876 h1:6ypwCtgRLK0v/hGWvnd847+KTo9BSkP9N0A4qSniP4E=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20210722110442-5061d6986876/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210730125129-73fcd4bb6a74 h1:w3uth5fTLC85L+kbllaEoUCdv0Y/sdBnaRyMK/C4410=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210730125129-73fcd4bb6a74/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210802103449-167a408ac8c9 h1:4okgxwwqhGffhzUmCKNiql9gULOUDrYrXkrYLeld7Zk=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210802103449-167a408ac8c9/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
 github.com/matrix-org/naffka v0.0.0-20210623111924-14ff508b58e0 h1:HZCzy4oVzz55e+cOMiX/JtSF2UOY1evBl2raaE7ACcU=
 github.com/matrix-org/naffka v0.0.0-20210623111924-14ff508b58e0/go.mod h1:sjyPyRxKM5uw1nD2cJ6O2OxI6GOqyVBfNXqKjBZTBZE=
 github.com/matrix-org/pinecone v0.0.0-20210623102758-74f885644c1b h1:5X5vdWQ13xrNkJVqaJHPsrt7rKkMJH5iac0EtfOuxSg=

keyserver/api/api.go

@@ -32,15 +32,25 @@ type KeyInternalAPI interface {
 	PerformUploadKeys(ctx context.Context, req *PerformUploadKeysRequest, res *PerformUploadKeysResponse)
 	// PerformClaimKeys claims one-time keys for use in pre-key messages
 	PerformClaimKeys(ctx context.Context, req *PerformClaimKeysRequest, res *PerformClaimKeysResponse)
+	PerformUploadDeviceKeys(ctx context.Context, req *PerformUploadDeviceKeysRequest, res *PerformUploadDeviceKeysResponse)
+	PerformUploadDeviceSignatures(ctx context.Context, req *PerformUploadDeviceSignaturesRequest, res *PerformUploadDeviceSignaturesResponse)
 	QueryKeys(ctx context.Context, req *QueryKeysRequest, res *QueryKeysResponse)
 	QueryKeyChanges(ctx context.Context, req *QueryKeyChangesRequest, res *QueryKeyChangesResponse)
 	QueryOneTimeKeys(ctx context.Context, req *QueryOneTimeKeysRequest, res *QueryOneTimeKeysResponse)
 	QueryDeviceMessages(ctx context.Context, req *QueryDeviceMessagesRequest, res *QueryDeviceMessagesResponse)
 }
 
+// Map of purpose -> public key
+type CrossSigningKeyMap map[gomatrixserverlib.CrossSigningKeyPurpose]gomatrixserverlib.Base64Bytes
+
+// Map of user ID -> key ID -> signature
+type CrossSigningSigMap map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes
+
 // KeyError is returned if there was a problem performing/querying the server
 type KeyError struct {
-	Err string
+	Err                string `json:"error"`
+	IsInvalidSignature bool   `json:"is_invalid_signature,omitempty"` // M_INVALID_SIGNATURE
+	IsMissingParam     bool   `json:"is_missing_param,omitempty"`     // M_MISSING_PARAM
 }
 
 func (k *KeyError) Error() string {
@@ -151,7 +161,30 @@ type PerformClaimKeysResponse struct {
 	Error *KeyError
 }
 
+type PerformUploadDeviceKeysRequest struct {
+	gomatrixserverlib.CrossSigningKeys
+	// The user that uploaded the key, should be populated by the clientapi.
+	UserID string `json:"user_id"`
+}
+
+type PerformUploadDeviceKeysResponse struct {
+	Error *KeyError
+}
+
+type PerformUploadDeviceSignaturesRequest struct {
+	Signatures map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice
+	// The user that uploaded the sig, should be populated by the clientapi.
+	UserID string `json:"user_id"`
+}
+
+type PerformUploadDeviceSignaturesResponse struct {
+	Error *KeyError
+}
+
 type QueryKeysRequest struct {
+	// The user ID asking for the keys, e.g. if from a client API request.
+	// Will not be populated if the key request came from federation.
+	UserID string
 	// Maps user IDs to a list of devices
 	UserToDevices map[string][]string
 	Timeout       time.Duration
@@ -162,6 +195,10 @@ type QueryKeysResponse struct {
 	Failures map[string]interface{}
 	// Map of user_id to device_id to device_key
 	DeviceKeys map[string]map[string]json.RawMessage
+	// Maps of user_id to cross signing key
+	MasterKeys      map[string]gomatrixserverlib.CrossSigningKey
+	SelfSigningKeys map[string]gomatrixserverlib.CrossSigningKey
+	UserSigningKeys map[string]gomatrixserverlib.CrossSigningKey
 	// Set if there was a fatal error processing this query
 	Error *KeyError
 }

keyserver/consumers/eduserver.go

@@ -0,0 +1,61 @@
+package consumers
+
+import (
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/storage"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/process"
+
+	"github.com/Shopify/sarama"
+)
+
+type OutputSigningKeyUpdateConsumer struct {
+	eduServerConsumer *internal.ContinualConsumer
+	keyDB             storage.Database
+	keyAPI            api.KeyInternalAPI
+	serverName        string
+}
+
+func NewOutputSigningKeyUpdateConsumer(
+	process *process.ProcessContext,
+	cfg *config.Dendrite,
+	kafkaConsumer sarama.Consumer,
+	keyDB storage.Database,
+	keyAPI api.KeyInternalAPI,
+) *OutputSigningKeyUpdateConsumer {
+	consumer := internal.ContinualConsumer{
+		Process:        process,
+		ComponentName:  "keyserver/eduserver",
+		Topic:          cfg.Global.Kafka.TopicFor(config.TopicOutputSigningKeyUpdate),
+		Consumer:       kafkaConsumer,
+		PartitionStore: keyDB,
+	}
+	s := &OutputSigningKeyUpdateConsumer{
+		eduServerConsumer: &consumer,
+		keyDB:             keyDB,
+		keyAPI:            keyAPI,
+		serverName:        string(cfg.Global.ServerName),
+	}
+	consumer.ProcessMessage = s.onMessage
+
+	return s
+}
+
+func (s *OutputSigningKeyUpdateConsumer) Start() error {
+	return s.eduServerConsumer.Start()
+}
+
+func (s *OutputSigningKeyUpdateConsumer) onMessage(msg *sarama.ConsumerMessage) error {
+	/*
+		var output eduapi.OutputSigningKeyUpdate
+		if err := json.Unmarshal(msg.Value, &output); err != nil {
+			log.WithError(err).Errorf("eduserver output log: message parse failure")
+			return nil
+		}
+		return nil
+	*/
+	return fmt.Errorf("TODO")
+}

keyserver/internal/cross_signing.go

@@ -0,0 +1,333 @@
+package internal
+
+import (
+	"context"
+	"crypto/ed25519"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+)
+
+func sanityCheckKey(key gomatrixserverlib.CrossSigningKey, userID string, purpose gomatrixserverlib.CrossSigningKeyPurpose) error {
+	// Is there exactly one key?
+	if len(key.Keys) != 1 {
+		return fmt.Errorf("should contain exactly one key")
+	}
+
+	// Does the key ID match the key value? Iterates exactly once
+	for keyID, keyData := range key.Keys {
+		b64 := keyData.Encode()
+		tokens := strings.Split(string(keyID), ":")
+		if len(tokens) != 2 {
+			return fmt.Errorf("key ID is incorrectly formatted")
+		}
+		if tokens[1] != b64 {
+			return fmt.Errorf("key ID isn't correct")
+		}
+	}
+
+	// Does the key claim to be from the right user?
+	if userID != key.UserID {
+		return fmt.Errorf("key has a user ID mismatch")
+	}
+
+	// Does the key contain the correct purpose?
+	useful := false
+	for _, usage := range key.Usage {
+		if usage == purpose {
+			useful = true
+		}
+	}
+	if !useful {
+		return fmt.Errorf("key does not contain correct usage purpose")
+	}
+
+	return nil
+}
+
+func (a *KeyInternalAPI) PerformUploadDeviceKeys(ctx context.Context, req *api.PerformUploadDeviceKeysRequest, res *api.PerformUploadDeviceKeysResponse) {
+	hasMasterKey := false
+
+	if len(req.MasterKey.Keys) > 0 {
+		if err := sanityCheckKey(req.MasterKey, req.UserID, gomatrixserverlib.CrossSigningKeyPurposeMaster); err != nil {
+			res.Error = &api.KeyError{
+				Err: "Master key sanity check failed: " + err.Error(),
+			}
+			return
+		}
+		hasMasterKey = true
+	}
+
+	if len(req.SelfSigningKey.Keys) > 0 {
+		if err := sanityCheckKey(req.SelfSigningKey, req.UserID, gomatrixserverlib.CrossSigningKeyPurposeSelfSigning); err != nil {
+			res.Error = &api.KeyError{
+				Err: "Self-signing key sanity check failed: " + err.Error(),
+			}
+			return
+		}
+	}
+
+	if len(req.UserSigningKey.Keys) > 0 {
+		if err := sanityCheckKey(req.UserSigningKey, req.UserID, gomatrixserverlib.CrossSigningKeyPurposeUserSigning); err != nil {
+			res.Error = &api.KeyError{
+				Err: "User-signing key sanity check failed: " + err.Error(),
+			}
+			return
+		}
+	}
+
+	// If the user hasn't given a new master key, then let's go and get their
+	// existing keys from the database.
+	var masterKey gomatrixserverlib.Base64Bytes
+	if !hasMasterKey {
+		existingKeys, err := a.DB.CrossSigningKeysForUser(ctx, req.UserID)
+		if err != nil {
+			res.Error = &api.KeyError{
+				Err: "User-signing key sanity check failed: " + err.Error(),
+			}
+			return
+		}
+
+		masterKey, hasMasterKey = existingKeys[gomatrixserverlib.CrossSigningKeyPurposeMaster]
+		if !hasMasterKey {
+			res.Error = &api.KeyError{
+				Err:            "No master key was found, either in the database or in the request!",
+				IsMissingParam: true,
+			}
+			return
+		}
+	} else {
+		for _, keyData := range req.MasterKey.Keys { // iterates once, see sanityCheckKey
+			masterKey = keyData
+		}
+	}
+	masterKeyID := gomatrixserverlib.KeyID(fmt.Sprintf("ed25519:%s", masterKey.Encode()))
+
+	// Work out which things we need to verify the signatures for.
+	toVerify := make(map[gomatrixserverlib.CrossSigningKeyPurpose]gomatrixserverlib.CrossSigningKey, 3)
+	toStore := api.CrossSigningKeyMap{}
+	if len(req.MasterKey.Keys) > 0 {
+		toVerify[gomatrixserverlib.CrossSigningKeyPurposeMaster] = req.MasterKey
+	}
+	if len(req.SelfSigningKey.Keys) > 0 {
+		toVerify[gomatrixserverlib.CrossSigningKeyPurposeSelfSigning] = req.SelfSigningKey
+	}
+	if len(req.SelfSigningKey.Keys) > 0 {
+		toVerify[gomatrixserverlib.CrossSigningKeyPurposeUserSigning] = req.UserSigningKey
+	}
+	for purpose, key := range toVerify {
+		// Collect together the key IDs we need to verify with. This will include
+		// all of the key IDs specified in the signatures. We don't do this for
+		// the master key because we have no means to verify the signatures - we
+		// instead just need to store them.
+		if purpose != gomatrixserverlib.CrossSigningKeyPurposeMaster {
+			// Marshal the specific key back into JSON so that we can verify the
+			// signature of it.
+			keyJSON, err := json.Marshal(key)
+			if err != nil {
+				res.Error = &api.KeyError{
+					Err: fmt.Sprintf("The JSON of the key section is invalid: %s", err.Error()),
+				}
+				return
+			}
+
+			// Now check if the subkey is signed by the master key.
+			if err := gomatrixserverlib.VerifyJSON(req.UserID, masterKeyID, ed25519.PublicKey(masterKey), keyJSON); err != nil {
+				res.Error = &api.KeyError{
+					Err:                fmt.Sprintf("The %q sub-key failed master key signature verification: %s", purpose, err.Error()),
+					IsInvalidSignature: true,
+				}
+				return
+			}
+		}
+
+		// If we've reached this point then all the signatures are valid so
+		// add the key to the list of keys to store.
+		for _, keyData := range key.Keys { // iterates once, see sanityCheckKey
+			toStore[purpose] = keyData
+		}
+	}
+
+	if err := a.DB.StoreCrossSigningKeysForUser(ctx, req.UserID, toStore); err != nil {
+		res.Error = &api.KeyError{
+			Err: fmt.Sprintf("a.DB.StoreCrossSigningKeysForUser: %s", err),
+		}
+	}
+}
+
+func (a *KeyInternalAPI) PerformUploadDeviceSignatures(ctx context.Context, req *api.PerformUploadDeviceSignaturesRequest, res *api.PerformUploadDeviceSignaturesResponse) {
+	selfSignatures := map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+	otherSignatures := map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+
+	for userID, forUserID := range req.Signatures {
+		for keyID, keyOrDevice := range forUserID {
+			switch key := keyOrDevice.CrossSigningBody.(type) {
+			case *gomatrixserverlib.CrossSigningKey:
+				if key.UserID == req.UserID {
+					if _, ok := selfSignatures[userID]; !ok {
+						selfSignatures[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+					}
+					selfSignatures[userID][keyID] = keyOrDevice
+				} else {
+					if _, ok := selfSignatures[userID]; !ok {
+						otherSignatures[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+					}
+					otherSignatures[userID][keyID] = keyOrDevice
+				}
+
+			case *gomatrixserverlib.DeviceKeys:
+				if key.UserID == req.UserID {
+					if _, ok := selfSignatures[userID]; !ok {
+						selfSignatures[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+					}
+					selfSignatures[userID][keyID] = keyOrDevice
+				} else {
+					if _, ok := selfSignatures[userID]; !ok {
+						otherSignatures[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice{}
+					}
+					otherSignatures[userID][keyID] = keyOrDevice
+				}
+
+			default:
+				continue
+			}
+		}
+	}
+
+	if err := a.processSelfSignatures(ctx, req.UserID, selfSignatures); err != nil {
+		res.Error = &api.KeyError{
+			Err: fmt.Sprintf("a.processSelfSignatures: %s", err),
+		}
+		return
+	}
+
+	if err := a.processOtherSignatures(ctx, req.UserID, otherSignatures); err != nil {
+		res.Error = &api.KeyError{
+			Err: fmt.Sprintf("a.processOtherSignatures: %s", err),
+		}
+		return
+	}
+}
+
+func (a *KeyInternalAPI) processSelfSignatures(
+	ctx context.Context, _ string,
+	signatures map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice,
+) error {
+	// Here we will process:
+	// * The user signing their own devices using their self-signing key
+	// * The user signing their master key using one of their devices
+
+	for targetUserID, forTargetUserID := range signatures {
+		for targetKeyID, signature := range forTargetUserID {
+			switch sig := signature.CrossSigningBody.(type) {
+			case *gomatrixserverlib.CrossSigningKey:
+				for originUserID, forOriginUserID := range sig.Signatures {
+					for originKeyID, originSig := range forOriginUserID {
+						if err := a.DB.StoreCrossSigningSigsForTarget(
+							ctx, originUserID, originKeyID, targetUserID, targetKeyID, originSig,
+						); err != nil {
+							return fmt.Errorf("a.DB.StoreCrossSigningKeysForTarget: %w", err)
+						}
+					}
+				}
+
+			case *gomatrixserverlib.DeviceKeys:
+				for originUserID, forOriginUserID := range sig.Signatures {
+					for originKeyID, originSig := range forOriginUserID {
+						if err := a.DB.StoreCrossSigningSigsForTarget(
+							ctx, originUserID, originKeyID, targetUserID, targetKeyID, originSig,
+						); err != nil {
+							return fmt.Errorf("a.DB.StoreCrossSigningKeysForTarget: %w", err)
+						}
+					}
+				}
+
+			default:
+				return fmt.Errorf("unexpected type assertion")
+			}
+		}
+	}
+
+	return nil
+}
+
+func (a *KeyInternalAPI) processOtherSignatures(
+	ctx context.Context, userID string,
+	signatures map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.CrossSigningForKeyOrDevice,
+) error {
+	// Here we will process:
+	// * A user signing someone else's master keys using their user-signing keys
+
+	return nil
+}
+
+func (a *KeyInternalAPI) crossSigningKeysFromDatabase(
+	ctx context.Context, req *api.QueryKeysRequest, res *api.QueryKeysResponse,
+) {
+	for userID := range req.UserToDevices {
+		keys, err := a.DB.CrossSigningKeysForUser(ctx, userID)
+		if err != nil {
+			logrus.WithError(err).Errorf("Failed to get cross-signing keys for user %q", userID)
+			continue
+		}
+
+		for keyType, keyData := range keys {
+			b64 := keyData.Encode()
+			keyID := gomatrixserverlib.KeyID("ed25519:" + b64)
+			key := gomatrixserverlib.CrossSigningKey{
+				UserID: userID,
+				Usage: []gomatrixserverlib.CrossSigningKeyPurpose{
+					keyType,
+				},
+				Keys: map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{
+					keyID: keyData,
+				},
+			}
+
+			sigs, err := a.DB.CrossSigningSigsForTarget(ctx, userID, keyID)
+			if err != nil {
+				logrus.WithError(err).Errorf("Failed to get cross-signing signatures for user %q key %q", userID, keyID)
+				continue
+			}
+
+			appendSignature := func(originUserID string, originKeyID gomatrixserverlib.KeyID, signature gomatrixserverlib.Base64Bytes) {
+				if key.Signatures == nil {
+					key.Signatures = api.CrossSigningSigMap{}
+				}
+				if _, ok := key.Signatures[originUserID]; !ok {
+					key.Signatures[originUserID] = make(map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes)
+				}
+				key.Signatures[originUserID][originKeyID] = signature
+			}
+
+			for originUserID, forOrigin := range sigs {
+				for originKeyID, signature := range forOrigin {
+					switch {
+					case req.UserID != "" && originUserID == req.UserID:
+						// Include signatures that we created
+						appendSignature(originUserID, originKeyID, signature)
+					case originUserID == userID:
+						// Include signatures that were created by the person whose key
+						// we are processing
+						appendSignature(originUserID, originKeyID, signature)
+					}
+				}
+			}
+
+			switch keyType {
+			case gomatrixserverlib.CrossSigningKeyPurposeMaster:
+				res.MasterKeys[userID] = key
+
+			case gomatrixserverlib.CrossSigningKeyPurposeSelfSigning:
+				res.SelfSigningKeys[userID] = key
+
+			case gomatrixserverlib.CrossSigningKeyPurposeUserSigning:
+				res.UserSigningKeys[userID] = key
+			}
+		}
+	}
+}

keyserver/internal/internal.go

@@ -221,9 +221,17 @@ func (a *KeyInternalAPI) QueryDeviceMessages(ctx context.Context, req *api.Query
 
 func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysRequest, res *api.QueryKeysResponse) {
 	res.DeviceKeys = make(map[string]map[string]json.RawMessage)
+	res.MasterKeys = make(map[string]gomatrixserverlib.CrossSigningKey)
+	res.SelfSigningKeys = make(map[string]gomatrixserverlib.CrossSigningKey)
+	res.UserSigningKeys = make(map[string]gomatrixserverlib.CrossSigningKey)
 	res.Failures = make(map[string]interface{})
+
+	// get cross-signing keys from the database
+	a.crossSigningKeysFromDatabase(ctx, req, res)
+
 	// make a map from domain to device keys
 	domainToDeviceKeys := make(map[string]map[string][]string)
+	domainToCrossSigningKeys := make(map[string]map[string]struct{})
 	for userID, deviceIDs := range req.UserToDevices {
 		_, serverName, err := gomatrixserverlib.SplitID('@', userID)
 		if err != nil {
@@ -274,14 +282,40 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques
 			domainToDeviceKeys[domain] = make(map[string][]string)
 			domainToDeviceKeys[domain][userID] = append(domainToDeviceKeys[domain][userID], deviceIDs...)
 		}
+		// work out if our cross-signing request for this user was
+		// satisfied, if not add them to the list of things to fetch
+		if _, ok := res.MasterKeys[userID]; !ok {
+			if _, ok := domainToCrossSigningKeys[domain]; !ok {
+				domainToCrossSigningKeys[domain] = make(map[string]struct{})
+			}
+			domainToCrossSigningKeys[domain][userID] = struct{}{}
+		}
+		if _, ok := res.SelfSigningKeys[userID]; !ok {
+			if _, ok := domainToCrossSigningKeys[domain]; !ok {
+				domainToCrossSigningKeys[domain] = make(map[string]struct{})
+			}
+			domainToCrossSigningKeys[domain][userID] = struct{}{}
+		}
 	}
 
 	// attempt to satisfy key queries from the local database first as we should get device updates pushed to us
 	domainToDeviceKeys = a.remoteKeysFromDatabase(ctx, res, domainToDeviceKeys)
-	if len(domainToDeviceKeys) == 0 {
+	if len(domainToDeviceKeys) == 0 && len(domainToCrossSigningKeys) == 0 {
 		return // nothing to query
 	}
 
+	// add in any cross-signing requests that need to be made to the list
+	for domain, forDomain := range domainToCrossSigningKeys {
+		for userID := range forDomain {
+			if _, ok := domainToDeviceKeys[domain]; !ok {
+				domainToDeviceKeys[domain] = make(map[string][]string)
+			}
+			if _, ok := domainToDeviceKeys[domain][userID]; !ok {
+				domainToDeviceKeys[domain][userID] = []string{}
+			}
+		}
+	}
+
 	// perform key queries for remote devices
 	a.queryRemoteKeys(ctx, req.Timeout, res, domainToDeviceKeys)
 }
@@ -344,6 +378,23 @@ func (a *KeyInternalAPI) queryRemoteKeys(
 				res.DeviceKeys[userID][deviceID] = keyJSON
 			}
 		}
+
+		for userID, body := range result.MasterKeys {
+			switch b := body.CrossSigningBody.(type) {
+			case *gomatrixserverlib.CrossSigningKey:
+				res.MasterKeys[userID] = *b
+			}
+		}
+
+		for userID, body := range result.SelfSigningKeys {
+			switch b := body.CrossSigningBody.(type) {
+			case *gomatrixserverlib.CrossSigningKey:
+				res.SelfSigningKeys[userID] = *b
+			}
+		}
+
+		// TODO: do we want to persist these somewhere now
+		// that we have fetched them?
 	}
 }
 
@@ -362,7 +413,7 @@ func (a *KeyInternalAPI) queryRemoteKeysOnServer(
 	for userID, deviceIDs := range devKeys {
 		if len(deviceIDs) == 0 {
 			userIDsForAllDevices = append(userIDsForAllDevices, userID)
-			delete(devKeys, userID)
+			//delete(devKeys, userID)
 		}
 	}
 	for _, userID := range userIDsForAllDevices {
@@ -393,9 +444,6 @@ func (a *KeyInternalAPI) queryRemoteKeysOnServer(
 			continue
 		}
 	}
-	if len(devKeys) == 0 {
-		return
-	}
 	queryKeysResp, err := a.FedClient.QueryKeys(fedCtx, gomatrixserverlib.ServerName(serverName), devKeys)
 	if err == nil {
 		resultCh <- &queryKeysResp

keyserver/inthttp/client.go

@@ -30,6 +30,8 @@ const (
 	InputDeviceListUpdatePath         = "/keyserver/inputDeviceListUpdate"
 	PerformUploadKeysPath             = "/keyserver/performUploadKeys"
 	PerformClaimKeysPath              = "/keyserver/performClaimKeys"
+	PerformUploadDeviceKeysPath       = "/keyserver/performUploadDeviceKeys"
+	PerformUploadDeviceSignaturesPath = "/keyserver/performUploadDeviceSignatures"
 	QueryKeysPath                     = "/keyserver/queryKeys"
 	QueryKeyChangesPath               = "/keyserver/queryKeyChanges"
 	QueryOneTimeKeysPath              = "/keyserver/queryOneTimeKeys"
@@ -175,3 +177,37 @@ func (h *httpKeyInternalAPI) QueryKeyChanges(
 		}
 	}
 }
+
+func (h *httpKeyInternalAPI) PerformUploadDeviceKeys(
+	ctx context.Context,
+	request *api.PerformUploadDeviceKeysRequest,
+	response *api.PerformUploadDeviceKeysResponse,
+) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "PerformUploadDeviceKeys")
+	defer span.Finish()
+
+	apiURL := h.apiURL + PerformUploadDeviceKeysPath
+	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+	if err != nil {
+		response.Error = &api.KeyError{
+			Err: err.Error(),
+		}
+	}
+}
+
+func (h *httpKeyInternalAPI) PerformUploadDeviceSignatures(
+	ctx context.Context,
+	request *api.PerformUploadDeviceSignaturesRequest,
+	response *api.PerformUploadDeviceSignaturesResponse,
+) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "PerformUploadDeviceSignatures")
+	defer span.Finish()
+
+	apiURL := h.apiURL + PerformUploadDeviceSignaturesPath
+	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+	if err != nil {
+		response.Error = &api.KeyError{
+			Err: err.Error(),
+		}
+	}
+}

keyserver/inthttp/server.go

@@ -58,6 +58,28 @@ func AddRoutes(internalAPIMux *mux.Router, s api.KeyInternalAPI) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(PerformUploadDeviceKeysPath,
+		httputil.MakeInternalAPI("performUploadDeviceKeys", func(req *http.Request) util.JSONResponse {
+			request := api.PerformUploadDeviceKeysRequest{}
+			response := api.PerformUploadDeviceKeysResponse{}
+			if err := json.NewDecoder(req.Body).Decode(&request.CrossSigningKeys); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			s.PerformUploadDeviceKeys(req.Context(), &request, &response)
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
+	internalAPIMux.Handle(PerformUploadDeviceSignaturesPath,
+		httputil.MakeInternalAPI("performUploadDeviceSignatures", func(req *http.Request) util.JSONResponse {
+			request := api.PerformUploadDeviceSignaturesRequest{}
+			response := api.PerformUploadDeviceSignaturesResponse{}
+			if err := json.NewDecoder(req.Body).Decode(&request.Signatures); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			s.PerformUploadDeviceSignatures(req.Context(), &request, &response)
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 	internalAPIMux.Handle(QueryKeysPath,
 		httputil.MakeInternalAPI("queryKeys", func(req *http.Request) util.JSONResponse {
 			request := api.QueryKeysRequest{}

keyserver/keyserver.go

@@ -18,10 +18,12 @@ import (
 	"github.com/gorilla/mux"
 	fedsenderapi "github.com/matrix-org/dendrite/federationsender/api"
 	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/consumers"
 	"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"
+	"github.com/matrix-org/dendrite/setup"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/kafka"
 	"github.com/sirupsen/logrus"
@@ -36,9 +38,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.KeyServer, fedClient fedsenderapi.FederationClient,
+	base *setup.BaseDendrite, cfg *config.KeyServer, fedClient fedsenderapi.FederationClient,
 ) api.KeyInternalAPI {
-	_, producer := kafka.SetupConsumerProducer(&cfg.Matrix.Kafka)
+	consumer, producer := kafka.SetupConsumerProducer(&cfg.Matrix.Kafka)
 
 	db, err := storage.NewDatabase(&cfg.Database)
 	if err != nil {
@@ -55,11 +57,21 @@ func NewInternalAPI(
 			logrus.WithError(err).Panicf("failed to start device list updater")
 		}
 	}()
-	return &internal.KeyInternalAPI{
+
+	ap := &internal.KeyInternalAPI{
 		DB:         db,
 		ThisServer: cfg.Matrix.ServerName,
 		FedClient:  fedClient,
 		Producer:   keyChangeProducer,
 		Updater:    updater,
 	}
+
+	keyconsumer := consumers.NewOutputSigningKeyUpdateConsumer(
+		base.ProcessContext, base.Cfg, consumer, db, ap,
+	)
+	if err := keyconsumer.Start(); err != nil {
+		logrus.WithError(err).Panicf("failed to start keyserver EDU server consumer")
+	}
+
+	return ap
 }

keyserver/storage/interface.go

@@ -18,11 +18,14 @@ import (
 	"context"
 	"encoding/json"
 
+	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
 type Database interface {
+	internal.PartitionStorer
+
 	// ExistingOneTimeKeys returns a map of keyIDWithAlgorithm to key JSON for the given parameters. If no keys exist with this combination
 	// of user/device/key/algorithm 4-uple then it is omitted from the map. Returns an error when failing to communicate with the database.
 	ExistingOneTimeKeys(ctx context.Context, userID, deviceID string, keyIDsWithAlgorithms []string) (map[string]json.RawMessage, error)
@@ -73,4 +76,10 @@ type Database interface {
 
 	// MarkDeviceListStale sets the stale bit for this user to isStale.
 	MarkDeviceListStale(ctx context.Context, userID string, isStale bool) error
+
+	CrossSigningKeysForUser(ctx context.Context, userID string) (api.CrossSigningKeyMap, error)
+	CrossSigningSigsForTarget(ctx context.Context, targetUserID string, targetKeyID gomatrixserverlib.KeyID) (api.CrossSigningSigMap, error)
+
+	StoreCrossSigningKeysForUser(ctx context.Context, userID string, keyMap api.CrossSigningKeyMap) error
+	StoreCrossSigningSigsForTarget(ctx context.Context, originUserID string, originKeyID gomatrixserverlib.KeyID, targetUserID string, targetKeyID gomatrixserverlib.KeyID, signature gomatrixserverlib.Base64Bytes) error
 }

keyserver/storage/postgres/cross_signing_keys_table.go

@@ -0,0 +1,97 @@
+// 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 postgres
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/storage/tables"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+var crossSigningKeysSchema = `
+CREATE TABLE IF NOT EXISTS keyserver_cross_signing_keys (
+    user_id TEXT NOT NULL,
+	key_type TEXT NOT NULL,
+	key_data TEXT NOT NULL,
+	PRIMARY KEY (user_id, key_type)
+);
+`
+
+const selectCrossSigningKeysForUserSQL = "" +
+	"SELECT key_type, key_data FROM keyserver_cross_signing_keys" +
+	" WHERE user_id = $1"
+
+const insertCrossSigningKeysForUserSQL = "" +
+	"INSERT INTO keyserver_cross_signing_keys (user_id, key_type, key_data)" +
+	" VALUES($1, $2, $3)" +
+	" ON CONFLICT (user_id, key_type) DO UPDATE SET key_data = $3"
+
+type crossSigningKeysStatements struct {
+	db                                *sql.DB
+	selectCrossSigningKeysForUserStmt *sql.Stmt
+	insertCrossSigningKeysForUserStmt *sql.Stmt
+}
+
+func NewPostgresCrossSigningKeysTable(db *sql.DB) (tables.CrossSigningKeys, error) {
+	s := &crossSigningKeysStatements{
+		db: db,
+	}
+	_, err := db.Exec(crossSigningKeysSchema)
+	if err != nil {
+		return nil, err
+	}
+	if s.selectCrossSigningKeysForUserStmt, err = db.Prepare(selectCrossSigningKeysForUserSQL); err != nil {
+		return nil, err
+	}
+	if s.insertCrossSigningKeysForUserStmt, err = db.Prepare(insertCrossSigningKeysForUserSQL); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *crossSigningKeysStatements) SelectCrossSigningKeysForUser(
+	ctx context.Context, txn *sql.Tx, userID string,
+) (r api.CrossSigningKeyMap, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningKeysForUserStmt).QueryContext(ctx, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectCrossSigningKeysForUserStmt: rows.close() failed")
+	r = api.CrossSigningKeyMap{}
+	for rows.Next() {
+		var keyType gomatrixserverlib.CrossSigningKeyPurpose
+		var keyData gomatrixserverlib.Base64Bytes
+		if err := rows.Scan(&keyType, &keyData); err != nil {
+			return nil, err
+		}
+		r[keyType] = keyData
+	}
+	return
+}
+
+func (s *crossSigningKeysStatements) InsertCrossSigningKeysForUser(
+	ctx context.Context, txn *sql.Tx, userID string, keyType gomatrixserverlib.CrossSigningKeyPurpose, keyData gomatrixserverlib.Base64Bytes,
+) error {
+	if _, err := sqlutil.TxStmt(txn, s.insertCrossSigningKeysForUserStmt).ExecContext(ctx, userID, keyType, keyData); err != nil {
+		return fmt.Errorf("s.insertCrossSigningKeysForUserStmt: %w", err)
+	}
+	return nil
+}

keyserver/storage/postgres/cross_signing_sigs_table.go

@@ -0,0 +1,106 @@
+// 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 postgres
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/storage/tables"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+var crossSigningSigsSchema = `
+CREATE TABLE IF NOT EXISTS keyserver_cross_signing_sigs (
+    origin_user_id TEXT NOT NULL,
+	origin_key_id TEXT NOT NULL,
+	target_user_id TEXT NOT NULL,
+	target_key_id TEXT NOT NULL,
+	signature TEXT NOT NULL,
+	PRIMARY KEY (origin_user_id, target_user_id, target_key_id)
+);
+`
+
+const selectCrossSigningSigsForTargetSQL = "" +
+	"SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" +
+	" WHERE target_user_id = $1 AND target_key_id = $2"
+
+const insertCrossSigningSigsForTargetSQL = "" +
+	"INSERT INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" +
+	" VALUES($1, $2, $3, $4, $5)" +
+	" ON CONFLICT (origin_user_id, target_user_id, target_key_id) DO UPDATE SET (origin_key_id, signature) = ($2, $5)"
+
+type crossSigningSigsStatements struct {
+	db                                  *sql.DB
+	selectCrossSigningSigsForTargetStmt *sql.Stmt
+	insertCrossSigningSigsForTargetStmt *sql.Stmt
+}
+
+func NewPostgresCrossSigningSigsTable(db *sql.DB) (tables.CrossSigningSigs, error) {
+	s := &crossSigningSigsStatements{
+		db: db,
+	}
+	_, err := db.Exec(crossSigningSigsSchema)
+	if err != nil {
+		return nil, err
+	}
+	if s.selectCrossSigningSigsForTargetStmt, err = db.Prepare(selectCrossSigningSigsForTargetSQL); err != nil {
+		return nil, err
+	}
+	if s.insertCrossSigningSigsForTargetStmt, err = db.Prepare(insertCrossSigningSigsForTargetSQL); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *crossSigningSigsStatements) SelectCrossSigningSigsForTarget(
+	ctx context.Context, txn *sql.Tx, targetUserID string, targetKeyID gomatrixserverlib.KeyID,
+) (r api.CrossSigningSigMap, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, targetUserID, targetKeyID)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectCrossSigningSigsForTargetStmt: rows.close() failed")
+	r = api.CrossSigningSigMap{}
+	for rows.Next() {
+		var userID string
+		var keyID gomatrixserverlib.KeyID
+		var signature gomatrixserverlib.Base64Bytes
+		if err := rows.Scan(&userID, &keyID, &signature); err != nil {
+			return nil, err
+		}
+		if _, ok := r[userID]; !ok {
+			r[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{}
+		}
+		r[userID][keyID] = signature
+	}
+	return
+}
+
+func (s *crossSigningSigsStatements) InsertCrossSigningSigsForTarget(
+	ctx context.Context, txn *sql.Tx,
+	originUserID string, originKeyID gomatrixserverlib.KeyID,
+	targetUserID string, targetKeyID gomatrixserverlib.KeyID,
+	signature gomatrixserverlib.Base64Bytes,
+) error {
+	if _, err := sqlutil.TxStmt(txn, s.insertCrossSigningSigsForTargetStmt).ExecContext(ctx, originUserID, originKeyID, targetUserID, targetKeyID, signature); err != nil {
+		return fmt.Errorf("s.insertCrossSigningSigsForTargetStmt: %w", err)
+	}
+	return nil
+}

keyserver/storage/postgres/storage.go

@@ -43,12 +43,26 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*shared.Database, error)
 	if err != nil {
 		return nil, err
 	}
-	return &shared.Database{
+	csk, err := NewPostgresCrossSigningKeysTable(db)
+	if err != nil {
+		return nil, err
+	}
+	css, err := NewPostgresCrossSigningSigsTable(db)
+	if err != nil {
+		return nil, err
+	}
+	d := &shared.Database{
 		DB:                    db,
 		Writer:                sqlutil.NewDummyWriter(),
 		OneTimeKeysTable:      otk,
 		DeviceKeysTable:       dk,
 		KeyChangesTable:       kc,
 		StaleDeviceListsTable: sdl,
-	}, nil
+		CrossSigningKeysTable: csk,
+		CrossSigningSigsTable: css,
+	}
+	if err = d.PartitionOffsetStatements.Prepare(db, d.Writer, "keyserver"); err != nil {
+		return nil, err
+	}
+	return d, nil
 }

keyserver/storage/shared/storage.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/keyserver/api"
@@ -32,6 +33,9 @@ type Database struct {
 	DeviceKeysTable       tables.DeviceKeys
 	KeyChangesTable       tables.KeyChanges
 	StaleDeviceListsTable tables.StaleDeviceLists
+	CrossSigningKeysTable tables.CrossSigningKeys
+	CrossSigningSigsTable tables.CrossSigningSigs
+	sqlutil.PartitionOffsetStatements
 }
 
 func (d *Database) ExistingOneTimeKeys(ctx context.Context, userID, deviceID string, keyIDsWithAlgorithms []string) (map[string]json.RawMessage, error) {
@@ -152,3 +156,40 @@ func (d *Database) MarkDeviceListStale(ctx context.Context, userID string, isSta
 		return d.StaleDeviceListsTable.InsertStaleDeviceList(ctx, userID, isStale)
 	})
 }
+
+// CrossSigningKeysForUser returns the latest known cross-signing keys for a user, if any.
+func (d *Database) CrossSigningKeysForUser(ctx context.Context, userID string) (api.CrossSigningKeyMap, error) {
+	return d.CrossSigningKeysTable.SelectCrossSigningKeysForUser(ctx, nil, userID)
+}
+
+// CrossSigningSigsForTarget returns the signatures for a given user's key ID, if any.
+func (d *Database) CrossSigningSigsForTarget(ctx context.Context, targetUserID string, targetKeyID gomatrixserverlib.KeyID) (api.CrossSigningSigMap, error) {
+	return d.CrossSigningSigsTable.SelectCrossSigningSigsForTarget(ctx, nil, targetUserID, targetKeyID)
+}
+
+// StoreCrossSigningKeysForUser stores the latest known cross-signing keys for a user.
+func (d *Database) StoreCrossSigningKeysForUser(ctx context.Context, userID string, keyMap api.CrossSigningKeyMap) error {
+	return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		for keyType, keyData := range keyMap {
+			if err := d.CrossSigningKeysTable.InsertCrossSigningKeysForUser(ctx, txn, userID, keyType, keyData); err != nil {
+				return fmt.Errorf("d.CrossSigningKeysTable.InsertCrossSigningKeysForUser: %w", err)
+			}
+		}
+		return nil
+	})
+}
+
+// StoreCrossSigningSigsForTarget stores a signature for a target user ID and key/dvice.
+func (d *Database) StoreCrossSigningSigsForTarget(
+	ctx context.Context,
+	originUserID string, originKeyID gomatrixserverlib.KeyID,
+	targetUserID string, targetKeyID gomatrixserverlib.KeyID,
+	signature gomatrixserverlib.Base64Bytes,
+) error {
+	return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		if err := d.CrossSigningSigsTable.InsertCrossSigningSigsForTarget(ctx, nil, originUserID, originKeyID, targetUserID, targetKeyID, signature); err != nil {
+			return fmt.Errorf("d.CrossSigningSigsTable.InsertCrossSigningSigsForTarget: %w", err)
+		}
+		return nil
+	})
+}

keyserver/storage/sqlite3/cross_signing_keys_table.go

@@ -0,0 +1,96 @@
+// 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 sqlite3
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/storage/tables"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+var crossSigningKeysSchema = `
+CREATE TABLE IF NOT EXISTS keyserver_cross_signing_keys (
+    user_id TEXT NOT NULL,
+	key_type TEXT NOT NULL,
+	key_data TEXT NOT NULL,
+	PRIMARY KEY (user_id, key_type)
+);
+`
+
+const selectCrossSigningKeysForUserSQL = "" +
+	"SELECT key_type, key_data FROM keyserver_cross_signing_keys" +
+	" WHERE user_id = $1"
+
+const insertCrossSigningKeysForUserSQL = "" +
+	"INSERT OR REPLACE INTO keyserver_cross_signing_keys (user_id, key_type, key_data)" +
+	" VALUES($1, $2, $3)"
+
+type crossSigningKeysStatements struct {
+	db                                *sql.DB
+	selectCrossSigningKeysForUserStmt *sql.Stmt
+	insertCrossSigningKeysForUserStmt *sql.Stmt
+}
+
+func NewSqliteCrossSigningKeysTable(db *sql.DB) (tables.CrossSigningKeys, error) {
+	s := &crossSigningKeysStatements{
+		db: db,
+	}
+	_, err := db.Exec(crossSigningKeysSchema)
+	if err != nil {
+		return nil, err
+	}
+	if s.selectCrossSigningKeysForUserStmt, err = db.Prepare(selectCrossSigningKeysForUserSQL); err != nil {
+		return nil, err
+	}
+	if s.insertCrossSigningKeysForUserStmt, err = db.Prepare(insertCrossSigningKeysForUserSQL); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *crossSigningKeysStatements) SelectCrossSigningKeysForUser(
+	ctx context.Context, txn *sql.Tx, userID string,
+) (r api.CrossSigningKeyMap, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningKeysForUserStmt).QueryContext(ctx, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectCrossSigningKeysForUserStmt: rows.close() failed")
+	r = api.CrossSigningKeyMap{}
+	for rows.Next() {
+		var keyType gomatrixserverlib.CrossSigningKeyPurpose
+		var keyData gomatrixserverlib.Base64Bytes
+		if err := rows.Scan(&keyType, &keyData); err != nil {
+			return nil, err
+		}
+		r[keyType] = keyData
+	}
+	return
+}
+
+func (s *crossSigningKeysStatements) InsertCrossSigningKeysForUser(
+	ctx context.Context, txn *sql.Tx, userID string, keyType gomatrixserverlib.CrossSigningKeyPurpose, keyData gomatrixserverlib.Base64Bytes,
+) error {
+	if _, err := sqlutil.TxStmt(txn, s.insertCrossSigningKeysForUserStmt).ExecContext(ctx, userID, keyType, keyData); err != nil {
+		return fmt.Errorf("s.insertCrossSigningKeysForUserStmt: %w", err)
+	}
+	return nil
+}

keyserver/storage/sqlite3/cross_signing_sigs_table.go

@@ -0,0 +1,105 @@
+// 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 sqlite3
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/storage/tables"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+var crossSigningSigsSchema = `
+CREATE TABLE IF NOT EXISTS keyserver_cross_signing_sigs (
+    origin_user_id TEXT NOT NULL,
+	origin_key_id TEXT NOT NULL,
+	target_user_id TEXT NOT NULL,
+	target_key_id TEXT NOT NULL,
+	signature TEXT NOT NULL,
+	PRIMARY KEY (origin_user_id, target_user_id, target_key_id)
+);
+`
+
+const selectCrossSigningSigsForTargetSQL = "" +
+	"SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" +
+	" WHERE target_user_id = $1 AND target_key_id = $2"
+
+const insertCrossSigningSigsForTargetSQL = "" +
+	"INSERT OR REPLACE INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" +
+	" VALUES($1, $2, $3, $4, $5)"
+
+type crossSigningSigsStatements struct {
+	db                                  *sql.DB
+	selectCrossSigningSigsForTargetStmt *sql.Stmt
+	insertCrossSigningSigsForTargetStmt *sql.Stmt
+}
+
+func NewSqliteCrossSigningSigsTable(db *sql.DB) (tables.CrossSigningSigs, error) {
+	s := &crossSigningSigsStatements{
+		db: db,
+	}
+	_, err := db.Exec(crossSigningSigsSchema)
+	if err != nil {
+		return nil, err
+	}
+	if s.selectCrossSigningSigsForTargetStmt, err = db.Prepare(selectCrossSigningSigsForTargetSQL); err != nil {
+		return nil, err
+	}
+	if s.insertCrossSigningSigsForTargetStmt, err = db.Prepare(insertCrossSigningSigsForTargetSQL); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *crossSigningSigsStatements) SelectCrossSigningSigsForTarget(
+	ctx context.Context, txn *sql.Tx, targetUserID string, targetKeyID gomatrixserverlib.KeyID,
+) (r api.CrossSigningSigMap, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, targetUserID, targetKeyID)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectCrossSigningSigsForTargetStmt: rows.close() failed")
+	r = api.CrossSigningSigMap{}
+	for rows.Next() {
+		var userID string
+		var keyID gomatrixserverlib.KeyID
+		var signature gomatrixserverlib.Base64Bytes
+		if err := rows.Scan(&userID, &keyID, &signature); err != nil {
+			return nil, err
+		}
+		if _, ok := r[userID]; !ok {
+			r[userID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{}
+		}
+		r[userID][keyID] = signature
+	}
+	return
+}
+
+func (s *crossSigningSigsStatements) InsertCrossSigningSigsForTarget(
+	ctx context.Context, txn *sql.Tx,
+	originUserID string, originKeyID gomatrixserverlib.KeyID,
+	targetUserID string, targetKeyID gomatrixserverlib.KeyID,
+	signature gomatrixserverlib.Base64Bytes,
+) error {
+	if _, err := sqlutil.TxStmt(txn, s.insertCrossSigningSigsForTargetStmt).ExecContext(ctx, originUserID, originKeyID, targetUserID, targetKeyID, signature); err != nil {
+		return fmt.Errorf("s.insertCrossSigningSigsForTargetStmt: %w", err)
+	}
+	return nil
+}

keyserver/storage/sqlite3/storage.go

@@ -41,12 +41,26 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*shared.Database, error)
 	if err != nil {
 		return nil, err
 	}
-	return &shared.Database{
+	csk, err := NewSqliteCrossSigningKeysTable(db)
+	if err != nil {
+		return nil, err
+	}
+	css, err := NewSqliteCrossSigningSigsTable(db)
+	if err != nil {
+		return nil, err
+	}
+	d := &shared.Database{
 		DB:                    db,
 		Writer:                sqlutil.NewExclusiveWriter(),
 		OneTimeKeysTable:      otk,
 		DeviceKeysTable:       dk,
 		KeyChangesTable:       kc,
 		StaleDeviceListsTable: sdl,
-	}, nil
+		CrossSigningKeysTable: csk,
+		CrossSigningSigsTable: css,
+	}
+	if err = d.PartitionOffsetStatements.Prepare(db, d.Writer, "keyserver"); err != nil {
+		return nil, err
+	}
+	return d, nil
 }

keyserver/storage/tables/interface.go

@@ -52,3 +52,15 @@ type StaleDeviceLists interface {
 	InsertStaleDeviceList(ctx context.Context, userID string, isStale bool) error
 	SelectUserIDsWithStaleDeviceLists(ctx context.Context, domains []gomatrixserverlib.ServerName) ([]string, error)
 }
+
+type CrossSigningKeys interface {
+	SelectCrossSigningKeysForUser(ctx context.Context, txn *sql.Tx, userID string) (r api.CrossSigningKeyMap, err error)
+	InsertCrossSigningKeysForUser(ctx context.Context, txn *sql.Tx, userID string, keyType gomatrixserverlib.CrossSigningKeyPurpose, keyData gomatrixserverlib.Base64Bytes) error
+}
+
+type CrossSigningSigs interface {
+	SelectCrossSigningSigsForTarget(ctx context.Context, txn *sql.Tx, targetUserID string, targetKeyID gomatrixserverlib.KeyID) (r api.CrossSigningSigMap, err error)
+	InsertCrossSigningSigsForTarget(ctx context.Context, txn *sql.Tx, originUserID string, originKeyID gomatrixserverlib.KeyID, targetUserID string, targetKeyID gomatrixserverlib.KeyID, signature gomatrixserverlib.Base64Bytes) error
+}
+
+type CrossSigningStreams interface{}

setup/config/config_kafka.go

@@ -10,6 +10,7 @@ const (
 	TopicOutputRoomEvent         = "OutputRoomEvent"
 	TopicOutputClientData        = "OutputClientData"
 	TopicOutputReceiptEvent      = "OutputReceiptEvent"
+	TopicOutputSigningKeyUpdate  = "OutputSigningKeyUpdate"
 )
 
 type Kafka struct {

syncapi/internal/keychange_test.go

@@ -33,6 +33,10 @@ func (k *mockKeyAPI) SetUserAPI(i userapi.UserInternalAPI) {}
 // PerformClaimKeys claims one-time keys for use in pre-key messages
 func (k *mockKeyAPI) PerformClaimKeys(ctx context.Context, req *keyapi.PerformClaimKeysRequest, res *keyapi.PerformClaimKeysResponse) {
 }
+func (k *mockKeyAPI) PerformUploadDeviceKeys(ctx context.Context, req *keyapi.PerformUploadDeviceKeysRequest, res *keyapi.PerformUploadDeviceKeysResponse) {
+}
+func (k *mockKeyAPI) PerformUploadDeviceSignatures(ctx context.Context, req *keyapi.PerformUploadDeviceSignaturesRequest, res *keyapi.PerformUploadDeviceSignaturesResponse) {
+}
 func (k *mockKeyAPI) QueryKeys(ctx context.Context, req *keyapi.QueryKeysRequest, res *keyapi.QueryKeysResponse) {
 }
 func (k *mockKeyAPI) QueryKeyChanges(ctx context.Context, req *keyapi.QueryKeyChangesRequest, res *keyapi.QueryKeyChangesResponse) {