MirrorHub/dendrite
0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-05-20 06:13:48 +02:00
Code Issues Releases Wiki Activity

Merge branch 'matrix-org:main' into presence-optimization

Browse source
This commit is contained in:
jjj333_p 2024-03-23 03:11:28 -10:00 committed by GitHub
parent b436d43133 ad0a7d09e8
commit cb76da4981
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
30 changed files with 1702 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
.golangci.yml
View file

@ -6,7 +6,7 @@ run:
concurrency: 4
# timeout for analysis, e.g. 30s, 5m, default is 1m
deadline: 30m
timeout: 5m
# exit code when at least one issue was found, default is 1
issues-exit-code: 1
@ -18,24 +18,6 @@ run:
#build-tags:
# - mytag
# which dirs to skip: they won't be analyzed;
# can use regexp here: generated.*, regexp is applied on full path;
# default value is empty list, but next dirs are always skipped independently
# from this option's value:
# vendor$, third_party$, testdata$, examples$, Godeps$, builtin$
skip-dirs:
- bin
- docs
# which files to skip: they will be analyzed, but issues from them
# won't be reported. Default value is empty list, but there is
# no need to include all autogenerated files, we confidently recognize
# autogenerated files. If it's not please let us know.
skip-files:
- ".*\\.md$"
- ".*\\.sh$"
- "^cmd/syncserver-integration-tests/testdata.go$"
# by default isn't set. If set we pass it to "go list -mod={option}". From "go help modules":
# If invoked with -mod=readonly, the go command is disallowed from the implicit
# automatic updating of go.mod described above. Instead, it fails when any changes
@ -50,7 +32,8 @@ run:
# output configuration options
output:
# colored-line-number|line-number|json|tab|checkstyle|code-climate, default is "colored-line-number"
format: colored-line-number
formats:
- format: colored-line-number
# print lines of code with issue, default is true
print-issued-lines: true
@ -79,9 +62,8 @@ linters-settings:
# see https://github.com/kisielk/errcheck#excluding-functions for details
#exclude: /path/to/file.txt
govet:
# report about shadowed variables
check-shadowing: true
enable:
- shadow
# settings per analyzer
settings:
printf: # analyzer name, run `go tool vet help` to see all analyzers
@ -217,6 +199,24 @@ linters:
issues:
# which files to skip: they will be analyzed, but issues from them
# won't be reported. Default value is empty list, but there is
# no need to include all autogenerated files, we confidently recognize
# autogenerated files. If it's not please let us know.
exclude-files:
- ".*\\.md$"
- ".*\\.sh$"
- "^cmd/syncserver-integration-tests/testdata.go$"
# which dirs to skip: they won't be analyzed;
# can use regexp here: generated.*, regexp is applied on full path;
# default value is empty list, but next dirs are always skipped independently
# from this option's value:
# vendor$, third_party$, testdata$, examples$, Godeps$, builtin$
exclude-dirs:
- bin
- docs
# List of regexps of issue texts to exclude, empty list by default.
# But independently from this option we use default exclude patterns,
# it can be disabled by `exclude-use-default: false`. To list all

381
clientapi/admin_test.go
View file

@ -2,10 +2,12 @@ package clientapi
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
@ -1092,3 +1094,382 @@ func TestAdminMarkAsStale(t *testing.T) {
}
})
}
func TestAdminQueryEventReports(t *testing.T) {
alice := test.NewUser(t, test.WithAccountType(uapi.AccountTypeAdmin))
bob := test.NewUser(t)
room := test.NewRoom(t, alice)
room2 := test.NewRoom(t, alice)
// room2 has a name and canonical alias
room2.CreateAndInsert(t, alice, spec.MRoomName, map[string]string{"name": "Testing"}, test.WithStateKey(""))
room2.CreateAndInsert(t, alice, spec.MRoomCanonicalAlias, map[string]string{"alias": "#testing"}, test.WithStateKey(""))
// Join the rooms with Bob
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
room2.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
// Create a few events to report
eventsToReportPerRoom := make(map[string][]string)
for i := 0; i < 10; i++ {
ev1 := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
ev2 := room2.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
eventsToReportPerRoom[room.ID] = append(eventsToReportPerRoom[room.ID], ev1.EventID())
eventsToReportPerRoom[room2.ID] = append(eventsToReportPerRoom[room2.ID], ev2.EventID())
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
/*if dbType == test.DBTypeSQLite {
t.Skip()
}*/
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use an actual roomserver for this
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room2.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
alice: {},
bob: {},
}
createAccessTokens(t, accessTokens, userAPI, processCtx.Context(), routers)
reqBody := map[string]any{
"reason": "baaad",
"score": -100,
}
body, err := json.Marshal(reqBody)
if err != nil {
t.Fatal(err)
}
w := httptest.NewRecorder()
var req *http.Request
// Report all events
for roomID, eventIDs := range eventsToReportPerRoom {
for _, eventID := range eventIDs {
req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/_matrix/client/v3/rooms/%s/report/%s", roomID, eventID), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[bob].accessToken)
routers.Client.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected report to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
}
}
type response struct {
EventReports []api.QueryAdminEventReportsResponse `json:"event_reports"`
Total int64 `json:"total"`
NextToken *int64 `json:"next_token,omitempty"`
}
t.Run("Can query all reports", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/_synapse/admin/v1/event_reports", strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting reports to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
var resp response
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
wantCount := 20
// Only validating the count
if len(resp.EventReports) != wantCount {
t.Fatalf("expected %d events, got %d", wantCount, len(resp.EventReports))
}
if resp.Total != int64(wantCount) {
t.Fatalf("expected total to be %d, got %d", wantCount, resp.Total)
}
})
t.Run("Can filter on room", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, fmt.Sprintf("/_synapse/admin/v1/event_reports?room_id=%s", room.ID), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting reports to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
var resp response
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
wantCount := 10
// Only validating the count
if len(resp.EventReports) != wantCount {
t.Fatalf("expected %d events, got %d", wantCount, len(resp.EventReports))
}
if resp.Total != int64(wantCount) {
t.Fatalf("expected total to be %d, got %d", wantCount, resp.Total)
}
})
t.Run("Can filter on user_id", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, fmt.Sprintf("/_synapse/admin/v1/event_reports?user_id=%s", "@doesnotexist:test"), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting reports to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
var resp response
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
// The user does not exist, so we expect no results
wantCount := 0
// Only validating the count
if len(resp.EventReports) != wantCount {
t.Fatalf("expected %d events, got %d", wantCount, len(resp.EventReports))
}
if resp.Total != int64(wantCount) {
t.Fatalf("expected total to be %d, got %d", wantCount, resp.Total)
}
})
t.Run("Can set direction=f", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, fmt.Sprintf("/_synapse/admin/v1/event_reports?room_id=%s&dir=f", room.ID), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting reports to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
var resp response
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
wantCount := 10
// Only validating the count
if len(resp.EventReports) != wantCount {
t.Fatalf("expected %d events, got %d", wantCount, len(resp.EventReports))
}
if resp.Total != int64(wantCount) {
t.Fatalf("expected total to be %d, got %d", wantCount, resp.Total)
}
// we now should have the first reported event
wantEventID := eventsToReportPerRoom[room.ID][0]
gotEventID := resp.EventReports[0].EventID
if gotEventID != wantEventID {
t.Fatalf("expected eventID to be %v, got %v", wantEventID, gotEventID)
}
})
t.Run("Can limit and paginate", func(t *testing.T) {
var from int64 = 0
var limit int64 = 5
var wantTotal int64 = 10 // We expect there to be 10 events in total
var resp response
for from+limit <= wantTotal {
resp = response{}
t.Logf("Getting reports starting from %d", from)
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, fmt.Sprintf("/_synapse/admin/v1/event_reports?room_id=%s&limit=%d&from=%d", room2.ID, limit, from), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting reports to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
wantCount := 5 // we are limited to 5
if len(resp.EventReports) != wantCount {
t.Fatalf("expected %d events, got %d", wantCount, len(resp.EventReports))
}
if resp.Total != int64(wantTotal) {
t.Fatalf("expected total to be %d, got %d", wantCount, resp.Total)
}
// We've reached the end
if (from + int64(len(resp.EventReports))) == wantTotal {
return
}
// The next_token should be set
if resp.NextToken == nil {
t.Fatal("expected nextToken to be set")
}
from = *resp.NextToken
}
})
})
}
func TestEventReportsGetDelete(t *testing.T) {
alice := test.NewUser(t, test.WithAccountType(uapi.AccountTypeAdmin))
bob := test.NewUser(t)
room := test.NewRoom(t, alice)
// Add a name and alias
roomName := "Testing"
alias := "#testing"
room.CreateAndInsert(t, alice, spec.MRoomName, map[string]string{"name": roomName}, test.WithStateKey(""))
room.CreateAndInsert(t, alice, spec.MRoomCanonicalAlias, map[string]string{"alias": alias}, test.WithStateKey(""))
// Join the rooms with Bob
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
// Create a few events to report
eventIDToReport := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use an actual roomserver for this
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
alice: {},
bob: {},
}
createAccessTokens(t, accessTokens, userAPI, processCtx.Context(), routers)
reqBody := map[string]any{
"reason": "baaad",
"score": -100,
}
body, err := json.Marshal(reqBody)
if err != nil {
t.Fatal(err)
}
w := httptest.NewRecorder()
var req *http.Request
// Report the event
req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/_matrix/client/v3/rooms/%s/report/%s", room.ID, eventIDToReport.EventID()), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[bob].accessToken)
routers.Client.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected report to succeed, got HTTP %d instead: %s", w.Code, w.Body.String())
}
t.Run("Can not query with invalid ID", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/_synapse/admin/v1/event_reports/abc", strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("expected getting report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
t.Run("Can query with valid ID", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/_synapse/admin/v1/event_reports/1", strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
resp := api.QueryAdminEventReportResponse{}
if err = json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
// test a few things
if resp.EventID != eventIDToReport.EventID() {
t.Fatalf("expected eventID to be %s, got %s instead", eventIDToReport.EventID(), resp.EventID)
}
if resp.RoomName != roomName {
t.Fatalf("expected roomName to be %s, got %s instead", roomName, resp.RoomName)
}
if resp.CanonicalAlias != alias {
t.Fatalf("expected alias to be %s, got %s instead", alias, resp.CanonicalAlias)
}
if reflect.DeepEqual(resp.EventJSON, eventIDToReport.JSON()) {
t.Fatal("mismatching eventJSON")
}
})
t.Run("Can delete with a valid ID", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodDelete, "/_synapse/admin/v1/event_reports/1", strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected getting report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
t.Run("Can not query deleted report", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/_synapse/admin/v1/event_reports/1", strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[alice].accessToken)
routers.SynapseAdmin.ServeHTTP(w, req)
if w.Code == http.StatusOK {
t.Fatalf("expected getting report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
})
}

89
clientapi/clientapi_test.go
View file

@ -2346,3 +2346,92 @@ func TestCreateRoomInvite(t *testing.T) {
}
})
}
func TestReportEvent(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := test.NewUser(t)
room := test.NewRoom(t, alice)
room.CreateAndInsert(t, charlie, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(charlie.ID))
eventToReport := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": "hello world"})
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use an actual roomserver for this
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
alice: {},
bob: {},
charlie: {},
}
createAccessTokens(t, accessTokens, userAPI, processCtx.Context(), routers)
reqBody := map[string]any{
"reason": "baaad",
"score": -100,
}
body, err := json.Marshal(reqBody)
if err != nil {
t.Fatal(err)
}
w := httptest.NewRecorder()
var req *http.Request
t.Run("Bob is not joined and should not be able to report the event", func(t *testing.T) {
req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/_matrix/client/v3/rooms/%s/report/%s", room.ID, eventToReport.EventID()), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[bob].accessToken)
routers.Client.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Fatalf("expected report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
t.Run("Charlie is joined but the event does not exist", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/_matrix/client/v3/rooms/%s/report/$doesNotExist", room.ID), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[charlie].accessToken)
routers.Client.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Fatalf("expected report to fail, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
t.Run("Charlie is joined and allowed to report the event", func(t *testing.T) {
w = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodPost, fmt.Sprintf("/_matrix/client/v3/rooms/%s/report/%s", room.ID, eventToReport.EventID()), strings.NewReader(string(body)))
req.Header.Set("Authorization", "Bearer "+accessTokens[charlie].accessToken)
routers.Client.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected report to be successful, got HTTP %d instead: %s", w.Code, w.Body.String())
}
})
})
}

