Implement unspecced server notices

clientapi/routing/routing.go

@@ -117,11 +117,49 @@ func Setup(
 		).Methods(http.MethodGet, http.MethodPost, http.MethodOptions)
 	}
 
+	synapseAdminRouter.Handle("/admin/v1/send_server_notice/{txnID}",
+		httputil.MakeAuthAPI("send_server_notice", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			// not specced, but ensure we're rate limiting requests to this endpoint
+			if r := rateLimits.Limit(req); r != nil {
+				return *r
+			}
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+			txnID := vars["txnID"]
+			return SendServerNotice(
+				req, &cfg.Matrix.ServerNotices,
+				cfg, userAPI, rsAPI, accountDB, asAPI,
+				device,
+				&txnID, transactionsCache,
+			)
+		}),
+	).Methods(http.MethodPut, http.MethodOptions)
+
+	synapseAdminRouter.Handle("/admin/v1/send_server_notice",
+		httputil.MakeAuthAPI("send_server_notice", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			// not specced, but ensure we're rate limiting requests to this endpoint
+			if r := rateLimits.Limit(req); r != nil {
+				return *r
+			}
+			return SendServerNotice(
+				req, &cfg.Matrix.ServerNotices,
+				cfg, userAPI, rsAPI, accountDB, asAPI,
+				device,
+				nil, transactionsCache,
+			)
+		}),
+	).Methods(http.MethodPost, http.MethodOptions)
+
 	r0mux := publicAPIMux.PathPrefix("/r0").Subrouter()
 	unstableMux := publicAPIMux.PathPrefix("/unstable").Subrouter()
 
 	r0mux.Handle("/createRoom",
 		httputil.MakeAuthAPI("createRoom", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			if r := rateLimits.Limit(req); r != nil {
+				return *r
+			}
 			return CreateRoom(req, device, cfg, accountDB, rsAPI, asAPI)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)

clientapi/routing/server_notices.go

@@ -0,0 +1,311 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// 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 routing
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/matrix-org/gomatrix"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/gomatrixserverlib/tokens"
+	"github.com/matrix-org/util"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
+
+	appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
+	"github.com/matrix-org/dendrite/clientapi/httputil"
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/internal/eventutil"
+	"github.com/matrix-org/dendrite/internal/transactions"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/config"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/dendrite/userapi/storage/accounts"
+)
+
+// Unspecced server notice request
+// https://github.com/matrix-org/synapse/blob/develop/docs/admin_api/server_notices.md
+type sendServerNoticeRequest struct {
+	UserID  string `json:"user_id,omitempty"`
+	Content struct {
+		MsgType string `json:"msgtype,omitempty"`
+		Body    string `json:"body,omitempty"`
+	} `json:"content,omitempty"`
+	Type     string `json:"type,omitempty"`
+	StateKey string `json:"state_key,omitempty"`
+}
+
+// SendServerNotice sends a message to a specific user. It can only be invoked by an admin.
+func SendServerNotice(
+	req *http.Request,
+	cfgNotices *config.ServerNotices,
+	cfgClient *config.ClientAPI,
+	userAPI userapi.UserInternalAPI,
+	rsAPI api.RoomserverInternalAPI,
+	accountsDB accounts.Database,
+	asAPI appserviceAPI.AppServiceQueryAPI,
+	device *userapi.Device,
+	txnID *string,
+	txnCache *transactions.Cache,
+) util.JSONResponse {
+	// TODO: Only allow admins to send notices
+	if !cfgNotices.Enabled {
+		return util.MessageResponse(http.StatusBadRequest, "Server notices are not enabled on this server.")
+	}
+
+	if txnID != nil {
+		// Try to fetch response from transactionsCache
+		if res, ok := txnCache.FetchTransaction(device.AccessToken, *txnID); ok {
+			return *res
+		}
+	}
+
+	ctx := req.Context()
+	var r sendServerNoticeRequest
+	resErr := httputil.UnmarshalJSONRequest(req, &r)
+	if resErr != nil {
+		return *resErr
+	}
+
+	// get rooms for specified user
+	allUserRooms := []string{}
+	userRooms := api.QueryRoomsForUserResponse{}
+	if err := rsAPI.QueryRoomsForUser(ctx, &api.QueryRoomsForUserRequest{
+		UserID:         r.UserID,
+		WantMembership: "join",
+	}, &userRooms); err != nil {
+		return util.ErrorResponse(err)
+	}
+	allUserRooms = append(allUserRooms, userRooms.RoomIDs...)
+	// get invites for specified user
+	if err := rsAPI.QueryRoomsForUser(ctx, &api.QueryRoomsForUserRequest{
+		UserID:         r.UserID,
+		WantMembership: "invite",
+	}, &userRooms); err != nil {
+		return util.ErrorResponse(err)
+	}
+	allUserRooms = append(allUserRooms, userRooms.RoomIDs...)
+
+	// get rooms of the sender
+	senderUserID := fmt.Sprintf("@%s:%s", cfgNotices.LocalPart, cfgClient.Matrix.ServerName)
+	senderRooms := api.QueryRoomsForUserResponse{}
+	if err := rsAPI.QueryRoomsForUser(ctx, &api.QueryRoomsForUserRequest{
+		UserID:         senderUserID,
+		WantMembership: "join",
+	}, &senderRooms); err != nil {
+		return util.ErrorResponse(err)
+	}
+
+	// check if we have rooms in common
+	commonRooms := []string{}
+	for _, userRoomID := range allUserRooms {
+		for _, senderRoomID := range senderRooms.RoomIDs {
+			if userRoomID == senderRoomID {
+				commonRooms = append(commonRooms, senderRoomID)
+			}
+		}
+	}
+
+	if len(commonRooms) > 1 {
+		return util.ErrorResponse(fmt.Errorf("expected to find one room, but got %d", len(commonRooms)))
+	}
+
+	var (
+		roomID      string
+		roomVersion = gomatrixserverlib.RoomVersionV6
+	)
+
+	adminDevice, err := getSenderDevice(ctx, userAPI, cfgClient)
+	if err != nil {
+		logrus.WithError(err).Error("unable to get device")
+		return util.ErrorResponse(err)
+	}
+	// create a new room for the user
+	if len(commonRooms) == 0 {
+		powerLevelContent := eventutil.InitialPowerLevelsContent(senderUserID)
+		powerLevelContent.Users[r.UserID] = -10
+		pl, err := json.Marshal(powerLevelContent)
+		if err != nil {
+			return util.ErrorResponse(err)
+		}
+		createContent := map[string]interface{}{}
+		createContent["m.federate"] = false
+		cc, err := json.Marshal(createContent)
+		if err != nil {
+			return util.ErrorResponse(err)
+		}
+		crReq := createRoomRequest{
+			Invite:                    []string{r.UserID},
+			Name:                      cfgNotices.RoomName,
+			Visibility:                "private",
+			Preset:                    presetPrivateChat,
+			CreationContent:           cc,
+			GuestCanJoin:              false,
+			RoomVersion:               roomVersion,
+			PowerLevelContentOverride: pl,
+		}
+
+		roomRes := createRoom(ctx, crReq, adminDevice, cfgClient, accountsDB, rsAPI, asAPI, time.Now())
+
+		switch data := roomRes.JSON.(type) {
+		case createRoomResponse:
+			roomID = data.RoomID
+
+			// tag the room, so we can later check if the user tries to reject an invite
+			serverAlertTag := gomatrix.TagContent{Tags: map[string]gomatrix.TagProperties{
+				"m.server_notice": {
+					Order: 1.0,
+				},
+			}}
+			if err = saveTagData(req, r.UserID, roomID, userAPI, serverAlertTag); err != nil {
+				util.GetLogger(req.Context()).WithError(err).Error("saveTagData failed")
+				return jsonerror.InternalServerError()
+			}
+
+		default:
+			return roomRes
+		}
+
+	} else {
+		roomID = commonRooms[0]
+	}
+
+	startedGeneratingEvent := time.Now()
+
+	request := map[string]interface{}{
+		"body":    r.Content.Body,
+		"msgtype": r.Content.MsgType,
+	}
+	e, resErr := generateSendEvent(ctx, request, adminDevice, roomID, "m.room.message", nil, cfgClient, rsAPI, time.Now())
+	if resErr != nil {
+		logrus.Errorf("failed to send message: %+v", resErr)
+		return *resErr
+	}
+	timeToGenerateEvent := time.Since(startedGeneratingEvent)
+
+	var txnAndSessionID *api.TransactionID
+	if txnID != nil {
+		txnAndSessionID = &api.TransactionID{
+			TransactionID: *txnID,
+			SessionID:     device.SessionID,
+		}
+	}
+
+	// pass the new event to the roomserver and receive the correct event ID
+	// event ID in case of duplicate transaction is discarded
+	startedSubmittingEvent := time.Now()
+	if err := api.SendEvents(
+		ctx, rsAPI,
+		api.KindNew,
+		[]*gomatrixserverlib.HeaderedEvent{
+			e.Headered(roomVersion),
+		},
+		cfgClient.Matrix.ServerName,
+		cfgClient.Matrix.ServerName,
+		txnAndSessionID,
+		false,
+	); err != nil {
+		util.GetLogger(ctx).WithError(err).Error("SendEvents failed")
+		return jsonerror.InternalServerError()
+	}
+	util.GetLogger(req.Context()).WithFields(logrus.Fields{
+		"event_id":     e.EventID(),
+		"room_id":      roomID,
+		"room_version": roomVersion,
+	}).Info("Sent event to roomserver")
+	timeToSubmitEvent := time.Since(startedSubmittingEvent)
+
+	res := util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: sendEventResponse{e.EventID()},
+	}
+	// Add response to transactionsCache
+	if txnID != nil {
+		txnCache.AddTransaction(device.AccessToken, *txnID, &res)
+	}
+
+	// Take a note of how long it took to generate the event vs submit
+	// it to the roomserver.
+	sendEventDuration.With(prometheus.Labels{"action": "build"}).Observe(float64(timeToGenerateEvent.Milliseconds()))
+	sendEventDuration.With(prometheus.Labels{"action": "submit"}).Observe(float64(timeToSubmitEvent.Milliseconds()))
+
+	return res
+}
+
+func (r sendServerNoticeRequest) validate() (ok bool) {
+	if r.UserID == "" {
+		return false
+	}
+	if r.Content.MsgType == "" || r.Content.Body == "" {
+		return false
+	}
+	return true
+}
+
+// getSenderDevice creates a user account to be used when sending server notices.
+// It returns an userapi.Device, which is used for building the
+func getSenderDevice(ctx context.Context, userAPI userapi.UserInternalAPI, cfg *config.ClientAPI) (*userapi.Device, error) {
+	var accRes userapi.PerformAccountCreationResponse
+	// create account if it doesn't exist
+	err := userAPI.PerformAccountCreation(ctx, &userapi.PerformAccountCreationRequest{
+		AccountType: userapi.AccountTypeUser,
+		Localpart:   cfg.Matrix.ServerNotices.LocalPart,
+		OnConflict:  userapi.ConflictUpdate,
+	}, &accRes)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check if we got existing devices
+	deviceRes := &userapi.QueryDevicesResponse{}
+	err = userAPI.QueryDevices(ctx, &userapi.QueryDevicesRequest{
+		UserID: accRes.Account.UserID,
+	}, deviceRes)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(deviceRes.Devices) > 0 {
+		return &deviceRes.Devices[0], nil
+	}
+
+	// create an AccessToken
+	token, err := tokens.GenerateLoginToken(tokens.TokenOptions{
+		ServerPrivateKey: cfg.Matrix.PrivateKey.Seed(),
+		ServerName:       string(cfg.Matrix.ServerName),
+		UserID:           accRes.Account.UserID,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// create a new device, if we didn't find any
+	var devRes userapi.PerformDeviceCreationResponse
+	err = userAPI.PerformDeviceCreation(ctx, &userapi.PerformDeviceCreationRequest{
+		Localpart:          cfg.Matrix.ServerNotices.LocalPart,
+		DeviceDisplayName:  &cfg.Matrix.ServerNotices.LocalPart,
+		AccessToken:        token,
+		NoDeviceListUpdate: true,
+	}, &devRes)
+
+	if err != nil {
+		return nil, err
+	}
+	return devRes.Device, nil
+}

clientapi/routing/server_notices_test.go

@@ -0,0 +1,77 @@
+package routing
+
+import (
+	"testing"
+)
+
+func Test_sendServerNoticeRequest_validate(t *testing.T) {
+	type fields struct {
+		UserID  string `json:"user_id,omitempty"`
+		Content struct {
+			MsgType string `json:"msgtype,omitempty"`
+			Body    string `json:"body,omitempty"`
+		} `json:"content,omitempty"`
+		Type     string `json:"type,omitempty"`
+		StateKey string `json:"state_key,omitempty"`
+	}
+
+	content := struct {
+		MsgType string `json:"msgtype,omitempty"`
+		Body    string `json:"body,omitempty"`
+	}{
+		MsgType: "m.text",
+		Body:    "Hello world!",
+	}
+
+	tests := []struct {
+		name   string
+		fields fields
+		wantOk bool
+	}{
+		{
+			name:   "empty request",
+			fields: fields{},
+		},
+		{
+			name: "msgtype empty",
+			fields: fields{
+				UserID: "@alice:localhost",
+			},
+		},
+		{
+			name: "msg body empty",
+			fields: fields{
+				UserID: "@alice:localhost",
+			},
+		},
+		{
+			name: "statekey empty",
+			fields: fields{
+				UserID:  "@alice:localhost",
+				Content: content,
+			},
+			wantOk: true,
+		},
+		{
+			name: "type empty",
+			fields: fields{
+				UserID:  "@alice:localhost",
+				Content: content,
+			},
+			wantOk: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := sendServerNoticeRequest{
+				UserID:   tt.fields.UserID,
+				Content:  tt.fields.Content,
+				Type:     tt.fields.Type,
+				StateKey: tt.fields.StateKey,
+			}
+			if gotOk := r.validate(); gotOk != tt.wantOk {
+				t.Errorf("validate() = %v, want %v", gotOk, tt.wantOk)
+			}
+		})
+	}
+}