More efficient server ACLs - hopefully

currentstateserver/acls/acls.go

@@ -0,0 +1,102 @@
+package acls
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+	"regexp"
+	"strings"
+	"sync"
+
+	"github.com/matrix-org/dendrite/currentstateserver/storage"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+)
+
+type ServerACLs struct {
+	acls      map[string]*serverACL // room ID -> ACL
+	aclsMutex sync.RWMutex          // protects the above
+}
+
+func NewServerACLs(db storage.Database) *ServerACLs {
+	ctx := context.TODO()
+	acls := &ServerACLs{
+		acls: make(map[string]*serverACL),
+	}
+	rooms, err := db.GetKnownRooms(ctx)
+	if err != nil {
+		logrus.WithError(err).Errorf("Failed to get known rooms")
+	}
+	for _, room := range rooms {
+		state, err := db.GetStateEvent(ctx, room, "m.room.server_acls", "")
+		if err != nil {
+			logrus.WithError(err).Errorf("Failed to get server ACLs for room %q", room)
+			continue
+		}
+		acls.OnServerACLUpdate(&state.Event)
+	}
+	return acls
+}
+
+type ServerACL struct {
+	Allowed         []string `json:"allow"`
+	Denied          []string `json:"deny"`
+	AllowIPLiterals bool     `json:"allow_ip_literals"`
+}
+
+type serverACL struct {
+	ServerACL
+	allowedRegexes []*regexp.Regexp
+	deniedRegexes  []*regexp.Regexp
+}
+
+func (s *ServerACLs) OnServerACLUpdate(state *gomatrixserverlib.Event) {
+	acls := &serverACL{}
+	if err := json.Unmarshal(state.Content(), &acls.ServerACL); err != nil {
+		return
+	}
+	for _, orig := range acls.Allowed {
+		escaped := regexp.QuoteMeta(orig)
+		escaped = strings.Replace(escaped, "\\?", "(.)", -1)
+		escaped = strings.Replace(escaped, "\\*", "(.*)", -1)
+		if expr, err := regexp.Compile(escaped); err == nil {
+			acls.allowedRegexes = append(acls.allowedRegexes, expr)
+		}
+	}
+	for _, orig := range acls.Denied {
+		escaped := regexp.QuoteMeta(orig)
+		escaped = strings.Replace(escaped, "\\?", "(.)", -1)
+		escaped = strings.Replace(escaped, "\\*", "(.*)", -1)
+		if expr, err := regexp.Compile(escaped); err == nil {
+			acls.deniedRegexes = append(acls.deniedRegexes, expr)
+		}
+	}
+	logrus.Infof("Update server ACLs for %q", state.RoomID())
+	s.aclsMutex.Lock()
+	defer s.aclsMutex.RUnlock()
+	s.acls[state.RoomID()] = acls
+}
+
+func (s *ServerACLs) IsServerBannedFromRoom(serverName gomatrixserverlib.ServerName, roomID string) bool {
+	acls, ok := s.acls[roomID]
+	if !ok {
+		return false
+	}
+	if _, _, err := net.ParseCIDR(fmt.Sprintf("%s/0", serverName)); err == nil {
+		if !acls.AllowIPLiterals {
+			return true
+		}
+	}
+	for _, expr := range acls.deniedRegexes {
+		if expr.MatchString(string(serverName)) {
+			return true
+		}
+	}
+	for _, expr := range acls.allowedRegexes {
+		if expr.MatchString(string(serverName)) {
+			return false
+		}
+	}
+	return true
+}

currentstateserver/api/api.go