90
clientapi/routing/admin.go
View file

@ -495,3 +495,93 @@ func AdminDownloadState(req *http.Request, device *api.Device, rsAPI roomserverA
JSON: struct{}{},
}
}
// GetEventReports returns reported events for a given user/room.
func GetEventReports(
req *http.Request,
rsAPI roomserverAPI.ClientRoomserverAPI,
from, limit uint64,
backwards bool,
userID, roomID string,
) util.JSONResponse {
eventReports, count, err := rsAPI.QueryAdminEventReports(req.Context(), from, limit, backwards, userID, roomID)
if err != nil {
logrus.WithError(err).Error("failed to query event reports")
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
resp := map[string]any{
"event_reports": eventReports,
"total": count,
}
// Add a next_token if there are still reports
if int64(from+limit) < count {
resp["next_token"] = int(from) + len(eventReports)
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: resp,
}
}
func GetEventReport(req *http.Request, rsAPI roomserverAPI.ClientRoomserverAPI, reportID string) util.JSONResponse {
parsedReportID, err := strconv.ParseUint(reportID, 10, 64)
if err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
// Given this is an admin endpoint, let them know what didn't work.
JSON: spec.InvalidParam(err.Error()),
}
}
report, err := rsAPI.QueryAdminEventReport(req.Context(), parsedReportID)
if err != nil {
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.Unknown(err.Error()),
}
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: report,
}
}
func DeleteEventReport(req *http.Request, rsAPI roomserverAPI.ClientRoomserverAPI, reportID string) util.JSONResponse {
parsedReportID, err := strconv.ParseUint(reportID, 10, 64)
if err != nil {
return util.JSONResponse{
Code: http.StatusBadRequest,
// Given this is an admin endpoint, let them know what didn't work.
JSON: spec.InvalidParam(err.Error()),
}
}
err = rsAPI.PerformAdminDeleteEventReport(req.Context(), parsedReportID)
if err != nil {
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.Unknown(err.Error()),
}
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: struct{}{},
}
}
func parseUint64OrDefault(input string, defaultValue uint64) uint64 {
v, err := strconv.ParseUint(input, 10, 64)
if err != nil {
return defaultValue
}
return v
}

93
clientapi/routing/report_event.go Normal file
View file

@ -0,0 +1,93 @@
// Copyright 2023 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package routing
import (
"net/http"
"github.com/matrix-org/dendrite/clientapi/httputil"
"github.com/matrix-org/dendrite/roomserver/api"
userAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
)
type reportEventRequest struct {
Reason string `json:"reason"`
Score int64 `json:"score"`
}
func ReportEvent(
req *http.Request,
device *userAPI.Device,
roomID, eventID string,
rsAPI api.ClientRoomserverAPI,
) util.JSONResponse {
defer req.Body.Close() // nolint: errcheck
deviceUserID, err := spec.NewUserID(device.UserID, true)
if err != nil {
return util.JSONResponse{
Code: http.StatusForbidden,
JSON: spec.NotFound("You don't have permission to report this event, bad userID"),
}
}
// The requesting user must be a member of the room
errRes := checkMemberInRoom(req.Context(), rsAPI, *deviceUserID, roomID)
if errRes != nil {
return util.JSONResponse{
Code: http.StatusNotFound, // Spec demands this...
JSON: spec.NotFound("The event was not found or you are not joined to the room."),
}
}
// Parse the request
report := reportEventRequest{}
if resErr := httputil.UnmarshalJSONRequest(req, &report); resErr != nil {
return *resErr
}
queryRes := &api.QueryEventsByIDResponse{}
if err = rsAPI.QueryEventsByID(req.Context(), &api.QueryEventsByIDRequest{
RoomID: roomID,
EventIDs: []string{eventID},
}, queryRes); err != nil {
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{Err: err.Error()},
}
}
// No event was found or it was already redacted
if len(queryRes.Events) == 0 || queryRes.Events[0].Redacted() {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: spec.NotFound("The event was not found or you are not joined to the room."),
}
}
_, err = rsAPI.InsertReportedEvent(req.Context(), roomID, eventID, device.UserID, report.Reason, report.Score)
if err != nil {
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{Err: err.Error()},
}
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: struct{}{},
}
}

44
clientapi/routing/routing.go
View file

@ -1523,4 +1523,48 @@ func Setup(
return GetJoinedMembers(req, device, vars["roomID"], rsAPI)
}),
).Methods(http.MethodGet, http.MethodOptions)
v3mux.Handle("/rooms/{roomID}/report/{eventID}",
httputil.MakeAuthAPI("report_event", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.ErrorResponse(err)
}
return ReportEvent(req, device, vars["roomID"], vars["eventID"], rsAPI)
}),
).Methods(http.MethodPost, http.MethodOptions)
synapseAdminRouter.Handle("/admin/v1/event_reports",
httputil.MakeAdminAPI("admin_report_events", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
from := parseUint64OrDefault(req.URL.Query().Get("from"), 0)
limit := parseUint64OrDefault(req.URL.Query().Get("limit"), 100)
dir := req.URL.Query().Get("dir")
userID := req.URL.Query().Get("user_id")
roomID := req.URL.Query().Get("room_id")
// Go backwards if direction is empty or "b"
backwards := dir == "" || dir == "b"
return GetEventReports(req, rsAPI, from, limit, backwards, userID, roomID)
}),
).Methods(http.MethodGet, http.MethodOptions)
synapseAdminRouter.Handle("/admin/v1/event_reports/{reportID}",
httputil.MakeAdminAPI("admin_report_event", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.ErrorResponse(err)
}
return GetEventReport(req, rsAPI, vars["reportID"])
}),
).Methods(http.MethodGet, http.MethodOptions)
synapseAdminRouter.Handle("/admin/v1/event_reports/{reportID}",
httputil.MakeAdminAPI("admin_report_event_delete", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.ErrorResponse(err)
}
return DeleteEventReport(req, rsAPI, vars["reportID"])
}),
).Methods(http.MethodDelete, http.MethodOptions)
}

