Perspective key fetching, some federation room join fixes (#975)

* Update gomatrixserverlib

* Test matrix.org as perspective key server

* Base64 decode better

* Optional strict validity checking in gmsl

* Update gomatrixserverlib

* Attempt to find missing auth events over federation (this shouldn't happen but I am guessing there is a synapse bug involved where we don't get all of the auth events)

* Update gomatrixserverlib, debug logging

* Remove debugging output

* More verbose debugging

* Print outliers

* Increase timeouts for testing, observe contexts before trying to join over more servers

* Don't block on roomserver (experimental)

* Don't block on roomserver

* Update gomatrixserverlib

* Update gomatrixserverlib

* Configurable perspective key fetchers

* Output number of configured keys for perspective

* Example perspective config included

* Undo debug stack trace

* Undo debug stack trace

* Restore original HTTP listener in monolith

* Fix lint

* Review comments

* Set default HTTP server timeout to 5 minutes now, block again when joining

* Don't use HTTP address for HTTPS whoops

* Update gomatrixserverlib

* Update gomatrixserverlib

* Update gomatrixserverlib

* Actually add perspectives

* Actually add perspectives

* Update gomatrixserverlib

clientapi/routing/joinroom.go

@@ -32,6 +32,7 @@ import (
 	"github.com/matrix-org/gomatrix"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
+	"github.com/sirupsen/logrus"
 )
 
 // JoinRoomByIDOrAlias implements the "/join/{roomIDOrAlias}" API.
@@ -290,8 +291,16 @@ func (r joinRoomReq) joinRoomUsingServers(
 			// There was a problem talking to one of the servers.
 			util.GetLogger(r.req.Context()).WithError(lastErr).WithField("server", server).Warn("Failed to join room using server")
 			// Try the next server.
+			if r.req.Context().Err() != nil {
+				// The request context has expired so don't bother trying any
+				// more servers - they will immediately fail due to the expired
+				// context.
+				break
+			} else {
+				// The request context hasn't expired yet so try the next server.
 				continue
 			}
+		}
 		return *response
 	}
 
@@ -365,16 +374,22 @@ func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib
 		return nil, fmt.Errorf("r.federation.SendJoin: %w", err)
 	}
 
-	if err = respSendJoin.Check(r.req.Context(), r.keyRing, event); err != nil {
+	if err = r.checkSendJoinResponse(event, server, respMakeJoin, respSendJoin); err != nil {
-		return nil, fmt.Errorf("respSendJoin: %w", err)
+		return nil, err
 	}
 
+	util.GetLogger(r.req.Context()).WithFields(logrus.Fields{
+		"room_id":          roomID,
+		"num_auth_events":  len(respSendJoin.AuthEvents),
+		"num_state_events": len(respSendJoin.StateEvents),
+	}).Info("Room join signature and auth verification passed")
+
 	if err = r.producer.SendEventWithState(
 		r.req.Context(),
 		gomatrixserverlib.RespState(respSendJoin.RespState),
 		event.Headered(respMakeJoin.RoomVersion),
 	); err != nil {
-		return nil, fmt.Errorf("r.producer.SendEventWithState: %w", err)
+		util.GetLogger(r.req.Context()).WithError(err).Error("r.producer.SendEventWithState")
 	}
 
 	return &util.JSONResponse{
@@ -385,3 +400,49 @@ func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib
 		}{roomID},
 	}, nil
 }
