2017-09-08 16:17:12 +02:00
|
|
|
// Copyright 2017 Vector Creations 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.
|
|
|
|
|
2017-10-11 19:16:53 +02:00
|
|
|
package routing
|
2017-06-07 15:32:53 +02:00
|
|
|
|
|
|
|
import (
|
2017-09-13 12:03:41 +02:00
|
|
|
"context"
|
2017-06-07 15:32:53 +02:00
|
|
|
"encoding/json"
|
|
|
|
"fmt"
|
2017-08-23 16:13:47 +02:00
|
|
|
"net/http"
|
2020-09-28 12:32:59 +02:00
|
|
|
"sync"
|
2020-09-07 13:32:40 +02:00
|
|
|
"time"
|
2017-08-23 16:13:47 +02:00
|
|
|
|
2017-06-07 15:32:53 +02:00
|
|
|
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
2021-06-30 13:05:58 +02:00
|
|
|
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
|
2022-03-29 14:14:35 +02:00
|
|
|
"github.com/matrix-org/dendrite/federationapi/producers"
|
|
|
|
"github.com/matrix-org/dendrite/federationapi/types"
|
2021-03-30 11:01:32 +02:00
|
|
|
"github.com/matrix-org/dendrite/internal"
|
2020-08-05 14:41:16 +02:00
|
|
|
keyapi "github.com/matrix-org/dendrite/keyserver/api"
|
2017-06-07 15:32:53 +02:00
|
|
|
"github.com/matrix-org/dendrite/roomserver/api"
|
2020-12-02 18:41:00 +01:00
|
|
|
"github.com/matrix-org/dendrite/setup/config"
|
2022-04-06 13:11:19 +02:00
|
|
|
syncTypes "github.com/matrix-org/dendrite/syncapi/types"
|
2017-06-07 15:32:53 +02:00
|
|
|
"github.com/matrix-org/gomatrixserverlib"
|
|
|
|
"github.com/matrix-org/util"
|
2021-03-23 12:33:36 +01:00
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
2020-05-05 16:48:37 +02:00
|
|
|
"github.com/sirupsen/logrus"
|
2017-06-07 15:32:53 +02:00
|
|
|
)
|
|
|
|
|
2021-03-23 16:22:00 +01:00
|
|
|
const (
|
|
|
|
// Event was passed to the roomserver
|
|
|
|
MetricsOutcomeOK = "ok"
|
|
|
|
// Event failed to be processed
|
|
|
|
MetricsOutcomeFail = "fail"
|
|
|
|
// Event failed auth checks
|
|
|
|
MetricsOutcomeRejected = "rejected"
|
|
|
|
// Terminated the transaction
|
|
|
|
MetricsOutcomeFatal = "fatal"
|
|
|
|
// The event has missing auth_events we need to fetch
|
|
|
|
MetricsWorkMissingAuthEvents = "missing_auth_events"
|
|
|
|
// No work had to be done as we had all prev/auth events
|
|
|
|
MetricsWorkDirect = "direct"
|
|
|
|
// The event has missing prev_events we need to call /g_m_e for
|
|
|
|
MetricsWorkMissingPrevEvents = "missing_prev_events"
|
|
|
|
)
|
|
|
|
|
2021-03-23 12:33:36 +01:00
|
|
|
var (
|
|
|
|
pduCountTotal = prometheus.NewCounterVec(
|
|
|
|
prometheus.CounterOpts{
|
|
|
|
Namespace: "dendrite",
|
|
|
|
Subsystem: "federationapi",
|
|
|
|
Name: "recv_pdus",
|
2021-03-23 16:22:00 +01:00
|
|
|
Help: "Number of incoming PDUs from remote servers with labels for success",
|
2021-03-23 12:33:36 +01:00
|
|
|
},
|
2021-03-23 16:22:00 +01:00
|
|
|
[]string{"status"}, // 'success' or 'total'
|
2021-03-23 12:33:36 +01:00
|
|
|
)
|
|
|
|
eduCountTotal = prometheus.NewCounter(
|
|
|
|
prometheus.CounterOpts{
|
|
|
|
Namespace: "dendrite",
|
|
|
|
Subsystem: "federationapi",
|
|
|
|
Name: "recv_edus",
|
2021-03-23 16:22:00 +01:00
|
|
|
Help: "Number of incoming EDUs from remote servers",
|
|
|
|
},
|
|
|
|
)
|
2021-03-23 12:33:36 +01:00
|
|
|
)
|
|
|
|
|
2021-11-08 10:24:16 +01:00
|
|
|
var inFlightTxnsPerOrigin sync.Map // transaction ID -> chan util.JSONResponse
|
2021-07-02 13:33:27 +02:00
|
|
|
|
2017-06-07 15:32:53 +02:00
|
|
|
// Send implements /_matrix/federation/v1/send/{txnID}
|
|
|
|
func Send(
|
2017-09-04 14:14:01 +02:00
|
|
|
httpReq *http.Request,
|
|
|
|
request *gomatrixserverlib.FederationRequest,
|
2017-06-07 15:32:53 +02:00
|
|
|
txnID gomatrixserverlib.TransactionID,
|
2020-08-10 15:18:04 +02:00
|
|
|
cfg *config.FederationAPI,
|
2022-05-05 20:30:38 +02:00
|
|
|
rsAPI api.FederationRoomserverAPI,
|
2022-05-06 13:39:26 +02:00
|
|
|
keyAPI keyapi.FederationKeyAPI,
|
2020-06-15 17:57:59 +02:00
|
|
|
keys gomatrixserverlib.JSONVerifier,
|
2022-05-17 14:23:35 +02:00
|
|
|
federation federationAPI.FederationClient,
|
2021-03-30 11:01:32 +02:00
|
|
|
mu *internal.MutexByRoom,
|
2021-06-30 13:05:58 +02:00
|
|
|
servers federationAPI.ServersInRoomProvider,
|
2022-03-29 14:14:35 +02:00
|
|
|
producer *producers.SyncAPIProducer,
|
2017-06-07 15:32:53 +02:00
|
|
|
) util.JSONResponse {
|
2021-11-08 10:24:16 +01:00
|
|
|
// First we should check if this origin has already submitted this
|
|
|
|
// txn ID to us. If they have and the txnIDs map contains an entry,
|
|
|
|
// the transaction is still being worked on. The new client can wait
|
|
|
|
// for it to complete rather than creating more work.
|
|
|
|
index := string(request.Origin()) + "\000" + string(txnID)
|
|
|
|
v, ok := inFlightTxnsPerOrigin.LoadOrStore(index, make(chan util.JSONResponse, 1))
|
|
|
|
ch := v.(chan util.JSONResponse)
|
|
|
|
if ok {
|
|
|
|
// This origin already submitted this txn ID to us, and the work
|
|
|
|
// is still taking place, so we'll just wait for it to finish.
|
|
|
|
ctx, cancel := context.WithTimeout(httpReq.Context(), time.Minute*5)
|
|
|
|
defer cancel()
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
// If the caller gives up then return straight away. We don't
|
|
|
|
// want to attempt to process what they sent us any further.
|
|
|
|
return util.JSONResponse{Code: http.StatusRequestTimeout}
|
|
|
|
case res := <-ch:
|
|
|
|
// The original task just finished processing so let's return
|
|
|
|
// the result of it.
|
|
|
|
if res.Code == 0 {
|
|
|
|
return util.JSONResponse{Code: http.StatusAccepted}
|
|
|
|
}
|
|
|
|
return res
|
|
|
|
}
|
|
|
|
}
|
|
|
|
// Otherwise, store that we're currently working on this txn from
|
|
|
|
// this origin. When we're done processing, close the channel.
|
|
|
|
defer close(ch)
|
|
|
|
defer inFlightTxnsPerOrigin.Delete(index)
|
|
|
|
|
2017-06-07 15:32:53 +02:00
|
|
|
t := txnReq{
|
2022-04-06 13:11:19 +02:00
|
|
|
rsAPI: rsAPI,
|
|
|
|
keys: keys,
|
2022-04-25 15:22:46 +02:00
|
|
|
ourServerName: cfg.Matrix.ServerName,
|
2022-04-06 13:11:19 +02:00
|
|
|
federation: federation,
|
|
|
|
servers: servers,
|
|
|
|
keyAPI: keyAPI,
|
|
|
|
roomsMu: mu,
|
|
|
|
producer: producer,
|
|
|
|
inboundPresenceEnabled: cfg.Matrix.Presence.EnableInbound,
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
2020-03-27 17:28:22 +01:00
|
|
|
|
|
|
|
var txnEvents struct {
|
2020-03-30 17:40:28 +02:00
|
|
|
PDUs []json.RawMessage `json:"pdus"`
|
|
|
|
EDUs []gomatrixserverlib.EDU `json:"edus"`
|
2020-03-27 17:28:22 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := json.Unmarshal(request.Content(), &txnEvents); err != nil {
|
2017-06-07 15:32:53 +02:00
|
|
|
return util.JSONResponse{
|
2018-03-13 16:55:45 +01:00
|
|
|
Code: http.StatusBadRequest,
|
2017-08-23 16:13:47 +02:00
|
|
|
JSON: jsonerror.NotJSON("The request body could not be decoded into valid JSON. " + err.Error()),
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
|
|
|
}
|
2020-06-23 14:15:15 +02:00
|
|
|
// Transactions are limited in size; they can have at most 50 PDUs and 100 EDUs.
|
|
|
|
// https://matrix.org/docs/spec/server_server/latest#transactions
|
|
|
|
if len(txnEvents.PDUs) > 50 || len(txnEvents.EDUs) > 100 {
|
|
|
|
return util.JSONResponse{
|
|
|
|
Code: http.StatusBadRequest,
|
|
|
|
JSON: jsonerror.BadJSON("max 50 pdus / 100 edus"),
|
|
|
|
}
|
|
|
|
}
|
2017-06-07 15:32:53 +02:00
|
|
|
|
2020-03-30 17:40:28 +02:00
|
|
|
// TODO: Really we should have a function to convert FederationRequest to txnReq
|
2020-03-27 17:28:22 +01:00
|
|
|
t.PDUs = txnEvents.PDUs
|
2020-03-30 17:40:28 +02:00
|
|
|
t.EDUs = txnEvents.EDUs
|
2017-06-07 15:32:53 +02:00
|
|
|
t.Origin = request.Origin()
|
|
|
|
t.TransactionID = txnID
|
2017-06-19 16:21:04 +02:00
|
|
|
t.Destination = cfg.Matrix.ServerName
|
2017-06-07 15:32:53 +02:00
|
|
|
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(httpReq.Context()).Debugf("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, request.Origin(), len(t.PDUs), len(t.EDUs))
|
2020-03-27 17:28:22 +01:00
|
|
|
|
2022-01-27 15:29:14 +01:00
|
|
|
resp, jsonErr := t.processTransaction(httpReq.Context())
|
2020-06-23 14:15:15 +02:00
|
|
|
if jsonErr != nil {
|
|
|
|
util.GetLogger(httpReq.Context()).WithField("jsonErr", jsonErr).Error("t.processTransaction failed")
|
|
|
|
return *jsonErr
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
2020-05-13 14:01:45 +02:00
|
|
|
|
|
|
|
// https://matrix.org/docs/spec/server_server/r0.1.3#put-matrix-federation-v1-send-txnid
|
|
|
|
// Status code 200:
|
|
|
|
// The result of processing the transaction. The server is to use this response
|
|
|
|
// even in the event of one or more PDUs failing to be processed.
|
2021-11-08 10:24:16 +01:00
|
|
|
res := util.JSONResponse{
|
2020-05-13 14:01:45 +02:00
|
|
|
Code: http.StatusOK,
|
|
|
|
JSON: resp,
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
2021-11-08 10:24:16 +01:00
|
|
|
ch <- res
|
|
|
|
return res
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
type txnReq struct {
|
|
|
|
gomatrixserverlib.Transaction
|
2022-05-05 20:30:38 +02:00
|
|
|
rsAPI api.FederationRoomserverAPI
|
2022-05-06 13:39:26 +02:00
|
|
|
keyAPI keyapi.FederationKeyAPI
|
2022-04-25 15:22:46 +02:00
|
|
|
ourServerName gomatrixserverlib.ServerName
|
2022-04-06 13:11:19 +02:00
|
|
|
keys gomatrixserverlib.JSONVerifier
|
|
|
|
federation txnFederationClient
|
|
|
|
roomsMu *internal.MutexByRoom
|
|
|
|
servers federationAPI.ServersInRoomProvider
|
|
|
|
producer *producers.SyncAPIProducer
|
|
|
|
inboundPresenceEnabled bool
|
2021-07-07 19:55:44 +02:00
|
|
|
}
|
|
|
|
|
2020-05-06 15:27:02 +02:00
|
|
|
// A subset of FederationClient functionality that txn requires. Useful for testing.
|
|
|
|
type txnFederationClient interface {
|
|
|
|
LookupState(ctx context.Context, s gomatrixserverlib.ServerName, roomID string, eventID string, roomVersion gomatrixserverlib.RoomVersion) (
|
|
|
|
res gomatrixserverlib.RespState, err error,
|
|
|
|
)
|
|
|
|
LookupStateIDs(ctx context.Context, s gomatrixserverlib.ServerName, roomID string, eventID string) (res gomatrixserverlib.RespStateIDs, err error)
|
|
|
|
GetEvent(ctx context.Context, s gomatrixserverlib.ServerName, eventID string) (res gomatrixserverlib.Transaction, err error)
|
2020-05-12 17:24:28 +02:00
|
|
|
LookupMissingEvents(ctx context.Context, s gomatrixserverlib.ServerName, roomID string, missing gomatrixserverlib.MissingEvents,
|
|
|
|
roomVersion gomatrixserverlib.RoomVersion) (res gomatrixserverlib.RespMissingEvents, err error)
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
|
|
|
|
2020-09-07 13:32:40 +02:00
|
|
|
func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.RespSend, *util.JSONResponse) {
|
2021-07-02 13:33:27 +02:00
|
|
|
var wg sync.WaitGroup
|
2022-01-27 15:29:14 +01:00
|
|
|
wg.Add(1)
|
|
|
|
go func() {
|
|
|
|
defer wg.Done()
|
|
|
|
t.processEDUs(ctx)
|
|
|
|
}()
|
|
|
|
|
|
|
|
results := make(map[string]gomatrixserverlib.PDUResult)
|
|
|
|
roomVersions := make(map[string]gomatrixserverlib.RoomVersion)
|
|
|
|
getRoomVersion := func(roomID string) gomatrixserverlib.RoomVersion {
|
|
|
|
if v, ok := roomVersions[roomID]; ok {
|
|
|
|
return v
|
|
|
|
}
|
|
|
|
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
|
|
|
|
verRes := api.QueryRoomVersionForRoomResponse{}
|
|
|
|
if err := t.rsAPI.QueryRoomVersionForRoom(ctx, &verReq, &verRes); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Transaction: Failed to query room version for room", verReq.RoomID)
|
2022-01-27 15:29:14 +01:00
|
|
|
return ""
|
|
|
|
}
|
|
|
|
roomVersions[roomID] = verRes.RoomVersion
|
|
|
|
return verRes.RoomVersion
|
|
|
|
}
|
2020-04-16 18:59:55 +02:00
|
|
|
|
2020-03-27 17:28:22 +01:00
|
|
|
for _, pdu := range t.PDUs {
|
2021-03-23 12:33:36 +01:00
|
|
|
pduCountTotal.WithLabelValues("total").Inc()
|
2020-03-27 17:28:22 +01:00
|
|
|
var header struct {
|
|
|
|
RoomID string `json:"room_id"`
|
|
|
|
}
|
|
|
|
if err := json.Unmarshal(pdu, &header); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Transaction: Failed to extract room ID from event")
|
2020-05-13 14:01:45 +02:00
|
|
|
// We don't know the event ID at this point so we can't return the
|
|
|
|
// failure in the PDU results
|
|
|
|
continue
|
2020-03-27 17:28:22 +01:00
|
|
|
}
|
2022-01-27 15:29:14 +01:00
|
|
|
roomVersion := getRoomVersion(header.RoomID)
|
|
|
|
event, err := gomatrixserverlib.NewEventFromUntrustedJSON(pdu, roomVersion)
|
2020-03-27 17:28:22 +01:00
|
|
|
if err != nil {
|
2020-06-23 14:15:15 +02:00
|
|
|
if _, ok := err.(gomatrixserverlib.BadJSONError); ok {
|
|
|
|
// Room version 6 states that homeservers should strictly enforce canonical JSON
|
|
|
|
// on PDUs.
|
|
|
|
//
|
|
|
|
// This enforces that the entire transaction is rejected if a single bad PDU is
|
|
|
|
// sent. It is unclear if this is the correct behaviour or not.
|
|
|
|
//
|
|
|
|
// See https://github.com/matrix-org/synapse/issues/7543
|
|
|
|
return nil, &util.JSONResponse{
|
|
|
|
Code: 400,
|
|
|
|
JSON: jsonerror.BadJSON("PDU contains bad JSON"),
|
|
|
|
}
|
2020-05-13 14:01:45 +02:00
|
|
|
}
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debugf("Transaction: Failed to parse event JSON of event %s", string(pdu))
|
2020-05-13 14:01:45 +02:00
|
|
|
continue
|
2020-03-27 17:28:22 +01:00
|
|
|
}
|
2022-02-11 16:18:14 +01:00
|
|
|
if event.Type() == gomatrixserverlib.MRoomCreate && event.StateKeyEquals("") {
|
|
|
|
continue
|
|
|
|
}
|
2020-09-07 13:32:40 +02:00
|
|
|
if api.IsServerBannedFromRoom(ctx, t.rsAPI, event.RoomID(), t.Origin) {
|
2020-08-11 19:19:11 +02:00
|
|
|
results[event.EventID()] = gomatrixserverlib.PDUResult{
|
|
|
|
Error: "Forbidden by server ACLs",
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
2021-11-02 11:13:38 +01:00
|
|
|
if err = event.VerifyEventSignatures(ctx, t.keys); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debugf("Transaction: Couldn't validate signature of event %q", event.EventID())
|
2020-05-13 14:01:45 +02:00
|
|
|
results[event.EventID()] = gomatrixserverlib.PDUResult{
|
|
|
|
Error: err.Error(),
|
|
|
|
}
|
|
|
|
continue
|
2020-03-27 17:28:22 +01:00
|
|
|
}
|
2021-07-02 13:33:27 +02:00
|
|
|
|
2022-01-27 15:29:14 +01:00
|
|
|
// pass the event to the roomserver which will do auth checks
|
|
|
|
// If the event fail auth checks, gmsl.NotAllowed error will be returned which we be silently
|
|
|
|
// discarded by the caller of this function
|
|
|
|
if err = api.SendEvents(
|
|
|
|
ctx,
|
|
|
|
t.rsAPI,
|
|
|
|
api.KindNew,
|
|
|
|
[]*gomatrixserverlib.HeaderedEvent{
|
|
|
|
event.Headered(roomVersion),
|
|
|
|
},
|
|
|
|
t.Origin,
|
|
|
|
api.DoNotSendToOtherServers,
|
|
|
|
nil,
|
|
|
|
true,
|
|
|
|
); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Errorf("Transaction: Couldn't submit event %q to input queue: %s", event.EventID(), err)
|
2022-01-27 15:29:14 +01:00
|
|
|
results[event.EventID()] = gomatrixserverlib.PDUResult{
|
|
|
|
Error: err.Error(),
|
2017-06-07 15:32:53 +02:00
|
|
|
}
|
2021-07-02 13:33:27 +02:00
|
|
|
continue
|
|
|
|
}
|
2020-05-13 14:01:45 +02:00
|
|
|
|
2022-01-27 15:29:14 +01:00
|
|
|
results[event.EventID()] = gomatrixserverlib.PDUResult{}
|
|
|
|
pduCountTotal.WithLabelValues("success").Inc()
|
|
|
|
}
|
2017-06-07 15:32:53 +02:00
|
|
|
|
2022-01-27 15:29:14 +01:00
|
|
|
wg.Wait()
|
|
|
|
return &gomatrixserverlib.RespSend{PDUs: results}, nil
|
2020-05-12 17:24:28 +02:00
|
|
|
}
|
|
|
|
|
2022-04-25 15:22:46 +02:00
|
|
|
// nolint:gocyclo
|
2020-09-07 13:32:40 +02:00
|
|
|
func (t *txnReq) processEDUs(ctx context.Context) {
|
|
|
|
for _, e := range t.EDUs {
|
2021-03-23 12:33:36 +01:00
|
|
|
eduCountTotal.Inc()
|
2020-03-30 17:40:28 +02:00
|
|
|
switch e.Type {
|
|
|
|
case gomatrixserverlib.MTyping:
|
|
|
|
// https://matrix.org/docs/spec/server_server/latest#typing-notifications
|
|
|
|
var typingPayload struct {
|
|
|
|
RoomID string `json:"room_id"`
|
|
|
|
UserID string `json:"user_id"`
|
|
|
|
Typing bool `json:"typing"`
|
|
|
|
}
|
|
|
|
if err := json.Unmarshal(e.Content, &typingPayload); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal typing event")
|
2020-03-30 17:40:28 +02:00
|
|
|
continue
|
|
|
|
}
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', typingPayload.UserID); err != nil {
|
2020-10-14 17:49:25 +02:00
|
|
|
continue
|
2022-04-25 15:22:46 +02:00
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
continue
|
|
|
|
} else if serverName != t.Origin {
|
2020-10-14 17:49:25 +02:00
|
|
|
continue
|
|
|
|
}
|
2022-03-29 14:14:35 +02:00
|
|
|
if err := t.producer.SendTyping(ctx, typingPayload.UserID, typingPayload.RoomID, typingPayload.Typing, 30*1000); err != nil {
|
|
|
|
util.GetLogger(ctx).WithError(err).Error("Failed to send typing event to JetStream")
|
2020-03-30 17:40:28 +02:00
|
|
|
}
|
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 18:50:19 +02:00
|
|
|
case gomatrixserverlib.MDirectToDevice:
|
|
|
|
// https://matrix.org/docs/spec/server_server/r0.1.3#m-direct-to-device-schema
|
|
|
|
var directPayload gomatrixserverlib.ToDeviceMessage
|
|
|
|
if err := json.Unmarshal(e.Content, &directPayload); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal send-to-device events")
|
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 18:50:19 +02:00
|
|
|
continue
|
|
|
|
}
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', directPayload.Sender); err != nil {
|
|
|
|
continue
|
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
continue
|
|
|
|
} else if serverName != t.Origin {
|
|
|
|
continue
|
|
|
|
}
|
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 18:50:19 +02:00
|
|
|
for userID, byUser := range directPayload.Messages {
|
|
|
|
for deviceID, message := range byUser {
|
|
|
|
// TODO: check that the user and the device actually exist here
|
2022-03-29 14:14:35 +02:00
|
|
|
if err := t.producer.SendToDevice(ctx, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil {
|
2020-09-07 13:32:40 +02:00
|
|
|
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
|
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 18:50:19 +02:00
|
|
|
"sender": directPayload.Sender,
|
|
|
|
"user_id": userID,
|
|
|
|
"device_id": deviceID,
|
2022-03-29 14:14:35 +02:00
|
|
|
}).Error("Failed to send send-to-device event to JetStream")
|
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 18:50:19 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2020-08-05 14:41:16 +02:00
|
|
|
case gomatrixserverlib.MDeviceListUpdate:
|
2020-09-07 13:32:40 +02:00
|
|
|
t.processDeviceListUpdate(ctx, e)
|
2020-11-09 19:46:11 +01:00
|
|
|
case gomatrixserverlib.MReceipt:
|
|
|
|
// https://matrix.org/docs/spec/server_server/r0.1.4#receipts
|
2022-03-29 14:14:35 +02:00
|
|
|
payload := map[string]types.FederationReceiptMRead{}
|
2020-11-09 19:46:11 +01:00
|
|
|
|
|
|
|
if err := json.Unmarshal(e.Content, &payload); err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal receipt event")
|
2020-11-09 19:46:11 +01:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
for roomID, receipt := range payload {
|
|
|
|
for userID, mread := range receipt.User {
|
|
|
|
_, domain, err := gomatrixserverlib.SplitID('@', userID)
|
|
|
|
if err != nil {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).WithError(err).Debug("Failed to split domain from receipt event sender")
|
2020-11-09 19:46:11 +01:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
if t.Origin != domain {
|
2022-01-31 11:48:28 +01:00
|
|
|
util.GetLogger(ctx).Debugf("Dropping receipt event where sender domain (%q) doesn't match origin (%q)", domain, t.Origin)
|
2020-11-09 19:46:11 +01:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
if err := t.processReceiptEvent(ctx, userID, roomID, "m.read", mread.Data.TS, mread.EventIDs); err != nil {
|
|
|
|
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
|
|
|
|
"sender": t.Origin,
|
|
|
|
"user_id": userID,
|
|
|
|
"room_id": roomID,
|
|
|
|
"events": mread.EventIDs,
|
2022-03-29 14:14:35 +02:00
|
|
|
}).Error("Failed to send receipt event to JetStream")
|
2020-11-09 19:46:11 +01:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2022-03-29 14:14:35 +02:00
|
|
|
case types.MSigningKeyUpdate:
|
2022-02-08 18:13:38 +01:00
|
|
|
if err := t.processSigningKeyUpdate(ctx, e); err != nil {
|
|
|
|
logrus.WithError(err).Errorf("Failed to process signing key update")
|
2021-08-17 14:44:30 +02:00
|
|
|
}
|
2022-04-06 13:11:19 +02:00
|
|
|
case gomatrixserverlib.MPresence:
|
|
|
|
if t.inboundPresenceEnabled {
|
|
|
|
if err := t.processPresence(ctx, e); err != nil {
|
|
|
|
logrus.WithError(err).Errorf("Failed to process presence update")
|
|
|
|
}
|
|
|
|
}
|
2020-03-30 17:40:28 +02:00
|
|
|
default:
|
2020-09-07 13:32:40 +02:00
|
|
|
util.GetLogger(ctx).WithField("type", e.Type).Debug("Unhandled EDU")
|
2020-03-30 17:40:28 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-04-06 13:11:19 +02:00
|
|
|
// processPresence handles m.receipt events
|
|
|
|
func (t *txnReq) processPresence(ctx context.Context, e gomatrixserverlib.EDU) error {
|
|
|
|
payload := types.Presence{}
|
|
|
|
if err := json.Unmarshal(e.Content, &payload); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
for _, content := range payload.Push {
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', content.UserID); err != nil {
|
|
|
|
continue
|
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
continue
|
|
|
|
} else if serverName != t.Origin {
|
|
|
|
continue
|
|
|
|
}
|
2022-04-06 13:11:19 +02:00
|
|
|
presence, ok := syncTypes.PresenceFromString(content.Presence)
|
|
|
|
if !ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
if err := t.producer.SendPresence(ctx, content.UserID, presence, content.StatusMsg, content.LastActiveAgo); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-02-08 18:13:38 +01:00
|
|
|
func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error {
|
2022-03-29 14:14:35 +02:00
|
|
|
var updatePayload keyapi.CrossSigningKeyUpdate
|
2022-02-08 18:13:38 +01:00
|
|
|
if err := json.Unmarshal(e.Content, &updatePayload); err != nil {
|
|
|
|
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
|
|
|
|
"user_id": updatePayload.UserID,
|
|
|
|
}).Debug("Failed to unmarshal signing key update")
|
|
|
|
return err
|
|
|
|
}
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', updatePayload.UserID); err != nil {
|
|
|
|
return nil
|
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
return nil
|
|
|
|
} else if serverName != t.Origin {
|
|
|
|
return nil
|
|
|
|
}
|
2022-02-08 18:13:38 +01:00
|
|
|
keys := gomatrixserverlib.CrossSigningKeys{}
|
|
|
|
if updatePayload.MasterKey != nil {
|
|
|
|
keys.MasterKey = *updatePayload.MasterKey
|
|
|
|
}
|
|
|
|
if updatePayload.SelfSigningKey != nil {
|
|
|
|
keys.SelfSigningKey = *updatePayload.SelfSigningKey
|
|
|
|
}
|
|
|
|
uploadReq := &keyapi.PerformUploadDeviceKeysRequest{
|
|
|
|
CrossSigningKeys: keys,
|
|
|
|
UserID: updatePayload.UserID,
|
|
|
|
}
|
|
|
|
uploadRes := &keyapi.PerformUploadDeviceKeysResponse{}
|
2022-08-11 16:29:33 +02:00
|
|
|
if err := t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2022-02-08 18:13:38 +01:00
|
|
|
if uploadRes.Error != nil {
|
|
|
|
return uploadRes.Error
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-03-29 14:14:35 +02:00
|
|
|
// processReceiptEvent sends receipt events to JetStream
|
2020-11-09 19:46:11 +01:00
|
|
|
func (t *txnReq) processReceiptEvent(ctx context.Context,
|
|
|
|
userID, roomID, receiptType string,
|
|
|
|
timestamp gomatrixserverlib.Timestamp,
|
|
|
|
eventIDs []string,
|
|
|
|
) error {
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', userID); err != nil {
|
|
|
|
return nil
|
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
return nil
|
|
|
|
} else if serverName != t.Origin {
|
|
|
|
return nil
|
|
|
|
}
|
2020-11-09 19:46:11 +01:00
|
|
|
// store every event
|
|
|
|
for _, eventID := range eventIDs {
|
2022-03-29 14:14:35 +02:00
|
|
|
if err := t.producer.SendReceipt(ctx, userID, roomID, eventID, receiptType, timestamp); err != nil {
|
2020-11-09 19:46:11 +01:00
|
|
|
return fmt.Errorf("unable to set receipt event: %w", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-09-07 13:32:40 +02:00
|
|
|
func (t *txnReq) processDeviceListUpdate(ctx context.Context, e gomatrixserverlib.EDU) {
|
2020-08-05 14:41:16 +02:00
|
|
|
var payload gomatrixserverlib.DeviceListUpdateEvent
|
|
|
|
if err := json.Unmarshal(e.Content, &payload); err != nil {
|
2020-09-07 13:32:40 +02:00
|
|
|
util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal device list update event")
|
2020-08-05 14:41:16 +02:00
|
|
|
return
|
|
|
|
}
|
2022-04-25 15:22:46 +02:00
|
|
|
if _, serverName, err := gomatrixserverlib.SplitID('@', payload.UserID); err != nil {
|
|
|
|
return
|
|
|
|
} else if serverName == t.ourServerName {
|
|
|
|
return
|
|
|
|
} else if serverName != t.Origin {
|
|
|
|
return
|
|
|
|
}
|
2022-06-15 15:27:07 +02:00
|
|
|
if err := t.producer.SendDeviceListUpdate(ctx, &payload); err != nil {
|
|
|
|
util.GetLogger(ctx).WithError(err).WithField("user_id", payload.UserID).Error("failed to InputDeviceListUpdate")
|
2020-08-05 14:41:16 +02:00
|
|
|
}
|
|
|
|
}
|