mirror of
https://github.com/hoernschen/dendrite.git
synced 2025-07-29 12:42:46 +00:00
Move Invite logic to GMSL (#3086)
This is both the federation receiving & sending side logic (which were previously entangeld in a single function)
This commit is contained in:
devonh
GitHub
parent
cbdc601f1b
commit
ea6b368ad4
15 changed files with 359 additions and 442 deletions
|
|
@ -376,7 +376,7 @@ func (c *Creator) PerformCreateRoom(ctx context.Context, userID spec.UserID, roo
|
|||
// If this is a direct message then we should invite the participants.
|
||||
if len(createRequest.InvitedUsers) > 0 {
|
||||
// Build some stripped state for the invite.
|
||||
var globalStrippedState []fclient.InviteV2StrippedState
|
||||
var globalStrippedState []gomatrixserverlib.InviteStrippedState
|
||||
for _, event := range builtEvents {
|
||||
// Chosen events from the spec:
|
||||
// https://spec.matrix.org/v1.3/client-server-api/#stripped-state
|
||||
|
|
@ -399,7 +399,7 @@ func (c *Creator) PerformCreateRoom(ctx context.Context, userID spec.UserID, roo
|
|||
ev := event.PDU
|
||||
globalStrippedState = append(
|
||||
globalStrippedState,
|
||||
fclient.NewInviteV2StrippedState(ev),
|
||||
gomatrixserverlib.NewInviteStrippedState(ev),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -443,7 +443,7 @@ func (c *Creator) PerformCreateRoom(ctx context.Context, userID spec.UserID, roo
|
|||
}
|
||||
inviteStrippedState := append(
|
||||
globalStrippedState,
|
||||
fclient.NewInviteV2StrippedState(inviteEvent.PDU),
|
||||
gomatrixserverlib.NewInviteStrippedState(inviteEvent.PDU),
|
||||
)
|
||||
// Send the invite event to the roomserver.
|
||||
event := inviteEvent
|
||||
|
|
|
|||
|
|
@ -28,186 +28,149 @@ import (
|
|||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/dendrite/setup/config"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/gomatrixserverlib/fclient"
|
||||
"github.com/matrix-org/gomatrixserverlib/spec"
|
||||
"github.com/matrix-org/util"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
type QueryState struct {
|
||||
storage.Database
|
||||
}
|
||||
|
||||
func (q *QueryState) GetAuthEvents(ctx context.Context, event gomatrixserverlib.PDU) (gomatrixserverlib.AuthEventProvider, error) {
|
||||
return helpers.GetAuthEvents(ctx, q.Database, event.Version(), event, event.AuthEventIDs())
|
||||
}
|
||||
|
||||
func (q *QueryState) GetState(ctx context.Context, roomID spec.RoomID, stateWanted []gomatrixserverlib.StateKeyTuple) ([]gomatrixserverlib.PDU, error) {
|
||||
info, err := q.Database.RoomInfo(ctx, roomID.String())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to load RoomInfo: %w", err)
|
||||
}
|
||||
if info != nil {
|
||||
roomState := state.NewStateResolution(q.Database, info)
|
||||
stateEntries, err := roomState.LoadStateAtSnapshotForStringTuples(
|
||||
ctx, info.StateSnapshotNID(), stateWanted,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil
|
||||
}
|
||||
stateNIDs := []types.EventNID{}
|
||||
for _, stateNID := range stateEntries {
|
||||
stateNIDs = append(stateNIDs, stateNID.EventNID)
|
||||
}
|
||||
stateEvents, err := q.Database.Events(ctx, info.RoomVersion, stateNIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to obtain required events: %w", err)
|
||||
}
|
||||
|
||||
events := []gomatrixserverlib.PDU{}
|
||||
for _, event := range stateEvents {
|
||||
events = append(events, event.PDU)
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
type Inviter struct {
|
||||
DB storage.Database
|
||||
Cfg *config.RoomServer
|
||||
FSAPI federationAPI.RoomserverFederationAPI
|
||||
RSAPI api.RoomserverInternalAPI
|
||||
Inputer *input.Inputer
|
||||
}
|
||||
|
||||
func (r *Inviter) IsKnownRoom(ctx context.Context, roomID spec.RoomID) (bool, error) {
|
||||
info, err := r.DB.RoomInfo(ctx, roomID.String())
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("failed to load RoomInfo: %w", err)
|
||||
}
|
||||
return (info != nil && !info.IsStub()), nil
|
||||
}
|
||||
|
||||
func (r *Inviter) StateQuerier() gomatrixserverlib.StateQuerier {
|
||||
return &QueryState{Database: r.DB}
|
||||
}
|
||||
|
||||
func (r *Inviter) ProcessInviteMembership(
|
||||
ctx context.Context, inviteEvent *types.HeaderedEvent,
|
||||
) ([]api.OutputEvent, error) {
|
||||
var outputUpdates []api.OutputEvent
|
||||
var updater *shared.MembershipUpdater
|
||||
_, domain, err := gomatrixserverlib.SplitID('@', *inviteEvent.StateKey())
|
||||
if err != nil {
|
||||
return nil, api.ErrInvalidID{Err: fmt.Errorf("the user ID %s is invalid", *inviteEvent.StateKey())}
|
||||
}
|
||||
isTargetLocal := r.Cfg.Matrix.IsLocalServerName(domain)
|
||||
if updater, err = r.DB.MembershipUpdater(ctx, inviteEvent.RoomID(), *inviteEvent.StateKey(), isTargetLocal, inviteEvent.Version()); err != nil {
|
||||
return nil, fmt.Errorf("r.DB.MembershipUpdater: %w", err)
|
||||
}
|
||||
outputUpdates, err = helpers.UpdateToInviteMembership(updater, &types.Event{
|
||||
EventNID: 0,
|
||||
PDU: inviteEvent.PDU,
|
||||
}, outputUpdates, inviteEvent.Version())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("updateToInviteMembership: %w", err)
|
||||
}
|
||||
if err = updater.Commit(); err != nil {
|
||||
return nil, fmt.Errorf("updater.Commit: %w", err)
|
||||
}
|
||||
return outputUpdates, nil
|
||||
}
|
||||
|
||||
// nolint:gocyclo
|
||||
func (r *Inviter) PerformInvite(
|
||||
ctx context.Context,
|
||||
req *api.PerformInviteRequest,
|
||||
) ([]api.OutputEvent, error) {
|
||||
var outputUpdates []api.OutputEvent
|
||||
) error {
|
||||
event := req.Event
|
||||
|
||||
sender, err := spec.NewUserID(event.Sender(), true)
|
||||
if err != nil {
|
||||
return spec.InvalidParam("The user ID is invalid")
|
||||
}
|
||||
if !r.Cfg.Matrix.IsLocalServerName(sender.Domain()) {
|
||||
return api.ErrInvalidID{Err: fmt.Errorf("the invite must be from a local user")}
|
||||
}
|
||||
|
||||
if event.StateKey() == nil {
|
||||
return nil, fmt.Errorf("invite must be a state event")
|
||||
return fmt.Errorf("invite must be a state event")
|
||||
}
|
||||
_, senderDomain, err := gomatrixserverlib.SplitID('@', event.Sender())
|
||||
invitedUser, err := spec.NewUserID(*event.StateKey(), true)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("sender %q is invalid", event.Sender())
|
||||
return spec.InvalidParam("The user ID is invalid")
|
||||
}
|
||||
isTargetLocal := r.Cfg.Matrix.IsLocalServerName(invitedUser.Domain())
|
||||
|
||||
roomID := event.RoomID()
|
||||
targetUserID := *event.StateKey()
|
||||
info, err := r.DB.RoomInfo(ctx, roomID)
|
||||
validRoomID, err := spec.NewRoomID(event.RoomID())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to load RoomInfo: %w", err)
|
||||
return err
|
||||
}
|
||||
|
||||
_, domain, err := gomatrixserverlib.SplitID('@', targetUserID)
|
||||
input := gomatrixserverlib.PerformInviteInput{
|
||||
RoomID: *validRoomID,
|
||||
InviteEvent: event.PDU,
|
||||
InvitedUser: *invitedUser,
|
||||
IsTargetLocal: isTargetLocal,
|
||||
StrippedState: req.InviteRoomState,
|
||||
MembershipQuerier: &api.MembershipQuerier{Roomserver: r.RSAPI},
|
||||
StateQuerier: &QueryState{r.DB},
|
||||
}
|
||||
inviteEvent, err := gomatrixserverlib.PerformInvite(ctx, input, r.FSAPI)
|
||||
if err != nil {
|
||||
return nil, api.ErrInvalidID{Err: fmt.Errorf("the user ID %s is invalid", targetUserID)}
|
||||
}
|
||||
isTargetLocal := r.Cfg.Matrix.IsLocalServerName(domain)
|
||||
isOriginLocal := r.Cfg.Matrix.IsLocalServerName(senderDomain)
|
||||
if !isOriginLocal && !isTargetLocal {
|
||||
return nil, api.ErrInvalidID{Err: fmt.Errorf("the invite must be either from or to a local user")}
|
||||
switch e := err.(type) {
|
||||
case spec.MatrixError:
|
||||
if e.ErrCode == spec.ErrorForbidden {
|
||||
return api.ErrNotAllowed{Err: fmt.Errorf("%s", e.Err)}
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
logger := util.GetLogger(ctx).WithFields(map[string]interface{}{
|
||||
"inviter": event.Sender(),
|
||||
"invitee": *event.StateKey(),
|
||||
"room_id": roomID,
|
||||
"event_id": event.EventID(),
|
||||
})
|
||||
logger.WithFields(log.Fields{
|
||||
"room_version": req.RoomVersion,
|
||||
"room_info_exists": info != nil,
|
||||
"target_local": isTargetLocal,
|
||||
"origin_local": isOriginLocal,
|
||||
}).Debug("processing invite event")
|
||||
|
||||
inviteState := req.InviteRoomState
|
||||
if len(inviteState) == 0 && info != nil {
|
||||
var is []fclient.InviteV2StrippedState
|
||||
if is, err = buildInviteStrippedState(ctx, r.DB, info, req); err == nil {
|
||||
inviteState = is
|
||||
}
|
||||
}
|
||||
if len(inviteState) == 0 {
|
||||
if err = event.SetUnsignedField("invite_room_state", struct{}{}); err != nil {
|
||||
return nil, fmt.Errorf("event.SetUnsignedField: %w", err)
|
||||
}
|
||||
} else {
|
||||
if err = event.SetUnsignedField("invite_room_state", inviteState); err != nil {
|
||||
return nil, fmt.Errorf("event.SetUnsignedField: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
updateMembershipTableManually := func() ([]api.OutputEvent, error) {
|
||||
var updater *shared.MembershipUpdater
|
||||
if updater, err = r.DB.MembershipUpdater(ctx, roomID, targetUserID, isTargetLocal, req.RoomVersion); err != nil {
|
||||
return nil, fmt.Errorf("r.DB.MembershipUpdater: %w", err)
|
||||
}
|
||||
outputUpdates, err = helpers.UpdateToInviteMembership(updater, &types.Event{
|
||||
EventNID: 0,
|
||||
PDU: event.PDU,
|
||||
}, outputUpdates, req.Event.Version())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("updateToInviteMembership: %w", err)
|
||||
}
|
||||
if err = updater.Commit(); err != nil {
|
||||
return nil, fmt.Errorf("updater.Commit: %w", err)
|
||||
}
|
||||
logger.Debugf("updated membership to invite and sending invite OutputEvent")
|
||||
return outputUpdates, nil
|
||||
}
|
||||
|
||||
if (info == nil || info.IsStub()) && !isOriginLocal && isTargetLocal {
|
||||
// The invite came in over federation for a room that we don't know about
|
||||
// yet. We need to handle this a bit differently to most invites because
|
||||
// we don't know the room state, therefore the roomserver can't process
|
||||
// an input event. Instead we will update the membership table with the
|
||||
// new invite and generate an output event.
|
||||
return updateMembershipTableManually()
|
||||
}
|
||||
|
||||
var isAlreadyJoined bool
|
||||
if info != nil {
|
||||
_, isAlreadyJoined, _, err = r.DB.GetMembership(ctx, info.RoomNID, *event.StateKey())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("r.DB.GetMembership: %w", err)
|
||||
}
|
||||
}
|
||||
if isAlreadyJoined {
|
||||
// If the user is joined to the room then that takes precedence over this
|
||||
// invite event. It makes little sense to move a user that is already
|
||||
// joined to the room into the invite state.
|
||||
// This could plausibly happen if an invite request raced with a join
|
||||
// request for a user. For example if a user was invited to a public
|
||||
// room and they joined the room at the same time as the invite was sent.
|
||||
// The other way this could plausibly happen is if an invite raced with
|
||||
// a kick. For example if a user was kicked from a room in error and in
|
||||
// response someone else in the room re-invited them then it is possible
|
||||
// for the invite request to race with the leave event so that the
|
||||
// target receives invite before it learns that it has been kicked.
|
||||
// There are a few ways this could be plausibly handled in the roomserver.
|
||||
// 1) Store the invite, but mark it as retired. That will result in the
|
||||
// permanent rejection of that invite event. So even if the target
|
||||
// user leaves the room and the invite is retransmitted it will be
|
||||
// ignored. However a new invite with a new event ID would still be
|
||||
// accepted.
|
||||
// 2) Silently discard the invite event. This means that if the event
|
||||
// was retransmitted at a later date after the target user had left
|
||||
// the room we would accept the invite. However since we hadn't told
|
||||
// the sending server that the invite had been discarded it would
|
||||
// have no reason to attempt to retry.
|
||||
// 3) Signal the sending server that the user is already joined to the
|
||||
// room.
|
||||
// For now we will implement option 2. Since in the abesence of a retry
|
||||
// mechanism it will be equivalent to option 1, and we don't have a
|
||||
// signalling mechanism to implement option 3.
|
||||
logger.Debugf("user already joined")
|
||||
return nil, api.ErrNotAllowed{Err: fmt.Errorf("user is already joined to room")}
|
||||
}
|
||||
|
||||
// If the invite originated remotely then we can't send an
|
||||
// InputRoomEvent for the invite as it will never pass auth checks
|
||||
// due to lacking room state, but we still need to tell the client
|
||||
// about the invite so we can accept it, hence we return an output
|
||||
// event to send to the Sync API.
|
||||
if !isOriginLocal {
|
||||
return updateMembershipTableManually()
|
||||
}
|
||||
|
||||
// The invite originated locally. Therefore we have a responsibility to
|
||||
// try and see if the user is allowed to make this invite. We can't do
|
||||
// this for invites coming in over federation - we have to take those on
|
||||
// trust.
|
||||
_, err = helpers.CheckAuthEvents(ctx, r.DB, info, event, event.AuthEventIDs())
|
||||
if err != nil {
|
||||
logger.WithError(err).WithField("event_id", event.EventID()).WithField("auth_event_ids", event.AuthEventIDs()).Error(
|
||||
"processInviteEvent.checkAuthEvents failed for event",
|
||||
)
|
||||
return nil, api.ErrNotAllowed{Err: err}
|
||||
}
|
||||
|
||||
// If the invite originated from us and the target isn't local then we
|
||||
// should try and send the invite over federation first. It might be
|
||||
// that the remote user doesn't exist, in which case we can give up
|
||||
// processing here.
|
||||
if req.SendAsServer != api.DoNotSendToOtherServers && !isTargetLocal {
|
||||
fsReq := &federationAPI.PerformInviteRequest{
|
||||
RoomVersion: req.RoomVersion,
|
||||
Event: event,
|
||||
InviteRoomState: inviteState,
|
||||
}
|
||||
fsRes := &federationAPI.PerformInviteResponse{}
|
||||
if err = r.FSAPI.PerformInvite(ctx, fsReq, fsRes); err != nil {
|
||||
logger.WithError(err).WithField("event_id", event.EventID()).Error("r.FSAPI.PerformInvite failed")
|
||||
return nil, api.ErrNotAllowed{Err: err}
|
||||
}
|
||||
event = fsRes.Event
|
||||
logger.Debugf("Federated PerformInvite success with event ID %s", event.EventID())
|
||||
// Use the returned event if there was one (due to federation), otherwise
|
||||
// send the original invite event to the roomserver.
|
||||
if inviteEvent == nil {
|
||||
inviteEvent = event
|
||||
}
|
||||
|
||||
// Send the invite event to the roomserver input stream. This will
|
||||
|
|
@ -219,67 +182,18 @@ func (r *Inviter) PerformInvite(
|
|||
InputRoomEvents: []api.InputRoomEvent{
|
||||
{
|
||||
Kind: api.KindNew,
|
||||
Event: event,
|
||||
Origin: senderDomain,
|
||||
Event: &types.HeaderedEvent{PDU: inviteEvent},
|
||||
Origin: sender.Domain(),
|
||||
SendAsServer: req.SendAsServer,
|
||||
},
|
||||
},
|
||||
}
|
||||
inputRes := &api.InputRoomEventsResponse{}
|
||||
r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes)
|
||||
if err = inputRes.Err(); err != nil {
|
||||
logger.WithError(err).WithField("event_id", event.EventID()).Error("r.InputRoomEvents failed")
|
||||
return nil, api.ErrNotAllowed{Err: err}
|
||||
if err := inputRes.Err(); err != nil {
|
||||
util.GetLogger(ctx).WithField("event_id", event.EventID()).Error("r.InputRoomEvents failed")
|
||||
return api.ErrNotAllowed{Err: err}
|
||||
}
|
||||
|
||||
// Don't notify the sync api of this event in the same way as a federated invite so the invitee
|
||||
// gets the invite, as the roomserver will do this when it processes the m.room.member invite.
|
||||
return outputUpdates, nil
|
||||
}
|
||||
|
||||
func buildInviteStrippedState(
|
||||
ctx context.Context,
|
||||
db storage.Database,
|
||||
info *types.RoomInfo,
|
||||
input *api.PerformInviteRequest,
|
||||
) ([]fclient.InviteV2StrippedState, error) {
|
||||
stateWanted := []gomatrixserverlib.StateKeyTuple{}
|
||||
// "If they are set on the room, at least the state for m.room.avatar, m.room.canonical_alias, m.room.join_rules, and m.room.name SHOULD be included."
|
||||
// https://matrix.org/docs/spec/client_server/r0.6.0#m-room-member
|
||||
for _, t := range []string{
|
||||
spec.MRoomName, spec.MRoomCanonicalAlias,
|
||||
spec.MRoomJoinRules, spec.MRoomAvatar,
|
||||
spec.MRoomEncryption, spec.MRoomCreate,
|
||||
} {
|
||||
stateWanted = append(stateWanted, gomatrixserverlib.StateKeyTuple{
|
||||
EventType: t,
|
||||
StateKey: "",
|
||||
})
|
||||
}
|
||||
roomState := state.NewStateResolution(db, info)
|
||||
stateEntries, err := roomState.LoadStateAtSnapshotForStringTuples(
|
||||
ctx, info.StateSnapshotNID(), stateWanted,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stateNIDs := []types.EventNID{}
|
||||
for _, stateNID := range stateEntries {
|
||||
stateNIDs = append(stateNIDs, stateNID.EventNID)
|
||||
}
|
||||
if info == nil {
|
||||
return nil, types.ErrorInvalidRoomInfo
|
||||
}
|
||||
stateEvents, err := db.Events(ctx, info.RoomVersion, stateNIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
inviteState := []fclient.InviteV2StrippedState{
|
||||
fclient.NewInviteV2StrippedState(input.Event.PDU),
|
||||
}
|
||||
stateEvents = append(stateEvents, types.Event{PDU: input.Event.PDU})
|
||||
for _, event := range stateEvents {
|
||||
inviteState = append(inviteState, fclient.NewInviteV2StrippedState(event.PDU))
|
||||
}
|
||||
return inviteState, nil
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue