mirror of
https://github.com/matrix-org/dendrite
synced 2024-11-10 07:04:24 +00:00
Merge branch 'master' into v0.1.0
This commit is contained in:
commit
15bf385141
59 changed files with 471 additions and 385 deletions
|
@ -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,12 +76,14 @@ func JoinRoomByIDOrAlias(
|
|||
}
|
||||
|
||||
// Ask the roomserver to perform the join.
|
||||
done := make(chan util.JSONResponse, 1)
|
||||
go func() {
|
||||
defer close(done)
|
||||
rsAPI.PerformJoin(req.Context(), &joinReq, &joinRes)
|
||||
if joinRes.Error != nil {
|
||||
return joinRes.Error.JSONResponse()
|
||||
}
|
||||
|
||||
return util.JSONResponse{
|
||||
done <- joinRes.Error.JSONResponse()
|
||||
} else {
|
||||
done <- util.JSONResponse{
|
||||
Code: http.StatusOK,
|
||||
// TODO: Put the response struct somewhere internal.
|
||||
JSON: struct {
|
||||
|
@ -87,3 +91,17 @@ func JoinRoomByIDOrAlias(
|
|||
}{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.PublicKey[:])] = struct{}{}
|
||||
nodemap[hex.EncodeToString(peer.SigPublicKey[:])] = 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,8 +153,8 @@ 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
|
||||
httpAPIAddr, // internal API
|
||||
httpAddr, // external API
|
||||
nil, nil, // TLS settings
|
||||
)
|
||||
}()
|
||||
|
@ -157,8 +162,8 @@ func main() {
|
|||
if *certFile != "" && *keyFile != "" {
|
||||
go func() {
|
||||
base.SetupAndServeHTTP(
|
||||
config.HTTPAddress(httpsAddr), // internal API
|
||||
config.HTTPAddress(httpsAddr), // external API
|
||||
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{
|
||||
|
|
|
@ -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
|
||||
```
|
||||
|
|
|
@ -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-20201005165532-90f2e192f625
|
||||
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-20201005165532-90f2e192f625 h1:9ShgY0ZkfLzqe3gv18V5WxDAZ4dgUvJwnGORycox680=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20201005165532-90f2e192f625/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,23 +272,22 @@ 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),
|
||||
WriteTimeout: HTTPServerTimeout,
|
||||
Handler: internalRouter,
|
||||
}
|
||||
externalServ := internalServ
|
||||
|
||||
if externalAddr != NoExternalListener && externalAddr != internalAddr {
|
||||
externalRouter = mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
externalServ = &http.Server{
|
||||
externalServ := &http.Server{
|
||||
Addr: string(externalAddr),
|
||||
WriteTimeout: HTTPServerTimeout,
|
||||
Handler: externalRouter,
|
||||
}
|
||||
internalServ := externalServ
|
||||
|
||||
if internalAddr != NoListener && externalAddr != internalAddr {
|
||||
internalRouter = mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||||
internalServ = &http.Server{
|
||||
Addr: string(internalAddr),
|
||||
Handler: internalRouter,
|
||||
}
|
||||
}
|
||||
|
||||
internalRouter.PathPrefix(httputil.InternalPathPrefix).Handler(b.InternalAPIMux)
|
||||
|
@ -301,8 +300,9 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
externalRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(b.PublicFederationAPIMux)
|
||||
externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(b.PublicMediaAPIMux)
|
||||
|
||||
if internalAddr != NoListener && internalAddr != externalAddr {
|
||||
go func() {
|
||||
logrus.Infof("Starting %s listener on %s", b.componentName, internalServ.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")
|
||||
|
@ -312,12 +312,13 @@ func (b *BaseDendrite) SetupAndServeHTTP(
|
|||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||||
}
|
||||
}
|
||||
logrus.Infof("Stopped %s listener on %s", b.componentName, internalServ.Addr)
|
||||
logrus.Infof("Stopped internal %s listener on %s", b.componentName, internalServ.Addr)
|
||||
}()
|
||||
}
|
||||
|
||||
if externalAddr != NoExternalListener && internalAddr != externalAddr {
|
||||
if externalAddr != NoListener {
|
||||
go func() {
|
||||
logrus.Infof("Starting %s listener on %s", b.componentName, externalServ.Addr)
|
||||
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,
|
||||
|
|
|
@ -34,12 +34,13 @@ type RoomserverInternalAPI struct {
|
|||
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{
|
||||
|
@ -47,6 +48,7 @@ func NewRoomserverAPI(
|
|||
Cfg: cfg,
|
||||
Cache: caches,
|
||||
ServerName: cfg.Matrix.ServerName,
|
||||
PerspectiveServerNames: perspectiveServerNames,
|
||||
KeyRing: keyRing,
|
||||
Queryer: &query.Queryer{
|
||||
DB: roomserverDB,
|
||||
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
@ -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,27 +197,12 @@ func (u *latestEventsUpdater) latestState() error {
|
|||
var err error
|
||||
roomState := state.NewStateResolution(u.api.DB, *u.roomInfo)
|
||||
|
||||
// Get a list of the current room state events if available.
|
||||
var currentState []types.StateEntry
|
||||
if u.roomInfo.StateSnapshotNID != 0 {
|
||||
currentState, _ = roomState.LoadStateAtSnapshot(u.ctx, u.roomInfo.StateSnapshotNID)
|
||||
}
|
||||
|
||||
// Get a list of the current latest events. This will include both
|
||||
// the current room state and the latest events after the input event.
|
||||
// The idea is that we will perform state resolution on this set and
|
||||
// any conflicting events will be resolved properly.
|
||||
latestStateAtEvents := make([]types.StateAtEvent, len(u.latest)+len(currentState))
|
||||
offset := 0
|
||||
for i := range currentState {
|
||||
latestStateAtEvents[i] = types.StateAtEvent{
|
||||
BeforeStateSnapshotNID: u.roomInfo.StateSnapshotNID,
|
||||
StateEntry: currentState[i],
|
||||
}
|
||||
offset++
|
||||
}
|
||||
// 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[offset+i] = u.latest[i].StateAtEvent
|
||||
latestStateAtEvents[i] = u.latest[i].StateAtEvent
|
||||
}
|
||||
|
||||
// Takes the NIDs of the latest events and creates a state snapshot
|
||||
|
@ -266,6 +233,14 @@ 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?
|
||||
return fmt.Errorf(
|
||||
"invalid state delta wants to remove %d state but only add %d state (between state snapshots %d and %d)",
|
||||
len(u.removed), len(u.added), u.oldStateNID, u.newStateNID,
|
||||
)
|
||||
}
|
||||
|
||||
// Also work out the state before the event removes and the event
|
||||
// adds.
|
||||
|
@ -279,42 +254,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 {
|
||||
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
|
||||
|
@ -218,6 +226,7 @@ type backfillRequester struct {
|
|||
db storage.Database
|
||||
fsAPI federationSenderAPI.FederationSenderInternalAPI
|
||||
thisServer gomatrixserverlib.ServerName
|
||||
preferServer map[gomatrixserverlib.ServerName]bool
|
||||
bwExtrems map[string][]string
|
||||
|
||||
// per-request state
|
||||
|
@ -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
|
||||
}
|
||||
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,
|
||||
|
|
|
@ -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"
|
||||
)
|
||||
|
|
@ -474,3 +474,5 @@ Inbound federation rejects invite rejections which include invalid JSON for room
|
|||
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
|
||||
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