Merge branch 'main' into devon/tiered-dag

federationapi/federationapi_test.go

@@ -106,7 +106,7 @@ func (f *fedClient) GetServerKeys(ctx context.Context, matrixServer spec.ServerN
 	return keys, nil
 }
 
-func (f *fedClient) MakeJoin(ctx context.Context, origin, s spec.ServerName, roomID, userID string, roomVersions []gomatrixserverlib.RoomVersion) (res fclient.RespMakeJoin, err error) {
+func (f *fedClient) MakeJoin(ctx context.Context, origin, s spec.ServerName, roomID, userID string) (res fclient.RespMakeJoin, err error) {
 	for _, r := range f.allowJoins {
 		if r.ID == roomID {
 			res.RoomVersion = r.Version

federationapi/internal/perform.go

@@ -73,12 +73,6 @@ func (r *FederationInternalAPI) PerformJoin(
 	r.joins.Store(j, nil)
 	defer r.joins.Delete(j)
 
-	// Look up the supported room versions.
-	var supportedVersions []gomatrixserverlib.RoomVersion
-	for version := range version.SupportedRoomVersions() {
-		supportedVersions = append(supportedVersions, version)
-	}
-
 	// Deduplicate the server names we were provided but keep the ordering
 	// as this encodes useful information about which servers are most likely
 	// to respond.
@@ -103,7 +97,6 @@ func (r *FederationInternalAPI) PerformJoin(
 			request.UserID,
 			request.Content,
 			serverName,
-			supportedVersions,
 			request.Unsigned,
 		); err != nil {
 			logrus.WithError(err).WithFields(logrus.Fields{
@@ -146,190 +139,74 @@ func (r *FederationInternalAPI) performJoinUsingServer(
 	roomID, userID string,
 	content map[string]interface{},
 	serverName spec.ServerName,
-	supportedVersions []gomatrixserverlib.RoomVersion,
 	unsigned map[string]interface{},
 ) error {
 	if !r.shouldAttemptDirectFederation(serverName) {
 		return fmt.Errorf("relay servers have no meaningful response for join.")
 	}
 
-	_, origin, err := r.cfg.Matrix.SplitLocalID('@', userID)
+	user, err := spec.NewUserID(userID, true)
 	if err != nil {
 		return err
 	}
 
-	// Try to perform a make_join using the information supplied in the
+	sendJoinInput := fclient.SendJoinInput{
-	// request.
+		UserID:        user,
-	respMakeJoin, err := r.federation.MakeJoin(
+		RoomID:        roomID,
-		ctx,
+		ServerName:    serverName,
-		origin,
+		Content:       content,
-		serverName,
+		Unsigned:      unsigned,
-		roomID,
+		PrivateKey:    r.cfg.Matrix.PrivateKey,
-		userID,
+		KeyID:         r.cfg.Matrix.KeyID,
-		supportedVersions,
+		KeyRing:       r.keyRing,
-	)
+		EventProvider: federatedEventProvider,
-	if err != nil {
-		// TODO: Check if the user was not allowed to join the room.
-		r.statistics.ForServer(serverName).Failure()
-		return fmt.Errorf("r.federation.MakeJoin: %w", err)
 	}
-	r.statistics.ForServer(serverName).Success(statistics.SendDirect)
+	callbacks := fclient.SendJoinCallbacks{
-
+		FederationFailure: func(server spec.ServerName) {
-	// Set all the fields to be what they should be, this should be a no-op
+			r.statistics.ForServer(server).Failure()
-	// but it's possible that the remote server returned us something "odd"
+		},
-	respMakeJoin.JoinEvent.Type = spec.MRoomMember
+		FederationSuccess: func(server spec.ServerName) {
-	respMakeJoin.JoinEvent.Sender = userID
+			r.statistics.ForServer(server).Success(statistics.SendDirect)
-	respMakeJoin.JoinEvent.StateKey = &userID
+		},
-	respMakeJoin.JoinEvent.RoomID = roomID
-	respMakeJoin.JoinEvent.Redacts = ""
-	if content == nil {
-		content = map[string]interface{}{}
-	}
-	_ = json.Unmarshal(respMakeJoin.JoinEvent.Content, &content)
-	content["membership"] = spec.Join
-	if err = respMakeJoin.JoinEvent.SetContent(content); err != nil {
-		return fmt.Errorf("respMakeJoin.JoinEvent.SetContent: %w", err)
-	}
-	if err = respMakeJoin.JoinEvent.SetUnsigned(struct{}{}); err != nil {
-		return fmt.Errorf("respMakeJoin.JoinEvent.SetUnsigned: %w", err)
 	}
 
-	// Work out if we support the room version that has been supplied in
+	event, respState, err := fclient.HandleSendJoin(ctx, r.federation, sendJoinInput, callbacks)
-	// the make_join response.
-	// "If not provided, the room version is assumed to be either "1" or "2"."
-	// https://matrix.org/docs/spec/server_server/unstable#get-matrix-federation-v1-make-join-roomid-userid
-	if respMakeJoin.RoomVersion == "" {
-		respMakeJoin.RoomVersion = setDefaultRoomVersionFromJoinEvent(respMakeJoin.JoinEvent)
-	}
-	verImpl, err := gomatrixserverlib.GetRoomVersion(respMakeJoin.RoomVersion)
 	if err != nil {
 		return err
 	}
 
-	// Build the join event.
-	event, err := respMakeJoin.JoinEvent.Build(
-		time.Now(),
-		origin,
-		r.cfg.Matrix.KeyID,
-		r.cfg.Matrix.PrivateKey,
-		respMakeJoin.RoomVersion,
-	)
-	if err != nil {
-		return fmt.Errorf("respMakeJoin.JoinEvent.Build: %w", err)
-	}
-
-	// Try to perform a send_join using the newly built event.
-	respSendJoin, err := r.federation.SendJoin(
-		context.Background(),
-		origin,
-		serverName,
-		event,
-	)
-	if err != nil {
-		r.statistics.ForServer(serverName).Failure()
-		return fmt.Errorf("r.federation.SendJoin: %w", err)
-	}
-	r.statistics.ForServer(serverName).Success(statistics.SendDirect)
-
-	// If the remote server returned an event in the "event" key of
-	// the send_join request then we should use that instead. It may
-	// contain signatures that we don't know about.
-	if len(respSendJoin.Event) > 0 {
-		var remoteEvent *gomatrixserverlib.Event
-		remoteEvent, err = verImpl.NewEventFromUntrustedJSON(respSendJoin.Event)
-		if err == nil && isWellFormedMembershipEvent(
-			remoteEvent, roomID, userID,
-		) {
-			event = remoteEvent
-		}
-	}
-
-	// Sanity-check the join response to ensure that it has a create
-	// event, that the room version is known, etc.
-	authEvents := respSendJoin.AuthEvents.UntrustedEvents(respMakeJoin.RoomVersion)
-	if err = sanityCheckAuthChain(authEvents); err != nil {
-		return fmt.Errorf("sanityCheckAuthChain: %w", err)
-	}
-
-	// Process the join response in a goroutine. The idea here is
-	// that we'll try and wait for as long as possible for the work
-	// to complete, but if the client does give up waiting, we'll
-	// still continue to process the join anyway so that we don't
-	// waste the effort.
-	// TODO: Can we expand Check here to return a list of missing auth
-	// events rather than failing one at a time?
-	var respState gomatrixserverlib.StateResponse
-	respState, err = gomatrixserverlib.CheckSendJoinResponse(
-		context.Background(),
-		respMakeJoin.RoomVersion, &respSendJoin,
-		r.keyRing,
-		event,
-		federatedAuthProvider(ctx, r.federation, r.keyRing, origin, serverName),
-	)
-	if err != nil {
-		return fmt.Errorf("respSendJoin.Check: %w", err)
-	}
-
 	// We need to immediately update our list of joined hosts for this room now as we are technically
 	// joined. We must do this synchronously: we cannot rely on the roomserver output events as they
 	// will happen asyncly. If we don't update this table, you can end up with bad failure modes like
 	// joining a room, waiting for 200 OK then changing device keys and have those keys not be sent
 	// to other servers (this was a cause of a flakey sytest "Local device key changes get to remote servers")
 	// The events are trusted now as we performed auth checks above.
-	joinedHosts, err := consumers.JoinedHostsFromEvents(respState.GetStateEvents().TrustedEvents(respMakeJoin.RoomVersion, false))
+	joinedHosts, err := consumers.JoinedHostsFromEvents(respState.GetStateEvents().TrustedEvents(event.RoomVersion, false))
 	if err != nil {
 		return fmt.Errorf("JoinedHostsFromEvents: failed to get joined hosts: %s", err)
 	}
+
 	logrus.WithField("room", roomID).Infof("Joined federated room with %d hosts", len(joinedHosts))
 	if _, err = r.db.UpdateRoom(context.Background(), roomID, joinedHosts, nil, true); err != nil {
 		return fmt.Errorf("UpdatedRoom: failed to update room with joined hosts: %s", err)
 	}
 
-	// If we successfully performed a send_join above then the other
-	// server now thinks we're a part of the room. Send the newly
-	// returned state to the roomserver to update our local view.
-	if unsigned != nil {
-		event, err = event.SetUnsigned(unsigned)
-		if err != nil {
-			// non-fatal, log and continue
-			logrus.WithError(err).Errorf("Failed to set unsigned content")
-		}
-	}
-
 	if err = roomserverAPI.SendEventWithState(
 		context.Background(),
 		r.rsAPI,
-		origin,
+		user.Domain(),
 		roomserverAPI.KindNew,
 		respState,
-		event.Headered(respMakeJoin.RoomVersion),
+		event,
 		serverName,
 		nil,
 		false,
 	); err != nil {
 		return fmt.Errorf("roomserverAPI.SendEventWithState: %w", err)
 	}
-
 	return nil
 }
 
-// isWellFormedMembershipEvent returns true if the event looks like a legitimate
-// membership event.
-func isWellFormedMembershipEvent(event *gomatrixserverlib.Event, roomID, userID string) bool {
-	if membership, err := event.Membership(); err != nil {
-		return false
-	} else if membership != spec.Join {
-		return false
-	}
-	if event.RoomID() != roomID {
-		return false
-	}
-	if !event.StateKeyEquals(userID) {
-		return false
-	}
-	return true
-}
-
 // PerformOutboundPeekRequest implements api.FederationInternalAPI
 func (r *FederationInternalAPI) PerformOutboundPeek(
 	ctx context.Context,
@@ -475,12 +352,12 @@ func (r *FederationInternalAPI) performOutboundPeekUsingServer(
 	// authenticate the state returned (check its auth events etc)
 	// the equivalent of CheckSendJoinResponse()
 	authEvents, stateEvents, err := gomatrixserverlib.CheckStateResponse(
-		ctx, &respPeek, respPeek.RoomVersion, r.keyRing, federatedAuthProvider(ctx, r.federation, r.keyRing, r.cfg.Matrix.ServerName, serverName),
+		ctx, &respPeek, respPeek.RoomVersion, r.keyRing, federatedEventProvider(ctx, r.federation, r.keyRing, r.cfg.Matrix.ServerName, serverName),
 	)
 	if err != nil {
 		return fmt.Errorf("error checking state returned from peeking: %w", err)
 	}
-	if err = sanityCheckAuthChain(authEvents); err != nil {
+	if err = checkEventsContainCreateEvent(authEvents); err != nil {
 		return fmt.Errorf("sanityCheckAuthChain: %w", err)
 	}
 
@@ -719,9 +596,9 @@ func (r *FederationInternalAPI) MarkServersAlive(destinations []spec.ServerName)
 	}
 }
 
-func sanityCheckAuthChain(authChain []*gomatrixserverlib.Event) error {
+func checkEventsContainCreateEvent(events []*gomatrixserverlib.Event) error {
 	// sanity check we have a create event and it has a known room version
-	for _, ev := range authChain {
+	for _, ev := range events {
 		if ev.Type() == spec.MRoomCreate && ev.StateKeyEquals("") {
 			// make sure the room version is known
 			content := ev.Content()
@@ -739,49 +616,25 @@ func sanityCheckAuthChain(authChain []*gomatrixserverlib.Event) error {
 			}
 			knownVersions := gomatrixserverlib.RoomVersions()
 			if _, ok := knownVersions[gomatrixserverlib.RoomVersion(verBody.Version)]; !ok {
-				return fmt.Errorf("auth chain m.room.create event has an unknown room version: %s", verBody.Version)
+				return fmt.Errorf("m.room.create event has an unknown room version: %s", verBody.Version)
 			}
 			return nil
 		}
 	}
-	return fmt.Errorf("auth chain response is missing m.room.create event")
+	return fmt.Errorf("response is missing m.room.create event")
 }
 
-func setDefaultRoomVersionFromJoinEvent(
+// FederatedEventProvider is an auth chain provider which fetches events from the server provided
-	joinEvent gomatrixserverlib.EventBuilder,
+func federatedEventProvider(
-) gomatrixserverlib.RoomVersion {
-	// if auth events are not event references we know it must be v3+
-	// we have to do these shenanigans to satisfy sytest, specifically for:
-	// "Outbound federation rejects m.room.create events with an unknown room version"
-	hasEventRefs := true
-	authEvents, ok := joinEvent.AuthEvents.([]interface{})
-	if ok {
-		if len(authEvents) > 0 {
-			_, ok = authEvents[0].(string)
-			if ok {
-				// event refs are objects, not strings, so we know we must be dealing with a v3+ room.
-				hasEventRefs = false
-			}
-		}
-	}
-
-	if hasEventRefs {
-		return gomatrixserverlib.RoomVersionV1
-	}
-	return gomatrixserverlib.RoomVersionV4
-}
-
-// FederatedAuthProvider is an auth chain provider which fetches events from the server provided
-func federatedAuthProvider(
 	ctx context.Context, federation fclient.FederationClient,
 	keyRing gomatrixserverlib.JSONVerifier, origin, server spec.ServerName,
-) gomatrixserverlib.AuthChainProvider {
+) gomatrixserverlib.EventProvider {
 	// A list of events that we have retried, if they were not included in
-	// the auth events supplied in the send_join.
+	// the events supplied in the send_join.
 	retries := map[string][]*gomatrixserverlib.Event{}
 
 	// Define a function which we can pass to Check to retrieve missing
-	// auth events inline. This greatly increases our chances of not having
+	// events inline. This greatly increases our chances of not having
 	// to repeat the entire set of checks just for a missing event or two.
 	return func(roomVersion gomatrixserverlib.RoomVersion, eventIDs []string) ([]*gomatrixserverlib.Event, error) {
 		returning := []*gomatrixserverlib.Event{}
@@ -796,7 +649,7 @@ func federatedAuthProvider(
 			// just append the results. We won't retry the request.
 			if retry, ok := retries[eventID]; ok {
 				if retry == nil {
-					return nil, fmt.Errorf("missingAuth: not retrying failed event ID %q", eventID)
+					return nil, fmt.Errorf("missingEvent: not retrying failed event ID %q", eventID)
 				}
 				returning = append(returning, retry...)
 				continue
@@ -810,7 +663,7 @@ func federatedAuthProvider(
 			// join response.
 			tx, txerr := federation.GetEvent(ctx, origin, server, eventID)
 			if txerr != nil {
-				return nil, fmt.Errorf("missingAuth r.federation.GetEvent: %w", txerr)
+				return nil, fmt.Errorf("missingEvent r.federation.GetEvent: %w", txerr)
 			}
 
 			// For each event returned, add it to the set of return events. We
@@ -820,12 +673,12 @@ func federatedAuthProvider(
 				// Try to parse the event.
 				ev, everr := verImpl.NewEventFromUntrustedJSON(pdu)
 				if everr != nil {
-					return nil, fmt.Errorf("missingAuth gomatrixserverlib.NewEventFromUntrustedJSON: %w", everr)
+					return nil, fmt.Errorf("missingEvent gomatrixserverlib.NewEventFromUntrustedJSON: %w", everr)
 				}
 
 				// Check the signatures of the event.
 				if err := ev.VerifyEventSignatures(ctx, keyRing); err != nil {
-					return nil, fmt.Errorf("missingAuth VerifyEventSignatures: %w", err)
+					return nil, fmt.Errorf("missingEvent VerifyEventSignatures: %w", err)
 				}
 
 				// If the event is OK then add it to the results and the retry map.

federationapi/routing/join.go

@@ -322,6 +322,9 @@ func SendJoin(
 		}
 	}
 
+	// TODO: (PowerDAG) query state & power DAG
+	// We return the entire power DAG with a send_join response
+
 	// Fetch the state and auth chain. We do this before we send the events
 	// on, in case this fails.
 	var stateAndAuthChainResponse api.QueryStateAndAuthChainResponse
@@ -349,6 +352,9 @@ func SendJoin(
 		}
 	}
 
+	// TODO: (Power DAG) will we have enough info in a join response to check membership?
+	// We should include the user membership event (if there is one) in the state response.
+
 	// Check if the user is already in the room. If they're already in then
 	// there isn't much point in sending another join event into the room.
 	// Also check to see if they are banned: if they are then we reject them.
@@ -435,6 +441,8 @@ func SendJoin(
 		}
 	}
 
+	// TODO: (PowerDAG) Adapt this to return the full power DAG & state + user membership event (if it exists)
+
 	// sort events deterministically by depth (lower is earlier)
 	// We also do this because sytest's basic federation server isn't good at using the correct
 	// state if these lists are randomised, resulting in flakey tests. :(

federationapi/routing/state.go

@@ -41,6 +41,7 @@ func GetState(
 		return *err
 	}
 
+	// TODO: (PowerDAG) return full power DAG instead of auth chains?
 	return util.JSONResponse{Code: http.StatusOK, JSON: &fclient.RespState{
 		AuthEvents:  gomatrixserverlib.NewEventJSONsFromHeaderedEvents(authChain),
 		StateEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(stateEvents),
@@ -67,6 +68,7 @@ func GetStateIDs(
 	stateEventIDs := getIDsFromEvent(stateEvents)
 	authEventIDs := getIDsFromEvent(authEvents)
 
+	// TODO: (PowerDAG) return full power DAG instead of auth chains?
 	return util.JSONResponse{Code: http.StatusOK, JSON: fclient.RespStateIDs{
 		StateEventIDs: stateEventIDs,
 		AuthEventIDs:  authEventIDs,
@@ -121,6 +123,7 @@ func getState(
 		return nil, nil, resErr
 	}
 
+	// TODO: (PowerDAG) query full power DAG instead of auth chains
 	var response api.QueryStateAndAuthChainResponse
 	err := rsAPI.QueryStateAndAuthChain(
 		ctx,

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
 	github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20230424155704-8daeaebaa0bc
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20230424210043-7e71cd67ae50
 	github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a
 	github.com/matrix-org/util v0.0.0-20221111132719-399730281e66
 	github.com/mattn/go-sqlite3 v1.14.16

go.sum

@@ -321,8 +321,8 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91 h1:s7fexw
 github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo=
 github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530 h1:kHKxCOLcHH8r4Fzarl4+Y3K5hjothkVW5z7T1dUM11U=
 github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20230424155704-8daeaebaa0bc h1:F73iHhpTZxWVO6qbyGZxd7Ch44v1gK6xNQZ7QVos/Es=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20230424210043-7e71cd67ae50 h1:3LoWLddWml6AmpT3+f4ciNA4SiNzQYKvmQAo8KDvIiE=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20230424155704-8daeaebaa0bc/go.mod h1:7HTbSZe+CIdmeqVyFMekwD5dFU8khWQyngKATvd12FU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20230424210043-7e71cd67ae50/go.mod h1:H9V9N3Uqn1bBJqYJNGK1noqtgJTaCEhtTdcH/mp50uU=
 github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a h1:awrPDf9LEFySxTLKYBMCiObelNx/cBuv/wzllvCCH3A=
 github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a/go.mod h1:HchJX9oKMXaT2xYFs0Ha/6Zs06mxLU8k6F1ODnrGkeQ=
 github.com/matrix-org/util v0.0.0-20221111132719-399730281e66 h1:6z4KxomXSIGWqhHcfzExgkH3Z3UkIXry4ibJS4Aqz2Y=