+
+// checkSendJoinResponse checks that all of the signatures are correct
+// and that the join is allowed by the supplied state.
+func (r joinRoomReq) checkSendJoinResponse(
+	event gomatrixserverlib.Event,
+	server gomatrixserverlib.ServerName,
+	respMakeJoin gomatrixserverlib.RespMakeJoin,
+	respSendJoin gomatrixserverlib.RespSendJoin,
+) error {
+	// A list of events that we have retried, if they were not included in
+	// the auth events supplied in the send_join.
+	retries := map[string]bool{}
+
+retryCheck:
+	// TODO: Can we expand Check here to return a list of missing auth
+	// events rather than failing one at a time?
+	if err := respSendJoin.Check(r.req.Context(), r.keyRing, event); err != nil {
+		switch e := err.(type) {
+		case gomatrixserverlib.MissingAuthEventError:
+			// Check that we haven't already retried for this event, prevents
+			// us from ending up in endless loops
+			if !retries[e.AuthEventID] {
+				// Ask the server that we're talking to right now for the event
+				tx, txerr := r.federation.GetEvent(r.req.Context(), server, e.AuthEventID)
+				if txerr != nil {
+					return fmt.Errorf("r.federation.GetEvent: %w", txerr)
+				}
+				// For each event returned, add it to the auth events.
+				for _, pdu := range tx.PDUs {
+					ev, everr := gomatrixserverlib.NewEventFromUntrustedJSON(pdu, respMakeJoin.RoomVersion)
+					if everr != nil {
+						return fmt.Errorf("gomatrixserverlib.NewEventFromUntrustedJSON: %w", everr)
+					}
+					respSendJoin.AuthEvents = append(respSendJoin.AuthEvents, ev)
+				}
+				// Mark the event as retried and then give the check another go.
+				retries[e.AuthEventID] = true
+				goto retryCheck
+			}
+			return fmt.Errorf("respSendJoin (after retries): %w", e)
+		default:
+			return fmt.Errorf("respSendJoin: %w", err)
+		}
+	}
+	return nil
+}

cmd/dendrite-client-api-server/main.go

