mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2024-11-26 00:01:55 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into s7evink/memberships

Browse source
This commit is contained in:
Devon Hudson 2023-07-21 11:37:50 -06:00
parent 16075ce657 c809e95335
commit 0aa1b6c218
No known key found for this signature in database
GPG key ID: CD06B18E77F6A628
72 changed files with 1299 additions and 940 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
appservice/appservice_test.go
View file

@ -134,7 +134,6 @@ func TestAppserviceInternalAPI(t *testing.T) {
}
as.CreateHTTPClient(cfg.AppServiceAPI.DisableTLSValidation)
cfg.AppServiceAPI.Derived.ApplicationServices = []config.ApplicationService{*as}
t.Cleanup(func() {
ctx.ShutdownDendrite()
ctx.WaitForShutdown()
@ -144,6 +143,7 @@ func TestAppserviceInternalAPI(t *testing.T) {
natsInstance := jetstream.NATSInstance{}
cm := sqlutil.NewConnectionManager(ctx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(ctx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
usrAPI := userapi.NewInternalAPI(ctx, cfg, cm, &natsInstance, rsAPI, nil)
asAPI := appservice.NewInternalAPI(ctx, cfg, &natsInstance, usrAPI, rsAPI)
@ -238,6 +238,7 @@ func TestAppserviceInternalAPI_UnixSocket_Simple(t *testing.T) {
natsInstance := jetstream.NATSInstance{}
cm := sqlutil.NewConnectionManager(ctx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(ctx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
usrAPI := userapi.NewInternalAPI(ctx, cfg, cm, &natsInstance, rsAPI, nil)
asAPI := appservice.NewInternalAPI(ctx, cfg, &natsInstance, usrAPI, rsAPI)

18
clientapi/admin_test.go
View file

@ -44,6 +44,7 @@ func TestAdminCreateToken(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
@ -194,6 +195,7 @@ func TestAdminListRegistrationTokens(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
@ -311,6 +313,7 @@ func TestAdminGetRegistrationToken(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
@ -411,6 +414,7 @@ func TestAdminDeleteRegistrationToken(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
@ -504,6 +508,7 @@ func TestAdminUpdateRegistrationToken(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
accessTokens := map[*test.User]userDevice{
@ -686,6 +691,7 @@ func TestAdminResetPassword(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
// Needed for changing the password/login
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the userAPI for this test, so nil for other APIs/caches etc.
@ -780,13 +786,14 @@ func TestPurgeRoom(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
fsAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(fsAPI, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
@ -851,12 +858,13 @@ func TestAdminEvacuateRoom(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
fsAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(fsAPI, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", api.DoNotSendToOtherServers, nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
@ -951,12 +959,13 @@ func TestAdminEvacuateUser(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
fsAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, basepkg.CreateFederationClient(cfg, nil), rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(fsAPI, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", api.DoNotSendToOtherServers, nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
@ -1045,6 +1054,7 @@ func TestAdminMarkAsStale(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.

9
clientapi/clientapi_test.go
View file

@ -120,6 +120,7 @@ func TestGetPutDevices(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
@ -168,6 +169,7 @@ func TestDeleteDevice(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI/ for this test, so nil for other APIs/caches etc.
@ -272,6 +274,7 @@ func TestDeleteDevices(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI/ for this test, so nil for other APIs/caches etc.
@ -947,6 +950,7 @@ func TestCapabilities(t *testing.T) {
// Needed to create accounts
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI/userAPI for this test, so nil for other APIs etc.
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
@ -993,6 +997,7 @@ func TestTurnserver(t *testing.T) {
// Needed to create accounts
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
//rsAPI.SetUserAPI(userAPI)
// We mostly need the rsAPI/userAPI for this test, so nil for other APIs etc.
@ -1090,6 +1095,7 @@ func Test3PID(t *testing.T) {
// Needed to create accounts
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI/userAPI for this test, so nil for other APIs etc.
AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
@ -1265,6 +1271,7 @@ func TestPushRules(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
@ -1651,6 +1658,7 @@ func TestKeys(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
@ -2112,6 +2120,7 @@ func TestKeyBackup(t *testing.T) {
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.

2
clientapi/routing/joinroom_test.go
View file

@ -35,9 +35,9 @@ func TestJoinRoomByIDOrAlias(t *testing.T) {
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil) // creates the rs.Inputer etc
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetFederationAPI(nil, nil) // creates the rs.Inputer etc
// Create the users in the userapi
for _, u := range []*test.User{alice, bob, charlie} {

1
clientapi/routing/login_test.go
View file

@ -47,6 +47,7 @@ func TestLogin(t *testing.T) {
routers := httputil.NewRouters()
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
// Needed for /login
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)

3
clientapi/routing/register_test.go
View file

@ -415,6 +415,7 @@ func Test_register(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
for _, tc := range testCases {
@ -594,6 +595,7 @@ func TestRegisterUserWithDisplayName(t *testing.T) {
natsInstance := jetstream.NATSInstance{}
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
deviceName, deviceID := "deviceName", "deviceID"
expectedDisplayName := "DisplayName"
@ -634,6 +636,7 @@ func TestRegisterAdminUsingSharedSecret(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
expectedDisplayName := "rabbit"

180
clientapi/routing/room_hierarchy.go Normal file
View file

@ -0,0 +1,180 @@
// 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"
"strconv"
"sync"
"github.com/google/uuid"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/types"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib/fclient"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
log "github.com/sirupsen/logrus"
)
// For storing pagination information for room hierarchies
type RoomHierarchyPaginationCache struct {
cache map[string]roomserverAPI.RoomHierarchyWalker
mu sync.Mutex
}
// Create a new, empty, pagination cache.
func NewRoomHierarchyPaginationCache() RoomHierarchyPaginationCache {
return RoomHierarchyPaginationCache{
cache: map[string]roomserverAPI.RoomHierarchyWalker{},
}
}
// Get a cached page, or nil if there is no associated page in the cache.
func (c *RoomHierarchyPaginationCache) Get(token string) *roomserverAPI.RoomHierarchyWalker {
c.mu.Lock()
defer c.mu.Unlock()
line, ok := c.cache[token]
if ok {
return &line
} else {
return nil
}
}
// Add a cache line to the pagination cache.
func (c *RoomHierarchyPaginationCache) AddLine(line roomserverAPI.RoomHierarchyWalker) string {
c.mu.Lock()
defer c.mu.Unlock()
token := uuid.NewString()
c.cache[token] = line
return token
}
// Query the hierarchy of a room/space
//
// Implements /_matrix/client/v1/rooms/{roomID}/hierarchy
func QueryRoomHierarchy(req *http.Request, device *userapi.Device, roomIDStr string, rsAPI roomserverAPI.ClientRoomserverAPI, paginationCache *RoomHierarchyPaginationCache) util.JSONResponse {
parsedRoomID, err := spec.NewRoomID(roomIDStr)
if err != nil {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: spec.InvalidParam("room is unknown/forbidden"),
}
}
roomID := *parsedRoomID
suggestedOnly := false // Defaults to false (spec-defined)
switch req.URL.Query().Get("suggested_only") {
case "true":
suggestedOnly = true
case "false":
case "": // Empty string is returned when query param is not set
default:
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: spec.InvalidParam("query parameter 'suggested_only', if set, must be 'true' or 'false'"),
}
}
limit := 1000 // Default to 1000
limitStr := req.URL.Query().Get("limit")
if limitStr != "" {
var maybeLimit int
maybeLimit, err = strconv.Atoi(limitStr)
if err != nil || maybeLimit < 0 {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: spec.InvalidParam("query parameter 'limit', if set, must be a positive integer"),
}
}
limit = maybeLimit
if limit > 1000 {
limit = 1000 // Maximum limit of 1000
}
}
maxDepth := -1 // '-1' representing no maximum depth
maxDepthStr := req.URL.Query().Get("max_depth")
if maxDepthStr != "" {
var maybeMaxDepth int
maybeMaxDepth, err = strconv.Atoi(maxDepthStr)
if err != nil || maybeMaxDepth < 0 {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: spec.InvalidParam("query parameter 'max_depth', if set, must be a positive integer"),
}
}
maxDepth = maybeMaxDepth
}
from := req.URL.Query().Get("from")
var walker roomserverAPI.RoomHierarchyWalker
if from == "" { // No pagination token provided, so start new hierarchy walker
walker = roomserverAPI.NewRoomHierarchyWalker(types.NewDeviceNotServerName(*device), roomID, suggestedOnly, maxDepth)
} else { // Attempt to resume cached walker
cachedWalker := paginationCache.Get(from)
if cachedWalker == nil || cachedWalker.SuggestedOnly != suggestedOnly || cachedWalker.MaxDepth != maxDepth {
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: spec.InvalidParam("pagination not found for provided token ('from') with given 'max_depth', 'suggested_only' and room ID"),
}
}
walker = *cachedWalker
}
discoveredRooms, nextWalker, err := rsAPI.QueryNextRoomHierarchyPage(req.Context(), walker, limit)
if err != nil {
switch err.(type) {
case roomserverAPI.ErrRoomUnknownOrNotAllowed:
util.GetLogger(req.Context()).WithError(err).Debugln("room unknown/forbidden when handling CS room hierarchy request")
return util.JSONResponse{
Code: http.StatusForbidden,
JSON: spec.Forbidden("room is unknown/forbidden"),
}
default:
log.WithError(err).Errorf("failed to fetch next page of room hierarchy (CS API)")
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.Unknown("internal server error"),
}
}
}
nextBatch := ""
// nextWalker will be nil if there's no more rooms left to walk
if nextWalker != nil {
nextBatch = paginationCache.AddLine(*nextWalker)
}
return util.JSONResponse{
Code: http.StatusOK,
JSON: RoomHierarchyClientResponse{
Rooms: discoveredRooms,
NextBatch: nextBatch,
},
}
}
// Success response for /_matrix/client/v1/rooms/{roomID}/hierarchy
type RoomHierarchyClientResponse struct {
Rooms []fclient.RoomHierarchyRoom `json:"rooms"`
NextBatch string `json:"next_batch,omitempty"`
}

15
clientapi/routing/routing.go
View file

@ -288,6 +288,8 @@ func Setup(
// Note that 'apiversion' is chosen because it must not collide with a variable used in any of the routing!
v3mux := publicAPIMux.PathPrefix("/{apiversion:(?:r0|v3)}/").Subrouter()
v1mux := publicAPIMux.PathPrefix("/v1/").Subrouter()
unstableMux := publicAPIMux.PathPrefix("/unstable").Subrouter()
v3mux.Handle("/createRoom",
@ -505,6 +507,19 @@ func Setup(
}, httputil.WithAllowGuests()),
).Methods(http.MethodPut, http.MethodOptions)
// Defined outside of handler to persist between calls
// TODO: clear based on some criteria
roomHierarchyPaginationCache := NewRoomHierarchyPaginationCache()
v1mux.Handle("/rooms/{roomID}/hierarchy",
httputil.MakeAuthAPI("spaces", 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 QueryRoomHierarchy(req, device, vars["roomID"], rsAPI, &roomHierarchyPaginationCache)
}, httputil.WithAllowGuests()),
).Methods(http.MethodGet, http.MethodOptions)
v3mux.Handle("/register", httputil.MakeExternalAPI("register", func(req *http.Request) util.JSONResponse {
if r := rateLimits.Limit(req, nil); r != nil {
return *r

7
cmd/dendrite-demo-pinecone/monolith/monolith.go
View file

@ -98,7 +98,7 @@ func GenerateDefaultConfig(sk ed25519.PrivateKey, storageDir string, cacheDir st
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", filepath.Join(storageDir, dbPrefix)))
cfg.FederationAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", filepath.Join(storageDir, dbPrefix)))
cfg.RelayAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-relayapi.db", filepath.Join(storageDir, dbPrefix)))
cfg.MSCs.MSCs = []string{"msc2836", "msc2946"}
cfg.MSCs.MSCs = []string{"msc2836"}
cfg.MSCs.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mscs.db", filepath.Join(storageDir, dbPrefix)))
cfg.ClientAPI.RegistrationDisabled = false
cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled = true
@ -126,7 +126,7 @@ func (p *P2PMonolith) SetupPinecone(sk ed25519.PrivateKey) {
}
func (p *P2PMonolith) SetupDendrite(
processCtx *process.ProcessContext, cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers,
processCtx *process.ProcessContext, cfg *config.Dendrite, cm *sqlutil.Connections, routers httputil.Routers,
port int, enableRelaying bool, enableMetrics bool, enableWebsockets bool) {
p.port = port
@ -143,13 +143,12 @@ func (p *P2PMonolith) SetupDendrite(
fsAPI := federationapi.NewInternalAPI(
processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
rsAPI.SetFederationAPI(fsAPI, keyRing)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetFederationAPI(fsAPI, keyRing)
userProvider := users.NewPineconeUserProvider(p.Router, p.Sessions, userAPI, federation)
roomProvider := rooms.NewPineconeRoomProvider(p.Router, p.Sessions, fsAPI, federation)

2
cmd/dendrite-demo-yggdrasil/main.go
View file

@ -134,7 +134,7 @@ func main() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", filepath.Join(*instanceDir, *instanceName)))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", filepath.Join(*instanceDir, *instanceName)))
cfg.FederationAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationapi.db", filepath.Join(*instanceDir, *instanceName)))
cfg.MSCs.MSCs = []string{"msc2836", "msc2946"}
cfg.MSCs.MSCs = []string{"msc2836"}
cfg.MSCs.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mscs.db", filepath.Join(*instanceDir, *instanceName)))
cfg.ClientAPI.RegistrationDisabled = false
cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled = true

7
cmd/dendrite/main.go
View file

@ -157,13 +157,14 @@ func main() {
keyRing := fsAPI.KeyRing()
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federationClient)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
// The underlying roomserver implementation needs to be able to call the fedsender.
// This is different to rsAPI which can be the http client which doesn't need this
// dependency. Other components also need updating after their dependencies are up.
rsAPI.SetFederationAPI(fsAPI, keyRing)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federationClient)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
rsAPI.SetUserAPI(userAPI)

2
cmd/generate-config/main.go
View file

@ -74,7 +74,7 @@ func main() {
// don't hit matrix.org when running tests!!!
cfg.FederationAPI.KeyPerspectives = config.KeyPerspectives{}
cfg.MediaAPI.BasePath = config.Path(filepath.Join(*dirPath, "media"))
cfg.MSCs.MSCs = []string{"msc2836", "msc2946", "msc2444", "msc2753"}
cfg.MSCs.MSCs = []string{"msc2836", "msc2444", "msc2753"}
cfg.Logging[0].Level = "trace"
cfg.Logging[0].Type = "std"
cfg.UserAPI.BCryptCost = bcrypt.MinCost

1
dendrite-sample.yaml
View file

@ -276,7 +276,6 @@ media_api:
mscs:
mscs:
# - msc2836 # (Threading, see https://github.com/matrix-org/matrix-doc/pull/2836)
# - msc2946 # (Spaces Summary, see https://github.com/matrix-org/matrix-doc/pull/2946)
# Configuration for the Sync API.
sync_api:

10
docs/FAQ.md
View file

@ -64,16 +64,18 @@ Use [dendrite.matrix.org](https://dendrite.matrix.org) which we officially suppo
## Does Dendrite support Space Summaries?
Yes, [Space Summaries](https://github.com/matrix-org/matrix-spec-proposals/pull/2946) were merged into the Matrix Spec as of 2022-01-17 however, they are still treated as an MSC (Matrix Specification Change) in Dendrite. In order to enable Space Summaries in Dendrite, you must add the MSC to the MSC configuration section in the configuration YAML. If the MSC is not enabled, a user will typically see a perpetual loading icon on the summary page. See below for a demonstration of how to add to the Dendrite configuration:
Yes
## Does Dendrite support Threads?
Yes, to enable them [msc2836](https://github.com/matrix-org/matrix-spec-proposals/pull/2836) would need to be added to mscs configuration in order to support Threading. Other MSCs are not currently supported.
```
mscs:
mscs:
- msc2946
- msc2836
```
Similarly, [msc2836](https://github.com/matrix-org/matrix-spec-proposals/pull/2836) would need to be added to mscs configuration in order to support Threading. Other MSCs are not currently supported.
Please note that MSCs should be considered experimental and can result in significant usability issues when enabled. If you'd like more details on how MSCs are ratified or the current status of MSCs, please see the [Matrix specification documentation](https://spec.matrix.org/proposals/) on the subject.
## Does Dendrite support push notifications?

3
federationapi/api/api.go
View file

@ -27,7 +27,6 @@ type FederationInternalAPI interface {
QueryServerKeys(ctx context.Context, request *QueryServerKeysRequest, response *QueryServerKeysResponse) error
LookupServerKeys(ctx context.Context, s spec.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]spec.Timestamp) ([]gomatrixserverlib.ServerKeys, error)
MSC2836EventRelationships(ctx context.Context, origin, dst spec.ServerName, r fclient.MSC2836EventRelationshipsRequest, roomVersion gomatrixserverlib.RoomVersion) (res fclient.MSC2836EventRelationshipsResponse, err error)
MSC2946Spaces(ctx context.Context, origin, dst spec.ServerName, roomID string, suggestedOnly bool) (res fclient.MSC2946SpacesResponse, err error)
// Broadcasts an EDU to all servers in rooms we are joined to. Used in the yggdrasil demos.
PerformBroadcastEDU(
@ -76,6 +75,8 @@ type RoomserverFederationAPI interface {
GetEvent(ctx context.Context, origin, s spec.ServerName, eventID string) (res gomatrixserverlib.Transaction, err error)
LookupMissingEvents(ctx context.Context, origin, s spec.ServerName, roomID string, missing fclient.MissingEvents, roomVersion gomatrixserverlib.RoomVersion) (res fclient.RespMissingEvents, err error)
CurrentMembership(ctx context.Context, roomID spec.RoomID, senderID spec.SenderID) (string, error)
RoomHierarchies(ctx context.Context, origin, dst spec.ServerName, roomID string, suggestedOnly bool) (res fclient.RoomHierarchyResponse, err error)
}
type P2PFederationAPI interface {

17
federationapi/consumers/roomserver.go
View file

@ -16,7 +16,9 @@ package consumers
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"strconv"
"time"
@ -411,13 +413,26 @@ func JoinedHostsFromEvents(ctx context.Context, evs []gomatrixserverlib.PDU, rsA
if err != nil {
return nil, err
}
var domain spec.ServerName
userID, err := rsAPI.QueryUserIDForSender(ctx, *validRoomID, spec.SenderID(*ev.StateKey()))
if err != nil {
if errors.As(err, new(base64.CorruptInputError)) {
// Fallback to using the "old" way of getting the user domain, avoids
// "illegal base64 data at input byte 0" errors
// FIXME: we should do this in QueryUserIDForSender instead
_, domain, err = gomatrixserverlib.SplitID('@', *ev.StateKey())
if err != nil {
return nil, err
}
} else {
return nil, err
}
} else {
domain = userID.Domain()
}
joinedHosts = append(joinedHosts, types.JoinedHost{
MemberEventID: ev.EventID(), ServerName: userID.Domain(),
MemberEventID: ev.EventID(), ServerName: domain,
})
}
return joinedHosts, nil

2
federationapi/federationapi.go
View file

@ -95,7 +95,7 @@ func AddPublicRoutes(
func NewInternalAPI(
processContext *process.ProcessContext,
dendriteCfg *config.Dendrite,
cm sqlutil.Connections,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
federation fclient.FederationClient,
rsAPI roomserverAPI.FederationRoomserverAPI,

10
federationapi/internal/federationclient.go
View file

@ -29,7 +29,7 @@ func (a *FederationInternalAPI) MakeJoin(
func (a *FederationInternalAPI) SendJoin(
ctx context.Context, origin, s spec.ServerName, event gomatrixserverlib.PDU,
) (res gomatrixserverlib.SendJoinResponse, err error) {
ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
defer cancel()
ires, err := a.federation.SendJoin(ctx, origin, s, event)
if err != nil {
@ -194,16 +194,16 @@ func (a *FederationInternalAPI) MSC2836EventRelationships(
return ires.(fclient.MSC2836EventRelationshipsResponse), nil
}
func (a *FederationInternalAPI) MSC2946Spaces(
func (a *FederationInternalAPI) RoomHierarchies(
ctx context.Context, origin, s spec.ServerName, roomID string, suggestedOnly bool,
) (res fclient.MSC2946SpacesResponse, err error) {
) (res fclient.RoomHierarchyResponse, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Minute)
defer cancel()
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
return a.federation.MSC2946Spaces(ctx, origin, s, roomID, suggestedOnly)
return a.federation.RoomHierarchy(ctx, origin, s, roomID, suggestedOnly)
})
if err != nil {
return res, err
}
return ires.(fclient.MSC2946SpacesResponse), nil
return ires.(fclient.RoomHierarchyResponse), nil
}

8
federationapi/routing/profile.go
View file

@ -15,9 +15,11 @@
package routing
import (
"errors"
"fmt"
"net/http"
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
@ -52,6 +54,12 @@ func GetProfile(
profile, err := userAPI.QueryProfile(httpReq.Context(), userID)
if err != nil {
if errors.Is(err, appserviceAPI.ErrProfileNotExists) {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: spec.NotFound("The user does not exist or does not have a profile."),
}
}
util.GetLogger(httpReq.Context()).WithError(err).Error("userAPI.QueryProfile failed")
return util.JSONResponse{
Code: http.StatusInternalServerError,

64
federationapi/routing/query.go
View file

@ -20,12 +20,14 @@ import (
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/fclient"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
log "github.com/sirupsen/logrus"
)
// RoomAliasToID converts the queried alias into a room ID and returns it
@ -116,3 +118,65 @@ func RoomAliasToID(
JSON: resp,
}
}
// Query the immediate children of a room/space
//
// Implements /_matrix/federation/v1/hierarchy/{roomID}
func QueryRoomHierarchy(httpReq *http.Request, request *fclient.FederationRequest, roomIDStr string, rsAPI roomserverAPI.FederationRoomserverAPI) util.JSONResponse {
parsedRoomID, err := spec.NewRoomID(roomIDStr)
if err != nil {
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: spec.InvalidParam("room is unknown/forbidden"),
}
}
roomID := *parsedRoomID
suggestedOnly := false // Defaults to false (spec-defined)
switch httpReq.URL.Query().Get("suggested_only") {
case "true":
suggestedOnly = true
case "false":
case "": // Empty string is returned when query param is not set
default:
return util.JSONResponse{
Code: http.StatusBadRequest,
JSON: spec.InvalidParam("query parameter 'suggested_only', if set, must be 'true' or 'false'"),
}
}
walker := roomserverAPI.NewRoomHierarchyWalker(types.NewServerNameNotDevice(request.Origin()), roomID, suggestedOnly, 1)
discoveredRooms, _, err := rsAPI.QueryNextRoomHierarchyPage(httpReq.Context(), walker, -1)
if err != nil {
switch err.(type) {
case roomserverAPI.ErrRoomUnknownOrNotAllowed:
util.GetLogger(httpReq.Context()).WithError(err).Debugln("room unknown/forbidden when handling SS room hierarchy request")
return util.JSONResponse{
Code: http.StatusNotFound,
JSON: spec.NotFound("room is unknown/forbidden"),
}
default:
log.WithError(err).Errorf("failed to fetch next page of room hierarchy (SS API)")
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.Unknown("internal server error"),
}
}
}
if len(discoveredRooms) == 0 {
util.GetLogger(httpReq.Context()).Debugln("no rooms found when handling SS room hierarchy request")
return util.JSONResponse{
Code: 404,
JSON: spec.NotFound("room is unknown/forbidden"),
}
}
return util.JSONResponse{
Code: 200,
JSON: fclient.RoomHierarchyResponse{
Room: discoveredRooms[0],
Children: discoveredRooms[1:],
},
}
}

7
federationapi/routing/routing.go
View file

@ -596,6 +596,13 @@ func Setup(
return GetOpenIDUserInfo(req, userAPI)
}),
).Methods(http.MethodGet)
v1fedmux.Handle("/hierarchy/{roomID}", MakeFedAPI(
"federation_room_hierarchy", cfg.Matrix.ServerName, cfg.Matrix.IsLocalServerName, keys, wakeup,
func(httpReq *http.Request, request *fclient.FederationRequest, vars map[string]string) util.JSONResponse {
return QueryRoomHierarchy(httpReq, request, vars["roomID"], rsAPI)
},
)).Methods(http.MethodGet)
}
func ErrorIfLocalServerNotInRoom(

2
federationapi/storage/postgres/storage.go
View file

@ -36,7 +36,7 @@ type Database struct {
}
// NewDatabase opens a new database
func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (*Database, error) {
func NewDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (*Database, error) {
var d Database
var err error
if d.db, d.writer, err = conMan.Connection(dbProperties); err != nil {

2
federationapi/storage/sqlite3/storage.go
View file

@ -34,7 +34,7 @@ type Database struct {
}
// NewDatabase opens a new database
func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (*Database, error) {
func NewDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (*Database, error) {
var d Database
var err error
if d.db, d.writer, err = conMan.Connection(dbProperties); err != nil {

2
federationapi/storage/storage.go
View file

@ -30,7 +30,7 @@ import (
)
// NewDatabase opens a new database
func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (Database, error) {
func NewDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.FederationCache, isLocalServerName func(spec.ServerName) bool) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewDatabase(ctx, conMan, dbProperties, cache, isLocalServerName)

4
go.mod
View file

@ -1,5 +1,7 @@
module github.com/matrix-org/dendrite
replace github.com/matrix-org/gomatrixserverlib => ../../../gomatrixserverlib/s7evink/memberships/
require (
github.com/Arceliar/ironwood v0.0.0-20221025225125-45b4281814c2
github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979
@ -22,7 +24,7 @@ require (
github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530
github.com/matrix-org/gomatrixserverlib v0.0.0-20230712091231-3a43f3ff9987
github.com/matrix-org/gomatrixserverlib v0.0.0-20230721154317-b5b0448aa378
github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a
github.com/matrix-org/util v0.0.0-20221111132719-399730281e66
github.com/mattn/go-sqlite3 v1.14.17

2
go.sum
View file

@ -207,8 +207,6 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91 h1:s7fexw
github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo=
github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530 h1:kHKxCOLcHH8r4Fzarl4+Y3K5hjothkVW5z7T1dUM11U=
github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
github.com/matrix-org/gomatrixserverlib v0.0.0-20230712091231-3a43f3ff9987 h1:X2dfvmGHXcZdbrt1Qcaocl7nfPEm32+9Es7/g518Dfg=
github.com/matrix-org/gomatrixserverlib v0.0.0-20230712091231-3a43f3ff9987/go.mod h1:H9V9N3Uqn1bBJqYJNGK1noqtgJTaCEhtTdcH/mp50uU=
github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a h1:awrPDf9LEFySxTLKYBMCiObelNx/cBuv/wzllvCCH3A=
github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a/go.mod h1:HchJX9oKMXaT2xYFs0Ha/6Zs06mxLU8k6F1ODnrGkeQ=
github.com/matrix-org/util v0.0.0-20221111132719-399730281e66 h1:6z4KxomXSIGWqhHcfzExgkH3Z3UkIXry4ibJS4Aqz2Y=

4
helm/dendrite/Chart.yaml
View file

@ -1,7 +1,7 @@
apiVersion: v2
name: dendrite
version: "0.13.0"
appVersion: "0.13.0"
version: "0.13.1"
appVersion: "0.13.1"
description: Dendrite Matrix Homeserver
type: application
keywords:

20
helm/dendrite/README.md
View file

@ -1,7 +1,7 @@
# dendrite
![Version: 0.13.0](https://img.shields.io/badge/Version-0.13.0-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.13.0](https://img.shields.io/badge/AppVersion-0.13.0-informational?style=flat-square)
![Version: 0.13.1](https://img.shields.io/badge/Version-0.13.1-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.13.1](https://img.shields.io/badge/AppVersion-0.13.1-informational?style=flat-square)
Dendrite Matrix Homeserver
Status: **NOT PRODUCTION READY**
@ -48,13 +48,16 @@ Create a folder `appservices` and place your configurations in there. The confi
| signing_key.create | bool | `true` | Create a new signing key, if not exists |
| signing_key.existingSecret | string | `""` | Use an existing secret |
| resources | object | sets some sane default values | Default resource requests/limits. |
| persistence.storageClass | string | `""` | The storage class to use for volume claims. Defaults to the cluster default storage class. |
| persistence.storageClass | string | `""` | The storage class to use for volume claims. Used unless specified at the specific component. Defaults to the cluster default storage class. |
| persistence.jetstream.existingClaim | string | `""` | Use an existing volume claim for jetstream |
| persistence.jetstream.capacity | string | `"1Gi"` | PVC Storage Request for the jetstream volume |
| persistence.jetstream.storageClass | string | `""` | The storage class to use for volume claims. Defaults to persistence.storageClass |
| persistence.media.existingClaim | string | `""` | Use an existing volume claim for media files |
| persistence.media.capacity | string | `"1Gi"` | PVC Storage Request for the media volume |
| persistence.media.storageClass | string | `""` | The storage class to use for volume claims. Defaults to persistence.storageClass |
| persistence.search.existingClaim | string | `""` | Use an existing volume claim for the fulltext search index |
| persistence.search.capacity | string | `"1Gi"` | PVC Storage Request for the search volume |
| persistence.search.storageClass | string | `""` | The storage class to use for volume claims. Defaults to persistence.storageClass |
| extraVolumes | list | `[]` | Add additional volumes to the Dendrite Pod |
| extraVolumeMounts | list | `[]` | Configure additional mount points volumes in the Dendrite Pod |
| strategy.type | string | `"RollingUpdate"` | Strategy to use for rolling updates (e.g. Recreate, RollingUpdate) If you are using ReadWriteOnce volumes, you should probably use Recreate |
@ -97,7 +100,7 @@ Create a folder `appservices` and place your configurations in there. The confi
| dendrite_config.global.dns_cache.cache_lifetime | string | `"10m"` | Duration for how long DNS cache items should be considered valid ([see time.ParseDuration](https://pkg.go.dev/time#ParseDuration) for more) |
| dendrite_config.global.profiling.enabled | bool | `false` | Enable pprof. You will need to manually create a port forwarding to the deployment to access PPROF, as it will only listen on localhost and the defined port. e.g. `kubectl port-forward deployments/dendrite 65432:65432` |
| dendrite_config.global.profiling.port | int | `65432` | pprof port, if enabled |
| dendrite_config.mscs | object | `{"mscs":["msc2946"]}` | Configuration for experimental MSC's. (Valid values are: msc2836 and msc2946) |
| dendrite_config.mscs | object | `{"mscs":[]}` | Configuration for experimental MSC's. (Valid values are: msc2836) |
| dendrite_config.app_service_api.disable_tls_validation | bool | `false` | Disable the validation of TLS certificates of appservices. This is not recommended in production since it may allow appservice traffic to be sent to an insecure endpoint. |
| dendrite_config.app_service_api.config_files | list | `[]` | Appservice config files to load on startup. (**NOTE**: This is overriden by Helm, if a folder `./appservices/` exists) |
| dendrite_config.client_api.registration_disabled | bool | `true` | Prevents new users from being able to register on this homeserver, except when using the registration shared secret below. |
@ -144,12 +147,11 @@ Create a folder `appservices` and place your configurations in there. The confi
| postgresql.auth.password | string | `"changeme"` | |
| postgresql.auth.database | string | `"dendrite"` | |
| postgresql.persistence.enabled | bool | `false` | |
| ingress.enabled | bool | `false` | Create an ingress for a monolith deployment |
| ingress.hosts | list | `[]` | |
| ingress.className | string | `""` | |
| ingress.hostName | string | `""` | |
| ingress.enabled | bool | `false` | Create an ingress for the deployment |
| ingress.className | string | `""` | The ingressClass to use. Will be converted to annotation if not yet supported. |
| ingress.annotations | object | `{}` | Extra, custom annotations |
| ingress.tls | list | `[]` | |
| ingress.hostName | string | `""` | The ingress hostname for your matrix server. Should align with the server_name and well_known_* hosts. If not set, generated from the dendrite_config values. |
| ingress.tls | list | `[]` | TLS configuration. Should contain information for the server_name and well-known hosts. Alternatively, set tls.generate=true to generate defaults based on the dendrite_config. |
| service.type | string | `"ClusterIP"` | |
| service.port | int | `8008` | |
| prometheus.servicemonitor.enabled | bool | `false` | Enable ServiceMonitor for Prometheus-Operator for scrape metric-endpoint |
@ -187,3 +189,5 @@ grafana:
```
PS: The label `release=kube-prometheus-stack` is setup with the helmchart of the Prometheus Operator. For Grafana Dashboards it may be necessary to enable scanning in the correct namespaces (or ALL), enabled by `sidecar.dashboards.searchNamespace` in [Helmchart of grafana](https://artifacthub.io/packages/helm/grafana/grafana) (which is part of PrometheusOperator, so `grafana.sidecar.dashboards.searchNamespace`)
----------------------------------------------
Autogenerated from chart metadata using [helm-docs v1.11.0](https://github.com/norwoodj/helm-docs/releases/v1.11.0)

6
helm/dendrite/values.yaml
View file

@ -211,14 +211,12 @@ dendrite_config:
# -- pprof port, if enabled
port: 65432
# -- Configuration for experimental MSC's. (Valid values are: msc2836 and msc2946)
# -- Configuration for experimental MSC's. (Valid values are: msc2836)
mscs:
mscs:
- msc2946
mscs: []
# A list of enabled MSC's
# Currently valid values are:
# - msc2836 (Threading, see https://github.com/matrix-org/matrix-doc/pull/2836)
# - msc2946 (Spaces Summary, see https://github.com/matrix-org/matrix-doc/pull/2946)
app_service_api:
# -- Disable the validation of TLS certificates of appservices. This is

1
internal/caching/cache_roomservernids.go
View file

@ -8,6 +8,7 @@ type RoomServerCaches interface {
RoomServerNIDsCache
RoomVersionCache
RoomServerEventsCache
RoomHierarchyCache
EventStateKeyCache
EventTypeCache
}

15
internal/caching/cache_space_rooms.go
View file

@ -2,15 +2,16 @@ package caching
import "github.com/matrix-org/gomatrixserverlib/fclient"
type SpaceSummaryRoomsCache interface {
GetSpaceSummary(roomID string) (r fclient.MSC2946SpacesResponse, ok bool)
StoreSpaceSummary(roomID string, r fclient.MSC2946SpacesResponse)
// RoomHierarchy cache caches responses to federated room hierarchy requests (A.K.A. 'space summaries')
type RoomHierarchyCache interface {
GetRoomHierarchy(roomID string) (r fclient.RoomHierarchyResponse, ok bool)
StoreRoomHierarchy(roomID string, r fclient.RoomHierarchyResponse)
}
func (c Caches) GetSpaceSummary(roomID string) (r fclient.MSC2946SpacesResponse, ok bool) {
return c.SpaceSummaryRooms.Get(roomID)
func (c Caches) GetRoomHierarchy(roomID string) (r fclient.RoomHierarchyResponse, ok bool) {
return c.RoomHierarchies.Get(roomID)
}
func (c Caches) StoreSpaceSummary(roomID string, r fclient.MSC2946SpacesResponse) {
c.SpaceSummaryRooms.Set(roomID, r)
func (c Caches) StoreRoomHierarchy(roomID string, r fclient.RoomHierarchyResponse) {
c.RoomHierarchies.Set(roomID, r)
}

2
internal/caching/caches.go
View file

@ -35,7 +35,7 @@ type Caches struct {
RoomServerEventTypes Cache[types.EventTypeNID, string] // eventType NID -> eventType
FederationPDUs Cache[int64, *types.HeaderedEvent] // queue NID -> PDU
FederationEDUs Cache[int64, *gomatrixserverlib.EDU] // queue NID -> EDU
SpaceSummaryRooms Cache[string, fclient.MSC2946SpacesResponse] // room ID -> space response
RoomHierarchies Cache[string, fclient.RoomHierarchyResponse] // room ID -> space response
LazyLoading Cache[lazyLoadingCacheKey, string] // composite key -> event ID
}

2
internal/caching/impl_ristretto.go
View file

@ -147,7 +147,7 @@ func NewRistrettoCache(maxCost config.DataUnit, maxAge time.Duration, enableProm
MaxAge: lesserOf(time.Hour/2, maxAge),
},
},
SpaceSummaryRooms: &RistrettoCachePartition[string, fclient.MSC2946SpacesResponse]{ // room ID -> space response
RoomHierarchies: &RistrettoCachePartition[string, fclient.RoomHierarchyResponse]{ // room ID -> space response
cache: cache,
Prefix: spaceSummaryRoomsCache,
Mutable: true,

51
internal/sqlutil/connection_manager.go
View file

@ -17,42 +17,59 @@ package sqlutil
import (
"database/sql"
"fmt"
"sync"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"
)
type Connections struct {
db *sql.DB
writer Writer
globalConfig config.DatabaseOptions
processContext *process.ProcessContext
existingConnections sync.Map
}
func NewConnectionManager(processCtx *process.ProcessContext, globalConfig config.DatabaseOptions) Connections {
return Connections{
type con struct {
db *sql.DB
writer Writer
}
func NewConnectionManager(processCtx *process.ProcessContext, globalConfig config.DatabaseOptions) *Connections {
return &Connections{
globalConfig: globalConfig,
processContext: processCtx,
}
}
func (c *Connections) Connection(dbProperties *config.DatabaseOptions) (*sql.DB, Writer, error) {
var err error
// If no connectionString was provided, try the global one
if dbProperties.ConnectionString == "" {
dbProperties = &c.globalConfig
// If we still don't have a connection string, that's a problem
if dbProperties.ConnectionString == "" {
return nil, nil, fmt.Errorf("no database connections configured")
}
}
writer := NewDummyWriter()
if dbProperties.ConnectionString.IsSQLite() {
writer = NewExclusiveWriter()
}
var err error
if dbProperties.ConnectionString == "" {
// if no connectionString was provided, try the global one
dbProperties = &c.globalConfig
existing, loaded := c.existingConnections.LoadOrStore(dbProperties.ConnectionString, &con{})
if loaded {
// We found an existing connection
ex := existing.(*con)
return ex.db, ex.writer, nil
}
if dbProperties.ConnectionString != "" || c.db == nil {
// Open a new database connection using the supplied config.
c.db, err = Open(dbProperties, writer)
db, err := Open(dbProperties, writer)
if err != nil {
return nil, nil, err
}
c.writer = writer
c.existingConnections.Store(dbProperties.ConnectionString, &con{db: db, writer: writer})
go func() {
if c.processContext == nil {
return
@ -61,15 +78,9 @@ func (c *Connections) Connection(dbProperties *config.DatabaseOptions) (*sql.DB,
// Dendrite to shut down to cleanly close the database connection.
c.processContext.ComponentStarted()
<-c.processContext.WaitForShutdown()
_ = c.db.Close()
_ = db.Close()
c.processContext.ComponentFinished()
}()
return c.db, c.writer, nil
}
if c.db != nil && c.writer != nil {
// Ignore the supplied config and return the global pool and
// writer.
return c.db, c.writer, nil
}
return nil, nil, fmt.Errorf("no database connections configured")
return db, writer, nil
}

96
internal/sqlutil/connection_manager_test.go
View file

@ -6,10 +6,13 @@ import (
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/test"
)
func TestConnectionManager(t *testing.T) {
t.Run("component defined connection string", func(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
conStr, close := test.PrepareDBConnectionString(t, dbType)
t.Cleanup(close)
@ -34,23 +37,104 @@ func TestConnectionManager(t *testing.T) {
}
}
// test global db pool
dbGlobal, writerGlobal, err := cm.Connection(&config.DatabaseOptions{})
// reuse existing connection
db2, writer2, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(db, dbGlobal) {
if !reflect.DeepEqual(db, db2) {
t.Fatalf("expected database connection to be reused")
}
if !reflect.DeepEqual(writer, writerGlobal) {
if !reflect.DeepEqual(writer, writer2) {
t.Fatalf("expected database writer to be reused")
}
// This test does not work with Postgres, because we can't just simply append
// "x" or replace the database to use.
if dbType == test.DBTypePostgres {
return
}
// Test different connection string
dbProps = &config.DatabaseOptions{ConnectionString: config.DataSource(conStr + "x")}
db3, _, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
}
if reflect.DeepEqual(db, db3) {
t.Fatalf("expected different database connection")
}
})
})
t.Run("global connection pool", func(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
conStr, close := test.PrepareDBConnectionString(t, dbType)
t.Cleanup(close)
cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{ConnectionString: config.DataSource(conStr)})
dbProps := &config.DatabaseOptions{}
db, writer, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
}
switch dbType {
case test.DBTypeSQLite:
_, ok := writer.(*sqlutil.ExclusiveWriter)
if !ok {
t.Fatalf("expected exclusive writer")
}
case test.DBTypePostgres:
_, ok := writer.(*sqlutil.DummyWriter)
if !ok {
t.Fatalf("expected dummy writer")
}
}
// reuse existing connection
db2, writer2, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(db, db2) {
t.Fatalf("expected database connection to be reused")
}
if !reflect.DeepEqual(writer, writer2) {
t.Fatalf("expected database writer to be reused")
}
})
})
t.Run("shutdown", func(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
conStr, close := test.PrepareDBConnectionString(t, dbType)
t.Cleanup(close)
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, config.DatabaseOptions{ConnectionString: config.DataSource(conStr)})
dbProps := &config.DatabaseOptions{}
_, _, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
}
processCtx.ShutdownDendrite()
processCtx.WaitForComponentsToFinish()
})
})
// test invalid connection string configured
cm2 := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
_, _, err = cm2.Connection(&config.DatabaseOptions{ConnectionString: "http://"})
_, _, err := cm2.Connection(&config.DatabaseOptions{ConnectionString: "http://"})
if err == nil {
t.Fatal("expected an error but got none")
}
// empty connection string is not allowed
_, _, err = cm2.Connection(&config.DatabaseOptions{})
if err == nil {
t.Fatal("expected an error but got none")
}
})
}

2
mediaapi/mediaapi.go
View file

@ -28,7 +28,7 @@ import (
// AddPublicRoutes sets up and registers HTTP handlers for the MediaAPI component.
func AddPublicRoutes(
mediaRouter *mux.Router,
cm sqlutil.Connections,
cm *sqlutil.Connections,
cfg *config.Dendrite,
userAPI userapi.MediaUserAPI,
client *fclient.Client,

2
mediaapi/storage/postgres/mediaapi.go
View file

@ -24,7 +24,7 @@ import (
)
// NewDatabase opens a postgres database.
func NewDatabase(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.Database, error) {
func NewDatabase(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.Database, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err

2
mediaapi/storage/sqlite3/mediaapi.go
View file

@ -23,7 +23,7 @@ import (
)
// NewDatabase opens a SQLIte database.
func NewDatabase(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.Database, error) {
func NewDatabase(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.Database, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err

2
mediaapi/storage/storage.go
View file

@ -27,7 +27,7 @@ import (
)
// NewMediaAPIDatasource opens a database connection.
func NewMediaAPIDatasource(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (Database, error) {
func NewMediaAPIDatasource(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewDatabase(conMan, dbProperties)

2
relayapi/relayapi.go
View file

@ -53,7 +53,7 @@ func AddPublicRoutes(
func NewRelayInternalAPI(
dendriteCfg *config.Dendrite,
cm sqlutil.Connections,
cm *sqlutil.Connections,
fedClient fclient.FederationClient,
rsAPI rsAPI.RoomserverInternalAPI,
keyRing *gomatrixserverlib.KeyRing,

2
relayapi/storage/postgres/storage.go
View file

@ -33,7 +33,7 @@ type Database struct {
// NewDatabase opens a new database
func NewDatabase(
conMan sqlutil.Connections,
conMan *sqlutil.Connections,
dbProperties *config.DatabaseOptions,
cache caching.FederationCache,
isLocalServerName func(spec.ServerName) bool,

2
relayapi/storage/sqlite3/storage.go
View file

@ -33,7 +33,7 @@ type Database struct {
// NewDatabase opens a new database
func NewDatabase(
conMan sqlutil.Connections,
conMan *sqlutil.Connections,
dbProperties *config.DatabaseOptions,
cache caching.FederationCache,
isLocalServerName func(spec.ServerName) bool,

2
relayapi/storage/storage.go
View file

@ -30,7 +30,7 @@ import (
// NewDatabase opens a new database
func NewDatabase(
conMan sqlutil.Connections,
conMan *sqlutil.Connections,
dbProperties *config.DatabaseOptions,
cache caching.FederationCache,
isLocalServerName func(spec.ServerName) bool,

24
roomserver/api/api.go
View file

@ -34,6 +34,17 @@ func (e ErrNotAllowed) Error() string {
return e.Err.Error()
}
// ErrRoomUnknownOrNotAllowed is an error return if either the provided
// room ID does not exist, or points to a room that the requester does
// not have access to.
type ErrRoomUnknownOrNotAllowed struct {
Err error
}
func (e ErrRoomUnknownOrNotAllowed) Error() string {
return e.Err.Error()
}
type RestrictedJoinAPI interface {
CurrentStateEvent(ctx context.Context, roomID spec.RoomID, eventType string, stateKey string) (gomatrixserverlib.PDU, error)
InvitePending(ctx context.Context, roomID spec.RoomID, senderID spec.SenderID) (bool, error)
@ -113,6 +124,17 @@ type QueryEventsAPI interface {
QueryCurrentState(ctx context.Context, req *QueryCurrentStateRequest, res *QueryCurrentStateResponse) error
}
type QueryRoomHierarchyAPI interface {
// Traverse the room hierarchy using the provided walker up to the provided limit,
// returning a new walker which can be used to fetch the next page.
//
// If limit is -1, this is treated as no limit, and the entire hierarchy will be traversed.
//
// If returned walker is nil, then there are no more rooms left to traverse. This method does not modify the provided walker, so it
// can be cached.
QueryNextRoomHierarchyPage(ctx context.Context, walker RoomHierarchyWalker, limit int) ([]fclient.RoomHierarchyRoom, *RoomHierarchyWalker, error)
}
// API functions required by the syncapi
type SyncRoomserverAPI interface {
QueryLatestEventsAndStateAPI
@ -187,6 +209,7 @@ type ClientRoomserverAPI interface {
QueryEventsAPI
QuerySenderIDAPI
UserRoomPrivateKeyCreator
QueryRoomHierarchyAPI
QueryMembershipForUser(ctx context.Context, req *QueryMembershipForUserRequest, res *QueryMembershipForUserResponse) error
QueryMembershipsForRoom(ctx context.Context, req *QueryMembershipsForRoomRequest, res *QueryMembershipsForRoomResponse) error
QueryRoomsForUser(ctx context.Context, req *QueryRoomsForUserRequest, res *QueryRoomsForUserResponse) error
@ -236,6 +259,7 @@ type FederationRoomserverAPI interface {
QueryLatestEventsAndStateAPI
QueryBulkStateContentAPI
QuerySenderIDAPI
QueryRoomHierarchyAPI
UserRoomPrivateKeyCreator
CurrentMembership(ctx context.Context, roomID spec.RoomID, senderID spec.SenderID) (string, error)
AssignRoomNID(ctx context.Context, roomID spec.RoomID, roomVersion gomatrixserverlib.RoomVersion) (roomNID types.RoomNID, err error)

76
roomserver/api/query.go
View file

@ -503,3 +503,79 @@ func (mq *MembershipQuerier) CurrentMembership(ctx context.Context, roomID spec.
}
return membership, err
}
type QueryRoomHierarchyRequest struct {
SuggestedOnly bool `json:"suggested_only"`
Limit int `json:"limit"`
MaxDepth int `json:"max_depth"`
From int `json:"json"`
}
// A struct storing the intermediate state of a room hierarchy query for pagination purposes.
//
// Used for implementing space summaries / room hierarchies
//
// Use NewRoomHierarchyWalker to construct this, and QueryNextRoomHierarchyPage on the roomserver API
// to traverse the room hierarchy.
type RoomHierarchyWalker struct {
RootRoomID spec.RoomID
Caller types.DeviceOrServerName
SuggestedOnly bool
MaxDepth int
Processed RoomSet
Unvisited []RoomHierarchyWalkerQueuedRoom
}
type RoomHierarchyWalkerQueuedRoom struct {
RoomID spec.RoomID
ParentRoomID *spec.RoomID
Depth int
Vias []string // vias to query this room by
}
// Create a new room hierarchy walker, starting from the provided root room ID.
//
// Use the resulting struct with QueryNextRoomHierarchyPage on the roomserver API to traverse the room hierarchy.
func NewRoomHierarchyWalker(caller types.DeviceOrServerName, roomID spec.RoomID, suggestedOnly bool, maxDepth int) RoomHierarchyWalker {
walker := RoomHierarchyWalker{
RootRoomID: roomID,
Caller: caller,
SuggestedOnly: suggestedOnly,
MaxDepth: maxDepth,
Unvisited: []RoomHierarchyWalkerQueuedRoom{{
RoomID: roomID,
ParentRoomID: nil,
Depth: 0,
}},
Processed: NewRoomSet(),
}
return walker
}
// A set of room IDs.
type RoomSet map[spec.RoomID]struct{}
// Create a new empty room set.
func NewRoomSet() RoomSet {
return RoomSet{}
}
// Check if a room ID is in a room set.
func (s RoomSet) Contains(val spec.RoomID) bool {
_, ok := s[val]
return ok
}
// Add a room ID to a room set.
func (s RoomSet) Add(val spec.RoomID) {
s[val] = struct{}{}
}
func (s RoomSet) Copy() RoomSet {
copied := make(RoomSet, len(s))
for k := range s {
copied.Add(k)
}
return copied
}

19
roomserver/internal/api.go
View file

@ -91,15 +91,8 @@ func NewRoomserverAPI(
NATSClient: nc,
Durable: dendriteCfg.Global.JetStream.Durable("RoomserverInputConsumer"),
ServerACLs: serverACLs,
Queryer: &query.Queryer{
DB: roomserverDB,
Cache: caches,
IsLocalServerName: dendriteCfg.Global.IsLocalServerName,
ServerACLs: serverACLs,
Cfg: dendriteCfg,
},
enableMetrics: enableMetrics,
// perform-er structs get initialised when we have a federation sender to use
// perform-er structs + queryer struct get initialised when we have a federation sender to use
}
return a
}
@ -111,6 +104,15 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
r.fsAPI = fsAPI
r.KeyRing = keyRing
r.Queryer = &query.Queryer{
DB: r.DB,
Cache: r.Cache,
IsLocalServerName: r.Cfg.Global.IsLocalServerName,
ServerACLs: r.ServerACLs,
Cfg: r.Cfg,
FSAPI: fsAPI,
}
r.Inputer = &input.Inputer{
Cfg: &r.Cfg.RoomServer,
ProcessContext: r.ProcessContext,
@ -123,6 +125,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
ServerName: r.ServerName,
SigningIdentity: r.SigningIdentityFor,
FSAPI: fsAPI,
RSAPI: r,
KeyRing: keyRing,
ACLs: r.ServerACLs,
Queryer: r.Queryer,

1
roomserver/internal/input/input.go
View file

@ -83,6 +83,7 @@ type Inputer struct {
ServerName spec.ServerName
SigningIdentity func(ctx context.Context, roomID spec.RoomID, senderID spec.UserID) (fclient.SigningIdentity, error)
FSAPI fedapi.RoomserverFederationAPI
RSAPI api.RoomserverInternalAPI
KeyRing gomatrixserverlib.JSONVerifier
ACLs *acls.ServerACLs
InputRoomEventTopic string

18
roomserver/internal/input/input_events.go
View file

@ -449,6 +449,24 @@ func (r *Inputer) processRoomEvent(
return nil
}
// TODO: Revist this to ensure we don't replace a current state mxid_mapping with an older one.
if event.Version() == gomatrixserverlib.RoomVersionPseudoIDs && event.Type() == spec.MRoomMember {
mapping := gomatrixserverlib.MemberContent{}
if err = json.Unmarshal(event.Content(), &mapping); err != nil {
return err
}
if mapping.MXIDMapping != nil {
storeUserID, userErr := spec.NewUserID(mapping.MXIDMapping.UserID, true)
if userErr != nil {
return userErr
}
err = r.RSAPI.StoreUserRoomPublicKey(ctx, mapping.MXIDMapping.UserRoomKey, *storeUserID, *validRoomID)
if err != nil {
return fmt.Errorf("failed storing user room public key: %w", err)
}
}
}
switch input.Kind {
case api.KindNew:
if err = r.updateLatestEvents(

6
roomserver/internal/perform/perform_join.go
View file

@ -274,7 +274,6 @@ func (r *Joiner) performJoinRoomByID(
// If we should do a forced federated join then do that.
var joinedVia spec.ServerName
if forceFederatedJoin {
// TODO : pseudoIDs - pass through userID here since we don't know what the senderID should be yet
joinedVia, err = r.performFederatedJoinRoomByID(ctx, req)
return req.RoomIDOrAlias, joinedVia, err
}
@ -286,10 +285,7 @@ func (r *Joiner) performJoinRoomByID(
// but everyone has since left. I suspect it does the wrong thing.
var buildRes rsAPI.QueryLatestEventsAndStateResponse
identity, err := r.RSAPI.SigningIdentityFor(ctx, *roomID, *userID)
if err != nil {
return "", "", fmt.Errorf("error joining local room: %q", err)
}
identity := r.Cfg.Matrix.SigningIdentity
// at this point we know we have an existing room
if inRoomRes.RoomVersion == gomatrixserverlib.RoomVersionPseudoIDs {

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

@ -32,6 +32,7 @@ import (
"github.com/matrix-org/dendrite/syncapi/synctypes"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/roomserver/acls"
"github.com/matrix-org/dendrite/roomserver/api"
@ -47,6 +48,7 @@ type Queryer struct {
IsLocalServerName func(spec.ServerName) bool
ServerACLs *acls.ServerACLs
Cfg *config.Dendrite
FSAPI fsAPI.RoomserverFederationAPI
}
func (r *Queryer) RestrictedRoomJoinInfo(ctx context.Context, roomID spec.RoomID, senderID spec.SenderID, localServerName spec.ServerName) (*gomatrixserverlib.RestrictedRoomJoinInfo, error) {

530
roomserver/internal/query/query_room_hierarchy.go Normal file
View file

@ -0,0 +1,530 @@
// 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 query
import (
"context"
"encoding/json"
"fmt"
"sort"
fs "github.com/matrix-org/dendrite/federationapi/api"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/types"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/fclient"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
"github.com/tidwall/gjson"
)
// Traverse the room hierarchy using the provided walker up to the provided limit,
// returning a new walker which can be used to fetch the next page.
//
// If limit is -1, this is treated as no limit, and the entire hierarchy will be traversed.
//
// If returned walker is nil, then there are no more rooms left to traverse. This method does not modify the provided walker, so it
// can be cached.
func (querier *Queryer) QueryNextRoomHierarchyPage(ctx context.Context, walker roomserver.RoomHierarchyWalker, limit int) ([]fclient.RoomHierarchyRoom, *roomserver.RoomHierarchyWalker, error) {
if authorised, _ := authorised(ctx, querier, walker.Caller, walker.RootRoomID, nil); !authorised {
return nil, nil, roomserver.ErrRoomUnknownOrNotAllowed{Err: fmt.Errorf("room is unknown/forbidden")}
}
discoveredRooms := []fclient.RoomHierarchyRoom{}
// Copy unvisited and processed to avoid modifying original walker (which is typically in cache)
unvisited := make([]roomserver.RoomHierarchyWalkerQueuedRoom, len(walker.Unvisited))
copy(unvisited, walker.Unvisited)
processed := walker.Processed.Copy()
// Depth first -> stack data structure
for len(unvisited) > 0 {
if len(discoveredRooms) >= limit && limit != -1 {
break
}
// pop the stack
queuedRoom := unvisited[len(unvisited)-1]
unvisited = unvisited[:len(unvisited)-1]
// If this room has already been processed, skip.
// If this room exceeds the specified depth, skip.
if processed.Contains(queuedRoom.RoomID) || (walker.MaxDepth > 0 && queuedRoom.Depth > walker.MaxDepth) {
continue
}
// Mark this room as processed.
processed.Add(queuedRoom.RoomID)
// if this room is not a space room, skip.
var roomType string
create := stateEvent(ctx, querier, queuedRoom.RoomID, spec.MRoomCreate, "")
if create != nil {
var createContent gomatrixserverlib.CreateContent
err := json.Unmarshal(create.Content(), &createContent)
if err != nil {
util.GetLogger(ctx).WithError(err).WithField("create_content", create.Content()).Warn("failed to unmarshal m.room.create event")
}
roomType = createContent.RoomType
}
// Collect rooms/events to send back (either locally or fetched via federation)
var discoveredChildEvents []fclient.RoomHierarchyStrippedEvent
// If we know about this room and the caller is authorised (joined/world_readable) then pull
// events locally
roomExists := roomExists(ctx, querier, queuedRoom.RoomID)
if !roomExists {
// attempt to query this room over federation, as either we've never heard of it before
// or we've left it and hence are not authorised (but info may be exposed regardless)
fedRes := federatedRoomInfo(ctx, querier, walker.Caller, walker.SuggestedOnly, queuedRoom.RoomID, queuedRoom.Vias)
if fedRes != nil {
discoveredChildEvents = fedRes.Room.ChildrenState
discoveredRooms = append(discoveredRooms, fedRes.Room)
if len(fedRes.Children) > 0 {
discoveredRooms = append(discoveredRooms, fedRes.Children...)
}
// mark this room as a space room as the federated server responded.
// we need to do this so we add the children of this room to the unvisited stack
// as these children may be rooms we do know about.
roomType = spec.MSpace
}
} else if authorised, isJoinedOrInvited := authorised(ctx, querier, walker.Caller, queuedRoom.RoomID, queuedRoom.ParentRoomID); authorised {
// Get all `m.space.child` state events for this room
events, err := childReferences(ctx, querier, walker.SuggestedOnly, queuedRoom.RoomID)
if err != nil {
util.GetLogger(ctx).WithError(err).WithField("room_id", queuedRoom.RoomID).Error("failed to extract references for room")
continue
}
discoveredChildEvents = events
pubRoom := publicRoomsChunk(ctx, querier, queuedRoom.RoomID)
discoveredRooms = append(discoveredRooms, fclient.RoomHierarchyRoom{
PublicRoom: *pubRoom,
RoomType: roomType,
ChildrenState: events,
})
// don't walk children if the user is not joined/invited to the space
if !isJoinedOrInvited {
continue
}
} else {
// room exists but user is not authorised
continue
}
// don't walk the children
// if the parent is not a space room
if roomType != spec.MSpace {
continue
}
// For each referenced room ID in the child events being returned to the caller
// add the room ID to the queue of unvisited rooms. Loop from the beginning.
// We need to invert the order here because the child events are lo->hi on the timestamp,
// so we need to ensure we pop in the same lo->hi order, which won't be the case if we
// insert the highest timestamp last in a stack.
for i := len(discoveredChildEvents) - 1; i >= 0; i-- {
spaceContent := struct {
Via []string `json:"via"`
}{}
ev := discoveredChildEvents[i]
_ = json.Unmarshal(ev.Content, &spaceContent)
childRoomID, err := spec.NewRoomID(ev.StateKey)
if err != nil {
util.GetLogger(ctx).WithError(err).WithField("invalid_room_id", ev.StateKey).WithField("parent_room_id", queuedRoom.RoomID).Warn("Invalid room ID in m.space.child state event")
} else {
unvisited = append(unvisited, roomserver.RoomHierarchyWalkerQueuedRoom{
RoomID: *childRoomID,
ParentRoomID: &queuedRoom.RoomID,
Depth: queuedRoom.Depth + 1,
Vias: spaceContent.Via,
})
}
}
}
if len(unvisited) == 0 {
// If no more rooms to walk, then don't return a walker for future pages
return discoveredRooms, nil, nil
} else {
// If there are more rooms to walk, then return a new walker to resume walking from (for querying more pages)
newWalker := roomserver.RoomHierarchyWalker{
RootRoomID: walker.RootRoomID,
Caller: walker.Caller,
SuggestedOnly: walker.SuggestedOnly,
MaxDepth: walker.MaxDepth,
Unvisited: unvisited,
Processed: processed,
}
return discoveredRooms, &newWalker, nil
}
}
// authorised returns true iff the user is joined this room or the room is world_readable
func authorised(ctx context.Context, querier *Queryer, caller types.DeviceOrServerName, roomID spec.RoomID, parentRoomID *spec.RoomID) (authed, isJoinedOrInvited bool) {
if clientCaller := caller.Device(); clientCaller != nil {
return authorisedUser(ctx, querier, clientCaller, roomID, parentRoomID)
} else {
return authorisedServer(ctx, querier, roomID, *caller.ServerName()), false
}
}
// authorisedServer returns true iff the server is joined this room or the room is world_readable, public, or knockable
func authorisedServer(ctx context.Context, querier *Queryer, roomID spec.RoomID, callerServerName spec.ServerName) bool {
// Check history visibility / join rules first
hisVisTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomHistoryVisibility,
StateKey: "",
}
joinRuleTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomJoinRules,
StateKey: "",
}
var queryRoomRes roomserver.QueryCurrentStateResponse
err := querier.QueryCurrentState(ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID.String(),
StateTuples: []gomatrixserverlib.StateKeyTuple{
hisVisTuple, joinRuleTuple,
},
}, &queryRoomRes)
if err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to QueryCurrentState")
return false
}
hisVisEv := queryRoomRes.StateEvents[hisVisTuple]
if hisVisEv != nil {
hisVis, _ := hisVisEv.HistoryVisibility()
if hisVis == "world_readable" {
return true
}
}
// check if this room is a restricted room and if so, we need to check if the server is joined to an allowed room ID
// in addition to the actual room ID (but always do the actual one first as it's quicker in the common case)
allowJoinedToRoomIDs := []spec.RoomID{roomID}
joinRuleEv := queryRoomRes.StateEvents[joinRuleTuple]
if joinRuleEv != nil {
rule, ruleErr := joinRuleEv.JoinRule()
if ruleErr != nil {
util.GetLogger(ctx).WithError(ruleErr).WithField("parent_room_id", roomID).Warn("failed to get join rule")
return false
}
if rule == spec.Public || rule == spec.Knock {
return true
}
if rule == spec.Restricted {
allowJoinedToRoomIDs = append(allowJoinedToRoomIDs, restrictedJoinRuleAllowedRooms(ctx, joinRuleEv)...)
}
}
// check if server is joined to any allowed room
for _, allowedRoomID := range allowJoinedToRoomIDs {
var queryRes fs.QueryJoinedHostServerNamesInRoomResponse
err = querier.FSAPI.QueryJoinedHostServerNamesInRoom(ctx, &fs.QueryJoinedHostServerNamesInRoomRequest{
RoomID: allowedRoomID.String(),
}, &queryRes)
if err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to QueryJoinedHostServerNamesInRoom")
continue
}
for _, srv := range queryRes.ServerNames {
if srv == callerServerName {
return true
}
}
}
return false
}
// authorisedUser returns true iff the user is invited/joined this room or the room is world_readable
// or if the room has a public or knock join rule.
// Failing that, if the room has a restricted join rule and belongs to the space parent listed, it will return true.
func authorisedUser(ctx context.Context, querier *Queryer, clientCaller *userapi.Device, roomID spec.RoomID, parentRoomID *spec.RoomID) (authed bool, isJoinedOrInvited bool) {
hisVisTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomHistoryVisibility,
StateKey: "",
}
joinRuleTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomJoinRules,
StateKey: "",
}
roomMemberTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomMember,
StateKey: clientCaller.UserID,
}
var queryRes roomserver.QueryCurrentStateResponse
err := querier.QueryCurrentState(ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID.String(),
StateTuples: []gomatrixserverlib.StateKeyTuple{
hisVisTuple, joinRuleTuple, roomMemberTuple,
},
}, &queryRes)
if err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to QueryCurrentState")
return false, false
}
memberEv := queryRes.StateEvents[roomMemberTuple]
if memberEv != nil {
membership, _ := memberEv.Membership()
if membership == spec.Join || membership == spec.Invite {
return true, true
}
}
hisVisEv := queryRes.StateEvents[hisVisTuple]
if hisVisEv != nil {
hisVis, _ := hisVisEv.HistoryVisibility()
if hisVis == "world_readable" {
return true, false
}
}
joinRuleEv := queryRes.StateEvents[joinRuleTuple]
if parentRoomID != nil && joinRuleEv != nil {
var allowed bool
rule, ruleErr := joinRuleEv.JoinRule()
if ruleErr != nil {
util.GetLogger(ctx).WithError(ruleErr).WithField("parent_room_id", parentRoomID).Warn("failed to get join rule")
} else if rule == spec.Public || rule == spec.Knock {
allowed = true
} else if rule == spec.Restricted {
allowedRoomIDs := restrictedJoinRuleAllowedRooms(ctx, joinRuleEv)
// check parent is in the allowed set
for _, a := range allowedRoomIDs {
if *parentRoomID == a {
allowed = true
break
}
}
}
if allowed {
// ensure caller is joined to the parent room
var queryRes2 roomserver.QueryCurrentStateResponse
err = querier.QueryCurrentState(ctx, &roomserver.QueryCurrentStateRequest{
RoomID: parentRoomID.String(),
StateTuples: []gomatrixserverlib.StateKeyTuple{
roomMemberTuple,
},
}, &queryRes2)
if err != nil {
util.GetLogger(ctx).WithError(err).WithField("parent_room_id", parentRoomID).Warn("failed to check user is joined to parent room")
} else {
memberEv = queryRes2.StateEvents[roomMemberTuple]
if memberEv != nil {
membership, _ := memberEv.Membership()
if membership == spec.Join {
return true, false
}
}
}
}
}
return false, false
}
// helper function to fetch a state event
func stateEvent(ctx context.Context, querier *Queryer, roomID spec.RoomID, evType, stateKey string) *types.HeaderedEvent {
var queryRes roomserver.QueryCurrentStateResponse
tuple := gomatrixserverlib.StateKeyTuple{
EventType: evType,
StateKey: stateKey,
}
err := querier.QueryCurrentState(ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID.String(),
StateTuples: []gomatrixserverlib.StateKeyTuple{tuple},
}, &queryRes)
if err != nil {
return nil
}
return queryRes.StateEvents[tuple]
}
// returns true if the current server is participating in the provided room
func roomExists(ctx context.Context, querier *Queryer, roomID spec.RoomID) bool {
var queryRes roomserver.QueryServerJoinedToRoomResponse
err := querier.QueryServerJoinedToRoom(ctx, &roomserver.QueryServerJoinedToRoomRequest{
RoomID: roomID.String(),
ServerName: querier.Cfg.Global.ServerName,
}, &queryRes)
if err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to QueryServerJoinedToRoom")
return false
}
// if the room exists but we aren't in the room then we might have stale data so we want to fetch
// it fresh via federation
return queryRes.RoomExists && queryRes.IsInRoom
}
// federatedRoomInfo returns more of the spaces graph from another server. Returns nil if this was
// unsuccessful.
func federatedRoomInfo(ctx context.Context, querier *Queryer, caller types.DeviceOrServerName, suggestedOnly bool, roomID spec.RoomID, vias []string) *fclient.RoomHierarchyResponse {
// only do federated requests for client requests
if caller.Device() == nil {
return nil
}
resp, ok := querier.Cache.GetRoomHierarchy(roomID.String())
if ok {
util.GetLogger(ctx).Debugf("Returning cached response for %s", roomID)
return &resp
}
util.GetLogger(ctx).Debugf("Querying %s via %+v", roomID, vias)
innerCtx := context.Background()
// query more of the spaces graph using these servers
for _, serverName := range vias {
if serverName == string(querier.Cfg.Global.ServerName) {
continue
}
res, err := querier.FSAPI.RoomHierarchies(innerCtx, querier.Cfg.Global.ServerName, spec.ServerName(serverName), roomID.String(), suggestedOnly)
if err != nil {
util.GetLogger(ctx).WithError(err).Warnf("failed to call RoomHierarchies on server %s", serverName)
continue
}
// ensure nil slices are empty as we send this to the client sometimes
if res.Room.ChildrenState == nil {
res.Room.ChildrenState = []fclient.RoomHierarchyStrippedEvent{}
}
for i := 0; i < len(res.Children); i++ {
child := res.Children[i]
if child.ChildrenState == nil {
child.ChildrenState = []fclient.RoomHierarchyStrippedEvent{}
}
res.Children[i] = child
}
querier.Cache.StoreRoomHierarchy(roomID.String(), res)
return &res
}
return nil
}
// references returns all child references pointing to or from this room.
func childReferences(ctx context.Context, querier *Queryer, suggestedOnly bool, roomID spec.RoomID) ([]fclient.RoomHierarchyStrippedEvent, error) {
createTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomCreate,
StateKey: "",
}
var res roomserver.QueryCurrentStateResponse
err := querier.QueryCurrentState(context.Background(), &roomserver.QueryCurrentStateRequest{
RoomID: roomID.String(),
AllowWildcards: true,
StateTuples: []gomatrixserverlib.StateKeyTuple{
createTuple, {
EventType: spec.MSpaceChild,
StateKey: "*",
},
},
}, &res)
if err != nil {
return nil, err
}
// don't return any child refs if the room is not a space room
if create := res.StateEvents[createTuple]; create != nil {
var createContent gomatrixserverlib.CreateContent
err := json.Unmarshal(create.Content(), &createContent)
if err != nil {
util.GetLogger(ctx).WithError(err).WithField("create_content", create.Content()).Warn("failed to unmarshal m.room.create event")
}
roomType := createContent.RoomType
if roomType != spec.MSpace {
return []fclient.RoomHierarchyStrippedEvent{}, nil
}
}
delete(res.StateEvents, createTuple)
el := make([]fclient.RoomHierarchyStrippedEvent, 0, len(res.StateEvents))
for _, ev := range res.StateEvents {
content := gjson.ParseBytes(ev.Content())
// only return events that have a `via` key as per MSC1772
// else we'll incorrectly walk redacted events (as the link
// is in the state_key)
if content.Get("via").Exists() {
strip := stripped(ev.PDU)
if strip == nil {
continue
}
// if suggested only and this child isn't suggested, skip it.
// if suggested only = false we include everything so don't need to check the content.
if suggestedOnly && !content.Get("suggested").Bool() {
continue
}
el = append(el, *strip)
}
}
// sort by origin_server_ts as per MSC2946
sort.Slice(el, func(i, j int) bool {
return el[i].OriginServerTS < el[j].OriginServerTS
})
return el, nil
}
// fetch public room information for provided room
func publicRoomsChunk(ctx context.Context, querier *Queryer, roomID spec.RoomID) *fclient.PublicRoom {
pubRooms, err := roomserver.PopulatePublicRooms(ctx, []string{roomID.String()}, querier)
if err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to PopulatePublicRooms")
return nil
}
if len(pubRooms) == 0 {
return nil
}
return &pubRooms[0]
}
func stripped(ev gomatrixserverlib.PDU) *fclient.RoomHierarchyStrippedEvent {
if ev.StateKey() == nil {
return nil
}
return &fclient.RoomHierarchyStrippedEvent{
Type: ev.Type(),
StateKey: *ev.StateKey(),
Content: ev.Content(),
Sender: string(ev.SenderID()),
OriginServerTS: ev.OriginServerTS(),
}
}
// given join_rule event, return list of rooms where membership of that room allows joining.
func restrictedJoinRuleAllowedRooms(ctx context.Context, joinRuleEv *types.HeaderedEvent) (allows []spec.RoomID) {
rule, _ := joinRuleEv.JoinRule()
if rule != spec.Restricted {
return nil
}
var jrContent gomatrixserverlib.JoinRuleContent
if err := json.Unmarshal(joinRuleEv.Content(), &jrContent); err != nil {
util.GetLogger(ctx).Warnf("failed to check join_rule on room %s: %s", joinRuleEv.RoomID(), err)
return nil
}
for _, allow := range jrContent.Allow {
if allow.Type == spec.MRoomMembership {
allowedRoomID, err := spec.NewRoomID(allow.RoomID)
if err != nil {
util.GetLogger(ctx).Warnf("invalid room ID '%s' found in join_rule on room %s: %s", allow.RoomID, joinRuleEv.RoomID(), err)
} else {
allows = append(allows, *allowedRoomID)
}
}
}
return
}

5
roomserver/roomserver.go
View file

@ -28,10 +28,13 @@ import (
)
// NewInternalAPI returns a concrete implementation of the internal API.
//
// Many of the methods provided by this API depend on access to a federation API, and so
// you may wish to call `SetFederationAPI` on the returned struct to avoid nil-dereference errors.
func NewInternalAPI(
processContext *process.ProcessContext,
cfg *config.Dendrite,
cm sqlutil.Connections,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
caches caching.RoomServerCaches,
enableMetrics bool,

7
roomserver/roomserver_test.go
View file

@ -249,13 +249,14 @@ func TestPurgeRoom(t *testing.T) {
defer jetstream.DeleteAllStreams(jsCtx, &cfg.Global.JetStream)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
fsAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(fsAPI, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
// Create the room
if err = api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
@ -1035,8 +1036,8 @@ func TestUpgrade(t *testing.T) {
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
rsAPI.SetFederationAPI(nil, nil)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
rsAPI.SetUserAPI(userAPI)
for _, tc := range testCases {

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

@ -37,7 +37,7 @@ type Database struct {
}
// Open a postgres database.
func Open(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) {
func Open(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) {
var d Database
var err error
db, writer, err := conMan.Connection(dbProperties)

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

@ -36,7 +36,7 @@ type Database struct {
}
// Open a sqlite database.
func Open(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) {
func Open(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) {
var d Database
var err error
db, writer, err := conMan.Connection(dbProperties)

2
roomserver/storage/storage.go
View file

@ -29,7 +29,7 @@ import (
)
// Open opens a database connection.
func Open(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (Database, error) {
func Open(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.Open(ctx, conMan, dbProperties, cache)

35
roomserver/types/types.go
View file

@ -22,7 +22,9 @@ import (
"strings"
"sync"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
"golang.org/x/crypto/blake2b"
)
@ -336,3 +338,36 @@ func (r *RoomInfo) CopyFrom(r2 *RoomInfo) {
}
var ErrorInvalidRoomInfo = fmt.Errorf("room info is invalid")
// Struct to represent a device or a server name.
//
// May be used to designate a caller for functions that can be called
// by a client (device) or by a server (server name).
//
// Exactly 1 of Device() and ServerName() will return a non-nil result.
type DeviceOrServerName struct {
device *userapi.Device
serverName *spec.ServerName
}
func NewDeviceNotServerName(device userapi.Device) DeviceOrServerName {
return DeviceOrServerName{
device: &device,
serverName: nil,
}
}
func NewServerNameNotDevice(serverName spec.ServerName) DeviceOrServerName {
return DeviceOrServerName{
device: nil,
serverName: &serverName,
}
}
func (s *DeviceOrServerName) Device() *userapi.Device {
return s.device
}
func (s *DeviceOrServerName) ServerName() *spec.ServerName {
return s.serverName
}

1
setup/config/config_mscs.go
View file

@ -7,7 +7,6 @@ type MSCs struct {
// 'msc2444': Peeking over federation - https://github.com/matrix-org/matrix-doc/pull/2444
// 'msc2753': Peeking via /sync - https://github.com/matrix-org/matrix-doc/pull/2753
// 'msc2836': Threading - https://github.com/matrix-org/matrix-doc/pull/2836
// 'msc2946': Spaces Summary - https://github.com/matrix-org/matrix-doc/pull/2946
MSCs []string `yaml:"mscs"`
Database DatabaseOptions `yaml:"database,omitempty"`

2
setup/monolith.go
View file

@ -61,7 +61,7 @@ func (m *Monolith) AddAllPublicRoutes(
processCtx *process.ProcessContext,
cfg *config.Dendrite,
routers httputil.Routers,
cm sqlutil.Connections,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
caches *caching.Caches,
enableMetrics bool,

2
setup/mscs/msc2836/msc2836.go
View file

@ -105,7 +105,7 @@ func toClientResponse(ctx context.Context, res *MSC2836EventRelationshipsRespons
// Enable this MSC
func Enable(
cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
cfg *config.Dendrite, cm *sqlutil.Connections, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
userAPI userapi.UserInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
) error {
db, err := NewDatabase(cm, &cfg.MSCs.Database)

6
setup/mscs/msc2836/storage.go
View file

@ -59,14 +59,14 @@ type DB struct {
}
// NewDatabase loads the database for msc2836
func NewDatabase(conMan sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
func NewDatabase(conMan *sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
if dbOpts.ConnectionString.IsPostgres() {
return newPostgresDatabase(conMan, dbOpts)
}
return newSQLiteDatabase(conMan, dbOpts)
}
func newPostgresDatabase(conMan sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
func newPostgresDatabase(conMan *sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
d := DB{}
var err error
if d.db, d.writer, err = conMan.Connection(dbOpts); err != nil {
@ -144,7 +144,7 @@ func newPostgresDatabase(conMan sqlutil.Connections, dbOpts *config.DatabaseOpti
return &d, err
}
func newSQLiteDatabase(conMan sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
func newSQLiteDatabase(conMan *sqlutil.Connections, dbOpts *config.DatabaseOptions) (Database, error) {
d := DB{}
var err error
if d.db, d.writer, err = conMan.Connection(dbOpts); err != nil {

744
setup/mscs/msc2946/msc2946.go
View file

@ -1,744 +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 msc2946 'Spaces Summary' implements https://github.com/matrix-org/matrix-doc/pull/2946
package msc2946
import (
"context"
"encoding/json"
"net/http"
"net/url"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/google/uuid"
"github.com/gorilla/mux"
fs "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/fclient"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/util"
"github.com/tidwall/gjson"
)
const (
ConstCreateEventContentKey = "type"
ConstCreateEventContentValueSpace = "m.space"
ConstSpaceChildEventType = "m.space.child"
ConstSpaceParentEventType = "m.space.parent"
)
type MSC2946ClientResponse struct {
Rooms []fclient.MSC2946Room `json:"rooms"`
NextBatch string `json:"next_batch,omitempty"`
}
// Enable this MSC
func Enable(
cfg *config.Dendrite, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
fsAPI fs.FederationInternalAPI, keyRing gomatrixserverlib.JSONVerifier, cache caching.SpaceSummaryRoomsCache,
) error {
clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, cache, cfg.Global.ServerName), httputil.WithAllowGuests())
routers.Client.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
routers.Client.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
fedAPI := httputil.MakeExternalAPI(
"msc2946_fed_spaces", func(req *http.Request) util.JSONResponse {
fedReq, errResp := fclient.VerifyHTTPRequest(
req, time.Now(), cfg.Global.ServerName, cfg.Global.IsLocalServerName, keyRing,
)
if fedReq == nil {
return errResp
}
// Extract the room ID from the request. Sanity check request data.
params, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.ErrorResponse(err)
}
roomID := params["roomID"]
return federatedSpacesHandler(req.Context(), fedReq, roomID, cache, rsAPI, fsAPI, cfg.Global.ServerName)
},
)
routers.Federation.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
routers.Federation.Handle("/v1/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
return nil
}
func federatedSpacesHandler(
ctx context.Context, fedReq *fclient.FederationRequest, roomID string,
cache caching.SpaceSummaryRoomsCache,
rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
thisServer spec.ServerName,
) util.JSONResponse {
u, err := url.Parse(fedReq.RequestURI())
if err != nil {
return util.JSONResponse{
Code: 400,
JSON: spec.InvalidParam("bad request uri"),
}
}
w := walker{
rootRoomID: roomID,
serverName: fedReq.Origin(),
thisServer: thisServer,
ctx: ctx,
cache: cache,
suggestedOnly: u.Query().Get("suggested_only") == "true",
limit: 1000,
// The main difference is that it does not recurse into spaces and does not support pagination.
// This is somewhat equivalent to a Client-Server request with a max_depth=1.
maxDepth: 1,
rsAPI: rsAPI,
fsAPI: fsAPI,
// inline cache as we don't have pagination in federation mode
paginationCache: make(map[string]paginationInfo),
}
return w.walk()
}
func spacesHandler(
rsAPI roomserver.RoomserverInternalAPI,
fsAPI fs.FederationInternalAPI,
cache caching.SpaceSummaryRoomsCache,
thisServer spec.ServerName,
) func(*http.Request, *userapi.Device) util.JSONResponse {
// declared outside the returned handler so it persists between calls
// TODO: clear based on... time?
paginationCache := make(map[string]paginationInfo)
return func(req *http.Request, device *userapi.Device) util.JSONResponse {
// Extract the room ID from the request. Sanity check request data.
params, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.ErrorResponse(err)
}
roomID := params["roomID"]
w := walker{
suggestedOnly: req.URL.Query().Get("suggested_only") == "true",
limit: parseInt(req.URL.Query().Get("limit"), 1000),
maxDepth: parseInt(req.URL.Query().Get("max_depth"), -1),
paginationToken: req.URL.Query().Get("from"),
rootRoomID: roomID,
caller: device,
thisServer: thisServer,
ctx: req.Context(),
cache: cache,
rsAPI: rsAPI,
fsAPI: fsAPI,
paginationCache: paginationCache,
}
return w.walk()
}
}
type paginationInfo struct {
processed set
unvisited []roomVisit
}
type walker struct {
rootRoomID string
caller *userapi.Device
serverName spec.ServerName
thisServer spec.ServerName
rsAPI roomserver.RoomserverInternalAPI
fsAPI fs.FederationInternalAPI
ctx context.Context
cache caching.SpaceSummaryRoomsCache
suggestedOnly bool
limit int
maxDepth int
paginationToken string
paginationCache map[string]paginationInfo
mu sync.Mutex
}
func (w *walker) newPaginationCache() (string, paginationInfo) {
p := paginationInfo{
processed: make(set),
unvisited: nil,
}
tok := uuid.NewString()
return tok, p
}
func (w *walker) loadPaginationCache(paginationToken string) *paginationInfo {
w.mu.Lock()
defer w.mu.Unlock()
p := w.paginationCache[paginationToken]
return &p
}
func (w *walker) storePaginationCache(paginationToken string, cache paginationInfo) {
w.mu.Lock()
defer w.mu.Unlock()
w.paginationCache[paginationToken] = cache
}
type roomVisit struct {
roomID string
parentRoomID string
depth int
vias []string // vias to query this room by
}
func (w *walker) walk() util.JSONResponse {
if authorised, _ := w.authorised(w.rootRoomID, ""); !authorised {
if w.caller != nil {
// CS API format
return util.JSONResponse{
Code: 403,
JSON: spec.Forbidden("room is unknown/forbidden"),
}
} else {
// SS API format
return util.JSONResponse{
Code: 404,
JSON: spec.NotFound("room is unknown/forbidden"),
}
}
}
var discoveredRooms []fclient.MSC2946Room
var cache *paginationInfo
if w.paginationToken != "" {
cache = w.loadPaginationCache(w.paginationToken)
if cache == nil {
return util.JSONResponse{
Code: 400,
JSON: spec.InvalidParam("invalid from"),
}
}
} else {
tok, c := w.newPaginationCache()
cache = &c
w.paginationToken = tok
// Begin walking the graph starting with the room ID in the request in a queue of unvisited rooms
c.unvisited = append(c.unvisited, roomVisit{
roomID: w.rootRoomID,
parentRoomID: "",
depth: 0,
})
}
processed := cache.processed
unvisited := cache.unvisited
// Depth first -> stack data structure
for len(unvisited) > 0 {
if len(discoveredRooms) >= w.limit {
break
}
// pop the stack
rv := unvisited[len(unvisited)-1]
unvisited = unvisited[:len(unvisited)-1]
// If this room has already been processed, skip.
// If this room exceeds the specified depth, skip.
if processed.isSet(rv.roomID) || rv.roomID == "" || (w.maxDepth > 0 && rv.depth > w.maxDepth) {
continue
}
// Mark this room as processed.
processed.set(rv.roomID)
// if this room is not a space room, skip.
var roomType string
create := w.stateEvent(rv.roomID, spec.MRoomCreate, "")
if create != nil {
// escape the `.`s so gjson doesn't think it's nested
roomType = gjson.GetBytes(create.Content(), strings.ReplaceAll(ConstCreateEventContentKey, ".", `\.`)).Str
}
// Collect rooms/events to send back (either locally or fetched via federation)
var discoveredChildEvents []fclient.MSC2946StrippedEvent
// If we know about this room and the caller is authorised (joined/world_readable) then pull
// events locally
roomExists := w.roomExists(rv.roomID)
if !roomExists {
// attempt to query this room over federation, as either we've never heard of it before
// or we've left it and hence are not authorised (but info may be exposed regardless)
fedRes := w.federatedRoomInfo(rv.roomID, rv.vias)
if fedRes != nil {
discoveredChildEvents = fedRes.Room.ChildrenState
discoveredRooms = append(discoveredRooms, fedRes.Room)
if len(fedRes.Children) > 0 {
discoveredRooms = append(discoveredRooms, fedRes.Children...)
}
// mark this room as a space room as the federated server responded.
// we need to do this so we add the children of this room to the unvisited stack
// as these children may be rooms we do know about.
roomType = ConstCreateEventContentValueSpace
}
} else if authorised, isJoinedOrInvited := w.authorised(rv.roomID, rv.parentRoomID); authorised {
// Get all `m.space.child` state events for this room
events, err := w.childReferences(rv.roomID)
if err != nil {
util.GetLogger(w.ctx).WithError(err).WithField("room_id", rv.roomID).Error("failed to extract references for room")
continue
}
discoveredChildEvents = events
pubRoom := w.publicRoomsChunk(rv.roomID)
discoveredRooms = append(discoveredRooms, fclient.MSC2946Room{
PublicRoom: *pubRoom,
RoomType: roomType,
ChildrenState: events,
})
// don't walk children if the user is not joined/invited to the space
if !isJoinedOrInvited {
continue
}
} else {
// room exists but user is not authorised
continue
}
// don't walk the children
// if the parent is not a space room
if roomType != ConstCreateEventContentValueSpace {
continue
}
// For each referenced room ID in the child events being returned to the caller
// add the room ID to the queue of unvisited rooms. Loop from the beginning.
// We need to invert the order here because the child events are lo->hi on the timestamp,
// so we need to ensure we pop in the same lo->hi order, which won't be the case if we
// insert the highest timestamp last in a stack.
for i := len(discoveredChildEvents) - 1; i >= 0; i-- {
spaceContent := struct {
Via []string `json:"via"`
}{}
ev := discoveredChildEvents[i]
_ = json.Unmarshal(ev.Content, &spaceContent)
unvisited = append(unvisited, roomVisit{
roomID: ev.StateKey,
parentRoomID: rv.roomID,
depth: rv.depth + 1,
vias: spaceContent.Via,
})
}
}
if len(unvisited) > 0 {
// we still have more rooms so we need to send back a pagination token,
// we probably hit a room limit
cache.processed = processed
cache.unvisited = unvisited
w.storePaginationCache(w.paginationToken, *cache)
} else {
// clear the pagination token so we don't send it back to the client
// Note we do NOT nuke the cache just in case this response is lost
// and the client retries it.
w.paginationToken = ""
}
if w.caller != nil {
// return CS API format
return util.JSONResponse{
Code: 200,
JSON: MSC2946ClientResponse{
Rooms: discoveredRooms,
NextBatch: w.paginationToken,
},
}
}
// return SS API format
// the first discovered room will be the room asked for, and subsequent ones the depth=1 children
if len(discoveredRooms) == 0 {
return util.JSONResponse{
Code: 404,
JSON: spec.NotFound("room is unknown/forbidden"),
}
}
return util.JSONResponse{
Code: 200,
JSON: fclient.MSC2946SpacesResponse{
Room: discoveredRooms[0],
Children: discoveredRooms[1:],
},
}
}
func (w *walker) stateEvent(roomID, evType, stateKey string) *types.HeaderedEvent {
var queryRes roomserver.QueryCurrentStateResponse
tuple := gomatrixserverlib.StateKeyTuple{
EventType: evType,
StateKey: stateKey,
}
err := w.rsAPI.QueryCurrentState(w.ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID,
StateTuples: []gomatrixserverlib.StateKeyTuple{tuple},
}, &queryRes)
if err != nil {
return nil
}
return queryRes.StateEvents[tuple]
}
func (w *walker) publicRoomsChunk(roomID string) *fclient.PublicRoom {
pubRooms, err := roomserver.PopulatePublicRooms(w.ctx, []string{roomID}, w.rsAPI)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Error("failed to PopulatePublicRooms")
return nil
}
if len(pubRooms) == 0 {
return nil
}
return &pubRooms[0]
}
// federatedRoomInfo returns more of the spaces graph from another server. Returns nil if this was
// unsuccessful.
func (w *walker) federatedRoomInfo(roomID string, vias []string) *fclient.MSC2946SpacesResponse {
// only do federated requests for client requests
if w.caller == nil {
return nil
}
resp, ok := w.cache.GetSpaceSummary(roomID)
if ok {
util.GetLogger(w.ctx).Debugf("Returning cached response for %s", roomID)
return &resp
}
util.GetLogger(w.ctx).Debugf("Querying %s via %+v", roomID, vias)
ctx := context.Background()
// query more of the spaces graph using these servers
for _, serverName := range vias {
if serverName == string(w.thisServer) {
continue
}
res, err := w.fsAPI.MSC2946Spaces(ctx, w.thisServer, spec.ServerName(serverName), roomID, w.suggestedOnly)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Warnf("failed to call MSC2946Spaces on server %s", serverName)
continue
}
// ensure nil slices are empty as we send this to the client sometimes
if res.Room.ChildrenState == nil {
res.Room.ChildrenState = []fclient.MSC2946StrippedEvent{}
}
for i := 0; i < len(res.Children); i++ {
child := res.Children[i]
if child.ChildrenState == nil {
child.ChildrenState = []fclient.MSC2946StrippedEvent{}
}
res.Children[i] = child
}
w.cache.StoreSpaceSummary(roomID, res)
return &res
}
return nil
}
func (w *walker) roomExists(roomID string) bool {
var queryRes roomserver.QueryServerJoinedToRoomResponse
err := w.rsAPI.QueryServerJoinedToRoom(w.ctx, &roomserver.QueryServerJoinedToRoomRequest{
RoomID: roomID,
ServerName: w.thisServer,
}, &queryRes)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Error("failed to QueryServerJoinedToRoom")
return false
}
// if the room exists but we aren't in the room then we might have stale data so we want to fetch
// it fresh via federation
return queryRes.RoomExists && queryRes.IsInRoom
}
// authorised returns true iff the user is joined this room or the room is world_readable
func (w *walker) authorised(roomID, parentRoomID string) (authed, isJoinedOrInvited bool) {
if w.caller != nil {
return w.authorisedUser(roomID, parentRoomID)
}
return w.authorisedServer(roomID), false
}
// authorisedServer returns true iff the server is joined this room or the room is world_readable, public, or knockable
func (w *walker) authorisedServer(roomID string) bool {
// Check history visibility / join rules first
hisVisTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomHistoryVisibility,
StateKey: "",
}
joinRuleTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomJoinRules,
StateKey: "",
}
var queryRoomRes roomserver.QueryCurrentStateResponse
err := w.rsAPI.QueryCurrentState(w.ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID,
StateTuples: []gomatrixserverlib.StateKeyTuple{
hisVisTuple, joinRuleTuple,
},
}, &queryRoomRes)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Error("failed to QueryCurrentState")
return false
}
hisVisEv := queryRoomRes.StateEvents[hisVisTuple]
if hisVisEv != nil {
hisVis, _ := hisVisEv.HistoryVisibility()
if hisVis == "world_readable" {
return true
}
}
// check if this room is a restricted room and if so, we need to check if the server is joined to an allowed room ID
// in addition to the actual room ID (but always do the actual one first as it's quicker in the common case)
allowJoinedToRoomIDs := []string{roomID}
joinRuleEv := queryRoomRes.StateEvents[joinRuleTuple]
if joinRuleEv != nil {
rule, ruleErr := joinRuleEv.JoinRule()
if ruleErr != nil {
util.GetLogger(w.ctx).WithError(ruleErr).WithField("parent_room_id", roomID).Warn("failed to get join rule")
return false
}
if rule == spec.Public || rule == spec.Knock {
return true
}
if rule == spec.Restricted {
allowJoinedToRoomIDs = append(allowJoinedToRoomIDs, w.restrictedJoinRuleAllowedRooms(joinRuleEv, "m.room_membership")...)
}
}
// check if server is joined to any allowed room
for _, allowedRoomID := range allowJoinedToRoomIDs {
var queryRes fs.QueryJoinedHostServerNamesInRoomResponse
err = w.fsAPI.QueryJoinedHostServerNamesInRoom(w.ctx, &fs.QueryJoinedHostServerNamesInRoomRequest{
RoomID: allowedRoomID,
}, &queryRes)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Error("failed to QueryJoinedHostServerNamesInRoom")
continue
}
for _, srv := range queryRes.ServerNames {
if srv == w.serverName {
return true
}
}
}
return false
}
// authorisedUser returns true iff the user is invited/joined this room or the room is world_readable
// or if the room has a public or knock join rule.
// Failing that, if the room has a restricted join rule and belongs to the space parent listed, it will return true.
func (w *walker) authorisedUser(roomID, parentRoomID string) (authed bool, isJoinedOrInvited bool) {
hisVisTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomHistoryVisibility,
StateKey: "",
}
joinRuleTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomJoinRules,
StateKey: "",
}
roomMemberTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomMember,
StateKey: w.caller.UserID,
}
var queryRes roomserver.QueryCurrentStateResponse
err := w.rsAPI.QueryCurrentState(w.ctx, &roomserver.QueryCurrentStateRequest{
RoomID: roomID,
StateTuples: []gomatrixserverlib.StateKeyTuple{
hisVisTuple, joinRuleTuple, roomMemberTuple,
},
}, &queryRes)
if err != nil {
util.GetLogger(w.ctx).WithError(err).Error("failed to QueryCurrentState")
return false, false
}
memberEv := queryRes.StateEvents[roomMemberTuple]
if memberEv != nil {
membership, _ := memberEv.Membership()
if membership == spec.Join || membership == spec.Invite {
return true, true
}
}
hisVisEv := queryRes.StateEvents[hisVisTuple]
if hisVisEv != nil {
hisVis, _ := hisVisEv.HistoryVisibility()
if hisVis == "world_readable" {
return true, false
}
}
joinRuleEv := queryRes.StateEvents[joinRuleTuple]
if parentRoomID != "" && joinRuleEv != nil {
var allowed bool
rule, ruleErr := joinRuleEv.JoinRule()
if ruleErr != nil {
util.GetLogger(w.ctx).WithError(ruleErr).WithField("parent_room_id", parentRoomID).Warn("failed to get join rule")
} else if rule == spec.Public || rule == spec.Knock {
allowed = true
} else if rule == spec.Restricted {
allowedRoomIDs := w.restrictedJoinRuleAllowedRooms(joinRuleEv, "m.room_membership")
// check parent is in the allowed set
for _, a := range allowedRoomIDs {
if parentRoomID == a {
allowed = true
break
}
}
}
if allowed {
// ensure caller is joined to the parent room
var queryRes2 roomserver.QueryCurrentStateResponse
err = w.rsAPI.QueryCurrentState(w.ctx, &roomserver.QueryCurrentStateRequest{
RoomID: parentRoomID,
StateTuples: []gomatrixserverlib.StateKeyTuple{
roomMemberTuple,
},
}, &queryRes2)
if err != nil {
util.GetLogger(w.ctx).WithError(err).WithField("parent_room_id", parentRoomID).Warn("failed to check user is joined to parent room")
} else {
memberEv = queryRes2.StateEvents[roomMemberTuple]
if memberEv != nil {
membership, _ := memberEv.Membership()
if membership == spec.Join {
return true, false
}
}
}
}
}
return false, false
}
func (w *walker) restrictedJoinRuleAllowedRooms(joinRuleEv *types.HeaderedEvent, allowType string) (allows []string) {
rule, _ := joinRuleEv.JoinRule()
if rule != spec.Restricted {
return nil
}
var jrContent gomatrixserverlib.JoinRuleContent
if err := json.Unmarshal(joinRuleEv.Content(), &jrContent); err != nil {
util.GetLogger(w.ctx).Warnf("failed to check join_rule on room %s: %s", joinRuleEv.RoomID(), err)
return nil
}
for _, allow := range jrContent.Allow {
if allow.Type == allowType {
allows = append(allows, allow.RoomID)
}
}
return
}
// references returns all child references pointing to or from this room.
func (w *walker) childReferences(roomID string) ([]fclient.MSC2946StrippedEvent, error) {
createTuple := gomatrixserverlib.StateKeyTuple{
EventType: spec.MRoomCreate,
StateKey: "",
}
var res roomserver.QueryCurrentStateResponse
err := w.rsAPI.QueryCurrentState(context.Background(), &roomserver.QueryCurrentStateRequest{
RoomID: roomID,
AllowWildcards: true,
StateTuples: []gomatrixserverlib.StateKeyTuple{
createTuple, {
EventType: ConstSpaceChildEventType,
StateKey: "*",
},
},
}, &res)
if err != nil {
return nil, err
}
// don't return any child refs if the room is not a space room
if res.StateEvents[createTuple] != nil {
// escape the `.`s so gjson doesn't think it's nested
roomType := gjson.GetBytes(res.StateEvents[createTuple].Content(), strings.ReplaceAll(ConstCreateEventContentKey, ".", `\.`)).Str
if roomType != ConstCreateEventContentValueSpace {
return []fclient.MSC2946StrippedEvent{}, nil
}
}
delete(res.StateEvents, createTuple)
el := make([]fclient.MSC2946StrippedEvent, 0, len(res.StateEvents))
for _, ev := range res.StateEvents {
content := gjson.ParseBytes(ev.Content())
// only return events that have a `via` key as per MSC1772
// else we'll incorrectly walk redacted events (as the link
// is in the state_key)
if content.Get("via").Exists() {
strip := stripped(ev.PDU)
if strip == nil {
continue
}
// if suggested only and this child isn't suggested, skip it.
// if suggested only = false we include everything so don't need to check the content.
if w.suggestedOnly && !content.Get("suggested").Bool() {
continue
}
el = append(el, *strip)
}
}
// sort by origin_server_ts as per MSC2946
sort.Slice(el, func(i, j int) bool {
return el[i].OriginServerTS < el[j].OriginServerTS
})
return el, nil
}
type set map[string]struct{}
func (s set) set(val string) {
s[val] = struct{}{}
}
func (s set) isSet(val string) bool {
_, ok := s[val]
return ok
}
func stripped(ev gomatrixserverlib.PDU) *fclient.MSC2946StrippedEvent {
if ev.StateKey() == nil {
return nil
}
return &fclient.MSC2946StrippedEvent{
Type: ev.Type(),
StateKey: *ev.StateKey(),
Content: ev.Content(),
Sender: string(ev.SenderID()),
OriginServerTS: ev.OriginServerTS(),
}
}
func parseInt(intstr string, defaultVal int) int {
i, err := strconv.ParseInt(intstr, 10, 32)
if err != nil {
return defaultVal
}
return int(i)
}

7
setup/mscs/mscs.go
View file

@ -25,12 +25,11 @@ import (
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs/msc2836"
"github.com/matrix-org/dendrite/setup/mscs/msc2946"
"github.com/matrix-org/util"
)
// Enable MSCs - returns an error on unknown MSCs
func Enable(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, caches *caching.Caches) error {
func Enable(cfg *config.Dendrite, cm *sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, caches *caching.Caches) error {
for _, msc := range cfg.MSCs.MSCs {
util.GetLogger(context.Background()).WithField("msc", msc).Info("Enabling MSC")
if err := EnableMSC(cfg, cm, routers, monolith, msc, caches); err != nil {
@ -40,12 +39,10 @@ func Enable(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Route
return nil
}
func EnableMSC(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
func EnableMSC(cfg *config.Dendrite, cm *sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
switch msc {
case "msc2836":
return msc2836.Enable(cfg, cm, routers, monolith.RoomserverAPI, monolith.FederationAPI, monolith.UserAPI, monolith.KeyRing)
case "msc2946":
return msc2946.Enable(cfg, routers, monolith.RoomserverAPI, monolith.UserAPI, monolith.FederationAPI, monolith.KeyRing, caches)
case "msc2444": // enabled inside federationapi
case "msc2753": // enabled inside clientapi
default:

33
syncapi/consumers/roomserver.go
View file

@ -17,18 +17,12 @@ package consumers
import (
"context"
"database/sql"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"github.com/getsentry/sentry-go"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/nats-io/nats.go"
"github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
"github.com/tidwall/gjson"
"github.com/tidwall/sjson"
"github.com/matrix-org/dendrite/internal/fulltext"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/api"
@ -40,6 +34,13 @@ import (
"github.com/matrix-org/dendrite/syncapi/storage"
"github.com/matrix-org/dendrite/syncapi/streams"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/nats-io/nats.go"
"github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
"github.com/tidwall/gjson"
"github.com/tidwall/sjson"
)
// OutputRoomEventConsumer consumes events that originated in the room server.
@ -143,7 +144,14 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Ms
)
}
if err != nil {
log.WithError(err).Error("roomserver output log: failed to process event")
if errors.As(err, new(base64.CorruptInputError)) {
// no matter how often we retry this event, we will always get this error, discard the event
return true
}
log.WithFields(log.Fields{
"type": output.Type,
}).WithError(err).Error("roomserver output log: failed to process event")
sentry.CaptureException(err)
return false
}
@ -239,21 +247,18 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
ev, err := s.updateStateEvent(ev)
if err != nil {
sentry.CaptureException(err)
return err
}
for i := range addsStateEvents {
addsStateEvents[i], err = s.updateStateEvent(addsStateEvents[i])
if err != nil {
sentry.CaptureException(err)
return err
}
}
if msg.RewritesState {
if err = s.db.PurgeRoomState(ctx, ev.RoomID()); err != nil {
sentry.CaptureException(err)
return fmt.Errorf("s.db.PurgeRoom: %w", err)
}
}
@ -291,7 +296,6 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
if pduPos, err = s.notifyJoinedPeeks(ctx, ev, pduPos); err != nil {
log.WithError(err).Errorf("Failed to notifyJoinedPeeks for PDU pos %d", pduPos)
sentry.CaptureException(err)
return err
}
@ -432,7 +436,6 @@ func (s *OutputRoomEventConsumer) onNewInviteEvent(
pduPos, err := s.db.AddInviteEvent(ctx, msg.Event)
if err != nil {
sentry.CaptureException(err)
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event_id": msg.Event.EventID(),
@ -454,7 +457,6 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent(
// It's possible we just haven't heard of this invite yet, so
// we should not panic if we try to retire it.
if err != nil && err != sql.ErrNoRows {
sentry.CaptureException(err)
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event_id": msg.EventID,
@ -498,7 +500,6 @@ func (s *OutputRoomEventConsumer) onNewPeek(
) {
sp, err := s.db.AddPeek(ctx, msg.RoomID, msg.UserID, msg.DeviceID)
if err != nil {
sentry.CaptureException(err)
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
log.ErrorKey: err,

2
syncapi/storage/postgres/syncserver.go
View file

@ -36,7 +36,7 @@ type SyncServerDatasource struct {
}
// NewDatabase creates a new sync server database
func NewDatabase(ctx context.Context, cm sqlutil.Connections, dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
func NewDatabase(ctx context.Context, cm *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
var d SyncServerDatasource
var err error
if d.db, d.writer, err = cm.Connection(dbProperties); err != nil {

2
syncapi/storage/sqlite3/syncserver.go
View file

@ -36,7 +36,7 @@ type SyncServerDatasource struct {
// NewDatabase creates a new sync server database
// nolint: gocyclo
func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
func NewDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
var d SyncServerDatasource
var err error

2
syncapi/storage/storage.go
View file

@ -28,7 +28,7 @@ import (
)
// NewSyncServerDatasource opens a database connection.
func NewSyncServerDatasource(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (Database, error) {
func NewSyncServerDatasource(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewDatabase(ctx, conMan, dbProperties)

2
syncapi/syncapi.go
View file

@ -45,7 +45,7 @@ func AddPublicRoutes(
processContext *process.ProcessContext,
routers httputil.Routers,
dendriteCfg *config.Dendrite,
cm sqlutil.Connections,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
userAPI userapi.SyncUserAPI,
rsAPI api.SyncRoomserverAPI,

4
userapi/storage/postgres/storage.go
View file

@ -31,7 +31,7 @@ import (
)
// NewDatabase creates a new accounts and profiles database
func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, serverName spec.ServerName, bcryptCost int, openIDTokenLifetimeMS int64, loginTokenLifetime time.Duration, serverNoticesLocalpart string) (*shared.Database, error) {
func NewDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, serverName spec.ServerName, bcryptCost int, openIDTokenLifetimeMS int64, loginTokenLifetime time.Duration, serverNoticesLocalpart string) (*shared.Database, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err
@ -140,7 +140,7 @@ func NewDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *
}, nil
}
func NewKeyDatabase(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.KeyDatabase, error) {
func NewKeyDatabase(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.KeyDatabase, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err

4
userapi/storage/sqlite3/storage.go
View file

@ -29,7 +29,7 @@ import (
)
// NewUserDatabase creates a new accounts and profiles database
func NewUserDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperties *config.DatabaseOptions, serverName spec.ServerName, bcryptCost int, openIDTokenLifetimeMS int64, loginTokenLifetime time.Duration, serverNoticesLocalpart string) (*shared.Database, error) {
func NewUserDatabase(ctx context.Context, conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions, serverName spec.ServerName, bcryptCost int, openIDTokenLifetimeMS int64, loginTokenLifetime time.Duration, serverNoticesLocalpart string) (*shared.Database, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err
@ -137,7 +137,7 @@ func NewUserDatabase(ctx context.Context, conMan sqlutil.Connections, dbProperti
}, nil
}
func NewKeyDatabase(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.KeyDatabase, error) {
func NewKeyDatabase(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (*shared.KeyDatabase, error) {
db, writer, err := conMan.Connection(dbProperties)
if err != nil {
return nil, err

4
userapi/storage/storage.go
View file

@ -34,7 +34,7 @@ import (
// and sets postgres connection parameters
func NewUserDatabase(
ctx context.Context,
conMan sqlutil.Connections,
conMan *sqlutil.Connections,
dbProperties *config.DatabaseOptions,
serverName spec.ServerName,
bcryptCost int,
@ -54,7 +54,7 @@ func NewUserDatabase(
// NewKeyDatabase opens a new Postgres or Sqlite database (base on dataSourceName) scheme)
// and sets postgres connection parameters.
func NewKeyDatabase(conMan sqlutil.Connections, dbProperties *config.DatabaseOptions) (KeyDatabase, error) {
func NewKeyDatabase(conMan *sqlutil.Connections, dbProperties *config.DatabaseOptions) (KeyDatabase, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewKeyDatabase(conMan, dbProperties)

5
userapi/userapi.go
View file

@ -36,10 +36,13 @@ import (
// NewInternalAPI returns a concrete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
//
// Creating a new instance of the user API requires a roomserver API with a federation API set
// using its `SetFederationAPI` method, other you may get nil-dereference errors.
func NewInternalAPI(
processContext *process.ProcessContext,
dendriteCfg *config.Dendrite,
cm sqlutil.Connections,
cm *sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
rsAPI rsapi.UserRoomserverAPI,
fedClient fedsenderapi.KeyserverFederationAPI,