1f349/dendrite
4
0
Fork 0
mirror of https://github.com/1f349/dendrite.git synced 2024-12-22 16:24:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Roomserver perform join (#1001)

Browse Source 
* Add PerformJoin template

* Try roomserver perform join

* Send correct server name to FS API

* Pass through content, try to handle multiple server names

* Fix local server checks

* Don't refer to non-existent error

* Add directory lookups of aliases

* Remove unneeded parameters

* Don't repeat join events into the roomserver

* Unmarshal the content, that would help

* Check if the user is already in the room in the fedeationapi too

* Return incompatible room version error

* Use Membership, don't try more servers than needed

* Review comments, make FS API take list of servernames, dedupe them, break out of loop properly on success

* Tweaks
This commit is contained in:
Neil Alexander 2020-05-04 13:53:47 +01:00 committed by GitHub
parent 36bbb25561
commit 5c894efd0e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 506 additions and 406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
clientapi/jsonerror/jsonerror.go
View File

@ -18,6 +18,7 @@ import (
"fmt"
"net/http"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
@ -124,6 +125,12 @@ func GuestAccessForbidden(msg string) *MatrixError {
return &MatrixError{"M_GUEST_ACCESS_FORBIDDEN", msg}
}
// IncompatibleRoomVersion is an error which is returned when the client
// requests a room with a version that is unsupported.
func IncompatibleRoomVersion(roomVersion gomatrixserverlib.RoomVersion) *MatrixError {
return &MatrixError{"M_INCOMPATIBLE_ROOM_VERSION", string(roomVersion)}
}
// UnsupportedRoomVersion is an error which is returned when the client
// requests a room with a version that is unsupported.
func UnsupportedRoomVersion(msg string) *MatrixError {

320
clientapi/routing/joinroom.go
View File

@ -15,332 +15,48 @@
package routing
import (
"fmt"
"net/http"
"strings"
"time"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
"github.com/matrix-org/dendrite/clientapi/httputil"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
"github.com/matrix-org/dendrite/clientapi/producers"
"github.com/matrix-org/dendrite/common"
"github.com/matrix-org/dendrite/common/config"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
// JoinRoomByIDOrAlias implements the "/join/{roomIDOrAlias}" API.
// https://matrix.org/docs/spec/client_server/r0.2.0.html#post-matrix-client-r0-join-roomidoralias
func JoinRoomByIDOrAlias(
req *http.Request,
device *authtypes.Device,
roomIDOrAlias string,
cfg *config.Dendrite,
federation *gomatrixserverlib.FederationClient,
producer *producers.RoomserverProducer,
rsAPI roomserverAPI.RoomserverInternalAPI,
fsAPI federationSenderAPI.FederationSenderInternalAPI,
keyRing gomatrixserverlib.KeyRing,
accountDB accounts.Database,
roomIDOrAlias string,
) util.JSONResponse {
var content map[string]interface{} // must be a JSON object
if resErr := httputil.UnmarshalJSONRequest(req, &content); resErr != nil {
return *resErr
// Prepare to ask the roomserver to perform the room join.
joinReq := roomserverAPI.PerformJoinRequest{
RoomIDOrAlias: roomIDOrAlias,
UserID: device.UserID,
}
joinRes := roomserverAPI.PerformJoinResponse{}
// If content was provided in the request then incude that
// in the request. It'll get used as a part of the membership
// event content.
if err := httputil.UnmarshalJSONRequest(req, &joinReq.Content); err != nil {
return *err
}
evTime, err := httputil.ParseTSParam(req)
if err != nil {
// Ask the roomserver to perform the join.
if err := rsAPI.PerformJoin(req.Context(), &joinReq, &joinRes); err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.InvalidArgumentValue(err.Error()),
JSON: jsonerror.Unknown(err.Error()),
}
}
localpart, _, err := gomatrixserverlib.SplitID('@', device.UserID)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
return jsonerror.InternalServerError()
}
profile, err := accountDB.GetProfileByLocalpart(req.Context(), localpart)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("accountDB.GetProfileByLocalpart failed")
return jsonerror.InternalServerError()
}
content["membership"] = gomatrixserverlib.Join
content["displayname"] = profile.DisplayName
content["avatar_url"] = profile.AvatarURL
r := joinRoomReq{
req, evTime, content, device.UserID, cfg, federation, producer,
rsAPI, fsAPI, keyRing,
}
if strings.HasPrefix(roomIDOrAlias, "!") {
return r.joinRoomByID(roomIDOrAlias)
}
if strings.HasPrefix(roomIDOrAlias, "#") {
return r.joinRoomByAlias(roomIDOrAlias)
}
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.BadJSON(
fmt.Sprintf("Invalid first character '%s' for room ID or alias",
string([]rune(roomIDOrAlias)[0])), // Wrapping with []rune makes this call UTF-8 safe
),
}
}
type joinRoomReq struct {
req *http.Request
evTime time.Time
content map[string]interface{}
userID string
cfg *config.Dendrite
federation *gomatrixserverlib.FederationClient
producer *producers.RoomserverProducer
rsAPI roomserverAPI.RoomserverInternalAPI
fsAPI federationSenderAPI.FederationSenderInternalAPI
keyRing gomatrixserverlib.KeyRing
}
// joinRoomByID joins a room by room ID
func (r joinRoomReq) joinRoomByID(roomID string) util.JSONResponse {
// A client should only join a room by room ID when it has an invite
// to the room. If the server is already in the room then we can
// lookup the invite and process the request as a normal state event.
// If the server is not in the room the we will need to look up the
// remote server the invite came from in order to request a join event
// from that server.
queryReq := roomserverAPI.QueryInvitesForUserRequest{
RoomID: roomID, TargetUserID: r.userID,
}
var queryRes roomserverAPI.QueryInvitesForUserResponse
if err := r.rsAPI.QueryInvitesForUser(r.req.Context(), &queryReq, &queryRes); err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("r.queryAPI.QueryInvitesForUser failed")
return jsonerror.InternalServerError()
}
servers := []gomatrixserverlib.ServerName{}
seenInInviterIDs := map[gomatrixserverlib.ServerName]bool{}
for _, userID := range queryRes.InviteSenderUserIDs {
_, domain, err := gomatrixserverlib.SplitID('@', userID)
if err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
return jsonerror.InternalServerError()
}
if !seenInInviterIDs[domain] {
servers = append(servers, domain)
seenInInviterIDs[domain] = true
}
}
// Also add the domain extracted from the roomID as a last resort to join
// in case the client is erroneously trying to join by ID without an invite
// or all previous attempts at domains extracted from the inviter IDs fail
// Note: It's no guarantee we'll succeed because a room isn't bound to the domain in its ID
_, domain, err := gomatrixserverlib.SplitID('!', roomID)
if err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
return jsonerror.InternalServerError()
}
if domain != r.cfg.Matrix.ServerName && !seenInInviterIDs[domain] {
servers = append(servers, domain)
}
return r.joinRoomUsingServers(roomID, servers)
}
// joinRoomByAlias joins a room using a room alias.
func (r joinRoomReq) joinRoomByAlias(roomAlias string) util.JSONResponse {
_, domain, err := gomatrixserverlib.SplitID('#', roomAlias)
if err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.BadJSON("Room alias must be in the form '#localpart:domain'"),
}
}
if domain == r.cfg.Matrix.ServerName {
queryReq := roomserverAPI.GetRoomIDForAliasRequest{Alias: roomAlias}
var queryRes roomserverAPI.GetRoomIDForAliasResponse
if err = r.rsAPI.GetRoomIDForAlias(r.req.Context(), &queryReq, &queryRes); err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("r.aliasAPI.GetRoomIDForAlias failed")
return jsonerror.InternalServerError()
}
if len(queryRes.RoomID) > 0 {
return r.joinRoomUsingServers(queryRes.RoomID, []gomatrixserverlib.ServerName{r.cfg.Matrix.ServerName})
}
// If the response doesn't contain a non-empty string, return an error
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: jsonerror.NotFound("Room alias " + roomAlias + " not found."),
}
}
// If the room isn't local, use federation to join
return r.joinRoomByRemoteAlias(domain, roomAlias)
}
func (r joinRoomReq) joinRoomByRemoteAlias(
domain gomatrixserverlib.ServerName, roomAlias string,
) util.JSONResponse {
resp, err := r.federation.LookupRoomAlias(r.req.Context(), domain, roomAlias)
if err != nil {
switch x := err.(type) {
case gomatrix.HTTPError:
if x.Code == http.StatusNotFound {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: jsonerror.NotFound("Room alias not found"),
}
}
}
util.GetLogger(r.req.Context()).WithError(err).Error("r.federation.LookupRoomAlias failed")
return jsonerror.InternalServerError()
}
return r.joinRoomUsingServers(resp.RoomID, resp.Servers)
}
func (r joinRoomReq) writeToBuilder(eb *gomatrixserverlib.EventBuilder, roomID string) error {
eb.Type = "m.room.member"
err := eb.SetContent(r.content)
if err != nil {
return err
}
err = eb.SetUnsigned(struct{}{})
if err != nil {
return err
}
eb.Sender = r.userID
eb.StateKey = &r.userID
eb.RoomID = roomID
eb.Redacts = ""
return nil
}
func (r joinRoomReq) joinRoomUsingServers(
roomID string, servers []gomatrixserverlib.ServerName,
) util.JSONResponse {
var eb gomatrixserverlib.EventBuilder
err := r.writeToBuilder(&eb, roomID)
if err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("r.writeToBuilder failed")
return jsonerror.InternalServerError()
}
queryRes := roomserverAPI.QueryLatestEventsAndStateResponse{}
event, err := common.BuildEvent(r.req.Context(), &eb, r.cfg, r.evTime, r.rsAPI, &queryRes)
if err == nil {
// If we have successfully built an event at this point then we can
// assert that the room is a local room, as BuildEvent was able to
// add prev_events etc successfully.
if _, err = r.producer.SendEvents(
r.req.Context(),
[]gomatrixserverlib.HeaderedEvent{
(*event).Headered(queryRes.RoomVersion),
},
r.cfg.Matrix.ServerName,
nil,
); err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("r.producer.SendEvents failed")
return jsonerror.InternalServerError()
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: struct {
RoomID string `json:"room_id"`
}{roomID},
}
}
// Otherwise, if we've reached here, then we haven't been able to populate
// prev_events etc for the room, therefore the room is probably federated.
// TODO: This needs to be re-thought, as in the case of an invite, the room
// will exist in the database in roomserver_rooms but won't have any state
// events, therefore this below check fails.
if err != common.ErrRoomNoExists {
util.GetLogger(r.req.Context()).WithError(err).Error("common.BuildEvent failed")
return jsonerror.InternalServerError()
}
if len(servers) == 0 {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: jsonerror.NotFound("No candidate servers found for room"),
}
}
var lastErr error
for _, server := range servers {
var response *util.JSONResponse
response, lastErr = r.joinRoomUsingServer(roomID, server)
if lastErr != nil {
// 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
}
// Every server we tried to join through resulted in an error.
// We return the error from the last server.
// TODO: Generate the correct HTTP status code for all different
// kinds of errors that could have happened.
// The possible errors include:
// 1) We can't connect to the remote servers.
// 2) None of the servers we could connect to think we are allowed
// to join the room.
// 3) The remote server returned something invalid.
// 4) We couldn't fetch the public keys needed to verify the
// signatures on the state events.
// 5) ...
util.GetLogger(r.req.Context()).WithError(lastErr).Error("failed to join through any server")
return jsonerror.InternalServerError()
}
// joinRoomUsingServer tries to join a remote room using a given matrix server.
// If there was a failure communicating with the server or the response from the
// server was invalid this returns an error.
// Otherwise this returns a JSONResponse.
func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib.ServerName) (*util.JSONResponse, error) {
fedJoinReq := federationSenderAPI.PerformJoinRequest{
RoomID: roomID,
UserID: r.userID,
ServerName: server,
}
fedJoinRes := federationSenderAPI.PerformJoinResponse{}
if err := r.fsAPI.PerformJoin(r.req.Context(), &fedJoinReq, &fedJoinRes); err != nil {
return nil, err
}
return &util.JSONResponse{
Code: http.StatusOK,
// TODO: Put the response struct somewhere common.
JSON: struct {
RoomID string `json:"room_id"`
}{roomID},
}, nil
}{joinReq.RoomIDOrAlias},
}
}

