2020-05-01 14:01:50 +02:00
|
|
|
package internal
|
2020-04-29 12:34:31 +02:00
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2020-10-10 01:21:15 +02:00
|
|
|
"encoding/json"
|
2020-06-25 16:04:48 +02:00
|
|
|
"errors"
|
2020-04-29 16:29:39 +02:00
|
|
|
"fmt"
|
|
|
|
"time"
|
2020-04-29 12:34:31 +02:00
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
"github.com/matrix-org/dendrite/federationapi/api"
|
2020-06-10 17:54:43 +02:00
|
|
|
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
2020-04-29 16:29:39 +02:00
|
|
|
"github.com/matrix-org/dendrite/roomserver/version"
|
2020-06-25 16:04:48 +02:00
|
|
|
"github.com/matrix-org/gomatrix"
|
2020-04-29 16:29:39 +02:00
|
|
|
"github.com/matrix-org/gomatrixserverlib"
|
2020-05-04 14:53:47 +02:00
|
|
|
"github.com/matrix-org/util"
|
|
|
|
"github.com/sirupsen/logrus"
|
2020-04-29 12:34:31 +02:00
|
|
|
)
|
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformLeaveRequest implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformDirectoryLookup(
|
2020-05-04 14:53:47 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformDirectoryLookupRequest,
|
|
|
|
response *api.PerformDirectoryLookupResponse,
|
|
|
|
) (err error) {
|
|
|
|
dir, err := r.federation.LookupRoomAlias(
|
|
|
|
ctx,
|
|
|
|
request.ServerName,
|
|
|
|
request.RoomAlias,
|
|
|
|
)
|
|
|
|
if err != nil {
|
Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007)
* Improve federation sender performance and behaviour, add backoff
* Tweaks
* Tweaks
* Tweaks
* Take copies of events before passing to destination queues
* Don't accidentally drop queued messages
* Don't take copies again
* Tidy up a bit
* Break out statistics (tracked component-wide), report success and failures from Perform actions
* Fix comment, use atomic add
* Improve logic a bit, don't block on wakeup, move idle check
* Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc
* Dedupe destinations, fix other bug hopefully
* Dispatch sends again
* Federation sender to ignore invites that are destined locally
* Loopback invite events
* Remodel a bit with channels
* Linter
* Only loopback invite event if we know the room
* We should tell other resident servers about the invite if we know about the room
* Correct invite signing
* Fix invite loopback
* Check HTTP response codes, push new invites to front of queue
* Review comments
2020-05-07 13:42:06 +02:00
|
|
|
r.statistics.ForServer(request.ServerName).Failure()
|
2020-05-04 14:53:47 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
response.RoomID = dir.RoomID
|
|
|
|
response.ServerNames = dir.Servers
|
Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007)
* Improve federation sender performance and behaviour, add backoff
* Tweaks
* Tweaks
* Tweaks
* Take copies of events before passing to destination queues
* Don't accidentally drop queued messages
* Don't take copies again
* Tidy up a bit
* Break out statistics (tracked component-wide), report success and failures from Perform actions
* Fix comment, use atomic add
* Improve logic a bit, don't block on wakeup, move idle check
* Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc
* Dedupe destinations, fix other bug hopefully
* Dispatch sends again
* Federation sender to ignore invites that are destined locally
* Loopback invite events
* Remodel a bit with channels
* Linter
* Only loopback invite event if we know the room
* We should tell other resident servers about the invite if we know about the room
* Correct invite signing
* Fix invite loopback
* Check HTTP response codes, push new invites to front of queue
* Review comments
2020-05-07 13:42:06 +02:00
|
|
|
r.statistics.ForServer(request.ServerName).Success()
|
2020-05-04 14:53:47 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-09-22 12:05:45 +02:00
|
|
|
type federatedJoin struct {
|
|
|
|
UserID string
|
|
|
|
RoomID string
|
|
|
|
}
|
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformJoin implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformJoin(
|
2020-04-29 12:34:31 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformJoinRequest,
|
|
|
|
response *api.PerformJoinResponse,
|
2020-06-25 16:04:48 +02:00
|
|
|
) {
|
2020-09-22 12:05:45 +02:00
|
|
|
// Check that a join isn't already in progress for this user/room.
|
|
|
|
j := federatedJoin{request.UserID, request.RoomID}
|
|
|
|
if _, found := r.joins.Load(j); found {
|
|
|
|
response.LastError = &gomatrix.HTTPError{
|
|
|
|
Code: 429,
|
|
|
|
Message: `{
|
|
|
|
"errcode": "M_LIMIT_EXCEEDED",
|
|
|
|
"error": "There is already a federated join to this room in progress. Please wait for it to finish."
|
|
|
|
}`, // TODO: Why do none of our error types play nicely with each other?
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
r.joins.Store(j, nil)
|
|
|
|
defer r.joins.Delete(j)
|
|
|
|
|
2020-04-29 16:29:39 +02:00
|
|
|
// Look up the supported room versions.
|
|
|
|
var supportedVersions []gomatrixserverlib.RoomVersion
|
|
|
|
for version := range version.SupportedRoomVersions() {
|
|
|
|
supportedVersions = append(supportedVersions, version)
|
|
|
|
}
|
|
|
|
|
2020-05-29 14:50:06 +02:00
|
|
|
// Deduplicate the server names we were provided but keep the ordering
|
|
|
|
// as this encodes useful information about which servers are most likely
|
|
|
|
// to respond.
|
|
|
|
seenSet := make(map[gomatrixserverlib.ServerName]bool)
|
|
|
|
var uniqueList []gomatrixserverlib.ServerName
|
|
|
|
for _, srv := range request.ServerNames {
|
2022-04-27 13:23:55 +02:00
|
|
|
if seenSet[srv] || srv == r.cfg.Matrix.ServerName {
|
2020-05-29 14:50:06 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
seenSet[srv] = true
|
|
|
|
uniqueList = append(uniqueList, srv)
|
|
|
|
}
|
|
|
|
request.ServerNames = uniqueList
|
2020-04-29 16:29:39 +02:00
|
|
|
|
2020-05-04 14:53:47 +02:00
|
|
|
// Try each server that we were provided until we land on one that
|
|
|
|
// successfully completes the make-join send-join dance.
|
2020-06-25 16:04:48 +02:00
|
|
|
var lastErr error
|
2020-05-04 14:53:47 +02:00
|
|
|
for _, serverName := range request.ServerNames {
|
2020-05-15 14:55:14 +02:00
|
|
|
if err := r.performJoinUsingServer(
|
2020-05-04 14:53:47 +02:00
|
|
|
ctx,
|
|
|
|
request.RoomID,
|
|
|
|
request.UserID,
|
2020-05-15 14:55:14 +02:00
|
|
|
request.Content,
|
2020-05-04 14:53:47 +02:00
|
|
|
serverName,
|
2020-05-15 14:55:14 +02:00
|
|
|
supportedVersions,
|
2020-05-04 14:53:47 +02:00
|
|
|
); err != nil {
|
2020-05-15 14:55:14 +02:00
|
|
|
logrus.WithError(err).WithFields(logrus.Fields{
|
|
|
|
"server_name": serverName,
|
|
|
|
"room_id": request.RoomID,
|
|
|
|
}).Warnf("Failed to join room through server")
|
2020-06-25 16:04:48 +02:00
|
|
|
lastErr = err
|
2020-05-04 14:53:47 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// We're all good.
|
2020-11-19 12:34:59 +01:00
|
|
|
response.JoinedVia = serverName
|
2020-06-25 16:04:48 +02:00
|
|
|
return
|
2020-04-29 16:29:39 +02:00
|
|
|
}
|
|
|
|
|
2020-05-04 14:53:47 +02:00
|
|
|
// If we reach here then we didn't complete a join for some reason.
|
2020-06-25 16:04:48 +02:00
|
|
|
var httpErr gomatrix.HTTPError
|
|
|
|
if ok := errors.As(lastErr, &httpErr); ok {
|
|
|
|
httpErr.Message = string(httpErr.Contents)
|
|
|
|
// Clear the wrapped error, else serialising to JSON (in polylith mode) will fail
|
|
|
|
httpErr.WrappedError = nil
|
|
|
|
response.LastError = &httpErr
|
|
|
|
} else {
|
|
|
|
response.LastError = &gomatrix.HTTPError{
|
|
|
|
Code: 0,
|
|
|
|
WrappedError: nil,
|
2020-09-15 12:17:46 +02:00
|
|
|
Message: "Unknown HTTP error",
|
|
|
|
}
|
|
|
|
if lastErr != nil {
|
|
|
|
response.LastError.Message = lastErr.Error()
|
2020-06-25 16:04:48 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
logrus.Errorf(
|
|
|
|
"failed to join user %q to room %q through %d server(s): last error %s",
|
|
|
|
request.UserID, request.RoomID, len(request.ServerNames), lastErr,
|
2020-05-04 14:53:47 +02:00
|
|
|
)
|
2020-04-29 12:34:31 +02:00
|
|
|
}
|
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
func (r *FederationInternalAPI) performJoinUsingServer(
|
2020-05-15 14:55:14 +02:00
|
|
|
ctx context.Context,
|
|
|
|
roomID, userID string,
|
|
|
|
content map[string]interface{},
|
|
|
|
serverName gomatrixserverlib.ServerName,
|
|
|
|
supportedVersions []gomatrixserverlib.RoomVersion,
|
|
|
|
) error {
|
|
|
|
// Try to perform a make_join using the information supplied in the
|
|
|
|
// request.
|
|
|
|
respMakeJoin, err := r.federation.MakeJoin(
|
|
|
|
ctx,
|
|
|
|
serverName,
|
|
|
|
roomID,
|
|
|
|
userID,
|
|
|
|
supportedVersions,
|
|
|
|
)
|
|
|
|
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()
|
|
|
|
|
|
|
|
// Set all the fields to be what they should be, this should be a no-op
|
|
|
|
// but it's possible that the remote server returned us something "odd"
|
|
|
|
respMakeJoin.JoinEvent.Type = gomatrixserverlib.MRoomMember
|
|
|
|
respMakeJoin.JoinEvent.Sender = userID
|
|
|
|
respMakeJoin.JoinEvent.StateKey = &userID
|
|
|
|
respMakeJoin.JoinEvent.RoomID = roomID
|
|
|
|
respMakeJoin.JoinEvent.Redacts = ""
|
|
|
|
if content == nil {
|
|
|
|
content = map[string]interface{}{}
|
|
|
|
}
|
|
|
|
content["membership"] = "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
|
|
|
|
// the make_join response.
|
2020-10-10 01:21:15 +02:00
|
|
|
// "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
|
2020-05-15 14:55:14 +02:00
|
|
|
if respMakeJoin.RoomVersion == "" {
|
2020-10-10 01:21:15 +02:00
|
|
|
respMakeJoin.RoomVersion = setDefaultRoomVersionFromJoinEvent(respMakeJoin.JoinEvent)
|
2020-05-15 14:55:14 +02:00
|
|
|
}
|
|
|
|
if _, err = respMakeJoin.RoomVersion.EventFormat(); err != nil {
|
|
|
|
return fmt.Errorf("respMakeJoin.RoomVersion.EventFormat: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Build the join event.
|
|
|
|
event, err := respMakeJoin.JoinEvent.Build(
|
|
|
|
time.Now(),
|
|
|
|
r.cfg.Matrix.ServerName,
|
|
|
|
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(
|
2022-02-02 18:33:36 +01:00
|
|
|
context.Background(),
|
2020-05-15 14:55:14 +02:00
|
|
|
serverName,
|
|
|
|
event,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
r.statistics.ForServer(serverName).Failure()
|
|
|
|
return fmt.Errorf("r.federation.SendJoin: %w", err)
|
|
|
|
}
|
|
|
|
r.statistics.ForServer(serverName).Success()
|
2020-10-13 12:53:20 +02:00
|
|
|
|
2022-02-09 21:31:24 +01:00
|
|
|
authEvents := respSendJoin.AuthEvents.UntrustedEvents(respMakeJoin.RoomVersion)
|
|
|
|
|
2020-10-13 12:53:20 +02:00
|
|
|
// Sanity-check the join response to ensure that it has a create
|
|
|
|
// event, that the room version is known, etc.
|
2022-02-09 21:31:24 +01:00
|
|
|
if err = sanityCheckAuthChain(authEvents); err != nil {
|
2021-01-22 18:16:35 +01:00
|
|
|
return fmt.Errorf("sanityCheckAuthChain: %w", err)
|
2020-10-10 01:21:15 +02:00
|
|
|
}
|
2020-05-15 14:55:14 +02:00
|
|
|
|
2020-09-22 12:05:45 +02:00
|
|
|
// 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.
|
2022-02-02 18:33:36 +01:00
|
|
|
// TODO: Can we expand Check here to return a list of missing auth
|
|
|
|
// events rather than failing one at a time?
|
|
|
|
var respState *gomatrixserverlib.RespState
|
|
|
|
respState, err = respSendJoin.Check(
|
|
|
|
context.Background(),
|
2022-02-09 21:31:24 +01:00
|
|
|
respMakeJoin.RoomVersion,
|
2022-02-02 18:33:36 +01:00
|
|
|
r.keyRing,
|
|
|
|
event,
|
|
|
|
federatedAuthProvider(ctx, r.federation, r.keyRing, serverName),
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("respSendJoin.Check: %w", err)
|
|
|
|
}
|
2020-09-22 12:05:45 +02:00
|
|
|
|
2022-02-02 18:33:36 +01:00
|
|
|
// 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 err = roomserverAPI.SendEventWithState(
|
|
|
|
context.Background(),
|
|
|
|
r.rsAPI,
|
|
|
|
roomserverAPI.KindNew,
|
|
|
|
respState,
|
|
|
|
event.Headered(respMakeJoin.RoomVersion),
|
|
|
|
serverName,
|
|
|
|
nil,
|
|
|
|
false,
|
|
|
|
); err != nil {
|
|
|
|
return fmt.Errorf("roomserverAPI.SendEventWithState: %w", err)
|
|
|
|
}
|
2020-05-15 14:55:14 +02:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
2021-01-22 15:55:08 +01:00
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformOutboundPeekRequest implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformOutboundPeek(
|
2021-01-22 15:55:08 +01:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformOutboundPeekRequest,
|
|
|
|
response *api.PerformOutboundPeekResponse,
|
|
|
|
) error {
|
|
|
|
// 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.
|
|
|
|
seenSet := make(map[gomatrixserverlib.ServerName]bool)
|
|
|
|
var uniqueList []gomatrixserverlib.ServerName
|
|
|
|
for _, srv := range request.ServerNames {
|
|
|
|
if seenSet[srv] {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
seenSet[srv] = true
|
|
|
|
uniqueList = append(uniqueList, srv)
|
|
|
|
}
|
|
|
|
request.ServerNames = uniqueList
|
|
|
|
|
|
|
|
// See if there's an existing outbound peek for this room ID with
|
|
|
|
// one of the specified servers.
|
|
|
|
if peeks, err := r.db.GetOutboundPeeks(ctx, request.RoomID); err == nil {
|
|
|
|
for _, peek := range peeks {
|
|
|
|
if _, ok := seenSet[peek.ServerName]; ok {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try each server that we were provided until we land on one that
|
|
|
|
// successfully completes the peek
|
|
|
|
var lastErr error
|
|
|
|
for _, serverName := range request.ServerNames {
|
|
|
|
if err := r.performOutboundPeekUsingServer(
|
|
|
|
ctx,
|
|
|
|
request.RoomID,
|
|
|
|
serverName,
|
|
|
|
supportedVersions,
|
|
|
|
); err != nil {
|
|
|
|
logrus.WithError(err).WithFields(logrus.Fields{
|
|
|
|
"server_name": serverName,
|
|
|
|
"room_id": request.RoomID,
|
|
|
|
}).Warnf("Failed to peek room through server")
|
|
|
|
lastErr = err
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// We're all good.
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// If we reach here then we didn't complete a peek for some reason.
|
|
|
|
var httpErr gomatrix.HTTPError
|
|
|
|
if ok := errors.As(lastErr, &httpErr); ok {
|
|
|
|
httpErr.Message = string(httpErr.Contents)
|
|
|
|
// Clear the wrapped error, else serialising to JSON (in polylith mode) will fail
|
|
|
|
httpErr.WrappedError = nil
|
|
|
|
response.LastError = &httpErr
|
|
|
|
} else {
|
|
|
|
response.LastError = &gomatrix.HTTPError{
|
|
|
|
Code: 0,
|
|
|
|
WrappedError: nil,
|
|
|
|
Message: lastErr.Error(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
logrus.Errorf(
|
|
|
|
"failed to peek room %q through %d server(s): last error %s",
|
|
|
|
request.RoomID, len(request.ServerNames), lastErr,
|
|
|
|
)
|
|
|
|
|
|
|
|
return lastErr
|
|
|
|
}
|
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
func (r *FederationInternalAPI) performOutboundPeekUsingServer(
|
2021-01-22 15:55:08 +01:00
|
|
|
ctx context.Context,
|
|
|
|
roomID string,
|
|
|
|
serverName gomatrixserverlib.ServerName,
|
|
|
|
supportedVersions []gomatrixserverlib.RoomVersion,
|
|
|
|
) error {
|
|
|
|
// create a unique ID for this peek.
|
|
|
|
// for now we just use the room ID again. In future, if we ever
|
|
|
|
// support concurrent peeks to the same room with different filters
|
|
|
|
// then we would need to disambiguate further.
|
|
|
|
peekID := roomID
|
|
|
|
|
|
|
|
// check whether we're peeking already to try to avoid needlessly
|
|
|
|
// re-peeking on the server. we don't need a transaction for this,
|
|
|
|
// given this is a nice-to-have.
|
|
|
|
outboundPeek, err := r.db.GetOutboundPeek(ctx, serverName, roomID, peekID)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
renewing := false
|
|
|
|
if outboundPeek != nil {
|
|
|
|
nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
|
|
|
|
if nowMilli > outboundPeek.RenewedTimestamp+outboundPeek.RenewalInterval {
|
|
|
|
logrus.Infof("stale outbound peek to %s for %s already exists; renewing", serverName, roomID)
|
|
|
|
renewing = true
|
|
|
|
} else {
|
|
|
|
logrus.Infof("live outbound peek to %s for %s already exists", serverName, roomID)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try to perform an outbound /peek using the information supplied in the
|
|
|
|
// request.
|
|
|
|
respPeek, err := r.federation.Peek(
|
|
|
|
ctx,
|
|
|
|
serverName,
|
|
|
|
roomID,
|
|
|
|
peekID,
|
|
|
|
supportedVersions,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
r.statistics.ForServer(serverName).Failure()
|
|
|
|
return fmt.Errorf("r.federation.Peek: %w", err)
|
|
|
|
}
|
|
|
|
r.statistics.ForServer(serverName).Success()
|
|
|
|
|
|
|
|
// Work out if we support the room version that has been supplied in
|
|
|
|
// the peek response.
|
|
|
|
if respPeek.RoomVersion == "" {
|
|
|
|
respPeek.RoomVersion = gomatrixserverlib.RoomVersionV1
|
|
|
|
}
|
|
|
|
if _, err = respPeek.RoomVersion.EventFormat(); err != nil {
|
|
|
|
return fmt.Errorf("respPeek.RoomVersion.EventFormat: %w", err)
|
|
|
|
}
|
|
|
|
|
2021-01-22 18:16:35 +01:00
|
|
|
// we have the peek state now so let's process regardless of whether upstream gives up
|
|
|
|
ctx = context.Background()
|
|
|
|
respState := respPeek.ToRespState()
|
2022-04-05 15:43:44 +02:00
|
|
|
|
2021-01-22 18:16:35 +01:00
|
|
|
// authenticate the state returned (check its auth events etc)
|
2021-01-22 15:55:08 +01:00
|
|
|
// the equivalent of CheckSendJoinResponse()
|
2022-04-05 15:43:44 +02:00
|
|
|
authEvents, _, err := respState.Check(ctx, respPeek.RoomVersion, r.keyRing, federatedAuthProvider(ctx, r.federation, r.keyRing, serverName))
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error checking state returned from peeking: %w", err)
|
|
|
|
}
|
2022-02-09 21:31:24 +01:00
|
|
|
if err = sanityCheckAuthChain(authEvents); err != nil {
|
2021-01-22 18:16:35 +01:00
|
|
|
return fmt.Errorf("sanityCheckAuthChain: %w", err)
|
|
|
|
}
|
2021-01-22 15:55:08 +01:00
|
|
|
|
|
|
|
// If we've got this far, the remote server is peeking.
|
|
|
|
if renewing {
|
|
|
|
if err = r.db.RenewOutboundPeek(ctx, serverName, roomID, peekID, respPeek.RenewalInterval); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if err = r.db.AddOutboundPeek(ctx, serverName, roomID, peekID, respPeek.RenewalInterval); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// logrus.Warnf("got respPeek %#v", respPeek)
|
|
|
|
// Send the newly returned state to the roomserver to update our local view.
|
|
|
|
if err = roomserverAPI.SendEventWithState(
|
|
|
|
ctx, r.rsAPI,
|
|
|
|
roomserverAPI.KindNew,
|
|
|
|
&respState,
|
|
|
|
respPeek.LatestEvent.Headered(respPeek.RoomVersion),
|
2022-01-27 15:29:14 +01:00
|
|
|
serverName,
|
|
|
|
nil,
|
|
|
|
false,
|
2021-01-22 15:55:08 +01:00
|
|
|
); err != nil {
|
|
|
|
return fmt.Errorf("r.producer.SendEventWithState: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
2020-05-15 14:55:14 +02:00
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformLeaveRequest implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformLeave(
|
2020-04-29 12:34:31 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformLeaveRequest,
|
|
|
|
response *api.PerformLeaveResponse,
|
|
|
|
) (err error) {
|
2020-05-04 19:34:09 +02:00
|
|
|
// Deduplicate the server names we were provided.
|
2020-05-07 18:14:32 +02:00
|
|
|
util.SortAndUnique(request.ServerNames)
|
2020-05-04 19:34:09 +02:00
|
|
|
|
|
|
|
// Try each server that we were provided until we land on one that
|
|
|
|
// successfully completes the make-leave send-leave dance.
|
|
|
|
for _, serverName := range request.ServerNames {
|
|
|
|
// Try to perform a make_leave using the information supplied in the
|
|
|
|
// request.
|
|
|
|
respMakeLeave, err := r.federation.MakeLeave(
|
|
|
|
ctx,
|
|
|
|
serverName,
|
|
|
|
request.RoomID,
|
|
|
|
request.UserID,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
// TODO: Check if the user was not allowed to leave the room.
|
|
|
|
logrus.WithError(err).Warnf("r.federation.MakeLeave failed")
|
Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007)
* Improve federation sender performance and behaviour, add backoff
* Tweaks
* Tweaks
* Tweaks
* Take copies of events before passing to destination queues
* Don't accidentally drop queued messages
* Don't take copies again
* Tidy up a bit
* Break out statistics (tracked component-wide), report success and failures from Perform actions
* Fix comment, use atomic add
* Improve logic a bit, don't block on wakeup, move idle check
* Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc
* Dedupe destinations, fix other bug hopefully
* Dispatch sends again
* Federation sender to ignore invites that are destined locally
* Loopback invite events
* Remodel a bit with channels
* Linter
* Only loopback invite event if we know the room
* We should tell other resident servers about the invite if we know about the room
* Correct invite signing
* Fix invite loopback
* Check HTTP response codes, push new invites to front of queue
* Review comments
2020-05-07 13:42:06 +02:00
|
|
|
r.statistics.ForServer(serverName).Failure()
|
2020-05-04 19:34:09 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Set all the fields to be what they should be, this should be a no-op
|
|
|
|
// but it's possible that the remote server returned us something "odd"
|
|
|
|
respMakeLeave.LeaveEvent.Type = gomatrixserverlib.MRoomMember
|
|
|
|
respMakeLeave.LeaveEvent.Sender = request.UserID
|
|
|
|
respMakeLeave.LeaveEvent.StateKey = &request.UserID
|
|
|
|
respMakeLeave.LeaveEvent.RoomID = request.RoomID
|
|
|
|
respMakeLeave.LeaveEvent.Redacts = ""
|
|
|
|
if respMakeLeave.LeaveEvent.Content == nil {
|
|
|
|
content := map[string]interface{}{
|
|
|
|
"membership": "leave",
|
|
|
|
}
|
|
|
|
if err = respMakeLeave.LeaveEvent.SetContent(content); err != nil {
|
|
|
|
logrus.WithError(err).Warnf("respMakeLeave.LeaveEvent.SetContent failed")
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if err = respMakeLeave.LeaveEvent.SetUnsigned(struct{}{}); err != nil {
|
|
|
|
logrus.WithError(err).Warnf("respMakeLeave.LeaveEvent.SetUnsigned failed")
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Work out if we support the room version that has been supplied in
|
|
|
|
// the make_leave response.
|
|
|
|
if _, err = respMakeLeave.RoomVersion.EventFormat(); err != nil {
|
|
|
|
return gomatrixserverlib.UnsupportedRoomVersionError{}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Build the leave event.
|
|
|
|
event, err := respMakeLeave.LeaveEvent.Build(
|
|
|
|
time.Now(),
|
|
|
|
r.cfg.Matrix.ServerName,
|
|
|
|
r.cfg.Matrix.KeyID,
|
|
|
|
r.cfg.Matrix.PrivateKey,
|
|
|
|
respMakeLeave.RoomVersion,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
logrus.WithError(err).Warnf("respMakeLeave.LeaveEvent.Build failed")
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Try to perform a send_leave using the newly built event.
|
|
|
|
err = r.federation.SendLeave(
|
|
|
|
ctx,
|
|
|
|
serverName,
|
|
|
|
event,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
logrus.WithError(err).Warnf("r.federation.SendLeave failed")
|
Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007)
* Improve federation sender performance and behaviour, add backoff
* Tweaks
* Tweaks
* Tweaks
* Take copies of events before passing to destination queues
* Don't accidentally drop queued messages
* Don't take copies again
* Tidy up a bit
* Break out statistics (tracked component-wide), report success and failures from Perform actions
* Fix comment, use atomic add
* Improve logic a bit, don't block on wakeup, move idle check
* Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc
* Dedupe destinations, fix other bug hopefully
* Dispatch sends again
* Federation sender to ignore invites that are destined locally
* Loopback invite events
* Remodel a bit with channels
* Linter
* Only loopback invite event if we know the room
* We should tell other resident servers about the invite if we know about the room
* Correct invite signing
* Fix invite loopback
* Check HTTP response codes, push new invites to front of queue
* Review comments
2020-05-07 13:42:06 +02:00
|
|
|
r.statistics.ForServer(serverName).Failure()
|
2020-05-04 19:34:09 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007)
* Improve federation sender performance and behaviour, add backoff
* Tweaks
* Tweaks
* Tweaks
* Take copies of events before passing to destination queues
* Don't accidentally drop queued messages
* Don't take copies again
* Tidy up a bit
* Break out statistics (tracked component-wide), report success and failures from Perform actions
* Fix comment, use atomic add
* Improve logic a bit, don't block on wakeup, move idle check
* Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc
* Dedupe destinations, fix other bug hopefully
* Dispatch sends again
* Federation sender to ignore invites that are destined locally
* Loopback invite events
* Remodel a bit with channels
* Linter
* Only loopback invite event if we know the room
* We should tell other resident servers about the invite if we know about the room
* Correct invite signing
* Fix invite loopback
* Check HTTP response codes, push new invites to front of queue
* Review comments
2020-05-07 13:42:06 +02:00
|
|
|
r.statistics.ForServer(serverName).Success()
|
2020-05-04 19:34:09 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// If we reach here then we didn't complete a leave for some reason.
|
|
|
|
return fmt.Errorf(
|
2021-09-08 18:31:03 +02:00
|
|
|
"failed to leave room %q through %d server(s)",
|
2020-05-04 19:34:09 +02:00
|
|
|
request.RoomID, len(request.ServerNames),
|
|
|
|
)
|
2020-04-29 12:34:31 +02:00
|
|
|
}
|
2020-06-01 19:34:08 +02:00
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformLeaveRequest implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformInvite(
|
2020-08-17 12:40:49 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformInviteRequest,
|
|
|
|
response *api.PerformInviteResponse,
|
|
|
|
) (err error) {
|
|
|
|
if request.Event.StateKey() == nil {
|
|
|
|
return errors.New("invite must be a state event")
|
|
|
|
}
|
|
|
|
|
|
|
|
_, destination, err := gomatrixserverlib.SplitID('@', *request.Event.StateKey())
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("gomatrixserverlib.SplitID: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
logrus.WithFields(logrus.Fields{
|
|
|
|
"event_id": request.Event.EventID(),
|
|
|
|
"user_id": *request.Event.StateKey(),
|
|
|
|
"room_id": request.Event.RoomID(),
|
|
|
|
"room_version": request.RoomVersion,
|
|
|
|
"destination": destination,
|
|
|
|
}).Info("Sending invite")
|
|
|
|
|
2020-11-16 16:44:53 +01:00
|
|
|
inviteReq, err := gomatrixserverlib.NewInviteV2Request(request.Event, request.InviteRoomState)
|
2020-08-17 12:40:49 +02:00
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("gomatrixserverlib.NewInviteV2Request: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
inviteRes, err := r.federation.SendInviteV2(ctx, destination, inviteReq)
|
|
|
|
if err != nil {
|
2022-02-09 21:31:24 +01:00
|
|
|
return fmt.Errorf("r.federation.SendInviteV2: failed to send invite: %w", err)
|
2020-08-17 12:40:49 +02:00
|
|
|
}
|
|
|
|
|
2022-02-09 21:31:24 +01:00
|
|
|
inviteEvent, err := inviteRes.Event.UntrustedEvent(request.RoomVersion)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("r.federation.SendInviteV2 failed to decode event response: %w", err)
|
|
|
|
}
|
|
|
|
response.Event = inviteEvent.Headered(request.RoomVersion)
|
2020-08-17 12:40:49 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformServersAlive implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformServersAlive(
|
2020-06-01 19:34:08 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformServersAliveRequest,
|
|
|
|
response *api.PerformServersAliveResponse,
|
|
|
|
) (err error) {
|
|
|
|
for _, srv := range request.Servers {
|
2021-05-24 12:43:24 +02:00
|
|
|
_ = r.db.RemoveServerFromBlacklist(srv)
|
2020-06-01 19:34:08 +02:00
|
|
|
r.queues.RetryServer(srv)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
2020-07-16 14:52:08 +02:00
|
|
|
|
2021-11-24 11:45:23 +01:00
|
|
|
// PerformServersAlive implements api.FederationInternalAPI
|
|
|
|
func (r *FederationInternalAPI) PerformBroadcastEDU(
|
2020-07-16 14:52:08 +02:00
|
|
|
ctx context.Context,
|
|
|
|
request *api.PerformBroadcastEDURequest,
|
|
|
|
response *api.PerformBroadcastEDUResponse,
|
|
|
|
) (err error) {
|
|
|
|
destinations, err := r.db.GetAllJoinedHosts(ctx)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("r.db.GetAllJoinedHosts: %w", err)
|
|
|
|
}
|
2020-08-06 17:00:42 +02:00
|
|
|
if len(destinations) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
logrus.WithContext(ctx).Infof("Sending wake-up EDU to %d destination(s)", len(destinations))
|
2020-07-16 14:52:08 +02:00
|
|
|
|
|
|
|
edu := &gomatrixserverlib.EDU{
|
|
|
|
Type: "org.matrix.dendrite.wakeup",
|
|
|
|
Origin: string(r.cfg.Matrix.ServerName),
|
|
|
|
}
|
|
|
|
if err = r.queues.SendEDU(edu, r.cfg.Matrix.ServerName, destinations); err != nil {
|
|
|
|
return fmt.Errorf("r.queues.SendEDU: %w", err)
|
|
|
|
}
|
|
|
|
|
2020-08-06 17:00:42 +02:00
|
|
|
wakeReq := &api.PerformServersAliveRequest{
|
|
|
|
Servers: destinations,
|
|
|
|
}
|
|
|
|
wakeRes := &api.PerformServersAliveResponse{}
|
|
|
|
if err := r.PerformServersAlive(ctx, wakeReq, wakeRes); err != nil {
|
|
|
|
return fmt.Errorf("r.PerformServersAlive: %w", err)
|
|
|
|
}
|
|
|
|
|
2020-07-16 14:52:08 +02:00
|
|
|
return nil
|
|
|
|
}
|
2020-10-10 01:21:15 +02:00
|
|
|
|
2021-01-22 18:16:35 +01:00
|
|
|
func sanityCheckAuthChain(authChain []*gomatrixserverlib.Event) error {
|
2020-10-10 01:21:15 +02:00
|
|
|
// sanity check we have a create event and it has a known room version
|
2021-01-22 18:16:35 +01:00
|
|
|
for _, ev := range authChain {
|
2020-10-10 01:21:15 +02:00
|
|
|
if ev.Type() == gomatrixserverlib.MRoomCreate && ev.StateKeyEquals("") {
|
|
|
|
// make sure the room version is known
|
|
|
|
content := ev.Content()
|
|
|
|
verBody := struct {
|
|
|
|
Version string `json:"room_version"`
|
|
|
|
}{}
|
|
|
|
err := json.Unmarshal(content, &verBody)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if verBody.Version == "" {
|
|
|
|
// https://matrix.org/docs/spec/client_server/r0.6.0#m-room-create
|
|
|
|
// The version of the room. Defaults to "1" if the key does not exist.
|
|
|
|
verBody.Version = "1"
|
|
|
|
}
|
|
|
|
knownVersions := gomatrixserverlib.RoomVersions()
|
|
|
|
if _, ok := knownVersions[gomatrixserverlib.RoomVersion(verBody.Version)]; !ok {
|
2021-01-22 18:16:35 +01:00
|
|
|
return fmt.Errorf("auth chain m.room.create event has an unknown room version: %s", verBody.Version)
|
2020-10-10 01:21:15 +02:00
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
2021-01-22 18:16:35 +01:00
|
|
|
return fmt.Errorf("auth chain response is missing m.room.create event")
|
2020-10-10 01:21:15 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func setDefaultRoomVersionFromJoinEvent(joinEvent gomatrixserverlib.EventBuilder) 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
|
|
|
|
}
|
2021-01-22 18:16:35 +01:00
|
|
|
|
|
|
|
// FederatedAuthProvider is an auth chain provider which fetches events from the server provided
|
|
|
|
func federatedAuthProvider(
|
|
|
|
ctx context.Context, federation *gomatrixserverlib.FederationClient,
|
|
|
|
keyRing gomatrixserverlib.JSONVerifier, server gomatrixserverlib.ServerName,
|
|
|
|
) gomatrixserverlib.AuthChainProvider {
|
|
|
|
// 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][]*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
|
|
|
|
// 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{}
|
|
|
|
|
|
|
|
// See if we have retry entries for each of the supplied event IDs.
|
|
|
|
for _, eventID := range eventIDs {
|
|
|
|
// If we've already satisfied a request for this event ID before then
|
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
returning = append(returning, retry...)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Make a note of the fact that we tried to do something with this
|
|
|
|
// event ID, even if we don't succeed.
|
|
|
|
retries[eventID] = nil
|
|
|
|
|
|
|
|
// Try to retrieve the event from the server that sent us the send
|
|
|
|
// join response.
|
|
|
|
tx, txerr := federation.GetEvent(ctx, server, eventID)
|
|
|
|
if txerr != nil {
|
|
|
|
return nil, fmt.Errorf("missingAuth r.federation.GetEvent: %w", txerr)
|
|
|
|
}
|
|
|
|
|
|
|
|
// For each event returned, add it to the set of return events. We
|
|
|
|
// also will populate the retries, in case someone asks for this
|
|
|
|
// event ID again.
|
|
|
|
for _, pdu := range tx.PDUs {
|
|
|
|
// Try to parse the event.
|
|
|
|
ev, everr := gomatrixserverlib.NewEventFromUntrustedJSON(pdu, roomVersion)
|
|
|
|
if everr != nil {
|
|
|
|
return nil, fmt.Errorf("missingAuth gomatrixserverlib.NewEventFromUntrustedJSON: %w", everr)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check the signatures of the event.
|
2021-11-02 11:13:38 +01:00
|
|
|
if err := ev.VerifyEventSignatures(ctx, keyRing); err != nil {
|
2021-01-22 18:16:35 +01:00
|
|
|
return nil, fmt.Errorf("missingAuth VerifyEventSignatures: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// If the event is OK then add it to the results and the retry map.
|
|
|
|
returning = append(returning, ev)
|
|
|
|
retries[ev.EventID()] = append(retries[ev.EventID()], ev)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return returning, nil
|
|
|
|
}
|
|
|
|
}
|