@@ -33,7 +33,7 @@ func main() {
 	deviceDB := base.CreateDeviceDB()
 	keyDB := base.CreateKeyDB()
 	federation := base.CreateFederationClient()
-	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
+	keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
 
 	asQuery := base.CreateHTTPAppServiceAPIs()
 	alias, input, query := base.CreateHTTPRoomserverAPIs()

cmd/dendrite-demo-libp2p/main.go

@@ -146,7 +146,7 @@ func main() {
 	deviceDB := base.Base.CreateDeviceDB()
 	keyDB := createKeyDB(base)
 	federation := createFederationClient(base)
-	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
+	keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
 
 	alias, input, query := roomserver.SetupRoomServerComponent(&base.Base)
 	eduInputAPI := eduserver.SetupEDUServerComponent(&base.Base, cache.New())

cmd/dendrite-federation-api-server/main.go

@@ -33,7 +33,7 @@ func main() {
 	keyDB := base.CreateKeyDB()
 	federation := base.CreateFederationClient()
 	federationSender := base.CreateHTTPFederationSenderAPIs()
-	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
+	keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
 
 	alias, input, query := base.CreateHTTPRoomserverAPIs()
 	asQuery := base.CreateHTTPAppServiceAPIs()

cmd/dendrite-monolith-server/main.go

@@ -55,7 +55,7 @@ func main() {
 	deviceDB := base.CreateDeviceDB()
 	keyDB := base.CreateKeyDB()
 	federation := base.CreateFederationClient()
-	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
+	keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
 
 	alias, input, query := roomserver.SetupRoomServerComponent(base)
 	eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New())
@@ -90,16 +90,26 @@ func main() {
 
 	// Expose the matrix APIs directly rather than putting them under a /api path.
 	go func() {
-		logrus.Info("Listening on ", *httpBindAddr)
+		serv := http.Server{
-		logrus.Fatal(http.ListenAndServe(*httpBindAddr, nil))
+			Addr:         *httpBindAddr,
+			WriteTimeout: basecomponent.HTTPServerTimeout,
+		}
+
+		logrus.Info("Listening on ", serv.Addr)
+		logrus.Fatal(serv.ListenAndServe())
 	}()
 	// Handle HTTPS if certificate and key are provided
-	go func() {
 	if *certFile != "" && *keyFile != "" {
-			logrus.Info("Listening on ", *httpsBindAddr)
+		go func() {
-			logrus.Fatal(http.ListenAndServeTLS(*httpsBindAddr, *certFile, *keyFile, nil))
+			serv := http.Server{
+				Addr:         *httpsBindAddr,
+				WriteTimeout: basecomponent.HTTPServerTimeout,
 			}
+
+			logrus.Info("Listening on ", serv.Addr)
+			logrus.Fatal(serv.ListenAndServeTLS(*certFile, *keyFile))
 		}()
+	}
 
 	// We want to block forever to let the HTTP and HTTPS handler serve the APIs
 	select {}

common/basecomponent/base.go

@@ -60,6 +60,9 @@ type BaseDendrite struct {
 	KafkaProducer sarama.SyncProducer
 }
 
+const HTTPServerTimeout = time.Minute * 5
+const HTTPClientTimeout = time.Second * 30
+
 // NewBaseDendrite creates a new instance to be used by a component.
 // The componentName is used for logging purposes, and should be a friendly name
 // of the compontent running, e.g. "SyncAPI"
@@ -80,14 +83,12 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
 		kafkaConsumer, kafkaProducer = setupKafka(cfg)
 	}
 
-	const defaultHTTPTimeout = 30 * time.Second
-
 	return &BaseDendrite{
 		componentName: componentName,
 		tracerCloser:  closer,
 		Cfg:           cfg,
 		APIMux:        mux.NewRouter().UseEncodedPath(),
-		httpClient:    &http.Client{Timeout: defaultHTTPTimeout},
+		httpClient:    &http.Client{Timeout: HTTPClientTimeout},
 		KafkaConsumer: kafkaConsumer,
 		KafkaProducer: kafkaProducer,
 	}
@@ -209,16 +210,20 @@ func (b *BaseDendrite) SetupAndServeHTTP(bindaddr string, listenaddr string) {
 		addr = listenaddr
 	}
 
+	serv := http.Server{
+		Addr:         addr,
+		WriteTimeout: HTTPServerTimeout,
+	}
+
 	common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(b.APIMux), b.Cfg)
-	logrus.Infof("Starting %s server on %s", b.componentName, addr)
+	logrus.Infof("Starting %s server on %s", b.componentName, serv.Addr)
-
-	err := http.ListenAndServe(addr, nil)
 
+	err := serv.ListenAndServe()
 	if err != nil {
 		logrus.WithError(err).Fatal("failed to serve http")
 	}
 
-	logrus.Infof("Stopped %s server on %s", b.componentName, addr)
+	logrus.Infof("Stopped %s server on %s", b.componentName, serv.Addr)
 }
 
 // setupKafka creates kafka consumer/producer pair from the config.

common/config/config.go

@@ -99,6 +99,9 @@ type Dendrite struct {
 		// If set disables new users from registering (except via shared
 		// secrets)
 		RegistrationDisabled bool `yaml:"registration_disabled"`
+		// Perspective keyservers, to use as a backup when direct key fetch
+		// requests don't succeed
+		KeyPerspectives KeyPerspectives `yaml:"key_perspectives"`
 	} `yaml:"matrix"`
 
 	// The configuration specific to the media repostitory.
@@ -285,6 +288,21 @@ type Dendrite struct {
 	} `yaml:"-"`
 }
 
+// KeyPerspectives are used to configure perspective key servers for
+// retrieving server keys.
+type KeyPerspectives []struct {
+	// The server name of the perspective key server
+	ServerName gomatrixserverlib.ServerName `yaml:"server_name"`
+	// Server keys for the perspective user, used to verify the
+	// keys have been signed by the perspective server
+	Keys []struct {
+		// The key ID, e.g. ed25519:auto
+		KeyID gomatrixserverlib.KeyID `yaml:"key_id"`
+		// The public key in base64 unpadded format
+		PublicKey string `yaml:"public_key"`
+	} `yaml:"keys"`
+}
+
 // A Path on the filesystem.
 type Path string
 

common/keydb/keyring.go

@@ -14,19 +14,61 @@
 
 package keydb
 
-import "github.com/matrix-org/gomatrixserverlib"
+import (
+	"encoding/base64"
+
+	"github.com/matrix-org/dendrite/common/config"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/crypto/ed25519"
+)
 
 // CreateKeyRing creates and configures a KeyRing object.
 //
 // It creates the necessary key fetchers and collects them into a KeyRing
 // backed by the given KeyDatabase.
 func CreateKeyRing(client gomatrixserverlib.Client,
-	keyDB gomatrixserverlib.KeyDatabase) gomatrixserverlib.KeyRing {
+	keyDB gomatrixserverlib.KeyDatabase,
-	return gomatrixserverlib.KeyRing{
+	cfg config.KeyPerspectives) gomatrixserverlib.KeyRing {
+
+	fetchers := gomatrixserverlib.KeyRing{
 		KeyFetchers: []gomatrixserverlib.KeyFetcher{
-			// TODO: Use perspective key fetchers for production.
+			&gomatrixserverlib.DirectKeyFetcher{
-			&gomatrixserverlib.DirectKeyFetcher{Client: client},
+				Client: client,
+			},
 		},
 		KeyDatabase: keyDB,
 	}
+
+	logrus.Info("Enabled direct key fetcher")
+
+	var b64e = base64.StdEncoding.WithPadding(base64.NoPadding)
+	for _, ps := range cfg {
+		perspective := &gomatrixserverlib.PerspectiveKeyFetcher{
+			PerspectiveServerName: ps.ServerName,
+			PerspectiveServerKeys: map[gomatrixserverlib.KeyID]ed25519.PublicKey{},
+			Client:                client,
+		}
+
+		for _, key := range ps.Keys {
+			rawkey, err := b64e.DecodeString(key.PublicKey)
+			if err != nil {
+				logrus.WithError(err).WithFields(logrus.Fields{
+					"server_name": ps.ServerName,
+					"public_key":  key.PublicKey,
+				}).Warn("Couldn't parse perspective key")
+				continue
+			}
+			perspective.PerspectiveServerKeys[key.KeyID] = rawkey
+		}
+
+		fetchers.KeyFetchers = append(fetchers.KeyFetchers, perspective)
+
+		logrus.WithFields(logrus.Fields{
+			"server_name":     ps.ServerName,
+			"num_public_keys": len(ps.Keys),
+		}).Info("Enabled perspective key fetcher")
+	}
+
+	return fetchers
 }

dendrite-config.yaml

@@ -17,6 +17,14 @@ matrix:
     trusted_third_party_id_servers:
       - vector.im
       - matrix.org
+    # Perspective key servers which are used when direct key requests fail
+    #key_perspectives:
+    #  - server_name: matrix.org
+    #    keys:
+    #      - key_id: ed25519:auto
+    #        public_key: Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw
+    #      - key_id: ed25519:a_RXGa
+    #        public_key: l8Hft5qXKn1vfHrg3p4+W8gELQVo8N13JkluMfmn2sQ
 
 # The media repository config
 media:

federationapi/routing/invite.go

@@ -66,6 +66,7 @@ func Invite(
 		ServerName:             event.Origin(),
 		Message:                redacted.JSON(),
 		AtTS:                   event.OriginServerTS(),
+		StrictValidityChecking: true,
 	}}
 	verifyResults, err := keys.VerifyJSONs(httpReq.Context(), verifyRequests)
 	if err != nil {

federationapi/routing/join.go

@@ -199,6 +199,7 @@ func SendJoin(
 		ServerName:             event.Origin(),
 		Message:                redacted.JSON(),
 		AtTS:                   event.OriginServerTS(),
+		StrictValidityChecking: true,
 	}}
 	verifyResults, err := keys.VerifyJSONs(httpReq.Context(), verifyRequests)
 	if err != nil {

federationapi/routing/leave.go

@@ -148,6 +148,7 @@ func SendLeave(
 		ServerName:             event.Origin(),
 		Message:                redacted.JSON(),
 		AtTS:                   event.OriginServerTS(),
+		StrictValidityChecking: true,
 	}}
 	verifyResults, err := keys.VerifyJSONs(httpReq.Context(), verifyRequests)
 	if err != nil {

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10
 	github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20200416165239-837ed63a0046
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20200420162430-9662fc15e7e2
 	github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1
 	github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7
 	github.com/mattn/go-sqlite3 v2.0.3+incompatible

go.sum

@@ -364,8 +364,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20200416165239-837ed63a0046 h1:R7iYuS8hhXdrqs5OSNF3Y3chaIFWU9KLpnzjjsdJkMU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20200420162430-9662fc15e7e2 h1:XTg0bCZ+fQlWZHi1CinNecYfg2SFhyRNnGxKxXbLubg=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20200416165239-837ed63a0046/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20200420162430-9662fc15e7e2/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
 github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk=
 github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A=
 github.com/matrix-org/util v0.0.0-20171127121716-2e2df66af2f5 h1:W7l5CP4V7wPyPb4tYE11dbmeAOwtFQBTW0rf4OonOS8=
@@ -721,4 +721,5 @@ gopkg.in/yaml.v2 v2.2.5 h1:ymVxjfMaHvXD8RqPRmzHHsB3VvucivSkIAvJFDI5O3c=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099 h1:XJP7lxbSxWLOMNdBE4B/STaqVy6L73o0knwj2vIlxnw=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

sytest-whitelist

@@ -252,3 +252,4 @@ Outbound federation can send invites via v2 API
 User can invite local user to room with version 3
 User can invite local user to room with version 4
 A pair of servers can establish a join in a v2 room
+Can logout all devices