4
go.mod
View file

@ -9,7 +9,7 @@ require (
github.com/blevesearch/bleve/v2 v2.3.8
github.com/codeclysm/extract v2.2.0+incompatible
github.com/dgraph-io/ristretto v0.1.1
github.com/docker/docker v24.0.7+incompatible
github.com/docker/docker v24.0.9+incompatible
github.com/docker/go-connections v0.4.0
github.com/getsentry/sentry-go v0.14.0
github.com/gologme/log v1.3.0
@ -128,7 +128,7 @@ require (
golang.org/x/text v0.14.0 // indirect
golang.org/x/time v0.5.0 // indirect
golang.org/x/tools v0.12.0 // indirect
google.golang.org/protobuf v1.30.0 // indirect
google.golang.org/protobuf v1.33.0 // indirect
gopkg.in/macaroon.v2 v2.1.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
lukechampine.com/uint128 v1.2.0 // indirect

8
go.sum
View file

@ -89,8 +89,8 @@ github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczC
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=
github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
github.com/docker/docker v24.0.7+incompatible h1:Wo6l37AuwP3JaMnZa226lzVXGA3F9Ig1seQen0cKYlM=
github.com/docker/docker v24.0.7+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
github.com/docker/docker v24.0.9+incompatible h1:HPGzNmwfLZWdxHqK9/II92pyi1EpYKsAqcl4G0Of9v0=
github.com/docker/docker v24.0.9+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@ -464,8 +464,8 @@ gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0/go.mod h1:wa6Ws7BG/ESfp6d
gonum.org/v1/plot v0.0.0-20190515093506-e2840ee46a6b/go.mod h1:Wt8AAjI+ypCyYX3nZBvf6cAIx93T+c/OS2HFAYskSZc=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

6
roomserver/acls/acls.go
View file

@ -32,8 +32,8 @@ import (
const MRoomServerACL = "m.room.server_acl"
type ServerACLDatabase interface {
// GetKnownRooms returns a list of all rooms we know about.
GetKnownRooms(ctx context.Context) ([]string, error)
// RoomsWithACLs returns all room IDs for rooms with ACLs
RoomsWithACLs(ctx context.Context) ([]string, error)
// GetBulkStateContent returns all state events which match a given room ID and a given state key tuple. Both must be satisfied for a match.
// If a tuple has the StateKey of '*' and allowWildcards=true then all state events with the EventType should be returned.
@ -57,7 +57,7 @@ func NewServerACLs(db ServerACLDatabase) *ServerACLs {
}
// Look up all of the rooms that the current state server knows about.
rooms, err := db.GetKnownRooms(ctx)
rooms, err := db.RoomsWithACLs(ctx)
if err != nil {
logrus.WithError(err).Fatalf("Failed to get known rooms")
}

2
roomserver/acls/acls_test.go
View file

@ -116,7 +116,7 @@ var (
type dummyACLDB struct{}
func (d dummyACLDB) GetKnownRooms(ctx context.Context) ([]string, error) {
func (d dummyACLDB) RoomsWithACLs(ctx context.Context) ([]string, error) {
return []string{"1", "2"}, nil
}

13
roomserver/api/api.go
View file

@ -86,6 +86,9 @@ type RoomserverInternalAPI interface {
req *QueryAuthChainRequest,
res *QueryAuthChainResponse,
) error
// RoomsWithACLs returns all room IDs for rooms with ACLs
RoomsWithACLs(ctx context.Context) ([]string, error)
}
type UserRoomPrivateKeyCreator interface {
@ -220,6 +223,7 @@ type ClientRoomserverAPI interface {
UserRoomPrivateKeyCreator
QueryRoomHierarchyAPI
DefaultRoomVersionAPI
QueryMembershipForUser(ctx context.Context, req *QueryMembershipForUserRequest, res *QueryMembershipForUserResponse) error
QueryMembershipsForRoom(ctx context.Context, req *QueryMembershipsForRoomRequest, res *QueryMembershipsForRoomResponse) error
QueryRoomsForUser(ctx context.Context, userID spec.UserID, desiredMembership string) ([]spec.RoomID, error)
@ -261,6 +265,15 @@ type ClientRoomserverAPI interface {
RemoveRoomAlias(ctx context.Context, senderID spec.SenderID, alias string) (aliasFound bool, aliasRemoved bool, err error)
SigningIdentityFor(ctx context.Context, roomID spec.RoomID, senderID spec.UserID) (fclient.SigningIdentity, error)
InsertReportedEvent(
ctx context.Context,
roomID, eventID, reportingUserID, reason string,
score int64,
) (int64, error)
QueryAdminEventReports(ctx context.Context, from, limit uint64, backwards bool, userID, roomID string) ([]QueryAdminEventReportsResponse, int64, error)
QueryAdminEventReport(ctx context.Context, reportID uint64) (QueryAdminEventReportResponse, error)
PerformAdminDeleteEventReport(ctx context.Context, reportID uint64) error
}
type UserRoomserverAPI interface {

22
roomserver/api/query.go
View file

@ -346,6 +346,28 @@ type QueryServerBannedFromRoomResponse struct {
Banned bool `json:"banned"`
}
type QueryAdminEventReportsResponse struct {
ID int64 `json:"id"`
Score int64 `json:"score"`
EventNID types.EventNID `json:"-"` // only used to query the state
RoomNID types.RoomNID `json:"-"` // only used to query the state
ReportingUserNID types.EventStateKeyNID `json:"-"` // only used in the DB
SenderNID types.EventStateKeyNID `json:"-"` // only used in the DB
RoomID string `json:"room_id"`
EventID string `json:"event_id"`
UserID string `json:"user_id"` // the user reporting the event
Reason string `json:"reason"`
Sender string `json:"sender"` // the user sending the reported event
CanonicalAlias string `json:"canonical_alias"`
RoomName string `json:"name"`
ReceivedTS spec.Timestamp `json:"received_ts"`
}
type QueryAdminEventReportResponse struct {
QueryAdminEventReportsResponse
EventJSON json.RawMessage `json:"event_json"`
}
// MarshalJSON stringifies the room ID and StateKeyTuple keys so they can be sent over the wire in HTTP API mode.
func (r *QueryBulkStateContentResponse) MarshalJSON() ([]byte, error) {
se := make(map[string]string)

8
roomserver/internal/api.go
View file

@ -340,3 +340,11 @@ func (r *RoomserverInternalAPI) SigningIdentityFor(ctx context.Context, roomID s
func (r *RoomserverInternalAPI) AssignRoomNID(ctx context.Context, roomID spec.RoomID, roomVersion gomatrixserverlib.RoomVersion) (roomNID types.RoomNID, err error) {
return r.DB.AssignRoomNID(ctx, roomID, roomVersion)
}
func (r *RoomserverInternalAPI) InsertReportedEvent(
ctx context.Context,
roomID, eventID, reportingUserID, reason string,
score int64,
) (int64, error) {
return r.DB.InsertReportedEvent(ctx, roomID, eventID, reportingUserID, reason, score)
}

4
roomserver/internal/perform/perform_admin.go
View file

@ -354,3 +354,7 @@ func (r *Admin) PerformAdminDownloadState(
return nil
}
func (r *Admin) PerformAdminDeleteEventReport(ctx context.Context, reportID uint64) error {
return r.DB.AdminDeleteEventReport(ctx, reportID)
}

15
roomserver/internal/query/query.go
View file

@ -1099,3 +1099,18 @@ func (r *Queryer) QueryUserIDForSender(ctx context.Context, roomID spec.RoomID,
return nil, nil
}
// RoomsWithACLs returns all room IDs for rooms with ACLs
func (r *Queryer) RoomsWithACLs(ctx context.Context) ([]string, error) {
return r.DB.RoomsWithACLs(ctx)
}
// QueryAdminEventReports returns event reports given a filter.
func (r *Queryer) QueryAdminEventReports(ctx context.Context, from uint64, limit uint64, backwards bool, userID, roomID string) ([]api.QueryAdminEventReportsResponse, int64, error) {
return r.DB.QueryAdminEventReports(ctx, from, limit, backwards, userID, roomID)
}
// QueryAdminEventReport returns a single event report.
func (r *Queryer) QueryAdminEventReport(ctx context.Context, reportID uint64) (api.QueryAdminEventReportResponse, error) {
return r.DB.QueryAdminEventReport(ctx, reportID)
}

35
roomserver/roomserver_test.go
View file

@ -1284,3 +1284,38 @@ func TestRoomConsumerRecreation(t *testing.T) {
wantAckWait := input.MaximumMissingProcessingTime + (time.Second * 10)
assert.Equal(t, wantAckWait, info.Config.AckWait)
}
func TestRoomsWithACLs(t *testing.T) {
ctx := context.Background()
alice := test.NewUser(t)
noACLRoom := test.NewRoom(t, alice)
aclRoom := test.NewRoom(t, alice)
aclRoom.CreateAndInsert(t, alice, "m.room.server_acl", map[string]any{
"deny": []string{"evilhost.test"},
"allow": []string{"*"},
}, test.WithStateKey(""))
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, closeDB := testrig.CreateConfig(t, dbType)
defer closeDB()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
natsInstance := &jetstream.NATSInstance{}
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
// start JetStream listeners
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
for _, room := range []*test.Room{noACLRoom, aclRoom} {
// Create the rooms
err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false)
assert.NoError(t, err)
}
// Validate that we only have one ACLd room.
roomsWithACLs, err := rsAPI.RoomsWithACLs(ctx)
assert.NoError(t, err)
assert.Equal(t, []string{aclRoom.ID}, roomsWithACLs)
})
}

17
roomserver/storage/interface.go
View file

@ -30,6 +30,7 @@ import (
type Database interface {
UserRoomKeys
ReportedEvents
// Do we support processing input events for more than one room at a time?
SupportsConcurrentRoomInputs() bool
AssignRoomNID(ctx context.Context, roomID spec.RoomID, roomVersion gomatrixserverlib.RoomVersion) (roomNID types.RoomNID, err error)
@ -170,8 +171,6 @@ type Database interface {
GetServerInRoom(ctx context.Context, roomNID types.RoomNID, serverName spec.ServerName) (bool, error)
// GetKnownUsers searches all users that userID knows about.
GetKnownUsers(ctx context.Context, userID, searchString string, limit int) ([]string, error)
// GetKnownRooms returns a list of all rooms we know about.
GetKnownRooms(ctx context.Context) ([]string, error)
// ForgetRoom sets a flag in the membership table, that the user wishes to forget a specific room
ForgetRoom(ctx context.Context, userID, roomID string, forget bool) error
@ -193,6 +192,12 @@ type Database interface {
MaybeRedactEvent(
ctx context.Context, roomInfo *types.RoomInfo, eventNID types.EventNID, event gomatrixserverlib.PDU, plResolver state.PowerLevelResolver, querier api.QuerySenderIDAPI,
) (gomatrixserverlib.PDU, gomatrixserverlib.PDU, error)
// RoomsWithACLs returns all room IDs for rooms with ACLs
RoomsWithACLs(ctx context.Context) ([]string, error)
QueryAdminEventReports(ctx context.Context, from uint64, limit uint64, backwards bool, userID string, roomID string) ([]api.QueryAdminEventReportsResponse, int64, error)
QueryAdminEventReport(ctx context.Context, reportID uint64) (api.QueryAdminEventReportResponse, error)
AdminDeleteEventReport(ctx context.Context, reportID uint64) error
}
type UserRoomKeys interface {
@ -256,3 +261,11 @@ type EventDatabase interface {
) (gomatrixserverlib.PDU, gomatrixserverlib.PDU, error)
StoreEvent(ctx context.Context, event gomatrixserverlib.PDU, roomInfo *types.RoomInfo, eventTypeNID types.EventTypeNID, eventStateKeyNID types.EventStateKeyNID, authEventNIDs []types.EventNID, isRejected bool) (types.EventNID, types.StateAtEvent, error)
}
type ReportedEvents interface {
InsertReportedEvent(
ctx context.Context,
roomID, eventID, reportingUserID, reason string,
score int64,
) (int64, error)
}

30
roomserver/storage/postgres/events_table.go
View file

@ -68,6 +68,10 @@ CREATE TABLE IF NOT EXISTS roomserver_events (
-- Create an index which helps in resolving membership events (event_type_nid = 5) - (used for history visibility)
CREATE INDEX IF NOT EXISTS roomserver_events_memberships_idx ON roomserver_events (room_nid, event_state_key_nid) WHERE (event_type_nid = 5);
-- The following indexes are used by bulkSelectStateEventByNIDSQL
CREATE INDEX IF NOT EXISTS roomserver_event_event_type_nid_idx ON roomserver_events (event_type_nid);
CREATE INDEX IF NOT EXISTS roomserver_event_state_key_nid_idx ON roomserver_events (event_state_key_nid);
`
const insertEventSQL = "" +
@ -147,6 +151,8 @@ const selectRoomNIDsForEventNIDsSQL = "" +
const selectEventRejectedSQL = "" +
"SELECT is_rejected FROM roomserver_events WHERE room_nid = $1 AND event_id = $2"
const selectRoomsWithEventTypeNIDSQL = `SELECT DISTINCT room_nid FROM roomserver_events WHERE event_type_nid = $1`
type eventStatements struct {
insertEventStmt *sql.Stmt
selectEventStmt *sql.Stmt
@ -166,6 +172,7 @@ type eventStatements struct {
selectMaxEventDepthStmt *sql.Stmt
selectRoomNIDsForEventNIDsStmt *sql.Stmt
selectEventRejectedStmt *sql.Stmt
selectRoomsWithEventTypeNIDStmt *sql.Stmt
}
func CreateEventsTable(db *sql.DB) error {
@ -206,6 +213,7 @@ func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
{&s.selectMaxEventDepthStmt, selectMaxEventDepthSQL},
{&s.selectRoomNIDsForEventNIDsStmt, selectRoomNIDsForEventNIDsSQL},
{&s.selectEventRejectedStmt, selectEventRejectedSQL},
{&s.selectRoomsWithEventTypeNIDStmt, selectRoomsWithEventTypeNIDSQL},
}.Prepare(db)
}
@ -582,3 +590,25 @@ func (s *eventStatements) SelectEventRejected(
err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
return
}
func (s *eventStatements) SelectRoomsWithEventTypeNID(
ctx context.Context, txn *sql.Tx, eventTypeNID types.EventTypeNID,
) ([]types.RoomNID, error) {
stmt := sqlutil.TxStmt(txn, s.selectRoomsWithEventTypeNIDStmt)
rows, err := stmt.QueryContext(ctx, eventTypeNID)
defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomsWithEventTypeNID: rows.close() failed")
if err != nil {
return nil, err
}
var roomNIDs []types.RoomNID
var roomNID types.RoomNID
for rows.Next() {
if err := rows.Scan(&roomNID); err != nil {
return nil, err
}
roomNIDs = append(roomNIDs, roomNID)
}
return roomNIDs, rows.Err()
}

221
roomserver/storage/postgres/reported_events_table.go Normal file
View file

@ -0,0 +1,221 @@
// Copyright 2023 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 postgres
import (
"context"
"database/sql"
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib/spec"
)
const reportedEventsScheme = `
CREATE SEQUENCE IF NOT EXISTS roomserver_reported_events_id_seq;
CREATE TABLE IF NOT EXISTS roomserver_reported_events
(
id BIGINT PRIMARY KEY DEFAULT nextval('roomserver_reported_events_id_seq'),
room_nid BIGINT NOT NULL,
event_nid BIGINT NOT NULL,
reporting_user_nid BIGINT NOT NULL, -- the user reporting the event
event_sender_nid BIGINT NOT NULL, -- the user who sent the reported event
reason TEXT,
score INTEGER,
received_ts BIGINT NOT NULL
);`
const insertReportedEventSQL = `
INSERT INTO roomserver_reported_events (room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id
`
const selectReportedEventsDescSQL = `
WITH countReports AS (
SELECT count(*) as report_count
FROM roomserver_reported_events
WHERE ($1::BIGINT IS NULL OR room_nid = $1::BIGINT) AND ($2::TEXT IS NULL OR reporting_user_nid = $2::BIGINT)
)
SELECT report_count, id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events, countReports
WHERE ($1::BIGINT IS NULL OR room_nid = $1::BIGINT) AND ($2::TEXT IS NULL OR reporting_user_nid = $2::BIGINT)
ORDER BY received_ts DESC
OFFSET $3
LIMIT $4
`
const selectReportedEventsAscSQL = `
WITH countReports AS (
SELECT count(*) as report_count
FROM roomserver_reported_events
WHERE ($1::BIGINT IS NULL OR room_nid = $1::BIGINT) AND ($2::TEXT IS NULL OR reporting_user_nid = $2::BIGINT)
)
SELECT report_count, id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events, countReports
WHERE ($1::BIGINT IS NULL OR room_nid = $1::BIGINT) AND ($2::TEXT IS NULL OR reporting_user_nid = $2::BIGINT)
ORDER BY received_ts ASC
OFFSET $3
LIMIT $4
`
const selectReportedEventSQL = `
SELECT id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events
WHERE id = $1
`
const deleteReportedEventSQL = `DELETE FROM roomserver_reported_events WHERE id = $1`
type reportedEventsStatements struct {
insertReportedEventsStmt *sql.Stmt
selectReportedEventsDescStmt *sql.Stmt
selectReportedEventsAscStmt *sql.Stmt
selectReportedEventStmt *sql.Stmt
deleteReportedEventStmt *sql.Stmt
}
func CreateReportedEventsTable(db *sql.DB) error {
_, err := db.Exec(reportedEventsScheme)
return err
}
func PrepareReportedEventsTable(db *sql.DB) (tables.ReportedEvents, error) {
s := &reportedEventsStatements{}
return s, sqlutil.StatementList{
{&s.insertReportedEventsStmt, insertReportedEventSQL},
{&s.selectReportedEventsDescStmt, selectReportedEventsDescSQL},
{&s.selectReportedEventsAscStmt, selectReportedEventsAscSQL},
{&s.selectReportedEventStmt, selectReportedEventSQL},
{&s.deleteReportedEventStmt, deleteReportedEventSQL},
}.Prepare(db)
}
func (r *reportedEventsStatements) InsertReportedEvent(
ctx context.Context,
txn *sql.Tx,
roomNID types.RoomNID,
eventNID types.EventNID,
reportingUserID types.EventStateKeyNID,
eventSenderID types.EventStateKeyNID,
reason string,
score int64,
) (int64, error) {
stmt := sqlutil.TxStmt(txn, r.insertReportedEventsStmt)
var reportID int64
err := stmt.QueryRowContext(ctx,
roomNID,
eventNID,
reportingUserID,
eventSenderID,
reason,
score,
spec.AsTimestamp(time.Now()),
).Scan(&reportID)
return reportID, err
}
func (r *reportedEventsStatements) SelectReportedEvents(
ctx context.Context,
txn *sql.Tx,
from, limit uint64,
backwards bool,
reportingUserID types.EventStateKeyNID,
roomNID types.RoomNID,
) ([]api.QueryAdminEventReportsResponse, int64, error) {
var stmt *sql.Stmt
if backwards {
stmt = sqlutil.TxStmt(txn, r.selectReportedEventsDescStmt)
} else {
stmt = sqlutil.TxStmt(txn, r.selectReportedEventsAscStmt)
}
var qryRoomNID *types.RoomNID
if roomNID > 0 {
qryRoomNID = &roomNID
}
var qryReportingUser *types.EventStateKeyNID
if reportingUserID > 0 {
qryReportingUser = &reportingUserID
}
rows, err := stmt.QueryContext(ctx,
qryRoomNID,
qryReportingUser,
from,
limit,
)
if err != nil {
return nil, 0, err
}
defer internal.CloseAndLogIfError(ctx, rows, "SelectReportedEvents: failed to close rows")
var result []api.QueryAdminEventReportsResponse
var row api.QueryAdminEventReportsResponse
var count int64
for rows.Next() {
if err = rows.Scan(
&count,
&row.ID,
&row.RoomNID,
&row.EventNID,
&row.ReportingUserNID,
&row.SenderNID,
&row.Reason,
&row.Score,
&row.ReceivedTS,
); err != nil {
return nil, 0, err
}
result = append(result, row)
}
return result, count, rows.Err()
}
func (r *reportedEventsStatements) SelectReportedEvent(
ctx context.Context,
txn *sql.Tx,
reportID uint64,
) (api.QueryAdminEventReportResponse, error) {
stmt := sqlutil.TxStmt(txn, r.selectReportedEventStmt)
var row api.QueryAdminEventReportResponse
if err := stmt.QueryRowContext(ctx, reportID).Scan(
&row.ID,
&row.RoomNID,
&row.EventNID,
&row.ReportingUserNID,
&row.SenderNID,
&row.Reason,
&row.Score,
&row.ReceivedTS,
); err != nil {
return api.QueryAdminEventReportResponse{}, err
}
return row, nil
}
func (r *reportedEventsStatements) DeleteReportedEvent(ctx context.Context, txn *sql.Tx, reportID uint64) error {
stmt := sqlutil.TxStmt(txn, r.deleteReportedEventStmt)
_, err := stmt.ExecContext(ctx, reportID)
return err
}

22
roomserver/storage/postgres/rooms_table.go
View file

@ -76,9 +76,6 @@ const selectRoomVersionsForRoomNIDsSQL = "" +
const selectRoomInfoSQL = "" +
"SELECT room_version, room_nid, state_snapshot_nid, latest_event_nids FROM roomserver_rooms WHERE room_id = $1"
const selectRoomIDsSQL = "" +
"SELECT room_id FROM roomserver_rooms WHERE array_length(latest_event_nids, 1) > 0"
const bulkSelectRoomIDsSQL = "" +
"SELECT room_id FROM roomserver_rooms WHERE room_nid = ANY($1)"
@ -94,7 +91,6 @@ type roomStatements struct {
updateLatestEventNIDsStmt *sql.Stmt
selectRoomVersionsForRoomNIDsStmt *sql.Stmt
selectRoomInfoStmt *sql.Stmt
selectRoomIDsStmt *sql.Stmt
bulkSelectRoomIDsStmt *sql.Stmt
bulkSelectRoomNIDsStmt *sql.Stmt
}
@ -116,29 +112,11 @@ func PrepareRoomsTable(db *sql.DB) (tables.Rooms, error) {
{&s.updateLatestEventNIDsStmt, updateLatestEventNIDsSQL},
{&s.selectRoomVersionsForRoomNIDsStmt, selectRoomVersionsForRoomNIDsSQL},
{&s.selectRoomInfoStmt, selectRoomInfoSQL},
{&s.selectRoomIDsStmt, selectRoomIDsSQL},
{&s.bulkSelectRoomIDsStmt, bulkSelectRoomIDsSQL},
{&s.bulkSelectRoomNIDsStmt, bulkSelectRoomNIDsSQL},
}.Prepare(db)
}
func (s *roomStatements) SelectRoomIDsWithEvents(ctx context.Context, txn *sql.Tx) ([]string, error) {
stmt := sqlutil.TxStmt(txn, s.selectRoomIDsStmt)
rows, err := stmt.QueryContext(ctx)
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsStmt: rows.close() failed")
var roomIDs []string
var roomID string
for rows.Next() {
if err = rows.Scan(&roomID); err != nil {
return nil, err
}
roomIDs = append(roomIDs, roomID)
}
return roomIDs, rows.Err()
}
func (s *roomStatements) InsertRoomNID(
ctx context.Context, txn *sql.Tx,
roomID string, roomVersion gomatrixserverlib.RoomVersion,

8
roomserver/storage/postgres/storage.go
View file

@ -134,6 +134,9 @@ func (d *Database) create(db *sql.DB) error {
if err := CreateUserRoomKeysTable(db); err != nil {
return err
}
if err := CreateReportedEventsTable(db); err != nil {
return err
}
return nil
}
@ -199,6 +202,10 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
if err != nil {
return err
}
reportedEvents, err := PrepareReportedEventsTable(db)
if err != nil {
return err
}
d.Database = shared.Database{
DB: db,
@ -212,6 +219,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
EventStateKeysTable: eventStateKeys,
PrevEventsTable: prevEvents,
RedactionsTable: redactions,
ReportedEventsTable: reportedEvents,
},
Cache: cache,
Writer: writer,

268
roomserver/storage/shared/storage.go
View file

@ -61,6 +61,7 @@ type EventDatabase struct {
EventStateKeysTable tables.EventStateKeys
PrevEventsTable tables.PreviousEvents
RedactionsTable tables.Redactions
ReportedEventsTable tables.ReportedEvents
}
func (d *Database) SupportsConcurrentRoomInputs() bool {
@ -1625,9 +1626,24 @@ func (d *Database) GetKnownUsers(ctx context.Context, userID, searchString strin
return d.MembershipTable.SelectKnownUsers(ctx, nil, stateKeyNID, searchString, limit)
}
// GetKnownRooms returns a list of all rooms we know about.
func (d *Database) GetKnownRooms(ctx context.Context) ([]string, error) {
return d.RoomsTable.SelectRoomIDsWithEvents(ctx, nil)
func (d *Database) RoomsWithACLs(ctx context.Context) ([]string, error) {
eventTypeNID, err := d.GetOrCreateEventTypeNID(ctx, "m.room.server_acl")
if err != nil {
return nil, err
}
roomNIDs, err := d.EventsTable.SelectRoomsWithEventTypeNID(ctx, nil, eventTypeNID)
if err != nil {
return nil, err
}
roomIDs, err := d.RoomsTable.BulkSelectRoomIDs(ctx, nil, roomNIDs)
if err != nil {
return nil, err
}
return roomIDs, nil
}
// ForgetRoom sets a users room to forgotten
@ -1867,6 +1883,252 @@ func (d *Database) SelectUserIDsForPublicKeys(ctx context.Context, publicKeys ma
return result, err
}
// InsertReportedEvent stores a reported event.
func (d *Database) InsertReportedEvent(
ctx context.Context,
roomID, eventID, reportingUserID, reason string,
score int64,
) (int64, error) {
roomInfo, err := d.roomInfo(ctx, nil, roomID)
if err != nil {
return 0, err
}
if roomInfo == nil {
return 0, fmt.Errorf("room does not exist")
}
events, err := d.eventsFromIDs(ctx, nil, roomInfo, []string{eventID}, NoFilter)
if err != nil {
return 0, err
}
if len(events) == 0 {
return 0, fmt.Errorf("unable to find requested event")
}
stateKeyNIDs, err := d.EventStateKeyNIDs(ctx, []string{reportingUserID, events[0].SenderID().ToUserID().String()})
if err != nil {
return 0, fmt.Errorf("failed to query eventStateKeyNIDs: %w", err)
}
// We expect exactly 2 stateKeyNIDs
if len(stateKeyNIDs) != 2 {
return 0, fmt.Errorf("expected 2 stateKeyNIDs, received %d", len(stateKeyNIDs))
}
var reportID int64
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
reportID, err = d.ReportedEventsTable.InsertReportedEvent(
ctx,
txn,
roomInfo.RoomNID,
events[0].EventNID,
stateKeyNIDs[reportingUserID],
stateKeyNIDs[events[0].SenderID().ToUserID().String()],
reason,
score,
)
if err != nil {
return err
}
return nil
})
return reportID, err
}
// QueryAdminEventReports returns event reports given a filter.
func (d *Database) QueryAdminEventReports(ctx context.Context, from uint64, limit uint64, backwards bool, userID string, roomID string) ([]api.QueryAdminEventReportsResponse, int64, error) {
// Filter on roomID, if requested
var roomNID types.RoomNID
if roomID != "" {
roomInfo, err := d.RoomInfo(ctx, roomID)
if err != nil {
return nil, 0, err
}
roomNID = roomInfo.RoomNID
}
// Same as above, but for userID
var userNID types.EventStateKeyNID
if userID != "" {
stateKeysMap, err := d.EventStateKeyNIDs(ctx, []string{userID})
if err != nil {
return nil, 0, err
}
if len(stateKeysMap) != 1 {
return nil, 0, fmt.Errorf("failed to get eventStateKeyNID for %s", userID)
}
userNID = stateKeysMap[userID]
}
// Query all reported events matching the filters
reports, count, err := d.ReportedEventsTable.SelectReportedEvents(ctx, nil, from, limit, backwards, userNID, roomNID)
if err != nil {
return nil, 0, fmt.Errorf("failed to SelectReportedEvents: %w", err)
}
// TODO: The below code may be inefficient due to many DB round trips and needs to be revisited.
// For the time being, this is "good enough".
qryRoomNIDs := make([]types.RoomNID, 0, len(reports))
qryEventNIDs := make([]types.EventNID, 0, len(reports))
qryStateKeyNIDs := make([]types.EventStateKeyNID, 0, len(reports))
for _, report := range reports {
qryRoomNIDs = append(qryRoomNIDs, report.RoomNID)
qryEventNIDs = append(qryEventNIDs, report.EventNID)
qryStateKeyNIDs = append(qryStateKeyNIDs, report.ReportingUserNID, report.SenderNID)
}
// This also de-dupes the roomIDs, otherwise we would query the same
// roomIDs in GetBulkStateContent multiple times
roomIDs, err := d.RoomsTable.BulkSelectRoomIDs(ctx, nil, qryRoomNIDs)
if err != nil {
return nil, 0, err
}
// TODO: replace this with something more efficient, as it loads the entire state snapshot.
stateContent, err := d.GetBulkStateContent(ctx, roomIDs, []gomatrixserverlib.StateKeyTuple{
{EventType: spec.MRoomName, StateKey: ""},
{EventType: spec.MRoomCanonicalAlias, StateKey: ""},
}, false)
if err != nil {
return nil, 0, err
}
eventIDMap, err := d.EventIDs(ctx, qryEventNIDs)
if err != nil {
logrus.WithError(err).Error("unable to map eventNIDs to eventIDs")
return nil, 0, err
}
if len(eventIDMap) != len(qryEventNIDs) {
return nil, 0, fmt.Errorf("expected %d eventIDs, got %d", len(qryEventNIDs), len(eventIDMap))
}
// Get a map from EventStateKeyNID to userID
userNIDMap, err := d.EventStateKeys(ctx, qryStateKeyNIDs)
if err != nil {
logrus.WithError(err).Error("unable to map userNIDs to userIDs")
return nil, 0, err
}
// Create a cache from roomNID to roomID to avoid hitting the DB again
roomNIDIDCache := make(map[types.RoomNID]string, len(roomIDs))
for i := 0; i < len(reports); i++ {
cachedRoomID := roomNIDIDCache[reports[i].RoomNID]
if cachedRoomID == "" {
// We need to query this again, as we otherwise don't have a way to match roomNID -> roomID
roomIDs, err = d.RoomsTable.BulkSelectRoomIDs(ctx, nil, []types.RoomNID{reports[i].RoomNID})
if err != nil {
return nil, 0, err
}
if len(roomIDs) == 0 || len(roomIDs) > 1 {
logrus.Warnf("unable to map roomNID %d to a roomID, was this room deleted?", roomNID)
continue
}
roomNIDIDCache[reports[i].RoomNID] = roomIDs[0]
cachedRoomID = roomIDs[0]
}
reports[i].EventID = eventIDMap[reports[i].EventNID]
reports[i].RoomID = cachedRoomID
roomName, canonicalAlias := findRoomNameAndCanonicalAlias(stateContent, cachedRoomID)
reports[i].RoomName = roomName
reports[i].CanonicalAlias = canonicalAlias
reports[i].Sender = userNIDMap[reports[i].SenderNID]
reports[i].UserID = userNIDMap[reports[i].ReportingUserNID]
}
return reports, count, nil
}
func (d *Database) QueryAdminEventReport(ctx context.Context, reportID uint64) (api.QueryAdminEventReportResponse, error) {
report, err := d.ReportedEventsTable.SelectReportedEvent(ctx, nil, reportID)
if err != nil {
return api.QueryAdminEventReportResponse{}, err
}
// Get a map from EventStateKeyNID to userID
userNIDMap, err := d.EventStateKeys(ctx, []types.EventStateKeyNID{report.ReportingUserNID, report.SenderNID})
if err != nil {
logrus.WithError(err).Error("unable to map userNIDs to userIDs")
return report, err
}
roomIDs, err := d.RoomsTable.BulkSelectRoomIDs(ctx, nil, []types.RoomNID{report.RoomNID})
if err != nil {
return report, err
}
if len(roomIDs) != 1 {
return report, fmt.Errorf("expected one roomID, got %d", len(roomIDs))
}
// TODO: replace this with something more efficient, as it loads the entire state snapshot.
stateContent, err := d.GetBulkStateContent(ctx, roomIDs, []gomatrixserverlib.StateKeyTuple{
{EventType: spec.MRoomName, StateKey: ""},
{EventType: spec.MRoomCanonicalAlias, StateKey: ""},
}, false)
if err != nil {
return report, err
}
eventIDMap, err := d.EventIDs(ctx, []types.EventNID{report.EventNID})
if err != nil {
logrus.WithError(err).Error("unable to map eventNIDs to eventIDs")
return report, err
}
if len(eventIDMap) != 1 {
return report, fmt.Errorf("expected %d eventIDs, got %d", 1, len(eventIDMap))
}
eventJSONs, err := d.EventJSONTable.BulkSelectEventJSON(ctx, nil, []types.EventNID{report.EventNID})
if err != nil {
return report, err
}
if len(eventJSONs) != 1 {
return report, fmt.Errorf("expected %d eventJSONs, got %d", 1, len(eventJSONs))
}
roomName, canonicalAlias := findRoomNameAndCanonicalAlias(stateContent, roomIDs[0])
report.Sender = userNIDMap[report.SenderNID]
report.UserID = userNIDMap[report.ReportingUserNID]
report.RoomID = roomIDs[0]
report.RoomName = roomName
report.CanonicalAlias = canonicalAlias
report.EventID = eventIDMap[report.EventNID]
report.EventJSON = eventJSONs[0].EventJSON
return report, nil
}
func (d *Database) AdminDeleteEventReport(ctx context.Context, reportID uint64) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.ReportedEventsTable.DeleteReportedEvent(ctx, txn, reportID)
})
}
// findRoomNameAndCanonicalAlias loops over events to find the corresponding room name and canonicalAlias
// for a given roomID.
func findRoomNameAndCanonicalAlias(events []tables.StrippedEvent, roomID string) (name, canonicalAlias string) {
for _, ev := range events {
if ev.RoomID != roomID {
continue
}
if ev.EventType == spec.MRoomName {
name = ev.ContentValue
}
if ev.EventType == spec.MRoomCanonicalAlias {
canonicalAlias = ev.ContentValue
}
// We found both wanted values, break the loop
if name != "" && canonicalAlias != "" {
break
}
}
return name, canonicalAlias
}
// FIXME TODO: Remove all this - horrible dupe with roomserver/state. Can't use the original impl because of circular loops
// it should live in this package!

34
roomserver/storage/sqlite3/events_table.go
View file

@ -44,6 +44,14 @@ const eventsSchema = `
auth_event_nids TEXT NOT NULL DEFAULT '[]',
is_rejected BOOLEAN NOT NULL DEFAULT FALSE
);
-- Create an index which helps in resolving membership events (event_type_nid = 5) - (used for history visibility)
CREATE INDEX IF NOT EXISTS roomserver_events_memberships_idx ON roomserver_events (room_nid, event_state_key_nid) WHERE (event_type_nid = 5);
-- The following indexes are used by bulkSelectStateEventByNIDSQL
CREATE INDEX IF NOT EXISTS roomserver_event_event_type_nid_idx ON roomserver_events (event_type_nid);
CREATE INDEX IF NOT EXISTS roomserver_event_state_key_nid_idx ON roomserver_events (event_state_key_nid);
`
const insertEventSQL = `
@ -120,6 +128,8 @@ const selectRoomNIDsForEventNIDsSQL = "" +
const selectEventRejectedSQL = "" +
"SELECT is_rejected FROM roomserver_events WHERE room_nid = $1 AND event_id = $2"
const selectRoomsWithEventTypeNIDSQL = `SELECT DISTINCT room_nid FROM roomserver_events WHERE event_type_nid = $1`
type eventStatements struct {
db *sql.DB
insertEventStmt *sql.Stmt
@ -135,6 +145,7 @@ type eventStatements struct {
bulkSelectStateAtEventAndReferenceStmt *sql.Stmt
bulkSelectEventIDStmt *sql.Stmt
selectEventRejectedStmt *sql.Stmt
selectRoomsWithEventTypeNIDStmt *sql.Stmt
//bulkSelectEventNIDStmt *sql.Stmt
//bulkSelectUnsentEventNIDStmt *sql.Stmt
//selectRoomNIDsForEventNIDsStmt *sql.Stmt
@ -192,6 +203,7 @@ func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
//{&s.bulkSelectUnsentEventNIDStmt, bulkSelectUnsentEventNIDSQL},
//{&s.selectRoomNIDForEventNIDStmt, selectRoomNIDForEventNIDSQL},
{&s.selectEventRejectedStmt, selectEventRejectedSQL},
{&s.selectRoomsWithEventTypeNIDStmt, selectRoomsWithEventTypeNIDSQL},
}.Prepare(db)
}
@ -682,3 +694,25 @@ func (s *eventStatements) SelectEventRejected(
err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
return
}
func (s *eventStatements) SelectRoomsWithEventTypeNID(
ctx context.Context, txn *sql.Tx, eventTypeNID types.EventTypeNID,
) ([]types.RoomNID, error) {
stmt := sqlutil.TxStmt(txn, s.selectRoomsWithEventTypeNIDStmt)
rows, err := stmt.QueryContext(ctx, eventTypeNID)
defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomsWithEventTypeNID: rows.close() failed")
if err != nil {
return nil, err
}
var roomNIDs []types.RoomNID
var roomNID types.RoomNID
for rows.Next() {
if err := rows.Scan(&roomNID); err != nil {
return nil, err
}
roomNIDs = append(roomNIDs, roomNID)
}
return roomNIDs, rows.Err()
}

221
roomserver/storage/sqlite3/reported_events_table.go Normal file
View file

@ -0,0 +1,221 @@
// Copyright 2023 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 sqlite3
import (
"context"
"database/sql"
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib/spec"
)
const reportedEventsScheme = `
CREATE TABLE IF NOT EXISTS roomserver_reported_events
(
id INTEGER PRIMARY KEY AUTOINCREMENT,
room_nid INTEGER NOT NULL,
event_nid INTEGER NOT NULL,
reporting_user_nid INTEGER NOT NULL, -- the user reporting the event
event_sender_nid INTEGER NOT NULL, -- the user who sent the reported event
reason TEXT,
score INTEGER,
received_ts INTEGER NOT NULL
);`
const insertReportedEventSQL = `
INSERT INTO roomserver_reported_events (room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id
`
const selectReportedEventsDescSQL = `
WITH countReports AS (
SELECT count(*) as report_count
FROM roomserver_reported_events
WHERE ($1 IS NULL OR room_nid = $1) AND ($2 IS NULL OR reporting_user_nid = $2)
)
SELECT report_count, id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events, countReports
WHERE ($1 IS NULL OR room_nid = $1) AND ($2 IS NULL OR reporting_user_nid = $2)
ORDER BY received_ts DESC
LIMIT $3
OFFSET $4
`
const selectReportedEventsAscSQL = `
WITH countReports AS (
SELECT count(*) as report_count
FROM roomserver_reported_events
WHERE ($1 IS NULL OR room_nid = $1) AND ($2 IS NULL OR reporting_user_nid = $2)
)
SELECT report_count, id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events, countReports
WHERE ($1 IS NULL OR room_nid = $1) AND ($2 IS NULL OR reporting_user_nid = $2)
ORDER BY received_ts ASC
LIMIT $3
OFFSET $4
`
const selectReportedEventSQL = `
SELECT id, room_nid, event_nid, reporting_user_nid, event_sender_nid, reason, score, received_ts
FROM roomserver_reported_events
WHERE id = $1
`
const deleteReportedEventSQL = `DELETE FROM roomserver_reported_events WHERE id = $1`
type reportedEventsStatements struct {
insertReportedEventsStmt *sql.Stmt
selectReportedEventsDescStmt *sql.Stmt
selectReportedEventsAscStmt *sql.Stmt
selectReportedEventStmt *sql.Stmt
deleteReportedEventStmt *sql.Stmt
}
func CreateReportedEventsTable(db *sql.DB) error {
_, err := db.Exec(reportedEventsScheme)
return err
}
func PrepareReportedEventsTable(db *sql.DB) (tables.ReportedEvents, error) {
s := &reportedEventsStatements{}
return s, sqlutil.StatementList{
{&s.insertReportedEventsStmt, insertReportedEventSQL},
{&s.selectReportedEventsDescStmt, selectReportedEventsDescSQL},
{&s.selectReportedEventsAscStmt, selectReportedEventsAscSQL},
{&s.selectReportedEventStmt, selectReportedEventSQL},
{&s.deleteReportedEventStmt, deleteReportedEventSQL},
}.Prepare(db)
}
func (r *reportedEventsStatements) InsertReportedEvent(
ctx context.Context,
txn *sql.Tx,
roomNID types.RoomNID,
eventNID types.EventNID,
reportingUserID types.EventStateKeyNID,
eventSenderID types.EventStateKeyNID,
reason string,
score int64,
) (int64, error) {
stmt := sqlutil.TxStmt(txn, r.insertReportedEventsStmt)
var reportID int64
err := stmt.QueryRowContext(ctx,
roomNID,
eventNID,
reportingUserID,
eventSenderID,
reason,
score,
spec.AsTimestamp(time.Now()),
).Scan(&reportID)
return reportID, err
}
func (r *reportedEventsStatements) SelectReportedEvents(
ctx context.Context,
txn *sql.Tx,
from, limit uint64,
backwards bool,
reportingUserID types.EventStateKeyNID,
roomNID types.RoomNID,
) ([]api.QueryAdminEventReportsResponse, int64, error) {
var stmt *sql.Stmt
if backwards {
stmt = sqlutil.TxStmt(txn, r.selectReportedEventsDescStmt)
} else {
stmt = sqlutil.TxStmt(txn, r.selectReportedEventsAscStmt)
}
var qryRoomNID *types.RoomNID
if roomNID > 0 {
qryRoomNID = &roomNID
}
var qryReportingUser *types.EventStateKeyNID
if reportingUserID > 0 {
qryReportingUser = &reportingUserID
}
rows, err := stmt.QueryContext(ctx,
qryRoomNID,
qryReportingUser,
limit,
from,
)
if err != nil {
return nil, 0, err
}
defer internal.CloseAndLogIfError(ctx, rows, "SelectReportedEvents: failed to close rows")
var result []api.QueryAdminEventReportsResponse
var row api.QueryAdminEventReportsResponse
var count int64
for rows.Next() {
if err = rows.Scan(
&count,
&row.ID,
&row.RoomNID,
&row.EventNID,
&row.ReportingUserNID,
&row.SenderNID,
&row.Reason,
&row.Score,
&row.ReceivedTS,
); err != nil {
return nil, 0, err
}
result = append(result, row)
}
return result, count, rows.Err()
}
func (r *reportedEventsStatements) SelectReportedEvent(
ctx context.Context,
txn *sql.Tx,
reportID uint64,
) (api.QueryAdminEventReportResponse, error) {
stmt := sqlutil.TxStmt(txn, r.selectReportedEventStmt)
var row api.QueryAdminEventReportResponse
if err := stmt.QueryRowContext(ctx, reportID).Scan(
&row.ID,
&row.RoomNID,
&row.EventNID,
&row.ReportingUserNID,
&row.SenderNID,
&row.Reason,
&row.Score,
&row.ReceivedTS,
); err != nil {
return api.QueryAdminEventReportResponse{}, err
}
return row, nil
}
func (r *reportedEventsStatements) DeleteReportedEvent(ctx context.Context, txn *sql.Tx, reportID uint64) error {
stmt := sqlutil.TxStmt(txn, r.deleteReportedEventStmt)
_, err := stmt.ExecContext(ctx, reportID)
return err
}

23
roomserver/storage/sqlite3/rooms_table.go
View file

@ -65,9 +65,6 @@ const selectRoomVersionsForRoomNIDsSQL = "" +
const selectRoomInfoSQL = "" +
"SELECT room_version, room_nid, state_snapshot_nid, latest_event_nids FROM roomserver_rooms WHERE room_id = $1"
const selectRoomIDsSQL = "" +
"SELECT room_id FROM roomserver_rooms WHERE latest_event_nids != '[]'"
const bulkSelectRoomIDsSQL = "" +
"SELECT room_id FROM roomserver_rooms WHERE room_nid IN ($1)"
@ -87,7 +84,6 @@ type roomStatements struct {
updateLatestEventNIDsStmt *sql.Stmt
//selectRoomVersionForRoomNIDStmt *sql.Stmt
selectRoomInfoStmt *sql.Stmt
selectRoomIDsStmt *sql.Stmt
}
func CreateRoomsTable(db *sql.DB) error {
@ -108,29 +104,10 @@ func PrepareRoomsTable(db *sql.DB) (tables.Rooms, error) {
{&s.updateLatestEventNIDsStmt, updateLatestEventNIDsSQL},
//{&s.selectRoomVersionForRoomNIDsStmt, selectRoomVersionForRoomNIDsSQL},
{&s.selectRoomInfoStmt, selectRoomInfoSQL},
{&s.selectRoomIDsStmt, selectRoomIDsSQL},
{&s.selectRoomNIDForUpdateStmt, selectRoomNIDForUpdateSQL},
}.Prepare(db)
}
func (s *roomStatements) SelectRoomIDsWithEvents(ctx context.Context, txn *sql.Tx) ([]string, error) {
stmt := sqlutil.TxStmt(txn, s.selectRoomIDsStmt)
rows, err := stmt.QueryContext(ctx)
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsStmt: rows.close() failed")
var roomIDs []string
var roomID string
for rows.Next() {
if err = rows.Scan(&roomID); err != nil {
return nil, err
}
roomIDs = append(roomIDs, roomID)
}
return roomIDs, rows.Err()
}
func (s *roomStatements) SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID string) (*types.RoomInfo, error) {
var info types.RoomInfo
var latestNIDsJSON string

9
roomserver/storage/sqlite3/storage.go
View file

@ -141,7 +141,9 @@ func (d *Database) create(db *sql.DB) error {
if err := CreateUserRoomKeysTable(db); err != nil {
return err
}
if err := CreateReportedEventsTable(db); err != nil {
return err
}
return nil
}
@ -206,6 +208,10 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
if err != nil {
return err
}
reportedEvents, err := PrepareReportedEventsTable(db)
if err != nil {
return err
}
d.Database = shared.Database{
DB: db,
@ -219,6 +225,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
EventJSONTable: eventJSON,
PrevEventsTable: prevEvents,
RedactionsTable: redactions,
ReportedEventsTable: reportedEvents,
},
Cache: cache,
Writer: writer,

36
roomserver/storage/tables/events_table_test.go
View file

@ -2,6 +2,7 @@ package tables_test
import (
"context"
"fmt"
"testing"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -147,3 +148,38 @@ func Test_EventsTable(t *testing.T) {
assert.Equal(t, int64(len(room.Events())+1), maxDepth)
})
}
func TestRoomsWithACL(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
eventStateKeys, closeEventStateKeys := mustCreateEventTypesTable(t, dbType)
defer closeEventStateKeys()
eventsTable, closeEventsTable := mustCreateEventsTable(t, dbType)
defer closeEventsTable()
ctx := context.Background()
// insert the m.room.server_acl event type
eventTypeNID, err := eventStateKeys.InsertEventTypeNID(ctx, nil, "m.room.server_acl")
assert.Nil(t, err)
// Create ACL'd rooms
var wantRoomNIDs []types.RoomNID
for i := 0; i < 10; i++ {
_, _, err = eventsTable.InsertEvent(ctx, nil, types.RoomNID(i), eventTypeNID, types.EmptyStateKeyNID, fmt.Sprintf("$1337+%d", i), nil, 0, false)
assert.Nil(t, err)
wantRoomNIDs = append(wantRoomNIDs, types.RoomNID(i))
}
// Create non-ACL'd rooms (eventTypeNID+1)
for i := 10; i < 20; i++ {
_, _, err = eventsTable.InsertEvent(ctx, nil, types.RoomNID(i), eventTypeNID+1, types.EmptyStateKeyNID, fmt.Sprintf("$1337+%d", i), nil, 0, false)
assert.Nil(t, err)
}
gotRoomNIDs, err := eventsTable.SelectRoomsWithEventTypeNID(ctx, nil, eventTypeNID)
assert.Nil(t, err)
assert.Equal(t, wantRoomNIDs, gotRoomNIDs)
})
}

31
roomserver/storage/tables/interface.go
View file

@ -6,6 +6,7 @@ import (
"database/sql"
"errors"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/tidwall/gjson"
@ -69,6 +70,8 @@ type Events interface {
SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error)
SelectRoomNIDsForEventNIDs(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (roomNIDs map[types.EventNID]types.RoomNID, err error)
SelectEventRejected(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventID string) (rejected bool, err error)
SelectRoomsWithEventTypeNID(ctx context.Context, txn *sql.Tx, eventTypeNID types.EventTypeNID) ([]types.RoomNID, error)
}
type Rooms interface {
@ -80,7 +83,6 @@ type Rooms interface {
UpdateLatestEventNIDs(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventNIDs []types.EventNID, lastEventSentNID types.EventNID, stateSnapshotNID types.StateSnapshotNID) error
SelectRoomVersionsForRoomNIDs(ctx context.Context, txn *sql.Tx, roomNID []types.RoomNID) (map[types.RoomNID]gomatrixserverlib.RoomVersion, error)
SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID string) (*types.RoomInfo, error)
SelectRoomIDsWithEvents(ctx context.Context, txn *sql.Tx) ([]string, error)
BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roomNIDs []types.RoomNID) ([]string, error)
BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, roomIDs []string) ([]types.RoomNID, error)
}
@ -126,6 +128,33 @@ type Invites interface {
SelectInviteActiveForUserInRoom(ctx context.Context, txn *sql.Tx, targetUserNID types.EventStateKeyNID, roomNID types.RoomNID) ([]types.EventStateKeyNID, []string, []byte, error)
}
type ReportedEvents interface {
InsertReportedEvent(
ctx context.Context,
txn *sql.Tx,
roomNID types.RoomNID,
eventNID types.EventNID,
reportingUserID types.EventStateKeyNID,
eventSenderID types.EventStateKeyNID,
reason string,
score int64,
) (int64, error)
SelectReportedEvents(
ctx context.Context,
txn *sql.Tx,
from, limit uint64,
backwards bool,
reportingUserID types.EventStateKeyNID,
roomNID types.RoomNID,
) ([]api.QueryAdminEventReportsResponse, int64, error)
SelectReportedEvent(
ctx context.Context,
txn *sql.Tx,
reportID uint64,
) (api.QueryAdminEventReportResponse, error)
DeleteReportedEvent(ctx context.Context, txn *sql.Tx, reportID uint64) error
}
type MembershipState int64
const (

7
roomserver/storage/tables/rooms_table_test.go
View file

@ -74,11 +74,6 @@ func TestRoomsTable(t *testing.T) {
assert.NoError(t, err)
assert.Nil(t, roomInfo)
// There are no rooms with latestEventNIDs yet
roomIDs, err := tab.SelectRoomIDsWithEvents(ctx, nil)
assert.NoError(t, err)
assert.Equal(t, 0, len(roomIDs))
roomVersions, err := tab.SelectRoomVersionsForRoomNIDs(ctx, nil, []types.RoomNID{wantRoomNID, 1337})
assert.NoError(t, err)
assert.Equal(t, roomVersions[wantRoomNID], room.Version)
@ -86,7 +81,7 @@ func TestRoomsTable(t *testing.T) {
_, ok := roomVersions[1337]
assert.False(t, ok)
roomIDs, err = tab.BulkSelectRoomIDs(ctx, nil, []types.RoomNID{wantRoomNID, 1337})
roomIDs, err := tab.BulkSelectRoomIDs(ctx, nil, []types.RoomNID{wantRoomNID, 1337})
assert.NoError(t, err)
assert.Equal(t, []string{room.ID}, roomIDs)

7
setup/jetstream/nats.go
View file

@ -38,7 +38,12 @@ func (s *NATSInstance) Prepare(process *process.ProcessContext, cfg *config.JetS
defer natsLock.Unlock()
// check if we need an in-process NATS Server
if len(cfg.Addresses) != 0 {
return setupNATS(process, cfg, nil)
// reuse existing connections
if s.nc != nil {
return s.js, s.nc
}
s.js, s.nc = setupNATS(process, cfg, nil)
return s.js, s.nc
}
if s.Server == nil {
var err error