0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-10-31 13:09:08 +01:00
dendrite/roomserver/roomserver_test.go

1322 lines
47 KiB
Go
Raw Permalink Normal View History

package roomserver_test
import (
"context"
"crypto/ed25519"
"reflect"
"testing"
"time"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
Use `AckExplicitPolicy` instead of `AckAllPolicy` (#3288) Fixes https://github.com/matrix-org/dendrite/issues/3240 and potentially a root cause for state resets. While testing, I've had added some more debug logging: ``` time="2023-12-16T18:13:11.319458084Z" level=warning msg="already processed event" event_id="$qFYMl_F2vb1N0yxmvlFAMhqhGhLKq4kA-o_YCQKH7tQ" kind=KindNew times=2 time="2023-12-16T18:13:14.537389126Z" level=warning msg="already processed event" event_id="$EU-LTsKErT6Mt1k12-p_3xOHfiLaK6gtwVDlZ35lSuo" kind=KindNew times=5 time="2023-12-16T18:13:16.789551206Z" level=warning msg="already processed event" event_id="$dIPuAfTL5x0VyG873LKPslQeljCSxFT1WKxUtjIMUGE" kind=KindNew times=5 time="2023-12-16T18:13:17.383838767Z" level=warning msg="already processed event" event_id="$7noSZiCkzerpkz_UBO3iatpRnaOiPx-3IXc0GPDQVGE" kind=KindNew times=2 time="2023-12-16T18:13:22.091946597Z" level=warning msg="already processed event" event_id="$3Lvo3Wbi2ol9-nNbQ93N-E2MuGQCJZo5397KkFH-W6E" kind=KindNew times=1 time="2023-12-16T18:13:23.026417446Z" level=warning msg="already processed event" event_id="$lj1xS46zsLBCChhKOLJEG-bu7z-_pq9i_Y2DUIjzGy4" kind=KindNew times=4 ``` So we did receive the same event over and over again. Given they are `KindNew`, we don't short circuit if we already processed them, which potentially caused the state to be calculated with a now wrong state snapshot. Also fixes the back pressure metric. We now correctly increment the counter once we sent the message to NATS and decrement it once we actually processed an event.
2023-12-19 08:25:47 +01:00
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/gomatrixserverlib/spec"
Use `AckExplicitPolicy` instead of `AckAllPolicy` (#3288) Fixes https://github.com/matrix-org/dendrite/issues/3240 and potentially a root cause for state resets. While testing, I've had added some more debug logging: ``` time="2023-12-16T18:13:11.319458084Z" level=warning msg="already processed event" event_id="$qFYMl_F2vb1N0yxmvlFAMhqhGhLKq4kA-o_YCQKH7tQ" kind=KindNew times=2 time="2023-12-16T18:13:14.537389126Z" level=warning msg="already processed event" event_id="$EU-LTsKErT6Mt1k12-p_3xOHfiLaK6gtwVDlZ35lSuo" kind=KindNew times=5 time="2023-12-16T18:13:16.789551206Z" level=warning msg="already processed event" event_id="$dIPuAfTL5x0VyG873LKPslQeljCSxFT1WKxUtjIMUGE" kind=KindNew times=5 time="2023-12-16T18:13:17.383838767Z" level=warning msg="already processed event" event_id="$7noSZiCkzerpkz_UBO3iatpRnaOiPx-3IXc0GPDQVGE" kind=KindNew times=2 time="2023-12-16T18:13:22.091946597Z" level=warning msg="already processed event" event_id="$3Lvo3Wbi2ol9-nNbQ93N-E2MuGQCJZo5397KkFH-W6E" kind=KindNew times=1 time="2023-12-16T18:13:23.026417446Z" level=warning msg="already processed event" event_id="$lj1xS46zsLBCChhKOLJEG-bu7z-_pq9i_Y2DUIjzGy4" kind=KindNew times=4 ``` So we did receive the same event over and over again. Given they are `KindNew`, we don't short circuit if we already processed them, which potentially caused the state to be calculated with a now wrong state snapshot. Also fixes the back pressure metric. We now correctly increment the counter once we sent the message to NATS and decrement it once we actually processed an event.
2023-12-19 08:25:47 +01:00
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/tidwall/gjson"
"github.com/matrix-org/dendrite/roomserver/acls"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/userapi"
userAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
)
var testIsBlacklistedOrBackingOff = func(s spec.ServerName) (*statistics.ServerStatistics, error) {
return &statistics.ServerStatistics{}, nil
}
type FakeQuerier struct {
api.QuerySenderIDAPI
}
func (f *FakeQuerier) QueryUserIDForSender(ctx context.Context, roomID spec.RoomID, senderID spec.SenderID) (*spec.UserID, error) {
return spec.NewUserID(string(senderID), true)
}
func TestUsers(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{}
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// SetFederationAPI starts the room event input consumer
rsAPI.SetFederationAPI(nil, nil)
t.Run("shared users", func(t *testing.T) {
testSharedUsers(t, rsAPI)
})
t.Run("kick users", func(t *testing.T) {
usrAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
rsAPI.SetUserAPI(usrAPI)
testKickUsers(t, rsAPI, usrAPI)
})
})
}
func testSharedUsers(t *testing.T, rsAPI api.RoomserverInternalAPI) {
alice := test.NewUser(t)
bob := test.NewUser(t)
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetTrustedPrivateChat))
// Invite and join Bob
room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "invite",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
ctx := context.Background()
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Query the shared users for Alice, there should only be Bob.
// This is used by the SyncAPI keychange consumer.
res := &api.QuerySharedUsersResponse{}
if err := rsAPI.QuerySharedUsers(ctx, &api.QuerySharedUsersRequest{UserID: alice.ID}, res); err != nil {
t.Errorf("unable to query known users: %v", err)
}
if _, ok := res.UserIDsToCount[bob.ID]; !ok {
t.Errorf("expected to find %s in shared users, but didn't: %+v", bob.ID, res.UserIDsToCount)
}
// Also verify that we get the expected result when specifying OtherUserIDs.
// This is used by the SyncAPI when getting device list changes.
if err := rsAPI.QuerySharedUsers(ctx, &api.QuerySharedUsersRequest{UserID: alice.ID, OtherUserIDs: []string{bob.ID}}, res); err != nil {
t.Errorf("unable to query known users: %v", err)
}
if _, ok := res.UserIDsToCount[bob.ID]; !ok {
t.Errorf("expected to find %s in shared users, but didn't: %+v", bob.ID, res.UserIDsToCount)
}
}
func testKickUsers(t *testing.T, rsAPI api.RoomserverInternalAPI, usrAPI userAPI.UserInternalAPI) {
// Create users and room; Bob is going to be the guest and kicked on revocation of guest access
alice := test.NewUser(t, test.WithAccountType(userAPI.AccountTypeUser))
bob := test.NewUser(t, test.WithAccountType(userAPI.AccountTypeGuest))
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetPublicChat), test.GuestsCanJoin(true))
// Join with the guest user
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
ctx := context.Background()
// Create the users in the userapi, so the RSAPI can query the account type later
for _, u := range []*test.User{alice, bob} {
localpart, serverName, _ := gomatrixserverlib.SplitID('@', u.ID)
userRes := &userAPI.PerformAccountCreationResponse{}
if err := usrAPI.PerformAccountCreation(ctx, &userAPI.PerformAccountCreationRequest{
AccountType: u.AccountType,
Localpart: localpart,
ServerName: serverName,
Password: "someRandomPassword",
}, userRes); err != nil {
t.Errorf("failed to create account: %s", err)
}
}
// Create the room in the database
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Get the membership events BEFORE revoking guest access
membershipRes := &api.QueryMembershipsForRoomResponse{}
if err := rsAPI.QueryMembershipsForRoom(ctx, &api.QueryMembershipsForRoomRequest{LocalOnly: true, JoinedOnly: true, RoomID: room.ID}, membershipRes); err != nil {
t.Errorf("failed to query membership for room: %s", err)
}
// revoke guest access
revokeEvent := room.CreateAndInsert(t, alice, spec.MRoomGuestAccess, map[string]string{"guest_access": "forbidden"}, test.WithStateKey(""))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{revokeEvent}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// TODO: Even though we are sending the events sync, the "kickUsers" function is sending the events async, so we need
// to loop and wait for the events to be processed by the roomserver.
for i := 0; i <= 20; i++ {
// Get the membership events AFTER revoking guest access
membershipRes2 := &api.QueryMembershipsForRoomResponse{}
if err := rsAPI.QueryMembershipsForRoom(ctx, &api.QueryMembershipsForRoomRequest{LocalOnly: true, JoinedOnly: true, RoomID: room.ID}, membershipRes2); err != nil {
t.Errorf("failed to query membership for room: %s", err)
}
// The membership events should NOT match, as Bob (guest user) should now be kicked from the room
if !reflect.DeepEqual(membershipRes, membershipRes2) {
return
}
time.Sleep(time.Millisecond * 10)
}
t.Errorf("memberships didn't change in time")
}
func Test_QueryLeftUsers(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetTrustedPrivateChat))
// Invite and join Bob
room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "invite",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{}
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// SetFederationAPI starts the room event input consumer
rsAPI.SetFederationAPI(nil, nil)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// Query the left users, there should only be "@idontexist:test",
// as Alice and Bob are still joined.
res := &api.QueryLeftUsersResponse{}
leftUserID := "@idontexist:test"
getLeftUsersList := []string{alice.ID, bob.ID, leftUserID}
testCase := func(rsAPI api.RoomserverInternalAPI) {
if err := rsAPI.QueryLeftUsers(ctx, &api.QueryLeftUsersRequest{StaleDeviceListUsers: getLeftUsersList}, res); err != nil {
t.Fatalf("unable to query left users: %v", err)
}
wantCount := 1
if count := len(res.LeftUsers); count > wantCount {
t.Fatalf("unexpected left users count: want %d, got %d", wantCount, count)
}
if res.LeftUsers[0] != leftUserID {
t.Fatalf("unexpected left users : want %s, got %s", leftUserID, res.LeftUsers[0])
}
}
testCase(rsAPI)
})
}
func TestPurgeRoom(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetTrustedPrivateChat))
roomID, err := spec.NewRoomID(room.ID)
if err != nil {
t.Fatal(err)
}
// Invite Bob
inviteEvent := room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "invite",
}, test.WithStateKey(bob.ID))
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
natsInstance := jetstream.NATSInstance{}
defer close()
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
if err != nil {
t.Fatal(err)
}
jsCtx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsCtx, &cfg.Global.JetStream)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// this starts the JetStream consumers
fsAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(fsAPI, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, fsAPI.IsBlacklistedOrBackingOff)
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
// Create the room
if err = api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// some dummy entries to validate after purging
if err = rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{RoomID: room.ID, Visibility: spec.Public}); err != nil {
t.Fatal(err)
}
isPublished, err := db.GetPublishedRoom(ctx, room.ID)
if err != nil {
t.Fatal(err)
}
if !isPublished {
t.Fatalf("room should be published before purging")
}
if _, err = rsAPI.SetRoomAlias(ctx, spec.SenderID(alice.ID), *roomID, "myalias"); err != nil {
t.Fatal(err)
}
// check the alias is actually there
aliasesResp := &api.GetAliasesForRoomIDResponse{}
if err = rsAPI.GetAliasesForRoomID(ctx, &api.GetAliasesForRoomIDRequest{RoomID: room.ID}, aliasesResp); err != nil {
t.Fatal(err)
}
wantAliases := 1
if gotAliases := len(aliasesResp.Aliases); gotAliases != wantAliases {
t.Fatalf("expected %d aliases, got %d", wantAliases, gotAliases)
}
// validate the room exists before purging
roomInfo, err := db.RoomInfo(ctx, room.ID)
if err != nil {
t.Fatal(err)
}
if roomInfo == nil {
t.Fatalf("room does not exist")
}
//
roomInfo2, err := db.RoomInfoByNID(ctx, roomInfo.RoomNID)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(roomInfo, roomInfo2) {
t.Fatalf("expected roomInfos to be the same, but they aren't")
}
// remember the roomInfo before purging
existingRoomInfo := roomInfo
// validate there is an invite for bob
nids, err := db.EventStateKeyNIDs(ctx, []string{bob.ID})
if err != nil {
t.Fatal(err)
}
bobNID, ok := nids[bob.ID]
if !ok {
t.Fatalf("%s does not exist", bob.ID)
}
_, inviteEventIDs, _, err := db.GetInvitesForUser(ctx, roomInfo.RoomNID, bobNID)
if err != nil {
t.Fatal(err)
}
wantInviteCount := 1
if inviteCount := len(inviteEventIDs); inviteCount != wantInviteCount {
t.Fatalf("expected there to be only %d invite events, got %d", wantInviteCount, inviteCount)
}
if inviteEventIDs[0] != inviteEvent.EventID() {
t.Fatalf("expected invite event ID %s, got %s", inviteEvent.EventID(), inviteEventIDs[0])
}
// purge the room from the database
if err = rsAPI.PerformAdminPurgeRoom(ctx, room.ID); err != nil {
t.Fatal(err)
}
// wait for all consumers to process the purge event
var sum = 1
timeout := time.Second * 5
deadline, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
for sum > 0 {
if deadline.Err() != nil {
t.Fatalf("test timed out after %s", timeout)
}
sum = 0
consumerCh := jsCtx.Consumers(cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
for x := range consumerCh {
sum += x.NumAckPending
}
time.Sleep(time.Millisecond)
}
roomInfo, err = db.RoomInfo(ctx, room.ID)
if err != nil {
t.Fatal(err)
}
if roomInfo != nil {
t.Fatalf("room should not exist after purging: %+v", roomInfo)
}
roomInfo2, err = db.RoomInfoByNID(ctx, existingRoomInfo.RoomNID)
if err == nil {
t.Fatalf("expected room to not exist, but it does: %#v", roomInfo2)
}
// validation below
// There should be no invite left
_, inviteEventIDs, _, err = db.GetInvitesForUser(ctx, existingRoomInfo.RoomNID, bobNID)
if err != nil {
t.Fatal(err)
}
if inviteCount := len(inviteEventIDs); inviteCount > 0 {
t.Fatalf("expected there to be only %d invite events, got %d", wantInviteCount, inviteCount)
}
// aliases should be deleted
aliases, err := db.GetAliasesForRoomID(ctx, room.ID)
if err != nil {
t.Fatal(err)
}
if aliasCount := len(aliases); aliasCount > 0 {
t.Fatalf("expected there to be only %d invite events, got %d", 0, aliasCount)
}
// published room should be deleted
isPublished, err = db.GetPublishedRoom(ctx, room.ID)
if err != nil {
t.Fatal(err)
}
if isPublished {
t.Fatalf("room should not be published after purging")
}
})
}
type fledglingEvent struct {
Type string
StateKey *string
2023-06-07 19:14:35 +02:00
SenderID string
RoomID string
Redacts string
Depth int64
PrevEvents []any
AuthEvents []any
Content map[string]any
}
func mustCreateEvent(t *testing.T, ev fledglingEvent) (result *types.HeaderedEvent) {
t.Helper()
roomVer := gomatrixserverlib.RoomVersionV9
seed := make([]byte, ed25519.SeedSize) // zero seed
key := ed25519.NewKeyFromSeed(seed)
eb := gomatrixserverlib.MustGetRoomVersion(roomVer).NewEventBuilderFromProtoEvent(&gomatrixserverlib.ProtoEvent{
2023-06-07 19:14:35 +02:00
SenderID: ev.SenderID,
Type: ev.Type,
StateKey: ev.StateKey,
RoomID: ev.RoomID,
Redacts: ev.Redacts,
Depth: ev.Depth,
PrevEvents: ev.PrevEvents,
})
if ev.Content == nil {
ev.Content = map[string]any{}
}
if ev.AuthEvents != nil {
eb.AuthEvents = ev.AuthEvents
}
err := eb.SetContent(ev.Content)
if err != nil {
t.Fatalf("mustCreateEvent: failed to marshal event content %v", err)
}
signedEvent, err := eb.Build(time.Now(), "localhost", "ed25519:test", key)
if err != nil {
t.Fatalf("mustCreateEvent: failed to sign event: %s", err)
}
h := &types.HeaderedEvent{PDU: signedEvent}
return h
}
func TestRedaction(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := test.NewUser(t, test.WithSigningServer("notlocalhost", "abc", test.PrivateKeyB))
testCases := []struct {
name string
additionalEvents func(t *testing.T, room *test.Room)
wantRedacted bool
}{
{
name: "can redact own message",
wantRedacted: true,
additionalEvents: func(t *testing.T, room *test.Room) {
redactedEvent := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
builderEv := mustCreateEvent(t, fledglingEvent{
Type: spec.MRoomRedaction,
2023-06-07 19:14:35 +02:00
SenderID: alice.ID,
RoomID: room.ID,
Redacts: redactedEvent.EventID(),
Depth: redactedEvent.Depth() + 1,
PrevEvents: []interface{}{redactedEvent.EventID()},
})
room.InsertEvent(t, builderEv)
},
},
{
name: "can redact others message, allowed by PL",
wantRedacted: true,
additionalEvents: func(t *testing.T, room *test.Room) {
redactedEvent := room.CreateAndInsert(t, bob, "m.room.message", map[string]interface{}{"body": "hello world"})
builderEv := mustCreateEvent(t, fledglingEvent{
Type: spec.MRoomRedaction,
2023-06-07 19:14:35 +02:00
SenderID: alice.ID,
RoomID: room.ID,
Redacts: redactedEvent.EventID(),
Depth: redactedEvent.Depth() + 1,
PrevEvents: []interface{}{redactedEvent.EventID()},
})
room.InsertEvent(t, builderEv)
},
},
{
name: "can redact others message, same server",
wantRedacted: true,
additionalEvents: func(t *testing.T, room *test.Room) {
redactedEvent := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
builderEv := mustCreateEvent(t, fledglingEvent{
Type: spec.MRoomRedaction,
2023-06-07 19:14:35 +02:00
SenderID: bob.ID,
RoomID: room.ID,
Redacts: redactedEvent.EventID(),
Depth: redactedEvent.Depth() + 1,
PrevEvents: []interface{}{redactedEvent.EventID()},
})
room.InsertEvent(t, builderEv)
},
},
{
name: "can not redact others message, missing PL",
additionalEvents: func(t *testing.T, room *test.Room) {
redactedEvent := room.CreateAndInsert(t, bob, "m.room.message", map[string]interface{}{"body": "hello world"})
builderEv := mustCreateEvent(t, fledglingEvent{
Type: spec.MRoomRedaction,
2023-06-07 19:14:35 +02:00
SenderID: charlie.ID,
RoomID: room.ID,
Redacts: redactedEvent.EventID(),
Depth: redactedEvent.Depth() + 1,
PrevEvents: []interface{}{redactedEvent.EventID()},
})
room.InsertEvent(t, builderEv)
},
},
}
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
if err != nil {
t.Fatal(err)
}
natsInstance := &jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
authEvents := []types.EventNID{}
var roomInfo *types.RoomInfo
var err error
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetPublicChat))
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, charlie, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(charlie.ID))
if tc.additionalEvents != nil {
tc.additionalEvents(t, room)
}
for _, ev := range room.Events() {
roomInfo, err = db.GetOrCreateRoomInfo(ctx, ev.PDU)
assert.NoError(t, err)
assert.NotNil(t, roomInfo)
evTypeNID, err := db.GetOrCreateEventTypeNID(ctx, ev.Type())
assert.NoError(t, err)
stateKeyNID, err := db.GetOrCreateEventStateKeyNID(ctx, ev.StateKey())
assert.NoError(t, err)
eventNID, stateAtEvent, err := db.StoreEvent(ctx, ev.PDU, roomInfo, evTypeNID, stateKeyNID, authEvents, false)
assert.NoError(t, err)
if ev.StateKey() != nil {
authEvents = append(authEvents, eventNID)
}
// Calculate the snapshotNID etc.
plResolver := state.NewStateResolution(db, roomInfo, rsAPI)
stateAtEvent.BeforeStateSnapshotNID, err = plResolver.CalculateAndStoreStateBeforeEvent(ctx, ev.PDU, false)
assert.NoError(t, err)
// Update the room
updater, err := db.GetRoomUpdater(ctx, roomInfo)
assert.NoError(t, err)
err = updater.SetState(ctx, eventNID, stateAtEvent.BeforeStateSnapshotNID)
assert.NoError(t, err)
err = updater.Commit()
assert.NoError(t, err)
_, redactedEvent, err := db.MaybeRedactEvent(ctx, roomInfo, eventNID, ev.PDU, &plResolver, &FakeQuerier{})
assert.NoError(t, err)
if redactedEvent != nil {
assert.Equal(t, ev.Redacts(), redactedEvent.EventID())
}
if ev.Type() == spec.MRoomRedaction {
nids, err := db.EventNIDs(ctx, []string{ev.Redacts()})
assert.NoError(t, err)
evs, err := db.Events(ctx, roomInfo.RoomVersion, []types.EventNID{nids[ev.Redacts()].EventNID})
assert.NoError(t, err)
assert.Equal(t, 1, len(evs))
assert.Equal(t, tc.wantRedacted, evs[0].Redacted())
}
}
})
}
})
}
func TestQueryRestrictedJoinAllowed(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
// a room we don't create in the database
allowedByRoomNotExists := test.NewRoom(t, alice)
// a room we create in the database, used for authorisation
allowedByRoomExists := test.NewRoom(t, alice)
allowedByRoomExists.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": spec.Join,
}, test.WithStateKey(bob.ID))
testCases := []struct {
name string
prepareRoomFunc func(t *testing.T) *test.Room
wantResponse string
wantError bool
}{
{
name: "public room unrestricted",
prepareRoomFunc: func(t *testing.T) *test.Room {
return test.NewRoom(t, alice)
},
wantResponse: "",
},
{
name: "room version without restrictions",
prepareRoomFunc: func(t *testing.T) *test.Room {
return test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV7))
},
},
{
name: "restricted only", // bob is not allowed to join
prepareRoomFunc: func(t *testing.T) *test.Room {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV8))
r.CreateAndInsert(t, alice, spec.MRoomJoinRules, map[string]interface{}{
"join_rule": spec.Restricted,
}, test.WithStateKey(""))
return r
},
wantError: true,
},
{
name: "knock_restricted",
prepareRoomFunc: func(t *testing.T) *test.Room {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV8))
r.CreateAndInsert(t, alice, spec.MRoomJoinRules, map[string]interface{}{
"join_rule": spec.KnockRestricted,
}, test.WithStateKey(""))
return r
},
wantError: true,
},
{
name: "restricted with pending invite", // bob should be allowed to join
prepareRoomFunc: func(t *testing.T) *test.Room {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV8))
r.CreateAndInsert(t, alice, spec.MRoomJoinRules, map[string]interface{}{
"join_rule": spec.Restricted,
}, test.WithStateKey(""))
r.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": spec.Invite,
}, test.WithStateKey(bob.ID))
return r
},
wantResponse: "",
},
{
name: "restricted with allowed room_id, but missing room", // bob should not be allowed to join, as we don't know about the room
prepareRoomFunc: func(t *testing.T) *test.Room {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV10))
r.CreateAndInsert(t, alice, spec.MRoomJoinRules, map[string]interface{}{
"join_rule": spec.KnockRestricted,
"allow": []map[string]interface{}{
{
"room_id": allowedByRoomNotExists.ID,
"type": spec.MRoomMembership,
},
},
}, test.WithStateKey(""))
r.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": spec.Join,
"join_authorised_via_users_server": alice.ID,
}, test.WithStateKey(bob.ID))
return r
},
wantError: true,
},
{
name: "restricted with allowed room_id", // bob should be allowed to join, as we know about the room
prepareRoomFunc: func(t *testing.T) *test.Room {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV10))
r.CreateAndInsert(t, alice, spec.MRoomJoinRules, map[string]interface{}{
"join_rule": spec.KnockRestricted,
"allow": []map[string]interface{}{
{
"room_id": allowedByRoomExists.ID,
"type": spec.MRoomMembership,
},
},
}, test.WithStateKey(""))
r.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": spec.Join,
"join_authorised_via_users_server": alice.ID,
}, test.WithStateKey(bob.ID))
return r
},
wantResponse: alice.ID,
},
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
natsInstance := jetstream.NATSInstance{}
defer close()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
if tc.prepareRoomFunc == nil {
t.Fatal("missing prepareRoomFunc")
}
testRoom := tc.prepareRoomFunc(t)
// Create the room
if err := api.SendEvents(processCtx.Context(), rsAPI, api.KindNew, testRoom.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
if err := api.SendEvents(processCtx.Context(), rsAPI, api.KindNew, allowedByRoomExists.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
roomID, _ := spec.NewRoomID(testRoom.ID)
userID, _ := spec.NewUserID(bob.ID, true)
got, err := rsAPI.QueryRestrictedJoinAllowed(processCtx.Context(), *roomID, spec.SenderID(userID.String()))
if tc.wantError && err == nil {
t.Fatal("expected error, got none")
}
if !tc.wantError && err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(tc.wantResponse, got) {
t.Fatalf("unexpected response, want %#v - got %#v", tc.wantResponse, got)
}
})
}
})
}
func TestUpgrade(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := test.NewUser(t)
ctx := context.Background()
spaceChild := test.NewRoom(t, alice)
validateTuples := []gomatrixserverlib.StateKeyTuple{
{EventType: spec.MRoomCreate},
{EventType: spec.MRoomPowerLevels},
{EventType: spec.MRoomJoinRules},
{EventType: spec.MRoomName},
{EventType: spec.MRoomCanonicalAlias},
{EventType: "m.room.tombstone"},
{EventType: "m.custom.event"},
{EventType: "m.space.child", StateKey: spaceChild.ID},
{EventType: "m.custom.event", StateKey: alice.ID},
{EventType: spec.MRoomMember, StateKey: charlie.ID}, // ban should be transferred
}
validate := func(t *testing.T, oldRoomID, newRoomID string, rsAPI api.RoomserverInternalAPI) {
oldRoomState := &api.QueryCurrentStateResponse{}
if err := rsAPI.QueryCurrentState(ctx, &api.QueryCurrentStateRequest{
RoomID: oldRoomID,
StateTuples: validateTuples,
}, oldRoomState); err != nil {
t.Fatal(err)
}
newRoomState := &api.QueryCurrentStateResponse{}
if err := rsAPI.QueryCurrentState(ctx, &api.QueryCurrentStateRequest{
RoomID: newRoomID,
StateTuples: validateTuples,
}, newRoomState); err != nil {
t.Fatal(err)
}
// the old room should have a tombstone event
ev := oldRoomState.StateEvents[gomatrixserverlib.StateKeyTuple{EventType: "m.room.tombstone"}]
replacementRoom := gjson.GetBytes(ev.Content(), "replacement_room").Str
if replacementRoom != newRoomID {
t.Fatalf("tombstone event has replacement_room '%s', expected '%s'", replacementRoom, newRoomID)
}
// the new room should have a predecessor equal to the old room
ev = newRoomState.StateEvents[gomatrixserverlib.StateKeyTuple{EventType: spec.MRoomCreate}]
predecessor := gjson.GetBytes(ev.Content(), "predecessor.room_id").Str
if predecessor != oldRoomID {
t.Fatalf("got predecessor room '%s', expected '%s'", predecessor, oldRoomID)
}
for _, tuple := range validateTuples {
// Skip create and powerlevel event (new room has e.g. predecessor event, old room has restricted powerlevels)
switch tuple.EventType {
case spec.MRoomCreate, spec.MRoomPowerLevels, spec.MRoomCanonicalAlias:
continue
}
oldEv, ok := oldRoomState.StateEvents[tuple]
if !ok {
t.Logf("skipping tuple %#v as it doesn't exist in the old room", tuple)
continue
}
newEv, ok := newRoomState.StateEvents[tuple]
if !ok {
t.Logf("skipping tuple %#v as it doesn't exist in the new room", tuple)
continue
}
if !reflect.DeepEqual(oldEv.Content(), newEv.Content()) {
t.Logf("OldEvent QueryCurrentState: %s", string(oldEv.Content()))
t.Logf("NewEvent QueryCurrentState: %s", string(newEv.Content()))
t.Errorf("event content mismatch")
}
}
}
testCases := []struct {
name string
upgradeUser string
roomFunc func(rsAPI api.RoomserverInternalAPI) string
validateFunc func(t *testing.T, oldRoomID, newRoomID string, rsAPI api.RoomserverInternalAPI)
wantNewRoom bool
}{
{
name: "invalid roomID",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
return "!doesnotexist:test"
},
},
{
name: "powerlevel too low",
upgradeUser: bob.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
room := test.NewRoom(t, alice)
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return room.ID
},
},
{
name: "successful upgrade on new room",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
room := test.NewRoom(t, alice)
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return room.ID
},
wantNewRoom: true,
validateFunc: validate,
},
{
name: "successful upgrade on new room with other state events",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice)
r.CreateAndInsert(t, alice, spec.MRoomName, map[string]interface{}{
"name": "my new name",
}, test.WithStateKey(""))
r.CreateAndInsert(t, alice, spec.MRoomCanonicalAlias, eventutil.CanonicalAliasContent{
Alias: "#myalias:test",
}, test.WithStateKey(""))
// this will be transferred
r.CreateAndInsert(t, alice, "m.custom.event", map[string]interface{}{
"random": "i should exist",
}, test.WithStateKey(""))
// the following will be ignored
r.CreateAndInsert(t, alice, "m.custom.event", map[string]interface{}{
"random": "i will be ignored",
}, test.WithStateKey(alice.ID))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: validate,
},
{
name: "with published room",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice)
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
if err := rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{
RoomID: r.ID,
Visibility: spec.Public,
}); err != nil {
t.Fatal(err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: func(t *testing.T, oldRoomID, newRoomID string, rsAPI api.RoomserverInternalAPI) {
validate(t, oldRoomID, newRoomID, rsAPI)
// check that the new room is published
res := &api.QueryPublishedRoomsResponse{}
if err := rsAPI.QueryPublishedRooms(ctx, &api.QueryPublishedRoomsRequest{RoomID: newRoomID}, res); err != nil {
t.Fatal(err)
}
if len(res.RoomIDs) == 0 {
t.Fatalf("expected room to be published, but wasn't: %#v", res.RoomIDs)
}
},
},
{
name: "with alias",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice)
roomID, err := spec.NewRoomID(r.ID)
if err != nil {
t.Fatal(err)
}
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
if _, err := rsAPI.SetRoomAlias(ctx, spec.SenderID(alice.ID),
*roomID,
"#myroomalias:test"); err != nil {
t.Fatal(err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: func(t *testing.T, oldRoomID, newRoomID string, rsAPI api.RoomserverInternalAPI) {
validate(t, oldRoomID, newRoomID, rsAPI)
// check that the old room has no aliases
res := &api.GetAliasesForRoomIDResponse{}
if err := rsAPI.GetAliasesForRoomID(ctx, &api.GetAliasesForRoomIDRequest{RoomID: oldRoomID}, res); err != nil {
t.Fatal(err)
}
if len(res.Aliases) != 0 {
t.Fatalf("expected old room aliases to be empty, but wasn't: %#v", res.Aliases)
}
// check that the new room has aliases
if err := rsAPI.GetAliasesForRoomID(ctx, &api.GetAliasesForRoomIDRequest{RoomID: newRoomID}, res); err != nil {
t.Fatal(err)
}
if len(res.Aliases) == 0 {
t.Fatalf("expected room aliases to be transferred, but wasn't: %#v", res.Aliases)
}
},
},
{
name: "bans are transferred",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice)
r.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": spec.Ban,
}, test.WithStateKey(charlie.ID))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: validate,
},
{
name: "space childs are transferred",
upgradeUser: alice.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice)
r.CreateAndInsert(t, alice, "m.space.child", map[string]interface{}{}, test.WithStateKey(spaceChild.ID))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: validate,
},
{
name: "custom state is not taken to the new room", // https://github.com/matrix-org/dendrite/issues/2912
upgradeUser: charlie.ID,
roomFunc: func(rsAPI api.RoomserverInternalAPI) string {
r := test.NewRoom(t, alice, test.RoomVersion(gomatrixserverlib.RoomVersionV6))
// Bob and Charlie join
r.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{"membership": spec.Join}, test.WithStateKey(bob.ID))
r.CreateAndInsert(t, charlie, spec.MRoomMember, map[string]interface{}{"membership": spec.Join}, test.WithStateKey(charlie.ID))
// make Charlie an admin so the room can be upgraded
r.CreateAndInsert(t, alice, spec.MRoomPowerLevels, gomatrixserverlib.PowerLevelContent{
Users: map[string]int64{
charlie.ID: 100,
},
}, test.WithStateKey(""))
// Alice creates a custom event
r.CreateAndInsert(t, alice, "m.custom.event", map[string]interface{}{
"random": "data",
}, test.WithStateKey(alice.ID))
r.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{"membership": spec.Leave}, test.WithStateKey(alice.ID))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, r.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
return r.ID
},
wantNewRoom: true,
validateFunc: validate,
},
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
natsInstance := jetstream.NATSInstance{}
defer close()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
rsAPI.SetUserAPI(userAPI)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
if tc.roomFunc == nil {
t.Fatalf("missing roomFunc")
}
if tc.upgradeUser == "" {
tc.upgradeUser = alice.ID
}
roomID := tc.roomFunc(rsAPI)
userID, err := spec.NewUserID(tc.upgradeUser, true)
if err != nil {
t.Fatalf("upgrade userID is invalid")
}
newRoomID, err := rsAPI.PerformRoomUpgrade(processCtx.Context(), roomID, *userID, rsAPI.DefaultRoomVersion())
if err != nil && tc.wantNewRoom {
t.Fatal(err)
}
if tc.wantNewRoom && newRoomID == "" {
t.Fatalf("expected a new room, but the upgrade failed")
}
if !tc.wantNewRoom && newRoomID != "" {
t.Fatalf("expected no new room, but the upgrade succeeded")
}
if tc.validateFunc != nil {
tc.validateFunc(t, roomID, newRoomID, rsAPI)
}
})
}
})
}
func TestStateReset(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := test.NewUser(t)
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
// Prepare APIs
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
natsInstance := jetstream.NATSInstance{}
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
// create a new room
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetPublicChat))
// join with Bob and Charlie
bobJoinEv := room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]any{"membership": "join"}, test.WithStateKey(bob.ID))
charlieJoinEv := room.CreateAndInsert(t, charlie, spec.MRoomMember, map[string]any{"membership": "join"}, test.WithStateKey(charlie.ID))
// Send and create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// send a message
bobMsg := room.CreateAndInsert(t, bob, "m.room.message", map[string]any{"body": "hello world"})
charlieMsg := room.CreateAndInsert(t, charlie, "m.room.message", map[string]any{"body": "hello world"})
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{bobMsg, charlieMsg}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Bob changes his name
expectedDisplayname := "Bob!"
bobDisplayname := room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]any{"membership": "join", "displayname": expectedDisplayname}, test.WithStateKey(bob.ID))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{bobDisplayname}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Change another state event
jrEv := room.CreateAndInsert(t, alice, spec.MRoomJoinRules, gomatrixserverlib.JoinRuleContent{JoinRule: "invite"}, test.WithStateKey(""))
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{jrEv}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// send a message
bobMsg = room.CreateAndInsert(t, bob, "m.room.message", map[string]any{"body": "hello world"})
charlieMsg = room.CreateAndInsert(t, charlie, "m.room.message", map[string]any{"body": "hello world"})
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{bobMsg, charlieMsg}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Craft the state reset message, which is using Bobs initial join event and the
// last message Charlie sent as the prev_events. This should trigger the recalculation
// of the "current" state, since the message event does not have state and no missing events in the DB.
stateResetMsg := mustCreateEvent(t, fledglingEvent{
Type: "m.room.message",
SenderID: charlie.ID,
RoomID: room.ID,
Depth: charlieMsg.Depth() + 1,
PrevEvents: []any{
bobJoinEv.EventID(),
charlieMsg.EventID(),
},
AuthEvents: []any{
room.Events()[0].EventID(), // create event
room.Events()[2].EventID(), // PL event
charlieJoinEv.EventID(), // Charlie join event
},
})
// Send the state reset message
if err := api.SendEvents(ctx, rsAPI, api.KindNew, []*types.HeaderedEvent{stateResetMsg}, "test", "test", "test", nil, false); err != nil {
t.Errorf("failed to send events: %v", err)
}
// Validate that there is a membership event for Bob
bobMembershipEv := api.GetStateEvent(ctx, rsAPI, room.ID, gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomMember,
StateKey: bob.ID,
})
if bobMembershipEv == nil {
t.Fatalf("Membership event for Bob does not exist. State reset?")
} else {
// Validate it's the correct membership event
if dn := gjson.GetBytes(bobMembershipEv.Content(), "displayname").Str; dn != expectedDisplayname {
t.Fatalf("Expected displayname to be %q, got %q", expectedDisplayname, dn)
}
}
})
}
func TestNewServerACLs(t *testing.T) {
alice := test.NewUser(t)
roomWithACL := test.NewRoom(t, alice)
roomWithACL.CreateAndInsert(t, alice, acls.MRoomServerACL, acls.ServerACL{
Allowed: []string{"*"},
Denied: []string{"localhost"},
AllowIPLiterals: false,
}, test.WithStateKey(""))
roomWithoutACL := test.NewRoom(t, alice)
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, closeDB := testrig.CreateConfig(t, dbType)
defer closeDB()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
natsInstance := &jetstream.NATSInstance{}
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
// start JetStream listeners
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
// let the RS create the events
err := api.SendEvents(context.Background(), rsAPI, api.KindNew, roomWithACL.Events(), "test", "test", "test", nil, false)
assert.NoError(t, err)
err = api.SendEvents(context.Background(), rsAPI, api.KindNew, roomWithoutACL.Events(), "test", "test", "test", nil, false)
assert.NoError(t, err)
db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
assert.NoError(t, err)
// create new server ACLs and verify server is banned/not banned
serverACLs := acls.NewServerACLs(db)
banned := serverACLs.IsServerBannedFromRoom("localhost", roomWithACL.ID)
assert.Equal(t, true, banned)
banned = serverACLs.IsServerBannedFromRoom("localhost", roomWithoutACL.ID)
assert.Equal(t, false, banned)
})
}
Use `AckExplicitPolicy` instead of `AckAllPolicy` (#3288) Fixes https://github.com/matrix-org/dendrite/issues/3240 and potentially a root cause for state resets. While testing, I've had added some more debug logging: ``` time="2023-12-16T18:13:11.319458084Z" level=warning msg="already processed event" event_id="$qFYMl_F2vb1N0yxmvlFAMhqhGhLKq4kA-o_YCQKH7tQ" kind=KindNew times=2 time="2023-12-16T18:13:14.537389126Z" level=warning msg="already processed event" event_id="$EU-LTsKErT6Mt1k12-p_3xOHfiLaK6gtwVDlZ35lSuo" kind=KindNew times=5 time="2023-12-16T18:13:16.789551206Z" level=warning msg="already processed event" event_id="$dIPuAfTL5x0VyG873LKPslQeljCSxFT1WKxUtjIMUGE" kind=KindNew times=5 time="2023-12-16T18:13:17.383838767Z" level=warning msg="already processed event" event_id="$7noSZiCkzerpkz_UBO3iatpRnaOiPx-3IXc0GPDQVGE" kind=KindNew times=2 time="2023-12-16T18:13:22.091946597Z" level=warning msg="already processed event" event_id="$3Lvo3Wbi2ol9-nNbQ93N-E2MuGQCJZo5397KkFH-W6E" kind=KindNew times=1 time="2023-12-16T18:13:23.026417446Z" level=warning msg="already processed event" event_id="$lj1xS46zsLBCChhKOLJEG-bu7z-_pq9i_Y2DUIjzGy4" kind=KindNew times=4 ``` So we did receive the same event over and over again. Given they are `KindNew`, we don't short circuit if we already processed them, which potentially caused the state to be calculated with a now wrong state snapshot. Also fixes the back pressure metric. We now correctly increment the counter once we sent the message to NATS and decrement it once we actually processed an event.
2023-12-19 08:25:47 +01:00
// Validate that changing the AckPolicy/AckWait of room consumers
// results in their recreation
func TestRoomConsumerRecreation(t *testing.T) {
alice := test.NewUser(t)
room := test.NewRoom(t, alice)
// As this is DB unrelated, just use SQLite
cfg, processCtx, closeDB := testrig.CreateConfig(t, test.DBTypeSQLite)
defer closeDB()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
natsInstance := &jetstream.NATSInstance{}
// Prepare a stream and consumer using the old configuration
jsCtx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
streamName := cfg.Global.JetStream.Prefixed(jetstream.InputRoomEvent)
consumer := cfg.Global.JetStream.Prefixed("RoomInput" + jetstream.Tokenise(room.ID))
subject := cfg.Global.JetStream.Prefixed(jetstream.InputRoomEventSubj(room.ID))
consumerConfig := &nats.ConsumerConfig{
Durable: consumer,
AckPolicy: nats.AckAllPolicy,
DeliverPolicy: nats.DeliverAllPolicy,
FilterSubject: subject,
AckWait: (time.Minute * 2) + (time.Second * 10),
InactiveThreshold: time.Hour * 24,
}
// Create the consumer with the old config
_, err := jsCtx.AddConsumer(streamName, consumerConfig)
assert.NoError(t, err)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
// start JetStream listeners
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
// let the RS create the events, this also recreates the Consumers
err = api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false)
assert.NoError(t, err)
// Validate that AckPolicy and AckWait has changed
info, err := jsCtx.ConsumerInfo(streamName, consumer)
assert.NoError(t, err)
assert.Equal(t, nats.AckExplicitPolicy, info.Config.AckPolicy)
wantAckWait := input.MaximumMissingProcessingTime + (time.Second * 10)
assert.Equal(t, wantAckWait, info.Config.AckWait)
}
func TestRoomsWithACLs(t *testing.T) {
ctx := context.Background()
alice := test.NewUser(t)
noACLRoom := test.NewRoom(t, alice)
aclRoom := test.NewRoom(t, alice)
aclRoom.CreateAndInsert(t, alice, "m.room.server_acl", map[string]any{
"deny": []string{"evilhost.test"},
"allow": []string{"*"},
}, test.WithStateKey(""))
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, closeDB := testrig.CreateConfig(t, dbType)
defer closeDB()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
natsInstance := &jetstream.NATSInstance{}
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
// start JetStream listeners
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
for _, room := range []*test.Room{noACLRoom, aclRoom} {
// Create the rooms
err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false)
assert.NoError(t, err)
}
// Validate that we only have one ACLd room.
roomsWithACLs, err := rsAPI.RoomsWithACLs(ctx)
assert.NoError(t, err)
assert.Equal(t, []string{aclRoom.ID}, roomsWithACLs)
})
}