mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-12 09:23:09 -06:00
Merge branch 'master' into ElyKar/add-debug-logs
This commit is contained in:
Tristan Claverie
commit
1194e09e22
5
.gitignore
vendored
5
.gitignore
vendored
|
|
@ -34,3 +34,8 @@ _testmain.go
|
|||
*.exe
|
||||
*.test
|
||||
*.prof
|
||||
|
||||
# Generated keys
|
||||
*.pem
|
||||
*.key
|
||||
*.crt
|
||||
|
|
|
|||
|
|
@ -69,6 +69,8 @@ Dendrite requires a postgres database engine, version 9.5 or later.
|
|||
done
|
||||
```
|
||||
|
||||
(On macOS, omit `sudo -u postgres` from the above commands.)
|
||||
|
||||
### Crypto key generation
|
||||
|
||||
Generate the keys:
|
||||
|
|
|
|||
10
docker/Dockerfile
Normal file
10
docker/Dockerfile
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
FROM golang:alpine3.6
|
||||
|
||||
RUN mkdir /build
|
||||
|
||||
WORKDIR /build
|
||||
|
||||
RUN apk --update --no-cache add openssl bash git && \
|
||||
go get github.com/constabulary/gb/...
|
||||
|
||||
CMD ["bash", "docker/build.sh"]
|
||||
100
docker/README.md
Normal file
100
docker/README.md
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
Development with Docker
|
||||
---
|
||||
|
||||
With `docker` and `docker-compose` you can easily spin up a development environment
|
||||
and start working on dendrite.
|
||||
|
||||
### Requirements
|
||||
|
||||
- docker
|
||||
- docker-compose (version 3+)
|
||||
|
||||
### Configuration
|
||||
|
||||
Copy the `dendrite-docker.yaml` file to the root of the project and rename it to
|
||||
`dendrite.yaml`. It already contains the defaults used in `docker-compose` for
|
||||
networking so you will only have to change things like the `server_name` or to
|
||||
toggle `naffka`.
|
||||
|
||||
You can run the following `docker-compose` commands either from the top directory
|
||||
specifying the `docker-compose` file
|
||||
```
|
||||
docker-compose -f docker/docker-compose.yml <cmd>
|
||||
```
|
||||
or from within the `docker` directory
|
||||
|
||||
```
|
||||
docker-compose <cmd>
|
||||
```
|
||||
|
||||
### Starting a monolith server
|
||||
|
||||
For the monolith server you would need a postgres instance
|
||||
|
||||
```
|
||||
docker-compose up postgres
|
||||
```
|
||||
|
||||
and the dendrite component from `bin/dendrite-monolith-server`
|
||||
|
||||
```
|
||||
docker-compose up monolith
|
||||
```
|
||||
|
||||
The monolith will be listening on `http://localhost:8008`.
|
||||
|
||||
You would also have to make the following adjustments to `dendrite.yaml`.
|
||||
- Set `use_naffka: true`
|
||||
- Uncomment the `database/naffka` postgres url.
|
||||
|
||||
### Starting a multiprocess server
|
||||
|
||||
The multiprocess server requires kafka, zookeeper and postgres
|
||||
|
||||
```
|
||||
docker-compose up kafka zookeeper postgres
|
||||
```
|
||||
|
||||
and the following dendrite components
|
||||
|
||||
```
|
||||
docker-compose up client_api media_api sync_api room_server public_rooms_api
|
||||
docker-compose up client_api_proxy
|
||||
```
|
||||
|
||||
The `client-api-proxy` will be listening on `http://localhost:8008`.
|
||||
|
||||
You would also have to make the following adjustments to `dendrite.yaml`.
|
||||
- Set `use_naffka: false`
|
||||
- Comment out the `database/naffka` postgres url.
|
||||
|
||||
### Starting federation
|
||||
|
||||
```
|
||||
docker-compose up federation_api federation_sender
|
||||
docker-compose up federation_api_proxy
|
||||
```
|
||||
|
||||
You can point other Matrix servers to `http://localhost:8448`.
|
||||
|
||||
### Creating a new component
|
||||
|
||||
You can create a new dendrite component by adding an entry to the `docker-compose.yml`
|
||||
file and creating a startup script for the component in `docker/services`.
|
||||
For more information refer to the official docker-compose [documentation](https://docs.docker.com/compose/).
|
||||
|
||||
```yaml
|
||||
new_component:
|
||||
container_name: dendrite_room_server
|
||||
hostname: new_component
|
||||
# Start up script.
|
||||
entrypoint: ["bash", "./docker/services/new-component.sh"]
|
||||
# Use the common Dockerfile for all the dendrite components.
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- another_component
|
||||
networks:
|
||||
- internal
|
||||
```
|
||||
18
docker/build.sh
Normal file
18
docker/build.sh
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
#!/bin/bash
|
||||
|
||||
gb build
|
||||
|
||||
# Generate the keys if they don't already exist.
|
||||
if [ ! -f server.key ] || [ ! -f server.crt ] || [ ! -f matrix_key.pem ]; then
|
||||
echo "Generating keys ..."
|
||||
|
||||
rm -f server.key server.crt matrix_key.pem
|
||||
|
||||
test -f server.key || openssl req -x509 -newkey rsa:4096 \
|
||||
-keyout server.key \
|
||||
-out server.crt \
|
||||
-days 3650 -nodes \
|
||||
-subj /CN=localhost
|
||||
|
||||
test -f matrix_key.pem || /build/bin/generate-keys -private-key matrix_key.pem
|
||||
fi
|
||||
126
docker/dendrite-docker.yml
Normal file
126
docker/dendrite-docker.yml
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
# The config file format version
|
||||
# This is used by dendrite to tell if it understands the config format.
|
||||
# This will change if the structure of the config file changes or if the meaning
|
||||
# of an existing config key changes.
|
||||
version: 0
|
||||
|
||||
# The matrix specific config
|
||||
matrix:
|
||||
# The name of the server. This is usually the domain name, e.g 'matrix.org', 'localhost'.
|
||||
server_name: "example.com"
|
||||
# The path to the PEM formatted matrix private key.
|
||||
private_key: "matrix_key.pem"
|
||||
# The x509 certificates used by the federation listeners for this server
|
||||
federation_certificates: ["server.crt"]
|
||||
# The list of identity servers trusted to verify third party identifiers by this server.
|
||||
# Defaults to no trusted servers.
|
||||
trusted_third_party_id_servers:
|
||||
- vector.im
|
||||
- matrix.org
|
||||
- riot.im
|
||||
|
||||
# The media repository config
|
||||
media:
|
||||
# The base path to where the media files will be stored. May be relative or absolute.
|
||||
base_path: /var/dendrite/media
|
||||
|
||||
# The maximum file size in bytes that is allowed to be stored on this server.
|
||||
# Note: if max_file_size_bytes is set to 0, the size is unlimited.
|
||||
# Note: if max_file_size_bytes is not set, it will default to 10485760 (10MB)
|
||||
max_file_size_bytes: 10485760
|
||||
|
||||
# Whether to dynamically generate thumbnails on-the-fly if the requested resolution is not already generated
|
||||
# NOTE: This is a possible denial-of-service attack vector - use at your own risk
|
||||
dynamic_thumbnails: false
|
||||
|
||||
# A list of thumbnail sizes to be pre-generated for downloaded remote / uploaded content
|
||||
# method is one of crop or scale. If omitted, it will default to scale.
|
||||
# crop scales to fill the requested dimensions and crops the excess.
|
||||
# scale scales to fit the requested dimensions and one dimension may be smaller than requested.
|
||||
thumbnail_sizes:
|
||||
- width: 32
|
||||
height: 32
|
||||
method: crop
|
||||
- width: 96
|
||||
height: 96
|
||||
method: crop
|
||||
- width: 320
|
||||
height: 240
|
||||
method: scale
|
||||
- width: 640
|
||||
height: 480
|
||||
method: scale
|
||||
- width: 800
|
||||
height: 600
|
||||
method: scale
|
||||
|
||||
# The config for the TURN server
|
||||
turn:
|
||||
# Whether or not guests can request TURN credentials
|
||||
turn_allow_guests: true
|
||||
# How long the authorization should last
|
||||
turn_user_lifetime: "1h"
|
||||
# The list of TURN URIs to pass to clients
|
||||
turn_uris: []
|
||||
|
||||
# Authorization via Shared Secret
|
||||
# The shared secret from coturn
|
||||
turn_shared_secret: "<SECRET STRING GOES HERE>"
|
||||
|
||||
# Authorization via Static Username & Password
|
||||
# Hardcoded Username and Password
|
||||
turn_username: ""
|
||||
turn_password: ""
|
||||
|
||||
# The config for communicating with kafka
|
||||
kafka:
|
||||
# Where the kafka servers are running.
|
||||
addresses: ["kafka:9092"]
|
||||
# Whether to use naffka instead of kafka.
|
||||
# Naffka can only be used when running dendrite as a single monolithic server.
|
||||
# Kafka can be used both with a monolithic server and when running the
|
||||
# components as separate servers.
|
||||
# If enabled database.naffka must also be specified.
|
||||
use_naffka: true
|
||||
# The names of the kafka topics to use.
|
||||
topics:
|
||||
output_room_event: roomserverOutput
|
||||
output_client_data: clientapiOutput
|
||||
user_updates: userUpdates
|
||||
|
||||
# The postgres connection configs for connecting to the databases e.g a postgres:// URI
|
||||
database:
|
||||
account: "postgres://dendrite:itsasecret@postgres/dendrite_account?sslmode=disable"
|
||||
device: "postgres://dendrite:itsasecret@postgres/dendrite_device?sslmode=disable"
|
||||
media_api: "postgres://dendrite:itsasecret@postgres/dendrite_mediaapi?sslmode=disable"
|
||||
sync_api: "postgres://dendrite:itsasecret@postgres/dendrite_syncapi?sslmode=disable"
|
||||
room_server: "postgres://dendrite:itsasecret@postgres/dendrite_roomserver?sslmode=disable"
|
||||
server_key: "postgres://dendrite:itsasecret@postgres/dendrite_serverkey?sslmode=disable"
|
||||
federation_sender: "postgres://dendrite:itsasecret@postgres/dendrite_federationsender?sslmode=disable"
|
||||
public_rooms_api: "postgres://dendrite:itsasecret@postgres/dendrite_publicroomsapi?sslmode=disable"
|
||||
# If using naffka you need to specify a naffka database
|
||||
naffka: "postgres://dendrite:itsasecret@postgres/dendrite_naffka?sslmode=disable"
|
||||
|
||||
# The TCP host:port pairs to bind the internal HTTP APIs to.
|
||||
# These shouldn't be exposed to the public internet.
|
||||
# These aren't needed when running dendrite as a monolithic server.
|
||||
listen:
|
||||
room_server: "room_server:7770"
|
||||
client_api: "client_api:7771"
|
||||
federation_api: "federation_api:7772"
|
||||
sync_api: "sync_api:7773"
|
||||
media_api: "media_api:7774"
|
||||
public_rooms_api: "public_rooms_api:7775"
|
||||
federation_sender: "federation_sender:7776"
|
||||
|
||||
# The configuration for tracing the dendrite components.
|
||||
tracing:
|
||||
# Config for the jaeger opentracing reporter.
|
||||
# See https://godoc.org/github.com/uber/jaeger-client-go/config#Configuration
|
||||
# for documtation.
|
||||
jaeger:
|
||||
disabled: true
|
||||
|
||||
# A list of application service config files to use
|
||||
application_services:
|
||||
config_files: []
|
||||
173
docker/docker-compose.yml
Normal file
173
docker/docker-compose.yml
Normal file
|
|
@ -0,0 +1,173 @@
|
|||
version: "3.4"
|
||||
services:
|
||||
monolith:
|
||||
container_name: dendrite_monolith
|
||||
hostname: monolith
|
||||
entrypoint: ["bash", "./docker/services/monolith.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
networks:
|
||||
- internal
|
||||
depends_on:
|
||||
- postgres
|
||||
ports:
|
||||
- "8008:8008"
|
||||
- "8448:8448"
|
||||
|
||||
client_api_proxy:
|
||||
container_name: dendrite_client_api_proxy
|
||||
hostname: client_api_proxy
|
||||
entrypoint: ["bash", "./docker/services/client-api-proxy.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
networks:
|
||||
- internal
|
||||
depends_on:
|
||||
- postgres
|
||||
- sync_api
|
||||
- client_api
|
||||
- media_api
|
||||
- public_rooms_api
|
||||
ports:
|
||||
- "8008:8008"
|
||||
|
||||
client_api:
|
||||
container_name: dendrite_client_api
|
||||
hostname: client_api
|
||||
entrypoint: ["bash", "./docker/services/client-api.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
- room_server
|
||||
networks:
|
||||
- internal
|
||||
|
||||
media_api:
|
||||
container_name: dendrite_media_api
|
||||
hostname: media_api
|
||||
entrypoint: ["bash", "./docker/services/media-api.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
public_rooms_api:
|
||||
container_name: dendrite_public_rooms_api
|
||||
hostname: public_rooms_api
|
||||
entrypoint: ["bash", "./docker/services/public-rooms-api.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
sync_api:
|
||||
container_name: dendrite_sync_api
|
||||
hostname: sync_api
|
||||
entrypoint: ["bash", "./docker/services/sync-api.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
room_server:
|
||||
container_name: dendrite_room_server
|
||||
hostname: room_server
|
||||
entrypoint: ["bash", "./docker/services/room-server.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
federation_api_proxy:
|
||||
container_name: dendrite_federation_api_proxy
|
||||
hostname: federation_api_proxy
|
||||
entrypoint: ["bash", "./docker/services/federation-api-proxy.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
- federation_api
|
||||
- federation_sender
|
||||
- media_api
|
||||
networks:
|
||||
- internal
|
||||
ports:
|
||||
- "8448:8448"
|
||||
|
||||
federation_api:
|
||||
container_name: dendrite_federation_api
|
||||
hostname: federation_api
|
||||
entrypoint: ["bash", "./docker/services/federation-api.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
federation_sender:
|
||||
container_name: dendrite_federation_sender
|
||||
hostname: federation_sender
|
||||
entrypoint: ["bash", "./docker/services/federation-sender.sh"]
|
||||
build: ./
|
||||
volumes:
|
||||
- ..:/build
|
||||
depends_on:
|
||||
- postgres
|
||||
networks:
|
||||
- internal
|
||||
|
||||
postgres:
|
||||
container_name: dendrite_postgres
|
||||
hostname: postgres
|
||||
image: postgres:9.5
|
||||
restart: always
|
||||
volumes:
|
||||
- ./postgres/create_db.sh:/docker-entrypoint-initdb.d/20-create_db.sh
|
||||
environment:
|
||||
POSTGRES_PASSWORD: itsasecret
|
||||
POSTGRES_USER: dendrite
|
||||
networks:
|
||||
- internal
|
||||
|
||||
zookeeper:
|
||||
container_name: dendrite_zk
|
||||
hostname: zookeeper
|
||||
image: zookeeper
|
||||
networks:
|
||||
- internal
|
||||
|
||||
kafka:
|
||||
container_name: dendrite_kafka
|
||||
hostname: kafka
|
||||
image: wurstmeister/kafka
|
||||
environment:
|
||||
KAFKA_ADVERTISED_HOST_NAME: "kafka"
|
||||
KAFKA_DELETE_TOPIC_ENABLE: "true"
|
||||
KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
|
||||
depends_on:
|
||||
- zookeeper
|
||||
networks:
|
||||
- internal
|
||||
|
||||
networks:
|
||||
internal:
|
||||
attachable: true
|
||||
5
docker/postgres/create_db.sh
Normal file
5
docker/postgres/create_db.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
for db in account device mediaapi syncapi roomserver serverkey federationsender publicroomsapi naffka; do
|
||||
createdb -O dendrite dendrite_$db
|
||||
done
|
||||
9
docker/services/client-api-proxy.sh
Normal file
9
docker/services/client-api-proxy.sh
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/client-api-proxy --bind-address ":8008" \
|
||||
--client-api-server-url "http://client_api:7771" \
|
||||
--sync-api-server-url "http://sync_api:7773" \
|
||||
--media-api-server-url "http://media_api:7774" \
|
||||
--public-rooms-api-server-url "http://public_rooms_api:7775" \
|
||||
5
docker/services/client-api.sh
Normal file
5
docker/services/client-api.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-client-api-server --config=dendrite.yaml
|
||||
7
docker/services/federation-api-proxy.sh
Normal file
7
docker/services/federation-api-proxy.sh
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/federation-api-proxy --bind-address ":8448" \
|
||||
--federation-api-url "http://federation_api_server:7772" \
|
||||
--media-api-server-url "http://media_api:7774" \
|
||||
5
docker/services/federation-api.sh
Normal file
5
docker/services/federation-api.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-federation-api-server --config dendrite.yaml
|
||||
5
docker/services/federation-sender.sh
Normal file
5
docker/services/federation-sender.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-federation-sender-server --config dendrite.yaml
|
||||
5
docker/services/media-api.sh
Normal file
5
docker/services/media-api.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-media-api-server --config dendrite.yaml
|
||||
5
docker/services/monolith.sh
Normal file
5
docker/services/monolith.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-monolith-server --tls-cert=server.crt --tls-key=server.key
|
||||
5
docker/services/public-rooms-api.sh
Normal file
5
docker/services/public-rooms-api.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-public-rooms-api-server --config dendrite.yaml
|
||||
5
docker/services/room-server.sh
Normal file
5
docker/services/room-server.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-room-server --config=dendrite.yaml
|
||||
5
docker/services/sync-api.sh
Normal file
5
docker/services/sync-api.sh
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
|
||||
bash ./docker/build.sh
|
||||
|
||||
./bin/dendrite-sync-api-server --config=dendrite.yaml
|
||||
|
|
@ -20,10 +20,11 @@ import (
|
|||
|
||||
// Account represents a Matrix account on this home server.
|
||||
type Account struct {
|
||||
UserID string
|
||||
Localpart string
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
Profile *Profile
|
||||
UserID string
|
||||
Localpart string
|
||||
ServerName gomatrixserverlib.ServerName
|
||||
Profile *Profile
|
||||
AppServiceID string
|
||||
// TODO: Other flags like IsAdmin, IsGuest
|
||||
// TODO: Devices
|
||||
// TODO: Associations (e.g. with application services)
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@ type LoginType string
|
|||
|
||||
// The relevant login types implemented in Dendrite
|
||||
const (
|
||||
LoginTypeDummy = "m.login.dummy"
|
||||
LoginTypeSharedSecret = "org.matrix.login.shared_secret"
|
||||
LoginTypeRecaptcha = "m.login.recaptcha"
|
||||
LoginTypeDummy = "m.login.dummy"
|
||||
LoginTypeSharedSecret = "org.matrix.login.shared_secret"
|
||||
LoginTypeRecaptcha = "m.login.recaptcha"
|
||||
LoginTypeApplicationService = "m.login.application_service"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -32,14 +32,16 @@ CREATE TABLE IF NOT EXISTS account_accounts (
|
|||
-- When this account was first created, as a unix timestamp (ms resolution).
|
||||
created_ts BIGINT NOT NULL,
|
||||
-- The password hash for this account. Can be NULL if this is a passwordless account.
|
||||
password_hash TEXT
|
||||
password_hash TEXT,
|
||||
-- Identifies which Application Service this account belongs to, if any.
|
||||
appservice_id TEXT
|
||||
-- TODO:
|
||||
-- is_guest, is_admin, appservice_id, upgraded_ts, devices, any email reset stuff?
|
||||
-- is_guest, is_admin, upgraded_ts, devices, any email reset stuff?
|
||||
);
|
||||
`
|
||||
|
||||
const insertAccountSQL = "" +
|
||||
"INSERT INTO account_accounts(localpart, created_ts, password_hash) VALUES ($1, $2, $3)"
|
||||
"INSERT INTO account_accounts(localpart, created_ts, password_hash, appservice_id) VALUES ($1, $2, $3, $4)"
|
||||
|
||||
const selectAccountByLocalpartSQL = "" +
|
||||
"SELECT localpart FROM account_accounts WHERE localpart = $1"
|
||||
|
|
@ -78,17 +80,26 @@ func (s *accountsStatements) prepare(db *sql.DB, server gomatrixserverlib.Server
|
|||
// this account will be passwordless. Returns an error if this account already exists. Returns the account
|
||||
// on success.
|
||||
func (s *accountsStatements) insertAccount(
|
||||
ctx context.Context, localpart, hash string,
|
||||
ctx context.Context, localpart, hash, appserviceID string,
|
||||
) (*authtypes.Account, error) {
|
||||
createdTimeMS := time.Now().UnixNano() / 1000000
|
||||
stmt := s.insertAccountStmt
|
||||
if _, err := stmt.ExecContext(ctx, localpart, createdTimeMS, hash); err != nil {
|
||||
|
||||
var err error
|
||||
if appserviceID == "" {
|
||||
_, err = stmt.ExecContext(ctx, localpart, createdTimeMS, hash, nil)
|
||||
} else {
|
||||
_, err = stmt.ExecContext(ctx, localpart, createdTimeMS, hash, appserviceID)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &authtypes.Account{
|
||||
Localpart: localpart,
|
||||
UserID: makeUserID(localpart, s.serverName),
|
||||
ServerName: s.serverName,
|
||||
Localpart: localpart,
|
||||
UserID: makeUserID(localpart, s.serverName),
|
||||
ServerName: s.serverName,
|
||||
AppServiceID: appserviceID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -121,11 +121,17 @@ func (d *Database) SetDisplayName(
|
|||
// for this account. If no password is supplied, the account will be a passwordless account. If the
|
||||
// account already exists, it will return nil, nil.
|
||||
func (d *Database) CreateAccount(
|
||||
ctx context.Context, localpart, plaintextPassword string,
|
||||
ctx context.Context, localpart, plaintextPassword, appserviceID string,
|
||||
) (*authtypes.Account, error) {
|
||||
hash, err := hashPassword(plaintextPassword)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
var err error
|
||||
|
||||
// Generate a password hash if this is not a password-less user
|
||||
hash := ""
|
||||
if plaintextPassword != "" {
|
||||
hash, err = hashPassword(plaintextPassword)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := d.profiles.insertProfile(ctx, localpart); err != nil {
|
||||
if common.IsUniqueConstraintViolationErr(err) {
|
||||
|
|
@ -133,7 +139,7 @@ func (d *Database) CreateAccount(
|
|||
}
|
||||
return nil, err
|
||||
}
|
||||
return d.accounts.insertAccount(ctx, localpart, hash)
|
||||
return d.accounts.insertAccount(ctx, localpart, hash, appserviceID)
|
||||
}
|
||||
|
||||
// SaveMembership saves the user matching a given localpart as a member of a given
|
||||
|
|
|
|||
66
src/github.com/matrix-org/dendrite/clientapi/clientapi.go
Normal file
66
src/github.com/matrix-org/dendrite/clientapi/clientapi.go
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package clientapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/clientapi/routing"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupClientAPIComponent sets up and registers HTTP handlers for the ClientAPI
|
||||
// component.
|
||||
func SetupClientAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
accountsDB *accounts.Database,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
keyRing *gomatrixserverlib.KeyRing,
|
||||
aliasAPI api.RoomserverAliasAPI,
|
||||
inputAPI api.RoomserverInputAPI,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
userUpdateProducer := &producers.UserUpdateProducer{
|
||||
Producer: base.KafkaProducer,
|
||||
Topic: string(base.Cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
|
||||
syncProducer := &producers.SyncAPIProducer{
|
||||
Producer: base.KafkaProducer,
|
||||
Topic: string(base.Cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, accountsDB, queryAPI,
|
||||
)
|
||||
if err := consumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start room server consumer")
|
||||
}
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, *base.Cfg, roomserverProducer,
|
||||
queryAPI, aliasAPI, accountsDB, deviceDB,
|
||||
federation, *keyRing,
|
||||
userUpdateProducer, syncProducer,
|
||||
)
|
||||
}
|
||||
|
|
@ -86,7 +86,7 @@ func MissingToken(msg string) *MatrixError {
|
|||
}
|
||||
|
||||
// UnknownToken is an error when the client tries to access a resource which
|
||||
// requires authentication and supplies a valid, but out-of-date token.
|
||||
// requires authentication and supplies an unrecognized token
|
||||
func UnknownToken(msg string) *MatrixError {
|
||||
return &MatrixError{"M_UNKNOWN_TOKEN", msg}
|
||||
}
|
||||
|
|
@ -109,6 +109,13 @@ func UserInUse(msg string) *MatrixError {
|
|||
return &MatrixError{"M_USER_IN_USE", msg}
|
||||
}
|
||||
|
||||
// ASExclusive is an error returned when an application service tries to
|
||||
// register an username that is outside of its registered namespace, or if a
|
||||
// user attempts to register a username within an exclusive namespace
|
||||
func ASExclusive(msg string) *MatrixError {
|
||||
return &MatrixError{"M_EXCLUSIVE", msg}
|
||||
}
|
||||
|
||||
// GuestAccessForbidden is an error which is returned when the client is
|
||||
// forbidden from accessing a resource as a guest.
|
||||
func GuestAccessForbidden(msg string) *MatrixError {
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@
|
|||
package routing
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
|
@ -43,10 +42,28 @@ type createRoomRequest struct {
|
|||
Topic string `json:"topic"`
|
||||
Preset string `json:"preset"`
|
||||
CreationContent map[string]interface{} `json:"creation_content"`
|
||||
InitialState json.RawMessage `json:"initial_state"` // TODO
|
||||
InitialState []fledglingEvent `json:"initial_state"`
|
||||
RoomAliasName string `json:"room_alias_name"`
|
||||
GuestCanJoin bool `json:"guest_can_join"`
|
||||
}
|
||||
|
||||
const (
|
||||
presetPrivateChat = "private_chat"
|
||||
presetTrustedPrivateChat = "trusted_private_chat"
|
||||
presetPublicChat = "public_chat"
|
||||
)
|
||||
|
||||
const (
|
||||
joinRulePublic = "public"
|
||||
joinRuleInvite = "invite"
|
||||
)
|
||||
const (
|
||||
historyVisibilityShared = "shared"
|
||||
// TODO: These should be implemented once history visibility is implemented
|
||||
// historyVisibilityWorldReadable = "world_readable"
|
||||
// historyVisibilityInvited = "invited"
|
||||
)
|
||||
|
||||
func (r createRoomRequest) Validate() *util.JSONResponse {
|
||||
whitespace := "\t\n\x0b\x0c\r " // https://docs.python.org/2/library/string.html#string.whitespace
|
||||
// https://github.com/matrix-org/synapse/blob/v0.19.2/synapse/handlers/room.py#L81
|
||||
|
|
@ -70,6 +87,16 @@ func (r createRoomRequest) Validate() *util.JSONResponse {
|
|||
}
|
||||
}
|
||||
}
|
||||
switch r.Preset {
|
||||
case presetPrivateChat, presetTrustedPrivateChat, presetPublicChat:
|
||||
break
|
||||
default:
|
||||
return &util.JSONResponse{
|
||||
Code: 400,
|
||||
JSON: jsonerror.BadJSON("preset must be any of 'private_chat', 'trusted_private_chat', 'public_chat'"),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -81,9 +108,9 @@ type createRoomResponse struct {
|
|||
|
||||
// fledglingEvent is a helper representation of an event used when creating many events in succession.
|
||||
type fledglingEvent struct {
|
||||
Type string
|
||||
StateKey string
|
||||
Content interface{}
|
||||
Type string `json:"type"`
|
||||
StateKey string `json:"state_key"`
|
||||
Content interface{} `json:"content"`
|
||||
}
|
||||
|
||||
// CreateRoom implements /createRoom
|
||||
|
|
@ -141,6 +168,20 @@ func createRoom(req *http.Request, device *authtypes.Device,
|
|||
AvatarURL: profile.AvatarURL,
|
||||
}
|
||||
|
||||
var joinRules, historyVisibility string
|
||||
switch r.Preset {
|
||||
case presetPrivateChat:
|
||||
joinRules = joinRuleInvite
|
||||
historyVisibility = historyVisibilityShared
|
||||
case presetTrustedPrivateChat:
|
||||
joinRules = joinRuleInvite
|
||||
historyVisibility = historyVisibilityShared
|
||||
// TODO If trusted_private_chat, all invitees are given the same power level as the room creator.
|
||||
case presetPublicChat:
|
||||
joinRules = joinRulePublic
|
||||
historyVisibility = historyVisibilityShared
|
||||
}
|
||||
|
||||
var builtEvents []gomatrixserverlib.Event
|
||||
|
||||
// send events into the room in order of:
|
||||
|
|
@ -151,7 +192,7 @@ func createRoom(req *http.Request, device *authtypes.Device,
|
|||
// 5- m.room.join_rules
|
||||
// 6- m.room.history_visibility
|
||||
// 7- m.room.guest_access (opt)
|
||||
// 8- other initial state items TODO
|
||||
// 8- other initial state items
|
||||
// 9- m.room.name (opt)
|
||||
// 10- m.room.topic (opt)
|
||||
// 11- invite events (opt) - with is_direct flag if applicable TODO
|
||||
|
|
@ -166,16 +207,22 @@ func createRoom(req *http.Request, device *authtypes.Device,
|
|||
{"m.room.member", userID, membershipContent},
|
||||
{"m.room.power_levels", "", common.InitialPowerLevelsContent(userID)},
|
||||
// TODO: m.room.canonical_alias
|
||||
{"m.room.join_rules", "", common.JoinRulesContent{JoinRule: "public"}}, // FIXME: Allow this to be changed
|
||||
{"m.room.history_visibility", "", common.HistoryVisibilityContent{HistoryVisibility: "joined"}}, // FIXME: Allow this to be changed
|
||||
{"m.room.guest_access", "", common.GuestAccessContent{GuestAccess: "can_join"}}, // FIXME: Allow this to be changed
|
||||
// TODO: Other initial state items
|
||||
{"m.room.name", "", common.NameContent{Name: r.Name}}, // FIXME: Only send the name event if a name is supplied, to avoid sending a false room name removal event
|
||||
{"m.room.topic", "", common.TopicContent{Topic: r.Topic}},
|
||||
// TODO: invite events
|
||||
// TODO: 3pid invite events
|
||||
// TODO: m.room.aliases
|
||||
{"m.room.join_rules", "", common.JoinRulesContent{JoinRule: joinRules}},
|
||||
{"m.room.history_visibility", "", common.HistoryVisibilityContent{HistoryVisibility: historyVisibility}},
|
||||
}
|
||||
if r.GuestCanJoin {
|
||||
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.guest_access", "", common.GuestAccessContent{GuestAccess: "can_join"}})
|
||||
}
|
||||
eventsToMake = append(eventsToMake, r.InitialState...)
|
||||
if r.Name != "" {
|
||||
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.name", "", common.NameContent{Name: r.Name}})
|
||||
}
|
||||
if r.Topic != "" {
|
||||
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.topic", "", common.TopicContent{Topic: r.Topic}})
|
||||
}
|
||||
// TODO: invite events
|
||||
// TODO: 3pid invite events
|
||||
// TODO: m.room.aliases
|
||||
|
||||
authEvents := gomatrixserverlib.NewAuthEvents(nil)
|
||||
for i, e := range eventsToMake {
|
||||
|
|
|
|||
|
|
@ -109,7 +109,7 @@ func Login(
|
|||
// but that would leak the existence of the user.
|
||||
return util.JSONResponse{
|
||||
Code: 403,
|
||||
JSON: jsonerror.Forbidden("username or password was incorrect, or the accouqnt does not exist"),
|
||||
JSON: jsonerror.Forbidden("username or password was incorrect, or the account does not exist"),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ var (
|
|||
// remembered. If ANY parameters are supplied, the server should REPLACE all knowledge of
|
||||
// previous parameters with the ones supplied. This mean you cannot "build up" request params.
|
||||
type registerRequest struct {
|
||||
// registration parameters.
|
||||
// registration parameters
|
||||
Password string `json:"password"`
|
||||
Username string `json:"username"`
|
||||
Admin bool `json:"admin"`
|
||||
|
|
@ -71,6 +71,10 @@ type registerRequest struct {
|
|||
Auth authDict `json:"auth"`
|
||||
|
||||
InitialDisplayName *string `json:"initial_device_display_name"`
|
||||
|
||||
// Application Services place Type in the root of their registration
|
||||
// request, whereas clients place it in the authDict struct.
|
||||
Type authtypes.LoginType `json:"type"`
|
||||
}
|
||||
|
||||
type authDict struct {
|
||||
|
|
@ -233,7 +237,110 @@ func validateRecaptcha(
|
|||
return nil
|
||||
}
|
||||
|
||||
// Register processes a /register request. http://matrix.org/speculator/spec/HEAD/client_server/unstable.html#post-matrix-client-unstable-register
|
||||
// UsernameIsWithinApplicationServiceNamespace checks to see if a username falls
|
||||
// within any of the namespaces of a given Application Service. If no
|
||||
// Application Service is given, it will check to see if it matches any
|
||||
// Application Service's namespace.
|
||||
func UsernameIsWithinApplicationServiceNamespace(
|
||||
cfg *config.Dendrite,
|
||||
username string,
|
||||
appservice *config.ApplicationService,
|
||||
) bool {
|
||||
if appservice != nil {
|
||||
// Loop through given Application Service's namespaces and see if any match
|
||||
for _, namespace := range appservice.NamespaceMap["users"] {
|
||||
// AS namespaces are checked for validity in config
|
||||
if namespace.RegexpObject.MatchString(username) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Loop through all known Application Service's namespaces and see if any match
|
||||
for _, knownAppservice := range cfg.Derived.ApplicationServices {
|
||||
for _, namespace := range knownAppservice.NamespaceMap["users"] {
|
||||
// AS namespaces are checked for validity in config
|
||||
if namespace.RegexpObject.MatchString(username) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// UsernameMatchesMultipleExclusiveNamespaces will check if a given username matches
|
||||
// more than one exclusive namespace. More than one is not allowed
|
||||
func UsernameMatchesMultipleExclusiveNamespaces(
|
||||
cfg *config.Dendrite,
|
||||
username string,
|
||||
) bool {
|
||||
// Check namespaces and see if more than one match
|
||||
matchCount := 0
|
||||
for _, appservice := range cfg.Derived.ApplicationServices {
|
||||
for _, namespaceSlice := range appservice.NamespaceMap {
|
||||
for _, namespace := range namespaceSlice {
|
||||
// Check if we have a match on this username
|
||||
if namespace.RegexpObject.MatchString(username) {
|
||||
matchCount++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return matchCount > 1
|
||||
}
|
||||
|
||||
// validateApplicationService checks if a provided application service token
|
||||
// corresponds to one that is registered. If so, then it checks if the desired
|
||||
// username is within that application service's namespace. As long as these
|
||||
// two requirements are met, no error will be returned.
|
||||
func validateApplicationService(
|
||||
cfg *config.Dendrite,
|
||||
req *http.Request,
|
||||
username string,
|
||||
) (string, *util.JSONResponse) {
|
||||
// Check if the token if the application service is valid with one we have
|
||||
// registered in the config.
|
||||
accessToken := req.URL.Query().Get("access_token")
|
||||
var matchedApplicationService *config.ApplicationService
|
||||
for _, appservice := range cfg.Derived.ApplicationServices {
|
||||
if appservice.ASToken == accessToken {
|
||||
matchedApplicationService = &appservice
|
||||
break
|
||||
}
|
||||
}
|
||||
if matchedApplicationService != nil {
|
||||
return "", &util.JSONResponse{
|
||||
Code: 401,
|
||||
JSON: jsonerror.UnknownToken("Supplied access_token does not match any known application service"),
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure the desired username is within at least one of the application service's namespaces.
|
||||
if !UsernameIsWithinApplicationServiceNamespace(cfg, username, matchedApplicationService) {
|
||||
// If we didn't find any matches, return M_EXCLUSIVE
|
||||
return "", &util.JSONResponse{
|
||||
Code: 401,
|
||||
JSON: jsonerror.ASExclusive(fmt.Sprintf(
|
||||
"Supplied username %s did not match any namespaces for application service ID: %s", username, matchedApplicationService.ID)),
|
||||
}
|
||||
}
|
||||
|
||||
// Check this user does not fit multiple application service namespaces
|
||||
if UsernameMatchesMultipleExclusiveNamespaces(cfg, username) {
|
||||
return "", &util.JSONResponse{
|
||||
Code: 401,
|
||||
JSON: jsonerror.ASExclusive(fmt.Sprintf(
|
||||
"Supplied username %s matches multiple exclusive application service namespaces. Only 1 match allowed", username)),
|
||||
}
|
||||
}
|
||||
|
||||
// No errors, registration valid
|
||||
return matchedApplicationService.ID, nil
|
||||
}
|
||||
|
||||
// Register processes a /register request.
|
||||
// http://matrix.org/speculator/spec/HEAD/client_server/unstable.html#post-matrix-client-unstable-register
|
||||
func Register(
|
||||
req *http.Request,
|
||||
accountDB *accounts.Database,
|
||||
|
|
@ -273,6 +380,17 @@ func Register(
|
|||
return *resErr
|
||||
}
|
||||
|
||||
// Make sure normal user isn't registering under an exclusive application
|
||||
// service namespace. Skip this check if no app services are registered.
|
||||
if r.Auth.Type != "m.login.application_service" &&
|
||||
len(cfg.Derived.ApplicationServices) != 0 &&
|
||||
cfg.Derived.ExclusiveApplicationServicesUsernameRegexp.MatchString(r.Username) {
|
||||
return util.JSONResponse{
|
||||
Code: 400,
|
||||
JSON: jsonerror.ASExclusive("This username is reserved by an application service."),
|
||||
}
|
||||
}
|
||||
|
||||
logger := util.GetLogger(req.Context())
|
||||
logger.WithFields(log.Fields{
|
||||
"username": r.Username,
|
||||
|
|
@ -294,7 +412,6 @@ func handleRegistrationFlow(
|
|||
deviceDB *devices.Database,
|
||||
) util.JSONResponse {
|
||||
// TODO: Shared secret registration (create new user scripts)
|
||||
// TODO: AS API registration
|
||||
// TODO: Enable registration config flag
|
||||
// TODO: Guest account upgrading
|
||||
|
||||
|
|
@ -331,6 +448,21 @@ func handleRegistrationFlow(
|
|||
// Add SharedSecret to the list of completed registration stages
|
||||
sessions[sessionID] = append(sessions[sessionID], authtypes.LoginTypeSharedSecret)
|
||||
|
||||
case authtypes.LoginTypeApplicationService:
|
||||
// Check Application Service register user request is valid.
|
||||
// The application service's ID is returned if so.
|
||||
appserviceID, err := validateApplicationService(cfg, req, r.Username)
|
||||
|
||||
if err != nil {
|
||||
return *err
|
||||
}
|
||||
|
||||
// If no error, application service was successfully validated.
|
||||
// Don't need to worry about appending to registration stages as
|
||||
// application service registration is entirely separate.
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB,
|
||||
r.Username, "", appserviceID, r.InitialDisplayName)
|
||||
|
||||
case authtypes.LoginTypeDummy:
|
||||
// there is nothing to do
|
||||
// Add Dummy to the list of completed registration stages
|
||||
|
|
@ -344,18 +476,36 @@ func handleRegistrationFlow(
|
|||
}
|
||||
|
||||
// Check if the user's registration flow has been completed successfully
|
||||
if !checkFlowCompleted(sessions[sessionID], cfg.Derived.Registration.Flows) {
|
||||
// There are still more stages to complete.
|
||||
// Return the flows and those that have been completed.
|
||||
return util.JSONResponse{
|
||||
Code: 401,
|
||||
JSON: newUserInteractiveResponse(sessionID,
|
||||
cfg.Derived.Registration.Flows, cfg.Derived.Registration.Params),
|
||||
}
|
||||
// A response with current registration flow and remaining available methods
|
||||
// will be returned if a flow has not been successfully completed yet
|
||||
return checkAndCompleteFlow(sessions[sessionID], req, r, sessionID, cfg, accountDB, deviceDB)
|
||||
}
|
||||
|
||||
// checkAndCompleteFlow checks if a given registration flow is completed given
|
||||
// a set of allowed flows. If so, registration is completed, otherwise a
|
||||
// response with
|
||||
func checkAndCompleteFlow(
|
||||
flow []authtypes.LoginType,
|
||||
req *http.Request,
|
||||
r registerRequest,
|
||||
sessionID string,
|
||||
cfg *config.Dendrite,
|
||||
accountDB *accounts.Database,
|
||||
deviceDB *devices.Database,
|
||||
) util.JSONResponse {
|
||||
if checkFlowCompleted(flow, cfg.Derived.Registration.Flows) {
|
||||
// This flow was completed, registration can continue
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB,
|
||||
r.Username, r.Password, "", r.InitialDisplayName)
|
||||
}
|
||||
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB,
|
||||
r.Username, r.Password, r.InitialDisplayName)
|
||||
// There are still more stages to complete.
|
||||
// Return the flows and those that have been completed.
|
||||
return util.JSONResponse{
|
||||
Code: 401,
|
||||
JSON: newUserInteractiveResponse(sessionID,
|
||||
cfg.Derived.Registration.Flows, cfg.Derived.Registration.Params),
|
||||
}
|
||||
}
|
||||
|
||||
// LegacyRegister process register requests from the legacy v1 API
|
||||
|
|
@ -396,10 +546,10 @@ func LegacyRegister(
|
|||
return util.MessageResponse(403, "HMAC incorrect")
|
||||
}
|
||||
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB, r.Username, r.Password, nil)
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB, r.Username, r.Password, "", nil)
|
||||
case authtypes.LoginTypeDummy:
|
||||
// there is nothing to do
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB, r.Username, r.Password, nil)
|
||||
return completeRegistration(req.Context(), accountDB, deviceDB, r.Username, r.Password, "", nil)
|
||||
default:
|
||||
return util.JSONResponse{
|
||||
Code: 501,
|
||||
|
|
@ -441,7 +591,7 @@ func completeRegistration(
|
|||
ctx context.Context,
|
||||
accountDB *accounts.Database,
|
||||
deviceDB *devices.Database,
|
||||
username, password string,
|
||||
username, password, appserviceID string,
|
||||
displayName *string,
|
||||
) util.JSONResponse {
|
||||
if username == "" {
|
||||
|
|
@ -450,14 +600,15 @@ func completeRegistration(
|
|||
JSON: jsonerror.BadJSON("missing username"),
|
||||
}
|
||||
}
|
||||
if password == "" {
|
||||
// Blank passwords are only allowed by registered application services
|
||||
if password == "" && appserviceID == "" {
|
||||
return util.JSONResponse{
|
||||
Code: 400,
|
||||
JSON: jsonerror.BadJSON("missing password"),
|
||||
}
|
||||
}
|
||||
|
||||
acc, err := accountDB.CreateAccount(ctx, username, password)
|
||||
acc, err := accountDB.CreateAccount(ctx, username, password, appserviceID)
|
||||
if err != nil {
|
||||
return util.JSONResponse{
|
||||
Code: 500,
|
||||
|
|
@ -580,7 +731,10 @@ func checkFlows(
|
|||
// checkFlowCompleted checks if a registration flow complies with any allowed flow
|
||||
// dictated by the server. Order of stages does not matter. A user may complete
|
||||
// extra stages as long as the required stages of at least one flow is met.
|
||||
func checkFlowCompleted(flow []authtypes.LoginType, allowedFlows []authtypes.Flow) bool {
|
||||
func checkFlowCompleted(
|
||||
flow []authtypes.LoginType,
|
||||
allowedFlows []authtypes.Flow,
|
||||
) bool {
|
||||
// Iterate through possible flows to check whether any have been fully completed.
|
||||
for _, allowedFlow := range allowedFlows {
|
||||
if checkFlows(flow, allowedFlow.Stages) {
|
||||
|
|
|
|||
|
|
@ -69,7 +69,7 @@ func main() {
|
|||
os.Exit(1)
|
||||
}
|
||||
|
||||
account, err := accountDB.CreateAccount(context.Background(), *username, *password)
|
||||
account, err := accountDB.CreateAccount(context.Background(), *username, *password, "")
|
||||
if err != nil {
|
||||
fmt.Println(err.Error())
|
||||
os.Exit(1)
|
||||
|
|
|
|||
|
|
@ -15,115 +15,29 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/clientapi/routing"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/clientapi"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
)
|
||||
|
||||
const componentName = "clientapi"
|
||||
|
||||
var (
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file, For more information see the config file in this repository")
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
|
||||
flag.Parse()
|
||||
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteClientAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
aliasAPI := api.NewRoomserverAliasAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
inputAPI := api.NewRoomserverInputAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
kafkaProducer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka producers")
|
||||
}
|
||||
|
||||
userUpdateProducer := &producers.UserUpdateProducer{
|
||||
Producer: kafkaProducer,
|
||||
Topic: string(cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
|
||||
syncProducer := &producers.SyncAPIProducer{
|
||||
Producer: kafkaProducer,
|
||||
Topic: string(cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
)
|
||||
|
||||
accountDB, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", cfg.Database.Account, err.Error())
|
||||
}
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup device database(%q): %s", cfg.Database.Device, err.Error())
|
||||
}
|
||||
keyDB, err := keydb.NewDatabase(string(cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
base := basecomponent.NewBaseDendrite(cfg, "ClientAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
accountDB := base.CreateAccountsDB()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, accountDB, queryAPI)
|
||||
if err = consumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
|
||||
log.Info("Starting client API server on ", cfg.Listen.ClientAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(
|
||||
api, *cfg, roomserverProducer,
|
||||
queryAPI, aliasAPI, accountDB, deviceDB, federation, keyRing,
|
||||
userUpdateProducer, syncProducer,
|
||||
clientapi.SetupClientAPIComponent(
|
||||
base, deviceDB, accountDB, federation, &keyRing,
|
||||
alias, input, query,
|
||||
)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.ClientAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.ClientAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,86 +15,27 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/dendrite/federationapi/routing"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const componentName = "federationapi"
|
||||
|
||||
var (
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/federationapi"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "FederationAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteFederationAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
federationapi.SetupFederationAPIComponent(
|
||||
base, accountDB, federation, &keyRing,
|
||||
alias, input, query,
|
||||
)
|
||||
|
||||
keyDB, err := keydb.NewDatabase(string(cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
|
||||
accountDB, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", cfg.Database.Account, err.Error())
|
||||
}
|
||||
|
||||
keyRing := gomatrixserverlib.KeyRing{
|
||||
KeyFetchers: []gomatrixserverlib.KeyFetcher{
|
||||
// TODO: Use perspective key fetchers for production.
|
||||
&gomatrixserverlib.DirectKeyFetcher{Client: federation.Client},
|
||||
},
|
||||
KeyDatabase: keyDB,
|
||||
}
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
inputAPI := api.NewRoomserverInputAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
aliasAPI := api.NewRoomserverAliasAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup kafka producers(%s): %s", cfg.Kafka.Addresses, err)
|
||||
}
|
||||
|
||||
log.Info("Starting federation API server on ", cfg.Listen.FederationAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, *cfg, queryAPI, aliasAPI, roomserverProducer, keyRing, federation, accountDB)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, api)
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.FederationAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,77 +15,22 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
"github.com/matrix-org/dendrite/federationsender/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationsender"
|
||||
)
|
||||
|
||||
const componentName = "federationsender"
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "FederationSender")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
federation := base.CreateFederationClient()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
_, _, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteFederationSender")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewDatabase(string(cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create federation sender database with data source %s : %s", cfg.Database.FederationSender, err)
|
||||
}
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
federationsender.SetupFederationSenderComponent(
|
||||
base, federation, query,
|
||||
)
|
||||
|
||||
queues := queue.NewOutgoingQueues(cfg.Matrix.ServerName, federation)
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, queues, db, queryAPI)
|
||||
if err = consumer.Start(); err != nil {
|
||||
log.WithError(err).Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
|
||||
api := mux.NewRouter()
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, api)
|
||||
|
||||
if err := http.ListenAndServe(string(cfg.Listen.FederationSender), nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationSender))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,64 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
"github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const componentName = "mediaapi"
|
||||
|
||||
var (
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/mediaapi"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "MediaAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
mediaapi.SetupMediaAPIComponent(base, deviceDB)
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteMediaAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
db, err := storage.Open(string(cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
log.WithError(err).Panic("Failed to open database")
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.WithError(err).Panicf("Failed to setup device database(%q)", cfg.Database.Device)
|
||||
}
|
||||
|
||||
client := gomatrixserverlib.NewClient()
|
||||
|
||||
log.Info("Starting media API server on ", cfg.Listen.MediaAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, cfg, db, deviceDB, client)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.MediaAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.MediaAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,56 +15,26 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/naffka"
|
||||
|
||||
mediaapi_routing "github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
mediaapi_storage "github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
|
||||
roomserver_alias "github.com/matrix-org/dendrite/roomserver/alias"
|
||||
roomserver_input "github.com/matrix-org/dendrite/roomserver/input"
|
||||
roomserver_query "github.com/matrix-org/dendrite/roomserver/query"
|
||||
roomserver_storage "github.com/matrix-org/dendrite/roomserver/storage"
|
||||
|
||||
clientapi_consumers "github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
clientapi_routing "github.com/matrix-org/dendrite/clientapi/routing"
|
||||
|
||||
syncapi_consumers "github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
syncapi_routing "github.com/matrix-org/dendrite/syncapi/routing"
|
||||
syncapi_storage "github.com/matrix-org/dendrite/syncapi/storage"
|
||||
syncapi_sync "github.com/matrix-org/dendrite/syncapi/sync"
|
||||
syncapi_types "github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
federationapi_routing "github.com/matrix-org/dendrite/federationapi/routing"
|
||||
|
||||
federationsender_consumers "github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
federationsender_storage "github.com/matrix-org/dendrite/federationsender/storage"
|
||||
|
||||
publicroomsapi_consumers "github.com/matrix-org/dendrite/publicroomsapi/consumers"
|
||||
publicroomsapi_routing "github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
publicroomsapi_storage "github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/clientapi"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationapi"
|
||||
"github.com/matrix-org/dendrite/federationsender"
|
||||
"github.com/matrix-org/dendrite/mediaapi"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/syncapi"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const componentName = "monolith"
|
||||
|
||||
var (
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
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")
|
||||
certFile = flag.String("tls-cert", "", "The PEM formatted X509 certificate to use for TLS")
|
||||
|
|
@ -72,287 +42,40 @@ var (
|
|||
)
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseMonolithFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "Monolith")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.LoadMonolithic(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
alias, input, query := roomserver.SetupRoomServerComponent(base)
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
clientapi.SetupClientAPIComponent(base, deviceDB, accountDB, federation, &keyRing, alias, input, query)
|
||||
federationapi.SetupFederationAPIComponent(base, accountDB, federation, &keyRing, alias, input, query)
|
||||
federationsender.SetupFederationSenderComponent(base, federation, query)
|
||||
mediaapi.SetupMediaAPIComponent(base, deviceDB)
|
||||
publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB)
|
||||
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteMonolith")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
m := newMonolith(cfg)
|
||||
m.setupDatabases()
|
||||
m.setupFederation()
|
||||
m.setupKafka()
|
||||
m.setupRoomServer()
|
||||
m.setupProducers()
|
||||
m.setupNotifiers()
|
||||
m.setupConsumers()
|
||||
m.setupAPIs()
|
||||
httpHandler := common.WrapHandlerInCORS(base.APIMux)
|
||||
|
||||
// Expose the matrix APIs directly rather than putting them under a /api path.
|
||||
go func() {
|
||||
log.Info("Listening on ", *httpBindAddr)
|
||||
log.Fatal(http.ListenAndServe(*httpBindAddr, common.WrapHandlerInCORS(m.api)))
|
||||
logrus.Info("Listening on ", *httpBindAddr)
|
||||
logrus.Fatal(http.ListenAndServe(*httpBindAddr, httpHandler))
|
||||
}()
|
||||
// Handle HTTPS if certificate and key are provided
|
||||
go func() {
|
||||
if *certFile != "" && *keyFile != "" {
|
||||
log.Info("Listening on ", *httpsBindAddr)
|
||||
log.Fatal(http.ListenAndServeTLS(*httpsBindAddr, *certFile, *keyFile, m.api))
|
||||
logrus.Info("Listening on ", *httpsBindAddr)
|
||||
logrus.Fatal(http.ListenAndServeTLS(*httpsBindAddr, *certFile, *keyFile, httpHandler))
|
||||
}
|
||||
}()
|
||||
|
||||
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
|
||||
select {}
|
||||
}
|
||||
|
||||
// A monolith contains all the dendrite components.
|
||||
// Some of the setup functions depend on previous setup functions, so they must
|
||||
// be called in the same order as they are defined in the file.
|
||||
type monolith struct {
|
||||
cfg *config.Dendrite
|
||||
api *mux.Router
|
||||
|
||||
roomServerDB *roomserver_storage.Database
|
||||
accountDB *accounts.Database
|
||||
deviceDB *devices.Database
|
||||
keyDB *keydb.Database
|
||||
mediaAPIDB *mediaapi_storage.Database
|
||||
syncAPIDB *syncapi_storage.SyncServerDatabase
|
||||
federationSenderDB *federationsender_storage.Database
|
||||
publicRoomsAPIDB *publicroomsapi_storage.PublicRoomsServerDatabase
|
||||
|
||||
federation *gomatrixserverlib.FederationClient
|
||||
keyRing gomatrixserverlib.KeyRing
|
||||
|
||||
inputAPI *roomserver_input.RoomserverInputAPI
|
||||
queryAPI *roomserver_query.RoomserverQueryAPI
|
||||
aliasAPI *roomserver_alias.RoomserverAliasAPI
|
||||
|
||||
naffka *naffka.Naffka
|
||||
kafkaProducer sarama.SyncProducer
|
||||
|
||||
roomServerProducer *producers.RoomserverProducer
|
||||
userUpdateProducer *producers.UserUpdateProducer
|
||||
syncProducer *producers.SyncAPIProducer
|
||||
|
||||
syncAPINotifier *syncapi_sync.Notifier
|
||||
}
|
||||
|
||||
func newMonolith(cfg *config.Dendrite) *monolith {
|
||||
return &monolith{cfg: cfg, api: mux.NewRouter()}
|
||||
}
|
||||
|
||||
func (m *monolith) setupDatabases() {
|
||||
var err error
|
||||
m.roomServerDB, err = roomserver_storage.Open(string(m.cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
m.accountDB, err = accounts.NewDatabase(string(m.cfg.Database.Account), m.cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", m.cfg.Database.Account, err.Error())
|
||||
}
|
||||
m.deviceDB, err = devices.NewDatabase(string(m.cfg.Database.Device), m.cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup device database(%q): %s", m.cfg.Database.Device, err.Error())
|
||||
}
|
||||
m.keyDB, err = keydb.NewDatabase(string(m.cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", m.cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
m.mediaAPIDB, err = mediaapi_storage.Open(string(m.cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup sync api database(%q): %s", m.cfg.Database.MediaAPI, err.Error())
|
||||
}
|
||||
m.syncAPIDB, err = syncapi_storage.NewSyncServerDatabase(string(m.cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup sync api database(%q): %s", m.cfg.Database.SyncAPI, err.Error())
|
||||
}
|
||||
m.federationSenderDB, err = federationsender_storage.NewDatabase(string(m.cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create federation sender database with data source %s : %s", m.cfg.Database.FederationSender, err)
|
||||
}
|
||||
m.publicRoomsAPIDB, err = publicroomsapi_storage.NewPublicRoomsServerDatabase(string(m.cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to setup public rooms api database with data source %s : %s", m.cfg.Database.PublicRoomsAPI, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupFederation() {
|
||||
m.federation = gomatrixserverlib.NewFederationClient(
|
||||
m.cfg.Matrix.ServerName, m.cfg.Matrix.KeyID, m.cfg.Matrix.PrivateKey,
|
||||
)
|
||||
|
||||
m.keyRing = keydb.CreateKeyRing(m.federation.Client, m.keyDB)
|
||||
}
|
||||
|
||||
func (m *monolith) setupKafka() {
|
||||
if m.cfg.Kafka.UseNaffka {
|
||||
db, err := sql.Open("postgres", string(m.cfg.Database.Naffka))
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to open naffka database")
|
||||
}
|
||||
|
||||
naffkaDB, err := naffka.NewPostgresqlDatabase(db)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to setup naffka database")
|
||||
}
|
||||
|
||||
naff, err := naffka.New(naffkaDB)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to setup naffka")
|
||||
}
|
||||
m.naffka = naff
|
||||
m.kafkaProducer = naff
|
||||
} else {
|
||||
var err error
|
||||
m.kafkaProducer, err = sarama.NewSyncProducer(m.cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": m.cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka producers")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) kafkaConsumer() sarama.Consumer {
|
||||
if m.cfg.Kafka.UseNaffka {
|
||||
return m.naffka
|
||||
}
|
||||
consumer, err := sarama.NewConsumer(m.cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": m.cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
return consumer
|
||||
}
|
||||
|
||||
func (m *monolith) setupRoomServer() {
|
||||
m.inputAPI = &roomserver_input.RoomserverInputAPI{
|
||||
DB: m.roomServerDB,
|
||||
Producer: m.kafkaProducer,
|
||||
OutputRoomEventTopic: string(m.cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
m.queryAPI = &roomserver_query.RoomserverQueryAPI{
|
||||
DB: m.roomServerDB,
|
||||
}
|
||||
|
||||
m.aliasAPI = &roomserver_alias.RoomserverAliasAPI{
|
||||
DB: m.roomServerDB,
|
||||
Cfg: m.cfg,
|
||||
InputAPI: m.inputAPI,
|
||||
QueryAPI: m.queryAPI,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupProducers() {
|
||||
m.roomServerProducer = producers.NewRoomserverProducer(m.inputAPI)
|
||||
m.userUpdateProducer = &producers.UserUpdateProducer{
|
||||
Producer: m.kafkaProducer,
|
||||
Topic: string(m.cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
m.syncProducer = &producers.SyncAPIProducer{
|
||||
Producer: m.kafkaProducer,
|
||||
Topic: string(m.cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupNotifiers() {
|
||||
pos, err := m.syncAPIDB.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to get latest sync stream position : %s", err)
|
||||
}
|
||||
|
||||
m.syncAPINotifier = syncapi_sync.NewNotifier(syncapi_types.StreamPosition(pos))
|
||||
if err = m.syncAPINotifier.Load(context.Background(), m.syncAPIDB); err != nil {
|
||||
log.Panicf("startup: failed to set up notifier: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupConsumers() {
|
||||
var err error
|
||||
|
||||
clientAPIConsumer := clientapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.accountDB, m.queryAPI,
|
||||
)
|
||||
if err = clientAPIConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
syncAPIRoomConsumer := syncapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.syncAPINotifier, m.syncAPIDB, m.queryAPI,
|
||||
)
|
||||
if err = syncAPIRoomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
syncAPIClientConsumer := syncapi_consumers.NewOutputClientDataConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.syncAPINotifier, m.syncAPIDB,
|
||||
)
|
||||
if err = syncAPIClientConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start client API server consumer: %s", err)
|
||||
}
|
||||
|
||||
publicRoomsAPIConsumer := publicroomsapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.publicRoomsAPIDB, m.queryAPI,
|
||||
)
|
||||
if err = publicRoomsAPIConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
federationSenderQueues := queue.NewOutgoingQueues(m.cfg.Matrix.ServerName, m.federation)
|
||||
|
||||
federationSenderRoomConsumer := federationsender_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), federationSenderQueues, m.federationSenderDB, m.queryAPI,
|
||||
)
|
||||
if err = federationSenderRoomConsumer.Start(); err != nil {
|
||||
log.WithError(err).Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupAPIs() {
|
||||
clientapi_routing.Setup(
|
||||
m.api, *m.cfg, m.roomServerProducer,
|
||||
m.queryAPI, m.aliasAPI, m.accountDB, m.deviceDB, m.federation, m.keyRing,
|
||||
m.userUpdateProducer, m.syncProducer,
|
||||
)
|
||||
|
||||
mediaapi_routing.Setup(
|
||||
m.api, m.cfg, m.mediaAPIDB, m.deviceDB, &m.federation.Client,
|
||||
)
|
||||
|
||||
syncapi_routing.Setup(m.api, syncapi_sync.NewRequestPool(
|
||||
m.syncAPIDB, m.syncAPINotifier, m.accountDB,
|
||||
), m.syncAPIDB, m.deviceDB)
|
||||
|
||||
federationapi_routing.Setup(
|
||||
m.api, *m.cfg, m.queryAPI, m.aliasAPI, m.roomServerProducer, m.keyRing, m.federation,
|
||||
m.accountDB,
|
||||
)
|
||||
|
||||
publicroomsapi_routing.Setup(m.api, m.deviceDB, m.publicRoomsAPIDB)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,80 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/consumers"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi"
|
||||
)
|
||||
|
||||
const componentName = "publicroomsapi"
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "PublicRoomsAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB)
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendritePublicRoomsAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewPublicRoomsServerDatabase(string(cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create public rooms server database with data source %s : %s", cfg.Database.PublicRoomsAPI, err)
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create device database with data source %s : %s", cfg.Database.Device, err)
|
||||
}
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, db, queryAPI)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create room server consumer: %s", err)
|
||||
}
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
log.Info("Starting public rooms server on ", cfg.Listen.PublicRoomsAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, deviceDB, db)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.PublicRoomsAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.PublicRoomsAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,87 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
_ "net/http/pprof"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/alias"
|
||||
"github.com/matrix-org/dendrite/roomserver/input"
|
||||
"github.com/matrix-org/dendrite/roomserver/query"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
)
|
||||
|
||||
const componentName = "roomserver"
|
||||
|
||||
var (
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "RoomServerAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
roomserver.SetupRoomServerComponent(base)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteRoomServer")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
db, err := storage.Open(string(cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
kafkaProducer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
inputAPI := input.RoomserverInputAPI{
|
||||
DB: db,
|
||||
Producer: kafkaProducer,
|
||||
OutputRoomEventTopic: string(cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
inputAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
queryAPI := query.RoomserverQueryAPI{DB: db}
|
||||
|
||||
queryAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
aliasAPI := alias.RoomserverAliasAPI{
|
||||
DB: db,
|
||||
Cfg: cfg,
|
||||
InputAPI: &inputAPI,
|
||||
QueryAPI: &queryAPI,
|
||||
}
|
||||
|
||||
aliasAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
// This is deprecated, but prometheus are still arguing on what to replace
|
||||
// it with. Alternatively we could set it up manually.
|
||||
http.DefaultServeMux.Handle("/metrics", prometheus.Handler()) // nolint: staticcheck, megacheck
|
||||
|
||||
log.Info("Started room server on ", cfg.Listen.RoomServer)
|
||||
|
||||
// TODO: Implement clean shutdown.
|
||||
if err := http.ListenAndServe(string(cfg.Listen.RoomServer), nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.RoomServer))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,100 +15,21 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
"github.com/matrix-org/dendrite/syncapi/routing"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage"
|
||||
"github.com/matrix-org/dendrite/syncapi/sync"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/syncapi"
|
||||
)
|
||||
|
||||
const componentName = "syncapi"
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupStdLogging()
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "SyncAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
_, _, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteSyncAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewSyncServerDatabase(string(cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create sync server database with data source %s : %s", cfg.Database.SyncAPI, err)
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create device database with data source %s : %s", cfg.Database.Device, err)
|
||||
}
|
||||
|
||||
adb, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create account database with data source %s : %s", cfg.Database.Account, err)
|
||||
}
|
||||
|
||||
pos, err := db.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to get latest sync stream position : %s", err)
|
||||
}
|
||||
|
||||
n := sync.NewNotifier(types.StreamPosition(pos))
|
||||
if err = n.Load(context.Background(), db); err != nil {
|
||||
log.Panicf("startup: failed to set up notifier: %s", err)
|
||||
}
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, n, db, queryAPI)
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
clientConsumer := consumers.NewOutputClientDataConsumer(cfg, kafkaConsumer, n, db)
|
||||
if err = clientConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start client API server consumer: %s", err)
|
||||
}
|
||||
|
||||
log.Info("Starting sync server on ", cfg.Listen.SyncAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, sync.NewRequestPool(db, n, adb), db, deviceDB)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.SyncAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.SyncAPI))
|
||||
}
|
||||
|
|
|
|||
182
src/github.com/matrix-org/dendrite/common/basecomponent/base.go
Normal file
182
src/github.com/matrix-org/dendrite/common/basecomponent/base.go
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
// Copyright 2017 New Vector Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package basecomponent
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/naffka"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// BaseDendrite is a base for creating new instances of dendrite. It parses
|
||||
// command line flags and config, and exposes methods for creating various
|
||||
// resources. All errors are handled by logging then exiting, so all methods
|
||||
// should only be used during start up.
|
||||
// Must be closed when shutting down.
|
||||
type BaseDendrite struct {
|
||||
componentName string
|
||||
tracerCloser io.Closer
|
||||
|
||||
// APIMux should be used to register new public matrix api endpoints
|
||||
APIMux *mux.Router
|
||||
Cfg *config.Dendrite
|
||||
KafkaConsumer sarama.Consumer
|
||||
KafkaProducer sarama.SyncProducer
|
||||
}
|
||||
|
||||
// NewBaseDendrite creates a new instance to be used by a component.
|
||||
// The componentName is used for logging purposes, and should be a friendly name
|
||||
// of the compontent running, e.g. "SyncAPI"
|
||||
func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
|
||||
common.SetupStdLogging()
|
||||
common.SetupHookLogging(cfg.Logging, componentName)
|
||||
|
||||
closer, err := cfg.SetupTracing("Dendrite" + componentName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start opentracing")
|
||||
}
|
||||
|
||||
kafkaConsumer, kafkaProducer := setupKafka(cfg)
|
||||
|
||||
return &BaseDendrite{
|
||||
componentName: componentName,
|
||||
tracerCloser: closer,
|
||||
Cfg: cfg,
|
||||
APIMux: mux.NewRouter(),
|
||||
KafkaConsumer: kafkaConsumer,
|
||||
KafkaProducer: kafkaProducer,
|
||||
}
|
||||
}
|
||||
|
||||
// Close implements io.Closer
|
||||
func (b *BaseDendrite) Close() error {
|
||||
return b.tracerCloser.Close()
|
||||
}
|
||||
|
||||
// CreateHTTPRoomserverAPIs returns the AliasAPI, InputAPI and QueryAPI to hit
|
||||
// the roomserver over HTTP.
|
||||
func (b *BaseDendrite) CreateHTTPRoomserverAPIs() (api.RoomserverAliasAPI, api.RoomserverInputAPI, api.RoomserverQueryAPI) {
|
||||
alias := api.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
input := api.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
query := api.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
return alias, input, query
|
||||
}
|
||||
|
||||
// CreateDeviceDB creates a new instance of the device database. Should only be
|
||||
// called once per component.
|
||||
func (b *BaseDendrite) CreateDeviceDB() *devices.Database {
|
||||
db, err := devices.NewDatabase(string(b.Cfg.Database.Device), b.Cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to devices db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateAccountsDB creates a new instance of the accounts database. Should only
|
||||
// be called once per component.
|
||||
func (b *BaseDendrite) CreateAccountsDB() *accounts.Database {
|
||||
db, err := accounts.NewDatabase(string(b.Cfg.Database.Account), b.Cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to accounts db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateKeyDB creates a new instance of the key database. Should only be called
|
||||
// once per component.
|
||||
func (b *BaseDendrite) CreateKeyDB() *keydb.Database {
|
||||
db, err := keydb.NewDatabase(string(b.Cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to keys db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateFederationClient creates a new federation client. Should only be called
|
||||
// once per component.
|
||||
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
|
||||
return gomatrixserverlib.NewFederationClient(
|
||||
b.Cfg.Matrix.ServerName, b.Cfg.Matrix.KeyID, b.Cfg.Matrix.PrivateKey,
|
||||
)
|
||||
}
|
||||
|
||||
// SetupAndServeHTTP sets up the HTTP server to serve endpoints registered on
|
||||
// ApiMux under /api/ and adds a prometheus handler under /metrics.
|
||||
func (b *BaseDendrite) SetupAndServeHTTP(addr string) {
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(b.APIMux))
|
||||
|
||||
logrus.Infof("Starting %s server on %s", b.componentName, addr)
|
||||
|
||||
err := http.ListenAndServe(addr, nil)
|
||||
|
||||
if err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve http")
|
||||
}
|
||||
|
||||
logrus.Infof("Stopped %s server on %s", b.componentName, addr)
|
||||
}
|
||||
|
||||
// setupKafka creates kafka consumer/producer pair from the config. Checks if
|
||||
// should use naffka.
|
||||
func setupKafka(cfg *config.Dendrite) (sarama.Consumer, sarama.SyncProducer) {
|
||||
if cfg.Kafka.UseNaffka {
|
||||
db, err := sql.Open("postgres", string(cfg.Database.Naffka))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to open naffka database")
|
||||
}
|
||||
|
||||
naffkaDB, err := naffka.NewPostgresqlDatabase(db)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka database")
|
||||
}
|
||||
|
||||
naff, err := naffka.New(naffkaDB)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka")
|
||||
}
|
||||
|
||||
return naff, naff
|
||||
}
|
||||
|
||||
consumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to start kafka consumer")
|
||||
}
|
||||
|
||||
producer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to setup kafka producers")
|
||||
}
|
||||
|
||||
return consumer, producer
|
||||
}
|
||||
|
|
@ -0,0 +1,61 @@
|
|||
// Copyright 2017 New Vector Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package basecomponent
|
||||
|
||||
import (
|
||||
"flag"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
// ParseFlags parses the commandline flags and uses them to create a config.
|
||||
// If running as a monolith use `ParseMonolithFlags` instead.
|
||||
func ParseFlags() *config.Dendrite {
|
||||
flag.Parse()
|
||||
|
||||
if *configPath == "" {
|
||||
logrus.Fatal("--config must be supplied")
|
||||
}
|
||||
|
||||
cfg, err := config.Load(*configPath)
|
||||
|
||||
if err != nil {
|
||||
logrus.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
||||
// ParseMonolithFlags parses the commandline flags and uses them to create a
|
||||
// config. Should only be used if running a monolith. See `ParseFlags`.
|
||||
func ParseMonolithFlags() *config.Dendrite {
|
||||
flag.Parse()
|
||||
|
||||
if *configPath == "" {
|
||||
logrus.Fatal("--config must be supplied")
|
||||
}
|
||||
|
||||
cfg, err := config.LoadMonolithic(*configPath)
|
||||
|
||||
if err != nil {
|
||||
logrus.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
|
@ -15,8 +15,11 @@
|
|||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"gopkg.in/yaml.v2"
|
||||
)
|
||||
|
|
@ -28,6 +31,8 @@ type ApplicationServiceNamespace struct {
|
|||
Exclusive bool `yaml:"exclusive"`
|
||||
// A regex pattern that represents the namespace
|
||||
Regex string `yaml:"regex"`
|
||||
// Regex object representing our pattern. Saves having to recompile every time
|
||||
RegexpObject *regexp.Regexp
|
||||
}
|
||||
|
||||
// ApplicationService represents a Matrix application service.
|
||||
|
|
@ -44,11 +49,12 @@ type ApplicationService struct {
|
|||
// Localpart of application service user
|
||||
SenderLocalpart string `yaml:"sender_localpart"`
|
||||
// Information about an application service's namespaces
|
||||
Namespaces map[string][]ApplicationServiceNamespace `yaml:"namespaces"`
|
||||
NamespaceMap map[string][]ApplicationServiceNamespace `yaml:"namespaces"`
|
||||
}
|
||||
|
||||
// loadAppservices iterates through all application service config files
|
||||
// and loads their data into the config object for later access.
|
||||
func loadAppservices(config *Dendrite) error {
|
||||
// Iterate through and return all the Application Services
|
||||
for _, configPath := range config.ApplicationServices.ConfigFiles {
|
||||
// Create a new application service
|
||||
var appservice ApplicationService
|
||||
|
|
@ -75,5 +81,116 @@ func loadAppservices(config *Dendrite) error {
|
|||
config.Derived.ApplicationServices, appservice)
|
||||
}
|
||||
|
||||
// Check for any errors in the loaded application services
|
||||
return checkErrors(config)
|
||||
}
|
||||
|
||||
// setupRegexps will create regex objects for exclusive and non-exclusive
|
||||
// usernames, aliases and rooms of all application services, so that other
|
||||
// methods can quickly check if a particular string matches any of them.
|
||||
func setupRegexps(cfg *Dendrite) {
|
||||
// Combine all exclusive namespaces for later string checking
|
||||
var exclusiveUsernameStrings, exclusiveAliasStrings, exclusiveRoomStrings []string
|
||||
|
||||
// If an application service's regex is marked as exclusive, add
|
||||
// it's contents to the overall exlusive regex string
|
||||
for _, appservice := range cfg.Derived.ApplicationServices {
|
||||
for key, namespaceSlice := range appservice.NamespaceMap {
|
||||
switch key {
|
||||
case "users":
|
||||
appendExclusiveNamespaceRegexs(&exclusiveUsernameStrings, namespaceSlice)
|
||||
case "aliases":
|
||||
appendExclusiveNamespaceRegexs(&exclusiveAliasStrings, namespaceSlice)
|
||||
case "rooms":
|
||||
appendExclusiveNamespaceRegexs(&exclusiveRoomStrings, namespaceSlice)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Join the regexes together into one big regex.
|
||||
// i.e. "app1.*", "app2.*" -> "(app1.*)|(app2.*)"
|
||||
// Later we can check if a username or some other string matches any exclusive
|
||||
// regex and deny access if it isn't from an application service
|
||||
exclusiveUsernames := strings.Join(exclusiveUsernameStrings, "|")
|
||||
|
||||
// If there are no exclusive username regexes, compile string so that it
|
||||
// will not match any valid usernames
|
||||
if exclusiveUsernames == "" {
|
||||
exclusiveUsernames = "^$"
|
||||
}
|
||||
|
||||
// TODO: Aliases and rooms. Needed?
|
||||
//exclusiveAliases := strings.Join(exclusiveAliasStrings, "|")
|
||||
//exclusiveRooms := strings.Join(exclusiveRoomStrings, "|")
|
||||
|
||||
cfg.Derived.ExclusiveApplicationServicesUsernameRegexp, _ = regexp.Compile(exclusiveUsernames)
|
||||
}
|
||||
|
||||
// concatenateExclusiveNamespaces takes a slice of strings and a slice of
|
||||
// namespaces and will append the regexes of only the exclusive namespaces
|
||||
// into the string slice
|
||||
func appendExclusiveNamespaceRegexs(
|
||||
exclusiveStrings *[]string, namespaces []ApplicationServiceNamespace,
|
||||
) {
|
||||
for _, namespace := range namespaces {
|
||||
if namespace.Exclusive {
|
||||
// We append parenthesis to later separate each regex when we compile
|
||||
// i.e. "app1.*", "app2.*" -> "(app1.*)|(app2.*)"
|
||||
*exclusiveStrings = append(*exclusiveStrings, "("+namespace.Regex+")")
|
||||
}
|
||||
|
||||
// Compile this regex into a Regexp object for later use
|
||||
namespace.RegexpObject, _ = regexp.Compile(namespace.Regex)
|
||||
}
|
||||
}
|
||||
|
||||
// checkErrors checks for any configuration errors amongst the loaded
|
||||
// application services according to the application service spec.
|
||||
func checkErrors(config *Dendrite) error {
|
||||
var idMap = make(map[string]bool)
|
||||
var tokenMap = make(map[string]bool)
|
||||
|
||||
// Check that no two application services have the same as_token or id
|
||||
for _, appservice := range config.Derived.ApplicationServices {
|
||||
// Check if we've already seen this ID
|
||||
if idMap[appservice.ID] {
|
||||
return configErrors([]string{fmt.Sprintf(
|
||||
"Application Service ID %s must be unique", appservice.ID,
|
||||
)})
|
||||
}
|
||||
if tokenMap[appservice.ASToken] {
|
||||
return configErrors([]string{fmt.Sprintf(
|
||||
"Application Service Token %s must be unique", appservice.ASToken,
|
||||
)})
|
||||
}
|
||||
|
||||
// Add the id/token to their respective maps if we haven't already
|
||||
// seen them.
|
||||
idMap[appservice.ID] = true
|
||||
tokenMap[appservice.ID] = true
|
||||
}
|
||||
|
||||
// Check that namespace(s) are valid regex
|
||||
for _, appservice := range config.Derived.ApplicationServices {
|
||||
for _, namespaceSlice := range appservice.NamespaceMap {
|
||||
for _, namespace := range namespaceSlice {
|
||||
if !IsValidRegex(namespace.Regex) {
|
||||
return configErrors([]string{fmt.Sprintf(
|
||||
"Invalid regex string for Application Service %s", appservice.ID,
|
||||
)})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
setupRegexps(config)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// IsValidRegex returns true or false based on whether the
|
||||
// given string is valid regex or not
|
||||
func IsValidRegex(regexString string) bool {
|
||||
_, err := regexp.Compile(regexString)
|
||||
|
||||
return err == nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import (
|
|||
"io"
|
||||
"io/ioutil"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
|
|
@ -233,6 +234,13 @@ type Dendrite struct {
|
|||
// Application Services parsed from their config files
|
||||
// The paths of which were given above in the main config file
|
||||
ApplicationServices []ApplicationService
|
||||
|
||||
// A meta-regex compiled from all exclusive Application Service
|
||||
// Regexes. When a user registers, we check that their username
|
||||
// does not match any exclusive Application Service namespaces
|
||||
ExclusiveApplicationServicesUsernameRegexp *regexp.Regexp
|
||||
|
||||
// TODO: Exclusive alias, room regexp's
|
||||
} `yaml:"-"`
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,44 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package federationapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
// TODO: Are we really wanting to pull in the producer from clientapi
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/federationapi/routing"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
// SetupFederationAPIComponent sets up and registers HTTP handlers for the
|
||||
// FederationAPI component.
|
||||
func SetupFederationAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
accountsDB *accounts.Database,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
keyRing *gomatrixserverlib.KeyRing,
|
||||
aliasAPI api.RoomserverAliasAPI,
|
||||
inputAPI api.RoomserverInputAPI,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, *base.Cfg, queryAPI, aliasAPI,
|
||||
roomserverProducer, *keyRing, federation, accountsDB,
|
||||
)
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package federationsender
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
"github.com/matrix-org/dendrite/federationsender/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupFederationSenderComponent sets up and registers HTTP handlers for the
|
||||
// FederationSender component.
|
||||
func SetupFederationSenderComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
federationSenderDB, err := storage.NewDatabase(string(base.Cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to connect to federation sender db")
|
||||
}
|
||||
|
||||
queues := queue.NewOutgoingQueues(base.Cfg.Matrix.ServerName, federation)
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, queues,
|
||||
federationSenderDB, queryAPI,
|
||||
)
|
||||
if err = consumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panic("failed to start room server consumer")
|
||||
}
|
||||
}
|
||||
40
src/github.com/matrix-org/dendrite/mediaapi/mediaapi.go
Normal file
40
src/github.com/matrix-org/dendrite/mediaapi/mediaapi.go
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package mediaapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
"github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupMediaAPIComponent sets up and registers HTTP handlers for the MediaAPI
|
||||
// component.
|
||||
func SetupMediaAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
) {
|
||||
mediaDB, err := storage.Open(string(base.Cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to media db")
|
||||
}
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, base.Cfg, mediaDB, deviceDB, gomatrixserverlib.NewClient(),
|
||||
)
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package publicroomsapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupPublicRoomsAPIComponent sets up and registers HTTP handlers for the PublicRoomsAPI
|
||||
// component.
|
||||
func SetupPublicRoomsAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
) {
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
|
||||
routing.Setup(base.APIMux, deviceDB, publicRoomsDB)
|
||||
}
|
||||
64
src/github.com/matrix-org/dendrite/roomserver/roomserver.go
Normal file
64
src/github.com/matrix-org/dendrite/roomserver/roomserver.go
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package roomserver
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/alias"
|
||||
"github.com/matrix-org/dendrite/roomserver/input"
|
||||
"github.com/matrix-org/dendrite/roomserver/query"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupRoomServerComponent sets up and registers HTTP handlers for the
|
||||
// RoomServer component. Returns instances of the various roomserver APIs,
|
||||
// allowing other components running in the same process to hit the query the
|
||||
// APIs directly instead of having to use HTTP.
|
||||
func SetupRoomServerComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
) (api.RoomserverAliasAPI, api.RoomserverInputAPI, api.RoomserverQueryAPI) {
|
||||
roomserverDB, err := storage.Open(string(base.Cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to room server db")
|
||||
}
|
||||
|
||||
inputAPI := input.RoomserverInputAPI{
|
||||
DB: roomserverDB,
|
||||
Producer: base.KafkaProducer,
|
||||
OutputRoomEventTopic: string(base.Cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
inputAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
queryAPI := query.RoomserverQueryAPI{DB: roomserverDB}
|
||||
|
||||
queryAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
aliasAPI := alias.RoomserverAliasAPI{
|
||||
DB: roomserverDB,
|
||||
Cfg: base.Cfg,
|
||||
InputAPI: &inputAPI,
|
||||
QueryAPI: &queryAPI,
|
||||
}
|
||||
|
||||
aliasAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
return &aliasAPI, &inputAPI, &queryAPI
|
||||
}
|
||||
|
|
@ -65,7 +65,7 @@ const selectEventsSQL = "" +
|
|||
const selectRecentEventsSQL = "" +
|
||||
"SELECT id, event_json, device_id, transaction_id FROM syncapi_output_room_events" +
|
||||
" WHERE room_id = $1 AND id > $2 AND id <= $3" +
|
||||
" ORDER BY id DESC LIMIT $4"
|
||||
" ORDER BY id ASC LIMIT $4"
|
||||
|
||||
const selectMaxEventIDSQL = "" +
|
||||
"SELECT MAX(id) FROM syncapi_output_room_events"
|
||||
|
|
@ -234,9 +234,7 @@ func (s *outputRoomEventsStatements) selectRecentEvents(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// reverse the order because [0] is the newest event due to the ORDER BY in SQL-land. The reverse order makes [0] the oldest event,
|
||||
// which is correct for /sync responses.
|
||||
return reverseEvents(events), nil
|
||||
return events, nil
|
||||
}
|
||||
|
||||
// Events returns the events for the given event IDs. Returns an error if any one of the event IDs given are missing
|
||||
|
|
@ -287,10 +285,3 @@ func rowsToStreamEvents(rows *sql.Rows) ([]streamEvent, error) {
|
|||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func reverseEvents(input []streamEvent) (output []streamEvent) {
|
||||
for i := len(input) - 1; i >= 0; i-- {
|
||||
output = append(output, input[i])
|
||||
}
|
||||
return
|
||||
}
|
||||
|
|
|
|||
75
src/github.com/matrix-org/dendrite/syncapi/syncapi.go
Normal file
75
src/github.com/matrix-org/dendrite/syncapi/syncapi.go
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package syncapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
"github.com/matrix-org/dendrite/syncapi/routing"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage"
|
||||
"github.com/matrix-org/dendrite/syncapi/sync"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
)
|
||||
|
||||
// SetupSyncAPIComponent sets up and registers HTTP handlers for the SyncAPI
|
||||
// component.
|
||||
func SetupSyncAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
accountsDB *accounts.Database,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
syncDB, err := storage.NewSyncServerDatabase(string(base.Cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to sync db")
|
||||
}
|
||||
|
||||
pos, err := syncDB.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to get stream position")
|
||||
}
|
||||
|
||||
notifier := sync.NewNotifier(types.StreamPosition(pos))
|
||||
err = notifier.Load(context.Background(), syncDB)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start notifier")
|
||||
}
|
||||
|
||||
requestPool := sync.NewRequestPool(syncDB, notifier, accountsDB)
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, notifier, syncDB, queryAPI,
|
||||
)
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start room server consumer")
|
||||
}
|
||||
|
||||
clientConsumer := consumers.NewOutputClientDataConsumer(
|
||||
base.Cfg, base.KafkaConsumer, notifier, syncDB,
|
||||
)
|
||||
if err = clientConsumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start client data consumer")
|
||||
}
|
||||
|
||||
routing.Setup(base.APIMux, requestPool, syncDB, deviceDB)
|
||||
}
|
||||
Loading…
Reference in a new issue