Merge branch 'master' into 653_implement_read_markers

This commit is contained in:
Kegsay 2020-10-08 14:48:34 +01:00 committed by GitHub
commit 9975631032
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
68 changed files with 534 additions and 431 deletions

View file

@ -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

View file

@ -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"
]

View file

@ -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 .

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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))

View file

@ -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
}
}

View file

@ -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,
)
}

View file

@ -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(

View file

@ -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))

View file

@ -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 {

View file

@ -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,
)
}

View file

@ -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()

View file

@ -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,
)
}

View file

@ -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,
)
}

View file

@ -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
)
}()
}

View file

@ -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,
)
}

View file

@ -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,
)
}

View file

@ -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,
)
}

View file

@ -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
View 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))
}

View file

@ -27,7 +27,7 @@ func main() {
},
},
}
cfg.ServerKeyAPI.KeyPerspectives = config.KeyPerspectives{
cfg.SigningKeyServer.KeyPerspectives = config.KeyPerspectives{
{
ServerName: "matrix.org",
Keys: []config.KeyPerspectiveTrustKey{

View file

@ -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)

View file

@ -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

View file

@ -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
```

View file

@ -40,6 +40,7 @@ func GetUserDevices(
response := gomatrixserverlib.RespUserDevices{
UserID: userID,
StreamID: res.StreamID,
Devices: []gomatrixserverlib.RespUserDevice{},
}
for _, dev := range res.Devices {

View file

@ -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
View file

@ -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
View file

@ -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=

View file

@ -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.

View file

@ -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

View file

@ -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)
}()
}

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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,

View file

@ -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,
}
}

View file

@ -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

View file

@ -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

View file

@ -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) {

View file

@ -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
}

View file

@ -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,
)
}

View file

@ -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,

View file

@ -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
}

View file

@ -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 {

View file

@ -6,7 +6,7 @@ import (
"github.com/matrix-org/gomatrixserverlib"
)
type ServerKeyInternalAPI interface {
type SigningKeyServerAPI interface {
gomatrixserverlib.KeyDatabase
KeyRing() *gomatrixserverlib.KeyRing

View file

@ -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,
)
}
}

View file

@ -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,

View file

@ -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{}

View file

@ -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

View file

@ -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,

View file

@ -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"
)

View file

@ -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"
)

View file

@ -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())
}
}

View file

@ -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

View file

@ -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{},

View file

@ -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")

View file

@ -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
}

View file

@ -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")
}
}

View file

@ -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

View file

@ -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
@ -460,6 +456,9 @@ After changing password, can log in with new password
After changing password, existing session still works
After changing password, different sessions can optionally be kept
After changing password, a different session no longer works by default
Read markers appear in incremental v2 /sync
Read markers appear in initial v2 /sync
Read markers can be updated
Local users can peek into world_readable rooms by room ID
We can't peek into rooms with shared history_visibility
We can't peek into rooms with invited history_visibility
@ -478,6 +477,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
Read markers appear in incremental v2 /sync
Read markers appear in initial v2 /sync
Read markers can be updated
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