mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-23 14:53:10 -06:00
Merge branch 'master' into reject-invalid-utf8
This commit is contained in:
commit
e7d399b613
|
|
@ -253,12 +253,12 @@ room_server:
|
|||
conn_max_lifetime: -1
|
||||
|
||||
# Configuration for the Server Key API (for server signing keys).
|
||||
server_key_api:
|
||||
signing_key_server:
|
||||
internal_api:
|
||||
listen: http://0.0.0.0:7780
|
||||
connect: http://server_key_api:7780
|
||||
connect: http://signing_key_server:7780
|
||||
database:
|
||||
connection_string: postgresql://dendrite:itsasecret@postgres/dendrite_serverkey?sslmode=disable
|
||||
connection_string: postgresql://dendrite:itsasecret@postgres/dendrite_signingkeyserver?sslmode=disable
|
||||
max_open_conns: 100
|
||||
max_idle_conns: 2
|
||||
conn_max_lifetime: -1
|
||||
|
|
|
|||
|
|
@ -128,9 +128,9 @@ services:
|
|||
networks:
|
||||
- internal
|
||||
|
||||
server_key_api:
|
||||
hostname: server_key_api
|
||||
image: matrixdotorg/dendrite:serverkeyapi
|
||||
signing_key_server:
|
||||
hostname: signing_key_server
|
||||
image: matrixdotorg/dendrite:signingkeyserver
|
||||
command: [
|
||||
"--config=dendrite.yaml"
|
||||
]
|
||||
|
|
|
|||
|
|
@ -17,5 +17,5 @@ docker build -t matrixdotorg/dendrite:keyserver --build-arg component=de
|
|||
docker build -t matrixdotorg/dendrite:mediaapi --build-arg component=dendrite-media-api-server -f build/docker/Dockerfile.component .
|
||||
docker build -t matrixdotorg/dendrite:roomserver --build-arg component=dendrite-room-server -f build/docker/Dockerfile.component .
|
||||
docker build -t matrixdotorg/dendrite:syncapi --build-arg component=dendrite-sync-api-server -f build/docker/Dockerfile.component .
|
||||
docker build -t matrixdotorg/dendrite:serverkeyapi --build-arg component=dendrite-server-key-api-server -f build/docker/Dockerfile.component .
|
||||
docker build -t matrixdotorg/dendrite:signingkeyserver --build-arg component=dendrite-signing-key-server -f build/docker/Dockerfile.component .
|
||||
docker build -t matrixdotorg/dendrite:userapi --build-arg component=dendrite-user-api-server -f build/docker/Dockerfile.component .
|
||||
|
|
|
|||
|
|
@ -13,4 +13,5 @@ docker pull matrixdotorg/dendrite:keyserver
|
|||
docker pull matrixdotorg/dendrite:mediaapi
|
||||
docker pull matrixdotorg/dendrite:roomserver
|
||||
docker pull matrixdotorg/dendrite:syncapi
|
||||
docker pull matrixdotorg/dendrite:signingkeyserver
|
||||
docker pull matrixdotorg/dendrite:userapi
|
||||
|
|
|
|||
|
|
@ -13,5 +13,5 @@ docker push matrixdotorg/dendrite:keyserver
|
|||
docker push matrixdotorg/dendrite:mediaapi
|
||||
docker push matrixdotorg/dendrite:roomserver
|
||||
docker push matrixdotorg/dendrite:syncapi
|
||||
docker push matrixdotorg/dendrite:serverkeyapi
|
||||
docker push matrixdotorg/dendrite:signingkeyserver
|
||||
docker push matrixdotorg/dendrite:userapi
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
for db in account device mediaapi syncapi roomserver serverkey keyserver federationsender appservice e2ekey naffka; do
|
||||
for db in account device mediaapi syncapi roomserver signingkeyserver keyserver federationsender appservice e2ekey naffka; do
|
||||
createdb -U dendrite -O dendrite dendrite_$db
|
||||
done
|
||||
|
|
|
|||
|
|
@ -94,7 +94,7 @@ func (m *DendriteMonolith) Start() {
|
|||
cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-mediaapi.db", m.StorageDirectory))
|
||||
cfg.SyncAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-syncapi.db", m.StorageDirectory))
|
||||
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-roomserver.db", m.StorageDirectory))
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-serverkey.db", m.StorageDirectory))
|
||||
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-signingkeyserver.db", m.StorageDirectory))
|
||||
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-keyserver.db", m.StorageDirectory))
|
||||
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-federationsender.db", m.StorageDirectory))
|
||||
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-appservice.db", m.StorageDirectory))
|
||||
|
|
|
|||
|
|
@ -16,9 +16,11 @@ package routing
|
|||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
|
||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/userapi/api"
|
||||
"github.com/matrix-org/dendrite/userapi/storage/accounts"
|
||||
|
|
@ -74,16 +76,32 @@ func JoinRoomByIDOrAlias(
|
|||
}
|
||||
|
||||
// Ask the roomserver to perform the join.
|
||||
rsAPI.PerformJoin(req.Context(), &joinReq, &joinRes)
|
||||
if joinRes.Error != nil {
|
||||
return joinRes.Error.JSONResponse()
|
||||
}
|
||||
done := make(chan util.JSONResponse, 1)
|
||||
go func() {
|
||||
defer close(done)
|
||||
rsAPI.PerformJoin(req.Context(), &joinReq, &joinRes)
|
||||
if joinRes.Error != nil {
|
||||
done <- joinRes.Error.JSONResponse()
|
||||
} else {
|
||||
done <- util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
// TODO: Put the response struct somewhere internal.
|
||||
JSON: struct {
|
||||
RoomID string `json:"room_id"`
|
||||
}{joinRes.RoomID},
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
// TODO: Put the response struct somewhere internal.
|
||||
JSON: struct {
|
||||
RoomID string `json:"room_id"`
|
||||
}{joinRes.RoomID},
|
||||
// Wait either for the join to finish, or for us to hit a reasonable
|
||||
// timeout, at which point we'll just return a 200 to placate clients.
|
||||
select {
|
||||
case <-time.After(time.Second * 20):
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusAccepted,
|
||||
JSON: jsonerror.Unknown("The room join will continue in the background."),
|
||||
}
|
||||
case result := <-done:
|
||||
return result
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,8 +31,8 @@ func main() {
|
|||
appservice.AddInternalRoutes(base.InternalAPIMux, intAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.AppServiceAPI.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.AppServiceAPI.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ import (
|
|||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/keyserver"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver"
|
||||
"github.com/matrix-org/dendrite/userapi"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
|
|
@ -125,7 +125,7 @@ func main() {
|
|||
cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName))
|
||||
cfg.SyncAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-syncapi.db", *instanceName))
|
||||
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-serverkey.db", *instanceName))
|
||||
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-signingkeyserver.db", *instanceName))
|
||||
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
|
||||
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
|
||||
cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
|
||||
|
|
@ -143,8 +143,8 @@ func main() {
|
|||
userAPI := userapi.NewInternalAPI(accountDB, &cfg.UserAPI, nil, keyAPI)
|
||||
keyAPI.SetUserAPI(userAPI)
|
||||
|
||||
serverKeyAPI := serverkeyapi.NewInternalAPI(
|
||||
&base.Base.Cfg.ServerKeyAPI, federation, base.Base.Caches,
|
||||
serverKeyAPI := signingkeyserver.NewInternalAPI(
|
||||
&base.Base.Cfg.SigningKeyServer, federation, base.Base.Caches,
|
||||
)
|
||||
keyRing := serverKeyAPI.KeyRing()
|
||||
createKeyDB(
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ func main() {
|
|||
cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName))
|
||||
cfg.SyncAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-syncapi.db", *instanceName))
|
||||
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-serverkey.db", *instanceName))
|
||||
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-signingkeyserver.db", *instanceName))
|
||||
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", *instanceName))
|
||||
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
|
||||
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
|
||||
|
|
|
|||
|
|
@ -205,13 +205,11 @@ func (n *Node) SessionCount() int {
|
|||
|
||||
func (n *Node) KnownNodes() []gomatrixserverlib.ServerName {
|
||||
nodemap := map[string]struct{}{
|
||||
"b5ae50589e50991dd9dd7d59c5c5f7a4521e8da5b603b7f57076272abc58b374": {},
|
||||
//"b5ae50589e50991dd9dd7d59c5c5f7a4521e8da5b603b7f57076272abc58b374": {},
|
||||
}
|
||||
for _, peer := range n.core.GetSwitchPeers() {
|
||||
nodemap[hex.EncodeToString(peer.SigPublicKey[:])] = struct{}{}
|
||||
}
|
||||
/*
|
||||
for _, peer := range n.core.GetSwitchPeers() {
|
||||
nodemap[hex.EncodeToString(peer.PublicKey[:])] = struct{}{}
|
||||
}
|
||||
*/
|
||||
n.sessions.Range(func(_, v interface{}) bool {
|
||||
session, ok := v.(quic.Session)
|
||||
if !ok {
|
||||
|
|
|
|||
|
|
@ -34,8 +34,8 @@ func main() {
|
|||
eduserver.AddInternalRoutes(base.InternalAPIMux, intAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.EDUServer.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.EDUServer.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ func main() {
|
|||
|
||||
userAPI := base.UserAPIClient()
|
||||
federation := base.CreateFederationClient()
|
||||
serverKeyAPI := base.ServerKeyAPIClient()
|
||||
serverKeyAPI := base.SigningKeyServerHTTPClient()
|
||||
keyRing := serverKeyAPI.KeyRing()
|
||||
fsAPI := base.FederationSenderHTTPClient()
|
||||
rsAPI := base.RoomserverHTTPClient()
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ func main() {
|
|||
|
||||
federation := base.CreateFederationClient()
|
||||
|
||||
serverKeyAPI := base.ServerKeyAPIClient()
|
||||
serverKeyAPI := base.SigningKeyServerHTTPClient()
|
||||
keyRing := serverKeyAPI.KeyRing()
|
||||
|
||||
rsAPI := base.RoomserverHTTPClient()
|
||||
|
|
@ -36,8 +36,8 @@ func main() {
|
|||
federationsender.AddInternalRoutes(base.InternalAPIMux, fsAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.FederationSender.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.FederationSender.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,8 +30,8 @@ func main() {
|
|||
keyserver.AddInternalRoutes(base.InternalAPIMux, intAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.KeyServer.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.KeyServer.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,13 +27,15 @@ import (
|
|||
"github.com/matrix-org/dendrite/keyserver"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver"
|
||||
"github.com/matrix-org/dendrite/userapi"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var (
|
||||
httpBindAddr = flag.String("http-bind-address", ":8008", "The HTTP listening port for the server")
|
||||
httpsBindAddr = flag.String("https-bind-address", ":8448", "The HTTPS listening port for the server")
|
||||
apiBindAddr = flag.String("api-bind-address", "localhost:18008", "The HTTP listening port for the internal HTTP APIs (if -api is enabled)")
|
||||
certFile = flag.String("tls-cert", "", "The PEM formatted X509 certificate to use for TLS")
|
||||
keyFile = flag.String("tls-key", "", "The PEM private key to use for TLS")
|
||||
enableHTTPAPIs = flag.Bool("api", false, "Use HTTP APIs instead of short-circuiting (warning: exposes API endpoints!)")
|
||||
|
|
@ -44,22 +46,25 @@ func main() {
|
|||
cfg := setup.ParseFlags(true)
|
||||
httpAddr := config.HTTPAddress("http://" + *httpBindAddr)
|
||||
httpsAddr := config.HTTPAddress("https://" + *httpsBindAddr)
|
||||
httpAPIAddr := httpAddr
|
||||
|
||||
if *enableHTTPAPIs {
|
||||
logrus.Warnf("DANGER! The -api option is enabled, exposing internal APIs on %q!", *apiBindAddr)
|
||||
httpAPIAddr = config.HTTPAddress("http://" + *apiBindAddr)
|
||||
// If the HTTP APIs are enabled then we need to update the Listen
|
||||
// statements in the configuration so that we know where to find
|
||||
// the API endpoints. They'll listen on the same port as the monolith
|
||||
// itself.
|
||||
cfg.AppServiceAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.ClientAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.EDUServer.InternalAPI.Connect = httpAddr
|
||||
cfg.FederationAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.FederationSender.InternalAPI.Connect = httpAddr
|
||||
cfg.KeyServer.InternalAPI.Connect = httpAddr
|
||||
cfg.MediaAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.RoomServer.InternalAPI.Connect = httpAddr
|
||||
cfg.ServerKeyAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.SyncAPI.InternalAPI.Connect = httpAddr
|
||||
cfg.AppServiceAPI.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.ClientAPI.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.EDUServer.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.FederationAPI.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.FederationSender.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.KeyServer.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.MediaAPI.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.RoomServer.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.SigningKeyServer.InternalAPI.Connect = httpAPIAddr
|
||||
cfg.SyncAPI.InternalAPI.Connect = httpAPIAddr
|
||||
}
|
||||
|
||||
base := setup.NewBaseDendrite(cfg, "Monolith", *enableHTTPAPIs)
|
||||
|
|
@ -68,14 +73,14 @@ func main() {
|
|||
accountDB := base.CreateAccountsDB()
|
||||
federation := base.CreateFederationClient()
|
||||
|
||||
serverKeyAPI := serverkeyapi.NewInternalAPI(
|
||||
&base.Cfg.ServerKeyAPI, federation, base.Caches,
|
||||
skAPI := signingkeyserver.NewInternalAPI(
|
||||
&base.Cfg.SigningKeyServer, federation, base.Caches,
|
||||
)
|
||||
if base.UseHTTPAPIs {
|
||||
serverkeyapi.AddInternalRoutes(base.InternalAPIMux, serverKeyAPI, base.Caches)
|
||||
serverKeyAPI = base.ServerKeyAPIClient()
|
||||
signingkeyserver.AddInternalRoutes(base.InternalAPIMux, skAPI, base.Caches)
|
||||
skAPI = base.SigningKeyServerHTTPClient()
|
||||
}
|
||||
keyRing := serverKeyAPI.KeyRing()
|
||||
keyRing := skAPI.KeyRing()
|
||||
|
||||
rsImpl := roomserver.NewInternalAPI(
|
||||
base, keyRing,
|
||||
|
|
@ -134,7 +139,7 @@ func main() {
|
|||
EDUInternalAPI: eduInputAPI,
|
||||
FederationSenderAPI: fsAPI,
|
||||
RoomserverAPI: rsAPI,
|
||||
ServerKeyAPI: serverKeyAPI,
|
||||
ServerKeyAPI: skAPI,
|
||||
UserAPI: userAPI,
|
||||
KeyAPI: keyAPI,
|
||||
}
|
||||
|
|
@ -148,18 +153,18 @@ func main() {
|
|||
// Expose the matrix APIs directly rather than putting them under a /api path.
|
||||
go func() {
|
||||
base.SetupAndServeHTTP(
|
||||
config.HTTPAddress(httpAddr), // internal API
|
||||
config.HTTPAddress(httpAddr), // external API
|
||||
nil, nil, // TLS settings
|
||||
httpAPIAddr, // internal API
|
||||
httpAddr, // external API
|
||||
nil, nil, // TLS settings
|
||||
)
|
||||
}()
|
||||
// Handle HTTPS if certificate and key are provided
|
||||
if *certFile != "" && *keyFile != "" {
|
||||
go func() {
|
||||
base.SetupAndServeHTTP(
|
||||
config.HTTPAddress(httpsAddr), // internal API
|
||||
config.HTTPAddress(httpsAddr), // external API
|
||||
certFile, keyFile, // TLS settings
|
||||
setup.NoListener, // internal API
|
||||
httpsAddr, // external API
|
||||
certFile, keyFile, // TLS settings
|
||||
)
|
||||
}()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ func main() {
|
|||
base := setup.NewBaseDendrite(cfg, "RoomServerAPI", true)
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
serverKeyAPI := base.ServerKeyAPIClient()
|
||||
serverKeyAPI := base.SigningKeyServerHTTPClient()
|
||||
keyRing := serverKeyAPI.KeyRing()
|
||||
|
||||
fsAPI := base.FederationSenderHTTPClient()
|
||||
|
|
@ -33,8 +33,8 @@ func main() {
|
|||
roomserver.AddInternalRoutes(base.InternalAPIMux, rsAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.RoomServer.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.RoomServer.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,22 +16,22 @@ package main
|
|||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver"
|
||||
)
|
||||
|
||||
func main() {
|
||||
cfg := setup.ParseFlags(false)
|
||||
base := setup.NewBaseDendrite(cfg, "ServerKeyAPI", true)
|
||||
base := setup.NewBaseDendrite(cfg, "SigningKeyServer", true)
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
federation := base.CreateFederationClient()
|
||||
|
||||
intAPI := serverkeyapi.NewInternalAPI(&base.Cfg.ServerKeyAPI, federation, base.Caches)
|
||||
serverkeyapi.AddInternalRoutes(base.InternalAPIMux, intAPI, base.Caches)
|
||||
intAPI := signingkeyserver.NewInternalAPI(&base.Cfg.SigningKeyServer, federation, base.Caches)
|
||||
signingkeyserver.AddInternalRoutes(base.InternalAPIMux, intAPI, base.Caches)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.ServerKeyAPI.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.SigningKeyServer.InternalAPI.Listen,
|
||||
setup.NoListener,
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
@ -31,8 +31,8 @@ func main() {
|
|||
userapi.AddInternalRoutes(base.InternalAPIMux, userAPI)
|
||||
|
||||
base.SetupAndServeHTTP(
|
||||
base.Cfg.UserAPI.InternalAPI.Listen,
|
||||
setup.NoExternalListener,
|
||||
base.Cfg.UserAPI.InternalAPI.Listen, // internal listener
|
||||
setup.NoListener, // external listener
|
||||
nil, nil,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -168,7 +168,7 @@ func main() {
|
|||
cfg.FederationSender.Database.ConnectionString = "file:/idb/dendritejs_fedsender.db"
|
||||
cfg.MediaAPI.Database.ConnectionString = "file:/idb/dendritejs_mediaapi.db"
|
||||
cfg.RoomServer.Database.ConnectionString = "file:/idb/dendritejs_roomserver.db"
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = "file:/idb/dendritejs_serverkey.db"
|
||||
cfg.SigningKeyServer.Database.ConnectionString = "file:/idb/dendritejs_signingkeyserver.db"
|
||||
cfg.SyncAPI.Database.ConnectionString = "file:/idb/dendritejs_syncapi.db"
|
||||
cfg.KeyServer.Database.ConnectionString = "file:/idb/dendritejs_e2ekey.db"
|
||||
cfg.Global.Kafka.UseNaffka = true
|
||||
|
|
|
|||
124
cmd/furl/main.go
Normal file
124
cmd/furl/main.go
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/url"
|
||||
"os"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
var requestFrom = flag.String("from", "", "the server name that the request should originate from")
|
||||
var requestKey = flag.String("key", "matrix_key.pem", "the private key to use when signing the request")
|
||||
var requestPost = flag.Bool("post", false, "send a POST request instead of GET (pipe input into stdin or type followed by Ctrl-D)")
|
||||
|
||||
// nolint:gocyclo
|
||||
func main() {
|
||||
flag.Parse()
|
||||
|
||||
if requestFrom == nil || *requestFrom == "" {
|
||||
fmt.Println("expecting: furl -from origin.com [-key matrix_key.pem] https://path/to/url")
|
||||
fmt.Println("supported flags:")
|
||||
flag.PrintDefaults()
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
data, err := ioutil.ReadFile(*requestKey)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
var privateKey ed25519.PrivateKey
|
||||
keyBlock, _ := pem.Decode(data)
|
||||
if keyBlock == nil {
|
||||
panic("keyBlock is nil")
|
||||
}
|
||||
if keyBlock.Type == "MATRIX PRIVATE KEY" {
|
||||
_, privateKey, err = ed25519.GenerateKey(bytes.NewReader(keyBlock.Bytes))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
} else {
|
||||
panic("unexpected key block")
|
||||
}
|
||||
|
||||
client := gomatrixserverlib.NewFederationClient(
|
||||
gomatrixserverlib.ServerName(*requestFrom),
|
||||
gomatrixserverlib.KeyID(keyBlock.Headers["Key-ID"]),
|
||||
privateKey,
|
||||
false,
|
||||
)
|
||||
|
||||
u, err := url.Parse(flag.Arg(0))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
var bodyObj interface{}
|
||||
var bodyBytes []byte
|
||||
method := "GET"
|
||||
if *requestPost {
|
||||
method = "POST"
|
||||
fmt.Println("Waiting for JSON input. Press Enter followed by Ctrl-D when done...")
|
||||
|
||||
scan := bufio.NewScanner(os.Stdin)
|
||||
for scan.Scan() {
|
||||
bytes := scan.Bytes()
|
||||
bodyBytes = append(bodyBytes, bytes...)
|
||||
}
|
||||
fmt.Println("Done!")
|
||||
if err = json.Unmarshal(bodyBytes, &bodyObj); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
req := gomatrixserverlib.NewFederationRequest(
|
||||
method,
|
||||
gomatrixserverlib.ServerName(u.Host),
|
||||
u.RequestURI(),
|
||||
)
|
||||
|
||||
if *requestPost {
|
||||
if err = req.SetContent(bodyObj); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
if err = req.Sign(
|
||||
gomatrixserverlib.ServerName(*requestFrom),
|
||||
gomatrixserverlib.KeyID(keyBlock.Headers["Key-ID"]),
|
||||
privateKey,
|
||||
); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
httpReq, err := req.HTTPRequest()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
var res interface{}
|
||||
err = client.DoRequestAndParseResponse(
|
||||
context.TODO(),
|
||||
httpReq,
|
||||
&res,
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
j, err := json.MarshalIndent(res, "", " ")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
fmt.Println(string(j))
|
||||
}
|
||||
|
|
@ -27,7 +27,7 @@ func main() {
|
|||
},
|
||||
},
|
||||
}
|
||||
cfg.ServerKeyAPI.KeyPerspectives = config.KeyPerspectives{
|
||||
cfg.SigningKeyServer.KeyPerspectives = config.KeyPerspectives{
|
||||
{
|
||||
ServerName: "matrix.org",
|
||||
Keys: []config.KeyPerspectiveTrustKey{
|
||||
|
|
|
|||
|
|
@ -104,4 +104,6 @@ You __must__ import the package in `/cmd/goose/main.go` so `func init()` gets ca
|
|||
#### Database limitations
|
||||
|
||||
- SQLite3 does NOT support `ALTER TABLE table_name DROP COLUMN` - you would have to rename the column or drop the table
|
||||
entirely and recreate it.
|
||||
entirely and recreate it. ([example](https://github.com/matrix-org/dendrite/blob/master/userapi/storage/accounts/sqlite3/deltas/20200929203058_is_active.sql))
|
||||
|
||||
More information: [sqlite.org](https://www.sqlite.org/lang_altertable.html)
|
||||
|
|
|
|||
|
|
@ -252,13 +252,13 @@ room_server:
|
|||
max_idle_conns: 2
|
||||
conn_max_lifetime: -1
|
||||
|
||||
# Configuration for the Server Key API (for server signing keys).
|
||||
server_key_api:
|
||||
# Configuration for the Signing Key Server (for server signing keys).
|
||||
signing_key_server:
|
||||
internal_api:
|
||||
listen: http://localhost:7780
|
||||
connect: http://localhost:7780
|
||||
database:
|
||||
connection_string: file:serverkeyapi.db
|
||||
connection_string: file:signingkeyserver.db
|
||||
max_open_conns: 100
|
||||
max_idle_conns: 2
|
||||
conn_max_lifetime: -1
|
||||
|
|
|
|||
|
|
@ -109,7 +109,7 @@ Assuming that Postgres 9.5 (or later) is installed:
|
|||
* Create the component databases:
|
||||
|
||||
```bash
|
||||
for i in account device mediaapi syncapi roomserver serverkey federationsender appservice e2ekey naffka; do
|
||||
for i in account device mediaapi syncapi roomserver signingkeyserver federationsender appservice e2ekey naffka; do
|
||||
sudo -u postgres createdb -O dendrite dendrite_$i
|
||||
done
|
||||
```
|
||||
|
|
|
|||
|
|
@ -18,6 +18,17 @@ server {
|
|||
return 200 '{ "m.homeserver": { "base_url": "https://my.hostname.com" } }';
|
||||
}
|
||||
|
||||
# route requests to:
|
||||
# /_matrix/client/.*/sync
|
||||
# /_matrix/client/.*/user/{userId}/filter
|
||||
# /_matrix/client/.*/user/{userId}/filter/{filterID}
|
||||
# /_matrix/client/.*/keys/changes
|
||||
# /_matrix/client/.*/rooms/{roomId}/messages
|
||||
# to sync_api
|
||||
location ~ /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/messages)$ {
|
||||
proxy_pass http://sync_api:8073;
|
||||
}
|
||||
|
||||
location /_matrix/client {
|
||||
proxy_pass http://client_api:8071;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ func GetUserDevices(
|
|||
response := gomatrixserverlib.RespUserDevices{
|
||||
UserID: userID,
|
||||
StreamID: res.StreamID,
|
||||
Devices: []gomatrixserverlib.RespUserDevice{},
|
||||
}
|
||||
|
||||
for _, dev := range res.Devices {
|
||||
|
|
|
|||
|
|
@ -232,7 +232,7 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer(
|
|||
// If we successfully performed a send_join above then the other
|
||||
// server now thinks we're a part of the room. Send the newly
|
||||
// returned state to the roomserver to update our local view.
|
||||
if err = roomserverAPI.SendEventWithRewrite(
|
||||
if err = roomserverAPI.SendEventWithState(
|
||||
ctx, r.rsAPI,
|
||||
respState,
|
||||
event.Headered(respMakeJoin.RoomVersion),
|
||||
|
|
|
|||
4
go.mod
4
go.mod
|
|
@ -22,7 +22,7 @@ require (
|
|||
github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4
|
||||
github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3
|
||||
github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201002084023-8bcafefa3290
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201006143701-222e7423a5e3
|
||||
github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91
|
||||
github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
|
||||
github.com/mattn/go-sqlite3 v1.14.2
|
||||
|
|
@ -37,7 +37,7 @@ require (
|
|||
github.com/tidwall/sjson v1.1.1
|
||||
github.com/uber/jaeger-client-go v2.25.0+incompatible
|
||||
github.com/uber/jaeger-lib v2.2.0+incompatible
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200806125501-cd4685a3b4de
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20201006093556-760d9a7fd5ee
|
||||
go.uber.org/atomic v1.6.0
|
||||
golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a
|
||||
gopkg.in/h2non/bimg.v1 v1.1.4
|
||||
|
|
|
|||
8
go.sum
8
go.sum
|
|
@ -569,8 +569,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
|
|||
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
|
||||
github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd h1:xVrqJK3xHREMNjwjljkAUaadalWc0rRbmVuQatzmgwg=
|
||||
github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201002084023-8bcafefa3290 h1:ilT9QNIh2KXfvzIALtAe31IvLVZH7mVjVtOOTxdd0tY=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201002084023-8bcafefa3290/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201006143701-222e7423a5e3 h1:lWR/w6rXKZJJU1yGHb2zem/EK7+aYhUcRgAOiouZAxk=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201006143701-222e7423a5e3/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
|
||||
github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91 h1:HJ6U3S3ljJqNffYMcIeAncp5qT/i+ZMiJ2JC2F0aXP4=
|
||||
github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91/go.mod h1:sjyPyRxKM5uw1nD2cJ6O2OxI6GOqyVBfNXqKjBZTBZE=
|
||||
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 h1:ntrLa/8xVzeSs8vHFHK25k0C+NV74sYMJnNSg5NoSRo=
|
||||
|
|
@ -851,8 +851,8 @@ github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhe
|
|||
github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
|
||||
github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
|
||||
github.com/yggdrasil-network/yggdrasil-extras v0.0.0-20200525205615-6c8a4a2e8855/go.mod h1:xQdsh08Io6nV4WRnOVTe6gI8/2iTvfLDQ0CYa5aMt+I=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200806125501-cd4685a3b4de h1:p91aw0Mvol825U+5bvV9BBPl+HQxIczj7wxIOxZs70M=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200806125501-cd4685a3b4de/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20201006093556-760d9a7fd5ee h1:Kot820OfxWfYrk5di5f4S5s0jXXrQj8w8BG5826HAv4=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20201006093556-760d9a7fd5ee/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
|
||||
go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
|
||||
go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
|
||||
go.opencensus.io v0.22.1/go.mod h1:Ap50jQcDJrx6rB6VgeeFPtuPIf3wMRvRfrfYDO6+BmA=
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ type Dendrite struct {
|
|||
KeyServer KeyServer `yaml:"key_server"`
|
||||
MediaAPI MediaAPI `yaml:"media_api"`
|
||||
RoomServer RoomServer `yaml:"room_server"`
|
||||
ServerKeyAPI ServerKeyAPI `yaml:"server_key_api"`
|
||||
SigningKeyServer SigningKeyServer `yaml:"signing_key_server"`
|
||||
SyncAPI SyncAPI `yaml:"sync_api"`
|
||||
UserAPI UserAPI `yaml:"user_api"`
|
||||
|
||||
|
|
@ -302,7 +302,7 @@ func (c *Dendrite) Defaults() {
|
|||
c.KeyServer.Defaults()
|
||||
c.MediaAPI.Defaults()
|
||||
c.RoomServer.Defaults()
|
||||
c.ServerKeyAPI.Defaults()
|
||||
c.SigningKeyServer.Defaults()
|
||||
c.SyncAPI.Defaults()
|
||||
c.UserAPI.Defaults()
|
||||
c.AppServiceAPI.Defaults()
|
||||
|
|
@ -318,7 +318,7 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
|||
&c.Global, &c.ClientAPI,
|
||||
&c.EDUServer, &c.FederationAPI, &c.FederationSender,
|
||||
&c.KeyServer, &c.MediaAPI, &c.RoomServer,
|
||||
&c.ServerKeyAPI, &c.SyncAPI, &c.UserAPI,
|
||||
&c.SigningKeyServer, &c.SyncAPI, &c.UserAPI,
|
||||
&c.AppServiceAPI,
|
||||
} {
|
||||
c.Verify(configErrs, isMonolith)
|
||||
|
|
@ -333,7 +333,7 @@ func (c *Dendrite) Wiring() {
|
|||
c.KeyServer.Matrix = &c.Global
|
||||
c.MediaAPI.Matrix = &c.Global
|
||||
c.RoomServer.Matrix = &c.Global
|
||||
c.ServerKeyAPI.Matrix = &c.Global
|
||||
c.SigningKeyServer.Matrix = &c.Global
|
||||
c.SyncAPI.Matrix = &c.Global
|
||||
c.UserAPI.Matrix = &c.Global
|
||||
c.AppServiceAPI.Matrix = &c.Global
|
||||
|
|
@ -524,13 +524,13 @@ func (config *Dendrite) FederationSenderURL() string {
|
|||
return string(config.FederationSender.InternalAPI.Connect)
|
||||
}
|
||||
|
||||
// ServerKeyAPIURL returns an HTTP URL for where the server key API is listening.
|
||||
func (config *Dendrite) ServerKeyAPIURL() string {
|
||||
// Hard code the server key API server to talk HTTP for now.
|
||||
// SigningKeyServerURL returns an HTTP URL for where the signing key server is listening.
|
||||
func (config *Dendrite) SigningKeyServerURL() string {
|
||||
// Hard code the signing key server to talk HTTP for now.
|
||||
// If we support HTTPS we need to think of a practical way to do certificate validation.
|
||||
// People setting up servers shouldn't need to get a certificate valid for the public
|
||||
// internet for an internal API.
|
||||
return string(config.ServerKeyAPI.InternalAPI.Connect)
|
||||
return string(config.SigningKeyServer.InternalAPI.Connect)
|
||||
}
|
||||
|
||||
// KeyServerURL returns an HTTP URL for where the key server is listening.
|
||||
|
|
|
|||
|
|
@ -2,12 +2,12 @@ package config
|
|||
|
||||
import "github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
type ServerKeyAPI struct {
|
||||
type SigningKeyServer struct {
|
||||
Matrix *Global `yaml:"-"`
|
||||
|
||||
InternalAPI InternalAPIOptions `yaml:"internal_api"`
|
||||
|
||||
// The ServerKey database caches the public keys of remote servers.
|
||||
// The SigningKeyServer database caches the public keys of remote servers.
|
||||
// It may be accessed by the FederationAPI, the ClientAPI, and the MediaAPI.
|
||||
Database DatabaseOptions `yaml:"database"`
|
||||
|
||||
|
|
@ -19,17 +19,17 @@ type ServerKeyAPI struct {
|
|||
PreferDirectFetch bool `yaml:"prefer_direct_fetch"`
|
||||
}
|
||||
|
||||
func (c *ServerKeyAPI) Defaults() {
|
||||
func (c *SigningKeyServer) Defaults() {
|
||||
c.InternalAPI.Listen = "http://localhost:7780"
|
||||
c.InternalAPI.Connect = "http://localhost:7780"
|
||||
c.Database.Defaults()
|
||||
c.Database.ConnectionString = "file:serverkeyapi.db"
|
||||
c.Database.ConnectionString = "file:signingkeyserver.db"
|
||||
}
|
||||
|
||||
func (c *ServerKeyAPI) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
||||
checkURL(configErrs, "server_key_api.internal_api.listen", string(c.InternalAPI.Listen))
|
||||
checkURL(configErrs, "server_key_api.internal_api.bind", string(c.InternalAPI.Connect))
|
||||
checkNotEmpty(configErrs, "server_key_api.database.connection_string", string(c.Database.ConnectionString))
|
||||
func (c *SigningKeyServer) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
||||
checkURL(configErrs, "signing_key_server.internal_api.listen", string(c.InternalAPI.Listen))
|
||||
checkURL(configErrs, "signing_key_server.internal_api.bind", string(c.InternalAPI.Connect))
|
||||
checkNotEmpty(configErrs, "signing_key_server.database.connection_string", string(c.Database.ConnectionString))
|
||||
}
|
||||
|
||||
// KeyPerspectives are used to configure perspective key servers for
|
||||
|
|
@ -46,8 +46,8 @@ import (
|
|||
keyinthttp "github.com/matrix-org/dendrite/keyserver/inthttp"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
rsinthttp "github.com/matrix-org/dendrite/roomserver/inthttp"
|
||||
serverKeyAPI "github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
skinthttp "github.com/matrix-org/dendrite/serverkeyapi/inthttp"
|
||||
skapi "github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
skinthttp "github.com/matrix-org/dendrite/signingkeyserver/inthttp"
|
||||
userapi "github.com/matrix-org/dendrite/userapi/api"
|
||||
userapiinthttp "github.com/matrix-org/dendrite/userapi/inthttp"
|
||||
"github.com/sirupsen/logrus"
|
||||
|
|
@ -80,7 +80,7 @@ type BaseDendrite struct {
|
|||
const HTTPServerTimeout = time.Minute * 5
|
||||
const HTTPClientTimeout = time.Second * 30
|
||||
|
||||
const NoExternalListener = ""
|
||||
const NoListener = ""
|
||||
|
||||
// NewBaseDendrite creates a new instance to be used by a component.
|
||||
// The componentName is used for logging purposes, and should be a friendly name
|
||||
|
|
@ -208,15 +208,15 @@ func (b *BaseDendrite) FederationSenderHTTPClient() federationSenderAPI.Federati
|
|||
return f
|
||||
}
|
||||
|
||||
// ServerKeyAPIClient returns ServerKeyInternalAPI for hitting the server key API over HTTP
|
||||
func (b *BaseDendrite) ServerKeyAPIClient() serverKeyAPI.ServerKeyInternalAPI {
|
||||
f, err := skinthttp.NewServerKeyClient(
|
||||
b.Cfg.ServerKeyAPIURL(),
|
||||
// SigningKeyServerHTTPClient returns SigningKeyServer for hitting the signing key server over HTTP
|
||||
func (b *BaseDendrite) SigningKeyServerHTTPClient() skapi.SigningKeyServerAPI {
|
||||
f, err := skinthttp.NewSigningKeyServerClient(
|
||||
b.Cfg.SigningKeyServerURL(),
|
||||
b.apiHttpClient,
|
||||
b.Caches,
|
||||
)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewServerKeyInternalAPIHTTP failed", b.httpClient)
|
||||
logrus.WithError(err).Panic("SigningKeyServerHTTPClient failed", b.httpClient)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
|
@ -272,22 +272,21 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
internalAddr, _ := internalHTTPAddr.Address()
|
||||
externalAddr, _ := externalHTTPAddr.Address()
|
||||
|
||||
internalRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
externalRouter := internalRouter
|
||||
externalRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
internalRouter := externalRouter
|
||||
|
||||
internalServ := &http.Server{
|
||||
Addr: string(internalAddr),
|
||||
externalServ := &http.Server{
|
||||
Addr: string(externalAddr),
|
||||
WriteTimeout: HTTPServerTimeout,
|
||||
Handler: internalRouter,
|
||||
Handler: externalRouter,
|
||||
}
|
||||
externalServ := internalServ
|
||||
internalServ := externalServ
|
||||
|
||||
if externalAddr != NoExternalListener && externalAddr != internalAddr {
|
||||
externalRouter = mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
externalServ = &http.Server{
|
||||
Addr: string(externalAddr),
|
||||
WriteTimeout: HTTPServerTimeout,
|
||||
Handler: externalRouter,
|
||||
if internalAddr != NoListener && externalAddr != internalAddr {
|
||||
internalRouter = mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
internalServ = &http.Server{
|
||||
Addr: string(internalAddr),
|
||||
Handler: internalRouter,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -301,23 +300,25 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
externalRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(b.PublicFederationAPIMux)
|
||||
externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(b.PublicMediaAPIMux)
|
||||
|
||||
go func() {
|
||||
logrus.Infof("Starting %s listener on %s", b.componentName, internalServ.Addr)
|
||||
if certFile != nil && keyFile != nil {
|
||||
if err := internalServ.ListenAndServeTLS(*certFile, *keyFile); err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve HTTPS")
|
||||
}
|
||||
} else {
|
||||
if err := internalServ.ListenAndServe(); err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||||
}
|
||||
}
|
||||
logrus.Infof("Stopped %s listener on %s", b.componentName, internalServ.Addr)
|
||||
}()
|
||||
|
||||
if externalAddr != NoExternalListener && internalAddr != externalAddr {
|
||||
if internalAddr != NoListener && internalAddr != externalAddr {
|
||||
go func() {
|
||||
logrus.Infof("Starting %s listener on %s", b.componentName, externalServ.Addr)
|
||||
logrus.Infof("Starting internal %s listener on %s", b.componentName, internalServ.Addr)
|
||||
if certFile != nil && keyFile != nil {
|
||||
if err := internalServ.ListenAndServeTLS(*certFile, *keyFile); err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve HTTPS")
|
||||
}
|
||||
} else {
|
||||
if err := internalServ.ListenAndServe(); err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||||
}
|
||||
}
|
||||
logrus.Infof("Stopped internal %s listener on %s", b.componentName, internalServ.Addr)
|
||||
}()
|
||||
}
|
||||
|
||||
if externalAddr != NoListener {
|
||||
go func() {
|
||||
logrus.Infof("Starting external %s listener on %s", b.componentName, externalServ.Addr)
|
||||
if certFile != nil && keyFile != nil {
|
||||
if err := externalServ.ListenAndServeTLS(*certFile, *keyFile); err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve HTTPS")
|
||||
|
|
@ -327,7 +328,7 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||||
}
|
||||
}
|
||||
logrus.Infof("Stopped %s listener on %s", b.componentName, externalServ.Addr)
|
||||
logrus.Infof("Stopped external %s listener on %s", b.componentName, externalServ.Addr)
|
||||
}()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ import (
|
|||
keyAPI "github.com/matrix-org/dendrite/keyserver/api"
|
||||
"github.com/matrix-org/dendrite/mediaapi"
|
||||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
serverKeyAPI "github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
serverKeyAPI "github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi"
|
||||
userapi "github.com/matrix-org/dendrite/userapi/api"
|
||||
"github.com/matrix-org/dendrite/userapi/storage/accounts"
|
||||
|
|
@ -50,7 +50,7 @@ type Monolith struct {
|
|||
EDUInternalAPI eduServerAPI.EDUServerInputAPI
|
||||
FederationSenderAPI federationSenderAPI.FederationSenderInternalAPI
|
||||
RoomserverAPI roomserverAPI.RoomserverInternalAPI
|
||||
ServerKeyAPI serverKeyAPI.ServerKeyInternalAPI
|
||||
ServerKeyAPI serverKeyAPI.SigningKeyServerAPI
|
||||
UserAPI userapi.UserInternalAPI
|
||||
KeyAPI keyAPI.KeyInternalAPI
|
||||
|
||||
|
|
|
|||
|
|
@ -92,7 +92,7 @@ func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*con
|
|||
cfg.KeyServer.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.MediaAPI.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.RoomServer.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.SyncAPI.Database.ConnectionString = config.DataSource(database)
|
||||
cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(database)
|
||||
cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(database)
|
||||
|
|
@ -104,7 +104,7 @@ func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*con
|
|||
cfg.KeyServer.InternalAPI.Listen = assignAddress()
|
||||
cfg.MediaAPI.InternalAPI.Listen = assignAddress()
|
||||
cfg.RoomServer.InternalAPI.Listen = assignAddress()
|
||||
cfg.ServerKeyAPI.InternalAPI.Listen = assignAddress()
|
||||
cfg.SigningKeyServer.InternalAPI.Listen = assignAddress()
|
||||
cfg.SyncAPI.InternalAPI.Listen = assignAddress()
|
||||
cfg.UserAPI.InternalAPI.Listen = assignAddress()
|
||||
|
||||
|
|
@ -115,7 +115,7 @@ func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*con
|
|||
cfg.KeyServer.InternalAPI.Connect = cfg.KeyServer.InternalAPI.Listen
|
||||
cfg.MediaAPI.InternalAPI.Connect = cfg.MediaAPI.InternalAPI.Listen
|
||||
cfg.RoomServer.InternalAPI.Connect = cfg.RoomServer.InternalAPI.Listen
|
||||
cfg.ServerKeyAPI.InternalAPI.Connect = cfg.ServerKeyAPI.InternalAPI.Listen
|
||||
cfg.SigningKeyServer.InternalAPI.Connect = cfg.SigningKeyServer.InternalAPI.Listen
|
||||
cfg.SyncAPI.InternalAPI.Connect = cfg.SyncAPI.InternalAPI.Listen
|
||||
cfg.UserAPI.InternalAPI.Connect = cfg.UserAPI.InternalAPI.Listen
|
||||
|
||||
|
|
|
|||
|
|
@ -32,13 +32,6 @@ const (
|
|||
// there was a new event that references an event that we don't
|
||||
// have a copy of.
|
||||
KindNew = 2
|
||||
// KindBackfill event extend the contiguous graph going backwards.
|
||||
// They always have state.
|
||||
KindBackfill = 3
|
||||
// KindRewrite events are used when rewriting the head of the room
|
||||
// graph with entirely new state. The output events generated will
|
||||
// be state events rather than timeline events.
|
||||
KindRewrite = 4
|
||||
)
|
||||
|
||||
// DoNotSendToOtherServers tells us not to send the event to other matrix
|
||||
|
|
|
|||
|
|
@ -80,99 +80,6 @@ func SendEventWithState(
|
|||
return SendInputRoomEvents(ctx, rsAPI, ires)
|
||||
}
|
||||
|
||||
// SendEventWithRewrite writes an event with KindNew to the roomserver along
|
||||
// with a number of rewrite and outlier events for state and auth events
|
||||
// respectively.
|
||||
func SendEventWithRewrite(
|
||||
ctx context.Context, rsAPI RoomserverInternalAPI, state *gomatrixserverlib.RespState,
|
||||
event gomatrixserverlib.HeaderedEvent, haveEventIDs map[string]bool,
|
||||
) error {
|
||||
isCurrentState := map[string]struct{}{}
|
||||
for _, se := range state.StateEvents {
|
||||
isCurrentState[se.EventID()] = struct{}{}
|
||||
}
|
||||
|
||||
authAndStateEvents, err := state.Events()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var ires []InputRoomEvent
|
||||
var stateIDs []string
|
||||
|
||||
// This function generates three things:
|
||||
// A - A set of "rewrite" events, which will form the newly rewritten
|
||||
// state before the event, which includes every rewrite event that
|
||||
// came before it in its state
|
||||
// B - A set of "outlier" events, which are auth events but not part
|
||||
// of the rewritten state
|
||||
// C - A "new" event, which include all of the rewrite events in its
|
||||
// state
|
||||
for _, authOrStateEvent := range authAndStateEvents {
|
||||
if authOrStateEvent.StateKey() == nil {
|
||||
continue
|
||||
}
|
||||
if haveEventIDs[authOrStateEvent.EventID()] {
|
||||
continue
|
||||
}
|
||||
if event.StateKey() == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// We will handle an event as if it's an outlier if one of the
|
||||
// following conditions is true:
|
||||
storeAsOutlier := false
|
||||
if _, ok := isCurrentState[authOrStateEvent.EventID()]; !ok {
|
||||
// The event is an auth event and isn't a part of the state set.
|
||||
// We'll send it as an outlier because we need it to be stored
|
||||
// in case something is referring to it as an auth event.
|
||||
storeAsOutlier = true
|
||||
}
|
||||
|
||||
if storeAsOutlier {
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindOutlier,
|
||||
Event: authOrStateEvent.Headered(event.RoomVersion),
|
||||
AuthEventIDs: authOrStateEvent.AuthEventIDs(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
// If the event isn't an outlier then we'll instead send it as a
|
||||
// rewrite event, so that it'll form part of the rewritten state.
|
||||
// These events will go through the membership and latest event
|
||||
// updaters and we will generate output events, but they will be
|
||||
// flagged as non-current (i.e. didn't just happen) events.
|
||||
// Each of these rewrite events includes all of the rewrite events
|
||||
// that came before in their StateEventIDs.
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindRewrite,
|
||||
Event: authOrStateEvent.Headered(event.RoomVersion),
|
||||
AuthEventIDs: authOrStateEvent.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
|
||||
// Add the event ID into the StateEventIDs of all subsequent
|
||||
// rewrite events, and the new event.
|
||||
stateIDs = append(stateIDs, authOrStateEvent.EventID())
|
||||
}
|
||||
|
||||
// Send the final event as a new event, which will generate
|
||||
// a timeline output event for it. All of the rewrite events
|
||||
// that came before will be sent as StateEventIDs, forming a
|
||||
// new clean state before the event.
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindNew,
|
||||
Event: event,
|
||||
AuthEventIDs: event.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
|
||||
return SendInputRoomEvents(ctx, rsAPI, ires)
|
||||
}
|
||||
|
||||
// SendInputRoomEvents to the roomserver.
|
||||
func SendInputRoomEvents(
|
||||
ctx context.Context, rsAPI RoomserverInternalAPI, ires []InputRoomEvent,
|
||||
|
|
|
|||
|
|
@ -26,28 +26,30 @@ type RoomserverInternalAPI struct {
|
|||
*perform.Leaver
|
||||
*perform.Publisher
|
||||
*perform.Backfiller
|
||||
DB storage.Database
|
||||
Cfg *config.RoomServer
|
||||
Producer sarama.SyncProducer
|
||||
Cache caching.RoomServerCaches
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
KeyRing gomatrixserverlib.JSONVerifier
|
||||
fsAPI fsAPI.FederationSenderInternalAPI
|
||||
OutputRoomEventTopic string // Kafka topic for new output room events
|
||||
DB storage.Database
|
||||
Cfg *config.RoomServer
|
||||
Producer sarama.SyncProducer
|
||||
Cache caching.RoomServerCaches
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
KeyRing gomatrixserverlib.JSONVerifier
|
||||
fsAPI fsAPI.FederationSenderInternalAPI
|
||||
OutputRoomEventTopic string // Kafka topic for new output room events
|
||||
PerspectiveServerNames []gomatrixserverlib.ServerName
|
||||
}
|
||||
|
||||
func NewRoomserverAPI(
|
||||
cfg *config.RoomServer, roomserverDB storage.Database, producer sarama.SyncProducer,
|
||||
outputRoomEventTopic string, caches caching.RoomServerCaches,
|
||||
keyRing gomatrixserverlib.JSONVerifier,
|
||||
keyRing gomatrixserverlib.JSONVerifier, perspectiveServerNames []gomatrixserverlib.ServerName,
|
||||
) *RoomserverInternalAPI {
|
||||
serverACLs := acls.NewServerACLs(roomserverDB)
|
||||
a := &RoomserverInternalAPI{
|
||||
DB: roomserverDB,
|
||||
Cfg: cfg,
|
||||
Cache: caches,
|
||||
ServerName: cfg.Matrix.ServerName,
|
||||
KeyRing: keyRing,
|
||||
DB: roomserverDB,
|
||||
Cfg: cfg,
|
||||
Cache: caches,
|
||||
ServerName: cfg.Matrix.ServerName,
|
||||
PerspectiveServerNames: perspectiveServerNames,
|
||||
KeyRing: keyRing,
|
||||
Queryer: &query.Queryer{
|
||||
DB: roomserverDB,
|
||||
Cache: caches,
|
||||
|
|
@ -105,6 +107,10 @@ func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSen
|
|||
DB: r.DB,
|
||||
FSAPI: r.fsAPI,
|
||||
KeyRing: r.KeyRing,
|
||||
// Perspective servers are trusted to not lie about server keys, so we will also
|
||||
// prefer these servers when backfilling (assuming they are in the room) rather
|
||||
// than trying random servers
|
||||
PreferServers: r.PerspectiveServerNames,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@ package helpers
|
|||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
|
@ -217,6 +219,9 @@ func CheckServerAllowedToSeeEvent(
|
|||
roomState := state.NewStateResolution(db, info)
|
||||
stateEntries, err := roomState.LoadStateAtEvent(ctx, eventID)
|
||||
if err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
|
||||
|
|
@ -304,7 +309,9 @@ BFSLoop:
|
|||
util.GetLogger(ctx).WithField("server", serverName).WithField("event_id", pre).WithError(err).Error(
|
||||
"Error checking if allowed to see event",
|
||||
)
|
||||
return resultNIDs, err
|
||||
// drop the error, as we will often error at the DB level if we don't have the prev_event itself. Let's
|
||||
// just return what we have.
|
||||
return resultNIDs, nil
|
||||
}
|
||||
|
||||
// If the event hasn't been seen before and the HS
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ func (r *Inputer) processRoomEvent(
|
|||
}
|
||||
|
||||
var softfail bool
|
||||
if input.Kind == api.KindBackfill || input.Kind == api.KindNew {
|
||||
if input.Kind == api.KindNew {
|
||||
// Check that the event passes authentication checks based on the
|
||||
// current room state.
|
||||
softfail, err = helpers.CheckForSoftFail(ctx, r.DB, headered, input.StateEventIDs)
|
||||
|
|
@ -136,15 +136,6 @@ func (r *Inputer) processRoomEvent(
|
|||
return event.EventID(), rejectionErr
|
||||
}
|
||||
|
||||
if input.Kind == api.KindRewrite {
|
||||
logrus.WithFields(logrus.Fields{
|
||||
"event_id": event.EventID(),
|
||||
"type": event.Type(),
|
||||
"room": event.RoomID(),
|
||||
}).Debug("Stored rewrite")
|
||||
return event.EventID(), nil
|
||||
}
|
||||
|
||||
if err = r.updateLatestEvents(
|
||||
ctx, // context
|
||||
roomInfo, // room info for the room being updated
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ import (
|
|||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// updateLatestEvents updates the list of latest events for this room in the database and writes the
|
||||
|
|
@ -116,7 +117,6 @@ type latestEventsUpdater struct {
|
|||
}
|
||||
|
||||
func (u *latestEventsUpdater) doUpdateLatestEvents() error {
|
||||
prevEvents := u.event.PrevEvents()
|
||||
u.lastEventIDSent = u.updater.LastEventIDSent()
|
||||
u.oldStateNID = u.updater.CurrentStateSnapshotNID()
|
||||
|
||||
|
|
@ -140,30 +140,12 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Update the roomserver_previous_events table with references. This
|
||||
// is effectively tracking the structure of the DAG.
|
||||
if err = u.updater.StorePreviousEvents(u.stateAtEvent.EventNID, prevEvents); err != nil {
|
||||
return fmt.Errorf("u.updater.StorePreviousEvents: %w", err)
|
||||
}
|
||||
|
||||
// Get the event reference for our new event. This will be used when
|
||||
// determining if the event is referenced by an existing event.
|
||||
eventReference := u.event.EventReference()
|
||||
|
||||
// Check if our new event is already referenced by an existing event
|
||||
// in the room. If it is then it isn't a latest event.
|
||||
alreadyReferenced, err := u.updater.IsReferenced(eventReference)
|
||||
if err != nil {
|
||||
return fmt.Errorf("u.updater.IsReferenced: %w", err)
|
||||
}
|
||||
|
||||
// Work out what the latest events are.
|
||||
u.latest = calculateLatest(
|
||||
// Work out what the latest events are. This will include the new
|
||||
// event if it is not already referenced.
|
||||
u.calculateLatest(
|
||||
oldLatest,
|
||||
alreadyReferenced,
|
||||
prevEvents,
|
||||
types.StateAtEventAndReference{
|
||||
EventReference: eventReference,
|
||||
EventReference: u.event.EventReference(),
|
||||
StateAtEvent: u.stateAtEvent,
|
||||
},
|
||||
)
|
||||
|
|
@ -215,7 +197,9 @@ func (u *latestEventsUpdater) latestState() error {
|
|||
var err error
|
||||
roomState := state.NewStateResolution(u.api.DB, *u.roomInfo)
|
||||
|
||||
// Get a list of the current latest events.
|
||||
// Get a list of the current latest events. This may or may not
|
||||
// include the new event from the input path, depending on whether
|
||||
// it is a forward extremity or not.
|
||||
latestStateAtEvents := make([]types.StateAtEvent, len(u.latest))
|
||||
for i := range u.latest {
|
||||
latestStateAtEvents[i] = u.latest[i].StateAtEvent
|
||||
|
|
@ -249,6 +233,18 @@ func (u *latestEventsUpdater) latestState() error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("roomState.DifferenceBetweenStateSnapshots: %w", err)
|
||||
}
|
||||
if len(u.removed) > len(u.added) {
|
||||
// This really shouldn't happen.
|
||||
// TODO: What is ultimately the best way to handle this situation?
|
||||
logrus.Errorf(
|
||||
"Invalid state delta on event %q wants to remove %d state but only add %d state (between state snapshots %d and %d)",
|
||||
u.event.EventID(), len(u.removed), len(u.added), u.oldStateNID, u.newStateNID,
|
||||
)
|
||||
u.added = u.added[:0]
|
||||
u.removed = u.removed[:0]
|
||||
u.newStateNID = u.oldStateNID
|
||||
return nil
|
||||
}
|
||||
|
||||
// Also work out the state before the event removes and the event
|
||||
// adds.
|
||||
|
|
@ -262,42 +258,49 @@ func (u *latestEventsUpdater) latestState() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func calculateLatest(
|
||||
func (u *latestEventsUpdater) calculateLatest(
|
||||
oldLatest []types.StateAtEventAndReference,
|
||||
alreadyReferenced bool,
|
||||
prevEvents []gomatrixserverlib.EventReference,
|
||||
newEvent types.StateAtEventAndReference,
|
||||
) []types.StateAtEventAndReference {
|
||||
var alreadyInLatest bool
|
||||
) {
|
||||
var newLatest []types.StateAtEventAndReference
|
||||
|
||||
// First of all, let's see if any of the existing forward extremities
|
||||
// now have entries in the previous events table. If they do then we
|
||||
// will no longer include them as forward extremities.
|
||||
for _, l := range oldLatest {
|
||||
keep := true
|
||||
for _, prevEvent := range prevEvents {
|
||||
if l.EventID == prevEvent.EventID && bytes.Equal(l.EventSHA256, prevEvent.EventSHA256) {
|
||||
// This event can be removed from the latest events cause we've found an event that references it.
|
||||
// (If an event is referenced by another event then it can't be one of the latest events in the room
|
||||
// because we have an event that comes after it)
|
||||
keep = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if l.EventNID == newEvent.EventNID {
|
||||
alreadyInLatest = true
|
||||
}
|
||||
if keep {
|
||||
// Keep the event in the latest events.
|
||||
referenced, err := u.updater.IsReferenced(l.EventReference)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Errorf("Failed to retrieve event reference for %q", l.EventID)
|
||||
} else if !referenced {
|
||||
newLatest = append(newLatest, l)
|
||||
}
|
||||
}
|
||||
|
||||
if !alreadyReferenced && !alreadyInLatest {
|
||||
// This event is not referenced by any of the events in the room
|
||||
// and the event is not already in the latest events.
|
||||
// Add it to the latest events
|
||||
// Then check and see if our new event is already included in that set.
|
||||
// This ordinarily won't happen but it covers the edge-case that we've
|
||||
// already seen this event before and it's a forward extremity, so rather
|
||||
// than adding a duplicate, we'll just return the set as complete.
|
||||
for _, l := range newLatest {
|
||||
if l.EventReference.EventID == newEvent.EventReference.EventID && bytes.Equal(l.EventReference.EventSHA256, newEvent.EventReference.EventSHA256) {
|
||||
// We've already referenced this new event so we can just return
|
||||
// the newly completed extremities at this point.
|
||||
u.latest = newLatest
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// At this point we've processed the old extremities, and we've checked
|
||||
// that our new event isn't already in that set. Therefore now we can
|
||||
// check if our *new* event is a forward extremity, and if it is, add
|
||||
// it in.
|
||||
referenced, err := u.updater.IsReferenced(newEvent.EventReference)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Errorf("Failed to retrieve event reference for %q", newEvent.EventReference.EventID)
|
||||
} else if !referenced || len(newLatest) == 0 {
|
||||
newLatest = append(newLatest, newEvent)
|
||||
}
|
||||
|
||||
return newLatest
|
||||
u.latest = newLatest
|
||||
}
|
||||
|
||||
func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error) {
|
||||
|
|
|
|||
|
|
@ -30,11 +30,19 @@ import (
|
|||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// the max number of servers to backfill from per request. If this is too low we may fail to backfill when
|
||||
// we could've from another server. If this is too high we may take far too long to successfully backfill
|
||||
// as we try dead servers.
|
||||
const maxBackfillServers = 5
|
||||
|
||||
type Backfiller struct {
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
DB storage.Database
|
||||
FSAPI federationSenderAPI.FederationSenderInternalAPI
|
||||
KeyRing gomatrixserverlib.JSONVerifier
|
||||
|
||||
// The servers which should be preferred above other servers when backfilling
|
||||
PreferServers []gomatrixserverlib.ServerName
|
||||
}
|
||||
|
||||
// PerformBackfill implements api.RoomServerQueryAPI
|
||||
|
|
@ -96,7 +104,7 @@ func (r *Backfiller) backfillViaFederation(ctx context.Context, req *api.Perform
|
|||
if info == nil || info.IsStub {
|
||||
return fmt.Errorf("backfillViaFederation: missing room info for room %s", req.RoomID)
|
||||
}
|
||||
requester := newBackfillRequester(r.DB, r.FSAPI, r.ServerName, req.BackwardsExtremities)
|
||||
requester := newBackfillRequester(r.DB, r.FSAPI, r.ServerName, req.BackwardsExtremities, r.PreferServers)
|
||||
// Request 100 items regardless of what the query asks for.
|
||||
// We don't want to go much higher than this.
|
||||
// We can't honour exactly the limit as some sytests rely on requesting more for tests to pass
|
||||
|
|
@ -195,7 +203,7 @@ func (r *Backfiller) fetchAndStoreMissingEvents(ctx context.Context, roomVer gom
|
|||
logger.Infof("returned %d PDUs which made events %+v", len(res.PDUs), result)
|
||||
for _, res := range result {
|
||||
if res.Error != nil {
|
||||
logger.WithError(err).Warn("event failed PDU checks")
|
||||
logger.WithError(res.Error).Warn("event failed PDU checks")
|
||||
continue
|
||||
}
|
||||
missingMap[id] = res.Event
|
||||
|
|
@ -215,10 +223,11 @@ func (r *Backfiller) fetchAndStoreMissingEvents(ctx context.Context, roomVer gom
|
|||
|
||||
// backfillRequester implements gomatrixserverlib.BackfillRequester
|
||||
type backfillRequester struct {
|
||||
db storage.Database
|
||||
fsAPI federationSenderAPI.FederationSenderInternalAPI
|
||||
thisServer gomatrixserverlib.ServerName
|
||||
bwExtrems map[string][]string
|
||||
db storage.Database
|
||||
fsAPI federationSenderAPI.FederationSenderInternalAPI
|
||||
thisServer gomatrixserverlib.ServerName
|
||||
preferServer map[gomatrixserverlib.ServerName]bool
|
||||
bwExtrems map[string][]string
|
||||
|
||||
// per-request state
|
||||
servers []gomatrixserverlib.ServerName
|
||||
|
|
@ -226,7 +235,14 @@ type backfillRequester struct {
|
|||
eventIDMap map[string]gomatrixserverlib.Event
|
||||
}
|
||||
|
||||
func newBackfillRequester(db storage.Database, fsAPI federationSenderAPI.FederationSenderInternalAPI, thisServer gomatrixserverlib.ServerName, bwExtrems map[string][]string) *backfillRequester {
|
||||
func newBackfillRequester(
|
||||
db storage.Database, fsAPI federationSenderAPI.FederationSenderInternalAPI, thisServer gomatrixserverlib.ServerName,
|
||||
bwExtrems map[string][]string, preferServers []gomatrixserverlib.ServerName,
|
||||
) *backfillRequester {
|
||||
preferServer := make(map[gomatrixserverlib.ServerName]bool)
|
||||
for _, p := range preferServers {
|
||||
preferServer[p] = true
|
||||
}
|
||||
return &backfillRequester{
|
||||
db: db,
|
||||
fsAPI: fsAPI,
|
||||
|
|
@ -234,6 +250,7 @@ func newBackfillRequester(db storage.Database, fsAPI federationSenderAPI.Federat
|
|||
eventIDToBeforeStateIDs: make(map[string][]string),
|
||||
eventIDMap: make(map[string]gomatrixserverlib.Event),
|
||||
bwExtrems: bwExtrems,
|
||||
preferServer: preferServer,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -436,8 +453,16 @@ FindSuccessor:
|
|||
if server == b.thisServer {
|
||||
continue
|
||||
}
|
||||
servers = append(servers, server)
|
||||
if b.preferServer[server] { // insert at the front
|
||||
servers = append([]gomatrixserverlib.ServerName{server}, servers...)
|
||||
} else { // insert at the back
|
||||
servers = append(servers, server)
|
||||
}
|
||||
}
|
||||
if len(servers) > maxBackfillServers {
|
||||
servers = servers[:maxBackfillServers]
|
||||
}
|
||||
|
||||
b.servers = servers
|
||||
return servers
|
||||
}
|
||||
|
|
|
|||
|
|
@ -41,6 +41,11 @@ func NewInternalAPI(
|
|||
) api.RoomserverInternalAPI {
|
||||
cfg := &base.Cfg.RoomServer
|
||||
|
||||
var perspectiveServerNames []gomatrixserverlib.ServerName
|
||||
for _, kp := range base.Cfg.SigningKeyServer.KeyPerspectives {
|
||||
perspectiveServerNames = append(perspectiveServerNames, kp.ServerName)
|
||||
}
|
||||
|
||||
roomserverDB, err := storage.Open(&cfg.Database, base.Caches)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to room server db")
|
||||
|
|
@ -48,6 +53,6 @@ func NewInternalAPI(
|
|||
|
||||
return internal.NewRoomserverAPI(
|
||||
cfg, roomserverDB, base.KafkaProducer, string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent)),
|
||||
base.Caches, keyRing,
|
||||
base.Caches, keyRing, perspectiveServerNames,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -238,7 +238,7 @@ func TestOutputRedactedEvent(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// This tests that rewriting state via KindRewrite works correctly.
|
||||
// This tests that rewriting state works correctly.
|
||||
// This creates a small room with a create/join/name state, then replays it
|
||||
// with a new room name. We expect the output events to contain the original events,
|
||||
// followed by a single OutputNewRoomEvent with RewritesState set to true with the
|
||||
|
|
@ -344,7 +344,7 @@ func TestOutputRewritesState(t *testing.T) {
|
|||
for i := 0; i < len(rewriteEvents)-1; i++ {
|
||||
ev := rewriteEvents[i]
|
||||
inputEvents = append(inputEvents, api.InputRoomEvent{
|
||||
Kind: api.KindRewrite,
|
||||
Kind: api.KindOutlier,
|
||||
Event: ev,
|
||||
AuthEventIDs: ev.AuthEventIDs(),
|
||||
HasState: true,
|
||||
|
|
|
|||
|
|
@ -118,7 +118,7 @@ func (v StateResolution) LoadCombinedStateAfterEvents(
|
|||
// the snapshot of the room state before them was the same.
|
||||
stateBlockNIDLists, err := v.db.StateBlockNIDs(ctx, uniqueStateSnapshotNIDs(stateNIDs))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("v.db.StateBlockNIDs: %w", err)
|
||||
}
|
||||
|
||||
var stateBlockNIDs []types.StateBlockNID
|
||||
|
|
@ -131,7 +131,7 @@ func (v StateResolution) LoadCombinedStateAfterEvents(
|
|||
// multiple snapshots.
|
||||
stateEntryLists, err := v.db.StateEntries(ctx, uniqueStateBlockNIDs(stateBlockNIDs))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("v.db.StateEntries: %w", err)
|
||||
}
|
||||
stateBlockNIDsMap := stateBlockNIDListMap(stateBlockNIDLists)
|
||||
stateEntriesMap := stateEntryListMap(stateEntryLists)
|
||||
|
|
@ -623,7 +623,7 @@ func (v StateResolution) calculateAndStoreStateAfterManyEvents(
|
|||
v.calculateStateAfterManyEvents(ctx, v.roomInfo.RoomVersion, prevStates)
|
||||
metrics.algorithm = algorithm
|
||||
if err != nil {
|
||||
return metrics.stop(0, err)
|
||||
return metrics.stop(0, fmt.Errorf("v.calculateStateAfterManyEvents: %w", err))
|
||||
}
|
||||
|
||||
// TODO: Check if we can encode the new state as a delta against the
|
||||
|
|
@ -642,6 +642,7 @@ func (v StateResolution) calculateStateAfterManyEvents(
|
|||
// First stage: load the state after each of the prev events.
|
||||
combined, err = v.LoadCombinedStateAfterEvents(ctx, prevStates)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("v.LoadCombinedStateAfterEvents: %w", err)
|
||||
algorithm = "_load_combined_state"
|
||||
return
|
||||
}
|
||||
|
|
@ -672,6 +673,7 @@ func (v StateResolution) calculateStateAfterManyEvents(
|
|||
var resolved []types.StateEntry
|
||||
resolved, err = v.resolveConflicts(ctx, roomVersion, notConflicted, conflicts)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("v.resolveConflits: %w", err)
|
||||
algorithm = "_resolve_conflicts"
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -474,6 +474,32 @@ func (d *Database) StoreEvent(
|
|||
return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("d.Writer.Do: %w", err)
|
||||
}
|
||||
|
||||
// We should attempt to update the previous events table with any
|
||||
// references that this new event makes. We do this using a latest
|
||||
// events updater because it somewhat works as a mutex, ensuring
|
||||
// that there's a row-level lock on the latest room events (well,
|
||||
// on Postgres at least).
|
||||
var roomInfo *types.RoomInfo
|
||||
var updater *LatestEventsUpdater
|
||||
if prevEvents := event.PrevEvents(); len(prevEvents) > 0 {
|
||||
roomInfo, err = d.RoomInfo(ctx, event.RoomID())
|
||||
if err != nil {
|
||||
return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("d.RoomInfo: %w", err)
|
||||
}
|
||||
if roomInfo == nil && len(prevEvents) > 0 {
|
||||
return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("expected room %q to exist", event.RoomID())
|
||||
}
|
||||
updater, err = d.GetLatestEventsForUpdate(ctx, *roomInfo)
|
||||
if err != nil {
|
||||
return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("NewLatestEventsUpdater: %w", err)
|
||||
}
|
||||
if err = updater.StorePreviousEvents(eventNID, prevEvents); err != nil {
|
||||
return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("updater.StorePreviousEvents: %w", err)
|
||||
}
|
||||
succeeded := true
|
||||
err = sqlutil.EndTransaction(updater, &succeeded)
|
||||
}
|
||||
|
||||
return roomNID, types.StateAtEvent{
|
||||
BeforeStateSnapshotNID: stateNID,
|
||||
StateEntry: types.StateEntry{
|
||||
|
|
@ -483,7 +509,7 @@ func (d *Database) StoreEvent(
|
|||
},
|
||||
EventNID: eventNID,
|
||||
},
|
||||
}, redactionEvent, redactedEventID, nil
|
||||
}, redactionEvent, redactedEventID, err
|
||||
}
|
||||
|
||||
func (d *Database) PublishRoom(ctx context.Context, roomID string, publish bool) error {
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import (
|
|||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
type ServerKeyInternalAPI interface {
|
||||
type SigningKeyServerAPI interface {
|
||||
gomatrixserverlib.KeyDatabase
|
||||
|
||||
KeyRing() *gomatrixserverlib.KeyRing
|
||||
|
|
@ -7,13 +7,13 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
type ServerKeyAPI struct {
|
||||
api.ServerKeyInternalAPI
|
||||
api.SigningKeyServerAPI
|
||||
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
ServerPublicKey ed25519.PublicKey
|
||||
|
|
@ -98,10 +98,6 @@ func (s *ServerKeyAPI) FetchKeys(
|
|||
// we've failed to satisfy it from local keys, database keys or from
|
||||
// all of the fetchers. Report an error.
|
||||
logrus.Warnf("Failed to retrieve key %q for server %q", req.KeyID, req.ServerName)
|
||||
return results, fmt.Errorf(
|
||||
"server key API failed to satisfy key request for server %q key ID %q",
|
||||
req.ServerName, req.KeyID,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -7,26 +7,26 @@ import (
|
|||
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
// HTTP paths for the internal HTTP APIs
|
||||
const (
|
||||
ServerKeyInputPublicKeyPath = "/serverkeyapi/inputPublicKey"
|
||||
ServerKeyQueryPublicKeyPath = "/serverkeyapi/queryPublicKey"
|
||||
ServerKeyInputPublicKeyPath = "/signingkeyserver/inputPublicKey"
|
||||
ServerKeyQueryPublicKeyPath = "/signingkeyserver/queryPublicKey"
|
||||
)
|
||||
|
||||
// NewServerKeyClient creates a ServerKeyInternalAPI implemented by talking to a HTTP POST API.
|
||||
// NewSigningKeyServerClient creates a SigningKeyServerAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil an error is returned
|
||||
func NewServerKeyClient(
|
||||
func NewSigningKeyServerClient(
|
||||
serverKeyAPIURL string,
|
||||
httpClient *http.Client,
|
||||
cache caching.ServerKeyCache,
|
||||
) (api.ServerKeyInternalAPI, error) {
|
||||
) (api.SigningKeyServerAPI, error) {
|
||||
if httpClient == nil {
|
||||
return nil, errors.New("NewRoomserverInternalAPIHTTP: httpClient is <nil>")
|
||||
return nil, errors.New("NewSigningKeyServerClient: httpClient is <nil>")
|
||||
}
|
||||
return &httpServerKeyInternalAPI{
|
||||
serverKeyAPIURL: serverKeyAPIURL,
|
||||
|
|
@ -7,11 +7,11 @@ import (
|
|||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
func AddRoutes(s api.ServerKeyInternalAPI, internalAPIMux *mux.Router, cache caching.ServerKeyCache) {
|
||||
func AddRoutes(s api.SigningKeyServerAPI, internalAPIMux *mux.Router, cache caching.ServerKeyCache) {
|
||||
internalAPIMux.Handle(ServerKeyQueryPublicKeyPath,
|
||||
httputil.MakeInternalAPI("queryPublicKeys", func(req *http.Request) util.JSONResponse {
|
||||
request := api.QueryPublicKeysRequest{}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package serverkeyapi
|
||||
package signingkeyserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
|
|
@ -16,18 +16,18 @@ import (
|
|||
"github.com/matrix-org/dendrite/federationapi/routing"
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
type server struct {
|
||||
name gomatrixserverlib.ServerName // server name
|
||||
validity time.Duration // key validity duration from now
|
||||
config *config.ServerKeyAPI // skeleton config, from TestMain
|
||||
config *config.SigningKeyServer // skeleton config, from TestMain
|
||||
fedconfig *config.FederationAPI //
|
||||
fedclient *gomatrixserverlib.FederationClient // uses MockRoundTripper
|
||||
cache *caching.Caches // server-specific cache
|
||||
api api.ServerKeyInternalAPI // server-specific server key API
|
||||
api api.SigningKeyServerAPI // server-specific server key API
|
||||
}
|
||||
|
||||
func (s *server) renew() {
|
||||
|
|
@ -76,8 +76,8 @@ func TestMain(m *testing.M) {
|
|||
cfg.Global.PrivateKey = testPriv
|
||||
cfg.Global.KeyID = serverKeyID
|
||||
cfg.Global.KeyValidityPeriod = s.validity
|
||||
cfg.ServerKeyAPI.Database.ConnectionString = config.DataSource("file::memory:")
|
||||
s.config = &cfg.ServerKeyAPI
|
||||
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource("file::memory:")
|
||||
s.config = &cfg.SigningKeyServer
|
||||
s.fedconfig = &cfg.FederationAPI
|
||||
|
||||
// Create a transport which redirects federation requests to
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package serverkeyapi
|
||||
package signingkeyserver
|
||||
|
||||
import (
|
||||
"crypto/ed25519"
|
||||
|
|
@ -7,28 +7,28 @@ import (
|
|||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/api"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/internal"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/inthttp"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/storage"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/storage/cache"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/api"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/internal"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/inthttp"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/storage"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/storage/cache"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
|
||||
// on the given input API.
|
||||
func AddInternalRoutes(router *mux.Router, intAPI api.ServerKeyInternalAPI, caches *caching.Caches) {
|
||||
func AddInternalRoutes(router *mux.Router, intAPI api.SigningKeyServerAPI, caches *caching.Caches) {
|
||||
inthttp.AddRoutes(intAPI, router, caches)
|
||||
}
|
||||
|
||||
// NewInternalAPI returns a concerete implementation of the internal API. Callers
|
||||
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
|
||||
func NewInternalAPI(
|
||||
cfg *config.ServerKeyAPI,
|
||||
cfg *config.SigningKeyServer,
|
||||
fedClient gomatrixserverlib.KeyClient,
|
||||
caches *caching.Caches,
|
||||
) api.ServerKeyInternalAPI {
|
||||
) api.SigningKeyServerAPI {
|
||||
innerDB, err := storage.NewDatabase(
|
||||
&cfg.Database,
|
||||
cfg.Matrix.ServerName,
|
||||
|
|
@ -22,8 +22,8 @@ import (
|
|||
"golang.org/x/crypto/ed25519"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/storage/postgres"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/storage/sqlite3"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/storage/postgres"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/storage/sqlite3"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
|
|
@ -23,7 +23,7 @@ import (
|
|||
"golang.org/x/crypto/ed25519"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi/storage/sqlite3"
|
||||
"github.com/matrix-org/dendrite/signingkeyserver/storage/sqlite3"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
|
|
@ -125,31 +125,3 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er
|
|||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *OutputKeyChangeEventConsumer) OnJoinEvent(ev *gomatrixserverlib.HeaderedEvent) {
|
||||
// work out who we are now sharing rooms with which we previously were not and notify them about the joining
|
||||
// users keys:
|
||||
changed, _, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, *ev.StateKey(), []string{ev.RoomID()}, nil)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("OnJoinEvent: failed to work out changed users")
|
||||
return
|
||||
}
|
||||
// TODO: f.e changed, wake up stream
|
||||
for _, userID := range changed {
|
||||
log.Infof("OnJoinEvent:Notify %s that %s should have device lists tracked", userID, *ev.StateKey())
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OutputKeyChangeEventConsumer) OnLeaveEvent(ev *gomatrixserverlib.HeaderedEvent) {
|
||||
// work out who we are no longer sharing any rooms with and notify them about the leaving user
|
||||
_, left, err := syncinternal.TrackChangedUsers(context.Background(), s.rsAPI, *ev.StateKey(), nil, []string{ev.RoomID()})
|
||||
if err != nil {
|
||||
log.WithError(err).Error("OnLeaveEvent: failed to work out left users")
|
||||
return
|
||||
}
|
||||
// TODO: f.e left, wake up stream
|
||||
for _, userID := range left {
|
||||
log.Infof("OnLeaveEvent:Notify %s that %s should no longer track device lists", userID, *ev.StateKey())
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,7 +38,6 @@ type OutputRoomEventConsumer struct {
|
|||
rsConsumer *internal.ContinualConsumer
|
||||
db storage.Database
|
||||
notifier *sync.Notifier
|
||||
keyChanges *OutputKeyChangeEventConsumer
|
||||
}
|
||||
|
||||
// NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call Start() to begin consuming from room servers.
|
||||
|
|
@ -48,7 +47,6 @@ func NewOutputRoomEventConsumer(
|
|||
n *sync.Notifier,
|
||||
store storage.Database,
|
||||
rsAPI api.RoomserverInternalAPI,
|
||||
keyChanges *OutputKeyChangeEventConsumer,
|
||||
) *OutputRoomEventConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
|
|
@ -63,7 +61,6 @@ func NewOutputRoomEventConsumer(
|
|||
db: store,
|
||||
notifier: n,
|
||||
rsAPI: rsAPI,
|
||||
keyChanges: keyChanges,
|
||||
}
|
||||
consumer.ProcessMessage = s.onMessage
|
||||
|
||||
|
|
@ -182,26 +179,9 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
|
|||
|
||||
s.notifier.OnNewEvent(&ev, "", nil, types.NewStreamToken(pduPos, 0, nil))
|
||||
|
||||
s.notifyKeyChanges(&ev)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *OutputRoomEventConsumer) notifyKeyChanges(ev *gomatrixserverlib.HeaderedEvent) {
|
||||
membership, err := ev.Membership()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
switch membership {
|
||||
case gomatrixserverlib.Join:
|
||||
s.keyChanges.OnJoinEvent(ev)
|
||||
case gomatrixserverlib.Ban:
|
||||
fallthrough
|
||||
case gomatrixserverlib.Leave:
|
||||
s.keyChanges.OnLeaveEvent(ev)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OutputRoomEventConsumer) notifyJoinedPeeks(ctx context.Context, ev *gomatrixserverlib.HeaderedEvent, sp types.StreamPosition) (types.StreamPosition, error) {
|
||||
if ev.Type() != gomatrixserverlib.MRoomMember {
|
||||
return sp, nil
|
||||
|
|
|
|||
|
|
@ -503,7 +503,7 @@ func (r *messagesReq) backfill(roomID string, backwardsExtremities map[string][]
|
|||
// up in responses to sync requests.
|
||||
for i := range res.Events {
|
||||
_, err = r.db.WriteEvent(
|
||||
r.ctx,
|
||||
context.Background(),
|
||||
&res.Events[i],
|
||||
[]gomatrixserverlib.HeaderedEvent{},
|
||||
[]string{},
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ func AddPublicRoutes(
|
|||
}
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(
|
||||
cfg, consumer, notifier, syncDB, rsAPI, keyChangeConsumer,
|
||||
cfg, consumer, notifier, syncDB, rsAPI,
|
||||
)
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start room server consumer")
|
||||
|
|
|
|||
|
|
@ -466,17 +466,31 @@ func NewJoinResponse() *JoinResponse {
|
|||
// InviteResponse represents a /sync response for a room which is under the 'invite' key.
|
||||
type InviteResponse struct {
|
||||
InviteState struct {
|
||||
Events json.RawMessage `json:"events"`
|
||||
Events []json.RawMessage `json:"events"`
|
||||
} `json:"invite_state"`
|
||||
}
|
||||
|
||||
// NewInviteResponse creates an empty response with initialised arrays.
|
||||
func NewInviteResponse(event gomatrixserverlib.HeaderedEvent) *InviteResponse {
|
||||
res := InviteResponse{}
|
||||
res.InviteState.Events = json.RawMessage{'[', ']'}
|
||||
res.InviteState.Events = []json.RawMessage{}
|
||||
|
||||
// First see if there's invite_room_state in the unsigned key of the invite.
|
||||
// If there is then unmarshal it into the response. This will contain the
|
||||
// partial room state such as join rules, room name etc.
|
||||
if inviteRoomState := gjson.GetBytes(event.Unsigned(), "invite_room_state"); inviteRoomState.Exists() {
|
||||
res.InviteState.Events = json.RawMessage(inviteRoomState.Raw)
|
||||
_ = json.Unmarshal([]byte(inviteRoomState.Raw), &res.InviteState.Events)
|
||||
}
|
||||
|
||||
// Then we'll see if we can create a partial of the invite event itself.
|
||||
// This is needed for clients to work out *who* sent the invite.
|
||||
format, _ := event.RoomVersion.EventFormat()
|
||||
inviteEvent := gomatrixserverlib.ToClientEvent(event.Unwrap(), format)
|
||||
inviteEvent.Unsigned = nil
|
||||
if ev, err := json.Marshal(inviteEvent); err == nil {
|
||||
res.InviteState.Events = append(res.InviteState.Events, ev)
|
||||
}
|
||||
|
||||
return &res
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
package types
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
func TestNewSyncTokenWithLogs(t *testing.T) {
|
||||
|
|
@ -87,3 +90,23 @@ func TestNewSyncTokenFromString(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewInviteResponse(t *testing.T) {
|
||||
event := `{"auth_events":["$SbSsh09j26UAXnjd3RZqf2lyA3Kw2sY_VZJVZQAV9yA","$EwL53onrLwQ5gL8Dv3VrOOCvHiueXu2ovLdzqkNi3lo","$l2wGmz9iAwevBDGpHT_xXLUA5O8BhORxWIGU1cGi1ZM","$GsWFJLXgdlF5HpZeyWkP72tzXYWW3uQ9X28HBuTztHE"],"content":{"avatar_url":"","displayname":"neilalexander","membership":"invite"},"depth":9,"hashes":{"sha256":"8p+Ur4f8vLFX6mkIXhxI0kegPG7X3tWy56QmvBkExAg"},"origin":"matrix.org","origin_server_ts":1602087113066,"prev_events":["$1v-O6tNwhOZcA8bvCYY-Dnj1V2ZDE58lLPxtlV97S28"],"prev_state":[],"room_id":"!XbeXirGWSPXbEaGokF:matrix.org","sender":"@neilalexander:matrix.org","signatures":{"dendrite.neilalexander.dev":{"ed25519:BMJi":"05KQ5lPw0cSFsE4A0x1z7vi/3cc8bG4WHUsFWYkhxvk/XkXMGIYAYkpNThIvSeLfdcHlbm/k10AsBSKH8Uq4DA"},"matrix.org":{"ed25519:a_RXGa":"jeovuHr9E/x0sHbFkdfxDDYV/EyoeLi98douZYqZ02iYddtKhfB7R3WLay/a+D3V3V7IW0FUmPh/A404x5sYCw"}},"state_key":"@neilalexander:dendrite.neilalexander.dev","type":"m.room.member","unsigned":{"age":2512,"invite_room_state":[{"content":{"join_rule":"invite"},"sender":"@neilalexander:matrix.org","state_key":"","type":"m.room.join_rules"},{"content":{"avatar_url":"mxc://matrix.org/BpDaozLwgLnlNStxDxvLzhPr","displayname":"neilalexander","membership":"join"},"sender":"@neilalexander:matrix.org","state_key":"@neilalexander:matrix.org","type":"m.room.member"},{"content":{"name":"Test room"},"sender":"@neilalexander:matrix.org","state_key":"","type":"m.room.name"}]},"_room_version":"5"}`
|
||||
expected := `{"invite_state":{"events":[{"content":{"join_rule":"invite"},"sender":"@neilalexander:matrix.org","state_key":"","type":"m.room.join_rules"},{"content":{"avatar_url":"mxc://matrix.org/BpDaozLwgLnlNStxDxvLzhPr","displayname":"neilalexander","membership":"join"},"sender":"@neilalexander:matrix.org","state_key":"@neilalexander:matrix.org","type":"m.room.member"},{"content":{"name":"Test room"},"sender":"@neilalexander:matrix.org","state_key":"","type":"m.room.name"},{"content":{"avatar_url":"","displayname":"neilalexander","membership":"invite"},"event_id":"$GQmw8e8-26CQv1QuFoHBHpKF1hQj61Flg3kvv_v_XWs","origin_server_ts":1602087113066,"sender":"@neilalexander:matrix.org","state_key":"@neilalexander:dendrite.neilalexander.dev","type":"m.room.member"}]}}`
|
||||
|
||||
ev, err := gomatrixserverlib.NewEventFromTrustedJSON([]byte(event), false, gomatrixserverlib.RoomVersionV5)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
res := NewInviteResponse(ev.Headered(gomatrixserverlib.RoomVersionV5))
|
||||
j, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if string(j) != expected {
|
||||
t.Fatalf("Invite response didn't contain correct info")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,4 +52,9 @@ Inbound federation accepts a second soft-failed event
|
|||
Outbound federation requests missing prev_events and then asks for /state_ids and resolves the state
|
||||
|
||||
# We don't implement lazy membership loading yet.
|
||||
The only membership state included in a gapped incremental sync is for senders in the timeline
|
||||
The only membership state included in a gapped incremental sync is for senders in the timeline
|
||||
|
||||
# Blacklisted out of flakiness after #1479
|
||||
Invited user can reject local invite after originator leaves
|
||||
Invited user can reject invite for empty room
|
||||
If user leaves room, remote user changes device and rejoins we see update in /sync and /keys/changes
|
||||
|
|
@ -400,8 +400,6 @@ Uninvited users cannot join the room
|
|||
Users cannot invite themselves to a room
|
||||
Users cannot invite a user that is already in the room
|
||||
Invited user can reject invite
|
||||
Invited user can reject invite for empty room
|
||||
Invited user can reject local invite after originator leaves
|
||||
PUT /rooms/:room_id/typing/:user_id sets typing notification
|
||||
Typing notification sent to local room members
|
||||
Typing notifications also sent to remote room members
|
||||
|
|
@ -431,7 +429,6 @@ A prev_batch token can be used in the v1 messages API
|
|||
We don't send redundant membership state across incremental syncs by default
|
||||
Typing notifications don't leak
|
||||
Users cannot kick users from a room they are not in
|
||||
Users cannot kick users who have already left a room
|
||||
User appears in user directory
|
||||
User directory correctly update on display name change
|
||||
User in shared private room does appear in user directory
|
||||
|
|
@ -451,7 +448,6 @@ Banned servers cannot backfill
|
|||
Inbound /v1/send_leave rejects leaves from other servers
|
||||
Guest users can accept invites to private rooms over federation
|
||||
AS user (not ghost) can join room without registering
|
||||
If user leaves room, remote user changes device and rejoins we see update in /sync and /keys/changes
|
||||
Can search public room list
|
||||
Can get remote public room list
|
||||
Asking for a remote rooms list, but supplying the local server's name, returns the local rooms list
|
||||
|
|
@ -479,3 +475,5 @@ GET /capabilities is present and well formed for registered user
|
|||
m.room.history_visibility == "joined" allows/forbids appropriately for Guest users
|
||||
m.room.history_visibility == "joined" allows/forbids appropriately for Real users
|
||||
POST rejects invalid utf-8 in JSON
|
||||
Users cannot kick users who have already left a room
|
||||
A prev_batch token from incremental sync can be used in the v1 messages API
|
||||
|
|
|
|||
Loading…
Reference in a new issue