diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md index 1204582e2..8014e9414 100644 --- a/.github/PULL_REQUEST_TEMPLATE.md +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -2,6 +2,6 @@ -* [ ] Pull request includes a [sign off](https://github.com/matrix-org/dendrite/blob/master/docs/CONTRIBUTING.md#sign-off) +* [ ] Pull request includes a [sign off](https://github.com/matrix-org/dendrite/blob/main/docs/CONTRIBUTING.md#sign-off) Signed-off-by: `Your Name ` diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml index a4ef8b395..de6c79ddc 100644 --- a/.github/workflows/codeql-analysis.yml +++ b/.github/workflows/codeql-analysis.yml @@ -2,9 +2,9 @@ name: "CodeQL" on: push: - branches: [master] + branches: [main] pull_request: - branches: [master] + branches: [main] jobs: analyze: @@ -14,21 +14,21 @@ jobs: strategy: fail-fast: false matrix: - language: ['go'] + language: ["go"] steps: - - name: Checkout repository - uses: actions/checkout@v2 - with: - fetch-depth: 2 + - name: Checkout repository + uses: actions/checkout@v2 + with: + fetch-depth: 2 - - run: git checkout HEAD^2 - if: ${{ github.event_name == 'pull_request' }} + - run: git checkout HEAD^2 + if: ${{ github.event_name == 'pull_request' }} - - name: Initialize CodeQL - uses: github/codeql-action/init@v1 - with: - languages: ${{ matrix.language }} + - name: Initialize CodeQL + uses: github/codeql-action/init@v1 + with: + languages: ${{ matrix.language }} - - name: Perform CodeQL Analysis - uses: github/codeql-action/analyze@v1 + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v1 diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index ad5a2660c..4a1720295 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -2,7 +2,7 @@ name: Tests on: push: - branches: [ 'master' ] + branches: ["main"] pull_request: concurrency: @@ -33,7 +33,7 @@ jobs: path: dendrite # Attempt to check out the same branch of Complement as the PR. If it - # doesn't exist, fallback to master. + # doesn't exist, fallback to main. - name: Checkout complement shell: bash run: | @@ -68,4 +68,4 @@ jobs: name: Run Complement Tests env: COMPLEMENT_BASE_IMAGE: complement-dendrite:latest - working-directory: complement \ No newline at end of file + working-directory: complement diff --git a/build.sh b/build.sh index 8196fc653..700e6434f 100755 --- a/build.sh +++ b/build.sh @@ -7,7 +7,7 @@ if [ -d ".git" ] then export BUILD=`git rev-parse --short HEAD || ""` export BRANCH=`(git symbolic-ref --short HEAD | tr -d \/ ) || ""` - if [ "$BRANCH" = master ] + if [ "$BRANCH" = main ] then export BRANCH="" fi diff --git a/build/docker/DendriteJS.Dockerfile b/build/docker/DendriteJS.Dockerfile index e8d742b7e..5e1cffcad 100644 --- a/build/docker/DendriteJS.Dockerfile +++ b/build/docker/DendriteJS.Dockerfile @@ -9,9 +9,9 @@ FROM golang:1.14-alpine AS gobuild # Download and build dendrite WORKDIR /build -ADD https://github.com/matrix-org/dendrite/archive/master.tar.gz /build/master.tar.gz -RUN tar xvfz master.tar.gz -WORKDIR /build/dendrite-master +ADD https://github.com/matrix-org/dendrite/archive/main.tar.gz /build/main.tar.gz +RUN tar xvfz main.tar.gz +WORKDIR /build/dendrite-main RUN GOOS=js GOARCH=wasm go build -o main.wasm ./cmd/dendritejs @@ -21,7 +21,7 @@ RUN apt-get update && apt-get -y install python # Download riot-web and libp2p repos WORKDIR /build -ADD https://github.com/matrix-org/go-http-js-libp2p/archive/master.tar.gz /build/libp2p.tar.gz +ADD https://github.com/matrix-org/go-http-js-libp2p/archive/main.tar.gz /build/libp2p.tar.gz RUN tar xvfz libp2p.tar.gz ADD https://github.com/vector-im/element-web/archive/matthew/p2p.tar.gz /build/p2p.tar.gz RUN tar xvfz p2p.tar.gz @@ -31,21 +31,21 @@ WORKDIR /build/element-web-matthew-p2p RUN yarn install RUN ln -s /build/go-http-js-libp2p-master /build/element-web-matthew-p2p/node_modules/go-http-js-libp2p RUN (cd node_modules/go-http-js-libp2p && yarn install) -COPY --from=gobuild /build/dendrite-master/main.wasm ./src/vector/dendrite.wasm +COPY --from=gobuild /build/dendrite-main/main.wasm ./src/vector/dendrite.wasm # build it all RUN yarn build:p2p SHELL ["/bin/bash", "-c"] RUN echo $'\ -{ \n\ + { \n\ "default_server_config": { \n\ - "m.homeserver": { \n\ - "base_url": "https://p2p.riot.im", \n\ - "server_name": "p2p.riot.im" \n\ - }, \n\ - "m.identity_server": { \n\ - "base_url": "https://vector.im" \n\ - } \n\ + "m.homeserver": { \n\ + "base_url": "https://p2p.riot.im", \n\ + "server_name": "p2p.riot.im" \n\ + }, \n\ + "m.identity_server": { \n\ + "base_url": "https://vector.im" \n\ + } \n\ }, \n\ "disable_custom_urls": false, \n\ "disable_guests": true, \n\ @@ -55,57 +55,57 @@ RUN echo $'\ "integrations_ui_url": "https://scalar.vector.im/", \n\ "integrations_rest_url": "https://scalar.vector.im/api", \n\ "integrations_widgets_urls": [ \n\ - "https://scalar.vector.im/_matrix/integrations/v1", \n\ - "https://scalar.vector.im/api", \n\ - "https://scalar-staging.vector.im/_matrix/integrations/v1", \n\ - "https://scalar-staging.vector.im/api", \n\ - "https://scalar-staging.riot.im/scalar/api" \n\ + "https://scalar.vector.im/_matrix/integrations/v1", \n\ + "https://scalar.vector.im/api", \n\ + "https://scalar-staging.vector.im/_matrix/integrations/v1", \n\ + "https://scalar-staging.vector.im/api", \n\ + "https://scalar-staging.riot.im/scalar/api" \n\ ], \n\ "integrations_jitsi_widget_url": "https://scalar.vector.im/api/widgets/jitsi.html", \n\ "bug_report_endpoint_url": "https://riot.im/bugreports/submit", \n\ "defaultCountryCode": "GB", \n\ "showLabsSettings": false, \n\ "features": { \n\ - "feature_pinning": "labs", \n\ - "feature_custom_status": "labs", \n\ - "feature_custom_tags": "labs", \n\ - "feature_state_counters": "labs" \n\ + "feature_pinning": "labs", \n\ + "feature_custom_status": "labs", \n\ + "feature_custom_tags": "labs", \n\ + "feature_state_counters": "labs" \n\ }, \n\ "default_federate": true, \n\ "default_theme": "light", \n\ "roomDirectory": { \n\ - "servers": [ \n\ - "matrix.org" \n\ - ] \n\ + "servers": [ \n\ + "matrix.org" \n\ + ] \n\ }, \n\ "welcomeUserId": "", \n\ "piwik": { \n\ - "url": "https://piwik.riot.im/", \n\ - "whitelistedHSUrls": ["https://matrix.org"], \n\ - "whitelistedISUrls": ["https://vector.im", "https://matrix.org"], \n\ - "siteId": 1 \n\ + "url": "https://piwik.riot.im/", \n\ + "whitelistedHSUrls": ["https://matrix.org"], \n\ + "whitelistedISUrls": ["https://vector.im", "https://matrix.org"], \n\ + "siteId": 1 \n\ }, \n\ "enable_presence_by_hs_url": { \n\ - "https://matrix.org": false, \n\ - "https://matrix-client.matrix.org": false \n\ + "https://matrix.org": false, \n\ + "https://matrix-client.matrix.org": false \n\ }, \n\ "settingDefaults": { \n\ - "breadcrumbs": true \n\ + "breadcrumbs": true \n\ } \n\ -}' > webapp/config.json + }' > webapp/config.json FROM nginx # Add "Service-Worker-Allowed: /" header so the worker can sniff traffic on this domain rather # than just the path this gets hosted under. NB this newline echo syntax only works on bash. SHELL ["/bin/bash", "-c"] RUN echo $'\ -server { \n\ + server { \n\ listen 80; \n\ add_header \'Service-Worker-Allowed\' \'/\'; \n\ location / { \n\ - root /usr/share/nginx/html; \n\ - index index.html index.htm; \n\ + root /usr/share/nginx/html; \n\ + index index.html index.htm; \n\ } \n\ -}' > /etc/nginx/conf.d/default.conf + }' > /etc/nginx/conf.d/default.conf RUN sed -i 's/}/ application\/wasm wasm;\n}/g' /etc/nginx/mime.types COPY --from=jsbuild /build/element-web-matthew-p2p/webapp /usr/share/nginx/html diff --git a/build/scripts/Complement.Dockerfile b/build/scripts/Complement.Dockerfile index 401695abf..a54fab1d4 100644 --- a/build/scripts/Complement.Dockerfile +++ b/build/scripts/Complement.Dockerfile @@ -2,6 +2,10 @@ FROM golang:1.16-stretch as build RUN apt-get update && apt-get install -y sqlite3 WORKDIR /build +# we will dump the binaries and config file to this location to ensure any local untracked files +# that come from the COPY . . file don't contaminate the build +RUN mkdir /dendrite + # Utilise Docker caching when downloading dependencies, this stops us needlessly # downloading dependencies every time. COPY go.mod . @@ -9,9 +13,11 @@ COPY go.sum . RUN go mod download COPY . . -RUN go build ./cmd/dendrite-monolith-server -RUN go build ./cmd/generate-keys -RUN go build ./cmd/generate-config +RUN go build -o /dendrite ./cmd/dendrite-monolith-server +RUN go build -o /dendrite ./cmd/generate-keys +RUN go build -o /dendrite ./cmd/generate-config + +WORKDIR /dendrite RUN ./generate-keys --private-key matrix_key.pem ENV SERVER_NAME=localhost diff --git a/clientapi/routing/membership.go b/clientapi/routing/membership.go index 4ce820797..58f187608 100644 --- a/clientapi/routing/membership.go +++ b/clientapi/routing/membership.go @@ -17,6 +17,7 @@ package routing import ( "context" "errors" + "fmt" "net/http" "time" @@ -459,13 +460,7 @@ func SendForget( if membershipRes.IsInRoom { return util.JSONResponse{ Code: http.StatusBadRequest, - JSON: jsonerror.Forbidden("user is still a member of the room"), - } - } - if !membershipRes.HasBeenInRoom { - return util.JSONResponse{ - Code: http.StatusBadRequest, - JSON: jsonerror.Forbidden("user did not belong to room"), + JSON: jsonerror.Unknown(fmt.Sprintf("User %s is in room %s", device.UserID, roomID)), } } diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md index ea4b2b27d..fe7127c76 100644 --- a/docs/CONTRIBUTING.md +++ b/docs/CONTRIBUTING.md @@ -37,7 +37,7 @@ If a job fails, click the "details" button and you should be taken to the job's logs. ![Click the details button on the failing build -step](https://raw.githubusercontent.com/matrix-org/dendrite/master/docs/images/details-button-location.jpg) +step](https://raw.githubusercontent.com/matrix-org/dendrite/main/docs/images/details-button-location.jpg) Scroll down to the failing step and you should see some log output. Scan the logs until you find what it's complaining about, fix it, submit a new commit, @@ -57,7 +57,7 @@ significant amount of CPU and RAM. Once the code builds, run [Sytest](https://github.com/matrix-org/sytest) according to the guide in -[docs/sytest.md](https://github.com/matrix-org/dendrite/blob/master/docs/sytest.md#using-a-sytest-docker-image) +[docs/sytest.md](https://github.com/matrix-org/dendrite/blob/main/docs/sytest.md#using-a-sytest-docker-image) so you can see whether something is being broken and whether there are newly passing tests. @@ -94,4 +94,4 @@ For more general questions please use We ask that everyone who contributes to the project signs off their contributions, in accordance with the -[DCO](https://github.com/matrix-org/matrix-doc/blob/master/CONTRIBUTING.rst#sign-off). +[DCO](https://github.com/matrix-org/matrix-doc/blob/main/CONTRIBUTING.rst#sign-off). diff --git a/docs/p2p.md b/docs/p2p.md index e858ba114..4e9a50524 100644 --- a/docs/p2p.md +++ b/docs/p2p.md @@ -6,7 +6,7 @@ These are the instructions for setting up P2P Dendrite, current as of May 2020. #### Build -- The `master` branch has a WASM-only binary for dendrite: `./cmd/dendritejs`. +- The `main` branch has a WASM-only binary for dendrite: `./cmd/dendritejs`. - Build it and copy assets to riot-web. ``` $ ./build-dendritejs.sh diff --git a/eduserver/api/input.go b/eduserver/api/input.go index 2fa253f4d..2aab107b2 100644 --- a/eduserver/api/input.go +++ b/eduserver/api/input.go @@ -100,10 +100,4 @@ type EDUServerInputAPI interface { request *InputReceiptEventRequest, response *InputReceiptEventResponse, ) error - - InputCrossSigningKeyUpdate( - ctx context.Context, - request *InputCrossSigningKeyUpdateRequest, - response *InputCrossSigningKeyUpdateResponse, - ) error } diff --git a/eduserver/eduserver.go b/eduserver/eduserver.go index febcf2864..9b7e21651 100644 --- a/eduserver/eduserver.go +++ b/eduserver/eduserver.go @@ -51,7 +51,6 @@ func NewInternalAPI( OutputTypingEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), OutputSendToDeviceEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), OutputReceiptEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), - OutputKeyChangeEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent), ServerName: cfg.Matrix.ServerName, } } diff --git a/eduserver/input/input.go b/eduserver/input/input.go index 4f8ab3e34..e58f0dd34 100644 --- a/eduserver/input/input.go +++ b/eduserver/input/input.go @@ -23,7 +23,6 @@ import ( "github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/cache" - keyapi "github.com/matrix-org/dendrite/keyserver/api" userapi "github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/gomatrixserverlib" "github.com/nats-io/nats.go" @@ -40,8 +39,6 @@ type EDUServerInputAPI struct { OutputSendToDeviceEventTopic string // The kafka topic to output new receipt events to OutputReceiptEventTopic string - // The kafka topic to output new key change events to - OutputKeyChangeEventTopic string // kafka producer JetStream nats.JetStreamContext // Internal user query API @@ -80,34 +77,6 @@ func (t *EDUServerInputAPI) InputSendToDeviceEvent( return t.sendToDeviceEvent(ise) } -// InputCrossSigningKeyUpdate implements api.EDUServerInputAPI -func (t *EDUServerInputAPI) InputCrossSigningKeyUpdate( - ctx context.Context, - request *api.InputCrossSigningKeyUpdateRequest, - response *api.InputCrossSigningKeyUpdateResponse, -) error { - eventJSON, err := json.Marshal(&keyapi.DeviceMessage{ - Type: keyapi.TypeCrossSigningUpdate, - OutputCrossSigningKeyUpdate: &api.OutputCrossSigningKeyUpdate{ - CrossSigningKeyUpdate: request.CrossSigningKeyUpdate, - }, - }) - if err != nil { - return err - } - - logrus.WithFields(logrus.Fields{ - "user_id": request.UserID, - }).Tracef("Producing to topic '%s'", t.OutputKeyChangeEventTopic) - - _, err = t.JetStream.PublishMsg(&nats.Msg{ - Subject: t.OutputKeyChangeEventTopic, - Header: nats.Header{}, - Data: eventJSON, - }) - return err -} - func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error { ev := &api.TypingEvent{ Type: gomatrixserverlib.MTyping, diff --git a/eduserver/inthttp/client.go b/eduserver/inthttp/client.go index 9a6f483c2..0690ed827 100644 --- a/eduserver/inthttp/client.go +++ b/eduserver/inthttp/client.go @@ -12,10 +12,9 @@ import ( // HTTP paths for the internal HTTP APIs const ( - EDUServerInputTypingEventPath = "/eduserver/input" - EDUServerInputSendToDeviceEventPath = "/eduserver/sendToDevice" - EDUServerInputReceiptEventPath = "/eduserver/receipt" - EDUServerInputCrossSigningKeyUpdatePath = "/eduserver/crossSigningKeyUpdate" + EDUServerInputTypingEventPath = "/eduserver/input" + EDUServerInputSendToDeviceEventPath = "/eduserver/sendToDevice" + EDUServerInputReceiptEventPath = "/eduserver/receipt" ) // NewEDUServerClient creates a EDUServerInputAPI implemented by talking to a HTTP POST API. @@ -69,16 +68,3 @@ func (h *httpEDUServerInputAPI) InputReceiptEvent( apiURL := h.eduServerURL + EDUServerInputReceiptEventPath return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response) } - -// InputCrossSigningKeyUpdate implements EDUServerInputAPI -func (h *httpEDUServerInputAPI) InputCrossSigningKeyUpdate( - ctx context.Context, - request *api.InputCrossSigningKeyUpdateRequest, - response *api.InputCrossSigningKeyUpdateResponse, -) error { - span, ctx := opentracing.StartSpanFromContext(ctx, "InputCrossSigningKeyUpdate") - defer span.Finish() - - apiURL := h.eduServerURL + EDUServerInputCrossSigningKeyUpdatePath - return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response) -} diff --git a/eduserver/inthttp/server.go b/eduserver/inthttp/server.go index a50ca84f9..a34943750 100644 --- a/eduserver/inthttp/server.go +++ b/eduserver/inthttp/server.go @@ -51,17 +51,4 @@ func AddRoutes(t api.EDUServerInputAPI, internalAPIMux *mux.Router) { return util.JSONResponse{Code: http.StatusOK, JSON: &response} }), ) - internalAPIMux.Handle(EDUServerInputCrossSigningKeyUpdatePath, - httputil.MakeInternalAPI("inputCrossSigningKeyUpdate", func(req *http.Request) util.JSONResponse { - var request api.InputCrossSigningKeyUpdateRequest - var response api.InputCrossSigningKeyUpdateResponse - if err := json.NewDecoder(req.Body).Decode(&request); err != nil { - return util.MessageResponse(http.StatusBadRequest, err.Error()) - } - if err := t.InputCrossSigningKeyUpdate(req.Context(), &request, &response); err != nil { - return util.ErrorResponse(err) - } - return util.JSONResponse{Code: http.StatusOK, JSON: &response} - }), - ) } diff --git a/federationapi/consumers/eduserver.go b/federationapi/consumers/eduserver.go index 22fedbeb4..1f81fa258 100644 --- a/federationapi/consumers/eduserver.go +++ b/federationapi/consumers/eduserver.go @@ -134,7 +134,7 @@ func (t *OutputEDUConsumer) onSendToDeviceEvent(ctx context.Context, msg *nats.M return true } - log.Infof("Sending send-to-device message into %q destination queue", destServerName) + log.Debugf("Sending send-to-device message into %q destination queue", destServerName) if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil { log.WithError(err).Error("failed to send EDU") return false diff --git a/federationapi/consumers/keychange.go b/federationapi/consumers/keychange.go index 1ec9f4c18..22dbc32da 100644 --- a/federationapi/consumers/keychange.go +++ b/federationapi/consumers/keychange.go @@ -127,6 +127,9 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { return true } + if len(destinations) == 0 { + return true + } // Pack the EDU and marshal it edu := &gomatrixserverlib.EDU{ Type: gomatrixserverlib.MDeviceListUpdate, @@ -146,7 +149,7 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { return true } - logger.Infof("Sending device list update message to %q", destinations) + logger.Debugf("Sending device list update message to %q", destinations) err = t.queues.SendEDU(edu, t.serverName, destinations) return err == nil } @@ -181,6 +184,10 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { return true } + if len(destinations) == 0 { + return true + } + // Pack the EDU and marshal it edu := &gomatrixserverlib.EDU{ Type: eduserverAPI.MSigningKeyUpdate, @@ -191,7 +198,7 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { return true } - logger.Infof("Sending cross-signing update message to %q", destinations) + logger.Debugf("Sending cross-signing update message to %q", destinations) err = t.queues.SendEDU(edu, t.serverName, destinations) return err == nil } diff --git a/federationapi/consumers/roomserver.go b/federationapi/consumers/roomserver.go index e9862000a..ac29f930b 100644 --- a/federationapi/consumers/roomserver.go +++ b/federationapi/consumers/roomserver.go @@ -114,11 +114,6 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) } } - case api.OutputTypeNewInviteEvent: - log.WithField("type", output.Type).Debug( - "received new invite, send device keys", - ) - case api.OutputTypeNewInboundPeek: if err := s.processInboundPeek(*output.NewInboundPeek); err != nil { log.WithFields(log.Fields{ diff --git a/federationapi/internal/perform.go b/federationapi/internal/perform.go index 7850f206c..c51ecf146 100644 --- a/federationapi/internal/perform.go +++ b/federationapi/internal/perform.go @@ -201,7 +201,6 @@ func (r *FederationInternalAPI) performJoinUsingServer( context.Background(), serverName, event, - respMakeJoin.RoomVersion, ) if err != nil { r.statistics.ForServer(serverName).Failure() @@ -209,9 +208,11 @@ func (r *FederationInternalAPI) performJoinUsingServer( } r.statistics.ForServer(serverName).Success() + authEvents := respSendJoin.AuthEvents.UntrustedEvents(respMakeJoin.RoomVersion) + // Sanity-check the join response to ensure that it has a create // event, that the room version is known, etc. - if err = sanityCheckAuthChain(respSendJoin.AuthEvents); err != nil { + if err = sanityCheckAuthChain(authEvents); err != nil { return fmt.Errorf("sanityCheckAuthChain: %w", err) } @@ -225,6 +226,7 @@ func (r *FederationInternalAPI) performJoinUsingServer( var respState *gomatrixserverlib.RespState respState, err = respSendJoin.Check( context.Background(), + respMakeJoin.RoomVersion, r.keyRing, event, federatedAuthProvider(ctx, r.federation, r.keyRing, serverName), @@ -392,12 +394,13 @@ func (r *FederationInternalAPI) performOutboundPeekUsingServer( ctx = context.Background() respState := respPeek.ToRespState() + authEvents := respState.AuthEvents.UntrustedEvents(respPeek.RoomVersion) // authenticate the state returned (check its auth events etc) // the equivalent of CheckSendJoinResponse() - if err = sanityCheckAuthChain(respState.AuthEvents); err != nil { + if err = sanityCheckAuthChain(authEvents); err != nil { return fmt.Errorf("sanityCheckAuthChain: %w", err) } - if err = respState.Check(ctx, r.keyRing, federatedAuthProvider(ctx, r.federation, r.keyRing, serverName)); err != nil { + if err = respState.Check(ctx, respPeek.RoomVersion, r.keyRing, federatedAuthProvider(ctx, r.federation, r.keyRing, serverName)); err != nil { return fmt.Errorf("error checking state returned from peeking: %w", err) } @@ -549,10 +552,15 @@ func (r *FederationInternalAPI) PerformInvite( inviteRes, err := r.federation.SendInviteV2(ctx, destination, inviteReq) if err != nil { - return fmt.Errorf("r.federation.SendInviteV2: %w", err) + return fmt.Errorf("r.federation.SendInviteV2: failed to send invite: %w", err) } + logrus.Infof("GOT INVITE RESPONSE %s", string(inviteRes.Event)) - response.Event = inviteRes.Event.Headered(request.RoomVersion) + inviteEvent, err := inviteRes.Event.UntrustedEvent(request.RoomVersion) + if err != nil { + return fmt.Errorf("r.federation.SendInviteV2 failed to decode event response: %w", err) + } + response.Event = inviteEvent.Headered(request.RoomVersion) return nil } diff --git a/federationapi/inthttp/client.go b/federationapi/inthttp/client.go index a65df906f..f9b2a33d2 100644 --- a/federationapi/inthttp/client.go +++ b/federationapi/inthttp/client.go @@ -387,14 +387,7 @@ func (h *httpFederationInternalAPI) LookupMissingEvents( if request.Err != nil { return res, request.Err } - res.Events = make([]*gomatrixserverlib.Event, 0, len(request.Res.Events)) - for _, js := range request.Res.Events { - ev, err := gomatrixserverlib.NewEventFromUntrustedJSON(js, roomVersion) - if err != nil { - return res, err - } - res.Events = append(res.Events, ev) - } + res.Events = request.Res.Events return res, nil } diff --git a/federationapi/queue/destinationqueue.go b/federationapi/queue/destinationqueue.go index 1306e8588..09814b31f 100644 --- a/federationapi/queue/destinationqueue.go +++ b/federationapi/queue/destinationqueue.go @@ -297,7 +297,7 @@ func (oq *destinationQueue) backgroundSend() { // We haven't backed off yet, so wait for the suggested amount of // time. duration := time.Until(*until) - logrus.Warnf("Backing off %q for %s", oq.destination, duration) + logrus.Debugf("Backing off %q for %s", oq.destination, duration) oq.backingOff.Store(true) destinationQueueBackingOff.Inc() select { diff --git a/federationapi/routing/eventauth.go b/federationapi/routing/eventauth.go index d92b66f4b..0a03a0cb4 100644 --- a/federationapi/routing/eventauth.go +++ b/federationapi/routing/eventauth.go @@ -65,7 +65,7 @@ func GetEventAuth( return util.JSONResponse{ Code: http.StatusOK, JSON: gomatrixserverlib.RespEventAuth{ - AuthEvents: gomatrixserverlib.UnwrapEventHeaders(response.AuthChainEvents), + AuthEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(response.AuthChainEvents), }, } } diff --git a/federationapi/routing/invite.go b/federationapi/routing/invite.go index 468659651..58bf99f4a 100644 --- a/federationapi/routing/invite.go +++ b/federationapi/routing/invite.go @@ -178,12 +178,12 @@ func processInvite( if isInviteV2 { return util.JSONResponse{ Code: http.StatusOK, - JSON: gomatrixserverlib.RespInviteV2{Event: &signedEvent}, + JSON: gomatrixserverlib.RespInviteV2{Event: signedEvent.JSON()}, } } else { return util.JSONResponse{ Code: http.StatusOK, - JSON: gomatrixserverlib.RespInvite{Event: &signedEvent}, + JSON: gomatrixserverlib.RespInvite{Event: signedEvent.JSON()}, } } default: diff --git a/federationapi/routing/join.go b/federationapi/routing/join.go index 7f8d31505..495b8c914 100644 --- a/federationapi/routing/join.go +++ b/federationapi/routing/join.go @@ -351,8 +351,8 @@ func SendJoin( return util.JSONResponse{ Code: http.StatusOK, JSON: gomatrixserverlib.RespSendJoin{ - StateEvents: gomatrixserverlib.UnwrapEventHeaders(stateAndAuthChainResponse.StateEvents), - AuthEvents: gomatrixserverlib.UnwrapEventHeaders(stateAndAuthChainResponse.AuthChainEvents), + StateEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(stateAndAuthChainResponse.StateEvents), + AuthEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(stateAndAuthChainResponse.AuthChainEvents), Origin: cfg.Matrix.ServerName, }, } diff --git a/federationapi/routing/missingevents.go b/federationapi/routing/missingevents.go index f79a2d2d8..dd3df7aa9 100644 --- a/federationapi/routing/missingevents.go +++ b/federationapi/routing/missingevents.go @@ -62,7 +62,7 @@ func GetMissingEvents( eventsResponse.Events = filterEvents(eventsResponse.Events, roomID) resp := gomatrixserverlib.RespMissingEvents{ - Events: gomatrixserverlib.UnwrapEventHeaders(eventsResponse.Events), + Events: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(eventsResponse.Events), } return util.JSONResponse{ diff --git a/federationapi/routing/peek.go b/federationapi/routing/peek.go index 511329997..827d1116d 100644 --- a/federationapi/routing/peek.go +++ b/federationapi/routing/peek.go @@ -88,8 +88,8 @@ func Peek( } respPeek := gomatrixserverlib.RespPeek{ - StateEvents: gomatrixserverlib.UnwrapEventHeaders(response.StateEvents), - AuthEvents: gomatrixserverlib.UnwrapEventHeaders(response.AuthChainEvents), + StateEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(response.StateEvents), + AuthEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(response.AuthChainEvents), RoomVersion: response.RoomVersion, LatestEvent: response.LatestEvent.Unwrap(), RenewalInterval: renewalInterval, diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go index 524fd510e..dd4fe13a8 100644 --- a/federationapi/routing/send.go +++ b/federationapi/routing/send.go @@ -382,20 +382,8 @@ func (t *txnReq) processEDUs(ctx context.Context) { } } case eduserverAPI.MSigningKeyUpdate: - var updatePayload eduserverAPI.CrossSigningKeyUpdate - if err := json.Unmarshal(e.Content, &updatePayload); err != nil { - util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{ - "user_id": updatePayload.UserID, - }).Debug("Failed to send signing key update to edu server") - continue - } - inputReq := &eduserverAPI.InputCrossSigningKeyUpdateRequest{ - CrossSigningKeyUpdate: updatePayload, - } - inputRes := &eduserverAPI.InputCrossSigningKeyUpdateResponse{} - if err := t.eduAPI.InputCrossSigningKeyUpdate(ctx, inputReq, inputRes); err != nil { - util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal cross-signing update") - continue + if err := t.processSigningKeyUpdate(ctx, e); err != nil { + logrus.WithError(err).Errorf("Failed to process signing key update") } default: util.GetLogger(ctx).WithField("type", e.Type).Debug("Unhandled EDU") @@ -403,6 +391,34 @@ func (t *txnReq) processEDUs(ctx context.Context) { } } +func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error { + var updatePayload eduserverAPI.CrossSigningKeyUpdate + 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 + } + + 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{} + t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes) + if uploadRes.Error != nil { + return uploadRes.Error + } + return nil +} + // processReceiptEvent sends receipt events to the edu server func (t *txnReq) processReceiptEvent(ctx context.Context, userID, roomID, receiptType string, diff --git a/federationapi/routing/state.go b/federationapi/routing/state.go index 128df6187..37cbb9d1e 100644 --- a/federationapi/routing/state.go +++ b/federationapi/routing/state.go @@ -35,12 +35,15 @@ func GetState( return *err } - state, err := getState(ctx, request, rsAPI, roomID, eventID) + stateEvents, authChain, err := getState(ctx, request, rsAPI, roomID, eventID) if err != nil { return *err } - return util.JSONResponse{Code: http.StatusOK, JSON: state} + return util.JSONResponse{Code: http.StatusOK, JSON: &gomatrixserverlib.RespState{ + AuthEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(authChain), + StateEvents: gomatrixserverlib.NewEventJSONsFromHeaderedEvents(stateEvents), + }} } // GetStateIDs returns state event IDs & auth event IDs for the roomID, eventID @@ -55,13 +58,13 @@ func GetStateIDs( return *err } - state, err := getState(ctx, request, rsAPI, roomID, eventID) + stateEvents, authEvents, err := getState(ctx, request, rsAPI, roomID, eventID) if err != nil { return *err } - stateEventIDs := getIDsFromEvent(state.StateEvents) - authEventIDs := getIDsFromEvent(state.AuthEvents) + stateEventIDs := getIDsFromEvent(stateEvents) + authEventIDs := getIDsFromEvent(authEvents) return util.JSONResponse{Code: http.StatusOK, JSON: gomatrixserverlib.RespStateIDs{ StateEventIDs: stateEventIDs, @@ -97,18 +100,18 @@ func getState( rsAPI api.RoomserverInternalAPI, roomID string, eventID string, -) (*gomatrixserverlib.RespState, *util.JSONResponse) { +) (stateEvents, authEvents []*gomatrixserverlib.HeaderedEvent, errRes *util.JSONResponse) { event, resErr := fetchEvent(ctx, rsAPI, eventID) if resErr != nil { - return nil, resErr + return nil, nil, resErr } if event.RoomID() != roomID { - return nil, &util.JSONResponse{Code: http.StatusNotFound, JSON: jsonerror.NotFound("event does not belong to this room")} + return nil, nil, &util.JSONResponse{Code: http.StatusNotFound, JSON: jsonerror.NotFound("event does not belong to this room")} } resErr = allowedToSeeEvent(ctx, request.Origin(), rsAPI, eventID) if resErr != nil { - return nil, resErr + return nil, nil, resErr } var response api.QueryStateAndAuthChainResponse @@ -123,20 +126,17 @@ func getState( ) if err != nil { resErr := util.ErrorResponse(err) - return nil, &resErr + return nil, nil, &resErr } if !response.RoomExists { - return nil, &util.JSONResponse{Code: http.StatusNotFound, JSON: nil} + return nil, nil, &util.JSONResponse{Code: http.StatusNotFound, JSON: nil} } - return &gomatrixserverlib.RespState{ - StateEvents: gomatrixserverlib.UnwrapEventHeaders(response.StateEvents), - AuthEvents: gomatrixserverlib.UnwrapEventHeaders(response.AuthChainEvents), - }, nil + return response.StateEvents, response.AuthChainEvents, nil } -func getIDsFromEvent(events []*gomatrixserverlib.Event) []string { +func getIDsFromEvent(events []*gomatrixserverlib.HeaderedEvent) []string { IDs := make([]string, len(events)) for i := range events { IDs[i] = events[i].EventID() diff --git a/federationapi/routing/threepid.go b/federationapi/routing/threepid.go index b16c68d25..8ae7130c3 100644 --- a/federationapi/routing/threepid.go +++ b/federationapi/routing/threepid.go @@ -170,13 +170,18 @@ func ExchangeThirdPartyInvite( util.GetLogger(httpReq.Context()).WithError(err).Error("federation.SendInvite failed") return jsonerror.InternalServerError() } + inviteEvent, err := signedEvent.Event.UntrustedEvent(verRes.RoomVersion) + if err != nil { + util.GetLogger(httpReq.Context()).WithError(err).Error("federation.SendInvite failed") + return jsonerror.InternalServerError() + } // Send the event to the roomserver if err = api.SendEvents( httpReq.Context(), rsAPI, api.KindNew, []*gomatrixserverlib.HeaderedEvent{ - signedEvent.Event.Headered(verRes.RoomVersion), + inviteEvent.Headered(verRes.RoomVersion), }, request.Origin(), cfg.Matrix.ServerName, diff --git a/go.mod b/go.mod index fc18ce07e..a1dc04084 100644 --- a/go.mod +++ b/go.mod @@ -41,7 +41,7 @@ require ( github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4 github.com/matrix-org/go-sqlite3-js v0.0.0-20210709140738-b0d1ba599a6d github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 - github.com/matrix-org/gomatrixserverlib v0.0.0-20220204110702-c559d2019275 + github.com/matrix-org/gomatrixserverlib v0.0.0-20220209202448-9805ef634335 github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4 github.com/mattn/go-sqlite3 v1.14.10 @@ -59,17 +59,17 @@ require ( github.com/prometheus/common v0.32.1 // indirect github.com/prometheus/procfs v0.7.3 // indirect github.com/sirupsen/logrus v1.8.1 - github.com/tidwall/gjson v1.13.0 + github.com/tidwall/gjson v1.14.0 github.com/tidwall/sjson v1.2.4 github.com/uber/jaeger-client-go v2.30.0+incompatible github.com/uber/jaeger-lib v2.4.1+incompatible github.com/yggdrasil-network/yggdrasil-go v0.4.2 go.uber.org/atomic v1.9.0 - golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8 + golang.org/x/crypto v0.0.0-20220209195652-db638375bc3a golang.org/x/image v0.0.0-20211028202545-6944b10bf410 golang.org/x/mobile v0.0.0-20220112015953-858099ff7816 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd - golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 // indirect + golang.org/x/sys v0.0.0-20220207234003-57398862261d // indirect golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 gopkg.in/h2non/bimg.v1 v1.1.5 gopkg.in/yaml.v2 v2.4.0 diff --git a/go.sum b/go.sum index 3f8a99f48..1483c792f 100644 --- a/go.sum +++ b/go.sum @@ -983,8 +983,8 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20210709140738-b0d1ba599a6d/go.mod h1 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0= github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5dLDCud4r0r55eP4j9FuUNpl60Gmntcop4= github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s= -github.com/matrix-org/gomatrixserverlib v0.0.0-20220204110702-c559d2019275 h1:f6Hh7D3EOTl1uUr76FiyHNA1h4pKBhcVUtyHbxn0hKA= -github.com/matrix-org/gomatrixserverlib v0.0.0-20220204110702-c559d2019275/go.mod h1:qFvhfbQ5orQxlH9vCiFnP4dW27xxnWHdNUBKyj/fbiY= +github.com/matrix-org/gomatrixserverlib v0.0.0-20220209202448-9805ef634335 h1:xzK9Q9VGqsZNGx5ANFOCWkJ8R+W1J2BOguxsVZw6m8M= +github.com/matrix-org/gomatrixserverlib v0.0.0-20220209202448-9805ef634335/go.mod h1:qFvhfbQ5orQxlH9vCiFnP4dW27xxnWHdNUBKyj/fbiY= github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf h1:/nqfHUdQHr3WVdbZieaYFvHF1rin5pvDTa/NOZ/qCyE= github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf/go.mod h1:r6dsL+ylE0yXe/7zh8y/Bdh6aBYI1r+u4yZni9A4iyk= github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U= @@ -1363,8 +1363,8 @@ github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpP github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA= github.com/tchap/go-patricia v2.2.6+incompatible/go.mod h1:bmLyhP68RS6kStMGxByiQ23RP/odRBOTVjwp2cDyi6I= github.com/tidwall/gjson v1.12.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= -github.com/tidwall/gjson v1.13.0 h1:3TFY9yxOQShrvmjdM76K+jc66zJeT6D3/VFFYCGQf7M= -github.com/tidwall/gjson v1.13.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/gjson v1.14.0 h1:6aeJ0bzojgWLa82gDQHcx3S0Lr/O51I9bJ5nv6JFx5w= +github.com/tidwall/gjson v1.14.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs= @@ -1509,8 +1509,8 @@ golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf/go.mod h1:P+XmwS30IXTQdn5 golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8= golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8 h1:kACShD3qhmr/3rLmg1yXyt+N4HcwutKyPRB93s54TIU= -golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= +golang.org/x/crypto v0.0.0-20220209195652-db638375bc3a h1:atOEWVSedO4ksXBe/UrlbSLVxQQ9RxM/tT2Jy10IaHo= +golang.org/x/crypto v0.0.0-20220209195652-db638375bc3a/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= @@ -1734,8 +1734,8 @@ golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 h1:XfKQ4OlFl8okEOr5UvAqFRVj8pY/4yfcXrddB8qAbU0= -golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220207234003-57398862261d h1:Bm7BNOQt2Qv7ZqysjeLjgCBanX+88Z/OtdvsrEv1Djc= +golang.org/x/sys v0.0.0-20220207234003-57398862261d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY= diff --git a/keyserver/consumers/cross_signing.go b/keyserver/consumers/cross_signing.go deleted file mode 100644 index aae69e960..000000000 --- a/keyserver/consumers/cross_signing.go +++ /dev/null @@ -1,123 +0,0 @@ -// Copyright 2021 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 consumers - -import ( - "context" - "encoding/json" - - "github.com/matrix-org/dendrite/keyserver/api" - "github.com/matrix-org/dendrite/keyserver/storage" - "github.com/matrix-org/dendrite/setup/config" - "github.com/matrix-org/dendrite/setup/jetstream" - "github.com/matrix-org/dendrite/setup/process" - "github.com/matrix-org/gomatrixserverlib" - "github.com/nats-io/nats.go" - "github.com/sirupsen/logrus" -) - -type OutputCrossSigningKeyUpdateConsumer struct { - ctx context.Context - keyDB storage.Database - keyAPI api.KeyInternalAPI - serverName string - jetstream nats.JetStreamContext - durable string - topic string -} - -func NewOutputCrossSigningKeyUpdateConsumer( - process *process.ProcessContext, - cfg *config.Dendrite, - js nats.JetStreamContext, - keyDB storage.Database, - keyAPI api.KeyInternalAPI, -) *OutputCrossSigningKeyUpdateConsumer { - // The keyserver both produces and consumes on the TopicOutputKeyChangeEvent - // topic. We will only produce events where the UserID matches our server name, - // and we will only consume events where the UserID does NOT match our server - // name (because the update came from a remote server). - s := &OutputCrossSigningKeyUpdateConsumer{ - ctx: process.Context(), - keyDB: keyDB, - jetstream: js, - durable: cfg.Global.JetStream.Durable("KeyServerCrossSigningConsumer"), - topic: cfg.Global.JetStream.TopicFor(jetstream.OutputKeyChangeEvent), - keyAPI: keyAPI, - serverName: string(cfg.Global.ServerName), - } - - return s -} - -func (s *OutputCrossSigningKeyUpdateConsumer) Start() error { - return jetstream.JetStreamConsumer( - s.ctx, s.jetstream, s.topic, s.durable, s.onMessage, - nats.DeliverAll(), nats.ManualAck(), - ) -} - -// onMessage is called in response to a message received on the -// key change events topic from the key server. -func (t *OutputCrossSigningKeyUpdateConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool { - var m api.DeviceMessage - if err := json.Unmarshal(msg.Data, &m); err != nil { - logrus.WithError(err).Errorf("failed to read device message from key change topic") - return true - } - if m.OutputCrossSigningKeyUpdate == nil { - // This probably shouldn't happen but stops us from panicking if we come - // across an update that doesn't satisfy either types. - return true - } - switch m.Type { - case api.TypeCrossSigningUpdate: - return t.onCrossSigningMessage(m) - default: - return true - } -} - -func (s *OutputCrossSigningKeyUpdateConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { - output := m.CrossSigningKeyUpdate - _, host, err := gomatrixserverlib.SplitID('@', output.UserID) - if err != nil { - logrus.WithError(err).Errorf("eduserver output log: user ID parse failure") - return true - } - if host == gomatrixserverlib.ServerName(s.serverName) { - // Ignore any messages that contain information about our own users, as - // they already originated from this server. - return true - } - uploadReq := &api.PerformUploadDeviceKeysRequest{ - UserID: output.UserID, - } - if output.MasterKey != nil { - uploadReq.MasterKey = *output.MasterKey - } - if output.SelfSigningKey != nil { - uploadReq.SelfSigningKey = *output.SelfSigningKey - } - uploadRes := &api.PerformUploadDeviceKeysResponse{} - s.keyAPI.PerformUploadDeviceKeys(context.TODO(), uploadReq, uploadRes) - if uploadRes.Error != nil { - // If the error is due to a missing or invalid parameter then we'd might - // as well just acknowledge the message, because otherwise otherwise we'll - // just keep getting delivered a faulty message over and over again. - return uploadRes.Error.IsMissingParam || uploadRes.Error.IsInvalidParam - } - return true -} diff --git a/keyserver/internal/cross_signing.go b/keyserver/internal/cross_signing.go index 1e1871b8b..bfb2037f8 100644 --- a/keyserver/internal/cross_signing.go +++ b/keyserver/internal/cross_signing.go @@ -219,25 +219,23 @@ func (a *KeyInternalAPI) PerformUploadDeviceKeys(ctx context.Context, req *api.P } // Finally, generate a notification that we updated the keys. - if _, host, err := gomatrixserverlib.SplitID('@', req.UserID); err == nil && host == a.ThisServer { - update := eduserverAPI.CrossSigningKeyUpdate{ - UserID: req.UserID, - } - if mk, ok := byPurpose[gomatrixserverlib.CrossSigningKeyPurposeMaster]; ok { - update.MasterKey = &mk - } - if ssk, ok := byPurpose[gomatrixserverlib.CrossSigningKeyPurposeSelfSigning]; ok { - update.SelfSigningKey = &ssk - } - if update.MasterKey == nil && update.SelfSigningKey == nil { - return - } - if err := a.Producer.ProduceSigningKeyUpdate(update); err != nil { - res.Error = &api.KeyError{ - Err: fmt.Sprintf("a.Producer.ProduceSigningKeyUpdate: %s", err), - } - return + update := eduserverAPI.CrossSigningKeyUpdate{ + UserID: req.UserID, + } + if mk, ok := byPurpose[gomatrixserverlib.CrossSigningKeyPurposeMaster]; ok { + update.MasterKey = &mk + } + if ssk, ok := byPurpose[gomatrixserverlib.CrossSigningKeyPurposeSelfSigning]; ok { + update.SelfSigningKey = &ssk + } + if update.MasterKey == nil && update.SelfSigningKey == nil { + return + } + if err := a.Producer.ProduceSigningKeyUpdate(update); err != nil { + res.Error = &api.KeyError{ + Err: fmt.Sprintf("a.Producer.ProduceSigningKeyUpdate: %s", err), } + return } } @@ -310,16 +308,18 @@ func (a *KeyInternalAPI) PerformUploadDeviceSignatures(ctx context.Context, req // Finally, generate a notification that we updated the signatures. for userID := range req.Signatures { - if _, host, err := gomatrixserverlib.SplitID('@', userID); err == nil && host == a.ThisServer { - update := eduserverAPI.CrossSigningKeyUpdate{ - UserID: userID, - } - if err := a.Producer.ProduceSigningKeyUpdate(update); err != nil { - res.Error = &api.KeyError{ - Err: fmt.Sprintf("a.Producer.ProduceSigningKeyUpdate: %s", err), - } - return + masterKey := queryRes.MasterKeys[userID] + selfSigningKey := queryRes.SelfSigningKeys[userID] + update := eduserverAPI.CrossSigningKeyUpdate{ + UserID: userID, + MasterKey: &masterKey, + SelfSigningKey: &selfSigningKey, + } + if err := a.Producer.ProduceSigningKeyUpdate(update); err != nil { + res.Error = &api.KeyError{ + Err: fmt.Sprintf("a.Producer.ProduceSigningKeyUpdate: %s", err), } + return } } } diff --git a/keyserver/internal/internal.go b/keyserver/internal/internal.go index 259249217..2536c1f76 100644 --- a/keyserver/internal/internal.go +++ b/keyserver/internal/internal.go @@ -326,8 +326,14 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques if err = json.Unmarshal(key, &deviceKey); err != nil { continue } + if deviceKey.Signatures == nil { + deviceKey.Signatures = map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{} + } for sourceUserID, forSourceUser := range sigMap { for sourceKeyID, sourceSig := range forSourceUser { + if _, ok := deviceKey.Signatures[sourceUserID]; !ok { + deviceKey.Signatures[sourceUserID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{} + } deviceKey.Signatures[sourceUserID][sourceKeyID] = sourceSig } } @@ -447,7 +453,6 @@ func (a *KeyInternalAPI) queryRemoteKeysOnServer( for userID, deviceIDs := range devKeys { if len(deviceIDs) == 0 { userIDsForAllDevices[userID] = struct{}{} - delete(devKeys, userID) } } // for cross-signing keys, it's probably easier just to hit /keys/query if we aren't already doing diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go index 61ccc0303..bd36fd9f9 100644 --- a/keyserver/keyserver.go +++ b/keyserver/keyserver.go @@ -18,7 +18,6 @@ import ( "github.com/gorilla/mux" fedsenderapi "github.com/matrix-org/dendrite/federationapi/api" "github.com/matrix-org/dendrite/keyserver/api" - "github.com/matrix-org/dendrite/keyserver/consumers" "github.com/matrix-org/dendrite/keyserver/internal" "github.com/matrix-org/dendrite/keyserver/inthttp" "github.com/matrix-org/dendrite/keyserver/producers" @@ -65,12 +64,5 @@ func NewInternalAPI( } }() - keyconsumer := consumers.NewOutputCrossSigningKeyUpdateConsumer( - base.ProcessContext, base.Cfg, js, db, ap, - ) - if err := keyconsumer.Start(); err != nil { - logrus.WithError(err).Panicf("failed to start keyserver EDU server consumer") - } - return ap } diff --git a/roomserver/api/wrapper.go b/roomserver/api/wrapper.go index e9b94e48c..012094c62 100644 --- a/roomserver/api/wrapper.go +++ b/roomserver/api/wrapper.go @@ -51,7 +51,7 @@ func SendEventWithState( state *gomatrixserverlib.RespState, event *gomatrixserverlib.HeaderedEvent, origin gomatrixserverlib.ServerName, haveEventIDs map[string]bool, async bool, ) error { - outliers, err := state.Events() + outliers, err := state.Events(event.RoomVersion) if err != nil { return err } @@ -68,9 +68,10 @@ func SendEventWithState( }) } - stateEventIDs := make([]string, len(state.StateEvents)) - for i := range state.StateEvents { - stateEventIDs[i] = state.StateEvents[i].EventID() + stateEvents := state.StateEvents.UntrustedEvents(event.RoomVersion) + stateEventIDs := make([]string, len(stateEvents)) + for i := range stateEvents { + stateEventIDs[i] = stateEvents[i].EventID() } ires = append(ires, InputRoomEvent{ diff --git a/roomserver/internal/input/input_events.go b/roomserver/internal/input/input_events.go index 0ca5c31a9..873a051cd 100644 --- a/roomserver/internal/input/input_events.go +++ b/roomserver/internal/input/input_events.go @@ -195,9 +195,26 @@ func (r *Inputer) processRoomEvent( authEventNIDs := make([]types.EventNID, 0, len(authEventIDs)) for _, authEventID := range authEventIDs { if _, ok := knownEvents[authEventID]; !ok { - return rollbackTransaction, fmt.Errorf("missing auth event %s", authEventID) + // Unknown auth events only really matter if the event actually failed + // auth. If it passed auth then we can assume that everything that was + // known was sufficient, even if extraneous auth events were specified + // but weren't found. + if isRejected { + if event.StateKey() != nil { + return commitTransaction, fmt.Errorf( + "missing auth event %s for state event %s (type %q, state key %q)", + authEventID, event.EventID(), event.Type(), *event.StateKey(), + ) + } else { + return commitTransaction, fmt.Errorf( + "missing auth event %s for timeline event %s (type %q)", + authEventID, event.EventID(), event.Type(), + ) + } + } + } else { + authEventNIDs = append(authEventNIDs, knownEvents[authEventID].EventNID) } - authEventNIDs = append(authEventNIDs, knownEvents[authEventID].EventNID) } var softfail bool @@ -238,13 +255,32 @@ func (r *Inputer) processRoomEvent( hadEvents: map[string]bool{}, haveEvents: map[string]*gomatrixserverlib.HeaderedEvent{}, } - if err := missingState.processEventWithMissingState(ctx, event, headered.RoomVersion); err != nil { + if stateSnapshot, err := missingState.processEventWithMissingState(ctx, event, headered.RoomVersion); err != nil { + // Something went wrong with retrieving the missing state, so we can't + // really do anything with the event other than reject it at this point. isRejected = true rejectionErr = fmt.Errorf("missingState.processEventWithMissingState: %w", err) + } else if stateSnapshot != nil { + // We retrieved some state and we ended up having to call /state_ids for + // the new event in question (probably because closing the gap by using + // /get_missing_events didn't do what we hoped) so we'll instead overwrite + // the state snapshot with the newly resolved state. + missingPrev = false + input.HasState = true + input.StateEventIDs = make([]string, 0, len(stateSnapshot.StateEvents)) + for _, e := range stateSnapshot.StateEvents { + input.StateEventIDs = append(input.StateEventIDs, e.EventID()) + } } else { + // We retrieved some state and it would appear that rolling forward the + // state did everything we needed it to do, so we can just resolve the + // state for the event in the normal way. missingPrev = false } } else { + // We're missing prev events or state for the event, but for some reason + // we don't know any servers to ask. In this case we can't do anything but + // reject the event and hope that it gets unrejected later. isRejected = true rejectionErr = fmt.Errorf("missing prev events and no other servers to ask") } @@ -282,7 +318,7 @@ func (r *Inputer) processRoomEvent( return rollbackTransaction, fmt.Errorf("updater.RoomInfo missing for room %s", event.RoomID()) } - if !missingPrev && stateAtEvent.BeforeStateSnapshotNID == 0 { + if input.HasState || (!missingPrev && stateAtEvent.BeforeStateSnapshotNID == 0) { // We haven't calculated a state for this event yet. // Lets calculate one. err = r.calculateAndSetState(ctx, updater, input, roomInfo, &stateAtEvent, event, isRejected) @@ -297,7 +333,10 @@ func (r *Inputer) processRoomEvent( "soft_fail": softfail, "missing_prev": missingPrev, }).Warn("Stored rejected event") - return commitTransaction, rejectionErr + if rejectionErr != nil { + return commitTransaction, types.RejectedError(rejectionErr.Error()) + } + return commitTransaction, nil } switch input.Kind { @@ -413,44 +452,42 @@ func (r *Inputer) fetchAuthEvents( return fmt.Errorf("no servers provided event auth for event ID %q, tried servers %v", event.EventID(), servers) } + // Reuse these to reduce allocations. + authEventNIDs := make([]types.EventNID, 0, 5) + isRejected := false +nextAuthEvent: for _, authEvent := range gomatrixserverlib.ReverseTopologicalOrdering( - res.AuthEvents, + res.AuthEvents.UntrustedEvents(event.RoomVersion), gomatrixserverlib.TopologicalOrderByAuthEvents, ) { // If we already know about this event from the database then we don't // need to store it again or do anything further with it, so just skip // over it rather than wasting cycles. if ev, ok := known[authEvent.EventID()]; ok && ev != nil { - continue + continue nextAuthEvent } // Check the signatures of the event. If this fails then we'll simply // skip it, because gomatrixserverlib.Allowed() will notice a problem // if a critical event is missing anyway. if err := authEvent.VerifyEventSignatures(ctx, r.FSAPI.KeyRing()); err != nil { - continue + continue nextAuthEvent } // In order to store the new auth event, we need to know its auth chain // as NIDs for the `auth_event_nids` column. Let's see if we can find those. - authEventNIDs := make([]types.EventNID, 0, len(authEvent.AuthEventIDs())) + authEventNIDs = authEventNIDs[:0] for _, eventID := range authEvent.AuthEventIDs() { knownEvent, ok := known[eventID] if !ok { - return fmt.Errorf("missing auth event %s for %s", eventID, authEvent.EventID()) + continue nextAuthEvent } authEventNIDs = append(authEventNIDs, knownEvent.EventNID) } - // Let's take a note of the fact that we now know about this event. - if err := auth.AddEvent(authEvent); err != nil { - return fmt.Errorf("auth.AddEvent: %w", err) - } - // Check if the auth event should be rejected. - isRejected := false - if err := gomatrixserverlib.Allowed(authEvent, auth); err != nil { - isRejected = true + err := gomatrixserverlib.Allowed(authEvent, auth) + if isRejected = err != nil; isRejected { logger.WithError(err).Warnf("Auth event %s rejected", authEvent.EventID()) } @@ -460,6 +497,14 @@ func (r *Inputer) fetchAuthEvents( return fmt.Errorf("updater.StoreEvent: %w", err) } + // Let's take a note of the fact that we now know about this event for + // authenticating future events. + if !isRejected { + if err := auth.AddEvent(authEvent); err != nil { + return fmt.Errorf("auth.AddEvent: %w", err) + } + } + // Now we know about this event, it was stored and the signatures were OK. known[authEvent.EventID()] = &types.Event{ EventNID: eventNID, @@ -483,16 +528,7 @@ func (r *Inputer) calculateAndSetState( roomState := state.NewStateResolution(updater, roomInfo) if input.HasState { - // Check here if we think we're in the room already. stateAtEvent.Overwrite = true - var joinEventNIDs []types.EventNID - // Request join memberships only for local users only. - if joinEventNIDs, err = updater.GetMembershipEventNIDsForRoom(ctx, roomInfo.RoomNID, true, true); err == nil { - // If we have no local users that are joined to the room then any state about - // the room that we have is quite possibly out of date. Therefore in that case - // we should overwrite it rather than merge it. - stateAtEvent.Overwrite = len(joinEventNIDs) == 0 - } // We've been told what the state at the event is so we don't need to calculate it. // Check that those state events are in the database and store the state. diff --git a/roomserver/internal/input/input_membership.go b/roomserver/internal/input/input_membership.go index ff3ed7e5d..3953586b2 100644 --- a/roomserver/internal/input/input_membership.go +++ b/roomserver/internal/input/input_membership.go @@ -48,7 +48,7 @@ func (r *Inputer) updateMemberships( // Load the event JSON so we can look up the "membership" key. // TODO: Maybe add a membership key to the events table so we can load that // key without having to load the entire event JSON? - events, err := r.DB.Events(ctx, eventNIDs) + events, err := updater.Events(ctx, eventNIDs) if err != nil { return nil, err } diff --git a/roomserver/internal/input/input_missing.go b/roomserver/internal/input/input_missing.go index 4d3306660..19771d4bd 100644 --- a/roomserver/internal/input/input_missing.go +++ b/roomserver/internal/input/input_missing.go @@ -12,11 +12,17 @@ import ( "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/internal/query" "github.com/matrix-org/dendrite/roomserver/storage/shared" + "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" "github.com/sirupsen/logrus" ) +type parsedRespState struct { + AuthEvents []*gomatrixserverlib.Event + StateEvents []*gomatrixserverlib.Event +} + type missingStateReq struct { origin gomatrixserverlib.ServerName db *shared.RoomUpdater @@ -34,9 +40,10 @@ type missingStateReq struct { // processEventWithMissingState is the entrypoint for a missingStateReq // request, as called from processRoomEvent. +// nolint:gocyclo func (t *missingStateReq) processEventWithMissingState( ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion, -) error { +) (*parsedRespState, error) { // We are missing the previous events for this events. // This means that there is a gap in our view of the history of the // room. There two ways that we can handle such a gap: @@ -62,100 +69,51 @@ func (t *missingStateReq) processEventWithMissingState( // - fill in the gap completely then process event `e` returning no backwards extremity // - fail to fill in the gap and tell us to terminate the transaction err=not nil // - fail to fill in the gap and tell us to fetch state at the new backwards extremity, and to not terminate the transaction - newEvents, isGapFilled, err := t.getMissingEvents(ctx, e, roomVersion) + newEvents, isGapFilled, prevStatesKnown, err := t.getMissingEvents(ctx, e, roomVersion) if err != nil { - return fmt.Errorf("t.getMissingEvents: %w", err) + return nil, fmt.Errorf("t.getMissingEvents: %w", err) } if len(newEvents) == 0 { - return fmt.Errorf("expected to find missing events but didn't") + return nil, fmt.Errorf("expected to find missing events but didn't") } if isGapFilled { - logger.Infof("gap filled by /get_missing_events, injecting %d new events", len(newEvents)) + logger.Infof("Gap filled by /get_missing_events, injecting %d new events", len(newEvents)) // we can just inject all the newEvents as new as we may have only missed 1 or 2 events and have filled // in the gap in the DAG for _, newEvent := range newEvents { _, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{ - Kind: api.KindNew, + Kind: api.KindOld, Event: newEvent.Headered(roomVersion), Origin: t.origin, SendAsServer: api.DoNotSendToOtherServers, }) if err != nil { - return fmt.Errorf("t.inputer.processRoomEvent: %w", err) + if _, ok := err.(types.RejectedError); !ok { + return nil, fmt.Errorf("t.inputer.processRoomEvent (filling gap): %w", err) + } } } - return nil } + // If we filled the gap *and* we know the state before the prev events + // then there's nothing else to do, we have everything we need to deal + // with the new event. + if isGapFilled && prevStatesKnown { + logger.Infof("Gap filled and state found for all prev events") + return nil, nil + } + + // Otherwise, if we've reached this point, it's possible that we've + // either not closed the gap, or we did but we still don't seem to + // know the events before the new event. Start by looking up the + // state at the event at the back of the gap and we'll try to roll + // forward the state first. backwardsExtremity := newEvents[0] newEvents = newEvents[1:] - type respState struct { - // A snapshot is considered trustworthy if it came from our own roomserver. - // That's because the state will have been through state resolution once - // already in QueryStateAfterEvent. - trustworthy bool - *gomatrixserverlib.RespState - } - - // at this point we know we're going to have a gap: we need to work out the room state at the new backwards extremity. - // Therefore, we cannot just query /state_ids with this event to get the state before. Instead, we need to query - // the state AFTER all the prev_events for this event, then apply state resolution to that to get the state before the event. - var states []*respState - for _, prevEventID := range backwardsExtremity.PrevEventIDs() { - // Look up what the state is after the backward extremity. This will either - // come from the roomserver, if we know all the required events, or it will - // come from a remote server via /state_ids if not. - prevState, trustworthy, lerr := t.lookupStateAfterEvent(ctx, roomVersion, backwardsExtremity.RoomID(), prevEventID) - if lerr != nil { - logger.WithError(lerr).Errorf("Failed to lookup state after prev_event: %s", prevEventID) - return lerr - } - // Append the state onto the collected state. We'll run this through the - // state resolution next. - states = append(states, &respState{trustworthy, prevState}) - } - - // Now that we have collected all of the state from the prev_events, we'll - // run the state through the appropriate state resolution algorithm for the - // room if needed. This does a couple of things: - // 1. Ensures that the state is deduplicated fully for each state-key tuple - // 2. Ensures that we pick the latest events from both sets, in the case that - // one of the prev_events is quite a bit older than the others - resolvedState := &gomatrixserverlib.RespState{} - switch len(states) { - case 0: - extremityIsCreate := backwardsExtremity.Type() == gomatrixserverlib.MRoomCreate && backwardsExtremity.StateKeyEquals("") - if !extremityIsCreate { - // There are no previous states and this isn't the beginning of the - // room - this is an error condition! - logger.Errorf("Failed to lookup any state after prev_events") - return fmt.Errorf("expected %d states but got %d", len(backwardsExtremity.PrevEventIDs()), len(states)) - } - case 1: - // There's only one previous state - if it's trustworthy (came from a - // local state snapshot which will already have been through state res), - // use it as-is. There's no point in resolving it again. - if states[0].trustworthy { - resolvedState = states[0].RespState - break - } - // Otherwise, if it isn't trustworthy (came from federation), run it through - // state resolution anyway for safety, in case there are duplicates. - fallthrough - default: - respStates := make([]*gomatrixserverlib.RespState, len(states)) - for i := range states { - respStates[i] = states[i].RespState - } - // There's more than one previous state - run them all through state res - t.roomsMu.Lock(e.RoomID()) - resolvedState, err = t.resolveStatesAndCheck(ctx, roomVersion, respStates, backwardsExtremity) - t.roomsMu.Unlock(e.RoomID()) - if err != nil { - logger.WithError(err).Errorf("Failed to resolve state conflicts for event %s", backwardsExtremity.EventID()) - return err - } + resolvedState, err := t.lookupResolvedStateBeforeEvent(ctx, backwardsExtremity, roomVersion) + if err != nil { + return nil, fmt.Errorf("t.lookupState (backwards extremity): %w", err) } hadEvents := map[string]bool{} @@ -165,27 +123,37 @@ func (t *missingStateReq) processEventWithMissingState( } t.hadEventsMutex.Unlock() - // Send outliers first so we can send the new backwards extremity without causing errors - outliers, err := resolvedState.Events() - if err != nil { - return err - } - var outlierRoomEvents []api.InputRoomEvent - for _, outlier := range outliers { - if hadEvents[outlier.EventID()] { - continue + sendOutliers := func(resolvedState *parsedRespState) error { + outliers, oerr := gomatrixserverlib.OrderAuthAndStateEvents(resolvedState.AuthEvents, resolvedState.StateEvents, roomVersion) + if oerr != nil { + return fmt.Errorf("gomatrixserverlib.OrderAuthAndStateEvents: %w", oerr) } - outlierRoomEvents = append(outlierRoomEvents, api.InputRoomEvent{ - Kind: api.KindOutlier, - Event: outlier.Headered(roomVersion), - Origin: t.origin, - }) - } - // TODO: we could do this concurrently? - for _, ire := range outlierRoomEvents { - if _, err = t.inputer.processRoomEvent(ctx, t.db, &ire); err != nil { - return fmt.Errorf("t.inputer.processRoomEvent[outlier]: %w", err) + var outlierRoomEvents []api.InputRoomEvent + for _, outlier := range outliers { + if hadEvents[outlier.EventID()] { + continue + } + outlierRoomEvents = append(outlierRoomEvents, api.InputRoomEvent{ + Kind: api.KindOutlier, + Event: outlier.Headered(roomVersion), + Origin: t.origin, + }) } + for _, ire := range outlierRoomEvents { + _, err = t.inputer.processRoomEvent(ctx, t.db, &ire) + if err != nil { + if _, ok := err.(types.RejectedError); !ok { + return fmt.Errorf("t.inputer.processRoomEvent (outlier): %w", err) + } + } + } + return nil + } + + // Send outliers first so we can send the state along with the new backwards + // extremity without any missing auth events. + if err = sendOutliers(resolvedState); err != nil { + return nil, fmt.Errorf("sendOutliers: %w", err) } // Now send the backward extremity into the roomserver with the @@ -205,7 +173,9 @@ func (t *missingStateReq) processEventWithMissingState( SendAsServer: api.DoNotSendToOtherServers, }) if err != nil { - return fmt.Errorf("t.inputer.processRoomEvent: %w", err) + if _, ok := err.(types.RejectedError); !ok { + return nil, fmt.Errorf("t.inputer.processRoomEvent (backward extremity): %w", err) + } } // Then send all of the newer backfilled events, of which will all be newer @@ -220,16 +190,115 @@ func (t *missingStateReq) processEventWithMissingState( SendAsServer: api.DoNotSendToOtherServers, }) if err != nil { - return fmt.Errorf("t.inputer.processRoomEvent: %w", err) + if _, ok := err.(types.RejectedError); !ok { + return nil, fmt.Errorf("t.inputer.processRoomEvent (fast forward): %w", err) + } } } - return nil + // Finally, check again if we know everything we need to know in order to + // make forward progress. If the prev state is known then we consider the + // rolled forward state to be sufficient — we now know all of the state + // before the prev events. If we don't then we need to look up the state + // before the new event as well, otherwise we will never make any progress. + if t.isPrevStateKnown(ctx, e) { + return nil, nil + } + + // If we still haven't got the state for the prev events then we'll go and + // ask the federation for it if needed. + resolvedState, err = t.lookupResolvedStateBeforeEvent(ctx, e, roomVersion) + if err != nil { + return nil, fmt.Errorf("t.lookupState (new event): %w", err) + } + + // Send the outliers for the retrieved state. + if err = sendOutliers(resolvedState); err != nil { + return nil, fmt.Errorf("sendOutliers: %w", err) + } + + // Then return the resolved state, for which the caller can replace the + // HasState with the event IDs to create a new state snapshot when we + // process the new event. + return resolvedState, nil +} + +func (t *missingStateReq) lookupResolvedStateBeforeEvent(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (*parsedRespState, error) { + type respState struct { + // A snapshot is considered trustworthy if it came from our own roomserver. + // That's because the state will have been through state resolution once + // already in QueryStateAfterEvent. + trustworthy bool + *parsedRespState + } + + // at this point we know we're going to have a gap: we need to work out the room state at the new backwards extremity. + // Therefore, we cannot just query /state_ids with this event to get the state before. Instead, we need to query + // the state AFTER all the prev_events for this event, then apply state resolution to that to get the state before the event. + var states []*respState + for _, prevEventID := range e.PrevEventIDs() { + // Look up what the state is after the backward extremity. This will either + // come from the roomserver, if we know all the required events, or it will + // come from a remote server via /state_ids if not. + prevState, trustworthy, err := t.lookupStateAfterEvent(ctx, roomVersion, e.RoomID(), prevEventID) + if err != nil { + return nil, fmt.Errorf("t.lookupStateAfterEvent: %w", err) + } + // Append the state onto the collected state. We'll run this through the + // state resolution next. + states = append(states, &respState{trustworthy, prevState}) + } + + // Now that we have collected all of the state from the prev_events, we'll + // run the state through the appropriate state resolution algorithm for the + // room if needed. This does a couple of things: + // 1. Ensures that the state is deduplicated fully for each state-key tuple + // 2. Ensures that we pick the latest events from both sets, in the case that + // one of the prev_events is quite a bit older than the others + resolvedState := &parsedRespState{} + switch len(states) { + case 0: + extremityIsCreate := e.Type() == gomatrixserverlib.MRoomCreate && e.StateKeyEquals("") + if !extremityIsCreate { + // There are no previous states and this isn't the beginning of the + // room - this is an error condition! + return nil, fmt.Errorf("expected %d states but got %d", len(e.PrevEventIDs()), len(states)) + } + case 1: + // There's only one previous state - if it's trustworthy (came from a + // local state snapshot which will already have been through state res), + // use it as-is. There's no point in resolving it again. Only trust a + // trustworthy state snapshot if it actually contains some state for all + // non-create events, otherwise we need to resolve what came from federation. + isCreate := e.Type() == gomatrixserverlib.MRoomCreate && e.StateKeyEquals("") + if states[0].trustworthy && (isCreate || len(states[0].StateEvents) > 0) { + resolvedState = states[0].parsedRespState + break + } + // Otherwise, if it isn't trustworthy (came from federation), run it through + // state resolution anyway for safety, in case there are duplicates. + fallthrough + default: + respStates := make([]*parsedRespState, len(states)) + for i := range states { + respStates[i] = states[i].parsedRespState + } + // There's more than one previous state - run them all through state res + var err error + t.roomsMu.Lock(e.RoomID()) + resolvedState, err = t.resolveStatesAndCheck(ctx, roomVersion, respStates, e) + t.roomsMu.Unlock(e.RoomID()) + if err != nil { + return nil, fmt.Errorf("t.resolveStatesAndCheck: %w", err) + } + } + + return resolvedState, nil } // lookupStateAfterEvent returns the room state after `eventID`, which is the state before eventID with the state of `eventID` (if it's a state event) // added into the mix. -func (t *missingStateReq) lookupStateAfterEvent(ctx context.Context, roomVersion gomatrixserverlib.RoomVersion, roomID, eventID string) (*gomatrixserverlib.RespState, bool, error) { +func (t *missingStateReq) lookupStateAfterEvent(ctx context.Context, roomVersion gomatrixserverlib.RoomVersion, roomID, eventID string) (*parsedRespState, bool, error) { // try doing all this locally before we resort to querying federation respState := t.lookupStateAfterEventLocally(ctx, roomID, eventID) if respState != nil { @@ -280,7 +349,7 @@ func (t *missingStateReq) cacheAndReturn(ev *gomatrixserverlib.HeaderedEvent) *g return ev } -func (t *missingStateReq) lookupStateAfterEventLocally(ctx context.Context, roomID, eventID string) *gomatrixserverlib.RespState { +func (t *missingStateReq) lookupStateAfterEventLocally(ctx context.Context, roomID, eventID string) *parsedRespState { var res api.QueryStateAfterEventsResponse err := t.queryer.QueryStateAfterEvents(ctx, &api.QueryStateAfterEventsRequest{ RoomID: roomID, @@ -335,7 +404,7 @@ func (t *missingStateReq) lookupStateAfterEventLocally(ctx context.Context, room queryRes.Events = nil } - return &gomatrixserverlib.RespState{ + return &parsedRespState{ StateEvents: gomatrixserverlib.UnwrapEventHeaders(stateEvents), AuthEvents: authEvents, } @@ -344,13 +413,13 @@ func (t *missingStateReq) lookupStateAfterEventLocally(ctx context.Context, room // lookuptStateBeforeEvent returns the room state before the event e, which is just /state_ids and/or /state depending on what // the server supports. func (t *missingStateReq) lookupStateBeforeEvent(ctx context.Context, roomVersion gomatrixserverlib.RoomVersion, roomID, eventID string) ( - *gomatrixserverlib.RespState, error) { + *parsedRespState, error) { // Attempt to fetch the missing state using /state_ids and /events return t.lookupMissingStateViaStateIDs(ctx, roomID, eventID, roomVersion) } -func (t *missingStateReq) resolveStatesAndCheck(ctx context.Context, roomVersion gomatrixserverlib.RoomVersion, states []*gomatrixserverlib.RespState, backwardsExtremity *gomatrixserverlib.Event) (*gomatrixserverlib.RespState, error) { +func (t *missingStateReq) resolveStatesAndCheck(ctx context.Context, roomVersion gomatrixserverlib.RoomVersion, states []*parsedRespState, backwardsExtremity *gomatrixserverlib.Event) (*parsedRespState, error) { var authEventList []*gomatrixserverlib.Event var stateEventList []*gomatrixserverlib.Event for _, state := range states { @@ -369,7 +438,7 @@ retryAllowedState: h, err2 := t.lookupEvent(ctx, roomVersion, backwardsExtremity.RoomID(), missing.AuthEventID, true) switch err2.(type) { case verifySigError: - return &gomatrixserverlib.RespState{ + return &parsedRespState{ AuthEvents: authEventList, StateEvents: resolvedStateEvents, }, nil @@ -385,7 +454,7 @@ retryAllowedState: } return nil, err } - return &gomatrixserverlib.RespState{ + return &parsedRespState{ AuthEvents: authEventList, StateEvents: resolvedStateEvents, }, nil @@ -393,22 +462,13 @@ retryAllowedState: // get missing events for `e`. If `isGapFilled`=true then `newEvents` contains all the events to inject, // without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events -func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled bool, err error) { +func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled, prevStateKnown bool, err error) { logger := util.GetLogger(ctx).WithField("event_id", e.EventID()).WithField("room_id", e.RoomID()) - needed := gomatrixserverlib.StateNeededForAuth([]*gomatrixserverlib.Event{e}) - // query latest events (our trusted forward extremities) - req := api.QueryLatestEventsAndStateRequest{ - RoomID: e.RoomID(), - StateToFetch: needed.Tuples(), - } - var res api.QueryLatestEventsAndStateResponse - if err = t.queryer.QueryLatestEventsAndState(ctx, &req, &res); err != nil { - logger.WithError(err).Warn("Failed to query latest events") - return nil, false, err - } - latestEvents := make([]string, len(res.LatestEvents)) - for i, ev := range res.LatestEvents { - latestEvents[i] = res.LatestEvents[i].EventID + + latest := t.db.LatestEvents() + latestEvents := make([]string, len(latest)) + for i, ev := range latest { + latestEvents[i] = ev.EventID t.hadEvent(ev.EventID) } @@ -429,7 +489,7 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve if errors.Is(err, context.DeadlineExceeded) { select { case <-ctx.Done(): // the parent request context timed out - return nil, false, context.DeadlineExceeded + return nil, false, false, context.DeadlineExceeded default: // this request exceed its own timeout continue } @@ -442,7 +502,7 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve "%s pushed us an event but %d server(s) couldn't give us details about prev_events via /get_missing_events - dropping this event until it can", t.origin, len(t.servers), ) - return nil, false, missingPrevEventsError{ + return nil, false, false, missingPrevEventsError{ eventID: e.EventID(), err: err, } @@ -451,12 +511,13 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve // Make sure events from the missingResp are using the cache - missing events // will be added and duplicates will be removed. logger.Debugf("get_missing_events returned %d events", len(missingResp.Events)) - for i, ev := range missingResp.Events { - missingResp.Events[i] = t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap() + missingEvents := make([]*gomatrixserverlib.Event, 0, len(missingResp.Events)) + for _, ev := range missingResp.Events.UntrustedEvents(roomVersion) { + missingEvents = append(missingEvents, t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap()) } // topologically sort and sanity check that we are making forward progress - newEvents = gomatrixserverlib.ReverseTopologicalOrdering(missingResp.Events, gomatrixserverlib.TopologicalOrderByPrevEvents) + newEvents = gomatrixserverlib.ReverseTopologicalOrdering(missingEvents, gomatrixserverlib.TopologicalOrderByPrevEvents) shouldHaveSomeEventIDs := e.PrevEventIDs() hasPrevEvent := false Event: @@ -474,52 +535,84 @@ Event: "%s pushed us an event but couldn't give us details about prev_events via /get_missing_events - dropping this event until it can", t.origin, ) - return nil, false, missingPrevEventsError{ + return nil, false, false, missingPrevEventsError{ eventID: e.EventID(), err: err, } } if len(newEvents) == 0 { - return nil, false, nil // TODO: error instead? + return nil, false, false, nil // TODO: error instead? } - // now check if we can fill the gap. Look to see if we have state snapshot IDs for the earliest event earliestNewEvent := newEvents[0] - if state, err := t.db.StateAtEventIDs(ctx, []string{earliestNewEvent.EventID()}); err != nil || len(state) == 0 { - if earliestNewEvent.Type() == gomatrixserverlib.MRoomCreate && earliestNewEvent.StateKeyEquals("") { - // we got to the beginning of the room so there will be no state! It's all good we can process this - return newEvents, true, nil - } - // we don't have the state at this earliest event from /g_m_e so we won't have state for later events either - return newEvents, false, nil + + // If we retrieved back to the beginning of the room then there's nothing else + // to do - we closed the gap. + if len(earliestNewEvent.PrevEventIDs()) == 0 && earliestNewEvent.Type() == gomatrixserverlib.MRoomCreate && earliestNewEvent.StateKeyEquals("") { + return newEvents, true, t.isPrevStateKnown(ctx, e), nil } - // StateAtEventIDs returned some kind of state for the earliest event so we can fill in the gap! - return newEvents, true, nil + + // If our backward extremity was not a known event to us then we obviously didn't + // close the gap. + if state, err := t.db.StateAtEventIDs(ctx, []string{earliestNewEvent.EventID()}); err != nil || len(state) == 0 && state[0].BeforeStateSnapshotNID == 0 { + return newEvents, false, false, nil + } + + // At this point we are satisfied that we know the state both at the earliest + // retrieved event and at the prev events of the new event. + return newEvents, true, t.isPrevStateKnown(ctx, e), nil } -func (t *missingStateReq) lookupMissingStateViaState(ctx context.Context, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion) ( - respState *gomatrixserverlib.RespState, err error) { +func (t *missingStateReq) isPrevStateKnown(ctx context.Context, e *gomatrixserverlib.Event) bool { + expected := len(e.PrevEventIDs()) + state, err := t.db.StateAtEventIDs(ctx, e.PrevEventIDs()) + if err != nil || len(state) != expected { + // We didn't get as many state snapshots as we expected, or there was an error, + // so we haven't completely solved the problem for the new event. + return false + } + // Check to see if we have a populated state snapshot for all of the prev events. + for _, stateAtEvent := range state { + if stateAtEvent.BeforeStateSnapshotNID == 0 { + // One of the prev events still has unknown state, so we haven't really + // solved the problem. + return false + } + } + return true +} + +func (t *missingStateReq) lookupMissingStateViaState( + ctx context.Context, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion, +) (respState *parsedRespState, err error) { state, err := t.federation.LookupState(ctx, t.origin, roomID, eventID, roomVersion) if err != nil { return nil, err } // Check that the returned state is valid. - if err := state.Check(ctx, t.keys, nil); err != nil { + if err := state.Check(ctx, roomVersion, t.keys, nil); err != nil { return nil, err } + parsedState := &parsedRespState{ + AuthEvents: make([]*gomatrixserverlib.Event, len(state.AuthEvents)), + StateEvents: make([]*gomatrixserverlib.Event, len(state.StateEvents)), + } // Cache the results of this state lookup and deduplicate anything we already // have in the cache, freeing up memory. - for i, ev := range state.AuthEvents { - state.AuthEvents[i] = t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap() + // We load these as trusted as we called state.Check before which loaded them as untrusted. + for i, evJSON := range state.AuthEvents { + ev, _ := gomatrixserverlib.NewEventFromTrustedJSON(evJSON, false, roomVersion) + parsedState.AuthEvents[i] = t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap() } - for i, ev := range state.StateEvents { - state.StateEvents[i] = t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap() + for i, evJSON := range state.StateEvents { + ev, _ := gomatrixserverlib.NewEventFromTrustedJSON(evJSON, false, roomVersion) + parsedState.StateEvents[i] = t.cacheAndReturn(ev.Headered(roomVersion)).Unwrap() } - return &state, nil + return parsedState, nil } func (t *missingStateReq) lookupMissingStateViaStateIDs(ctx context.Context, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion) ( - *gomatrixserverlib.RespState, error) { + *parsedRespState, error) { util.GetLogger(ctx).WithField("room_id", roomID).Infof("lookupMissingStateViaStateIDs %s", eventID) // fetch the state event IDs at the time of the event stateIDs, err := t.federation.LookupStateIDs(ctx, t.origin, roomID, eventID) @@ -651,13 +744,14 @@ func (t *missingStateReq) lookupMissingStateViaStateIDs(ctx context.Context, roo return resp, err } -func (t *missingStateReq) createRespStateFromStateIDs(stateIDs gomatrixserverlib.RespStateIDs) ( - *gomatrixserverlib.RespState, error) { // nolint:unparam +func (t *missingStateReq) createRespStateFromStateIDs( + stateIDs gomatrixserverlib.RespStateIDs, +) (*parsedRespState, error) { // nolint:unparam t.haveEventsMutex.Lock() defer t.haveEventsMutex.Unlock() // create a RespState response using the response to /state_ids as a guide - respState := gomatrixserverlib.RespState{} + respState := parsedRespState{} for i := range stateIDs.StateEventIDs { ev, ok := t.haveEvents[stateIDs.StateEventIDs[i]] diff --git a/roomserver/internal/query/query.go b/roomserver/internal/query/query.go index 6b4cb5816..05cd686f4 100644 --- a/roomserver/internal/query/query.go +++ b/roomserver/internal/query/query.go @@ -149,7 +149,8 @@ func (r *Queryer) QueryMissingAuthPrevEvents( } for _, prevEventID := range request.PrevEventIDs { - if state, err := r.DB.StateAtEventIDs(ctx, []string{prevEventID}); err != nil || len(state) == 0 { + state, err := r.DB.StateAtEventIDs(ctx, []string{prevEventID}) + if err != nil || len(state) == 0 || (!state[0].IsCreate() && state[0].BeforeStateSnapshotNID == 0) { response.MissingPrevEventIDs = append(response.MissingPrevEventIDs, prevEventID) } } diff --git a/roomserver/storage/postgres/events_table.go b/roomserver/storage/postgres/events_table.go index 6c3847752..c136f039a 100644 --- a/roomserver/storage/postgres/events_table.go +++ b/roomserver/storage/postgres/events_table.go @@ -71,10 +71,10 @@ CREATE TABLE IF NOT EXISTS roomserver_events ( ` const insertEventSQL = "" + - "INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth, is_rejected)" + + "INSERT INTO roomserver_events AS e (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth, is_rejected)" + " VALUES ($1, $2, $3, $4, $5, $6, $7, $8)" + - " ON CONFLICT ON CONSTRAINT roomserver_event_id_unique" + - " DO NOTHING" + + " ON CONFLICT ON CONSTRAINT roomserver_event_id_unique DO UPDATE" + + " SET is_rejected = $8 WHERE e.event_id = $4 AND e.is_rejected = FALSE" + " RETURNING event_nid, state_snapshot_nid" const selectEventSQL = "" + @@ -192,7 +192,8 @@ func (s *eventStatements) InsertEvent( ) (types.EventNID, types.StateSnapshotNID, error) { var eventNID int64 var stateNID int64 - err := s.insertEventStmt.QueryRowContext( + stmt := sqlutil.TxStmt(txn, s.insertEventStmt) + err := stmt.QueryRowContext( ctx, int64(roomNID), int64(eventTypeNID), int64(eventStateKeyNID), eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth, isRejected, diff --git a/roomserver/storage/shared/membership_updater.go b/roomserver/storage/shared/membership_updater.go index f1f589a31..66ac2f5b6 100644 --- a/roomserver/storage/shared/membership_updater.go +++ b/roomserver/storage/shared/membership_updater.go @@ -136,7 +136,7 @@ func (u *MembershipUpdater) SetToJoin(senderUserID string, eventID string, isUpd } // Look up the NID of the new join event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) + nIDs, err := u.d.eventNIDs(u.ctx, u.txn, []string{eventID}) if err != nil { return fmt.Errorf("u.d.EventNIDs: %w", err) } @@ -170,7 +170,7 @@ func (u *MembershipUpdater) SetToLeave(senderUserID string, eventID string) ([]s } // Look up the NID of the new leave event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) + nIDs, err := u.d.eventNIDs(u.ctx, u.txn, []string{eventID}) if err != nil { return fmt.Errorf("u.d.EventNIDs: %w", err) } @@ -196,7 +196,7 @@ func (u *MembershipUpdater) SetToKnock(event *gomatrixserverlib.Event) (bool, er } if u.membership != tables.MembershipStateKnock { // Look up the NID of the new knock event - nIDs, err := u.d.EventNIDs(u.ctx, []string{event.EventID()}) + nIDs, err := u.d.eventNIDs(u.ctx, u.txn, []string{event.EventID()}) if err != nil { return fmt.Errorf("u.d.EventNIDs: %w", err) } diff --git a/roomserver/storage/shared/room_updater.go b/roomserver/storage/shared/room_updater.go index bb9f5dc62..fc75a2606 100644 --- a/roomserver/storage/shared/room_updater.go +++ b/roomserver/storage/shared/room_updater.go @@ -187,6 +187,12 @@ func (u *RoomUpdater) EventIDs( return u.d.EventsTable.BulkSelectEventID(ctx, u.txn, eventNIDs) } +func (u *RoomUpdater) EventNIDs( + ctx context.Context, eventIDs []string, +) (map[string]types.EventNID, error) { + return u.d.eventNIDs(ctx, u.txn, eventIDs) +} + func (u *RoomUpdater) StateAtEventIDs( ctx context.Context, eventIDs []string, ) ([]types.StateAtEvent, error) { diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go index 127cd1f52..8319de265 100644 --- a/roomserver/storage/shared/storage.go +++ b/roomserver/storage/shared/storage.go @@ -603,6 +603,8 @@ func (d *Database) storeEvent( if err == sql.ErrNoRows { // We've already inserted the event so select the numeric event ID eventNID, stateNID, err = d.EventsTable.SelectEvent(ctx, txn, event.EventID()) + } else if err != nil { + return fmt.Errorf("d.EventsTable.InsertEvent: %w", err) } if err != nil { return fmt.Errorf("d.EventsTable.SelectEvent: %w", err) diff --git a/roomserver/storage/sqlite3/events_table.go b/roomserver/storage/sqlite3/events_table.go index e1e6a597c..cef09fe60 100644 --- a/roomserver/storage/sqlite3/events_table.go +++ b/roomserver/storage/sqlite3/events_table.go @@ -49,7 +49,8 @@ const eventsSchema = ` const insertEventSQL = ` INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth, is_rejected) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT DO NOTHING + ON CONFLICT DO UPDATE + SET is_rejected = $8 WHERE is_rejected = 0 RETURNING event_nid, state_snapshot_nid; ` diff --git a/roomserver/types/types.go b/roomserver/types/types.go index d7e03ad61..5d52ccfcd 100644 --- a/roomserver/types/types.go +++ b/roomserver/types/types.go @@ -83,6 +83,10 @@ type StateKeyTuple struct { EventStateKeyNID EventStateKeyNID } +func (a StateKeyTuple) IsCreate() bool { + return a.EventTypeNID == MRoomCreateNID && a.EventStateKeyNID == EmptyStateKeyNID +} + // LessThan returns true if this state key is less than the other state key. // The ordering is arbitrary and is used to implement binary search and to efficiently deduplicate entries. func (a StateKeyTuple) LessThan(b StateKeyTuple) bool { @@ -209,6 +213,12 @@ type MissingEventError string func (e MissingEventError) Error() string { return string(e) } +// A RejectedError is returned when an event is stored as rejected. The error +// contains the reason why. +type RejectedError string + +func (e RejectedError) Error() string { return string(e) } + // RoomInfo contains metadata about a room type RoomInfo struct { RoomNID RoomNID diff --git a/setup/mscs/msc2836/msc2836.go b/setup/mscs/msc2836/msc2836.go index 8a35e4143..0af22c19a 100644 --- a/setup/mscs/msc2836/msc2836.go +++ b/setup/mscs/msc2836/msc2836.go @@ -82,9 +82,15 @@ type EventRelationshipResponse struct { Limited bool `json:"limited"` } -func toClientResponse(res *gomatrixserverlib.MSC2836EventRelationshipsResponse) *EventRelationshipResponse { +type MSC2836EventRelationshipsResponse struct { + gomatrixserverlib.MSC2836EventRelationshipsResponse + ParsedEvents []*gomatrixserverlib.Event + ParsedAuthChain []*gomatrixserverlib.Event +} + +func toClientResponse(res *MSC2836EventRelationshipsResponse) *EventRelationshipResponse { out := &EventRelationshipResponse{ - Events: gomatrixserverlib.ToClientEvents(res.Events, gomatrixserverlib.FormatAll), + Events: gomatrixserverlib.ToClientEvents(res.ParsedEvents, gomatrixserverlib.FormatAll), Limited: res.Limited, NextBatch: res.NextBatch, } @@ -210,7 +216,7 @@ func federatedEventRelationship( // add auth chain information requiredAuthEventsSet := make(map[string]bool) var requiredAuthEvents []string - for _, ev := range res.Events { + for _, ev := range res.ParsedEvents { for _, a := range ev.AuthEventIDs() { if requiredAuthEventsSet[a] { continue @@ -227,19 +233,24 @@ func federatedEventRelationship( // they may already have the auth events so don't fail this request util.GetLogger(ctx).WithError(err).Error("Failed to QueryAuthChain") } - res.AuthChain = make([]*gomatrixserverlib.Event, len(queryRes.AuthChain)) + res.AuthChain = make(gomatrixserverlib.EventJSONs, len(queryRes.AuthChain)) for i := range queryRes.AuthChain { - res.AuthChain[i] = queryRes.AuthChain[i].Unwrap() + res.AuthChain[i] = queryRes.AuthChain[i].JSON() + } + + res.Events = make(gomatrixserverlib.EventJSONs, len(res.ParsedEvents)) + for i := range res.ParsedEvents { + res.Events[i] = res.ParsedEvents[i].JSON() } return util.JSONResponse{ Code: 200, - JSON: res, + JSON: res.MSC2836EventRelationshipsResponse, } } -func (rc *reqCtx) process() (*gomatrixserverlib.MSC2836EventRelationshipsResponse, *util.JSONResponse) { - var res gomatrixserverlib.MSC2836EventRelationshipsResponse +func (rc *reqCtx) process() (*MSC2836EventRelationshipsResponse, *util.JSONResponse) { + var res MSC2836EventRelationshipsResponse var returnEvents []*gomatrixserverlib.HeaderedEvent // Can the user see (according to history visibility) event_id? If no, reject the request, else continue. event := rc.getLocalEvent(rc.req.EventID) @@ -290,11 +301,11 @@ func (rc *reqCtx) process() (*gomatrixserverlib.MSC2836EventRelationshipsRespons ) returnEvents = append(returnEvents, events...) } - res.Events = make([]*gomatrixserverlib.Event, len(returnEvents)) + res.ParsedEvents = make([]*gomatrixserverlib.Event, len(returnEvents)) for i, ev := range returnEvents { // for each event, extract the children_count | hash and add it as unsigned data. rc.addChildMetadata(ev) - res.Events[i] = ev.Unwrap() + res.ParsedEvents[i] = ev.Unwrap() } res.Limited = remaining == 0 || walkLimited return &res, nil @@ -357,7 +368,7 @@ func (rc *reqCtx) fetchUnknownEvent(eventID, roomID string) *gomatrixserverlib.H continue } rc.injectResponseToRoomserver(res) - for _, ev := range res.Events { + for _, ev := range res.ParsedEvents { if ev.EventID() == eventID { return ev.Headered(ev.Version()) } @@ -384,7 +395,7 @@ func (rc *reqCtx) includeChildren(db Database, parentID string, limit int, recen if rc.hasUnexploredChildren(parentID) { // we need to do a remote request to pull in the children as we are missing them locally. serversToQuery := rc.getServersForEventID(parentID) - var result *gomatrixserverlib.MSC2836EventRelationshipsResponse + var result *MSC2836EventRelationshipsResponse for _, srv := range serversToQuery { res, err := rc.fsAPI.MSC2836EventRelationships(rc.ctx, srv, gomatrixserverlib.MSC2836EventRelationshipsRequest{ EventID: parentID, @@ -397,7 +408,12 @@ func (rc *reqCtx) includeChildren(db Database, parentID string, limit int, recen if err != nil { util.GetLogger(rc.ctx).WithError(err).WithField("server", srv).Error("includeChildren: failed to call MSC2836EventRelationships") } else { - result = &res + mscRes := &MSC2836EventRelationshipsResponse{ + MSC2836EventRelationshipsResponse: res, + } + mscRes.ParsedEvents = res.Events.UntrustedEvents(rc.roomVersion) + mscRes.ParsedAuthChain = res.AuthChain.UntrustedEvents(rc.roomVersion) + result = mscRes break } } @@ -467,7 +483,7 @@ func walkThread( } // MSC2836EventRelationships performs an /event_relationships request to a remote server -func (rc *reqCtx) MSC2836EventRelationships(eventID string, srv gomatrixserverlib.ServerName, ver gomatrixserverlib.RoomVersion) (*gomatrixserverlib.MSC2836EventRelationshipsResponse, error) { +func (rc *reqCtx) MSC2836EventRelationships(eventID string, srv gomatrixserverlib.ServerName, ver gomatrixserverlib.RoomVersion) (*MSC2836EventRelationshipsResponse, error) { res, err := rc.fsAPI.MSC2836EventRelationships(rc.ctx, srv, gomatrixserverlib.MSC2836EventRelationshipsRequest{ EventID: eventID, DepthFirst: rc.req.DepthFirst, @@ -481,7 +497,12 @@ func (rc *reqCtx) MSC2836EventRelationships(eventID string, srv gomatrixserverli util.GetLogger(rc.ctx).WithError(err).Error("Failed to call MSC2836EventRelationships") return nil, err } - return &res, nil + mscRes := &MSC2836EventRelationshipsResponse{ + MSC2836EventRelationshipsResponse: res, + } + mscRes.ParsedEvents = res.Events.UntrustedEvents(ver) + mscRes.ParsedAuthChain = res.AuthChain.UntrustedEvents(ver) + return mscRes, nil } @@ -550,12 +571,12 @@ func (rc *reqCtx) getServersForEventID(eventID string) []gomatrixserverlib.Serve return serversToQuery } -func (rc *reqCtx) remoteEventRelationships(eventID string) *gomatrixserverlib.MSC2836EventRelationshipsResponse { +func (rc *reqCtx) remoteEventRelationships(eventID string) *MSC2836EventRelationshipsResponse { if rc.isFederatedRequest { return nil // we don't query remote servers for remote requests } serversToQuery := rc.getServersForEventID(eventID) - var res *gomatrixserverlib.MSC2836EventRelationshipsResponse + var res *MSC2836EventRelationshipsResponse var err error for _, srv := range serversToQuery { res, err = rc.MSC2836EventRelationships(eventID, srv, rc.roomVersion) @@ -577,7 +598,7 @@ func (rc *reqCtx) lookForEvent(eventID string) *gomatrixserverlib.HeaderedEvent if queryRes != nil { // inject all the events into the roomserver then return the event in question rc.injectResponseToRoomserver(queryRes) - for _, ev := range queryRes.Events { + for _, ev := range queryRes.ParsedEvents { if ev.EventID() == eventID && rc.req.RoomID == ev.RoomID() { return ev.Headered(ev.Version()) } @@ -619,12 +640,12 @@ func (rc *reqCtx) getLocalEvent(eventID string) *gomatrixserverlib.HeaderedEvent // injectResponseToRoomserver injects the events // into the roomserver as KindOutlier, with auth chains. -func (rc *reqCtx) injectResponseToRoomserver(res *gomatrixserverlib.MSC2836EventRelationshipsResponse) { - var stateEvents []*gomatrixserverlib.Event +func (rc *reqCtx) injectResponseToRoomserver(res *MSC2836EventRelationshipsResponse) { + var stateEvents gomatrixserverlib.EventJSONs var messageEvents []*gomatrixserverlib.Event - for _, ev := range res.Events { + for _, ev := range res.ParsedEvents { if ev.StateKey() != nil { - stateEvents = append(stateEvents, ev) + stateEvents = append(stateEvents, ev.JSON()) } else { messageEvents = append(messageEvents, ev) } @@ -633,7 +654,7 @@ func (rc *reqCtx) injectResponseToRoomserver(res *gomatrixserverlib.MSC2836Event AuthEvents: res.AuthChain, StateEvents: stateEvents, } - eventsInOrder, err := respState.Events() + eventsInOrder, err := respState.Events(rc.roomVersion) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("failed to calculate order to send events in MSC2836EventRelationshipsResponse") return diff --git a/syncapi/internal/keychange.go b/syncapi/internal/keychange.go index fa1064b70..37a9e2d39 100644 --- a/syncapi/internal/keychange.go +++ b/syncapi/internal/keychange.go @@ -282,6 +282,8 @@ func membershipEvents(res *types.Response) (joinUserIDs, leaveUserIDs []string) if ev.Type == gomatrixserverlib.MRoomMember && ev.StateKey != nil { if strings.Contains(string(ev.Content), `"join"`) { joinUserIDs = append(joinUserIDs, *ev.StateKey) + } else if strings.Contains(string(ev.Content), `"invite"`) { + joinUserIDs = append(joinUserIDs, *ev.StateKey) } else if strings.Contains(string(ev.Content), `"leave"`) { leaveUserIDs = append(leaveUserIDs, *ev.StateKey) } else if strings.Contains(string(ev.Content), `"ban"`) { diff --git a/sytest-whitelist b/sytest-whitelist index 7d26c610e..04b1bbf36 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -590,3 +590,5 @@ Can reject invites over federation for rooms with version 9 Can receive redactions from regular users over federation in room version 9 Forward extremities remain so even after the next events are populated as outliers If a device list update goes missing, the server resyncs on the next one +uploading self-signing key notifies over federation +uploading signed devices gets propagated over federation