Add filtering to /sync and /context

Some cleanup

syncapi/internal/history_visibility.go

@@ -23,6 +23,7 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
+	"github.com/tidwall/gjson"
 )
 
 type HistoryVisibility string
@@ -35,20 +36,26 @@ const (
 	Invited       HistoryVisibility = "invited"
 )
 
+var historyVisibilityPriority = map[HistoryVisibility]uint8{
+	WorldReadable: 0,
+	Shared:        1,
+	Default:       1, // as per the spec, default == shared
+	Invited:       2,
+	Joined:        3,
+}
+
 // EventVisibility contains the history visibility and membership state at a given event
 type EventVisibility struct {
-	Visibility         HistoryVisibility
+	Visibility        HistoryVisibility
-	MembershipAtEvent  string
+	MembershipAtEvent string
-	MembershipCurrent  string
+	MembershipCurrent string
-	MembershipPosition int // the topological position of the membership event
-	HistoryPosition    int // the topological position of the history event
 }
 
 // Visibility is a map from event_id to EvVis, which contains the history visibility and membership for a given user.
 type Visibility map[string]EventVisibility
 
-// Allowed checks the Visibility map if the user is allowed to see the given event.
+// allowed checks the Visibility map if the user is allowed to see the given event.
-func (v Visibility) Allowed(eventID string) (allowed bool) {
+func (v Visibility) allowed(eventID string) (allowed bool) {
 	ev, ok := v[eventID]
 	if !ok {
 		return false
@@ -84,50 +91,70 @@ func (v Visibility) Allowed(eventID string) (allowed bool) {
 	}
 }
 
-// ApplyHistoryVisibilityFilter applies the room history visibility filter on gomatrixserverlib.ClientEvents.
+// ApplyHistoryVisibilityFilter applies the room history visibility filter on gomatrixserverlib.HeaderedEvents.
 // Returns the filtered events and an error, if any.
 func ApplyHistoryVisibilityFilter(
 	ctx context.Context,
 	syncDB storage.Database,
-	clientEvents []gomatrixserverlib.ClientEvent,
+	events []*gomatrixserverlib.HeaderedEvent,
 	userID string,
-) ([]gomatrixserverlib.ClientEvent, error) {
+) ([]*gomatrixserverlib.HeaderedEvent, error) {
-	clientEventsFiltered := []gomatrixserverlib.ClientEvent{}
+	eventsFiltered := make([]*gomatrixserverlib.HeaderedEvent, 0, len(events))
-	stateForEvents, err := getStateForEvents(ctx, syncDB, clientEvents, userID)
+	stateForEvents, err := getStateForEvents(ctx, syncDB, events, userID)
 	if err != nil {
-		return clientEventsFiltered, err
+		return eventsFiltered, err
 	}
-	for _, ev := range clientEvents {
+	for _, ev := range events {
-		if stateForEvents.Allowed(ev.EventID) {
+		// NOTSPEC: Always allow user to see their own membership events (spec contains more "rules")
-			clientEventsFiltered = append(clientEventsFiltered, ev)
+		if _, err = ev.Membership(); err == nil && ev.Sender() == userID {
+			eventsFiltered = append(eventsFiltered, ev)
+			continue
+		}
+		// Handle history visibility changes
+		if hisVis, err := ev.HistoryVisibility(); err == nil {
+			prevHisVis := gjson.GetBytes(ev.Unsigned(), "prev_content.history_visibility").String()
+			if oldPrio, ok := historyVisibilityPriority[HistoryVisibility(prevHisVis)]; ok {
+				// no OK check, since this should have been validated when setting the value
+				newPrio := historyVisibilityPriority[HistoryVisibility(hisVis)]
+				if oldPrio < newPrio {
+					sfe := stateForEvents[ev.EventID()]
+					sfe.Visibility = HistoryVisibility(prevHisVis)
+					stateForEvents[ev.EventID()] = sfe
+				}
+			}
+		}
+		if stateForEvents.allowed(ev.EventID()) {
+			eventsFiltered = append(eventsFiltered, ev)
 		}
 	}
-	return clientEventsFiltered, nil
+	return eventsFiltered, nil
 }
 
 // getStateForEvents returns a Visibility map containing the state before and at the given events.
-func getStateForEvents(ctx context.Context, db storage.Database, events []gomatrixserverlib.ClientEvent, userID string) (Visibility, error) {
+func getStateForEvents(ctx context.Context, db storage.Database, events []*gomatrixserverlib.HeaderedEvent, userID string) (Visibility, error) {
 	result := make(map[string]EventVisibility, len(events))
+	if len(events) == 0 {
+		return result, nil
+	}
 	var (
 		membershipCurrent string
 		err               error
 	)
 	// try to get the current membership of the user
-	if len(events) > 0 {
+	membershipCurrent, _, err = db.SelectMembershipForUser(ctx, events[0].RoomID(), userID, math.MaxInt64)
-		membershipCurrent, _, err = db.SelectMembershipForUser(ctx, events[0].RoomID, userID, math.MaxInt64)
+	if err != nil {
-		if err != nil {
+		return nil, err
-			return nil, err
-		}
 	}
+
 	for _, ev := range events {
 		// get the event topology position
-		pos, err := db.EventPositionInTopology(ctx, ev.EventID)
+		pos, err := db.EventPositionInTopology(ctx, ev.EventID())
 		if err != nil {
 			return nil, fmt.Errorf("initial event does not exist: %w", err)
 		}
 		// By default if no history_visibility is set, or if the value is not understood, the visibility is assumed to be shared
 		var hisVis = "shared"
-		historyEvent, historyPos, err := db.SelectTopologicalEvent(ctx, int(pos.Depth), "m.room.history_visibility", ev.RoomID)
+		historyEvent, _, err := db.SelectTopologicalEvent(ctx, int(pos.Depth), "m.room.history_visibility", ev.RoomID())
 		if err != nil {
 			if err != sql.ErrNoRows {
 				return nil, err
@@ -141,17 +168,15 @@ func getStateForEvents(ctx context.Context, db storage.Database, events []gomatr
 		}
 		// get the membership event
 		var membership string
-		membership, memberPos, err := db.SelectMembershipForUser(ctx, ev.RoomID, userID, int64(pos.Depth))
+		membership, _, err = db.SelectMembershipForUser(ctx, ev.RoomID(), userID, int64(pos.Depth))
 		if err != nil {
 			return nil, err
 		}
 		// finally create the mapping
-		result[ev.EventID] = EventVisibility{
+		result[ev.EventID()] = EventVisibility{
-			Visibility:         HistoryVisibility(hisVis),
+			Visibility:        HistoryVisibility(hisVis),
-			MembershipAtEvent:  membership,
+			MembershipAtEvent: membership,
-			MembershipCurrent:  membershipCurrent,
+			MembershipCurrent: membershipCurrent,
-			MembershipPosition: memberPos,
-			HistoryPosition:    int(historyPos.Depth),
 		}
 	}
 

syncapi/routing/context.go

@@ -24,6 +24,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/internal/caching"
 	roomserver "github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/internal"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -93,24 +94,6 @@ func Context(
 		ContainsURL:             filter.ContainsURL,
 	}
 
-	// TODO: Get the actual state at the last event returned by SelectContextAfterEvent
-	state, _ := syncDB.CurrentState(ctx, roomID, &stateFilter, nil)
-	// verify the user is allowed to see the context for this room/event
-	for _, x := range state {
-		var hisVis string
-		hisVis, err = x.HistoryVisibility()
-		if err != nil {
-			continue
-		}
-		allowed := hisVis == gomatrixserverlib.WorldReadable || membershipRes.Membership == gomatrixserverlib.Join
-		if !allowed {
-			return util.JSONResponse{
-				Code: http.StatusForbidden,
-				JSON: jsonerror.Forbidden("User is not allowed to query context"),
-			}
-		}
-	}
-
 	id, requestedEvent, err := syncDB.SelectContextEvent(ctx, roomID, eventID)
 	if err != nil {
 		if err == sql.ErrNoRows {
@@ -123,6 +106,19 @@ func Context(
 		return jsonerror.InternalServerError()
 	}
 
+	// verify the user is allowed to see the context for this room/event
+	filteredEvent, err := internal.ApplyHistoryVisibilityFilter(ctx, syncDB, []*gomatrixserverlib.HeaderedEvent{&requestedEvent}, device.UserID)
+	if err != nil {
+		logrus.WithError(err).Error("unable to apply history visibility filter")
+		return jsonerror.InternalServerError()
+	}
+	if len(filteredEvent) == 0 {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("User is not allowed to query context"),
+		}
+	}
+
 	eventsBefore, err := syncDB.SelectContextBeforeEvent(ctx, id, roomID, filter)
 	if err != nil && err != sql.ErrNoRows {
 		logrus.WithError(err).Error("unable to fetch before events")
@@ -135,8 +131,26 @@ func Context(
 		return jsonerror.InternalServerError()
 	}
 
-	eventsBeforeClient := gomatrixserverlib.HeaderedToClientEvents(eventsBefore, gomatrixserverlib.FormatAll)
+	eventsBeforeFiltered, err := internal.ApplyHistoryVisibilityFilter(ctx, syncDB, eventsBefore, device.UserID)
-	eventsAfterClient := gomatrixserverlib.HeaderedToClientEvents(eventsAfter, gomatrixserverlib.FormatAll)
+	if err != nil {
+		logrus.WithError(err).Error("unable to apply history visibility filter")
+		return jsonerror.InternalServerError()
+	}
+	eventsAfterFiltered, err := internal.ApplyHistoryVisibilityFilter(ctx, syncDB, eventsAfter, device.UserID)
+	if err != nil {
+		logrus.WithError(err).Error("unable to apply history visibility filter")
+		return jsonerror.InternalServerError()
+	}
+
+	// TODO: Get the actual state at the last event returned by SelectContextAfterEvent
+	state, err := syncDB.CurrentState(ctx, roomID, &stateFilter, nil)
+	if err != nil {
+		logrus.WithError(err).Error("unable to fetch current room state")
+		return jsonerror.InternalServerError()
+	}
+
+	eventsBeforeClient := gomatrixserverlib.HeaderedToClientEvents(eventsBeforeFiltered, gomatrixserverlib.FormatAll)
+	eventsAfterClient := gomatrixserverlib.HeaderedToClientEvents(eventsAfterFiltered, gomatrixserverlib.FormatAll)
 	newState := applyLazyLoadMembers(device, filter, eventsAfterClient, eventsBeforeClient, state, lazyLoadCache)
 
 	response := ContextRespsonse{

syncapi/routing/messages.go

@@ -328,10 +328,10 @@ func (r *messagesReq) retrieveEvents() (
 		return []gomatrixserverlib.ClientEvent{}, *r.from, *r.to, nil
 	}
 
-	// Convert all events into client events and filter them.
+	// Apply room history visibility filter
-	clientEvents, err = internal.ApplyHistoryVisibilityFilter(r.ctx, r.db, gomatrixserverlib.HeaderedToClientEvents(events, gomatrixserverlib.FormatAll), r.device.UserID)
+	filteredEvents, err := internal.ApplyHistoryVisibilityFilter(r.ctx, r.db, events, r.device.UserID)
 
-	return clientEvents, start, end, err
+	return gomatrixserverlib.HeaderedToClientEvents(filteredEvents, gomatrixserverlib.FormatAll), start, end, err
 }
 
 func (r *messagesReq) getStartEnd(events []*gomatrixserverlib.HeaderedEvent) (start, end types.TopologyToken, err error) {

syncapi/streams/stream_pdu.go

@@ -10,9 +10,11 @@ import (
 
 	"github.com/matrix-org/dendrite/internal/caching"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/internal"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
 	"github.com/tidwall/gjson"
 	"go.uber.org/atomic"
 )
@@ -301,8 +303,13 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 			p.addRoomSummary(ctx, jr, delta.RoomID, device.UserID, latestPosition)
 		}
 		jr.Timeline.PrevBatch = &prevBatch
-		jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
+		events, err := internal.ApplyHistoryVisibilityFilter(ctx, p.DB, recentEvents, device.UserID)
-		jr.Timeline.Limited = limited
+		if err != nil {
+			logrus.WithError(err).Error("unable to apply history visibility filter")
+			return r.From, err
+		}
+		jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(events, gomatrixserverlib.FormatSync)
+		jr.Timeline.Limited = limited && len(events) == len(recentEvents)
 		jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(delta.StateEvents, gomatrixserverlib.FormatSync)
 		res.Rooms.Join[delta.RoomID] = *jr
 
@@ -392,33 +399,6 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 		return
 	}
 
-	// TODO FIXME: We don't fully implement history visibility yet. To avoid leaking events which the
-	// user shouldn't see, we check the recent events and remove any prior to the join event of the user
-	// which is equiv to history_visibility: joined
-	joinEventIndex := -1
-	for i := len(recentStreamEvents) - 1; i >= 0; i-- {
-		ev := recentStreamEvents[i]
-		if ev.Type() == gomatrixserverlib.MRoomMember && ev.StateKeyEquals(device.UserID) {
-			membership, _ := ev.Membership()
-			if membership == "join" {
-				joinEventIndex = i
-				if i > 0 {
-					// the create event happens before the first join, so we should cut it at that point instead
-					if recentStreamEvents[i-1].Type() == gomatrixserverlib.MRoomCreate && recentStreamEvents[i-1].StateKeyEquals("") {
-						joinEventIndex = i - 1
-						break
-					}
-				}
-				break
-			}
-		}
-	}
-	if joinEventIndex != -1 {
-		// cut all events earlier than the join (but not the join itself)
-		recentStreamEvents = recentStreamEvents[joinEventIndex:]
-		limited = false // so clients know not to try to backpaginate
-	}
-
 	// Work our way through the timeline events and pick out the event IDs
 	// of any state events that appear in the timeline. We'll specifically
 	// exclude them at the next step, so that we don't get duplicate state
@@ -462,6 +442,12 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	recentEvents := p.DB.StreamEventsToEvents(device, recentStreamEvents)
 	stateEvents = removeDuplicates(stateEvents, recentEvents)
 
+	events, err := internal.ApplyHistoryVisibilityFilter(ctx, p.DB, recentEvents, device.UserID)
+	if err != nil {
+		return nil, err
+	}
+	limited = limited && len(events) == len(recentEvents)
+
 	if stateFilter.LazyLoadMembers {
 		if err != nil {
 			return nil, err
@@ -476,7 +462,8 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	}
 
 	jr.Timeline.PrevBatch = prevBatch
-	jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
+
+	jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(events, gomatrixserverlib.FormatSync)
 	jr.Timeline.Limited = limited
 	jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(stateEvents, gomatrixserverlib.FormatSync)
 	return jr, nil