3
clientapi/routing/routing.go
View File

@ -100,8 +100,7 @@ func Setup(
return util.ErrorResponse(err)
}
return JoinRoomByIDOrAlias(
req, device, vars["roomIDOrAlias"], cfg, federation, producer,
rsAPI, federationSender, keyRing, accountDB,
req, device, rsAPI, vars["roomIDOrAlias"],
)
}),
).Methods(http.MethodPost, http.MethodOptions)

53
federationapi/routing/join.go
View File

@ -61,9 +61,7 @@ func MakeJoin(
if !remoteSupportsVersion {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.UnsupportedRoomVersion(
fmt.Sprintf("Joining server does not support room version %s", verRes.RoomVersion),
),
JSON: jsonerror.IncompatibleRoomVersion(verRes.RoomVersion),
}
}
@ -132,6 +130,9 @@ func MakeJoin(
}
// SendJoin implements the /send_join API
// The make-join send-join dance makes much more sense as a single
// flow so the cyclomatic complexity is high:
// nolint:gocyclo
func SendJoin(
httpReq *http.Request,
request *gomatrixserverlib.FederationRequest,
@ -159,6 +160,16 @@ func SendJoin(
}
}
// Check that a state key is provided.
if event.StateKey() == nil || (event.StateKey() != nil && *event.StateKey() == "") {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.BadJSON(
fmt.Sprintf("No state key was provided in the join event."),
),
}
}
// Check that the room ID is correct.
if event.RoomID() != roomID {
return util.JSONResponse{
@ -234,20 +245,34 @@ func SendJoin(
}
}
// 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.
alreadyJoined := false
for _, se := range stateAndAuthChainResponse.StateEvents {
if membership, merr := se.Membership(); merr == nil {
if se.StateKey() != nil && *se.StateKey() == *event.StateKey() {
alreadyJoined = (membership == "join")
break
}
}
}
// Send the events to the room server.
// We are responsible for notifying other servers that the user has joined
// the room, so set SendAsServer to cfg.Matrix.ServerName
_, err = producer.SendEvents(
httpReq.Context(),
[]gomatrixserverlib.HeaderedEvent{
event.Headered(stateAndAuthChainResponse.RoomVersion),
},
cfg.Matrix.ServerName,
nil,
)
if err != nil {
util.GetLogger(httpReq.Context()).WithError(err).Error("producer.SendEvents failed")
return jsonerror.InternalServerError()
if !alreadyJoined {
_, err = producer.SendEvents(
httpReq.Context(),
[]gomatrixserverlib.HeaderedEvent{
event.Headered(stateAndAuthChainResponse.RoomVersion),
},
cfg.Matrix.ServerName,
nil,
)
if err != nil {
util.GetLogger(httpReq.Context()).WithError(err).Error("producer.SendEvents failed")
return jsonerror.InternalServerError()
}
}
return util.JSONResponse{

6
federationsender/api/api.go
View File

@ -8,6 +8,12 @@ import (
// FederationSenderInternalAPI is used to query information from the federation sender.
type FederationSenderInternalAPI interface {
// PerformDirectoryLookup looks up a remote room ID from a room alias.
PerformDirectoryLookup(
ctx context.Context,
request *PerformDirectoryLookupRequest,
response *PerformDirectoryLookupResponse,
) error
// Query the joined hosts and the membership events accounting for their participation in a room.
// Note that if a server has multiple users in the room, it will have multiple entries in the returned slice.
// See `QueryJoinedHostServerNamesInRoom` for a de-duplicated version.

35
federationsender/api/perform.go
View File

@ -4,11 +4,15 @@ import (
"context"
commonHTTP "github.com/matrix-org/dendrite/common/http"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/opentracing/opentracing-go"
)
const (
// FederationSenderPerformJoinRequestPath is the HTTP path for the PerformJoinRequest API.
FederationSenderPerformDirectoryLookupRequestPath = "/api/federationsender/performDirectoryLookup"
// FederationSenderPerformJoinRequestPath is the HTTP path for the PerformJoinRequest API.
FederationSenderPerformJoinRequestPath = "/api/federationsender/performJoinRequest"
@ -16,11 +20,34 @@ const (
FederationSenderPerformLeaveRequestPath = "/api/federationsender/performLeaveRequest"
)
type PerformJoinRequest struct {
RoomID string `json:"room_id"`
UserID string `json:"user_id"`
type PerformDirectoryLookupRequest struct {
RoomAlias string `json:"room_alias"`
ServerName gomatrixserverlib.ServerName `json:"server_name"`
Content map[string]interface{} `json:"content"`
}
type PerformDirectoryLookupResponse struct {
RoomID string `json:"room_id"`
ServerNames []gomatrixserverlib.ServerName `json:"server_names"`
}
// Handle an instruction to make_join & send_join with a remote server.
func (h *httpFederationSenderInternalAPI) PerformDirectoryLookup(
ctx context.Context,
request *PerformDirectoryLookupRequest,
response *PerformDirectoryLookupResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformDirectoryLookup")
defer span.Finish()
apiURL := h.federationSenderURL + FederationSenderPerformDirectoryLookupRequestPath
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
type PerformJoinRequest struct {
RoomID string `json:"room_id"`
UserID string `json:"user_id"`
ServerNames types.ServerNames `json:"server_names"`
Content map[string]interface{} `json:"content"`
}
type PerformJoinResponse struct {

205
federationsender/internal/perform.go
View File

@ -9,8 +9,29 @@ import (
"github.com/matrix-org/dendrite/federationsender/internal/perform"
"github.com/matrix-org/dendrite/roomserver/version"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/sirupsen/logrus"
)
// PerformLeaveRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformDirectoryLookup(
ctx context.Context,
request *api.PerformDirectoryLookupRequest,
response *api.PerformDirectoryLookupResponse,
) (err error) {
dir, err := r.federation.LookupRoomAlias(
ctx,
request.ServerName,
request.RoomAlias,
)
if err != nil {
return err
}
response.RoomID = dir.RoomID
response.ServerNames = dir.Servers
return nil
}
// PerformJoinRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformJoin(
ctx context.Context,
@ -23,91 +44,107 @@ func (r *FederationSenderInternalAPI) PerformJoin(
supportedVersions = append(supportedVersions, version)
}
// Try to perform a make_join using the information supplied in the
// request.
respMakeJoin, err := r.federation.MakeJoin(
ctx,
request.ServerName,
request.RoomID,
request.UserID,
supportedVersions,
// Deduplicate the server names we were provided.
util.Unique(request.ServerNames)
// Try each server that we were provided until we land on one that
// successfully completes the make-join send-join dance.
for _, serverName := range request.ServerNames {
// Try to perform a make_join using the information supplied in the
// request.
respMakeJoin, err := r.federation.MakeJoin(
ctx,
serverName,
request.RoomID,
request.UserID,
supportedVersions,
)
if err != nil {
// TODO: Check if the user was not allowed to join the room.
return fmt.Errorf("r.federation.MakeJoin: %w", err)
}
// 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 = request.UserID
respMakeJoin.JoinEvent.StateKey = &request.UserID
respMakeJoin.JoinEvent.RoomID = request.RoomID
respMakeJoin.JoinEvent.Redacts = ""
if request.Content == nil {
request.Content = map[string]interface{}{}
}
request.Content["membership"] = "join"
if err = respMakeJoin.JoinEvent.SetContent(request.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.
if respMakeJoin.RoomVersion == "" {
respMakeJoin.RoomVersion = gomatrixserverlib.RoomVersionV1
}
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(
ctx,
serverName,
event,
respMakeJoin.RoomVersion,
)
if err != nil {
logrus.WithError(err).Warnf("r.federation.SendJoin failed")
continue
}
// Check that the send_join response was valid.
joinCtx := perform.JoinContext(r.federation, r.keyRing)
if err = joinCtx.CheckSendJoinResponse(
ctx, event, serverName, respMakeJoin, respSendJoin,
); err != nil {
logrus.WithError(err).Warnf("joinCtx.CheckSendJoinResponse failed")
continue
}
// 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 = r.producer.SendEventWithState(
ctx,
respSendJoin.ToRespState(),
event.Headered(respMakeJoin.RoomVersion),
); err != nil {
logrus.WithError(err).Warnf("r.producer.SendEventWithState failed")
continue
}
// We're all good.
return nil
}
// If we reach here then we didn't complete a join for some reason.
return fmt.Errorf(
"failed to join user %q to room %q through %d server(s)",
request.UserID, request.RoomID, len(request.ServerNames),
)
if err != nil {
// TODO: Check if the user was not allowed to join the room.
return fmt.Errorf("r.federation.MakeJoin: %w", err)
}
// 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 = "m.room.member"
respMakeJoin.JoinEvent.Sender = request.UserID
respMakeJoin.JoinEvent.StateKey = &request.UserID
respMakeJoin.JoinEvent.RoomID = request.RoomID
respMakeJoin.JoinEvent.Redacts = ""
if request.Content == nil {
request.Content = map[string]interface{}{}
}
request.Content["membership"] = "join"
if err = respMakeJoin.JoinEvent.SetContent(request.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.
if respMakeJoin.RoomVersion == "" {
respMakeJoin.RoomVersion = gomatrixserverlib.RoomVersionV1
}
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(
ctx,
request.ServerName,
event,
respMakeJoin.RoomVersion,
)
if err != nil {
return fmt.Errorf("r.federation.SendJoin: %w", err)
}
// Check that the send_join response was valid.
joinCtx := perform.JoinContext(r.federation, r.keyRing)
if err = joinCtx.CheckSendJoinResponse(
ctx, event, request.ServerName, respMakeJoin, respSendJoin,
); err != nil {
return fmt.Errorf("perform.JoinRequest.CheckSendJoinResponse: %w", 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 err = r.producer.SendEventWithState(
ctx,
respSendJoin.ToRespState(),
event.Headered(respMakeJoin.RoomVersion),
); err != nil {
return fmt.Errorf("r.producer.SendEventWithState: %w", err)
}
// Everything went to plan.
return nil
}
// PerformLeaveRequest implements api.FederationSenderInternalAPI

6
federationsender/types/types.go
View File

@ -28,6 +28,12 @@ type JoinedHost struct {
ServerName gomatrixserverlib.ServerName
}
type ServerNames []gomatrixserverlib.ServerName
func (s ServerNames) Len() int { return len(s) }
func (s ServerNames) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s ServerNames) Less(i, j int) bool { return s[i] < s[j] }
// A EventIDMismatchError indicates that we have got out of sync with the
// room server.
type EventIDMismatchError struct {

6
roomserver/api/api.go
View File

@ -18,6 +18,12 @@ type RoomserverInternalAPI interface {
response *InputRoomEventsResponse,
) error
PerformJoin(
ctx context.Context,
req *PerformJoinRequest,
res *PerformJoinResponse,
) error
// Query the latest events and state for a room from the room server.
QueryLatestEventsAndState(
ctx context.Context,

59
roomserver/api/perform.go Normal file
View File

@ -0,0 +1,59 @@
package api
import (
"context"
commonHTTP "github.com/matrix-org/dendrite/common/http"
"github.com/matrix-org/gomatrixserverlib"
"github.com/opentracing/opentracing-go"
)
const (
// RoomserverPerformJoinPath is the HTTP path for the PerformJoin API.
RoomserverPerformJoinPath = "/api/roomserver/performJoin"
// RoomserverPerformLeavePath is the HTTP path for the PerformLeave API.
RoomserverPerformLeavePath = "/api/roomserver/performLeave"
)
type PerformJoinRequest struct {
RoomIDOrAlias string `json:"room_id_or_alias"`
UserID string `json:"user_id"`
Content map[string]interface{} `json:"content"`
ServerNames []gomatrixserverlib.ServerName `json:"server_names"`
}
type PerformJoinResponse struct {
}
func (h *httpRoomserverInternalAPI) PerformJoin(
ctx context.Context,
request *PerformJoinRequest,
response *PerformJoinResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformJoin")
defer span.Finish()
apiURL := h.roomserverURL + RoomserverPerformJoinPath
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
type PerformLeaveRequest struct {
RoomID string `json:"room_id"`
UserID string `json:"user_id"`
}
type PerformLeaveResponse struct {
}
func (h *httpRoomserverInternalAPI) PerformLeave(
ctx context.Context,
request *PerformLeaveRequest,
response *PerformLeaveResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformLeave")
defer span.Finish()
apiURL := h.roomserverURL + RoomserverPerformLeavePath
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}

13
roomserver/internal/api.go
View File

@ -46,6 +46,19 @@ func (r *RoomserverInternalAPI) SetupHTTP(servMux *http.ServeMux) {
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
servMux.Handle(api.RoomserverPerformJoinPath,
common.MakeInternalAPI("performJoin", func(req *http.Request) util.JSONResponse {
var request api.PerformJoinRequest
var response api.PerformJoinResponse
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())
}
if err := r.PerformJoin(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
servMux.Handle(
api.RoomserverQueryLatestEventsAndStatePath,
common.MakeInternalAPI("queryLatestEventsAndState", func(req *http.Request) util.JSONResponse {

199
roomserver/internal/perform_join.go Normal file
View File

@ -0,0 +1,199 @@
package internal
import (
"context"
"fmt"
"strings"
"time"
"github.com/matrix-org/dendrite/common"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
)
// WriteOutputEvents implements OutputRoomEventWriter
func (r *RoomserverInternalAPI) PerformJoin(
ctx context.Context,
req *api.PerformJoinRequest,
res *api.PerformJoinResponse,
) error {
_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil {
return fmt.Errorf("Supplied user ID %q in incorrect format", req.UserID)
}
if domain != r.Cfg.Matrix.ServerName {
return fmt.Errorf("User %q does not belong to this homeserver", req.UserID)
}
if strings.HasPrefix(req.RoomIDOrAlias, "!") {
return r.performJoinRoomByID(ctx, req, res)
}
if strings.HasPrefix(req.RoomIDOrAlias, "#") {
return r.performJoinRoomByAlias(ctx, req, res)
}
return fmt.Errorf("Room ID or alias %q is invalid", req.RoomIDOrAlias)
}
func (r *RoomserverInternalAPI) performJoinRoomByAlias(
ctx context.Context,
req *api.PerformJoinRequest,
res *api.PerformJoinResponse,
) error {
// Get the domain part of the room alias.
_, domain, err := gomatrixserverlib.SplitID('#', req.RoomIDOrAlias)
if err != nil {
return fmt.Errorf("Alias %q is not in the correct format", req.RoomIDOrAlias)
}
req.ServerNames = append(req.ServerNames, domain)
// Check if this alias matches our own server configuration. If it
// doesn't then we'll need to try a federated join.
var roomID string
if domain != r.Cfg.Matrix.ServerName {
// The alias isn't owned by us, so we will need to try joining using
// a remote server.
dirReq := fsAPI.PerformDirectoryLookupRequest{
RoomAlias: req.RoomIDOrAlias, // the room alias to lookup
ServerName: domain, // the server to ask
}
dirRes := fsAPI.PerformDirectoryLookupResponse{}
err = r.fsAPI.PerformDirectoryLookup(ctx, &dirReq, &dirRes)
if err != nil {
logrus.WithError(err).Errorf("error looking up alias %q", req.RoomIDOrAlias)
return fmt.Errorf("Looking up alias %q over federation failed: %w", req.RoomIDOrAlias, err)
}
roomID = dirRes.RoomID
req.ServerNames = append(req.ServerNames, dirRes.ServerNames...)
} else {
// Otherwise, look up if we know this room alias locally.
roomID, err = r.DB.GetRoomIDForAlias(ctx, req.RoomIDOrAlias)
if err != nil {
return fmt.Errorf("Lookup room alias %q failed: %w", req.RoomIDOrAlias, err)
}
}
// If the room ID is empty then we failed to look up the alias.
if roomID == "" {
return fmt.Errorf("Alias %q not found", req.RoomIDOrAlias)
}
// If we do, then pluck out the room ID and continue the join.
req.RoomIDOrAlias = roomID
return r.performJoinRoomByID(ctx, req, res)
}
// TODO: Break this function up a bit
// nolint:gocyclo
func (r *RoomserverInternalAPI) performJoinRoomByID(
ctx context.Context,
req *api.PerformJoinRequest,
res *api.PerformJoinResponse, // nolint:unparam
) error {
// Get the domain part of the room ID.
_, domain, err := gomatrixserverlib.SplitID('!', req.RoomIDOrAlias)
if err != nil {
return fmt.Errorf("Room ID %q is invalid", req.RoomIDOrAlias)
}
req.ServerNames = append(req.ServerNames, domain)
// Prepare the template for the join event.
userID := req.UserID
eb := gomatrixserverlib.EventBuilder{
Type: gomatrixserverlib.MRoomMember,
Sender: userID,
StateKey: &userID,
RoomID: req.RoomIDOrAlias,
Redacts: "",
}
if err = eb.SetUnsigned(struct{}{}); err != nil {
return fmt.Errorf("eb.SetUnsigned: %w", err)
}
// It is possible for the request to include some "content" for the
// event. We'll always overwrite the "membership" key, but the rest,
// like "display_name" or "avatar_url", will be kept if supplied.
if req.Content == nil {
req.Content = map[string]interface{}{}
}
req.Content["membership"] = "join"
if err = eb.SetContent(req.Content); err != nil {
return fmt.Errorf("eb.SetContent: %w", err)
}
// Try to construct an actual join event from the template.
// If this succeeds then it is a sign that the room already exists
// locally on the homeserver.
// TODO: Check what happens if the room exists on the server
// but everyone has since left. I suspect it does the wrong thing.
buildRes := api.QueryLatestEventsAndStateResponse{}
event, err := common.BuildEvent(
ctx, // the request context
&eb, // the template join event
r.Cfg, // the server configuration
time.Now(), // the event timestamp to use
r, // the roomserver API to use
&buildRes, // the query response
)
switch err {
case nil:
// The room join is local. Send the new join event into the
// roomserver. First of all check that the user isn't already
// a member of the room.
alreadyJoined := false
for _, se := range buildRes.StateEvents {
if membership, merr := se.Membership(); merr == nil {
if se.StateKey() != nil && *se.StateKey() == *event.StateKey() {
alreadyJoined = (membership == "join")
break
}
}
}
// If we haven't already joined the room then send an event
// into the room changing our membership status.
if !alreadyJoined {
inputReq := api.InputRoomEventsRequest{
InputRoomEvents: []api.InputRoomEvent{
api.InputRoomEvent{
Kind: api.KindNew,
Event: event.Headered(buildRes.RoomVersion),
AuthEventIDs: event.AuthEventIDs(),
SendAsServer: string(r.Cfg.Matrix.ServerName),
},
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
return fmt.Errorf("r.InputRoomEvents: %w", err)
}
}
case common.ErrRoomNoExists:
// The room doesn't exist. First of all check if the room is a local
// room. If it is then there's nothing more to do - the room just
// hasn't been created yet.
if domain == r.Cfg.Matrix.ServerName {
return fmt.Errorf("Room ID %q does not exist", req.RoomIDOrAlias)
}
// Try joining by all of the supplied server names.
fedReq := fsAPI.PerformJoinRequest{
RoomID: req.RoomIDOrAlias, // the room ID to try and join
UserID: req.UserID, // the user ID joining the room
ServerNames: req.ServerNames, // the server to try joining with
Content: req.Content, // the membership event content
}
fedRes := fsAPI.PerformJoinResponse{}
err = r.fsAPI.PerformJoin(ctx, &fedReq, &fedRes)
if err != nil {
return fmt.Errorf("Error joining federated room: %q", err)
}
default:
return fmt.Errorf("Error joining room %q: %w", req.RoomIDOrAlias, err)
}
return nil
}