Peeking updates (#1607)

* Add unpeek

* Don't allow peeks into encrypted rooms

* Fix send tests

* Update consumers

9 files changed, 200 insertions, 2 deletions

diff --git a/roomserver/api/api.go b/roomserver/api/api.go
index 2683918a..bef2bb3f 100644
--- a/roomserver/api/api.go
+++ b/roomserver/api/api.go
@@ -42,6 +42,12 @@ type RoomserverInternalAPI interface {
 		res *PerformPeekResponse,
 	)
 
+	PerformUnpeek(
+		ctx context.Context,
+		req *PerformUnpeekRequest,
+		res *PerformUnpeekResponse,
+	)
+
 	PerformPublish(
 		ctx context.Context,
 		req *PerformPublishRequest,
diff --git a/roomserver/api/api_trace.go b/roomserver/api/api_trace.go
index e625fb04..eb2b2e1d 100644
--- a/roomserver/api/api_trace.go
+++ b/roomserver/api/api_trace.go
@@ -46,6 +46,15 @@ func (t *RoomserverInternalAPITrace) PerformPeek(
 	util.GetLogger(ctx).Infof("PerformPeek req=%+v res=%+v", js(req), js(res))
 }
 
+func (t *RoomserverInternalAPITrace) PerformUnpeek(
+	ctx context.Context,
+	req *PerformUnpeekRequest,
+	res *PerformUnpeekResponse,
+) {
+	t.Impl.PerformUnpeek(ctx, req, res)
+	util.GetLogger(ctx).Infof("PerformUnpeek req=%+v res=%+v", js(req), js(res))
+}
+
 func (t *RoomserverInternalAPITrace) PerformJoin(
 	ctx context.Context,
 	req *PerformJoinRequest,
diff --git a/roomserver/api/output.go b/roomserver/api/output.go
index fb512a93..2993813c 100644
--- a/roomserver/api/output.go
+++ b/roomserver/api/output.go
@@ -51,6 +51,8 @@ const (
 
 	// OutputTypeNewPeek indicates that the kafka event is an OutputNewPeek
 	OutputTypeNewPeek OutputType = "new_peek"
+	// OutputTypeRetirePeek indicates that the kafka event is an OutputRetirePeek
+	OutputTypeRetirePeek OutputType = "retire_peek"
 )
 
 // An OutputEvent is an entry in the roomserver output kafka log.
@@ -70,6 +72,8 @@ type OutputEvent struct {
 	RedactedEvent *OutputRedactedEvent `json:"redacted_event,omitempty"`
 	// The content of event with type OutputTypeNewPeek
 	NewPeek *OutputNewPeek `json:"new_peek,omitempty"`
+	// The content of event with type OutputTypeRetirePeek
+	RetirePeek *OutputRetirePeek `json:"retire_peek,omitempty"`
 }
 
 // Type of the OutputNewRoomEvent.
@@ -240,3 +244,10 @@ type OutputNewPeek struct {
 	UserID   string
 	DeviceID string
 }
+
+// An OutputRetirePeek is written whenever a user stops peeking into a room.
+type OutputRetirePeek struct {
+	RoomID   string
+	UserID   string
+	DeviceID string
+}
diff --git a/roomserver/api/perform.go b/roomserver/api/perform.go
index ec561f11..ae2d6d97 100644
--- a/roomserver/api/perform.go
+++ b/roomserver/api/perform.go
@@ -123,6 +123,17 @@ type PerformPeekResponse struct {
 	Error *PerformError
 }
 
+type PerformUnpeekRequest struct {
+	RoomID   string `json:"room_id"`
+	UserID   string `json:"user_id"`
+	DeviceID string `json:"device_id"`
+}
+
+type PerformUnpeekResponse struct {
+	// If non-nil, the join request failed. Contains more information why it failed.
+	Error *PerformError
+}
+
 // PerformBackfillRequest is a request to PerformBackfill.
 type PerformBackfillRequest struct {
 	// The room to backfill
diff --git a/roomserver/internal/api.go b/roomserver/internal/api.go
index c825c13d..1ad971ec 100644
--- a/roomserver/internal/api.go
+++ b/roomserver/internal/api.go
@@ -23,6 +23,7 @@ type RoomserverInternalAPI struct {
 	*perform.Inviter
 	*perform.Joiner
 	*perform.Peeker
+	*perform.Unpeeker
 	*perform.Leaver
 	*perform.Publisher
 	*perform.Backfiller
@@ -94,6 +95,13 @@ func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSen
 		FSAPI:      r.fsAPI,
 		Inputer:    r.Inputer,
 	}
+	r.Unpeeker = &perform.Unpeeker{
+		ServerName: r.Cfg.Matrix.ServerName,
+		Cfg:        r.Cfg,
+		DB:         r.DB,
+		FSAPI:      r.fsAPI,
+		Inputer:    r.Inputer,
+	}
 	r.Leaver = &perform.Leaver{
 		Cfg:     r.Cfg,
 		DB:      r.DB,
diff --git a/roomserver/internal/perform/perform_peek.go b/roomserver/internal/perform/perform_peek.go
index 66d1bdb2..2f4694c8 100644
--- a/roomserver/internal/perform/perform_peek.go
+++ b/roomserver/internal/perform/perform_peek.go
@@ -163,8 +163,7 @@ func (r *Peeker) performPeekRoomByID(
 	// XXX: we should probably factor out history_visibility checks into a common utility method somewhere
 	// which handles the default value etc.
 	var worldReadable = false
-	ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.history_visibility", "")
-	if ev != nil {
+	if ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.history_visibility", ""); ev != nil {
 		content := map[string]string{}
 		if err = json.Unmarshal(ev.Content(), &content); err != nil {
 			util.GetLogger(ctx).WithError(err).Error("json.Unmarshal for history visibility failed")
@@ -182,6 +181,13 @@ func (r *Peeker) performPeekRoomByID(
 		}
 	}
 
+	if ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.encryption", ""); ev != nil {
+		return "", &api.PerformError{
+			Code: api.PerformErrorNotAllowed,
+			Msg:  "Cannot peek into an encrypted room",
+		}
+	}
+
 	// TODO: handle federated peeks
 
 	err = r.Inputer.WriteOutputEvents(roomID, []api.OutputEvent{
diff --git a/roomserver/internal/perform/perform_unpeek.go b/roomserver/internal/perform/perform_unpeek.go
new file mode 100644
index 00000000..f71e0007
--- /dev/null
+++ b/roomserver/internal/perform/perform_unpeek.go
@@ -0,0 +1,118 @@
+// Copyright 2020 New Vector Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package perform
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	fsAPI "github.com/matrix-org/dendrite/federationsender/api"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/roomserver/internal/input"
+	"github.com/matrix-org/dendrite/roomserver/storage"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+type Unpeeker struct {
+	ServerName gomatrixserverlib.ServerName
+	Cfg        *config.RoomServer
+	FSAPI      fsAPI.FederationSenderInternalAPI
+	DB         storage.Database
+
+	Inputer *input.Inputer
+}
+
+// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationsender.
+func (r *Unpeeker) PerformUnpeek(
+	ctx context.Context,
+	req *api.PerformUnpeekRequest,
+	res *api.PerformUnpeekResponse,
+) {
+	if err := r.performUnpeek(ctx, req); err != nil {
+		perr, ok := err.(*api.PerformError)
+		if ok {
+			res.Error = perr
+		} else {
+			res.Error = &api.PerformError{
+				Msg: err.Error(),
+			}
+		}
+	}
+}
+
+func (r *Unpeeker) performUnpeek(
+	ctx context.Context,
+	req *api.PerformUnpeekRequest,
+) error {
+	// FIXME: there's way too much duplication with performJoin
+	_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
+	if err != nil {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("Supplied user ID %q in incorrect format", req.UserID),
+		}
+	}
+	if domain != r.Cfg.Matrix.ServerName {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("User %q does not belong to this homeserver", req.UserID),
+		}
+	}
+	if strings.HasPrefix(req.RoomID, "!") {
+		return r.performUnpeekRoomByID(ctx, req)
+	}
+	return &api.PerformError{
+		Code: api.PerformErrorBadRequest,
+		Msg:  fmt.Sprintf("Room ID %q is invalid", req.RoomID),
+	}
+}
+
+func (r *Unpeeker) performUnpeekRoomByID(
+	_ context.Context,
+	req *api.PerformUnpeekRequest,
+) (err error) {
+	// Get the domain part of the room ID.
+	_, _, err = gomatrixserverlib.SplitID('!', req.RoomID)
+	if err != nil {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("Room ID %q is invalid: %s", req.RoomID, err),
+		}
+	}
+
+	// TODO: handle federated peeks
+
+	err = r.Inputer.WriteOutputEvents(req.RoomID, []api.OutputEvent{
+		{
+			Type: api.OutputTypeRetirePeek,
+			RetirePeek: &api.OutputRetirePeek{
+				RoomID:   req.RoomID,
+				UserID:   req.UserID,
+				DeviceID: req.DeviceID,
+			},
+		},
+	})
+	if err != nil {
+		return
+	}
+
+	// By this point, if req.RoomIDOrAlias contained an alias, then
+	// it will have been overwritten with a room ID by performPeekRoomByAlias.
+	// We should now include this in the response so that the CS API can
+	// return the right room ID.
+	return nil
+}
diff --git a/roomserver/inthttp/client.go b/roomserver/inthttp/client.go
index f5b66ca6..e496b81e 100644
--- a/roomserver/inthttp/client.go
+++ b/roomserver/inthttp/client.go
@@ -27,6 +27,7 @@ const (
 	// Perform operations
 	RoomserverPerformInvitePath   = "/roomserver/performInvite"
 	RoomserverPerformPeekPath     = "/roomserver/performPeek"
+	RoomserverPerformUnpeekPath   = "/roomserver/performUnpeek"
 	RoomserverPerformJoinPath     = "/roomserver/performJoin"
 	RoomserverPerformLeavePath    = "/roomserver/performLeave"
 	RoomserverPerformBackfillPath = "/roomserver/performBackfill"
@@ -209,6 +210,23 @@ func (h *httpRoomserverInternalAPI) PerformPeek(
 	}
 }
 
+func (h *httpRoomserverInternalAPI) PerformUnpeek(
+	ctx context.Context,
+	request *api.PerformUnpeekRequest,
+	response *api.PerformUnpeekResponse,
+) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "PerformUnpeek")
+	defer span.Finish()
+
+	apiURL := h.roomserverURL + RoomserverPerformUnpeekPath
+	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+	if err != nil {
+		response.Error = &api.PerformError{
+			Msg: fmt.Sprintf("failed to communicate with roomserver: %s", err),
+		}
+	}
+}
+
 func (h *httpRoomserverInternalAPI) PerformLeave(
 	ctx context.Context,
 	request *api.PerformLeaveRequest,
diff --git a/roomserver/inthttp/server.go b/roomserver/inthttp/server.go
index 2bc8f82d..ac1fc25b 100644
--- a/roomserver/inthttp/server.go
+++ b/roomserver/inthttp/server.go
@@ -72,6 +72,17 @@ func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(RoomserverPerformPeekPath,
+		httputil.MakeInternalAPI("performUnpeek", func(req *http.Request) util.JSONResponse {
+			var request api.PerformUnpeekRequest
+			var response api.PerformUnpeekResponse
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			r.PerformUnpeek(req.Context(), &request, &response)
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 	internalAPIMux.Handle(RoomserverPerformPublishPath,
 		httputil.MakeInternalAPI("performPublish", func(req *http.Request) util.JSONResponse {
 			var request api.PerformPublishRequest