diff options
Diffstat (limited to 'synapse')
| -rwxr-xr-x | synapse/app/homeserver.py | 89 | ||||
| -rw-r--r-- | synapse/config/__init__.py | 14 | ||||
| -rw-r--r-- | synapse/config/_base.py | 99 | ||||
| -rw-r--r-- | synapse/config/database.py | 36 | ||||
| -rw-r--r-- | synapse/config/homeserver.py | 26 | ||||
| -rw-r--r-- | synapse/config/logger.py | 67 | ||||
| -rw-r--r-- | synapse/config/server.py | 75 | ||||
| -rw-r--r-- | synapse/config/tls.py | 106 | ||||
| -rw-r--r-- | synapse/crypto/config.py | 160 | ||||
| -rw-r--r-- | synapse/federation/replication.py | 5 | ||||
| -rw-r--r-- | synapse/handlers/events.py | 6 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 2 | ||||
| -rw-r--r-- | synapse/handlers/message.py | 7 | ||||
| -rw-r--r-- | synapse/handlers/presence.py | 228 | ||||
| -rw-r--r-- | synapse/notifier.py | 1 | ||||
| -rw-r--r-- | synapse/storage/stream.py | 9 | ||||
| -rw-r--r-- | synapse/streams/config.py | 2 | ||||
| -rw-r--r-- | synapse/util/logutils.py | 53 |
18 files changed, 652 insertions, 333 deletions
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 6d292ccf9a..f56dde846d 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -29,6 +29,7 @@ from synapse.http.client import TwistedHttpClient from synapse.api.urls import ( CLIENT_PREFIX, FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX ) +from synapse.config.homeserver import HomeServerConfig from daemonize import Daemonize import twisted.manhole.telnet @@ -211,32 +212,7 @@ class SynapseHomeServer(HomeServer): logger.info("Synapse now listening on port %d", port) -def setup_logging(verbosity=0, filename=None, config_path=None): - """ Sets up logging with verbosity levels. - Args: - verbosity: The verbosity level. - filename: Log to the given file rather than to the console. - config_path: Path to a python logging config file. - """ - - if config_path is None: - log_format = ( - '%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s' - ) - - level = logging.INFO - if verbosity: - level = logging.DEBUG - - # FIXME: we need a logging.WARN for a -q quiet option - - logging.basicConfig(level=level, filename=filename, format=log_format) - else: - logging.config.fileConfig(config_path) - - observer = PythonLoggingObserver() - observer.start() def run(): @@ -244,78 +220,49 @@ def run(): def setup(): - parser = argparse.ArgumentParser() - parser.add_argument("-p", "--port", dest="port", type=int, default=8080, - help="The port to listen on.") - parser.add_argument("-d", "--database", dest="db", default="homeserver.db", - help="The database name.") - parser.add_argument("-H", "--host", dest="host", default="localhost", - help="The hostname of the server.") - parser.add_argument('-v', '--verbose', dest="verbose", action='count', - help="The verbosity level.") - parser.add_argument('-f', '--log-file', dest="log_file", default=None, - help="File to log to.") - parser.add_argument('--log-config', dest="log_config", default=None, - help="Python logging config") - parser.add_argument('-D', '--daemonize', action='store_true', - default=False, help="Daemonize the home server") - parser.add_argument('--pid-file', dest="pid", help="When running as a " - "daemon, the file to store the pid in", - default="hs.pid") - parser.add_argument("-W", "--webclient", dest="webclient", default=True, - action="store_false", help="Don't host a web client.") - parser.add_argument("--manhole", dest="manhole", type=int, default=None, - help="Turn on the twisted telnet manhole service.") - args = parser.parse_args() - - verbosity = int(args.verbose) if args.verbose else None - - # Because if/when we daemonize we change to root dir. - db_name = os.path.abspath(args.db) - log_file = args.log_file - if log_file: - log_file = os.path.abspath(log_file) - - setup_logging( + config = HomeServerConfig.load_config("Synapse Homeserver", sys.argv[1:]) + + config.setup_logging( verbosity=verbosity, filename=log_file, config_path=args.log_config, ) - logger.info("Server hostname: %s", args.host) + logger.info("Server hostname: %s", config.server_name) - if re.search(":[0-9]+$", args.host): - domain_with_port = args.host + if re.search(":[0-9]+$", config.server_name): + domain_with_port = config.server_name else: - domain_with_port = "%s:%s" % (args.host, args.port) + domain_with_port = "%s:%s" % (args.server_name, config.bind_port) hs = SynapseHomeServer( - args.host, + config.server_name, domain_with_port=domain_with_port, upload_dir=os.path.abspath("uploads"), - db_name=db_name, + db_name=config.database_path, ) hs.register_servlets() hs.create_resource_tree( - web_client=args.webclient, - redirect_root_to_web_client=True) - hs.start_listening(args.port) + web_client=config.webclient, + redirect_root_to_web_client=True, + ) + hs.start_listening(config.bind_port) hs.get_db_pool() - if args.manhole: + if config.manhole: f = twisted.manhole.telnet.ShellFactory() f.username = "matrix" f.password = "rabbithole" f.namespace['hs'] = hs - reactor.listenTCP(args.manhole, f, interface='127.0.0.1') + reactor.listenTCP(config.manhole, f, interface='127.0.0.1') - if args.daemonize: + if config.daemonize: daemon = Daemonize( app="synapse-homeserver", - pid=args.pid, + pid=config.pid_file, action=run, auto_close_fds=False, verbose=True, diff --git a/synapse/config/__init__.py b/synapse/config/__init__.py new file mode 100644 index 0000000000..fe8a073cd3 --- /dev/null +++ b/synapse/config/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. diff --git a/synapse/config/_base.py b/synapse/config/_base.py new file mode 100644 index 0000000000..b4cf0262f4 --- /dev/null +++ b/synapse/config/_base.py @@ -0,0 +1,99 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + + +import ConfigParser as configparser +import argparse +import sys +import os + + +class Config(object): + def __init__(self, args): + pass + + @staticmethod + def read_file(file_path): + with open(file_path) as file_stream: + return file_stream.read() + + @staticmethod + def read_config_file(file_path): + config = configparser.SafeConfigParser() + config.read([file_path]) + config_dict = {} + for section in config.sections(): + config_dict.update(config.items(section)) + return config_dict + + @classmethod + def add_arguments(cls, parser): + pass + + @classmethod + def generate_config(cls, args, config_dir_path): + pass + + @classmethod + def load_config(cls, description, argv, generate_section=None): + config_parser = argparse.ArgumentParser(add_help=False) + config_parser.add_argument( + "-c", "--config-path", + metavar="CONFIG_FILE", + help="Specify config file" + ) + config_args, remaining_args = config_parser.parse_known_args(argv) + + if generate_section: + if not config_args.config_path: + config_parser.error( + "Must specify where to generate the config file" + ) + config_dir_path = os.path.dirname(config_args.config_path) + if os.path.exists(config_args.config_path): + defaults = cls.read_config_file(config_args.config_path) + else: + if config_args.config_path: + defaults = cls.read_config_file(config_args.config_path) + else: + defaults = {} + + parser = argparse.ArgumentParser( + parents=[config_parser], + description=description, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.set_defaults(**defaults) + + + cls.add_arguments(parser) + args = parser.parse_args(remaining_args) + + if generate_section: + config_dir_path = os.path.dirname(config_args.config_path) + config_dir_path = os.path.abspath(config_dir_path) + cls.generate_config(args, config_dir_path) + config = configparser.SafeConfigParser() + config.add_section(generate_section) + for key, value in vars(args).items(): + if key != "config_path" and value is not None: + config.set(generate_section, key, str(value)) + with open(config_args.config_path, "w") as config_file: + config.write(config_file) + + return cls(args) + + + diff --git a/synapse/config/database.py b/synapse/config/database.py new file mode 100644 index 0000000000..43f54be437 --- /dev/null +++ b/synapse/config/database.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + +from ._base import Config +import os + +class DatabaseConfig(Config): + def __init__(self, args): + self.db_path = os.path.abspath(args.database_path) + + @classmethod + def add_arguments(cls, parser): + super(DatabaseConfig, cls).add_arguments(parser) + db_group = parser.add_argument_group("database") + db_group.add_argument( + "-d", "--database", dest="database_path", default="homeserver.db", + help="The database name." + ) + + @classmethod + def generate_config(cls, args, config_dir_path): + super(DatabaseConfig, cls).generate_config(args, config_dir_path) + args.database_path = os.path.abspath(args.database_path) + diff --git a/synapse/config/homeserver.py b/synapse/config/homeserver.py new file mode 100644 index 0000000000..18072e3196 --- /dev/null +++ b/synapse/config/homeserver.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + +from .tls import TlsConfig +from .server import ServerConfig +from .logger import LoggingConfig +from .database import DatabaseConfig + +class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig): + pass + +if __name__=='__main__': + import sys + HomeServerConfig.load_config("Generate config", sys.argv[1:], "HomeServer") diff --git a/synapse/config/logger.py b/synapse/config/logger.py new file mode 100644 index 0000000000..d34532c41a --- /dev/null +++ b/synapse/config/logger.py @@ -0,0 +1,67 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + +from ._base import Config + +from twisted.python.log import PythonLoggingObserver +import logging +import logging.config +import os + +class LoggingConfig(Config): + def __init__(self, args): + self.verbosity = int(args.verbose) if args.verbose else None + self.log_config = os.path.abspath(args.log_config) + self.log_file = os.path.abspath(args.log_file) + + @classmethod + def add_arguments(cls, parser): + super(LoggingConfig, cls).add_arguments(parser) + logging_group = parser.add_argument_group("logging") + logging_group.add_argument( + '-v', '--verbose', dest="verbose", action='count', + help="The verbosity level." + ) + logging_group.add_argument( + '-f', '--log-file', dest="log_file", default=None, + help="File to log to." + ) + logging_group.add_argument( + '--log-config', dest="log_config", default=None, + help="Python logging config file" + ) + + def setup_logging(self): + log_format = ( + '%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s' + ) + if self.config_path is None: + + level = logging.INFO + if verbosity: + level = logging.DEBUG + + # FIXME: we need a logging.WARN for a -q quiet option + + logging.basicConfig( + level=level, + filename=filename, + format=log_format + ) + else: + logging.config.fileConfig(config_path) + + observer = PythonLoggingObserver() + observer.start() diff --git a/synapse/config/server.py b/synapse/config/server.py new file mode 100644 index 0000000000..4a656b06ab --- /dev/null +++ b/synapse/config/server.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + +import nacl.signing +import socket +import os +from ._base import Config +from syutil.base64util import encode_base64, decode_base64 + + +class ServerConfig(Config): + def __init__(self, args): + super(ServerConfig, self).__init__(args) + self.server_name = args.server_name + self.signing_key = self.read_signing_key(args.signing_key_path) + self.bind_port = args.bind_port + self.bind_host = args.bind_host + self.daemonize = args.daemonize + self.pid_file = os.path.abspath(args.pid_file) + + @classmethod + def add_arguments(cls, parser): + super(ServerConfig, cls).add_arguments(parser) + server_group = parser.add_argument_group("server") + server_group.add_argument("-H", "--server-name", default="localhost", + help="The name of the server") + server_group.add_argument("--signing-key-path", + help="The signing key to sign messages with") + server_group.add_argument("-p", "--bind-port", type=int, + help="TCP port to listen on") + server_group.add_argument("--bind-host", default="", + help="Local interface to listen on") + server_group.add_argument("-D", "--daemonize", action='store_true', + help="Daemonize the home server") + server_group.add_argument('--pid-file', default = "hs.pid", + help="When running as a daemon, the file to" + " store the pid in") + server_group.add_argument("-W", "--no-webclient", dest="webclient", + default=True, action="store_false", + help="Don't host a web client.") + server_group.add_argument("--manhole", dest="manhole", type=int, + help="Turn on the twisted telnet manhole" + " service on the given port.") + + def read_signing_key(self, signing_key_path): + signing_key_base64 = self.read_file(signing_key_path) + signing_key_bytes = decode_base64(signing_key_base64) + return nacl.signing.SigningKey(signing_key_bytes) + + @classmethod + def generate_config(cls, args, config_dir_path): + super(ServerConfig, cls).generate_config(args, config_dir_path) + base_key_name = os.path.join(config_dir_path, args.server_name) + + args.pid_file = os.path.abspath(args.pid_file) + + if not args.signing_key_path: + args.signing_key_path = base_key_name + ".signing.key" + + if not os.path.exists(args.signing_key_path): + with open(args.signing_key_path, "w") as signing_key_file: + key = nacl.signing.SigningKey.generate() + signing_key_file.write(encode_base64(key.encode())) diff --git a/synapse/config/tls.py b/synapse/config/tls.py new file mode 100644 index 0000000000..c65487ceb9 --- /dev/null +++ b/synapse/config/tls.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 matrix.org +# +# 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. + +from ._base import Config + +from OpenSSL import crypto +import subprocess +import os + +class TlsConfig(Config): + def __init__(self, args): + super(TlsConfig, self).__init__(args) + self.tls_certificate = self.read_tls_certificate( + args.tls_certificate_path + ) + self.tls_private_key = self.read_tls_private_key( + args.tls_private_key_path + ) + self.tls_dh_params_path = args.tls_dh_params_path + + @classmethod + def add_arguments(cls, parser): + super(TlsConfig, cls).add_arguments(parser) + tls_group = parser.add_argument_group("tls") + tls_group.add_argument("--tls-certificate-path", + help="PEM encoded X509 certificate for TLS") + tls_group.add_argument("--tls-private-key-path", + help="PEM encoded private key for TLS") + tls_group.add_argument("--tls-dh-params-path", + help="PEM dh parameters for ephemeral keys") + + def read_tls_certificate(self, cert_path): + cert_pem = self.read_file(cert_path) + return crypto.load_certificate(crypto.FILETYPE_PEM, cert_pem) + + def read_tls_private_key(self, private_key_path): + private_key_pem = self.read_file(private_key_path) + return crypto.load_privatekey(crypto.FILETYPE_PEM, private_key_pem) + + @classmethod + def generate_config(cls, args, config_dir_path): + super(TlsConfig, cls).generate_config(args, config_dir_path) + base_key_name = os.path.join(config_dir_path, args.server_name) + + if args.tls_certificate_path is None: + args.tls_certificate_path = base_key_name + ".tls.crt" + + if args.tls_private_key_path is None: + args.tls_private_key_path = base_key_name + ".tls.key" + + if args.tls_dh_params_path is None: + args.tls_dh_params_path = base_key_name + ".tls.dh" + + if not os.path.exists(args.tls_private_key_path): + with open(args.tls_private_key_path, "w") as private_key_file: + tls_private_key = crypto.PKey() + tls_private_key.generate_key(crypto.TYPE_RSA, 2048) + private_key_pem = crypto.dump_privatekey( + crypto.FILETYPE_PEM, tls_private_key + ) + private_key_file.write(private_key_pem) + else: + with open(args.tls_private_key_path) as private_key_file: + private_key_pem = private_key_file.read() + tls_private_key = crypto.load_privatekey( + crypto.FILETYPE_PEM, private_key_pem + ) + + if not os.path.exists(args.tls_certificate_path): + with open(args.tls_certificate_path, "w") as certifcate_file: + cert = crypto.X509() + subject = cert.get_subject() + subject.CN = args.server_name + + cert.set_serial_number(1000) + cert.gmtime_adj_notBefore(0) + cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) + cert.set_issuer(cert.get_subject()) + cert.set_pubkey(tls_private_key) + + cert.sign(tls_private_key, 'sha256') + + cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) + + certifcate_file.write(cert_pem) + + if not os.path.exists(args.tls_dh_params_path): + subprocess.check_call([ + "openssl", "dhparam", + "-outform", "PEM", + "-out", args.tls_dh_params_path, + "2048" + ]) + diff --git a/synapse/crypto/config.py b/synapse/crypto/config.py deleted file mode 100644 index 2330133e71..0000000000 --- a/synapse/crypto/config.py +++ /dev/null @@ -1,160 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 matrix.org -# -# 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. - - -import ConfigParser as configparser -import argparse -import socket -import sys -import os -from OpenSSL import crypto -import nacl.signing -from syutil.base64util import encode_base64 -import subprocess - - -def load_config(description, argv): - config_parser = argparse.ArgumentParser(add_help=False) - config_parser.add_argument("-c", "--config-path", metavar="CONFIG_FILE", - help="Specify config file") - config_args, remaining_args = config_parser.parse_known_args(argv) - if config_args.config_path: - config = configparser.SafeConfigParser() - config.read([config_args.config_path]) - defaults = dict(config.items("KeyServer")) - else: - defaults = {} - parser = argparse.ArgumentParser( - parents=[config_parser], - description=description, - formatter_class=argparse.RawDescriptionHelpFormatter, - ) - parser.set_defaults(**defaults) - parser.add_argument("--server-name", default=socket.getfqdn(), - help="The name of the server") - parser.add_argument("--signing-key-path", - help="The signing key to sign responses with") - parser.add_argument("--tls-certificate-path", - help="PEM encoded X509 certificate for TLS") - parser.add_argument("--tls-private-key-path", - help="PEM encoded private key for TLS") - parser.add_argument("--tls-dh-params-path", - help="PEM encoded dh parameters for ephemeral keys") - parser.add_argument("--bind-port", type=int, - help="TCP port to listen on") - parser.add_argument("--bind-host", default="", - help="Local interface to listen on") - - args = parser.parse_args(remaining_args) - - server_config = vars(args) - del server_config["config_path"] - return server_config - - -def generate_config(argv): - parser = argparse.ArgumentParser() - parser.add_argument("-c", "--config-path", help="Specify config file", - metavar="CONFIG_FILE", required=True) - parser.add_argument("--server-name", default=socket.getfqdn(), - help="The name of the server") - parser.add_argument("--signing-key-path", - help="The signing key to sign responses with") - parser.add_argument("--tls-certificate-path", - help="PEM encoded X509 certificate for TLS") - parser.add_argument("--tls-private-key-path", - help="PEM encoded private key for TLS") - parser.add_argument("--tls-dh-params-path", - help="PEM encoded dh parameters for ephemeral keys") - parser.add_argument("--bind-port", type=int, required=True, - help="TCP port to listen on") - parser.add_argument("--bind-host", default="", - help="Local interface to listen on") - - args = parser.parse_args(argv) - - dir_name = os.path.dirname(args.config_path) - base_key_name = os.path.join(dir_name, args.server_name) - - if args.signing_key_path is None: - args.signing_key_path = base_key_name + ".signing.key" - - if args.tls_certificate_path is None: - args.tls_certificate_path = base_key_name + ".tls.crt" - - if args.tls_private_key_path is None: - args.tls_private_key_path = base_key_name + ".tls.key" - - if args.tls_dh_params_path is None: - args.tls_dh_params_path = base_key_name + ".tls.dh" - - if not os.path.exists(args.signing_key_path): - with open(args.signing_key_path, "w") as signing_key_file: - key = nacl.signing.SigningKey.generate() - signing_key_file.write(encode_base64(key.encode())) - - if not os.path.exists(args.tls_private_key_path): - with open(args.tls_private_key_path, "w") as private_key_file: - tls_private_key = crypto.PKey() - tls_private_key.generate_key(crypto.TYPE_RSA, 2048) - private_key_pem = crypto.dump_privatekey( - crypto.FILETYPE_PEM, tls_private_key - ) - private_key_file.write(private_key_pem) - else: - with open(args.tls_private_key_path) as private_key_file: - private_key_pem = private_key_file.read() - tls_private_key = crypto.load_privatekey( - crypto.FILETYPE_PEM, private_key_pem - ) - - if not os.path.exists(args.tls_certificate_path): - with open(args.tls_certificate_path, "w") as certifcate_file: - cert = crypto.X509() - subject = cert.get_subject() - subject.CN = args.server_name - - cert.set_serial_number(1000) - cert.gmtime_adj_notBefore(0) - cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) - cert.set_issuer(cert.get_subject()) - cert.set_pubkey(tls_private_key) - - cert.sign(tls_private_key, 'sha256') - - cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) - - certifcate_file.write(cert_pem) - - if not os.path.exists(args.tls_dh_params_path): - subprocess.check_call([ - "openssl", "dhparam", - "-outform", "PEM", - "-out", args.tls_dh_params_path, - "2048" - ]) - - config = configparser.SafeConfigParser() - config.add_section("KeyServer") - for key, value in vars(args).items(): - if key != "config_path": - config.set("KeyServer", key, str(value)) - - with open(args.config_path, "w") as config_file: - config.write(config_file) - - -if __name__ == "__main__": - generate_config(sys.argv[1:]) diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 38ae360bcd..cadf574b3b 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -541,7 +541,10 @@ class _TransactionQueue(object): ) def eb(failure): - deferred.errback(failure) + if not deferred.called: + deferred.errback(failure) + else: + logger.exception("Failed to send edu", failure) self._attempt_new_transaction(destination).addErrback(eb) return deferred diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index e08231406d..980a169b25 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -16,6 +16,7 @@ from twisted.internet import defer from synapse.api.events import SynapseEvent +from synapse.util.logutils import log_function from ._base import BaseHandler @@ -44,6 +45,7 @@ class EventStreamHandler(BaseHandler): self.notifier = hs.get_notifier() @defer.inlineCallbacks + @log_function def get_stream(self, auth_user_id, pagin_config, timeout=0): auth_user = self.hs.parse_userid(auth_user_id) @@ -90,13 +92,15 @@ class EventStreamHandler(BaseHandler): # 10 seconds of grace to allow the client to reconnect again # before we think they're gone def _later(): + logger.debug("_later stopped_user_eventstream %s", auth_user) self.distributor.fire( "stopped_user_eventstream", auth_user ) del self._stop_timer_per_user[auth_user] + logger.debug("Scheduling _later: for %s", auth_user) self._stop_timer_per_user[auth_user] = ( - self.clock.call_later(5, _later) + self.clock.call_later(30, _later) ) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 1cc820fb5b..eac110419c 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -146,7 +146,7 @@ class FederationHandler(BaseHandler): # Huh, let's try and get the current state try: yield self.replication_layer.get_state_for_context( - origin, event.room_id + event.origin, event.room_id ) hosts = yield self.store.get_joined_hosts_for_room( diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 32548e66fb..3d7f97bcff 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -277,10 +277,13 @@ class MessageHandler(BaseRoomHandler): end_token=now_token.events_key, ) + start_token = now_token.copy_and_replace("events_key", token[0]) + end_token = now_token.copy_and_replace("events_key", token[1]) + d["messages"] = { "chunk": [m.get_dict() for m in messages], - "start": token[0], - "end": token[1], + "start": start_token.to_string(), + "end": end_token.to_string(), } current_state = yield self.store.get_current_state( diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index c479908f61..7731de85c0 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -18,6 +18,8 @@ from twisted.internet import defer from synapse.api.errors import SynapseError, AuthError from synapse.api.constants import PresenceState +from synapse.util.logutils import log_function + from ._base import BaseHandler import logging @@ -142,7 +144,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def is_presence_visible(self, observer_user, observed_user): defer.returnValue(True) - return + # return # FIXME (erikj): This code path absolutely kills the database. assert(observed_user.is_mine) @@ -188,8 +190,9 @@ class PresenceHandler(BaseHandler): defer.returnValue(state) @defer.inlineCallbacks + @log_function def set_state(self, target_user, auth_user, state): - return + # return # TODO (erikj): Turn this back on. Why did we end up sending EDUs # everywhere? @@ -245,33 +248,42 @@ class PresenceHandler(BaseHandler): self.push_presence(user, statuscache=statuscache) + @log_function def started_user_eventstream(self, user): # TODO(paul): Use "last online" state self.set_state(user, user, {"state": PresenceState.ONLINE}) + @log_function def stopped_user_eventstream(self, user): # TODO(paul): Save current state as "last online" state self.set_state(user, user, {"state": PresenceState.OFFLINE}) @defer.inlineCallbacks def user_joined_room(self, user, room_id): - localusers = set() - remotedomains = set() - - rm_handler = self.homeserver.get_handlers().room_member_handler - yield rm_handler.fetch_room_distributions_into(room_id, - localusers=localusers, remotedomains=remotedomains, - ignore_user=user) if user.is_mine: - yield self._send_presence_to_distribution(srcuser=user, - localusers=localusers, remotedomains=remotedomains, + self.push_update_to_local_and_remote( + observed_user=user, + room_ids=[room_id], + statuscache=self._get_or_offline_usercache(user), + ) + + else: + self.push_update_to_clients( + observed_user=user, + room_ids=[room_id], statuscache=self._get_or_offline_usercache(user), ) - for srcuser in localusers: - yield self._send_presence(srcuser=srcuser, destuser=user, - statuscache=self._get_or_offline_usercache(srcuser), + # We also want to tell them about current presence of people. + rm_handler = self.homeserver.get_handlers().room_member_handler + curr_users = yield rm_handler.get_room_members(room_id) + + for local_user in [c for c in curr_users if c.is_mine]: + self.push_update_to_local_and_remote( + observed_user=local_user, + users_to_push=[user], + statuscache=self._get_or_offline_usercache(local_user), ) @defer.inlineCallbacks @@ -382,11 +394,13 @@ class PresenceHandler(BaseHandler): defer.returnValue(presence) @defer.inlineCallbacks + @log_function def start_polling_presence(self, user, target_user=None, state=None): logger.debug("Start polling for presence from %s", user) if target_user: target_users = set([target_user]) + room_ids = [] else: presence = yield self.store.get_presence_list( user.localpart, accepted=True @@ -400,23 +414,37 @@ class PresenceHandler(BaseHandler): rm_handler = self.homeserver.get_handlers().room_member_handler room_ids = yield rm_handler.get_rooms_for_user(user) - for room_id in room_ids: - for member in (yield rm_handler.get_room_members(room_id)): - target_users.add(member) - if state is None: state = yield self.store.get_presence_state(user.localpart) - - localusers, remoteusers = partitionbool( - target_users, - lambda u: u.is_mine + else: +# statuscache = self._get_or_make_usercache(user) +# self._user_cachemap_latest_serial += 1 +# statuscache.update(state, self._user_cachemap_latest_serial) + pass + + yield self.push_update_to_local_and_remote( + observed_user=user, + users_to_push=target_users, + room_ids=room_ids, + statuscache=self._get_or_make_usercache(user), ) - for target_user in localusers: - self._start_polling_local(user, target_user) + for target_user in target_users: + if target_user.is_mine: + self._start_polling_local(user, target_user) + + # We want to tell the person that just came online + # presence state of people they are interested in? + self.push_update_to_clients( + observed_user=target_user, + users_to_push=[user], + statuscache=self._get_or_offline_usercache(target_user), + ) deferreds = [] - remoteusers_by_domain = partition(remoteusers, lambda u: u.domain) + remote_users = [u for u in target_users if not u.is_mine] + remoteusers_by_domain = partition(remote_users, lambda u: u.domain) + # Only poll for people in our get_presence_list for domain in remoteusers_by_domain: remoteusers = remoteusers_by_domain[domain] @@ -438,25 +466,26 @@ class PresenceHandler(BaseHandler): self._local_pushmap[target_localpart].add(user) - self.push_update_to_clients( - observer_user=user, - observed_user=target_user, - statuscache=self._get_or_offline_usercache(target_user), - ) - def _start_polling_remote(self, user, domain, remoteusers): + to_poll = set() + for u in remoteusers: if u not in self._remote_recvmap: self._remote_recvmap[u] = set() + to_poll.add(u) self._remote_recvmap[u].add(user) + if not to_poll: + return defer.succeed(None) + return self.federation.send_edu( destination=domain, edu_type="m.presence", - content={"poll": [u.to_string() for u in remoteusers]} + content={"poll": [u.to_string() for u in to_poll]} ) + @log_function def stop_polling_presence(self, user, target_user=None): logger.debug("Stop polling for presence from %s", user) @@ -496,20 +525,28 @@ class PresenceHandler(BaseHandler): if not self._local_pushmap[localpart]: del self._local_pushmap[localpart] + @log_function def _stop_polling_remote(self, user, domain, remoteusers): + to_unpoll = set() + for u in remoteusers: self._remote_recvmap[u].remove(user) if not self._remote_recvmap[u]: del self._remote_recvmap[u] + to_unpoll.add(u) + + if not to_unpoll: + return defer.succeed(None) return self.federation.send_edu( destination=domain, edu_type="m.presence", - content={"unpoll": [u.to_string() for u in remoteusers]} + content={"unpoll": [u.to_string() for u in to_unpoll]} ) @defer.inlineCallbacks + @log_function def push_presence(self, user, statuscache): assert(user.is_mine) @@ -525,53 +562,17 @@ class PresenceHandler(BaseHandler): rm_handler = self.homeserver.get_handlers().room_member_handler room_ids = yield rm_handler.get_rooms_for_user(user) - for room_id in room_ids: - yield rm_handler.fetch_room_distributions_into( - room_id, localusers=localusers, remotedomains=remotedomains, - ignore_user=user, - ) - - if not localusers and not remotedomains: + if not localusers and not room_ids: defer.returnValue(None) - yield self._send_presence_to_distribution(user, - localusers=localusers, remotedomains=remotedomains, - statuscache=statuscache + yield self.push_update_to_local_and_remote( + observed_user=user, + users_to_push=localusers, + remote_domains=remotedomains, + room_ids=room_ids, + statuscache=statuscache, ) - def _send_presence(self, srcuser, destuser, statuscache): - if destuser.is_mine: - self.push_update_to_clients( - observer_user=destuser, - observed_user=srcuser, - statuscache=statuscache) - return defer.succeed(None) - else: - return self._push_presence_remote(srcuser, destuser.domain, - state=statuscache.get_state() - ) - - @defer.inlineCallbacks - def _send_presence_to_distribution(self, srcuser, localusers=set(), - remotedomains=set(), statuscache=None): - - for u in localusers: - logger.debug(" | push to local user %s", u) - self.push_update_to_clients( - observer_user=u, - observed_user=srcuser, - statuscache=statuscache, - ) - - deferreds = [] - for domain in remotedomains: - logger.debug(" | push to remote domain %s", domain) - deferreds.append(self._push_presence_remote(srcuser, domain, - state=statuscache.get_state()) - ) - - yield defer.DeferredList(deferreds) - @defer.inlineCallbacks def _push_presence_remote(self, user, destination, state=None): if state is None: @@ -587,12 +588,17 @@ class PresenceHandler(BaseHandler): self.clock.time_msec() - state.pop("mtime") ) + user_state = { + "user_id": user.to_string(), + } + user_state.update(**state) + yield self.federation.send_edu( destination=destination, edu_type="m.presence", content={ "push": [ - dict(user_id=user.to_string(), **state), + user_state, ], } ) @@ -611,12 +617,7 @@ class PresenceHandler(BaseHandler): rm_handler = self.homeserver.get_handlers().room_member_handler room_ids = yield rm_handler.get_rooms_for_user(user) - for room_id in room_ids: - yield rm_handler.fetch_room_distributions_into( - room_id, localusers=observers, ignore_user=user - ) - - if not observers: + if not observers and not room_ids: break state = dict(push) @@ -632,12 +633,12 @@ class PresenceHandler(BaseHandler): self._user_cachemap_latest_serial += 1 statuscache.update(state, serial=self._user_cachemap_latest_serial) - for observer_user in observers: - self.push_update_to_clients( - observer_user=observer_user, - observed_user=user, - statuscache=statuscache, - ) + self.push_update_to_clients( + observed_user=user, + users_to_push=observers, + room_ids=room_ids, + statuscache=statuscache, + ) if state["state"] == PresenceState.OFFLINE: del self._user_cachemap[user] @@ -671,12 +672,53 @@ class PresenceHandler(BaseHandler): yield defer.DeferredList(deferreds) - def push_update_to_clients(self, observer_user, observed_user, - statuscache): - statuscache.make_event(user=observed_user, clock=self.clock) + @defer.inlineCallbacks + def push_update_to_local_and_remote(self, observed_user, + users_to_push=[], room_ids=[], + remote_domains=[], + statuscache=None): + + localusers, remoteusers = partitionbool( + users_to_push, + lambda u: u.is_mine + ) + + localusers = set(localusers) + + self.push_update_to_clients( + observed_user=observed_user, + users_to_push=localusers, + room_ids=room_ids, + statuscache=statuscache, + ) + + remote_domains = set(remote_domains) + remote_domains |= set([r.domain for r in remoteusers]) + for room_id in room_ids: + remote_domains.update( + (yield self.store.get_joined_hosts_for_room(room_id)) + ) + + remote_domains.discard(self.hs.hostname) + + deferreds = [] + for domain in remote_domains: + logger.debug(" | push to remote domain %s", domain) + deferreds.append( + self._push_presence_remote( + observed_user, domain, state=statuscache.get_state() + ) + ) + + yield defer.DeferredList(deferreds) + + defer.returnValue((localusers, remote_domains)) + def push_update_to_clients(self, observed_user, users_to_push=[], + room_ids=[], statuscache=None): self.notifier.on_new_user_event( - [observer_user], + users_to_push, + room_ids, ) diff --git a/synapse/notifier.py b/synapse/notifier.py index 3d3fcdabdb..b6d5ec4820 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -119,6 +119,7 @@ class Notifier(object): ) @defer.inlineCallbacks + @log_function def on_new_user_event(self, users=[], rooms=[]): """ Used to inform listeners that something has happend presence/user event wise. diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 4f42afc015..0b78222827 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -205,8 +205,11 @@ class StreamStore(SQLBaseStore): with_feedback=False): # TODO (erikj): Handle compressed feedback - from_comp = '<' if direction =='b' else '>' - to_comp = '>' if direction =='b' else '<' + # Tokens really represent positions between elements, but we use + # the convention of pointing to the event before the gap. Hence + # we have a bit of asymmetry when it comes to equalities. + from_comp = '<=' if direction =='b' else '>' + to_comp = '>' if direction =='b' else '<=' order = "DESC" if direction == 'b' else "ASC" args = [room_id] @@ -294,7 +297,7 @@ class StreamStore(SQLBaseStore): logger.debug("get_room_events_max_id: %s", res) if not res or not res[0] or not res[0]["m"]: - return "s1" + return "s0" key = res[0]["m"] return "s%d" % (key,) diff --git a/synapse/streams/config.py b/synapse/streams/config.py index 2434844d80..01bab568ff 100644 --- a/synapse/streams/config.py +++ b/synapse/streams/config.py @@ -81,4 +81,4 @@ class PaginationConfig(object): return ( "<PaginationConfig from_tok=%s, to_tok=%s, " "direction=%s, limit=%s>" - ) % (self.from_tok, self.to_tok, self.direction, self.limit) + ) % (self.from_token, self.to_token, self.direction, self.limit) diff --git a/synapse/util/logutils.py b/synapse/util/logutils.py index 021649071b..b94a749786 100644 --- a/synapse/util/logutils.py +++ b/synapse/util/logutils.py @@ -18,6 +18,8 @@ from inspect import getcallargs from functools import wraps import logging +import inspect +import traceback def log_function(f): @@ -65,4 +67,55 @@ def log_function(f): return f(*args, **kwargs) + wrapped.__name__ = func_name + return wrapped + + +def trace_function(f): + func_name = f.__name__ + linenum = f.func_code.co_firstlineno + pathname = f.func_code.co_filename + + def wrapped(*args, **kwargs): + name = f.__module__ + logger = logging.getLogger(name) + level = logging.DEBUG + + s = inspect.currentframe().f_back + + to_print = [ + "\t%s:%s %s. Args: args=%s, kwargs=%s" % ( + pathname, linenum, func_name, args, kwargs + ) + ] + while s: + if True or s.f_globals["__name__"].startswith("synapse"): + filename, lineno, function, _, _ = inspect.getframeinfo(s) + args_string = inspect.formatargvalues(*inspect.getargvalues(s)) + + to_print.append( + "\t%s:%d %s. Args: %s" % ( + filename, lineno, function, args_string + ) + ) + + s = s.f_back + + msg = "\nTraceback for %s:\n" % (func_name,) + "\n".join(to_print) + + record = logging.LogRecord( + name=name, + level=level, + pathname=pathname, + lineno=lineno, + msg=msg, + args=None, + exc_info=None + ) + + logger.handle(record) + + return f(*args, **kwargs) + + wrapped.__name__ = func_name return wrapped |