From b8f91485b47ac6e92a90988b394e8f3611735250 Mon Sep 17 00:00:00 2001 From: Till <2353100+S7evinK@users.noreply.github.com> Date: Wed, 22 Nov 2023 15:38:04 +0100 Subject: [PATCH] Update ACLs when received as outliers (#3008) This should fix #3004 by making sure we also update our in-memory ACLs after joining a new room. Also makes use of more caching in `GetStateEvent` Bonus: Adds some tests, as I was about to use `GetBulkStateContent`, but turns out that `GetStateEvent` is basically doing the same, just that it only gets the `eventTypeNID`/`eventStateKeyNID` once and not for every call. --- federationapi/federationapi.go | 2 +- .../internal/federationclient_test.go | 10 +-- federationapi/internal/perform_test.go | 10 +-- federationapi/queue/destinationqueue.go | 2 - federationapi/queue/queue.go | 36 --------- federationapi/queue/queue_test.go | 13 +--- roomserver/acls/acls.go | 4 +- roomserver/internal/input/input_events.go | 22 ++++++ roomserver/producers/roomevent.go | 2 +- roomserver/roomserver_test.go | 41 ++++++++++ roomserver/storage/tables/interface_test.go | 76 +++++++++++++++++++ 11 files changed, 155 insertions(+), 63 deletions(-) create mode 100644 roomserver/storage/tables/interface_test.go diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go index e2524f66a..efbfa3315 100644 --- a/federationapi/federationapi.go +++ b/federationapi/federationapi.go @@ -125,7 +125,7 @@ func NewInternalAPI( queues := queue.NewOutgoingQueues( federationDB, processContext, cfg.Matrix.DisableFederation, - cfg.Matrix.ServerName, federation, rsAPI, &stats, + cfg.Matrix.ServerName, federation, &stats, signingInfo, ) diff --git a/federationapi/internal/federationclient_test.go b/federationapi/internal/federationclient_test.go index 8c562dd61..fe8d84ffb 100644 --- a/federationapi/internal/federationclient_test.go +++ b/federationapi/internal/federationclient_test.go @@ -65,7 +65,7 @@ func TestFederationClientQueryKeys(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedapi := FederationInternalAPI{ @@ -96,7 +96,7 @@ func TestFederationClientQueryKeysBlacklisted(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedapi := FederationInternalAPI{ @@ -126,7 +126,7 @@ func TestFederationClientQueryKeysFailure(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedapi := FederationInternalAPI{ @@ -156,7 +156,7 @@ func TestFederationClientClaimKeys(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedapi := FederationInternalAPI{ @@ -187,7 +187,7 @@ func TestFederationClientClaimKeysBlacklisted(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedapi := FederationInternalAPI{ diff --git a/federationapi/internal/perform_test.go b/federationapi/internal/perform_test.go index 656755f96..2795a018a 100644 --- a/federationapi/internal/perform_test.go +++ b/federationapi/internal/perform_test.go @@ -70,7 +70,7 @@ func TestPerformWakeupServers(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedAPI := NewFederationInternalAPI( @@ -116,7 +116,7 @@ func TestQueryRelayServers(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedAPI := NewFederationInternalAPI( @@ -157,7 +157,7 @@ func TestRemoveRelayServers(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedAPI := NewFederationInternalAPI( @@ -197,7 +197,7 @@ func TestPerformDirectoryLookup(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedAPI := NewFederationInternalAPI( @@ -236,7 +236,7 @@ func TestPerformDirectoryLookupRelaying(t *testing.T) { queues := queue.NewOutgoingQueues( testDB, process.NewProcessContext(), false, - cfg.Matrix.ServerName, fedClient, nil, &stats, + cfg.Matrix.ServerName, fedClient, &stats, nil, ) fedAPI := NewFederationInternalAPI( diff --git a/federationapi/queue/destinationqueue.go b/federationapi/queue/destinationqueue.go index 880aee0d3..f51e849fa 100644 --- a/federationapi/queue/destinationqueue.go +++ b/federationapi/queue/destinationqueue.go @@ -31,7 +31,6 @@ import ( "github.com/matrix-org/dendrite/federationapi/statistics" "github.com/matrix-org/dendrite/federationapi/storage" "github.com/matrix-org/dendrite/federationapi/storage/shared/receipt" - "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/dendrite/setup/process" ) @@ -53,7 +52,6 @@ type destinationQueue struct { db storage.Database process *process.ProcessContext signing map[spec.ServerName]*fclient.SigningIdentity - rsAPI api.FederationRoomserverAPI client fclient.FederationClient // federation client origin spec.ServerName // origin of requests destination spec.ServerName // destination of requests diff --git a/federationapi/queue/queue.go b/federationapi/queue/queue.go index 24b3efd2d..892c26a2c 100644 --- a/federationapi/queue/queue.go +++ b/federationapi/queue/queue.go @@ -27,12 +27,10 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus" - "github.com/tidwall/gjson" "github.com/matrix-org/dendrite/federationapi/statistics" "github.com/matrix-org/dendrite/federationapi/storage" "github.com/matrix-org/dendrite/federationapi/storage/shared/receipt" - "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/dendrite/setup/process" ) @@ -43,7 +41,6 @@ type OutgoingQueues struct { db storage.Database process *process.ProcessContext disabled bool - rsAPI api.FederationRoomserverAPI origin spec.ServerName client fclient.FederationClient statistics *statistics.Statistics @@ -90,7 +87,6 @@ func NewOutgoingQueues( disabled bool, origin spec.ServerName, client fclient.FederationClient, - rsAPI api.FederationRoomserverAPI, statistics *statistics.Statistics, signing []*fclient.SigningIdentity, ) *OutgoingQueues { @@ -98,7 +94,6 @@ func NewOutgoingQueues( disabled: disabled, process: process, db: db, - rsAPI: rsAPI, origin: origin, client: client, statistics: statistics, @@ -162,7 +157,6 @@ func (oqs *OutgoingQueues) getQueue(destination spec.ServerName) *destinationQue queues: oqs, db: oqs.db, process: oqs.process, - rsAPI: oqs.rsAPI, origin: oqs.origin, destination: destination, client: oqs.client, @@ -213,18 +207,6 @@ func (oqs *OutgoingQueues) SendEvent( delete(destmap, local) } - // Check if any of the destinations are prohibited by server ACLs. - for destination := range destmap { - if api.IsServerBannedFromRoom( - oqs.process.Context(), - oqs.rsAPI, - ev.RoomID().String(), - destination, - ) { - delete(destmap, destination) - } - } - // If there are no remaining destinations then give up. if len(destmap) == 0 { return nil @@ -303,24 +285,6 @@ func (oqs *OutgoingQueues) SendEDU( delete(destmap, local) } - // There is absolutely no guarantee that the EDU will have a room_id - // field, as it is not required by the spec. However, if it *does* - // (e.g. typing notifications) then we should try to make sure we don't - // bother sending them to servers that are prohibited by the server - // ACLs. - if result := gjson.GetBytes(e.Content, "room_id"); result.Exists() { - for destination := range destmap { - if api.IsServerBannedFromRoom( - oqs.process.Context(), - oqs.rsAPI, - result.Str, - destination, - ) { - delete(destmap, destination) - } - } - } - // If there are no remaining destinations then give up. if len(destmap) == 0 { return nil diff --git a/federationapi/queue/queue_test.go b/federationapi/queue/queue_test.go index e75615e05..73d3b0598 100644 --- a/federationapi/queue/queue_test.go +++ b/federationapi/queue/queue_test.go @@ -34,7 +34,6 @@ import ( "github.com/matrix-org/dendrite/federationapi/statistics" "github.com/matrix-org/dendrite/federationapi/storage" - rsapi "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/process" @@ -65,15 +64,6 @@ func mustCreateFederationDatabase(t *testing.T, dbType test.DBType, realDatabase } } -type stubFederationRoomServerAPI struct { - rsapi.FederationRoomserverAPI -} - -func (r *stubFederationRoomServerAPI) QueryServerBannedFromRoom(ctx context.Context, req *rsapi.QueryServerBannedFromRoomRequest, res *rsapi.QueryServerBannedFromRoomResponse) error { - res.Banned = false - return nil -} - type stubFederationClient struct { fclient.FederationClient shouldTxSucceed bool @@ -126,7 +116,6 @@ func testSetup(failuresUntilBlacklist uint32, failuresUntilAssumedOffline uint32 txCount: *atomic.NewUint32(0), txRelayCount: *atomic.NewUint32(0), } - rs := &stubFederationRoomServerAPI{} stats := statistics.NewStatistics(db, failuresUntilBlacklist, failuresUntilAssumedOffline) signingInfo := []*fclient.SigningIdentity{ @@ -136,7 +125,7 @@ func testSetup(failuresUntilBlacklist uint32, failuresUntilAssumedOffline uint32 ServerName: "localhost", }, } - queues := NewOutgoingQueues(db, processContext, false, "localhost", fc, rs, &stats, signingInfo) + queues := NewOutgoingQueues(db, processContext, false, "localhost", fc, &stats, signingInfo) return db, fc, queues, processContext, close } diff --git a/roomserver/acls/acls.go b/roomserver/acls/acls.go index 601ce9063..e247c7553 100644 --- a/roomserver/acls/acls.go +++ b/roomserver/acls/acls.go @@ -29,6 +29,8 @@ import ( "github.com/sirupsen/logrus" ) +const MRoomServerACL = "m.room.server_acl" + type ServerACLDatabase interface { // GetKnownRooms returns a list of all rooms we know about. GetKnownRooms(ctx context.Context) ([]string, error) @@ -57,7 +59,7 @@ func NewServerACLs(db ServerACLDatabase) *ServerACLs { // do then we'll process it into memory so that we have the regexes to // hand. for _, room := range rooms { - state, err := db.GetStateEvent(ctx, room, "m.room.server_acl", "") + state, err := db.GetStateEvent(ctx, room, MRoomServerACL, "") if err != nil { logrus.WithError(err).Errorf("Failed to get server ACLs for room %q", room) continue diff --git a/roomserver/internal/input/input_events.go b/roomserver/internal/input/input_events.go index 77b50d0e2..520f82a80 100644 --- a/roomserver/internal/input/input_events.go +++ b/roomserver/internal/input/input_events.go @@ -33,6 +33,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" + "github.com/matrix-org/dendrite/roomserver/acls" "github.com/matrix-org/dendrite/roomserver/internal/helpers" userAPI "github.com/matrix-org/dendrite/userapi/api" @@ -491,6 +492,27 @@ func (r *Inputer) processRoomEvent( } } + // If this is a membership event, it is possible we newly joined a federated room and eventually + // missed to update our m.room.server_acl - the following ensures we set the ACLs + // TODO: This probably performs badly in benchmarks + if event.Type() == spec.MRoomMember { + membership, _ := event.Membership() + if membership == spec.Join { + _, serverName, _ := gomatrixserverlib.SplitID('@', *event.StateKey()) + // only handle local membership events + if r.Cfg.Matrix.IsLocalServerName(serverName) { + var aclEvent *types.HeaderedEvent + aclEvent, err = r.DB.GetStateEvent(ctx, event.RoomID().String(), acls.MRoomServerACL, "") + if err != nil { + logrus.WithError(err).Error("failed to get server ACLs") + } + if aclEvent != nil { + r.ACLs.OnServerACLUpdate(aclEvent) + } + } + } + } + // Handle remote room upgrades, e.g. remove published room if event.Type() == "m.room.tombstone" && event.StateKeyEquals("") && !r.Cfg.Matrix.IsLocalServerName(senderDomain) { if err = r.handleRemoteRoomUpgrade(ctx, event); err != nil { diff --git a/roomserver/producers/roomevent.go b/roomserver/producers/roomevent.go index 165304d49..af7e10580 100644 --- a/roomserver/producers/roomevent.go +++ b/roomserver/producers/roomevent.go @@ -73,7 +73,7 @@ func (r *RoomEventProducer) ProduceRoomEvents(roomID string, updates []api.Outpu } } - if eventType == "m.room.server_acl" && update.NewRoomEvent.Event.StateKeyEquals("") { + if eventType == acls.MRoomServerACL && update.NewRoomEvent.Event.StateKeyEquals("") { ev := update.NewRoomEvent.Event.PDU defer r.ACLs.OnServerACLUpdate(ev) } diff --git a/roomserver/roomserver_test.go b/roomserver/roomserver_test.go index 218a0d8a9..e9cd926d7 100644 --- a/roomserver/roomserver_test.go +++ b/roomserver/roomserver_test.go @@ -16,6 +16,7 @@ import ( "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" @@ -1190,3 +1191,43 @@ func TestStateReset(t *testing.T) { } }) } + +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) + }) +} diff --git a/roomserver/storage/tables/interface_test.go b/roomserver/storage/tables/interface_test.go new file mode 100644 index 000000000..8727e2436 --- /dev/null +++ b/roomserver/storage/tables/interface_test.go @@ -0,0 +1,76 @@ +package tables + +import ( + "testing" + + "github.com/matrix-org/dendrite/roomserver/types" + "github.com/matrix-org/dendrite/test" + "github.com/matrix-org/gomatrixserverlib/spec" + "github.com/stretchr/testify/assert" +) + +func TestExtractContentValue(t *testing.T) { + alice := test.NewUser(t) + room := test.NewRoom(t, alice) + + tests := []struct { + name string + event *types.HeaderedEvent + want string + }{ + { + name: "returns creator ID for create events", + event: room.Events()[0], + want: alice.ID, + }, + { + name: "returns the alias for canonical alias events", + event: room.CreateEvent(t, alice, spec.MRoomCanonicalAlias, map[string]string{"alias": "#test:test"}), + want: "#test:test", + }, + { + name: "returns the history_visibility for history visibility events", + event: room.CreateEvent(t, alice, spec.MRoomHistoryVisibility, map[string]string{"history_visibility": "shared"}), + want: "shared", + }, + { + name: "returns the join rules for join_rules events", + event: room.CreateEvent(t, alice, spec.MRoomJoinRules, map[string]string{"join_rule": "public"}), + want: "public", + }, + { + name: "returns the membership for room_member events", + event: room.CreateEvent(t, alice, spec.MRoomMember, map[string]string{"membership": "join"}, test.WithStateKey(alice.ID)), + want: "join", + }, + { + name: "returns the room name for room_name events", + event: room.CreateEvent(t, alice, spec.MRoomName, map[string]string{"name": "testing"}, test.WithStateKey(alice.ID)), + want: "testing", + }, + { + name: "returns the room avatar for avatar events", + event: room.CreateEvent(t, alice, spec.MRoomAvatar, map[string]string{"url": "mxc://testing"}, test.WithStateKey(alice.ID)), + want: "mxc://testing", + }, + { + name: "returns the room topic for topic events", + event: room.CreateEvent(t, alice, spec.MRoomTopic, map[string]string{"topic": "testing"}, test.WithStateKey(alice.ID)), + want: "testing", + }, + { + name: "returns guest_access for guest access events", + event: room.CreateEvent(t, alice, "m.room.guest_access", map[string]string{"guest_access": "forbidden"}, test.WithStateKey(alice.ID)), + want: "forbidden", + }, + { + name: "returns empty string if key can't be found or unknown event", + event: room.CreateEvent(t, alice, "idontexist", nil), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equalf(t, tt.want, ExtractContentValue(tt.event), "ExtractContentValue(%v)", tt.event) + }) + } +}