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

@@ -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,
 	)
 

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{

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(

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

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

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
 }

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

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 {

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)
 			}

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)
+	})
+}

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)
+		})
+	}
+}