@@ -36,6 +36,8 @@ type CurrentStateInternalAPI interface {
 	QuerySharedUsers(ctx context.Context, req *QuerySharedUsersRequest, res *QuerySharedUsersResponse) error
 	// QueryKnownUsers returns a list of users that we know about from our joined rooms.
 	QueryKnownUsers(ctx context.Context, req *QueryKnownUsersRequest, res *QueryKnownUsersResponse) error
+	// QueryServerBannedFromRoom returns whether a server is banned from a room by server ACLs.
+	QueryServerBannedFromRoom(ctx context.Context, req *QueryServerBannedFromRoomRequest, res *QueryServerBannedFromRoomResponse) error
 }
 
 type QuerySharedUsersRequest struct {
@@ -101,6 +103,15 @@ type QueryKnownUsersResponse struct {
 	Users []authtypes.FullyQualifiedProfile `json:"profiles"`
 }
 
+type QueryServerBannedFromRoomRequest struct {
+	ServerName gomatrixserverlib.ServerName `json:"server_name"`
+	RoomID     string                       `json:"room_id"`
+}
+
+type QueryServerBannedFromRoomResponse struct {
+	Banned bool `json:"banned"`
+}
+
 // MarshalJSON stringifies the StateKeyTuple keys so they can be sent over the wire in HTTP API mode.
 func (r *QueryCurrentStateResponse) MarshalJSON() ([]byte, error) {
 	se := make(map[string]*gomatrixserverlib.HeaderedEvent, len(r.StateEvents))

currentstateserver/api/wrapper.go

@@ -2,15 +2,9 @@ package api
 
 import (
 	"context"
-	"encoding/json"
-	"fmt"
-	"net"
-	"regexp"
-	"strings"
 
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
-	"github.com/sirupsen/logrus"
 )
 
 // GetEvent returns the current state event in the room or nil.
@@ -33,70 +27,13 @@ func GetEvent(ctx context.Context, stateAPI CurrentStateInternalAPI, roomID stri
 
 // IsServerBannedFromRoom returns whether the server is banned from a room by server ACLs.
 func IsServerBannedFromRoom(ctx context.Context, stateAPI CurrentStateInternalAPI, roomID string, serverName gomatrixserverlib.ServerName) bool {
-	tuple := gomatrixserverlib.StateKeyTuple{
-		EventType: "m.room.server_acl",
-		StateKey:  "",
-	}
-	req := &QueryCurrentStateRequest{
-		RoomID:      roomID,
-		StateTuples: []gomatrixserverlib.StateKeyTuple{tuple},
-	}
-	res := &QueryCurrentStateResponse{}
-	if err := stateAPI.QueryCurrentState(ctx, req, res); err != nil {
-		logrus.WithError(err).Error("Failed to query current state for server ACL")
+	req := &QueryServerBannedFromRoomRequest{}
+	res := &QueryServerBannedFromRoomResponse{}
+	if err := stateAPI.QueryServerBannedFromRoom(ctx, req, res); err != nil {
+		util.GetLogger(ctx).WithError(err).Error("Failed to QueryServerBannedFromRoom")
 		return true
 	}
-	state, ok := res.StateEvents[tuple]
-	if !ok {
-		return false
-	}
-	acls := struct {
-		Allowed         []string `json:"allow"`
-		Denied          []string `json:"deny"`
-		AllowIPLiterals bool     `json:"allow_ip_literals"`
-	}{}
-	if err := json.Unmarshal(state.Content(), &acls); err != nil {
-		return true
-	}
-	// First, check to see if this is an IP literal.
-	if _, _, err := net.ParseCIDR(fmt.Sprintf("%s/0", serverName)); err == nil {
-		if !acls.AllowIPLiterals {
-			return true
-		}
-	}
-	// Next, build up a list of regular expressions for allowed and denied.
-	allowed := []*regexp.Regexp{}
-	denied := []*regexp.Regexp{}
-	for _, orig := range acls.Allowed {
-		escaped := regexp.QuoteMeta(orig)
-		escaped = strings.Replace(escaped, "\\?", "(.)", -1)
-		escaped = strings.Replace(escaped, "\\*", "(.*)", -1)
-		if expr, err := regexp.Compile(escaped); err == nil {
-			allowed = append(allowed, expr)
-		}
-	}
-	for _, orig := range acls.Denied {
-		escaped := regexp.QuoteMeta(orig)
-		escaped = strings.Replace(escaped, "\\?", "(.)", -1)
-		escaped = strings.Replace(escaped, "\\*", "(.*)", -1)
-		if expr, err := regexp.Compile(escaped); err == nil {
-			denied = append(denied, expr)
-		}
-	}
-	// Now see if we match any of the denied hosts.
-	for _, expr := range denied {
-		if expr.MatchString(string(serverName)) {
-			return true
-		}
-	}
-	// Finally, see if we match any of the allowed hosts.
-	for _, expr := range allowed {
-		if expr.MatchString(string(serverName)) {
-			return false
-		}
-	}
-	// Failing all else deny.
-	return true
+	return res.Banned
 }
 
 // PopulatePublicRooms extracts PublicRoom information for all the provided room IDs. The IDs are not checked to see if they are visible in the

currentstateserver/consumers/roomserver.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 
 	"github.com/Shopify/sarama"
+	"github.com/matrix-org/dendrite/currentstateserver/acls"
 	"github.com/matrix-org/dendrite/currentstateserver/storage"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -30,9 +31,10 @@ import (
 type OutputRoomEventConsumer struct {
 	rsConsumer *internal.ContinualConsumer
 	db         storage.Database
+	acls       *acls.ServerACLs
 }
 
-func NewOutputRoomEventConsumer(topicName string, kafkaConsumer sarama.Consumer, store storage.Database) *OutputRoomEventConsumer {
+func NewOutputRoomEventConsumer(topicName string, kafkaConsumer sarama.Consumer, store storage.Database, acls *acls.ServerACLs) *OutputRoomEventConsumer {
 	consumer := &internal.ContinualConsumer{
 		Topic:          topicName,
 		Consumer:       kafkaConsumer,
@@ -88,6 +90,7 @@ func (c *OutputRoomEventConsumer) onNewRoomEvent(
 		if err != nil {
 			return err
 		}
+		c.acls.OnServerACLUpdate(&addsStateEvents[i].Event)
 	}
 
 	err = c.db.StoreStateEvents(

currentstateserver/currentstateserver.go

@@ -17,6 +17,7 @@ package currentstateserver
 import (
 	"github.com/Shopify/sarama"
 	"github.com/gorilla/mux"
+	"github.com/matrix-org/dendrite/currentstateserver/acls"
 	"github.com/matrix-org/dendrite/currentstateserver/api"
 	"github.com/matrix-org/dendrite/currentstateserver/consumers"
 	"github.com/matrix-org/dendrite/currentstateserver/internal"
@@ -39,13 +40,15 @@ func NewInternalAPI(cfg *config.CurrentStateServer, consumer sarama.Consumer) ap
 	if err != nil {
 		logrus.WithError(err).Panicf("failed to open database")
 	}
+	serverACLs := acls.NewServerACLs(csDB)
 	roomConsumer := consumers.NewOutputRoomEventConsumer(
-		cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent), consumer, csDB,
+		cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent), consumer, csDB, serverACLs,
 	)
 	if err = roomConsumer.Start(); err != nil {
 		logrus.WithError(err).Panicf("failed to start room server consumer")
 	}
 	return &internal.CurrentStateInternalAPI{
-		DB: csDB,
+		DB:         csDB,
+		ServerACLs: serverACLs,
 	}
 }

currentstateserver/internal/api.go

@@ -18,13 +18,15 @@ import (
 	"context"
 
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/currentstateserver/acls"
 	"github.com/matrix-org/dendrite/currentstateserver/api"
 	"github.com/matrix-org/dendrite/currentstateserver/storage"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
 type CurrentStateInternalAPI struct {
-	DB storage.Database
+	DB         storage.Database
+	ServerACLs *acls.ServerACLs
 }
 
 func (a *CurrentStateInternalAPI) QueryCurrentState(ctx context.Context, req *api.QueryCurrentStateRequest, res *api.QueryCurrentStateResponse) error {
@@ -112,3 +114,8 @@ func (a *CurrentStateInternalAPI) QuerySharedUsers(ctx context.Context, req *api
 	res.UserIDsToCount = users
 	return nil
 }
+
+func (a *CurrentStateInternalAPI) QueryServerBannedFromRoom(ctx context.Context, req *api.QueryServerBannedFromRoomRequest, res *api.QueryServerBannedFromRoomResponse) error {
+	res.Banned = a.ServerACLs.IsServerBannedFromRoom(req.ServerName, req.RoomID)
+	return nil
+}

currentstateserver/inthttp/client.go

@@ -26,11 +26,12 @@ import (
 
 // HTTP paths for the internal HTTP APIs
 const (
-	QueryCurrentStatePath     = "/currentstateserver/queryCurrentState"
-	QueryRoomsForUserPath     = "/currentstateserver/queryRoomsForUser"
-	QueryBulkStateContentPath = "/currentstateserver/queryBulkStateContent"
-	QuerySharedUsersPath      = "/currentstateserver/querySharedUsers"
-	QueryKnownUsersPath       = "/currentstateserver/queryKnownUsers"
+	QueryCurrentStatePath         = "/currentstateserver/queryCurrentState"
+	QueryRoomsForUserPath         = "/currentstateserver/queryRoomsForUser"
+	QueryBulkStateContentPath     = "/currentstateserver/queryBulkStateContent"
+	QuerySharedUsersPath          = "/currentstateserver/querySharedUsers"
+	QueryKnownUsersPath           = "/currentstateserver/queryKnownUsers"
+	QueryServerBannedFromRoomPath = "/currentstateserver/queryServerBannedFromRoom"
 )
 
 // NewCurrentStateAPIClient creates a CurrentStateInternalAPI implemented by talking to a HTTP POST API.
@@ -108,3 +109,13 @@ func (h *httpCurrentStateInternalAPI) QueryKnownUsers(
 	apiURL := h.apiURL + QueryKnownUsersPath
 	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, req, res)
 }
+
+func (h *httpCurrentStateInternalAPI) QueryServerBannedFromRoom(
+	ctx context.Context, req *api.QueryServerBannedFromRoomRequest, res *api.QueryServerBannedFromRoomResponse,
+) error {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "QueryKnownUsers")
+	defer span.Finish()
+
+	apiURL := h.apiURL + QueryServerBannedFromRoomPath
+	return httputil.PostJSON(ctx, span, h.httpClient, apiURL, req, res)
+}

currentstateserver/inthttp/server.go

@@ -90,4 +90,17 @@ func AddRoutes(internalAPIMux *mux.Router, intAPI api.CurrentStateInternalAPI) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(QuerySharedUsersPath,
+		httputil.MakeInternalAPI("queryServerBannedFromRoom", func(req *http.Request) util.JSONResponse {
+			request := api.QueryServerBannedFromRoomRequest{}
+			response := api.QueryServerBannedFromRoomResponse{}
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			if err := intAPI.QueryServerBannedFromRoom(req.Context(), &request, &response); err != nil {
+				return util.ErrorResponse(err)
+			}
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 }

currentstateserver/storage/interface.go

@@ -41,4 +41,6 @@ type Database interface {
 	JoinedUsersSetInRooms(ctx context.Context, roomIDs []string) (map[string]int, error)
 	// GetKnownUsers searches all users that userID knows about.
 	GetKnownUsers(ctx context.Context, userID, searchString string, limit int) ([]string, error)
+	// GetKnownRooms returns a list of all rooms we know about.
+	GetKnownRooms(ctx context.Context) ([]string, error)
 }

currentstateserver/storage/postgres/current_room_state_table.go

@@ -82,6 +82,9 @@ const selectJoinedUsersSetForRoomsSQL = "" +
 	"SELECT state_key, COUNT(room_id) FROM currentstate_current_room_state WHERE room_id = ANY($1) AND" +
 	" type = 'm.room.member' and content_value = 'join' GROUP BY state_key"
 
+const selectKnownRoomsSQL = "" +
+	"SELECT DISTINCT room_id FROM currentsate_current_room_state"
+
 // selectKnownUsersSQL uses a sub-select statement here to find rooms that the user is
 // joined to. Since this information is used to populate the user directory, we will
 // only return users that the user would ordinarily be able to see anyway.
@@ -99,6 +102,7 @@ type currentRoomStateStatements struct {
 	selectBulkStateContentStmt       *sql.Stmt
 	selectBulkStateContentWildStmt   *sql.Stmt
 	selectJoinedUsersSetForRoomsStmt *sql.Stmt
+	selectKnownRoomsStmt             *sql.Stmt
 	selectKnownUsersStmt             *sql.Stmt
 }
 
@@ -132,6 +136,9 @@ func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, erro
 	if s.selectJoinedUsersSetForRoomsStmt, err = db.Prepare(selectJoinedUsersSetForRoomsSQL); err != nil {
 		return nil, err
 	}
+	if s.selectKnownRoomsStmt, err = db.Prepare(selectKnownRoomsSQL); err != nil {
+		return nil, err
+	}
 	if s.selectKnownUsersStmt, err = db.Prepare(selectKnownUsersSQL); err != nil {
 		return nil, err
 	}
@@ -325,3 +332,20 @@ func (s *currentRoomStateStatements) SelectKnownUsers(ctx context.Context, userI
 	}
 	return result, rows.Err()
 }
+
+func (s *currentRoomStateStatements) SelectKnownRooms(ctx context.Context) ([]string, error) {
+	rows, err := s.selectKnownUsersStmt.QueryContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	result := []string{}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectKnownRooms: rows.close() failed")
+	for rows.Next() {
+		var roomID string
+		if err := rows.Scan(&roomID); err != nil {
+			return nil, err
+		}
+		result = append(result, roomID)
+	}
+	return result, rows.Err()
+}

currentstateserver/storage/shared/storage.go

@@ -93,3 +93,7 @@ func (d *Database) JoinedUsersSetInRooms(ctx context.Context, roomIDs []string)
 func (d *Database) GetKnownUsers(ctx context.Context, userID, searchString string, limit int) ([]string, error) {
 	return d.CurrentRoomState.SelectKnownUsers(ctx, userID, searchString, limit)
 }
+
+func (d *Database) GetKnownRooms(ctx context.Context) ([]string, error) {
+	return d.CurrentRoomState.SelectKnownRooms(ctx)
+}

currentstateserver/storage/sqlite3/current_room_state_table.go

@@ -70,6 +70,9 @@ const selectBulkStateContentWildSQL = "" +
 const selectJoinedUsersSetForRoomsSQL = "" +
 	"SELECT state_key, COUNT(room_id) FROM currentstate_current_room_state WHERE room_id IN ($1) AND type = 'm.room.member' and content_value = 'join' GROUP BY state_key"
 
+const selectKnownRoomsSQL = "" +
+	"SELECT DISTINCT room_id FROM currentsate_current_room_state"
+
 // selectKnownUsersSQL uses a sub-select statement here to find rooms that the user is
 // joined to. Since this information is used to populate the user directory, we will
 // only return users that the user would ordinarily be able to see anyway.
@@ -86,6 +89,7 @@ type currentRoomStateStatements struct {
 	selectRoomIDsWithMembershipStmt  *sql.Stmt
 	selectStateEventStmt             *sql.Stmt
 	selectJoinedUsersSetForRoomsStmt *sql.Stmt
+	selectKnownRoomsStmt             *sql.Stmt
 	selectKnownUsersStmt             *sql.Stmt
 }
 
@@ -113,6 +117,9 @@ func NewSqliteCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, error)
 	if s.selectJoinedUsersSetForRoomsStmt, err = db.Prepare(selectJoinedUsersSetForRoomsSQL); err != nil {
 		return nil, err
 	}
+	if s.selectKnownRoomsStmt, err = db.Prepare(selectKnownRoomsSQL); err != nil {
+		return nil, err
+	}
 	if s.selectKnownUsersStmt, err = db.Prepare(selectKnownUsersSQL); err != nil {
 		return nil, err
 	}
@@ -345,3 +352,20 @@ func (s *currentRoomStateStatements) SelectKnownUsers(ctx context.Context, userI
 	}
 	return result, rows.Err()
 }
+
+func (s *currentRoomStateStatements) SelectKnownRooms(ctx context.Context) ([]string, error) {
+	rows, err := s.selectKnownUsersStmt.QueryContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	result := []string{}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectKnownRooms: rows.close() failed")
+	for rows.Next() {
+		var roomID string
+		if err := rows.Scan(&roomID); err != nil {
+			return nil, err
+		}
+		result = append(result, roomID)
+	}
+	return result, rows.Err()
+}

currentstateserver/storage/tables/interface.go

@@ -41,6 +41,8 @@ type CurrentRoomState interface {
 	SelectJoinedUsersSetForRooms(ctx context.Context, roomIDs []string) (map[string]int, error)
 	// SelectKnownUsers searches all users that userID knows about.
 	SelectKnownUsers(ctx context.Context, userID, searchString string, limit int) ([]string, error)
+	// SelectKnownRooms returns all rooms that we know about.
+	SelectKnownRooms(ctx context.Context) ([]string, error)
 }
 
 // StrippedEvent represents a stripped event for returning extracted content values.