mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-16 19:33:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into bad-json-body

Browse source
This commit is contained in:
Alex Chen 2020-04-11 21:45:48 +08:00 committed by GitHub
parent b0045a9c22 dacee648f7
commit 37a58fd688
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
33 changed files with 594 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
appservice/api/query.go
View file

@ -20,6 +20,7 @@ package api
import (
"context"
"database/sql"
"errors"
"net/http"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
@ -97,15 +98,15 @@ type httpAppServiceQueryAPI struct {
// NewAppServiceQueryAPIHTTP creates a AppServiceQueryAPI implemented by talking
// to a HTTP POST API.
// If httpClient is nil then it uses http.DefaultClient
// If httpClient is nil an error is returned
func NewAppServiceQueryAPIHTTP(
appserviceURL string,
httpClient *http.Client,
) AppServiceQueryAPI {
) (AppServiceQueryAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewRoomserverAliasAPIHTTP: httpClient is <nil>")
}
return &httpAppServiceQueryAPI{appserviceURL, httpClient}
return &httpAppServiceQueryAPI{appserviceURL, httpClient}, nil
}
// RoomAliasExists implements AppServiceQueryAPI

14
clientapi/producers/roomserver.go
View file

@ -104,18 +104,14 @@ func (c *RoomserverProducer) SendInputRoomEvents(
// This should only be needed for invite events that occur outside of a known room.
// If we are in the room then the event should be sent using the SendEvents method.
func (c *RoomserverProducer) SendInvite(
ctx context.Context, inviteEvent gomatrixserverlib.Event,
ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
inviteRoomState []gomatrixserverlib.InviteV2StrippedState,
) error {
verReq := api.QueryRoomVersionForRoomRequest{RoomID: inviteEvent.RoomID()}
verRes := api.QueryRoomVersionForRoomResponse{}
err := c.QueryAPI.QueryRoomVersionForRoom(ctx, &verReq, &verRes)
if err != nil {
return err
}
request := api.InputRoomEventsRequest{
InputInviteEvents: []api.InputInviteEvent{{
Event: inviteEvent.Headered(verRes.RoomVersion),
Event: inviteEvent,
InviteRoomState: inviteRoomState,
RoomVersion: inviteEvent.RoomVersion,
}},
}
var response api.InputRoomEventsResponse

2
clientapi/routing/capabilities.go
View file

@ -29,7 +29,7 @@ func GetCapabilities(
req *http.Request, queryAPI roomserverAPI.RoomserverQueryAPI,
) util.JSONResponse {
roomVersionsQueryReq := roomserverAPI.QueryRoomVersionCapabilitiesRequest{}
var roomVersionsQueryRes roomserverAPI.QueryRoomVersionCapabilitiesResponse
roomVersionsQueryRes := roomserverAPI.QueryRoomVersionCapabilitiesResponse{}
if err := queryAPI.QueryRoomVersionCapabilities(
req.Context(),
&roomVersionsQueryReq,

26
clientapi/routing/joinroom.go
View file

@ -242,6 +242,9 @@ func (r joinRoomReq) joinRoomUsingServers(
queryRes := roomserverAPI.QueryLatestEventsAndStateResponse{}
event, err := common.BuildEvent(r.req.Context(), &eb, r.cfg, r.evTime, r.queryAPI, &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{
@ -260,6 +263,13 @@ func (r joinRoomReq) joinRoomUsingServers(
}{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()
@ -320,14 +330,14 @@ func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib
respMakeJoin, err := r.federation.MakeJoin(r.req.Context(), server, roomID, r.userID, supportedVersions)
if err != nil {
// TODO: Check if the user was not allowed to join the room.
return nil, err
return nil, 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"
err = r.writeToBuilder(&respMakeJoin.JoinEvent, roomID)
if err != nil {
return nil, err
return nil, fmt.Errorf("r.writeToBuilder: %w", err)
}
if respMakeJoin.RoomVersion == "" {
@ -347,18 +357,16 @@ func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib
r.cfg.Matrix.PrivateKey, respMakeJoin.RoomVersion,
)
if err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("respMakeJoin.JoinEvent.Build failed")
res := jsonerror.InternalServerError()
return &res, nil
return nil, fmt.Errorf("respMakeJoin.JoinEvent.Build: %w", err)
}
respSendJoin, err := r.federation.SendJoin(r.req.Context(), server, event, respMakeJoin.RoomVersion)
if err != nil {
return nil, err
return nil, fmt.Errorf("r.federation.SendJoin: %w", err)
}
if err = respSendJoin.Check(r.req.Context(), r.keyRing, event); err != nil {
return nil, err
return nil, fmt.Errorf("respSendJoin: %w", err)
}
if err = r.producer.SendEventWithState(
@ -366,9 +374,7 @@ func (r joinRoomReq) joinRoomUsingServer(roomID string, server gomatrixserverlib
gomatrixserverlib.RespState(respSendJoin.RespState),
event.Headered(respMakeJoin.RoomVersion),
); err != nil {
util.GetLogger(r.req.Context()).WithError(err).Error("gomatrixserverlib.RespState failed")
res := jsonerror.InternalServerError()
return &res, nil
return nil, fmt.Errorf("r.producer.SendEventWithState: %w", err)
}
return &util.JSONResponse{

18
cmd/roomserver-integration-tests/main.go
View file

@ -44,6 +44,8 @@ var (
// This needs to be high enough to account for the time it takes to create
// the postgres database tables which can take a while on travis.
timeoutString = defaulting(os.Getenv("TIMEOUT"), "60s")
// Timeout for http client
timeoutHTTPClient = defaulting(os.Getenv("TIMEOUT_HTTP"), "30s")
// The name of maintenance database to connect to in order to create the test database.
postgresDatabase = defaulting(os.Getenv("POSTGRES_DATABASE"), "postgres")
// The name of the test database to create.
@ -68,7 +70,10 @@ func defaulting(value, defaultValue string) string {
return value
}
var timeout time.Duration
var (
timeout time.Duration
timeoutHTTP time.Duration
)
func init() {
var err error
@ -76,6 +81,10 @@ func init() {
if err != nil {
panic(err)
}
timeoutHTTP, err = time.ParseDuration(timeoutHTTPClient)
if err != nil {
panic(err)
}
}
func createDatabase(database string) error {
@ -199,7 +208,10 @@ func writeToRoomServer(input []string, roomserverURL string) error {
return err
}
}
x := api.NewRoomserverInputAPIHTTP(roomserverURL, nil)
x, err := api.NewRoomserverInputAPIHTTP(roomserverURL, &http.Client{Timeout: timeoutHTTP})
if err != nil {
return err
}
return x.InputRoomEvents(context.Background(), &request, &response)
}
@ -258,7 +270,7 @@ func testRoomserver(input []string, wantOutput []string, checkQueries func(api.R
cmd.Args = []string{"dendrite-room-server", "--config", filepath.Join(dir, test.ConfigFile)}
gotOutput, err := runAndReadFromTopic(cmd, cfg.RoomServerURL()+"/metrics", doInput, outputTopic, len(wantOutput), func() {
queryAPI := api.NewRoomserverQueryAPIHTTP("http://"+string(cfg.Listen.RoomServer), nil)
queryAPI, _ := api.NewRoomserverQueryAPIHTTP("http://"+string(cfg.Listen.RoomServer), &http.Client{Timeout: timeoutHTTP})
checkQueries(queryAPI)
})
if err != nil {

39
common/basecomponent/base.go
View file

@ -19,6 +19,7 @@ import (
"io"
"net/http"
"net/url"
"time"
"golang.org/x/crypto/ed25519"
@ -52,6 +53,7 @@ type BaseDendrite struct {
// APIMux should be used to register new public matrix api endpoints
APIMux *mux.Router
httpClient *http.Client
Cfg *config.Dendrite
KafkaConsumer sarama.Consumer
KafkaProducer sarama.SyncProducer
@ -77,11 +79,14 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
kafkaConsumer, kafkaProducer = setupKafka(cfg)
}
const defaultHTTPTimeout = 30 * time.Second
return &BaseDendrite{
componentName: componentName,
tracerCloser: closer,
Cfg: cfg,
APIMux: mux.NewRouter().UseEncodedPath(),
httpClient: &http.Client{Timeout: defaultHTTPTimeout},
KafkaConsumer: kafkaConsumer,
KafkaProducer: kafkaProducer,
}
@ -95,7 +100,11 @@ func (b *BaseDendrite) Close() error {
// CreateHTTPAppServiceAPIs returns the QueryAPI for hitting the appservice
// component over HTTP.
func (b *BaseDendrite) CreateHTTPAppServiceAPIs() appserviceAPI.AppServiceQueryAPI {
return appserviceAPI.NewAppServiceQueryAPIHTTP(b.Cfg.AppServiceURL(), nil)
a, err := appserviceAPI.NewAppServiceQueryAPIHTTP(b.Cfg.AppServiceURL(), b.httpClient)
if err != nil {
logrus.WithError(err).Panic("CreateHTTPAppServiceAPIs failed")
}
return a
}
// CreateHTTPRoomserverAPIs returns the AliasAPI, InputAPI and QueryAPI for hitting
@ -105,22 +114,40 @@ func (b *BaseDendrite) CreateHTTPRoomserverAPIs() (
roomserverAPI.RoomserverInputAPI,
roomserverAPI.RoomserverQueryAPI,
) {
alias := roomserverAPI.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), nil)
input := roomserverAPI.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), nil)
query := roomserverAPI.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
alias, err := roomserverAPI.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), b.httpClient)
if err != nil {
logrus.WithError(err).Panic("NewRoomserverAliasAPIHTTP failed")
}
input, err := roomserverAPI.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), b.httpClient)
if err != nil {
logrus.WithError(err).Panic("NewRoomserverInputAPIHTTP failed", b.httpClient)
}
query, err := roomserverAPI.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
if err != nil {
logrus.WithError(err).Panic("NewRoomserverQueryAPIHTTP failed", b.httpClient)
}
return alias, input, query
}
// CreateHTTPEDUServerAPIs returns eduInputAPI for hitting the EDU
// server over HTTP
func (b *BaseDendrite) CreateHTTPEDUServerAPIs() eduServerAPI.EDUServerInputAPI {
return eduServerAPI.NewEDUServerInputAPIHTTP(b.Cfg.EDUServerURL(), nil)
e, err := eduServerAPI.NewEDUServerInputAPIHTTP(b.Cfg.EDUServerURL(), nil)
if err != nil {
logrus.WithError(err).Panic("NewEDUServerInputAPIHTTP failed", b.httpClient)
}
return e
}
// CreateHTTPFederationSenderAPIs returns FederationSenderQueryAPI for hitting
// the federation sender over HTTP
func (b *BaseDendrite) CreateHTTPFederationSenderAPIs() federationSenderAPI.FederationSenderQueryAPI {
return federationSenderAPI.NewFederationSenderQueryAPIHTTP(b.Cfg.FederationSenderURL(), nil)
f, err := federationSenderAPI.NewFederationSenderQueryAPIHTTP(b.Cfg.FederationSenderURL(), nil)
if err != nil {
logrus.WithError(err).Panic("NewFederationSenderQueryAPIHTTP failed", b.httpClient)
}
return f
}
// CreateDeviceDB creates a new instance of the device database. Should only be

42
common/events.go
View file

@ -17,6 +17,7 @@ package common
import (
"context"
"errors"
"fmt"
"time"
"github.com/matrix-org/dendrite/common/config"
@ -46,6 +47,7 @@ func BuildEvent(
err := AddPrevEventsToEvent(ctx, builder, queryAPI, queryRes)
if err != nil {
// This can pass through a ErrRoomNoExists to the caller
return nil, err
}
@ -68,7 +70,7 @@ func AddPrevEventsToEvent(
) error {
eventsNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(builder)
if err != nil {
return err
return fmt.Errorf("gomatrixserverlib.StateNeededForEventBuilder: %w", err)
}
// Ask the roomserver for information about this room
@ -77,7 +79,7 @@ func AddPrevEventsToEvent(
StateToFetch: eventsNeeded.Tuples(),
}
if err = queryAPI.QueryLatestEventsAndState(ctx, &queryReq, queryRes); err != nil {
return err
return fmt.Errorf("queryAPI.QueryLatestEventsAndState: %w", err)
}
if !queryRes.RoomExists {
@ -86,7 +88,7 @@ func AddPrevEventsToEvent(
eventFormat, err := queryRes.RoomVersion.EventFormat()
if err != nil {
return err
return fmt.Errorf("queryRes.RoomVersion.EventFormat: %w", err)
}
builder.Depth = queryRes.Depth
@ -96,26 +98,26 @@ func AddPrevEventsToEvent(
for i := range queryRes.StateEvents {
err = authEvents.AddEvent(&queryRes.StateEvents[i].Event)
if err != nil {
return err
return fmt.Errorf("authEvents.AddEvent: %w", err)
}
}
refs, err := eventsNeeded.AuthEventReferences(&authEvents)
if err != nil {
return err
return fmt.Errorf("eventsNeeded.AuthEventReferences: %w", err)
}
truncAuth, truncPrev := truncateAuthAndPrevEvents(refs, queryRes.LatestEvents)
switch eventFormat {
case gomatrixserverlib.EventFormatV1:
builder.AuthEvents = refs
builder.PrevEvents = queryRes.LatestEvents
builder.AuthEvents = truncAuth
builder.PrevEvents = truncPrev
case gomatrixserverlib.EventFormatV2:
v2AuthRefs := []string{}
v2PrevRefs := []string{}
for _, ref := range refs {
v2AuthRefs, v2PrevRefs := []string{}, []string{}
for _, ref := range truncAuth {
v2AuthRefs = append(v2AuthRefs, ref.EventID)
}
for _, ref := range queryRes.LatestEvents {
for _, ref := range truncPrev {
v2PrevRefs = append(v2PrevRefs, ref.EventID)
}
builder.AuthEvents = v2AuthRefs
@ -124,3 +126,21 @@ func AddPrevEventsToEvent(
return nil
}
// truncateAuthAndPrevEvents limits the number of events we add into
// an event as prev_events or auth_events.
// NOTSPEC: The limits here feel a bit arbitrary but they are currently
// here because of https://github.com/matrix-org/matrix-doc/issues/2307
// and because Synapse will just drop events that don't comply.
func truncateAuthAndPrevEvents(auth, prev []gomatrixserverlib.EventReference) (
truncAuth, truncPrev []gomatrixserverlib.EventReference,
) {
truncAuth, truncPrev = auth, prev
if len(truncAuth) > 10 {
truncAuth = truncAuth[:10]
}
if len(truncPrev) > 20 {
truncPrev = truncPrev[:20]
}
return
}

7
eduserver/api/input.go
View file

@ -15,6 +15,7 @@ package api
import (
"context"
"errors"
"net/http"
commonHTTP "github.com/matrix-org/dendrite/common/http"
@ -57,11 +58,11 @@ type EDUServerInputAPI interface {
const EDUServerInputTypingEventPath = "/api/eduserver/input"
// NewEDUServerInputAPIHTTP creates a EDUServerInputAPI implemented by talking to a HTTP POST API.
func NewEDUServerInputAPIHTTP(eduServerURL string, httpClient *http.Client) EDUServerInputAPI {
func NewEDUServerInputAPIHTTP(eduServerURL string, httpClient *http.Client) (EDUServerInputAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewTypingServerInputAPIHTTP: httpClient is <nil>")
}
return &httpEDUServerInputAPI{eduServerURL, httpClient}
return &httpEDUServerInputAPI{eduServerURL, httpClient}, nil
}
type httpEDUServerInputAPI struct {

37
federationapi/routing/invite.go
View file

@ -15,18 +15,17 @@
package routing
import (
"context"
"encoding/json"
"net/http"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
"github.com/matrix-org/dendrite/clientapi/producers"
"github.com/matrix-org/dendrite/common/config"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
// Invite implements /_matrix/federation/v1/invite/{roomID}/{eventID}
// Invite implements /_matrix/federation/v2/invite/{roomID}/{eventID}
func Invite(
httpReq *http.Request,
request *gomatrixserverlib.FederationRequest,
@ -36,24 +35,14 @@ func Invite(
producer *producers.RoomserverProducer,
keys gomatrixserverlib.KeyRing,
) util.JSONResponse {
// Look up the room version for the room.
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
verRes := api.QueryRoomVersionForRoomResponse{}
if err := producer.QueryAPI.QueryRoomVersionForRoom(context.Background(), &verReq, &verRes); err != nil {
inviteReq := gomatrixserverlib.InviteV2Request{}
if err := json.Unmarshal(request.Content(), &inviteReq); err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.UnsupportedRoomVersion(err.Error()),
}
}
// Decode the event JSON from the request.
event, err := gomatrixserverlib.NewEventFromUntrustedJSON(request.Content(), verRes.RoomVersion)
if err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.NotJSON("The request body could not be decoded into valid JSON. " + err.Error()),
JSON: jsonerror.NotJSON("The request body could not be decoded into an invite request. " + err.Error()),
}
}
event := inviteReq.Event()
// Check that the room ID is correct.
if event.RoomID() != roomID {
@ -71,14 +60,6 @@ func Invite(
}
}
// Check that the event is from the server sending the request.
if event.Origin() != request.Origin() {
return util.JSONResponse{
Code: http.StatusForbidden,
JSON: jsonerror.Forbidden("The invite must be sent by the server it originated on"),
}
}
// Check that the event is signed by the server sending the request.
redacted := event.Redact()
verifyRequests := []gomatrixserverlib.VerifyJSONRequest{{
@ -104,7 +85,11 @@ func Invite(
)
// Add the invite event to the roomserver.
if err = producer.SendInvite(httpReq.Context(), signedEvent); err != nil {
if err = producer.SendInvite(
httpReq.Context(),
signedEvent.Headered(inviteReq.RoomVersion()),
inviteReq.InviteRoomState(),
); err != nil {
util.GetLogger(httpReq.Context()).WithError(err).Error("producer.SendInvite failed")
return jsonerror.InternalServerError()
}

38
federationapi/routing/join.go
View file

@ -15,6 +15,7 @@
package routing
import (
"fmt"
"net/http"
"time"
@ -34,6 +35,7 @@ func MakeJoin(
cfg *config.Dendrite,
query api.RoomserverQueryAPI,
roomID, userID string,
remoteVersions []gomatrixserverlib.RoomVersion,
) util.JSONResponse {
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
verRes := api.QueryRoomVersionForRoomResponse{}
@ -44,6 +46,27 @@ func MakeJoin(
}
}
// Check that the room that the remote side is trying to join is actually
// one of the room versions that they listed in their supported ?ver= in
// the make_join URL.
// https://matrix.org/docs/spec/server_server/r0.1.3#get-matrix-federation-v1-make-join-roomid-userid
remoteSupportsVersion := false
for _, v := range remoteVersions {
if v == verRes.RoomVersion {
remoteSupportsVersion = true
break
}
}
// If it isn't, stop trying to join the room.
if !remoteSupportsVersion {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.UnsupportedRoomVersion(
fmt.Sprintf("Joining server does not support room version %s", verRes.RoomVersion),
),
}
}
_, domain, err := gomatrixserverlib.SplitID('@', userID)
if err != nil {
return util.JSONResponse{
@ -140,7 +163,12 @@ func SendJoin(
if event.RoomID() != roomID {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.BadJSON("The room ID in the request path must match the room ID in the join event JSON"),
JSON: jsonerror.BadJSON(
fmt.Sprintf(
"The room ID in the request path (%q) must match the room ID in the join event JSON (%q)",
roomID, event.RoomID(),
),
),
}
}
@ -148,7 +176,12 @@ func SendJoin(
if event.EventID() != eventID {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: jsonerror.BadJSON("The event ID in the request path must match the event ID in the join event JSON"),
JSON: jsonerror.BadJSON(
fmt.Sprintf(
"The event ID in the request path (%q) must match the event ID in the join event JSON (%q)",
eventID, event.EventID(),
),
),
}
}
@ -186,6 +219,7 @@ func SendJoin(
PrevEventIDs: event.PrevEventIDs(),
AuthEventIDs: event.AuthEventIDs(),
RoomID: roomID,
ResolveState: true,
}, &stateAndAuthChainResponse)
if err != nil {
util.GetLogger(httpReq.Context()).WithError(err).Error("query.QueryStateAndAuthChain failed")

40
federationapi/routing/routing.go
View file

@ -85,7 +85,7 @@ func Setup(
},
)).Methods(http.MethodPut, http.MethodOptions)
v1fedmux.Handle("/invite/{roomID}/{eventID}", common.MakeFedAPI(
v2fedmux.Handle("/invite/{roomID}/{eventID}", common.MakeFedAPI(
"federation_invite", cfg.Matrix.ServerName, keys,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
@ -198,7 +198,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/make_join/{roomID}/{userID}", common.MakeFedAPI(
v1fedmux.Handle("/make_join/{roomID}/{eventID}", common.MakeFedAPI(
"federation_make_join", cfg.Matrix.ServerName, keys,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
@ -206,14 +206,28 @@ func Setup(
return util.ErrorResponse(err)
}
roomID := vars["roomID"]
userID := vars["userID"]
eventID := vars["eventID"]
queryVars := httpReq.URL.Query()
remoteVersions := []gomatrixserverlib.RoomVersion{}
if vers, ok := queryVars["ver"]; ok {
// The remote side supplied a ?=ver so use that to build up the list
// of supported room versions
for _, v := range vers {
remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersion(v))
}
} else {
// The remote side didn't supply a ?ver= so just assume that they only
// support room version 1, as per the spec
// https://matrix.org/docs/spec/server_server/r0.1.3#get-matrix-federation-v1-make-join-roomid-userid
remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersionV1)
}
return MakeJoin(
httpReq, request, cfg, query, roomID, userID,
httpReq, request, cfg, query, roomID, eventID, remoteVersions,
)
},
)).Methods(http.MethodGet)
v2fedmux.Handle("/send_join/{roomID}/{userID}", common.MakeFedAPI(
v2fedmux.Handle("/send_join/{roomID}/{eventID}", common.MakeFedAPI(
"federation_send_join", cfg.Matrix.ServerName, keys,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
@ -221,14 +235,14 @@ func Setup(
return util.ErrorResponse(err)
}
roomID := vars["roomID"]
userID := vars["userID"]
eventID := vars["eventID"]
return SendJoin(
httpReq, request, cfg, query, producer, keys, roomID, userID,
httpReq, request, cfg, query, producer, keys, roomID, eventID,
)
},
)).Methods(http.MethodPut)
v1fedmux.Handle("/make_leave/{roomID}/{userID}", common.MakeFedAPI(
v1fedmux.Handle("/make_leave/{roomID}/{eventID}", common.MakeFedAPI(
"federation_make_leave", cfg.Matrix.ServerName, keys,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
@ -236,14 +250,14 @@ func Setup(
return util.ErrorResponse(err)
}
roomID := vars["roomID"]
userID := vars["userID"]
eventID := vars["eventID"]
return MakeLeave(
httpReq, request, cfg, query, roomID, userID,
httpReq, request, cfg, query, roomID, eventID,
)
},
)).Methods(http.MethodGet)
v2fedmux.Handle("/send_leave/{roomID}/{userID}", common.MakeFedAPI(
v2fedmux.Handle("/send_leave/{roomID}/{eventID}", common.MakeFedAPI(
"federation_send_leave", cfg.Matrix.ServerName, keys,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
@ -251,9 +265,9 @@ func Setup(
return util.ErrorResponse(err)
}
roomID := vars["roomID"]
userID := vars["userID"]
eventID := vars["eventID"]
return SendLeave(
httpReq, request, cfg, producer, keys, roomID, userID,
httpReq, request, cfg, producer, keys, roomID, eventID,
)
},
)).Methods(http.MethodPut)

3
federationapi/routing/state.go
View file

@ -107,7 +107,6 @@ func getState(
return nil, &util.JSONResponse{Code: http.StatusNotFound, JSON: nil}
}
prevEventIDs := getIDsFromEventRef(event.PrevEvents())
authEventIDs := getIDsFromEventRef(event.AuthEvents())
var response api.QueryStateAndAuthChainResponse
@ -115,7 +114,7 @@ func getState(
ctx,
&api.QueryStateAndAuthChainRequest{
RoomID: roomID,
PrevEventIDs: prevEventIDs,
PrevEventIDs: []string{eventID},
AuthEventIDs: authEventIDs,
},
&response,

9
federationsender/api/query.go
View file

@ -2,6 +2,7 @@ package api
import (
"context"
"errors"
"net/http"
commonHTTP "github.com/matrix-org/dendrite/common/http"
@ -58,12 +59,12 @@ const FederationSenderQueryJoinedHostsInRoomPath = "/api/federationsender/queryJ
const FederationSenderQueryJoinedHostServerNamesInRoomPath = "/api/federationsender/queryJoinedHostServerNamesInRoom"
// NewFederationSenderQueryAPIHTTP creates a FederationSenderQueryAPI implemented by talking to a HTTP POST API.
// If httpClient is nil then it uses the http.DefaultClient
func NewFederationSenderQueryAPIHTTP(federationSenderURL string, httpClient *http.Client) FederationSenderQueryAPI {
// If httpClient is nil an error is returned
func NewFederationSenderQueryAPIHTTP(federationSenderURL string, httpClient *http.Client) (FederationSenderQueryAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewFederationSenderQueryAPIHTTP: httpClient is <nil>")
}
return &httpFederationSenderQueryAPI{federationSenderURL, httpClient}
return &httpFederationSenderQueryAPI{federationSenderURL, httpClient}, nil
}
type httpFederationSenderQueryAPI struct {

120
federationsender/consumers/roomserver.go
View file

@ -32,6 +32,7 @@ import (
// OutputRoomEventConsumer consumes events that originated in the room server.
type OutputRoomEventConsumer struct {
cfg *config.Dendrite
roomServerConsumer *common.ContinualConsumer
db storage.Database
queues *queue.OutgoingQueues
@ -52,6 +53,7 @@ func NewOutputRoomEventConsumer(
PartitionStore: store,
}
s := &OutputRoomEventConsumer{
cfg: cfg,
roomServerConsumer: &consumer,
db: store,
queues: queues,
@ -79,29 +81,48 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
log.WithError(err).Errorf("roomserver output log: message parse failure")
return nil
}
if output.Type != api.OutputTypeNewRoomEvent {
switch output.Type {
case api.OutputTypeNewRoomEvent:
ev := &output.NewRoomEvent.Event
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"room_id": ev.RoomID(),
"send_as_server": output.NewRoomEvent.SendAsServer,
}).Info("received room event from roomserver")
if err := s.processMessage(*output.NewRoomEvent); err != nil {
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event": string(ev.JSON()),
"add": output.NewRoomEvent.AddsStateEventIDs,
"del": output.NewRoomEvent.RemovesStateEventIDs,
log.ErrorKey: err,
}).Panicf("roomserver output log: write room event failure")
return nil
}
case api.OutputTypeNewInviteEvent:
ev := &output.NewInviteEvent.Event
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"room_id": ev.RoomID(),
"state_key": ev.StateKey(),
}).Info("received invite event from roomserver")
if err := s.processInvite(*output.NewInviteEvent); err != nil {
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event": string(ev.JSON()),
log.ErrorKey: err,
}).Panicf("roomserver output log: write invite event failure")
return nil
}
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
)
return nil
}
ev := &output.NewRoomEvent.Event
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"room_id": ev.RoomID(),
"send_as_server": output.NewRoomEvent.SendAsServer,
}).Info("received event from roomserver")
if err := s.processMessage(*output.NewRoomEvent); err != nil {
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event": string(ev.JSON()),
log.ErrorKey: err,
"add": output.NewRoomEvent.AddsStateEventIDs,
"del": output.NewRoomEvent.RemovesStateEventIDs,
}).Panicf("roomserver output log: write event failure")
return nil
}
return nil
}
@ -159,6 +180,69 @@ func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) err
)
}
// processInvite handles an invite event for sending over federation.
func (s *OutputRoomEventConsumer) processInvite(oie api.OutputNewInviteEvent) error {
// Don't try to reflect and resend invites that didn't originate from us.
if s.cfg.Matrix.ServerName != oie.Event.Origin() {
return nil
}
// When sending a v2 invite, the inviting server should try and include
// a "stripped down" version of the room state. This is pretty much just
// enough information for the remote side to show something useful to the
// user, like the room name, aliases etc.
strippedState := []gomatrixserverlib.InviteV2StrippedState{}
stateWanted := []string{
gomatrixserverlib.MRoomName, gomatrixserverlib.MRoomCanonicalAlias,
gomatrixserverlib.MRoomAliases, gomatrixserverlib.MRoomJoinRules,
}
// For each of the state keys that we want to try and send, ask the
// roomserver if we have a state event for that room that matches the
// state key.
for _, wanted := range stateWanted {
queryReq := api.QueryLatestEventsAndStateRequest{
RoomID: oie.Event.RoomID(),
StateToFetch: []gomatrixserverlib.StateKeyTuple{
gomatrixserverlib.StateKeyTuple{
EventType: wanted,
StateKey: "",
},
},
}
// If this fails then we just move onto the next event - we don't
// actually know at this point whether the room even has that type
// of state.
queryRes := api.QueryLatestEventsAndStateResponse{}
if err := s.query.QueryLatestEventsAndState(context.TODO(), &queryReq, &queryRes); err != nil {
log.WithFields(log.Fields{
"room_id": queryReq.RoomID,
"event_type": wanted,
}).WithError(err).Info("couldn't find state to strip")
continue
}
// Append the stripped down copy of the state to our list.
for _, headeredEvent := range queryRes.StateEvents {
event := headeredEvent.Unwrap()
strippedState = append(strippedState, gomatrixserverlib.NewInviteV2StrippedState(&event))
log.WithFields(log.Fields{
"room_id": queryReq.RoomID,
"event_type": event.Type(),
}).Info("adding stripped state")
}
}
// Build the invite request with the info we've got.
inviteReq, err := gomatrixserverlib.NewInviteV2Request(&oie.Event, strippedState)
if err != nil {
return fmt.Errorf("gomatrixserverlib.NewInviteV2Request: %w", err)
}
// Send the event.
return s.queues.SendInvite(&inviteReq)
}
// joinedHostsAtEvent works out a list of matrix servers that were joined to
// the room at the event.
// It is important to use the state at the event for sending messages because:

98
federationsender/queue/destinationqueue.go
View file

@ -24,6 +24,7 @@ import (
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
log "github.com/sirupsen/logrus"
"go.uber.org/atomic"
)
// destinationQueue is a queue of events for a single destination.
@ -34,14 +35,15 @@ type destinationQueue struct {
client *gomatrixserverlib.FederationClient
origin gomatrixserverlib.ServerName
destination gomatrixserverlib.ServerName
// The running mutex protects running, sentCounter, lastTransactionIDs and
running atomic.Bool
// The running mutex protects sentCounter, lastTransactionIDs and
// pendingEvents, pendingEDUs.
runningMutex sync.Mutex
running bool
sentCounter int
lastTransactionIDs []gomatrixserverlib.TransactionID
pendingEvents []*gomatrixserverlib.HeaderedEvent
pendingEDUs []*gomatrixserverlib.EDU
pendingInvites []*gomatrixserverlib.InviteV2Request
}
// Send event adds the event to the pending queue for the destination.
@ -51,29 +53,43 @@ func (oq *destinationQueue) sendEvent(ev *gomatrixserverlib.HeaderedEvent) {
oq.runningMutex.Lock()
defer oq.runningMutex.Unlock()
oq.pendingEvents = append(oq.pendingEvents, ev)
if !oq.running {
oq.running = true
if !oq.running.Load() {
go oq.backgroundSend()
}
}
// sendEDU adds the EDU event to the pending queue for the destination.
// If the queue is empty then it starts a background goroutine to
// start sending event to that destination.
// start sending events to that destination.
func (oq *destinationQueue) sendEDU(e *gomatrixserverlib.EDU) {
oq.runningMutex.Lock()
defer oq.runningMutex.Unlock()
oq.pendingEDUs = append(oq.pendingEDUs, e)
if !oq.running {
oq.running = true
if !oq.running.Load() {
go oq.backgroundSend()
}
}
// sendInvite adds the invite event to the pending queue for the
// destination. If the queue is empty then it starts a background
// goroutine to start sending events to that destination.
func (oq *destinationQueue) sendInvite(ev *gomatrixserverlib.InviteV2Request) {
oq.runningMutex.Lock()
defer oq.runningMutex.Unlock()
oq.pendingInvites = append(oq.pendingInvites, ev)
if !oq.running.Load() {
go oq.backgroundSend()
}
}
// backgroundSend is the worker goroutine for sending events.
func (oq *destinationQueue) backgroundSend() {
oq.running.Store(true)
defer oq.running.Store(false)
for {
t := oq.next()
if t == nil {
transaction, invites := oq.nextTransaction(), oq.nextInvites()
if !transaction && !invites {
// If the queue is empty then stop processing for this destination.
// TODO: Remove this destination from the queue map.
return
@ -81,29 +97,18 @@ func (oq *destinationQueue) backgroundSend() {
// TODO: handle retries.
// TODO: blacklist uncooperative servers.
util.GetLogger(context.TODO()).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs))
_, err := oq.client.SendTransaction(context.TODO(), *t)
if err != nil {
log.WithFields(log.Fields{
"destination": oq.destination,
log.ErrorKey: err,
}).Info("problem sending transaction")
}
}
}
// next creates a new transaction from the pending event queue
// and flushes the queue.
// Returns nil if the queue was empty.
func (oq *destinationQueue) next() *gomatrixserverlib.Transaction {
// nextTransaction creates a new transaction from the pending event
// queue and sends it. Returns true if a transaction was sent or
// false otherwise.
func (oq *destinationQueue) nextTransaction() bool {
oq.runningMutex.Lock()
defer oq.runningMutex.Unlock()
if len(oq.pendingEvents) == 0 && len(oq.pendingEDUs) == 0 {
oq.running = false
return nil
return false
}
t := gomatrixserverlib.Transaction{
@ -136,5 +141,46 @@ func (oq *destinationQueue) next() *gomatrixserverlib.Transaction {
oq.pendingEDUs = nil
oq.sentCounter += len(t.EDUs)
return &t
util.GetLogger(context.TODO()).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs))
_, err := oq.client.SendTransaction(context.TODO(), t)
if err != nil {
log.WithFields(log.Fields{
"destination": oq.destination,
log.ErrorKey: err,
}).Info("problem sending transaction")
}
return true
}
// nextInvite takes pending invite events from the queue and sends
// them. Returns true if a transaction was sent or false otherwise.
func (oq *destinationQueue) nextInvites() bool {
oq.runningMutex.Lock()
defer oq.runningMutex.Unlock()
if len(oq.pendingInvites) == 0 {
return false
}
for _, inviteReq := range oq.pendingInvites {
ev := inviteReq.Event()
if _, err := oq.client.SendInviteV2(
context.TODO(),
oq.destination,
*inviteReq,
); err != nil {
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"state_key": ev.StateKey(),
"destination": oq.destination,
}).WithError(err).Error("failed to send invite")
}
}
oq.pendingInvites = nil
return true
}

43
federationsender/queue/queue.go
View file

@ -80,6 +80,49 @@ func (oqs *OutgoingQueues) SendEvent(
return nil
}
// SendEvent sends an event to the destinations
func (oqs *OutgoingQueues) SendInvite(
inviteReq *gomatrixserverlib.InviteV2Request,
) error {
ev := inviteReq.Event()
stateKey := ev.StateKey()
if stateKey == nil {
log.WithFields(log.Fields{
"event_id": ev.EventID(),
}).Info("invite had no state key, dropping")
return nil
}
_, destination, err := gomatrixserverlib.SplitID('@', *stateKey)
if err != nil {
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"state_key": stateKey,
}).Info("failed to split destination from state key")
return nil
}
log.WithFields(log.Fields{
"event_id": ev.EventID(),
}).Info("Sending invite")
oqs.queuesMutex.Lock()
defer oqs.queuesMutex.Unlock()
oq := oqs.queues[destination]
if oq == nil {
oq = &destinationQueue{
origin: oqs.origin,
destination: destination,
client: oqs.client,
}
oqs.queues[destination] = oq
}
oq.sendInvite(inviteReq)
return nil
}
// SendEDU sends an EDU event to the destinations
func (oqs *OutgoingQueues) SendEDU(
e *gomatrixserverlib.EDU, origin gomatrixserverlib.ServerName,

6
go.mod
View file

@ -9,7 +9,7 @@ require (
github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f
github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
github.com/matrix-org/gomatrixserverlib v0.0.0-20200327155501-33fb4c7049dc
github.com/matrix-org/gomatrixserverlib v0.0.0-20200409140603-8b9a51fe9b89
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7
github.com/mattn/go-sqlite3 v2.0.2+incompatible
@ -23,9 +23,9 @@ require (
github.com/tidwall/pretty v1.0.1 // indirect
github.com/uber/jaeger-client-go v2.22.1+incompatible
github.com/uber/jaeger-lib v2.2.0+incompatible
go.uber.org/atomic v1.6.0 // indirect
go.uber.org/atomic v1.6.0
golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b // indirect
golang.org/x/tools v0.0.0-20200402223321-bcf690261a44 // indirect
gopkg.in/Shopify/sarama.v1 v1.20.1
gopkg.in/h2non/bimg.v1 v1.0.18
gopkg.in/yaml.v2 v2.2.5

15
go.sum
View file

@ -130,8 +130,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200327155501-33fb4c7049dc h1:qrRu4/AlulnldLiyGpYYm+ELIkrP51XCRlA3txWpN30=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200327155501-33fb4c7049dc/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200409140603-8b9a51fe9b89 h1:YAlUJK/Ty2ZrP/DL41CiR0Cp3pteshnyIS420KVs220=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200409140603-8b9a51fe9b89/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk=
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A=
github.com/matrix-org/util v0.0.0-20171127121716-2e2df66af2f5 h1:W7l5CP4V7wPyPb4tYE11dbmeAOwtFQBTW0rf4OonOS8=
@ -253,6 +253,7 @@ github.com/uber/jaeger-lib v1.5.0/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/Aaua
github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=
github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
go.uber.org/atomic v1.3.0 h1:vs7fgriifsPbGdK3bNuMWapNn3qnZhCRXc19NRdq010=
go.uber.org/atomic v1.3.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
@ -276,6 +277,8 @@ golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvx
golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/mod v0.2.0 h1:KU7oHjnv3XNWfa5COkzUifxZmxp1TyI7ImMXqFxLwvQ=
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@ -288,8 +291,8 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b h1:XfVGCX+0T4WOStkaOsJRllbsiImhB2jgVBGc9L0lPGc=
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b h1:0mm1VjtFUOIlE1SbDlwjYaDxZVDP2S5ou6y0gSgXHu8=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@ -324,7 +327,11 @@ golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c h1:IGkKhmfzcztjm6gYkykvu/NiS8kaqbCWAEWWAyf8J5U=
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200402223321-bcf690261a44 h1:bMm0eoDiGkM5VfIyKjxDvoflW5GLp7+VCo+60n8F+TE=
golang.org/x/tools v0.0.0-20200402223321-bcf690261a44/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=

9
roomserver/api/alias.go
View file

@ -16,6 +16,7 @@ package api
import (
"context"
"errors"
"net/http"
commonHTTP "github.com/matrix-org/dendrite/common/http"
@ -139,12 +140,12 @@ const RoomserverGetCreatorIDForAliasPath = "/api/roomserver/GetCreatorIDForAlias
const RoomserverRemoveRoomAliasPath = "/api/roomserver/removeRoomAlias"
// NewRoomserverAliasAPIHTTP creates a RoomserverAliasAPI implemented by talking to a HTTP POST API.
// If httpClient is nil then it uses the http.DefaultClient
func NewRoomserverAliasAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverAliasAPI {
// If httpClient is nil an error is returned
func NewRoomserverAliasAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverAliasAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewRoomserverAliasAPIHTTP: httpClient is <nil>")
}
return &httpRoomserverAliasAPI{roomserverURL, httpClient}
return &httpRoomserverAliasAPI{roomserverURL, httpClient}, nil
}
type httpRoomserverAliasAPI struct {

13
roomserver/api/input.go
View file

@ -17,6 +17,7 @@ package api
import (
"context"
"errors"
"net/http"
commonHTTP "github.com/matrix-org/dendrite/common/http"
@ -85,7 +86,9 @@ type TransactionID struct {
// the usual context a matrix room event would have. We usually do not have
// access to the events needed to check the event auth rules for the invite.
type InputInviteEvent struct {
Event gomatrixserverlib.HeaderedEvent `json:"event"`
RoomVersion gomatrixserverlib.RoomVersion `json:"room_version"`
Event gomatrixserverlib.HeaderedEvent `json:"event"`
InviteRoomState []gomatrixserverlib.InviteV2StrippedState `json:"invite_room_state"`
}
// InputRoomEventsRequest is a request to InputRoomEvents
@ -112,12 +115,12 @@ type RoomserverInputAPI interface {
const RoomserverInputRoomEventsPath = "/api/roomserver/inputRoomEvents"
// NewRoomserverInputAPIHTTP creates a RoomserverInputAPI implemented by talking to a HTTP POST API.
// If httpClient is nil then it uses the http.DefaultClient
func NewRoomserverInputAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverInputAPI {
// If httpClient is nil an error is returned
func NewRoomserverInputAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverInputAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewRoomserverInputAPIHTTP: httpClient is <nil>")
}
return &httpRoomserverInputAPI{roomserverURL, httpClient}
return &httpRoomserverInputAPI{roomserverURL, httpClient}, nil
}
type httpRoomserverInputAPI struct {

2
roomserver/api/output.go
View file

@ -116,6 +116,8 @@ type OutputNewRoomEvent struct {
// Invite events can be received outside of an existing room so have to be
// tracked separately from the room events themselves.
type OutputNewInviteEvent struct {
// The room version of the invited room.
RoomVersion gomatrixserverlib.RoomVersion `json:"room_version"`
// The "m.room.member" invite event.
Event gomatrixserverlib.HeaderedEvent `json:"event"`
}

12
roomserver/api/query.go
View file

@ -18,6 +18,7 @@ package api
import (
"context"
"errors"
"net/http"
commonHTTP "github.com/matrix-org/dendrite/common/http"
@ -202,6 +203,9 @@ type QueryStateAndAuthChainRequest struct {
PrevEventIDs []string `json:"prev_event_ids"`
// The list of auth events for the event. Used to calculate the auth chain
AuthEventIDs []string `json:"auth_event_ids"`
// Should state resolution be ran on the result events?
// TODO: check call sites and remove if we always want to do state res
ResolveState bool `json:"resolve_state"`
}
// QueryStateAndAuthChainResponse is a response to QueryStateAndAuthChain
@ -406,12 +410,12 @@ const RoomserverQueryRoomVersionCapabilitiesPath = "/api/roomserver/queryRoomVer
const RoomserverQueryRoomVersionForRoomPath = "/api/roomserver/queryRoomVersionForRoom"
// NewRoomserverQueryAPIHTTP creates a RoomserverQueryAPI implemented by talking to a HTTP POST API.
// If httpClient is nil then it uses the http.DefaultClient
func NewRoomserverQueryAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverQueryAPI {
// If httpClient is nil an error is returned
func NewRoomserverQueryAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverQueryAPI, error) {
if httpClient == nil {
httpClient = http.DefaultClient
return nil, errors.New("NewRoomserverQueryAPIHTTP: httpClient is <nil>")
}
return &httpRoomserverQueryAPI{roomserverURL, httpClient}
return &httpRoomserverQueryAPI{roomserverURL, httpClient}, nil
}
type httpRoomserverQueryAPI struct {

18
roomserver/input/events.go
View file

@ -26,6 +26,7 @@ import (
"github.com/matrix-org/dendrite/roomserver/state/database"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
log "github.com/sirupsen/logrus"
)
// A RoomEventDatabase has the storage APIs needed to store a room event.
@ -64,6 +65,7 @@ type RoomEventDatabase interface {
// Build a membership updater for the target user in a room.
MembershipUpdater(
ctx context.Context, roomID, targerUserID string,
roomVersion gomatrixserverlib.RoomVersion,
) (types.MembershipUpdater, error)
// Look up event ID by transaction's info.
// This is used to determine if the room event is processed/processing already.
@ -193,7 +195,14 @@ func processInviteEvent(
roomID := input.Event.RoomID()
targetUserID := *input.Event.StateKey()
updater, err := db.MembershipUpdater(ctx, roomID, targetUserID)
log.WithFields(log.Fields{
"event_id": input.Event.EventID(),
"room_id": roomID,
"room_version": input.RoomVersion,
"target_user_id": targetUserID,
}).Info("processing invite event")
updater, err := db.MembershipUpdater(ctx, roomID, targetUserID, input.RoomVersion)
if err != nil {
return err
}
@ -237,7 +246,12 @@ func processInviteEvent(
}
event := input.Event.Unwrap()
outputUpdates, err := updateToInviteMembership(updater, &event, nil)
if err = event.SetUnsignedField("invite_room_state", input.InviteRoomState); err != nil {
return err
}
outputUpdates, err := updateToInviteMembership(updater, &event, nil, input.Event.RoomVersion)
if err != nil {
return err
}

7
roomserver/input/membership.go
View file

@ -112,7 +112,7 @@ func updateMembership(
switch newMembership {
case gomatrixserverlib.Invite:
return updateToInviteMembership(mu, add, updates)
return updateToInviteMembership(mu, add, updates, updater.RoomVersion())
case gomatrixserverlib.Join:
return updateToJoinMembership(mu, add, updates)
case gomatrixserverlib.Leave, gomatrixserverlib.Ban:
@ -126,6 +126,7 @@ func updateMembership(
func updateToInviteMembership(
mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
roomVersion gomatrixserverlib.RoomVersion,
) ([]api.OutputEvent, error) {
// We may have already sent the invite to the user, either because we are
// reprocessing this event, or because the we received this invite from a
@ -136,14 +137,14 @@ func updateToInviteMembership(
return nil, err
}
if needsSending {
roomVersion := gomatrixserverlib.RoomVersionV1
// We notify the consumers using a special event even though we will
// notify them about the change in current state as part of the normal
// room event stream. This ensures that the consumers only have to
// consider a single stream of events when determining whether a user
// is invited, rather than having to combine multiple streams themselves.
onie := api.OutputNewInviteEvent{
Event: (*add).Headered(roomVersion),
Event: (*add).Headered(roomVersion),
RoomVersion: roomVersion,
}
updates = append(updates, api.OutputEvent{
Type: api.OutputTypeNewInviteEvent,

10
roomserver/query/query.go
View file

@ -132,7 +132,7 @@ func (r *RoomserverQueryAPI) QueryLatestEventsAndState(
return err
}
// Look up the currrent state for the requested tuples.
// Look up the current state for the requested tuples.
stateEntries, err := roomState.LoadStateAtSnapshotForStringTuples(
ctx, currentStateSnapshotNID, request.StateToFetch,
)
@ -736,6 +736,14 @@ func (r *RoomserverQueryAPI) QueryStateAndAuthChain(
return err
}
if request.ResolveState {
if stateEvents, err = state.ResolveConflictsAdhoc(
roomVersion, stateEvents, authEvents,
); err != nil {
return err
}
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(roomVersion))
}

80
roomserver/state/state.go
View file

@ -18,7 +18,6 @@ package state
import (
"context"
"errors"
"fmt"
"sort"
"time"
@ -681,6 +680,83 @@ func (v StateResolution) calculateStateAfterManyEvents(
return
}
// ResolveConflictsAdhoc is a helper function to assist the query API in
// performing state resolution when requested. This is a different code
// path to the rest of state.go because this assumes you already have
// gomatrixserverlib.Event objects and not just a bunch of NIDs like
// elsewhere in the state resolution.
// TODO: Some of this can possibly be deduplicated
func ResolveConflictsAdhoc(
version gomatrixserverlib.RoomVersion,
events []gomatrixserverlib.Event,
authEvents []gomatrixserverlib.Event,
) ([]gomatrixserverlib.Event, error) {
type stateKeyTuple struct {
Type string
StateKey string
}
// Prepare our data structures.
eventMap := make(map[stateKeyTuple][]gomatrixserverlib.Event)
var conflicted, notConflicted, resolved []gomatrixserverlib.Event
// Run through all of the events that we were given and sort them
// into a map, sorted by (event_type, state_key) tuple. This means
// that we can easily spot events that are "conflicted", e.g.
// there are duplicate values for the same tuple key.
for _, event := range events {
if event.StateKey() == nil {
// Ignore events that are not state events.
continue
}
// Append the events if there is already a conflicted list for
// this tuple key, create it if not.
tuple := stateKeyTuple{event.Type(), *event.StateKey()}
if _, ok := eventMap[tuple]; ok {
eventMap[tuple] = append(eventMap[tuple], event)
} else {
eventMap[tuple] = []gomatrixserverlib.Event{event}
}
}
// Split out the events in the map into conflicted and unconflicted
// buckets. The conflicted events will be ran through state res,
// whereas unconfliced events will always going to appear in the
// final resolved state.
for _, list := range eventMap {
if len(list) > 1 {
conflicted = append(conflicted, list...)
} else {
notConflicted = append(notConflicted, list...)
}
}
// Work out which state resolution algorithm we want to run for
// the room version.
stateResAlgo, err := version.StateResAlgorithm()
if err != nil {
return nil, err
}
switch stateResAlgo {
case gomatrixserverlib.StateResV1:
// Currently state res v1 doesn't handle unconflicted events
// for us, like state res v2 does, so we will need to add the
// unconflicted events into the state ourselves.
// TODO: Fix state res v1 so this is handled for the caller.
resolved = gomatrixserverlib.ResolveStateConflicts(conflicted, authEvents)
resolved = append(resolved, notConflicted...)
case gomatrixserverlib.StateResV2:
// TODO: auth difference here?
resolved = gomatrixserverlib.ResolveStateConflictsV2(conflicted, notConflicted, authEvents, authEvents)
default:
return nil, fmt.Errorf("unsupported state resolution algorithm %v", stateResAlgo)
}
// Return the final resolved state events, including both the
// resolved set of conflicted events, and the unconflicted events.
return resolved, nil
}
func (v StateResolution) resolveConflicts(
ctx context.Context, version gomatrixserverlib.RoomVersion,
notConflicted, conflicted []types.StateEntry,
@ -695,7 +771,7 @@ func (v StateResolution) resolveConflicts(
case gomatrixserverlib.StateResV2:
return v.resolveConflictsV2(ctx, notConflicted, conflicted)
}
return nil, errors.New("unsupported state resolution algorithm")
return nil, fmt.Errorf("unsupported state resolution algorithm %v", stateResAlgo)
}
// resolveConflicts resolves a list of conflicted state entries. It takes two lists.

2
roomserver/storage/interface.go
View file

@ -41,7 +41,7 @@ type Database interface {
GetAliasesForRoomID(ctx context.Context, roomID string) ([]string, error)
GetCreatorIDForAlias(ctx context.Context, alias string) (string, error)
RemoveRoomAlias(ctx context.Context, alias string) error
MembershipUpdater(ctx context.Context, roomID, targetUserID string) (types.MembershipUpdater, error)
MembershipUpdater(ctx context.Context, roomID, targetUserID string, roomVersion gomatrixserverlib.RoomVersion) (types.MembershipUpdater, error)
GetMembership(ctx context.Context, roomNID types.RoomNID, requestSenderUserID string) (membershipEventNID types.EventNID, stillInRoom bool, err error)
GetMembershipEventNIDsForRoom(ctx context.Context, roomNID types.RoomNID, joinOnly bool) ([]types.EventNID, error)
EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error)

10
roomserver/storage/postgres/storage.go
View file

@ -393,6 +393,12 @@ type roomRecentEventsUpdater struct {
currentStateSnapshotNID types.StateSnapshotNID
}
// RoomVersion implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
return
}
// LatestEvents implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
return u.latestEvents
@ -534,6 +540,7 @@ func (d *Database) StateEntriesForTuples(
// MembershipUpdater implements input.RoomEventDatabase
func (d *Database) MembershipUpdater(
ctx context.Context, roomID, targetUserID string,
roomVersion gomatrixserverlib.RoomVersion,
) (types.MembershipUpdater, error) {
txn, err := d.db.Begin()
if err != nil {
@ -546,8 +553,7 @@ func (d *Database) MembershipUpdater(
}
}()
// TODO: Room version here
roomNID, err := d.assignRoomNID(ctx, txn, roomID, "1")
roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
if err != nil {
return nil, err
}

10
roomserver/storage/sqlite3/storage.go
View file

@ -486,6 +486,12 @@ type roomRecentEventsUpdater struct {
currentStateSnapshotNID types.StateSnapshotNID
}
// RoomVersion implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
return
}
// LatestEvents implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
return u.latestEvents
@ -657,6 +663,7 @@ func (d *Database) StateEntriesForTuples(
// MembershipUpdater implements input.RoomEventDatabase
func (d *Database) MembershipUpdater(
ctx context.Context, roomID, targetUserID string,
roomVersion gomatrixserverlib.RoomVersion,
) (updater types.MembershipUpdater, err error) {
var txn *sql.Tx
txn, err = d.db.Begin()
@ -682,8 +689,7 @@ func (d *Database) MembershipUpdater(
}
}()
// TODO: Room version here
roomNID, err := d.assignRoomNID(ctx, txn, roomID, "1")
roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
if err != nil {
return nil, err
}

2
roomserver/types/types.go
View file

@ -140,6 +140,8 @@ type StateEntryList struct {
// (On postgresql this wraps a database transaction that holds a "FOR UPDATE"
// lock on the row in the rooms table holding the latest events for the room.)
type RoomRecentEventsUpdater interface {
// The room version of the room.
RoomVersion() gomatrixserverlib.RoomVersion
// The latest event IDs and state in the room.
LatestEvents() []StateAtEventAndReference
// The event ID of the latest event written to the output log in the room.

8
roomserver/version/version.go
View file

@ -43,12 +43,12 @@ var roomVersions = map[gomatrixserverlib.RoomVersion]RoomVersionDescription{
Stable: true,
},
gomatrixserverlib.RoomVersionV3: RoomVersionDescription{
Supported: false,
Stable: false,
Supported: true,
Stable: true,
},
gomatrixserverlib.RoomVersionV4: RoomVersionDescription{
Supported: false,
Stable: false,
Supported: true,
Stable: true,
},
gomatrixserverlib.RoomVersionV5: RoomVersionDescription{
Supported: false,

5
sytest-blacklist
View file

@ -31,4 +31,7 @@ Alias creators can delete canonical alias with no ops
# Blacklisted because we need to implement v2 invite endpoints for room versions
# to be supported (currently fails with M_UNSUPPORTED_ROOM_VERSION)
Inbound federation rejects invites which are not signed by the sender
Inbound federation rejects invites which are not signed by the sender
# Blacklisted because we don't support ignores yet
Ignore invite in incremental sync

18
sytest-whitelist
View file

@ -215,7 +215,6 @@ Guest users can sync from default guest_access rooms if joined
Real non-joined users cannot room initalSync for non-world_readable rooms
Push rules come down in an initial /sync
Regular users can add and delete aliases in the default room configuration
Regular users can add and delete aliases when m.room.aliases is restricted
GET /r0/capabilities is not public
GET /joined_rooms lists newly-created room
/joined_rooms returns only joined rooms
@ -225,9 +224,24 @@ Remote user can backfill in a room with version 1
POST /createRoom creates a room with the given version
POST /createRoom rejects attempts to create rooms with numeric versions
POST /createRoom rejects attempts to create rooms with unknown versions
Regular users can add and delete aliases when m.room.aliases is restricted
User can create and send/receive messages in a room with version 2
local user can join room with version 2
remote user can join room with version 2
User can invite local user to room with version 2
Remote user can backfill in a room with version 2
Inbound federation accepts attempts to join v2 rooms from servers with support
Inbound federation accepts attempts to join v2 rooms from servers with support
Outbound federation can send invites via v2 API
User can create and send/receive messages in a room with version 3
local user can join room with version 3
Remote user can backfill in a room with version 3
User can create and send/receive messages in a room with version 4
local user can join room with version 4
remote user can join room with version 3
remote user can join room with version 4
Remote user can backfill in a room with version 4
# We don't support ignores yet, so ignore this for now - ha ha.
# Ignore invite in incremental sync
Outbound federation can send invites via v2 API
User can invite local user to room with version 3
User can invite local user to room with version 4