Merge branch 'erikj/ldap3_auth'

This only makes a difference for versions of ldap3 before 1.0, but a)
its best to be explicit and b) there are distributions that package
ancient versions for ldap3 (e.g. debian).

.gitignore

@@ -24,10 +24,10 @@ homeserver*.yaml
 .coverage
 htmlcov
 
-demo/*.db
-demo/*.log
-demo/*.log.*
-demo/*.pid
+demo/*/*.db
+demo/*/*.log
+demo/*/*.log.*
+demo/*/*.pid
 demo/media_store.*
 demo/etc
 
@@ -42,3 +42,7 @@ build/
 
 localhost-800*/
 static/client/register/register_config.js
+.tox
+
+env/
+*.config

AUTHORS.rst

@@ -29,9 +29,34 @@ Matthew Hodgson <matthew at matrix.org>
 
 Emmanuel Rohee <manu at matrix.org>
  * Supporting iOS clients (testability and fallback registration)
-  
+
 Turned to Dust <dwinslow86 at gmail.com>
  * ArchLinux installation instructions
 
 Brabo <brabo at riseup.net>
  * Installation instruction fixes
+
+Ivan Shapovalov <intelfx100 at gmail.com>
+ * contrib/systemd: a sample systemd unit file and a logger configuration
+
+Eric Myhre <hash at exultant.us>
+ * Fix bug where ``media_store_path`` config option was ignored by v0 content
+   repository API.
+
+Muthu Subramanian <muthu.subramanian.karunanidhi at ericsson.com>
+ * Add SAML2 support for registration and login.
+
+Steven Hammerton <steven.hammerton at openmarket.com>
+ * Add CAS support for registration and login.
+
+Mads Robin Christensen <mads at v42 dot dk>
+ * CentOS 7 installation instructions.
+
+Florent Violleau <floviolleau at gmail dot com>
+ * Add Raspberry Pi installation instructions and general troubleshooting items
+
+Niklas Riekenbrauck <nikriek at gmail dot.com>
+ * Add JWT support for registration and login
+
+Christoph Witzany <christoph at web.crofting.com>
+ * Add LDAP support for authentication

MANIFEST.in

@@ -3,12 +3,28 @@ include LICENSE
 include VERSION
 include *.rst
 include demo/README
+include demo/demo.tls.dh
+include demo/*.py
+include demo/*.sh
 
 recursive-include synapse/storage/schema *.sql
+recursive-include synapse/storage/schema *.py
 
-recursive-include demo *.dh
-recursive-include demo *.py
-recursive-include demo *.sh
 recursive-include docs *
+recursive-include res *
 recursive-include scripts *
+recursive-include scripts-dev *
+recursive-include synapse *.pyi
 recursive-include tests *.py
+
+recursive-include synapse/static *.css
+recursive-include synapse/static *.gif
+recursive-include synapse/static *.html
+recursive-include synapse/static *.js
+
+exclude jenkins.sh
+exclude jenkins*.sh
+exclude jenkins*
+recursive-exclude jenkins *.sh
+
+prune demo/etc

README.rst

@@ -7,12 +7,12 @@ Matrix is an ambitious new ecosystem for open federated Instant Messaging and
 VoIP.  The basics you need to know to get up and running are:
 
 - Everything in Matrix happens in a room.  Rooms are distributed and do not
-  exist on any single server.  Rooms can be located using convenience aliases 
+  exist on any single server.  Rooms can be located using convenience aliases
   like ``#matrix:matrix.org`` or ``#test:localhost:8448``.
 
 - Matrix user IDs look like ``@matthew:matrix.org`` (although in the future
-  you will normally refer to yourself and others using a 3PID: email
-  address, phone number, etc rather than manipulating Matrix user IDs)
+  you will normally refer to yourself and others using a third party identifier 
+  (3PID): email address, phone number, etc rather than manipulating Matrix user IDs)
 
 The overall architecture is::
 
@@ -20,10 +20,10 @@ The overall architecture is::
              https://somewhere.org/_matrix      https://elsewhere.net/_matrix
 
 ``#matrix:matrix.org`` is the official support room for Matrix, and can be
-accessed by the web client at http://matrix.org/beta or via an IRC bridge at
-irc://irc.freenode.net/matrix.
+accessed by any client from https://matrix.org/blog/try-matrix-now or via IRC
+bridge at irc://irc.freenode.net/matrix.
 
-Synapse is currently in rapid development, but as of version 0.5 we believe it 
+Synapse is currently in rapid development, but as of version 0.5 we believe it
 is sufficiently stable to be run as an internet-facing service for real usage!
 
 About Matrix
@@ -58,12 +58,13 @@ the spec in the context of a codebase and let you run your own homeserver and
 generally help bootstrap the ecosystem.
 
 In Matrix, every user runs one or more Matrix clients, which connect through to
-a Matrix homeserver which stores all their personal chat history and user
-account information - much as a mail client connects through to an IMAP/SMTP
-server. Just like email, you can either run your own Matrix homeserver and
-control and own your own communications and history or use one hosted by
-someone else (e.g. matrix.org) - there is no single point of control or
-mandatory service provider in Matrix, unlike WhatsApp, Facebook, Hangouts, etc.
+a Matrix homeserver. The homeserver stores all their personal chat history and
+user account information - much as a mail client connects through to an
+IMAP/SMTP server. Just like email, you can either run your own Matrix
+homeserver and control and own your own communications and history or use one
+hosted by someone else (e.g. matrix.org) - there is no single point of control
+or mandatory service provider in Matrix, unlike WhatsApp, Facebook, Hangouts,
+etc.
 
 Synapse ships with two basic demo Matrix clients: webclient (a basic group chat
 web client demo implemented in AngularJS) and cmdclient (a basic Python
@@ -77,14 +78,14 @@ Meanwhile, iOS and Android SDKs and clients are available from:
 - https://github.com/matrix-org/matrix-android-sdk
 
 We'd like to invite you to join #matrix:matrix.org (via
-https://matrix.org/beta), run a homeserver, take a look at the Matrix spec at
-https://matrix.org/docs/spec and API docs at https://matrix.org/docs/api,
-experiment with the APIs and the demo clients, and report any bugs via
-https://matrix.org/jira.
+https://matrix.org/blog/try-matrix-now), run a homeserver, take a look at the
+Matrix spec at https://matrix.org/docs/spec and API docs at
+https://matrix.org/docs/api, experiment with the APIs and the demo clients, and
+report any bugs via https://matrix.org/jira.
 
 Thanks for using Matrix!
 
-[1] End-to-end encryption is currently in development
+[1] End-to-end encryption is currently in development - see https://matrix.org/git/olm
 
 Synapse Installation
 ====================
@@ -94,6 +95,7 @@ Synapse is the reference python/twisted Matrix homeserver implementation.
 System requirements:
 - POSIX-compliant system (tested on Linux & OS X)
 - Python 2.7
+- At least 1GB of free RAM if you want to join large public rooms like #matrix:matrix.org
 
 Synapse is written in python but some of the libraries is uses are written in
 C. So before we can install synapse itself we need a working C compiler and the
@@ -101,41 +103,73 @@ header files for python C extensions.
 
 Installing prerequisites on Ubuntu or Debian::
 
-    $ sudo apt-get install build-essential python2.7-dev libffi-dev \
-                           python-pip python-setuptools sqlite3 \
-                           libssl-dev python-virtualenv libjpeg-dev
-                           
+    sudo apt-get install build-essential python2.7-dev libffi-dev \
+                         python-pip python-setuptools sqlite3 \
+                         libssl-dev python-virtualenv libjpeg-dev libxslt1-dev
+
 Installing prerequisites on ArchLinux::
 
-    $ sudo pacman -S base-devel python2 python-pip \
-                     python-setuptools python-virtualenv sqlite3
+    sudo pacman -S base-devel python2 python-pip \
+                   python-setuptools python-virtualenv sqlite3
+
+Installing prerequisites on CentOS 7::
+
+    sudo yum install libtiff-devel libjpeg-devel libzip-devel freetype-devel \
+                     lcms2-devel libwebp-devel tcl-devel tk-devel \
+                     python-virtualenv libffi-devel openssl-devel
+    sudo yum groupinstall "Development Tools"
 
 Installing prerequisites on Mac OS X::
 
-    $ xcode-select --install
-    $ sudo pip install virtualenv
-    
+    xcode-select --install
+    sudo easy_install pip
+    sudo pip install virtualenv
+
+Installing prerequisites on Raspbian::
+
+    sudo apt-get install build-essential python2.7-dev libffi-dev \
+                         python-pip python-setuptools sqlite3 \
+                         libssl-dev python-virtualenv libjpeg-dev
+    sudo pip install --upgrade pip
+    sudo pip install --upgrade ndg-httpsclient
+    sudo pip install --upgrade virtualenv
+
+Installing prerequisites on openSUSE::
+
+    sudo zypper in -t pattern devel_basis
+    sudo zypper in python-pip python-setuptools sqlite3 python-virtualenv \
+                   python-devel libffi-devel libopenssl-devel libjpeg62-devel
+
 To install the synapse homeserver run::
 
-    $ virtualenv ~/.synapse
-    $ source ~/.synapse/bin/activate
-    $ pip install --process-dependency-links https://github.com/matrix-org/synapse/tarball/master
+    virtualenv -p python2.7 ~/.synapse
+    source ~/.synapse/bin/activate
+    pip install --upgrade setuptools
+    pip install https://github.com/matrix-org/synapse/tarball/master
 
 This installs synapse, along with the libraries it uses, into a virtual
-environment under ``~/.synapse``.
+environment under ``~/.synapse``.  Feel free to pick a different directory
+if you prefer.
+
+In case of problems, please see the _Troubleshooting section below.
 
 Alternatively, Silvio Fricke has contributed a Dockerfile to automate the
 above in Docker at https://registry.hub.docker.com/u/silviof/docker-matrix/.
 
+Also, Martin Giess has created an auto-deployment process with vagrant/ansible, 
+tested with VirtualBox/AWS/DigitalOcean - see https://github.com/EMnify/matrix-synapse-auto-deploy 
+for details.
+
 To set up your homeserver, run (in your virtualenv, as before)::
 
-    $ cd ~/.synapse
-    $ python -m synapse.app.homeserver \
+    cd ~/.synapse
+    python -m synapse.app.homeserver \
         --server-name machine.my.domain.name \
         --config-path homeserver.yaml \
-        --generate-config
+        --generate-config \
+        --report-stats=[yes|no]
 
-Substituting your host and domain name as appropriate.
+...substituting your host and domain name as appropriate.
 
 This will generate you a config file that you can then customise, but it will
 also generate a set of keys for you. These keys will allow your Home Server to
@@ -143,15 +177,15 @@ identify itself to other Home Servers, so don't lose or delete them. It would be
 wise to back them up somewhere safe. If, for whatever reason, you do need to
 change your Home Server's keys, you may find that other Home Servers have the
 old key cached. If you update the signing key, you should change the name of the
-key in the <server name>.signing.key file (the second word, which by default is
-, 'auto') to something different.
+key in the <server name>.signing.key file (the second word) to something different.
 
 By default, registration of new users is disabled. You can either enable
 registration in the config by specifying ``enable_registration: true``
-(it is then recommended to also set up CAPTCHA), or
+(it is then recommended to also set up CAPTCHA - see docs/CAPTCHA_SETUP), or
 you can use the command line to register new users::
 
     $ source ~/.synapse/bin/activate
+    $ synctl start # if not already running
     $ register_new_matrix_user -c homeserver.yaml https://localhost:8448
     New user localpart: erikj
     Password:
@@ -161,6 +195,31 @@ you can use the command line to register new users::
 For reliable VoIP calls to be routed via this homeserver, you MUST configure
 a TURN server.  See docs/turn-howto.rst for details.
 
+Running Synapse
+===============
+
+To actually run your new homeserver, pick a working directory for Synapse to
+run (e.g. ``~/.synapse``), and::
+
+    cd ~/.synapse
+    source ./bin/activate
+    synctl start
+
+Security Note
+=============
+
+Matrix serves raw user generated data in some APIs - specifically the content
+repository endpoints: http://matrix.org/docs/spec/client_server/r0.2.0.html#get-matrix-media-r0-download-servername-mediaid
+Whilst we have tried to mitigate against possible XSS attacks (e.g.
+https://github.com/matrix-org/synapse/pull/1021) we recommend running
+matrix homeservers on a dedicated domain name, to limit any malicious user generated
+content served to web browsers a matrix API from being able to attack webapps hosted
+on the same domain.  This is particularly true of sharing a matrix webclient and
+server on the same domain.
+
+See https://github.com/vector-im/vector-web/issues/1977 and
+https://developer.github.com/changes/2014-04-25-user-content-security for more details.
+
 Using PostgreSQL
 ================
 
@@ -170,32 +229,32 @@ traditionally used for convenience and simplicity.
 
 The advantages of Postgres include:
 
- * significant performance improvements due to the superior threading and
-   caching model, smarter query optimiser
- * allowing the DB to be run on separate hardware
- * allowing basic active/backup high-availability with a "hot spare" synapse
-   pointing at the same DB master, as well as enabling DB replication in
-   synapse itself.
-   
-The only disadvantage is that the code is relatively new as of April 2015 and
-may have a few regressions relative to SQLite.
+* significant performance improvements due to the superior threading and
+  caching model, smarter query optimiser
+* allowing the DB to be run on separate hardware
+* allowing basic active/backup high-availability with a "hot spare" synapse
+  pointing at the same DB master, as well as enabling DB replication in
+  synapse itself.
 
 For information on how to install and use PostgreSQL, please see
 `docs/postgres.rst <docs/postgres.rst>`_.
 
-Running Synapse
-===============
-
-To actually run your new homeserver, pick a working directory for Synapse to run 
-(e.g. ``~/.synapse``), and::
-
-    $ cd ~/.synapse
-    $ source ./bin/activate
-    $ synctl start
-
 Platform Specific Instructions
 ==============================
 
+Debian
+------
+
+Matrix provides official Debian packages via apt from http://matrix.org/packages/debian/.
+Note that these packages do not include a client - choose one from
+https://matrix.org/blog/try-matrix-now/ (or build your own with one of our SDKs :)
+
+Fedora
+------
+
+Oleg Girko provides Fedora RPMs at
+https://obs.infoserver.lv/project/monitor/matrix-synapse
+
 ArchLinux
 ---------
 
@@ -209,48 +268,63 @@ defaults to python 3, but synapse currently assumes python 2.7 by default:
 
 pip may be outdated (6.0.7-1 and needs to be upgraded to 6.0.8-1 )::
 
-    $ sudo pip2.7 install --upgrade pip
-    
+    sudo pip2.7 install --upgrade pip
+
 You also may need to explicitly specify python 2.7 again during the install
 request::
 
-    $ pip2.7 install --process-dependency-links \
-        https://github.com/matrix-org/synapse/tarball/master
-    
+    pip2.7 install https://github.com/matrix-org/synapse/tarball/master
+
 If you encounter an error with lib bcrypt causing an Wrong ELF Class:
 ELFCLASS32 (x64 Systems), you may need to reinstall py-bcrypt to correctly
 compile it under the right architecture. (This should not be needed if
 installing under virtualenv)::
 
-    $ sudo pip2.7 uninstall py-bcrypt
-    $ sudo pip2.7 install py-bcrypt
-    
+    sudo pip2.7 uninstall py-bcrypt
+    sudo pip2.7 install py-bcrypt
+
 During setup of Synapse you need to call python2.7 directly again::
 
-    $ cd ~/.synapse
-    $ python2.7 -m synapse.app.homeserver \
+    cd ~/.synapse
+    python2.7 -m synapse.app.homeserver \
       --server-name machine.my.domain.name \
       --config-path homeserver.yaml \
       --generate-config
-        
+
 ...substituting your host and domain name as appropriate.
 
+FreeBSD
+-------
+
+Synapse can be installed via FreeBSD Ports or Packages contributed by Brendan Molloy from:
+
+ - Ports: ``cd /usr/ports/net/py-matrix-synapse && make install clean``
+ - Packages: ``pkg install py27-matrix-synapse``
+
+NixOS
+-----
+
+Robin Lambertz has packaged Synapse for NixOS at:
+https://github.com/NixOS/nixpkgs/blob/master/nixos/modules/services/misc/matrix-synapse.nix
+
 Windows Install
 ---------------
 Synapse can be installed on Cygwin. It requires the following Cygwin packages:
 
- - gcc
- - git
- - libffi-devel
- - openssl (and openssl-devel, python-openssl)
- - python
- - python-setuptools
+- gcc
+- git
+- libffi-devel
+- openssl (and openssl-devel, python-openssl)
+- python
+- python-setuptools
 
 The content repository requires additional packages and will be unable to process
 uploads without them:
- - libjpeg8
- - libjpeg8-devel
- - zlib
+
+- libjpeg8
+- libjpeg8-devel
+- zlib
+
 If you choose to install Synapse without these packages, you will need to reinstall
 ``pillow`` for changes to be applied, e.g. ``pip uninstall pillow`` ``pip install
 pillow --user``
@@ -272,33 +346,49 @@ Troubleshooting
 Troubleshooting Installation
 ----------------------------
 
-Synapse requires pip 1.7 or later, so if your OS provides too old a version and 
-you get errors about ``error: no such option: --process-dependency-links`` you 
+Synapse requires pip 1.7 or later, so if your OS provides too old a version you
 may need to manually upgrade it::
 
-    $ sudo pip install --upgrade pip
+    sudo pip install --upgrade pip
+
+Installing may fail with ``Could not find any downloads that satisfy the requirement pymacaroons-pynacl (from matrix-synapse==0.12.0)``.
+You can fix this by manually upgrading pip and virtualenv::
+
+    sudo pip install --upgrade virtualenv
+
+You can next rerun ``virtualenv -p python2.7 synapse`` to update the virtual env.
+
+Installing may fail during installing virtualenv with ``InsecurePlatformWarning: A true SSLContext object is not available. This prevents urllib3 from configuring SSL appropriately and may cause certain SSL connections to fail. For more information, see https://urllib3.readthedocs.org/en/latest/security.html#insecureplatformwarning.``
+You can fix this  by manually installing ndg-httpsclient::
+
+    pip install --upgrade ndg-httpsclient
+
+Installing may fail with ``mock requires setuptools>=17.1. Aborting installation``.
+You can fix this by upgrading setuptools::
+
+    pip install --upgrade setuptools
 
 If pip crashes mid-installation for reason (e.g. lost terminal), pip may
 refuse to run until you remove the temporary installation directory it
 created. To reset the installation::
 
-    $ rm -rf /tmp/pip_install_matrix
+    rm -rf /tmp/pip_install_matrix
 
-pip seems to leak *lots* of memory during installation.  For instance, a Linux 
-host with 512MB of RAM may run out of memory whilst installing Twisted.  If this 
-happens, you will have to individually install the dependencies which are 
+pip seems to leak *lots* of memory during installation.  For instance, a Linux
+host with 512MB of RAM may run out of memory whilst installing Twisted.  If this
+happens, you will have to individually install the dependencies which are
 failing, e.g.::
 
-    $ pip install twisted
+    pip install twisted
 
-On OSX, if you encounter clang: error: unknown argument: '-mno-fused-madd' you
+On OS X, if you encounter clang: error: unknown argument: '-mno-fused-madd' you
 will need to export CFLAGS=-Qunused-arguments.
 
 Troubleshooting Running
 -----------------------
 
-If synapse fails with ``missing "sodium.h"`` crypto errors, you may need 
-to manually upgrade PyNaCL, as synapse uses NaCl (http://nacl.cr.yp.to/) for 
+If synapse fails with ``missing "sodium.h"`` crypto errors, you may need
+to manually upgrade PyNaCL, as synapse uses NaCl (http://nacl.cr.yp.to/) for
 encryption and digital signatures.
 Unfortunately PyNACL currently has a few issues
 (https://github.com/pyca/pynacl/issues/53) and
@@ -307,10 +397,11 @@ correctly, causing all tests to fail with errors about missing "sodium.h". To
 fix try re-installing from PyPI or directly from
 (https://github.com/pyca/pynacl)::
 
-    $ # Install from PyPI
-    $ pip install --user --upgrade --force pynacl
-    $ # Install from github
-    $ pip install --user https://github.com/pyca/pynacl/tarball/master
+    # Install from PyPI
+    pip install --user --upgrade --force pynacl
+
+    # Install from github
+    pip install --user https://github.com/pyca/pynacl/tarball/master
 
 ArchLinux
 ~~~~~~~~~
@@ -318,8 +409,8 @@ ArchLinux
 If running `$ synctl start` fails with 'returned non-zero exit status 1',
 you will need to explicitly call Python2.7 - either running as::
 
-    $ python2.7 -m synapse.app.homeserver --daemonize -c homeserver.yaml --pid-file homeserver.pid
-    
+    python2.7 -m synapse.app.homeserver --daemonize -c homeserver.yaml
+
 ...or by editing synctl with the correct python executable.
 
 Synapse Development
@@ -328,16 +419,16 @@ Synapse Development
 To check out a synapse for development, clone the git repo into a working
 directory of your choice::
 
-    $ git clone https://github.com/matrix-org/synapse.git
-    $ cd synapse
+    git clone https://github.com/matrix-org/synapse.git
+    cd synapse
 
 Synapse has a number of external dependencies, that are easiest
 to install using pip and a virtualenv::
 
-    $ virtualenv env
-    $ source env/bin/activate
-    $ python synapse/python_dependencies.py | xargs -n1 pip install
-    $ pip install setuptools_trial mock
+    virtualenv env
+    source env/bin/activate
+    python synapse/python_dependencies.py | xargs -n1 pip install
+    pip install setuptools_trial mock
 
 This will run a process of downloading and installing all the needed
 dependencies into a virtual env.
@@ -345,7 +436,7 @@ dependencies into a virtual env.
 Once this is done, you may wish to run Synapse's unit tests, to
 check that everything is installed as it should be::
 
-    $ python setup.py test
+    python setup.py test
 
 This should end with a 'PASSED' result::
 
@@ -357,14 +448,11 @@ This should end with a 'PASSED' result::
 Upgrading an existing Synapse
 =============================
 
-IMPORTANT: Before upgrading an existing synapse to a new version, please
-refer to UPGRADE.rst for any additional instructions.
-
-Otherwise, simply re-install the new codebase over the current one - e.g.
-by ``pip install --process-dependency-links
-https://github.com/matrix-org/synapse/tarball/master``
-if using pip, or by ``git pull`` if running off a git working copy.
+The instructions for upgrading synapse are in `UPGRADE.rst`_.
+Please check these instructions as upgrading may require extra steps for some
+versions of synapse.
 
+.. _UPGRADE.rst: UPGRADE.rst
 
 Setting up Federation
 =====================
@@ -375,7 +463,7 @@ You have two choices here, which will influence the form of your Matrix user
 IDs:
 
 1) Use the machine's own hostname as available on public DNS in the form of
-   its A or AAAA records. This is easier to set up initially, perhaps for
+   its A records. This is easier to set up initially, perhaps for
    testing, but lacks the flexibility of SRV.
 
 2) Set up a SRV record for your domain name. This requires you create a SRV
@@ -386,11 +474,11 @@ IDs:
 For the first form, simply pass the required hostname (of the machine) as the
 --server-name parameter::
 
-    $ python -m synapse.app.homeserver \
+    python -m synapse.app.homeserver \
         --server-name machine.my.domain.name \
         --config-path homeserver.yaml \
         --generate-config
-    $ python -m synapse.app.homeserver --config-path homeserver.yaml
+    python -m synapse.app.homeserver --config-path homeserver.yaml
 
 Alternatively, you can run ``synctl start`` to guide you through the process.
 
@@ -407,14 +495,17 @@ record would then look something like::
 At this point, you should then run the homeserver with the hostname of this
 SRV record, as that is the name other machines will expect it to have::
 
-    $ python -m synapse.app.homeserver \
+    python -m synapse.app.homeserver \
         --server-name YOURDOMAIN \
-        --bind-port 8448 \
         --config-path homeserver.yaml \
         --generate-config
-    $ python -m synapse.app.homeserver --config-path homeserver.yaml
+    python -m synapse.app.homeserver --config-path homeserver.yaml
 
 
+If you've already generated the config file, you need to edit the "server_name"
+in you  ```homeserver.yaml``` file. If you've already started Synapse and a
+database has been created, you will have to recreate the database.
+
 You may additionally want to pass one or more "-v" options, in order to
 increase the verbosity of logging output; at least for initial testing.
 
@@ -426,8 +517,8 @@ private federation (``localhost:8080``, ``localhost:8081`` and
 ``localhost:8082``) which you can then access through the webclient running at
 http://localhost:8080. Simply run::
 
-    $ demo/start.sh
-    
+    demo/start.sh
+
 This is mainly useful just for development purposes.
 
 Running The Demo Web Client
@@ -466,7 +557,6 @@ Logging In To An Existing Account
 Just enter the ``@localpart:my.domain.here`` Matrix user ID and password into
 the form and click the Login button.
 
-
 Identity Servers
 ================
 
@@ -487,10 +577,47 @@ we are running a single identity server (https://matrix.org) at the current
 time.
 
 
+URL Previews
+============
+
+Synapse 0.15.0 introduces an experimental new API for previewing URLs at
+/_matrix/media/r0/preview_url.  This is disabled by default.  To turn it on
+you must enable the `url_preview_enabled: True` config parameter and explicitly
+specify the IP ranges that Synapse is not allowed to spider for previewing in
+the `url_preview_ip_range_blacklist` configuration parameter.  This is critical
+from a security perspective to stop arbitrary Matrix users spidering 'internal'
+URLs on your network.  At the very least we recommend that your loopback and
+RFC1918 IP addresses are blacklisted.
+
+This also requires the optional lxml and netaddr python dependencies to be
+installed.
+
+
+Password reset
+==============
+
+If a user has registered an email address to their account using an identity
+server, they can request a password-reset token via clients such as Vector.
+
+A manual password reset can be done via direct database access as follows.
+
+First calculate the hash of the new password:
+
+    $ source ~/.synapse/bin/activate
+    $ ./scripts/hash_password
+    Password: 
+    Confirm password: 
+    $2a$12$xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+
+Then update the `users` table in the database:
+
+    UPDATE users SET password_hash='$2a$12$xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
+        WHERE name='@test:test.com';
+
 Where's the spec?!
 ==================
 
-The source of the matrix spec lives at https://github.com/matrix-org/matrix-doc.  
+The source of the matrix spec lives at https://github.com/matrix-org/matrix-doc.
 A recent HTML snapshot of this lives at http://matrix.org/docs/spec
 
 
@@ -500,10 +627,27 @@ Building Internal API Documentation
 Before building internal API documentation install sphinx and
 sphinxcontrib-napoleon::
 
-    $ pip install sphinx
-    $ pip install sphinxcontrib-napoleon
+    pip install sphinx
+    pip install sphinxcontrib-napoleon
 
 Building internal API documentation::
 
-    $ python setup.py build_sphinx
+    python setup.py build_sphinx
+
+
+
+Help!! Synapse eats all my RAM!
+===============================
+
+Synapse's architecture is quite RAM hungry currently - we deliberately
+cache a lot of recent room data and metadata in RAM in order to speed up
+common requests.  We'll improve this in future, but for now the easiest
+way to either reduce the RAM usage (at the risk of slowing things down)
+is to set the almost-undocumented ``SYNAPSE_CACHE_FACTOR`` environment
+variable.  Roughly speaking, a SYNAPSE_CACHE_FACTOR of 1.0 will max out
+at around 3-4GB of resident memory - this is what we currently run the
+matrix.org on.  The default setting is currently 0.1, which is probably
+around a ~700MB footprint.  You can dial it down further to 0.02 if
+desired, which targets roughly ~512MB.  Conversely you can dial it up if
+you need performance for lots of users and have a box with a lot of RAM.
 

UPGRADE.rst

@@ -1,4 +1,58 @@
-Upgrading to v0.x.x
+Upgrading Synapse
+=================
+
+Before upgrading check if any special steps are required to upgrade from the
+what you currently have installed to current version of synapse. The extra
+instructions that may be required are listed later in this document.
+
+If synapse was installed in a virtualenv then active that virtualenv before
+upgrading. If synapse is installed in a virtualenv in ``~/.synapse/`` then run:
+
+.. code:: bash
+
+    source ~/.synapse/bin/activate
+
+If synapse was installed using pip then upgrade to the latest version by
+running:
+
+.. code:: bash
+
+    pip install --upgrade --process-dependency-links https://github.com/matrix-org/synapse/tarball/master
+
+If synapse was installed using git then upgrade to the latest version by
+running:
+
+.. code:: bash
+
+    # Pull the latest version of the master branch.
+    git pull
+    # Update the versions of synapse's python dependencies.
+    python synapse/python_dependencies.py | xargs -n1 pip install --upgrade
+
+
+Upgrading to v0.15.0
+====================
+
+If you want to use the new URL previewing API (/_matrix/media/r0/preview_url)
+then you have to explicitly enable it in the config and update your dependencies
+dependencies.  See README.rst for details.
+
+
+Upgrading to v0.11.0
+====================
+
+This release includes the option to send anonymous usage stats to matrix.org,
+and requires that administrators explictly opt in or out by setting the
+``report_stats`` option to either ``true`` or ``false``.
+
+We would really appreciate it if you could help our project out by reporting
+anonymized usage statistics from your homeserver. Only very basic aggregate
+data (e.g. number of users) will be reported, but it helps us to track the
+growth of the Matrix community, and helps us to make Matrix a success, as well
+as to convince other networks that they should peer with us.
+
+
+Upgrading to v0.9.0
 ===================
 
 Application services have had a breaking API change in this version.

contrib/cmdclient/console.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/cmdclient/http.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/experiments/cursesio.py

@@ -1,4 +1,4 @@
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/experiments/test_messaging.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/graph/graph.py

@@ -1,4 +1,4 @@
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/graph/graph2.py

@@ -1,4 +1,4 @@
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

contrib/graph/graph3.py

@@ -0,0 +1,151 @@
+# Copyright 2016 OpenMarket 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.
+
+
+import pydot
+import cgi
+import simplejson as json
+import datetime
+import argparse
+
+from synapse.events import FrozenEvent
+from synapse.util.frozenutils import unfreeze
+
+
+def make_graph(file_name, room_id, file_prefix, limit):
+    print "Reading lines"
+    with open(file_name) as f:
+        lines = f.readlines()
+
+    print "Read lines"
+
+    events = [FrozenEvent(json.loads(line)) for line in lines]
+
+    print "Loaded events."
+
+    events.sort(key=lambda e: e.depth)
+
+    print "Sorted events"
+
+    if limit:
+        events = events[-int(limit):]
+
+    node_map = {}
+
+    graph = pydot.Dot(graph_name="Test")
+
+    for event in events:
+        t = datetime.datetime.fromtimestamp(
+            float(event.origin_server_ts) / 1000
+        ).strftime('%Y-%m-%d %H:%M:%S,%f')
+
+        content = json.dumps(unfreeze(event.get_dict()["content"]), indent=4)
+        content = content.replace("\n", "<br/>\n")
+
+        print content
+        content = []
+        for key, value in unfreeze(event.get_dict()["content"]).items():
+            if value is None:
+                value = "<null>"
+            elif isinstance(value, basestring):
+                pass
+            else:
+                value = json.dumps(value)
+
+            content.append(
+                "<b>%s</b>: %s," % (
+                    cgi.escape(key, quote=True).encode("ascii", 'xmlcharrefreplace'),
+                    cgi.escape(value, quote=True).encode("ascii", 'xmlcharrefreplace'),
+                )
+            )
+
+        content = "<br/>\n".join(content)
+
+        print content
+
+        label = (
+            "<"
+            "<b>%(name)s </b><br/>"
+            "Type: <b>%(type)s </b><br/>"
+            "State key: <b>%(state_key)s </b><br/>"
+            "Content: <b>%(content)s </b><br/>"
+            "Time: <b>%(time)s </b><br/>"
+            "Depth: <b>%(depth)s </b><br/>"
+            ">"
+        ) % {
+            "name": event.event_id,
+            "type": event.type,
+            "state_key": event.get("state_key", None),
+            "content": content,
+            "time": t,
+            "depth": event.depth,
+        }
+
+        node = pydot.Node(
+            name=event.event_id,
+            label=label,
+        )
+
+        node_map[event.event_id] = node
+        graph.add_node(node)
+
+    print "Created Nodes"
+
+    for event in events:
+        for prev_id, _ in event.prev_events:
+            try:
+                end_node = node_map[prev_id]
+            except:
+                end_node = pydot.Node(
+                    name=prev_id,
+                    label="<<b>%s</b>>" % (prev_id,),
+                )
+
+                node_map[prev_id] = end_node
+                graph.add_node(end_node)
+
+            edge = pydot.Edge(node_map[event.event_id], end_node)
+            graph.add_edge(edge)
+
+    print "Created edges"
+
+    graph.write('%s.dot' % file_prefix, format='raw', prog='dot')
+
+    print "Created Dot"
+
+    graph.write_svg("%s.svg" % file_prefix, prog='dot')
+
+    print "Created svg"
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Generate a PDU graph for a given room by reading "
+                    "from a file with line deliminated events. \n"
+                    "Requires pydot."
+    )
+    parser.add_argument(
+        "-p", "--prefix", dest="prefix",
+        help="String to prefix output files with",
+        default="graph_output"
+    )
+    parser.add_argument(
+        "-l", "--limit",
+        help="Only retrieve the last N events.",
+    )
+    parser.add_argument('event_file')
+    parser.add_argument('room')
+
+    args = parser.parse_args()
+
+    make_graph(args.event_file, args.room, args.prefix, args.limit)

contrib/systemd/log_config.yaml

@@ -21,3 +21,5 @@ handlers:
 root:
     level: INFO
     handlers: [journal]
+
+disable_existing_loggers: False

contrib/systemd/synapse.service

@@ -9,6 +9,7 @@ Description=Synapse Matrix homeserver
 Type=simple
 User=synapse
 Group=synapse
+EnvironmentFile=-/etc/sysconfig/synapse
 WorkingDirectory=/var/lib/synapse
 ExecStart=/usr/bin/python2.7 -m synapse.app.homeserver --config-path=/etc/synapse/homeserver.yaml --log-config=/etc/synapse/log_config.yaml
 

contrib/vertobot/bot.pl

@@ -126,12 +126,26 @@ sub on_unknown_event
         if (!$bridgestate->{$room_id}->{gathered_candidates}) {
             $bridgestate->{$room_id}->{gathered_candidates} = 1;
             my $offer = $bridgestate->{$room_id}->{offer};
-            my $candidate_block = "";
+            my $candidate_block = {
+                audio => '',
+                video => '',
+            };
             foreach (@{$event->{content}->{candidates}}) {
-                $candidate_block .= "a=" . $_->{candidate} . "\r\n";
+                if ($_->{sdpMid}) {
+                    $candidate_block->{$_->{sdpMid}} .= "a=" . $_->{candidate} . "\r\n";
+                }
+                else {
+                    $candidate_block->{audio} .= "a=" . $_->{candidate} . "\r\n";
+                    $candidate_block->{video} .= "a=" . $_->{candidate} . "\r\n";
+                }
             }
-            # XXX: collate using the right m= line - for now assume audio call
-            $offer =~ s/(a=rtcp.*[\r\n]+)/$1$candidate_block/;
+
+            # XXX: assumes audio comes first
+            #$offer =~ s/(a=rtcp-mux[\r\n]+)/$1$candidate_block->{audio}/;
+            #$offer =~ s/(a=rtcp-mux[\r\n]+)/$1$candidate_block->{video}/;
+
+            $offer =~ s/(m=video)/$candidate_block->{audio}$1/;
+            $offer =~ s/(.$)/$1\n$candidate_block->{video}$1/;
             
             my $f = send_verto_json_request("verto.invite", {
                 "sdp" => $offer,
@@ -172,22 +186,18 @@ sub on_room_message
     warn "[Matrix] in $room_id: $from: " . $content->{body} . "\n";    
 }
 
-my $verto_connecting = $loop->new_future;
-$bot_verto->connect(
-    %{ $CONFIG{"verto-bot"} },
-    on_connect_error => sub { die "Cannot connect to verto - $_[-1]" },
-    on_resolve_error => sub { die "Cannot resolve to verto - $_[-1]" },        
-)->then( sub { 
-    warn("[Verto] connected to websocket");
-    $verto_connecting->done($bot_verto) if not $verto_connecting->is_done;
-});
-
 Future->needs_all(
     $bot_matrix->login( %{ $CONFIG{"matrix-bot"} } )->then( sub {
         $bot_matrix->start;
     }),
     
-    $verto_connecting,
+    $bot_verto->connect(
+        %{ $CONFIG{"verto-bot"} },
+        on_connect_error => sub { die "Cannot connect to verto - $_[-1]" },
+        on_resolve_error => sub { die "Cannot resolve to verto - $_[-1]" },        
+    )->on_done( sub { 
+        warn("[Verto] connected to websocket");
+    }),
 )->get;
 
 $loop->attach_signal(

contrib/vertobot/cpanfile

@@ -11,7 +11,4 @@ requires 'YAML', 0;
 requires 'JSON', 0;
 requires 'Getopt::Long', 0;
 
-on 'test' => sub {
-	requires 'Test::More', '>= 0.98';
-};
 

demo/clean.sh

@@ -11,7 +11,9 @@ if [ -f $PID_FILE ]; then
     exit 1
 fi
 
-find "$DIR" -name "*.log" -delete
-find "$DIR" -name "*.db" -delete
+for port in 8080 8081 8082; do
+    rm -rf $DIR/$port
+    rm -rf $DIR/media_store.$port
+done
 
 rm -rf $DIR/etc

demo/start.sh

@@ -8,38 +8,49 @@ cd "$DIR/.."
 
 mkdir -p demo/etc
 
-# Check the --no-rate-limit param
-PARAMS=""
-if [ $# -eq 1 ]; then
-    if [ $1 = "--no-rate-limit" ]; then
-	    PARAMS="--rc-messages-per-second 1000 --rc-message-burst-count 1000"
-    fi
-fi
+export PYTHONPATH=$(readlink -f $(pwd))
+
+
+echo $PYTHONPATH
 
 for port in 8080 8081 8082; do
     echo "Starting server on port $port... "
 
     https_port=$((port + 400))
+    mkdir -p demo/$port
+    pushd demo/$port
 
+    #rm $DIR/etc/$port.config
     python -m synapse.app.homeserver \
         --generate-config \
-        --config-path "demo/etc/$port.config" \
-        -p "$https_port" \
-        --unsecure-port "$port" \
         -H "localhost:$https_port" \
-        -f "$DIR/$port.log" \
-        -d "$DIR/$port.db" \
-        -D --pid-file "$DIR/$port.pid" \
-        --manhole $((port + 1000)) \
-        --tls-dh-params-path "demo/demo.tls.dh" \
-        --media-store-path "demo/media_store.$port" \
-        $PARAMS $SYNAPSE_PARAMS \
-        --enable-registration
+        --config-path "$DIR/etc/$port.config" \
+        --report-stats no
+
+    # Check script parameters
+    if [ $# -eq 1 ]; then
+        if [ $1 = "--no-rate-limit" ]; then
+            # Set high limits in config file to disable rate limiting
+            perl -p -i -e 's/rc_messages_per_second.*/rc_messages_per_second: 1000/g' $DIR/etc/$port.config
+            perl -p -i -e 's/rc_message_burst_count.*/rc_message_burst_count: 1000/g' $DIR/etc/$port.config
+        fi
+    fi
+
+    perl -p -i -e 's/^enable_registration:.*/enable_registration: true/g' $DIR/etc/$port.config
+
+    if ! grep -F "full_twisted_stacktraces" -q  $DIR/etc/$port.config; then
+        echo "full_twisted_stacktraces: true" >> $DIR/etc/$port.config
+    fi
+    if ! grep -F "report_stats" -q  $DIR/etc/$port.config ; then
+        echo "report_stats: false" >> $DIR/etc/$port.config
+    fi
 
     python -m synapse.app.homeserver \
-        --config-path "demo/etc/$port.config" \
+        --config-path "$DIR/etc/$port.config" \
+        -D \
         -vv \
 
+    popd
 done
 
 cd "$CWD"

docs/admin_api/README.rst

@@ -0,0 +1,12 @@
+Admin APIs
+==========
+
+This directory includes documentation for the various synapse specific admin
+APIs available.
+
+Only users that are server admins can use these APIs. A user can be marked as a
+server admin by updating the database directly, e.g.:
+
+``UPDATE users SET admin = 1 WHERE name = '@foo:bar.com'``
+
+Restarting may be required for the changes to register.

docs/admin_api/purge_history_api.rst

@@ -0,0 +1,15 @@
+Purge History API
+=================
+
+The purge history API allows server admins to purge historic events from their
+database, reclaiming disk space.
+
+Depending on the amount of history being purged a call to the API may take
+several minutes or longer. During this period users will not be able to
+paginate further back in the room from the point being purged from.
+
+The API is simply:
+
+``POST /_matrix/client/r0/admin/purge_history/<room_id>/<event_id>``
+
+including an ``access_token`` of a server admin.

docs/admin_api/purge_remote_media.rst

@@ -0,0 +1,19 @@
+Purge Remote Media API
+======================
+
+The purge remote media API allows server admins to purge old cached remote
+media. 
+
+The API is::
+
+    POST /_matrix/client/r0/admin/purge_media_cache
+
+    {
+        "before_ts": <unix_timestamp_in_ms>
+    }
+
+Which will remove all cached media that was last accessed before
+``<unix_timestamp_in_ms>``.
+
+If the user re-requests purged remote media, synapse will re-request the media
+from the originating server.

docs/application_services.rst

@@ -0,0 +1,35 @@
+Registering an Application Service
+==================================
+
+The registration of new application services depends on the homeserver used. 
+In synapse, you need to create a new configuration file for your AS and add it
+to the list specified under the ``app_service_config_files`` config
+option in your synapse config.
+
+For example:
+
+.. code-block:: yaml
+
+  app_service_config_files:
+  - /home/matrix/.synapse/<your-AS>.yaml
+
+
+The format of the AS configuration file is as follows:
+
+..  code-block:: yaml
+
+    url: <base url of AS>
+    as_token: <token AS will add to requests to HS>
+    hs_token: <token HS will add to requests to AS>
+    sender_localpart: <localpart of AS user>
+    namespaces:
+      users:  # List of users we're interested in
+        - exclusive: <bool>
+          regex: <regex>
+        - ...
+      aliases: []  # List of aliases we're interested in
+      rooms: [] # List of room ids we're interested in
+
+See the spec_ for further details on how application services work.
+
+.. _spec: https://matrix.org/docs/spec/application_service/unstable.html

docs/code_style.rst

@@ -43,7 +43,10 @@ Basically, PEP8
   together, or want to deliberately extend or preserve vertical/horizontal
   space)
 
-Comments should follow the google code style. This is so that we can generate
-documentation with sphinx (http://sphinxcontrib-napoleon.readthedocs.org/en/latest/) 
+Comments should follow the `google code style <http://google.github.io/styleguide/pyguide.html?showone=Comments#Comments>`_.
+This is so that we can generate documentation with 
+`sphinx <http://sphinxcontrib-napoleon.readthedocs.org/en/latest/>`_. See the
+`examples <http://sphinxcontrib-napoleon.readthedocs.io/en/latest/example_google.html>`_
+in the sphinx documentation.
 
 Code should pass pep8 --max-line-length=100 without any warnings.

docs/log_contexts.rst

@@ -0,0 +1,10 @@
+What do I do about "Unexpected logging context" debug log-lines everywhere?
+
+<Mjark> The logging context lives in thread local storage
+<Mjark> Sometimes it gets out of sync with what it should actually be, usually because something scheduled something to run on the reactor without preserving the logging context. 
+<Matthew> what is the impact of it getting out of sync? and how and when should we preserve log context?
+<Mjark> The impact is that some of the CPU and database metrics will be under-reported, and some log lines will be mis-attributed.
+<Mjark> It should happen auto-magically in all the APIs that do IO or otherwise defer to the reactor.
+<Erik> Mjark: the other place is if we branch, e.g. using defer.gatherResults
+
+Unanswered: how and when should we preserve log context?

docs/metrics-howto.rst

@@ -15,36 +15,45 @@ How to monitor Synapse metrics using Prometheus
 
   Restart synapse
 
-3: Check out synapse-prometheus-config
-  https://github.com/matrix-org/synapse-prometheus-config
+3: Add a prometheus target for synapse. It needs to set the ``metrics_path``
+   to a non-default value::
 
-4: Add ``synapse.html`` and ``synapse.rules``
-  The ``.html`` file needs to appear in prometheus's ``consoles`` directory,
-  and the ``.rules`` file needs to be invoked somewhere in the main config
-  file. A symlink to each from the git checkout into the prometheus directory
-  might be easiest to ensure ``git pull`` keeps it updated.
+    - job_name: "synapse"
+      metrics_path: "/_synapse/metrics"
+      static_configs:
+        - targets:
+            "my.server.here:9092"
 
-5: Add a prometheus target for synapse
-  This is easiest if prometheus runs on the same machine as synapse, as it can
-  then just use localhost::
+Standard Metric Names
+---------------------
 
-    global: {
-      rule_file: "synapse.rules"
-    }
+As of synapse version 0.18.2, the format of the process-wide metrics has been
+changed to fit prometheus standard naming conventions. Additionally the units
+have been changed to seconds, from miliseconds.
 
-    job: {
-      name: "synapse"
+================================== =============================
+New name                           Old name
+---------------------------------- -----------------------------
+process_cpu_user_seconds_total     process_resource_utime / 1000
+process_cpu_system_seconds_total   process_resource_stime / 1000
+process_open_fds (no 'type' label) process_fds
+================================== =============================
 
-      target_group: {
-        target: "http://localhost:9092/"
-      }
-    }
+The python-specific counts of garbage collector performance have been renamed.
 
-6: Start prometheus::
+=========================== ======================
+New name                    Old name
+--------------------------- ----------------------
+python_gc_time              reactor_gc_time      
+python_gc_unreachable_total reactor_gc_unreachable
+python_gc_counts            reactor_gc_counts
+=========================== ======================
 
-   ./prometheus -config.file=prometheus.conf
+The twisted-specific reactor metrics have been renamed.
 
-7: Wait a few seconds for it to start and perform the first scrape,
-   then visit the console:
-
-    http://server-where-prometheus-runs:9090/consoles/synapse.html
+==================================== =================
+New name                             Old name
+------------------------------------ -----------------
+python_twisted_reactor_pending_calls reactor_tick_time
+python_twisted_reactor_tick_time     reactor_tick_time
+==================================== =================

docs/postgres.rst

@@ -18,8 +18,8 @@ encoding use, e.g.::
 This would create an appropriate database named ``synapse`` owned by the
 ``synapse_user`` user (which must already exist).
 
-Set up client
-=============
+Set up client in Debian/Ubuntu
+===========================
 
 Postgres support depends on the postgres python connector ``psycopg2``. In the
 virtual env::
@@ -27,6 +27,19 @@ virtual env::
     sudo apt-get install libpq-dev
     pip install psycopg2
 
+Set up client in RHEL/CentOs 7
+==============================
+
+Make sure you have the appropriate version of postgres-devel installed. For a
+postgres 9.4, use the postgres 9.4 packages from
+[here](https://wiki.postgresql.org/wiki/YUM_Installation).
+
+As with Debian/Ubuntu, postgres support depends on the postgres python connector
+``psycopg2``. In the virtual env::
+
+    sudo yum install postgresql-devel libpqxx-devel.x86_64
+    export PATH=/usr/pgsql-9.4/bin/:$PATH
+    pip install psycopg2
 
 Synapse config
 ==============
@@ -34,19 +47,15 @@ Synapse config
 When you are ready to start using PostgreSQL, add the following line to your
 config file::
 
-    database_config: <db_config_file>
-
-Where ``<db_config_file>`` is the file name that points to a yaml file of the
-following form::
-
-    name: psycopg2
-    args:
-        user: <user>
-        password: <pass>
-        database: <db>
-        host: <host>
-        cp_min: 5
-        cp_max: 10
+    database:
+        name: psycopg2
+        args:
+            user: <user>
+            password: <pass>
+            database: <db>
+            host: <host>
+            cp_min: 5
+            cp_max: 10
 
 All key, values in ``args`` are passed to the ``psycopg2.connect(..)``
 function, except keys beginning with ``cp_``, which are consumed by the twisted
@@ -59,9 +68,8 @@ Porting from SQLite
 Overview
 ~~~~~~~~
 
-The script ``port_from_sqlite_to_postgres.py`` allows porting an existing
-synapse server backed by SQLite to using PostgreSQL. This is done in as a two
-phase process:
+The script ``synapse_port_db`` allows porting an existing synapse server
+backed by SQLite to using PostgreSQL. This is done in as a two phase process:
 
 1. Copy the existing SQLite database to a separate location (while the server
    is down) and running the port script against that offline database.
@@ -86,13 +94,12 @@ complete, restart synapse.  For instance::
     cp homeserver.db homeserver.db.snapshot
     ./synctl start
 
-Assuming your database config file (as described in the section *Synapse
-config*) is named ``database_config.yaml`` and the SQLite snapshot is at
+Assuming your new config file (as described in the section *Synapse config*)
+is named ``homeserver-postgres.yaml`` and the SQLite snapshot is at
 ``homeserver.db.snapshot`` then simply run::
 
-    python scripts/port_from_sqlite_to_postgres.py \
-        --sqlite-database homeserver.db.snapshot \
-        --postgres-config database_config.yaml
+    synapse_port_db --sqlite-database homeserver.db.snapshot \
+        --postgres-config homeserver-postgres.yaml
 
 The flag ``--curses`` displays a coloured curses progress UI.
 
@@ -104,8 +111,7 @@ To complete the conversion shut down the synapse server and run the port
 script one last time, e.g. if the SQLite database is at  ``homeserver.db``
 run::
 
-    python scripts/port_from_sqlite_to_postgres.py \
-        --sqlite-database homeserver.db \
+    synapse_port_db --sqlite-database homeserver.db \
         --postgres-config database_config.yaml
 
 Once that has completed, change the synapse config to point at the PostgreSQL

docs/replication.rst

@@ -0,0 +1,58 @@
+Replication Architecture
+========================
+
+Motivation
+----------
+
+We'd like to be able to split some of the work that synapse does into multiple
+python processes. In theory multiple synapse processes could share a single
+postgresql database and we'd scale up by running more synapse processes.
+However much of synapse assumes that only one process is interacting with the
+database, both for assigning unique identifiers when inserting into tables,
+notifying components about new updates, and for invalidating its caches.
+
+So running multiple copies of the current code isn't an option. One way to
+run multiple processes would be to have a single writer process and multiple
+reader processes connected to the same database. In order to do this we'd need
+a way for the reader process to invalidate its in-memory caches when an update
+happens on the writer. One way to do this is for the writer to present an
+append-only log of updates which the readers can consume to invalidate their
+caches and to push updates to listening clients or pushers.
+
+Synapse already stores much of its data as an append-only log so that it can
+correctly respond to /sync requests so the amount of code changes needed to
+expose the append-only log to the readers should be fairly minimal.
+
+Architecture
+------------
+
+The Replication API
+~~~~~~~~~~~~~~~~~~~
+
+Synapse will optionally expose a long poll HTTP API for extracting updates. The
+API will have a similar shape to /sync in that clients provide tokens
+indicating where in the log they have reached and a timeout. The synapse server
+then either responds with updates immediately if it already has updates or it
+waits until the timeout for more updates. If the timeout expires and nothing
+happened then the server returns an empty response.
+
+However unlike the /sync API this replication API is returning synapse specific
+data rather than trying to implement a matrix specification. The replication
+results are returned as arrays of rows where the rows are mostly lifted
+directly from the database. This avoids unnecessary JSON parsing on the server
+and hopefully avoids an impedance mismatch between the data returned and the
+required updates to the datastore.
+
+This does not replicate all the database tables as many of the database tables
+are indexes that can be recovered from the contents of other tables.
+
+The format and parameters for the api are documented in
+``synapse/replication/resource.py``.
+
+
+The Slaved DataStore
+~~~~~~~~~~~~~~~~~~~~
+
+There are read-only version of the synapse storage layer in
+``synapse/replication/slave/storage`` that use the response of the replication
+API to invalidate their caches.

docs/turn-howto.rst

@@ -9,31 +9,35 @@ the Home Server to generate credentials that are valid for use on the TURN
 server through the use of a secret shared between the Home Server and the
 TURN server.
 
-This document described how to install coturn
-(https://code.google.com/p/coturn/) which also supports the TURN REST API,
+This document describes how to install coturn
+(https://github.com/coturn/coturn) which also supports the TURN REST API,
 and integrate it with synapse.
 
 coturn Setup
 ============
 
+You may be able to setup coturn via your package manager,  or set it up manually using the usual ``configure, make, make install`` process.  
+
  1. Check out coturn::
-      svn checkout http://coturn.googlecode.com/svn/trunk/ coturn
+ 
+      git clone https://github.com/coturn/coturn.git coturn
       cd coturn
 
  2. Configure it::
+ 
       ./configure
 
-    You may need to install libevent2: if so, you should do so
+    You may need to install ``libevent2``: if so, you should do so
     in the way recommended by your operating system.
     You can ignore warnings about lack of database support: a
     database is unnecessary for this purpose.
 
  3. Build and install it::
+ 
       make
       make install
 
- 4. Make a config file in /etc/turnserver.conf. You can customise
-    a config file from turnserver.conf.default. The relevant
+ 4. Create or edit the config file in ``/etc/turnserver.conf``. The relevant
     lines, with example values, are::
 
       lt-cred-mech
@@ -41,7 +45,7 @@ coturn Setup
       static-auth-secret=[your secret key here]
       realm=turn.myserver.org
 
-    See turnserver.conf.default for explanations of the options.
+    See turnserver.conf for explanations of the options.
     One way to generate the static-auth-secret is with pwgen::
 
        pwgen -s 64 1
@@ -54,6 +58,7 @@ coturn Setup
     import your private key and certificate.
 
  7. Start the turn server::
+ 
        bin/turnserver -o
 
 

docs/url_previews.rst

@@ -0,0 +1,74 @@
+URL Previews
+============
+
+Design notes on a URL previewing service for Matrix:
+
+Options are:
+
+ 1. Have an AS which listens for URLs, downloads them, and inserts an event that describes their metadata.
+   * Pros:
+     * Decouples the implementation entirely from Synapse.
+     * Uses existing Matrix events & content repo to store the metadata.
+   * Cons:
+     * Which AS should provide this service for a room, and why should you trust it?
+     * Doesn't work well with E2E; you'd have to cut the AS into every room
+     * the AS would end up subscribing to every room anyway.
+
+ 2. Have a generic preview API (nothing to do with Matrix) that provides a previewing service:
+   * Pros:
+     * Simple and flexible; can be used by any clients at any point
+   * Cons:
+     * If each HS provides one of these independently, all the HSes in a room may needlessly DoS the target URI
+     * We need somewhere to store the URL metadata rather than just using Matrix itself
+     * We can't piggyback on matrix to distribute the metadata between HSes.
+
+ 3. Make the synapse of the sending user responsible for spidering the URL and inserting an event asynchronously which describes the metadata.
+   * Pros:
+     * Works transparently for all clients
+     * Piggy-backs nicely on using Matrix for distributing the metadata.
+     * No confusion as to which AS
+   * Cons:
+     * Doesn't work with E2E
+     * We might want to decouple the implementation of the spider from the HS, given spider behaviour can be quite complicated and evolve much more rapidly than the HS.  It's more like a bot than a core part of the server.
+
+ 4. Make the sending client use the preview API and insert the event itself when successful.
+   * Pros:
+      * Works well with E2E
+      * No custom server functionality
+      * Lets the client customise the preview that they send (like on FB)
+   * Cons:
+      * Entirely specific to the sending client, whereas it'd be nice if /any/ URL was correctly previewed if clients support it.
+
+ 5. Have the option of specifying a shared (centralised) previewing service used by a room, to avoid all the different HSes in the room DoSing the target.
+
+Best solution is probably a combination of both 2 and 4.
+ * Sending clients do their best to create and send a preview at the point of sending the message, perhaps delaying the message until the preview is computed?  (This also lets the user validate the preview before sending)
+ * Receiving clients have the option of going and creating their own preview if one doesn't arrive soon enough (or if the original sender didn't create one)
+
+This is a bit magical though in that the preview could come from two entirely different sources - the sending HS or your local one.  However, this can always be exposed to users: "Generate your own URL previews if none are available?"
+
+This is tantamount also to senders calculating their own thumbnails for sending in advance of the main content - we are trusting the sender not to lie about the content in the thumbnail.  Whereas currently thumbnails are calculated by the receiving homeserver to avoid this attack.
+
+However, this kind of phishing attack does exist whether we let senders pick their thumbnails or not, in that a malicious sender can send normal text messages around the attachment claiming it to be legitimate.  We could rely on (future) reputation/abuse management to punish users who phish (be it with bogus metadata or bogus descriptions).   Bogus metadata is particularly bad though, especially if it's avoidable.
+
+As a first cut, let's do #2 and have the receiver hit the API to calculate its own previews (as it does currently for image thumbnails).  We can then extend/optimise this to option 4 as a special extra if needed.
+
+API
+---
+
+GET /_matrix/media/r0/preview_url?url=http://wherever.com
+200 OK
+{
+    "og:type"        : "article"
+    "og:url"         : "https://twitter.com/matrixdotorg/status/684074366691356672"
+    "og:title"       : "Matrix on Twitter"
+    "og:image"       : "https://pbs.twimg.com/profile_images/500400952029888512/yI0qtFi7_400x400.png"
+    "og:description" : "“Synapse 0.12 is out! Lots of polishing, performance & bugfixes: /sync API, /r0 prefix, fulltext search, 3PID invites https://t.co/5alhXLLEGP”"
+    "og:site_name"   : "Twitter"
+}
+
+* Downloads the URL
+  * If HTML, just stores it in RAM and parses it for OG meta tags
+    * Download any media OG meta tags to the media repo, and refer to them in the OG via mxc:// URIs.
+  * If a media filetype we know we can thumbnail: store it on disk, and hand it to the thumbnailer. Generate OG meta tags from the thumbnailer contents.
+  * Otherwise, don't bother downloading further.

docs/workers.rst

@@ -0,0 +1,98 @@
+Scaling synapse via workers
+---------------------------
+
+Synapse has experimental support for splitting out functionality into
+multiple separate python processes, helping greatly with scalability.  These
+processes are called 'workers', and are (eventually) intended to scale
+horizontally independently.
+
+All processes continue to share the same database instance, and as such, workers
+only work with postgres based synapse deployments (sharing a single sqlite
+across multiple processes is a recipe for disaster, plus you should be using
+postgres anyway if you care about scalability).
+
+The workers communicate with the master synapse process via a synapse-specific
+HTTP protocol called 'replication' - analogous to MySQL or Postgres style
+database replication; feeding a stream of relevant data to the workers so they
+can be kept in sync with the main synapse process and database state.
+
+To enable workers, you need to add a replication listener to the master synapse, e.g.::
+
+    listeners:
+      - port: 9092
+        bind_address: '127.0.0.1'
+        type: http
+        tls: false
+        x_forwarded: false
+        resources:
+          - names: [replication]
+            compress: false
+
+Under **no circumstances** should this replication API listener be exposed to the
+public internet; it currently implements no authentication whatsoever and is
+unencrypted HTTP.
+
+You then create a set of configs for the various worker processes.  These should be
+worker configuration files should be stored in a dedicated subdirectory, to allow
+synctl to manipulate them.
+
+The current available worker applications are:
+ * synapse.app.pusher - handles sending push notifications to sygnal and email
+ * synapse.app.synchrotron - handles /sync endpoints.  can scales horizontally through multiple instances.
+ * synapse.app.appservice - handles output traffic to Application Services
+ * synapse.app.federation_reader - handles receiving federation traffic (including public_rooms API)
+ * synapse.app.media_repository - handles the media repository.
+ * synapse.app.client_reader - handles client API endpoints like /publicRooms
+
+Each worker configuration file inherits the configuration of the main homeserver
+configuration file.  You can then override configuration specific to that worker,
+e.g. the HTTP listener that it provides (if any); logging configuration; etc.
+You should minimise the number of overrides though to maintain a usable config.
+
+You must specify the type of worker application (worker_app) and the replication
+endpoint that it's talking to on the main synapse process (worker_replication_url).
+
+For instance::
+
+    worker_app: synapse.app.synchrotron
+
+    # The replication listener on the synapse to talk to.
+    worker_replication_url: http://127.0.0.1:9092/_synapse/replication
+
+    worker_listeners:
+     - type: http
+       port: 8083
+       resources:
+         - names:
+           - client
+
+    worker_daemonize: True
+    worker_pid_file: /home/matrix/synapse/synchrotron.pid
+    worker_log_config: /home/matrix/synapse/config/synchrotron_log_config.yaml
+
+...is a full configuration for a synchrotron worker instance, which will expose a
+plain HTTP /sync endpoint on port 8083 separately from the /sync endpoint provided
+by the main synapse.
+
+Obviously you should configure your loadbalancer to route the /sync endpoint to
+the synchrotron instance(s) in this instance.
+
+Finally, to actually run your worker-based synapse, you must pass synctl the -a
+commandline option to tell it to operate on all the worker configurations found
+in the given directory, e.g.::
+
+    synctl -a $CONFIG/workers start
+
+Currently one should always restart all workers when restarting or upgrading
+synapse, unless you explicitly know it's safe not to.  For instance, restarting
+synapse without restarting all the synchrotrons may result in broken typing
+notifications.
+
+To manipulate a specific worker, you pass the -w option to synctl::
+
+    synctl -w $CONFIG/workers/synchrotron.yaml restart
+
+All of the above is highly experimental and subject to change as Synapse evolves,
+but documenting it here to help folks needing highly scalable Synapses similar
+to the one running matrix.org!
+

jenkins-dendron-postgres.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+
+set -eux
+
+: ${WORKSPACE:="$(pwd)"}
+
+export WORKSPACE
+export PYTHONDONTWRITEBYTECODE=yep
+export SYNAPSE_CACHE_FACTOR=1
+
+./jenkins/prepare_synapse.sh
+./jenkins/clone.sh sytest https://github.com/matrix-org/sytest.git
+./jenkins/clone.sh dendron https://github.com/matrix-org/dendron.git
+./dendron/jenkins/build_dendron.sh
+./sytest/jenkins/prep_sytest_for_postgres.sh
+
+./sytest/jenkins/install_and_run.sh \
+    --synapse-directory $WORKSPACE \
+    --dendron $WORKSPACE/dendron/bin/dendron \
+    --pusher \
+    --synchrotron \
+    --federation-reader \
+    --client-reader \
+    --appservice \

jenkins-flake8.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+
+set -eux
+
+: ${WORKSPACE:="$(pwd)"}
+
+export PYTHONDONTWRITEBYTECODE=yep
+export SYNAPSE_CACHE_FACTOR=1
+
+# Output test results as junit xml
+export TRIAL_FLAGS="--reporter=subunit"
+export TOXSUFFIX="| subunit-1to2 | subunit2junitxml --no-passthrough --output-to=results.xml"
+# Write coverage reports to a separate file for each process
+export COVERAGE_OPTS="-p"
+export DUMP_COVERAGE_COMMAND="coverage help"
+
+# Output flake8 violations to violations.flake8.log
+export PEP8SUFFIX="--output-file=violations.flake8.log"
+
+rm .coverage* || echo "No coverage files to remove"
+
+tox -e packaging -e pep8

jenkins-postgres.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -eux
+
+: ${WORKSPACE:="$(pwd)"}
+
+export WORKSPACE
+export PYTHONDONTWRITEBYTECODE=yep
+export SYNAPSE_CACHE_FACTOR=1
+
+./jenkins/prepare_synapse.sh
+./jenkins/clone.sh sytest https://github.com/matrix-org/sytest.git
+
+./sytest/jenkins/prep_sytest_for_postgres.sh
+
+./sytest/jenkins/install_and_run.sh \
+    --synapse-directory $WORKSPACE \

jenkins-sqlite.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+set -eux
+
+: ${WORKSPACE:="$(pwd)"}
+
+export WORKSPACE
+export PYTHONDONTWRITEBYTECODE=yep
+export SYNAPSE_CACHE_FACTOR=1
+
+./jenkins/prepare_synapse.sh
+./jenkins/clone.sh sytest https://github.com/matrix-org/sytest.git
+
+./sytest/jenkins/install_and_run.sh \
+    --synapse-directory $WORKSPACE \

jenkins-unittests.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+
+set -eux
+
+: ${WORKSPACE:="$(pwd)"}
+
+export PYTHONDONTWRITEBYTECODE=yep
+export SYNAPSE_CACHE_FACTOR=1
+
+# Output test results as junit xml
+export TRIAL_FLAGS="--reporter=subunit"
+export TOXSUFFIX="| subunit-1to2 | subunit2junitxml --no-passthrough --output-to=results.xml"
+# Write coverage reports to a separate file for each process
+export COVERAGE_OPTS="-p"
+export DUMP_COVERAGE_COMMAND="coverage help"
+
+# Output flake8 violations to violations.flake8.log
+# Don't exit with non-0 status code on Jenkins,
+# so that the build steps continue and a later step can decided whether to
+# UNSTABLE or FAILURE this build.
+export PEP8SUFFIX="--output-file=violations.flake8.log || echo flake8 finished with status code \$?"
+
+rm .coverage* || echo "No coverage files to remove"
+
+tox --notest -e py27
+TOX_BIN=$WORKSPACE/.tox/py27/bin
+python synapse/python_dependencies.py | xargs -n1 $TOX_BIN/pip install
+$TOX_BIN/pip install lxml
+
+tox -e py27

jenkins/clone.sh

@@ -0,0 +1,44 @@
+#! /bin/bash
+
+# This clones a project from github into a named subdirectory
+# If the project has a branch with the same name as this branch
+# then it will checkout that branch after cloning.
+# Otherwise it will checkout "origin/develop."
+# The first argument is the name of the directory to checkout
+# the branch into.
+# The second argument is the URL of the remote repository to checkout.
+# Usually something like https://github.com/matrix-org/sytest.git
+
+set -eux
+
+NAME=$1
+PROJECT=$2
+BASE=".$NAME-base"
+
+# Update our mirror.
+if [ ! -d ".$NAME-base" ]; then
+  # Create a local mirror of the source repository.
+  # This saves us from having to download the entire repository
+  # when this script is next run.
+  git clone "$PROJECT" "$BASE" --mirror
+else
+  # Fetch any updates from the source repository.
+  (cd "$BASE"; git fetch -p)
+fi
+
+# Remove the existing repository so that we have a clean copy
+rm -rf "$NAME"
+# Cloning with --shared means that we will share portions of the
+# .git directory with our local mirror.
+git clone "$BASE" "$NAME" --shared
+
+# Jenkins may have supplied us with the name of the branch in the
+# environment. Otherwise we will have to guess based on the current
+# commit.
+: ${GIT_BRANCH:="origin/$(git rev-parse --abbrev-ref HEAD)"}
+cd "$NAME"
+# check out the relevant branch
+git checkout "${GIT_BRANCH}" || (
+    echo >&2 "No ref ${GIT_BRANCH} found, falling back to develop"
+    git checkout "origin/develop"
+)

jenkins/prepare_synapse.sh

@@ -0,0 +1,20 @@
+#! /bin/bash
+
+cd "`dirname $0`/.."
+
+TOX_DIR=$WORKSPACE/.tox
+
+mkdir -p $TOX_DIR
+
+if ! [ $TOX_DIR -ef .tox ]; then
+    ln -s "$TOX_DIR" .tox
+fi
+
+# set up the virtualenv
+tox -e py27 --notest -v
+
+TOX_BIN=$TOX_DIR/py27/bin
+$TOX_BIN/pip install setuptools
+python synapse/python_dependencies.py | xargs -n1 $TOX_BIN/pip install
+$TOX_BIN/pip install lxml
+$TOX_BIN/pip install psycopg2

res/templates/mail-Vector.css

@@ -0,0 +1,7 @@
+.header {
+    border-bottom: 4px solid #e4f7ed ! important;
+}
+
+.notif_link a, .footer a {
+    color: #76CFA6 ! important;
+}

res/templates/mail.css

@@ -0,0 +1,156 @@
+body {
+    margin: 0px;
+}
+
+pre, code {
+    word-break: break-word;
+    white-space: pre-wrap;
+}
+
+#page {
+    font-family: 'Open Sans', Helvetica, Arial, Sans-Serif;
+    font-color: #454545;
+    font-size: 12pt;
+    width: 100%;
+    padding: 20px;
+}
+
+#inner {
+    width: 640px;
+}
+
+.header {
+    width: 100%;
+    height: 87px;
+    color: #454545;
+    border-bottom: 4px solid #e5e5e5;
+}
+
+.logo {
+    text-align: right;
+    margin-left: 20px;
+}
+
+.salutation {
+    padding-top: 10px;
+    font-weight: bold;
+}
+
+.summarytext {
+}
+
+.room {
+    width: 100%;
+    color: #454545;
+    border-bottom: 1px solid #e5e5e5;
+}
+
+.room_header td {
+    padding-top: 38px;
+    padding-bottom: 10px;
+    border-bottom: 1px solid #e5e5e5;
+}
+
+.room_name {
+    vertical-align: middle;
+    font-size: 18px;
+    font-weight: bold;
+}
+
+.room_header h2 {
+    margin-top: 0px;
+    margin-left: 75px;
+    font-size: 20px;
+}
+
+.room_avatar {
+    width: 56px;
+    line-height: 0px;
+    text-align: center;
+    vertical-align: middle;
+}
+
+.room_avatar img {
+    width: 48px;
+    height: 48px;
+    object-fit: cover;
+    border-radius: 24px;
+}
+
+.notif {
+    border-bottom: 1px solid #e5e5e5;
+    margin-top: 16px;
+    padding-bottom: 16px;
+}
+
+.historical_message .sender_avatar {
+    opacity: 0.3;
+}
+
+/* spell out opacity and historical_message class names for Outlook aka Word */
+.historical_message .sender_name {
+    color: #e3e3e3;
+}
+
+.historical_message .message_time {
+    color: #e3e3e3;
+}
+
+.historical_message .message_body {
+    color: #c7c7c7;
+}
+
+.historical_message td,
+.message td {
+    padding-top: 10px;
+}
+
+.sender_avatar {
+    width: 56px;
+    text-align: center;
+    vertical-align: top;
+}
+
+.sender_avatar img {
+    margin-top: -2px;
+    width: 32px;
+    height: 32px;
+    border-radius: 16px;
+}
+
+.sender_name  {
+    display: inline;
+    font-size: 13px;
+    color: #a2a2a2;
+}
+
+.message_time  {
+    text-align: right;
+    width: 100px;
+    font-size: 11px;
+    color: #a2a2a2;
+}
+
+.message_body {
+}
+
+.notif_link td {
+    padding-top: 10px;
+    padding-bottom: 10px;
+    font-weight: bold;
+}
+
+.notif_link a, .footer a {
+    color: #454545;
+    text-decoration: none;
+}
+
+.debug {
+    font-size: 10px;
+    color: #888;
+}
+
+.footer {
+    margin-top: 20px;
+    text-align: center;
+}

res/templates/notif.html

@@ -0,0 +1,45 @@
+{% for message in notif.messages %}
+    <tr class="{{ "historical_message" if message.is_historical else "message" }}">
+        <td class="sender_avatar">
+            {% if loop.index0 == 0 or notif.messages[loop.index0 - 1].sender_name != notif.messages[loop.index0].sender_name %}
+                {% if message.sender_avatar_url %}
+                    <img alt="" class="sender_avatar" src="{{ message.sender_avatar_url|mxc_to_http(32,32) }}"  />
+                {% else %}
+                    {% if message.sender_hash % 3 == 0 %}
+                        <img class="sender_avatar" src="https://vector.im/beta/img/76cfa6.png"  />
+                    {% elif message.sender_hash % 3 == 1 %}
+                        <img class="sender_avatar" src="https://vector.im/beta/img/50e2c2.png"  />
+                    {% else %}
+                        <img class="sender_avatar" src="https://vector.im/beta/img/f4c371.png"  />
+                    {% endif %}
+                {% endif %}
+            {% endif %}
+        </td>
+        <td class="message_contents">
+            {% if loop.index0 == 0 or notif.messages[loop.index0 - 1].sender_name != notif.messages[loop.index0].sender_name %}
+                <div class="sender_name">{% if message.msgtype == "m.emote" %}*{% endif %} {{ message.sender_name }}</div>
+            {% endif %}
+            <div class="message_body">
+                {% if message.msgtype == "m.text" %}
+                    {{ message.body_text_html }}
+                {% elif message.msgtype == "m.emote" %}
+                    {{ message.body_text_html }}
+                {% elif message.msgtype == "m.notice" %}
+                    {{ message.body_text_html }}
+                {% elif message.msgtype == "m.image" %}
+                    <img src="{{ message.image_url|mxc_to_http(640, 480, scale) }}" />
+                {% elif message.msgtype == "m.file" %}
+                    <span class="filename">{{ message.body_text_plain }}</span>
+                {% endif %}
+            </div>
+        </td>
+        <td class="message_time">{{ message.ts|format_ts("%H:%M") }}</td>
+    </tr>
+{% endfor %}
+<tr class="notif_link">
+    <td></td>
+    <td>
+        <a href="{{ notif.link }}">View {{ room.title }}</a>
+    </td>
+    <td></td>
+</tr>

res/templates/notif.txt

@@ -0,0 +1,16 @@
+{% for message in notif.messages %}
+{% if message.msgtype == "m.emote" %}* {% endif %}{{ message.sender_name }} ({{ message.ts|format_ts("%H:%M") }})
+{% if message.msgtype == "m.text" %}
+{{ message.body_text_plain }}
+{% elif message.msgtype == "m.emote" %}
+{{ message.body_text_plain }}
+{% elif message.msgtype == "m.notice" %}
+{{ message.body_text_plain }}
+{% elif message.msgtype == "m.image" %}
+{{ message.body_text_plain }}
+{% elif message.msgtype == "m.file" %}
+{{ message.body_text_plain }}
+{% endif %}
+{% endfor %}
+
+View {{ room.title }} at {{ notif.link }}

res/templates/notif_mail.html

@@ -0,0 +1,55 @@
+<!doctype html>
+<html lang="en">
+    <head>
+        <style type="text/css">
+            {% include 'mail.css' without context %}
+            {% include "mail-%s.css" % app_name ignore missing without context %}
+        </style>
+    </head>
+    <body>
+        <table id="page">
+            <tr>
+                <td> </td>
+                <td id="inner">
+                    <table class="header">
+                        <tr>
+                            <td>
+                                <div class="salutation">Hi {{ user_display_name }},</div>
+                                <div class="summarytext">{{ summary_text }}</div>
+                            </td>
+                            <td class="logo">
+                                {% if app_name == "Riot" %}
+                                    <img src="http://matrix.org/img/riot-logo-email.png" width="83" height="83" alt="[Riot]"/>
+                                {% elif app_name == "Vector" %}
+                                    <img src="http://matrix.org/img/vector-logo-email.png" width="64" height="83" alt="[Vector]"/>
+                                {% else %}
+                                    <img src="http://matrix.org/img/matrix-120x51.png" width="120" height="51" alt="[matrix]"/>
+                                {% endif %}
+                            </td>
+                        </tr>
+                    </table>
+                    {% for room in rooms %}
+                        {% include 'room.html' with context %}
+                    {% endfor %}
+                    <div class="footer">
+                        <a href="{{ unsubscribe_link }}">Unsubscribe</a>
+                        <br/>
+                        <br/>
+                        <div class="debug">
+                            Sending email at {{ reason.now|format_ts("%c") }} due to activity in room {{ reason.room_name }} because
+                            an event was received at {{ reason.received_at|format_ts("%c") }}
+                            which is more than {{ "%.1f"|format(reason.delay_before_mail_ms / (60*1000)) }} ({{ reason.delay_before_mail_ms }}) mins ago,
+                            {% if reason.last_sent_ts %}
+                                and the last time we sent a mail for this room was {{ reason.last_sent_ts|format_ts("%c") }},
+                                which is more than {{ "%.1f"|format(reason.throttle_ms / (60*1000)) }} (current throttle_ms) mins ago.
+                            {% else %}
+                                and we don't have a last time we sent a mail for this room.
+                            {% endif %}
+                        </div>
+                    </div>
+                </td>
+                <td> </td>
+            </tr>
+        </table>
+    </body>
+</html>

res/templates/notif_mail.txt

@@ -0,0 +1,10 @@
+Hi {{ user_display_name }},
+
+{{ summary_text }}
+
+{% for room in rooms %}
+{% include 'room.txt' with context %}
+{% endfor %}
+
+You can disable these notifications at {{ unsubscribe_link }}
+

res/templates/room.html

@@ -0,0 +1,33 @@
+<table class="room">
+    <tr class="room_header">
+        <td class="room_avatar">
+            {% if room.avatar_url %}
+                <img alt="" src="{{ room.avatar_url|mxc_to_http(48,48) }}" />
+            {% else %}
+                {% if room.hash % 3 == 0 %}
+                    <img alt="" src="https://vector.im/beta/img/76cfa6.png"  />
+                {% elif room.hash % 3 == 1 %}
+                    <img alt="" src="https://vector.im/beta/img/50e2c2.png"  />
+                {% else %}
+                    <img alt="" src="https://vector.im/beta/img/f4c371.png"  />
+                {% endif %}
+            {% endif %}
+        </td>
+        <td class="room_name" colspan="2">
+            {{ room.title }}
+        </td>
+    </tr>
+    {% if room.invite %}
+        <tr>
+            <td></td>
+            <td>
+                <a href="{{ room.link }}">Join the conversation.</a>
+            </td>
+            <td></td>
+        </tr>
+    {% else %}
+        {% for notif in room.notifs %}
+            {% include 'notif.html' with context %}
+        {% endfor %}
+    {% endif %}
+</table>

res/templates/room.txt

@@ -0,0 +1,9 @@
+{{ room.title }}
+
+{% if room.invite %}
+    You've been invited, join at {{ room.link }}
+{% else %}
+    {% for notif in room.notifs %}
+        {% include 'notif.txt' with context %}
+    {% endfor %}
+{% endif %}

scripts-dev/check_auth.py

@@ -56,10 +56,9 @@ if __name__ == '__main__':
 
     js = json.load(args.json)
 
-
     auth = Auth(Mock())
     check_auth(
         auth,
         [FrozenEvent(d) for d in js["auth_chain"]],
-        [FrozenEvent(d) for d in js["pdus"]],
+        [FrozenEvent(d) for d in js.get("pdus", [])],
     )

scripts-dev/check_event_hash.py

@@ -1,5 +1,5 @@
 from synapse.crypto.event_signing import *
-from syutil.base64util import encode_base64
+from unpaddedbase64 import encode_base64
 
 import argparse
 import hashlib

scripts-dev/check_signature.py

@@ -1,9 +1,7 @@
 
-from syutil.crypto.jsonsign import verify_signed_json
-from syutil.crypto.signing_key import (
-    decode_verify_key_bytes, write_signing_keys
-)
-from syutil.base64util import decode_base64
+from signedjson.sign import verify_signed_json
+from signedjson.key import decode_verify_key_bytes, write_signing_keys
+from unpaddedbase64 import decode_base64
 
 import urllib2
 import json

scripts-dev/convert_server_keys.py

@@ -0,0 +1,116 @@
+import psycopg2
+import yaml
+import sys
+import json
+import time
+import hashlib
+from unpaddedbase64 import encode_base64
+from signedjson.key import read_signing_keys
+from signedjson.sign import sign_json
+from canonicaljson import encode_canonical_json
+
+
+def select_v1_keys(connection):
+    cursor = connection.cursor()
+    cursor.execute("SELECT server_name, key_id, verify_key FROM server_signature_keys")
+    rows = cursor.fetchall()
+    cursor.close()
+    results = {}
+    for server_name, key_id, verify_key in rows:
+        results.setdefault(server_name, {})[key_id] = encode_base64(verify_key)
+    return results
+
+
+def select_v1_certs(connection):
+    cursor = connection.cursor()
+    cursor.execute("SELECT server_name, tls_certificate FROM server_tls_certificates")
+    rows = cursor.fetchall()
+    cursor.close()
+    results = {}
+    for server_name, tls_certificate in rows:
+        results[server_name] = tls_certificate
+    return results
+
+
+def select_v2_json(connection):
+    cursor = connection.cursor()
+    cursor.execute("SELECT server_name, key_id, key_json FROM server_keys_json")
+    rows = cursor.fetchall()
+    cursor.close()
+    results = {}
+    for server_name, key_id, key_json in rows:
+        results.setdefault(server_name, {})[key_id] = json.loads(str(key_json).decode("utf-8"))
+    return results
+
+
+def convert_v1_to_v2(server_name, valid_until, keys, certificate):
+    return {
+        "old_verify_keys": {},
+        "server_name": server_name,
+        "verify_keys": {
+            key_id: {"key": key}
+            for key_id, key in keys.items()
+        },
+        "valid_until_ts": valid_until,
+        "tls_fingerprints": [fingerprint(certificate)],
+    }
+
+
+def fingerprint(certificate):
+    finger = hashlib.sha256(certificate)
+    return {"sha256": encode_base64(finger.digest())}
+
+
+def rows_v2(server, json):
+    valid_until = json["valid_until_ts"]
+    key_json = encode_canonical_json(json)
+    for key_id in json["verify_keys"]:
+        yield (server, key_id, "-", valid_until, valid_until, buffer(key_json))
+
+
+def main():
+    config = yaml.load(open(sys.argv[1]))
+    valid_until = int(time.time() / (3600 * 24)) * 1000 * 3600 * 24
+
+    server_name = config["server_name"]
+    signing_key = read_signing_keys(open(config["signing_key_path"]))[0]
+
+    database = config["database"]
+    assert database["name"] == "psycopg2", "Can only convert for postgresql"
+    args = database["args"]
+    args.pop("cp_max")
+    args.pop("cp_min")
+    connection = psycopg2.connect(**args)
+    keys = select_v1_keys(connection)
+    certificates = select_v1_certs(connection)
+    json = select_v2_json(connection)
+
+    result = {}
+    for server in keys:
+        if not server in json:
+            v2_json = convert_v1_to_v2(
+                server, valid_until, keys[server], certificates[server]
+            )
+            v2_json = sign_json(v2_json, server_name, signing_key)
+            result[server] = v2_json
+
+    yaml.safe_dump(result, sys.stdout, default_flow_style=False)
+
+    rows = list(
+        row for server, json in result.items()
+        for row in rows_v2(server, json)
+    )
+
+    cursor = connection.cursor()
+    cursor.executemany(
+        "INSERT INTO server_keys_json ("
+        " server_name, key_id, from_server,"
+        " ts_added_ms, ts_valid_until_ms, key_json"
+        ") VALUES (%s, %s, %s, %s, %s, %s)",
+        rows
+    )
+    connection.commit()
+
+
+if __name__ == '__main__':
+    main()

scripts-dev/copyrighter-sql.pl

@@ -1,5 +1,5 @@
 #!/usr/bin/perl -pi
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
 # limitations under the License.
 
 $copyright = <<EOT;
-/* Copyright 2015 OpenMarket Ltd
+/* Copyright 2016 OpenMarket Ltd
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

scripts-dev/copyrighter.pl

@@ -1,5 +1,5 @@
 #!/usr/bin/perl -pi
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
 # limitations under the License.
 
 $copyright = <<EOT;
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

scripts-dev/definitions.py

@@ -0,0 +1,196 @@
+#! /usr/bin/python
+
+import ast
+import yaml
+
+class DefinitionVisitor(ast.NodeVisitor):
+    def __init__(self):
+        super(DefinitionVisitor, self).__init__()
+        self.functions = {}
+        self.classes = {}
+        self.names = {}
+        self.attrs = set()
+        self.definitions = {
+            'def': self.functions,
+            'class': self.classes,
+            'names': self.names,
+            'attrs': self.attrs,
+        }
+
+    def visit_Name(self, node):
+        self.names.setdefault(type(node.ctx).__name__, set()).add(node.id)
+
+    def visit_Attribute(self, node):
+        self.attrs.add(node.attr)
+        for child in ast.iter_child_nodes(node):
+            self.visit(child)
+
+    def visit_ClassDef(self, node):
+        visitor = DefinitionVisitor()
+        self.classes[node.name] = visitor.definitions
+        for child in ast.iter_child_nodes(node):
+            visitor.visit(child)
+
+    def visit_FunctionDef(self, node):
+        visitor = DefinitionVisitor()
+        self.functions[node.name] = visitor.definitions
+        for child in ast.iter_child_nodes(node):
+            visitor.visit(child)
+
+
+def non_empty(defs):
+    functions = {name: non_empty(f) for name, f in defs['def'].items()}
+    classes = {name: non_empty(f) for name, f in defs['class'].items()}
+    result = {}
+    if functions: result['def'] = functions
+    if classes: result['class'] = classes
+    names = defs['names']
+    uses = []
+    for name in names.get('Load', ()):
+        if name not in names.get('Param', ()) and name not in names.get('Store', ()):
+            uses.append(name)
+    uses.extend(defs['attrs'])
+    if uses: result['uses'] = uses
+    result['names'] = names
+    result['attrs'] = defs['attrs']
+    return result
+
+
+def definitions_in_code(input_code):
+    input_ast = ast.parse(input_code)
+    visitor = DefinitionVisitor()
+    visitor.visit(input_ast)
+    definitions = non_empty(visitor.definitions)
+    return definitions
+
+
+def definitions_in_file(filepath):
+    with open(filepath) as f:
+        return definitions_in_code(f.read())
+
+
+def defined_names(prefix, defs, names):
+    for name, funcs in defs.get('def', {}).items():
+        names.setdefault(name, {'defined': []})['defined'].append(prefix + name)
+        defined_names(prefix + name + ".", funcs, names)
+
+    for name, funcs in defs.get('class', {}).items():
+        names.setdefault(name, {'defined': []})['defined'].append(prefix + name)
+        defined_names(prefix + name + ".", funcs, names)
+
+
+def used_names(prefix, item, defs, names):
+    for name, funcs in defs.get('def', {}).items():
+        used_names(prefix + name + ".", name, funcs, names)
+
+    for name, funcs in defs.get('class', {}).items():
+        used_names(prefix + name + ".", name, funcs, names)
+
+    path = prefix.rstrip('.')
+    for used in defs.get('uses', ()):
+        if used in names:
+            if item:
+                names[item].setdefault('uses', []).append(used)
+            names[used].setdefault('used', {}).setdefault(item, []).append(path)
+
+
+if __name__ == '__main__':
+    import sys, os, argparse, re
+
+    parser = argparse.ArgumentParser(description='Find definitions.')
+    parser.add_argument(
+        "--unused", action="store_true", help="Only list unused definitions"
+    )
+    parser.add_argument(
+        "--ignore", action="append", metavar="REGEXP", help="Ignore a pattern"
+    )
+    parser.add_argument(
+        "--pattern", action="append", metavar="REGEXP",
+        help="Search for a pattern"
+    )
+    parser.add_argument(
+        "directories", nargs='+', metavar="DIR",
+        help="Directories to search for definitions"
+    )
+    parser.add_argument(
+        "--referrers", default=0, type=int,
+        help="Include referrers up to the given depth"
+    )
+    parser.add_argument(
+        "--referred", default=0, type=int,
+        help="Include referred down to the given depth"
+    )
+    parser.add_argument(
+        "--format", default="yaml",
+        help="Output format, one of 'yaml' or 'dot'"
+    )
+    args = parser.parse_args()
+
+    definitions = {}
+    for directory in args.directories:
+        for root, dirs, files in os.walk(directory):
+            for filename in files:
+                if filename.endswith(".py"):
+                    filepath = os.path.join(root, filename)
+                    definitions[filepath] = definitions_in_file(filepath)
+
+    names = {}
+    for filepath, defs in definitions.items():
+        defined_names(filepath + ":", defs, names)
+
+    for filepath, defs in definitions.items():
+        used_names(filepath + ":", None, defs, names)
+
+    patterns = [re.compile(pattern) for pattern in args.pattern or ()]
+    ignore = [re.compile(pattern) for pattern in args.ignore or ()]
+
+    result = {}
+    for name, definition in names.items():
+        if patterns and not any(pattern.match(name) for pattern in patterns):
+            continue
+        if ignore and any(pattern.match(name) for pattern in ignore):
+            continue
+        if args.unused and definition.get('used'):
+            continue
+        result[name] = definition
+
+    referrer_depth = args.referrers
+    referrers = set()
+    while referrer_depth:
+        referrer_depth -= 1
+        for entry in result.values():
+            for used_by in entry.get("used", ()):
+                referrers.add(used_by)
+        for name, definition in names.items():
+            if not name in referrers:
+                continue
+            if ignore and any(pattern.match(name) for pattern in ignore):
+                continue
+            result[name] = definition
+
+    referred_depth = args.referred
+    referred = set()
+    while referred_depth:
+        referred_depth -= 1
+        for entry in result.values():
+            for uses in entry.get("uses", ()):
+                referred.add(uses)
+        for name, definition in names.items():
+            if not name in referred:
+                continue
+            if ignore and any(pattern.match(name) for pattern in ignore):
+                continue
+            result[name] = definition
+
+    if args.format == 'yaml':
+        yaml.dump(result, sys.stdout, default_flow_style=False)
+    elif args.format == 'dot':
+        print "digraph {"
+        for name, entry in result.items():
+            print name
+            for used_by in entry.get("used", ()):
+                if used_by in result:
+                    print used_by, "->", name
+        print "}"
+    else:
+        raise ValueError("Unknown format %r" % (args.format))

scripts-dev/dump_macaroon.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python2
+
+import pymacaroons
+import sys
+
+if len(sys.argv) == 1:
+    sys.stderr.write("usage: %s macaroon [key]\n" % (sys.argv[0],))
+    sys.exit(1)
+
+macaroon_string = sys.argv[1]
+key = sys.argv[2] if len(sys.argv) > 2 else None
+
+macaroon = pymacaroons.Macaroon.deserialize(macaroon_string)
+print macaroon.inspect()
+
+print ""
+
+verifier = pymacaroons.Verifier()
+verifier.satisfy_general(lambda c: True)
+try:
+    verifier.verify(macaroon, key)
+    print "Signature is correct"
+except Exception as e:
+    print e.message

scripts-dev/federation_client.py

@@ -116,17 +116,19 @@ def get_json(origin_name, origin_key, destination, path):
     authorization_headers = []
 
     for key, sig in signed_json["signatures"][origin_name].items():
-        authorization_headers.append(bytes(
-            "X-Matrix origin=%s,key=\"%s\",sig=\"%s\"" % (
-                origin_name, key, sig,
-            )
-        ))
+        header = "X-Matrix origin=%s,key=\"%s\",sig=\"%s\"" % (
+            origin_name, key, sig,
+        )
+        authorization_headers.append(bytes(header))
+        sys.stderr.write(header)
+        sys.stderr.write("\n")
 
     result = requests.get(
         lookup(destination, path),
         headers={"Authorization": authorization_headers[0]},
         verify=False,
     )
+    sys.stderr.write("Status Code: %d\n" % (result.status_code,))
     return result.json()
 
 
@@ -141,6 +143,7 @@ def main():
     )
 
     json.dump(result, sys.stdout)
+    print ""
 
 if __name__ == "__main__":
     main()

scripts-dev/hash_history.py

@@ -6,8 +6,8 @@ from synapse.crypto.event_signing import (
     add_event_pdu_content_hash, compute_pdu_event_reference_hash
 )
 from synapse.api.events.utils import prune_pdu
-from syutil.base64util import encode_base64, decode_base64
-from syutil.jsonutil import encode_canonical_json
+from unpaddedbase64 import encode_base64, decode_base64
+from canonicaljson import encode_canonical_json
 import sqlite3
 import sys
 

scripts-dev/list_url_patterns.py

@@ -0,0 +1,62 @@
+#! /usr/bin/python
+
+import ast
+import argparse
+import os
+import sys
+import yaml
+
+PATTERNS_V1 = []
+PATTERNS_V2 = []
+
+RESULT = {
+    "v1": PATTERNS_V1,
+    "v2": PATTERNS_V2,
+}
+
+class CallVisitor(ast.NodeVisitor):
+    def visit_Call(self, node):
+        if isinstance(node.func, ast.Name):
+            name = node.func.id
+        else:
+            return
+
+
+        if name == "client_path_patterns":
+            PATTERNS_V1.append(node.args[0].s)
+        elif name == "client_v2_patterns":
+            PATTERNS_V2.append(node.args[0].s)
+
+
+def find_patterns_in_code(input_code):
+    input_ast = ast.parse(input_code)
+    visitor = CallVisitor()
+    visitor.visit(input_ast)
+
+
+def find_patterns_in_file(filepath):
+    with open(filepath) as f:
+        find_patterns_in_code(f.read())
+
+
+parser = argparse.ArgumentParser(description='Find url patterns.')
+
+parser.add_argument(
+    "directories", nargs='+', metavar="DIR",
+    help="Directories to search for definitions"
+)
+
+args = parser.parse_args()
+
+
+for directory in args.directories:
+    for root, dirs, files in os.walk(directory):
+        for filename in files:
+            if filename.endswith(".py"):
+                filepath = os.path.join(root, filename)
+                find_patterns_in_file(filepath)
+
+PATTERNS_V1.sort()
+PATTERNS_V2.sort()
+
+yaml.dump(RESULT, sys.stdout, default_flow_style=False)

scripts-dev/tail-synapse.py

@@ -0,0 +1,67 @@
+import requests
+import collections
+import sys
+import time
+import json
+
+Entry = collections.namedtuple("Entry", "name position rows")
+
+ROW_TYPES = {}
+
+
+def row_type_for_columns(name, column_names):
+    column_names = tuple(column_names)
+    row_type = ROW_TYPES.get((name, column_names))
+    if row_type is None:
+        row_type = collections.namedtuple(name, column_names)
+        ROW_TYPES[(name, column_names)] = row_type
+    return row_type
+
+
+def parse_response(content):
+    streams = json.loads(content)
+    result = {}
+    for name, value in streams.items():
+        row_type = row_type_for_columns(name, value["field_names"])
+        position = value["position"]
+        rows = [row_type(*row) for row in value["rows"]]
+        result[name] = Entry(name, position, rows)
+    return result
+
+
+def replicate(server, streams):
+    return parse_response(requests.get(
+        server + "/_synapse/replication",
+        verify=False,
+        params=streams
+    ).content)
+
+
+def main():
+    server = sys.argv[1]
+
+    streams = None
+    while not streams:
+        try:
+            streams = {
+                row.name: row.position
+                for row in replicate(server, {"streams":"-1"})["streams"].rows
+            }
+        except requests.exceptions.ConnectionError as e:
+            time.sleep(0.1)
+
+    while True:
+        try:
+            results = replicate(server, streams)
+        except:
+            sys.stdout.write("connection_lost("+ repr(streams) + ")\n")
+            break
+        for update in results.values():
+            for row in update.rows:
+                sys.stdout.write(repr(row) + "\n")
+            streams[update.name] = update.position
+
+
+
+if __name__=='__main__':
+    main()

scripts/database-prepare-for-0.0.1.sh

@@ -1,21 +0,0 @@
-#!/bin/bash
-
-# This is will prepare a synapse database for running with v0.0.1 of synapse. 
-# It will store all the user information, but will *delete* all messages and
-# room data.
-
-set -e
-
-cp "$1" "$1.bak"
-
-DUMP=$(sqlite3 "$1" << 'EOF'
-.dump users
-.dump access_tokens
-.dump presence
-.dump profiles
-EOF
-)
-
-rm "$1"
-
-sqlite3 "$1" <<< "$DUMP" 

scripts/database-prepare-for-0.5.0.sh

@@ -1,21 +0,0 @@
-#!/bin/bash
-
-# This is will prepare a synapse database for running with v0.5.0 of synapse. 
-# It will store all the user information, but will *delete* all messages and
-# room data.
-
-set -e
-
-cp "$1" "$1.bak"
-
-DUMP=$(sqlite3 "$1" << 'EOF'
-.dump users
-.dump access_tokens
-.dump presence
-.dump profiles
-EOF
-)
-
-rm "$1"
-
-sqlite3 "$1" <<< "$DUMP" 

scripts/hash_password

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+
+import argparse
+
+import sys
+
+import bcrypt
+import getpass
+
+import yaml
+
+bcrypt_rounds=12
+password_pepper = ""
+
+def prompt_for_pass():
+    password = getpass.getpass("Password: ")
+
+    if not password:
+        raise Exception("Password cannot be blank.")
+
+    confirm_password = getpass.getpass("Confirm password: ")
+
+    if password != confirm_password:
+        raise Exception("Passwords do not match.")
+
+    return password
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Calculate the hash of a new password, so that passwords"
+                    " can be reset")
+    parser.add_argument(
+        "-p", "--password",
+        default=None,
+        help="New password for user. Will prompt if omitted.",
+    )
+    parser.add_argument(
+        "-c", "--config",
+        type=argparse.FileType('r'),
+        help="Path to server config file. Used to read in bcrypt_rounds and password_pepper.",
+    )
+
+    args = parser.parse_args()
+    if "config" in args and args.config:
+        config = yaml.safe_load(args.config)
+        bcrypt_rounds = config.get("bcrypt_rounds", bcrypt_rounds)
+        password_config = config.get("password_config", {})
+        password_pepper = password_config.get("pepper", password_pepper)
+    password = args.password
+
+    if not password:
+        password = prompt_for_pass()
+
+    print bcrypt.hashpw(password + password_pepper, bcrypt.gensalt(bcrypt_rounds))
+

scripts/register_new_matrix_user

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -25,17 +25,26 @@ import urllib2
 import yaml
 
 
-def request_registration(user, password, server_location, shared_secret):
+def request_registration(user, password, server_location, shared_secret, admin=False):
     mac = hmac.new(
         key=shared_secret,
-        msg=user,
         digestmod=hashlib.sha1,
-    ).hexdigest()
+    )
+
+    mac.update(user)
+    mac.update("\x00")
+    mac.update(password)
+    mac.update("\x00")
+    mac.update("admin" if admin else "notadmin")
+
+    mac = mac.hexdigest()
 
     data = {
-        "username": user,
+        "user": user,
         "password": password,
         "mac": mac,
+        "type": "org.matrix.login.shared_secret",
+        "admin": admin,
     }
 
     server_location = server_location.rstrip("/")
@@ -43,7 +52,7 @@ def request_registration(user, password, server_location, shared_secret):
     print "Sending registration request..."
 
     req = urllib2.Request(
-        "%s/_matrix/client/v2_alpha/register" % (server_location,),
+        "%s/_matrix/client/api/v1/register" % (server_location,),
         data=json.dumps(data),
         headers={'Content-Type': 'application/json'}
     )
@@ -67,7 +76,7 @@ def request_registration(user, password, server_location, shared_secret):
         sys.exit(1)
 
 
-def register_new_user(user, password, server_location, shared_secret):
+def register_new_user(user, password, server_location, shared_secret, admin):
     if not user:
         try:
             default_user = getpass.getuser()
@@ -98,7 +107,14 @@ def register_new_user(user, password, server_location, shared_secret):
             print "Passwords do not match"
             sys.exit(1)
 
-    request_registration(user, password, server_location, shared_secret)
+    if not admin:
+        admin = raw_input("Make admin [no]: ")
+        if admin in ("y", "yes", "true"):
+            admin = True
+        else:
+            admin = False
+
+    request_registration(user, password, server_location, shared_secret, bool(admin))
 
 
 if __name__ == "__main__":
@@ -118,6 +134,11 @@ if __name__ == "__main__":
         default=None,
         help="New password for user. Will prompt if omitted.",
     )
+    parser.add_argument(
+        "-a", "--admin",
+        action="store_true",
+        help="Register new user as an admin. Will prompt if omitted.",
+    )
 
     group = parser.add_mutually_exclusive_group(required=True)
     group.add_argument(
@@ -150,4 +171,4 @@ if __name__ == "__main__":
     else:
         secret = args.shared_secret
 
-    register_new_user(args.user, args.password, args.server_url, secret)
+    register_new_user(args.user, args.password, args.server_url, secret, args.admin)

scripts/synapse_port_db

@@ -1,5 +1,6 @@
+#!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -18,6 +19,7 @@ from twisted.enterprise import adbapi
 
 from synapse.storage._base import LoggingTransaction, SQLBaseStore
 from synapse.storage.engines import create_engine
+from synapse.storage.prepare_database import prepare_database
 
 import argparse
 import curses
@@ -28,14 +30,16 @@ import traceback
 import yaml
 
 
-logger = logging.getLogger("port_from_sqlite_to_postgres")
+logger = logging.getLogger("synapse_port_db")
 
 
 BOOLEAN_COLUMNS = {
-    "events": ["processed", "outlier"],
+    "events": ["processed", "outlier", "contains_url"],
     "rooms": ["is_public"],
     "event_edges": ["is_state"],
     "presence_list": ["accepted"],
+    "presence_stream": ["currently_active"],
+    "public_room_list_stream": ["visibility"],
 }
 
 
@@ -67,6 +71,15 @@ APPEND_ONLY_TABLES = [
     "state_groups_state",
     "event_to_state_groups",
     "rejections",
+    "event_search",
+    "presence_stream",
+    "push_rules_stream",
+    "current_state_resets",
+    "ex_outlier_stream",
+    "cache_invalidation_stream",
+    "public_room_list_stream",
+    "state_group_edges",
+    "stream_ordering_to_exterm",
 ]
 
 
@@ -88,14 +101,16 @@ class Store(object):
 
     _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"]
     _simple_select_onecol = SQLBaseStore.__dict__["_simple_select_onecol"]
+    _simple_select_one = SQLBaseStore.__dict__["_simple_select_one"]
+    _simple_select_one_txn = SQLBaseStore.__dict__["_simple_select_one_txn"]
     _simple_select_one_onecol = SQLBaseStore.__dict__["_simple_select_one_onecol"]
-    _simple_select_one_onecol_txn = SQLBaseStore.__dict__["_simple_select_one_onecol_txn"]
+    _simple_select_one_onecol_txn = SQLBaseStore.__dict__[
+        "_simple_select_one_onecol_txn"
+    ]
 
     _simple_update_one = SQLBaseStore.__dict__["_simple_update_one"]
     _simple_update_one_txn = SQLBaseStore.__dict__["_simple_update_one_txn"]
 
-    _execute_and_decode = SQLBaseStore.__dict__["_execute_and_decode"]
-
     def runInteraction(self, desc, func, *args, **kwargs):
         def r(conn):
             try:
@@ -105,7 +120,7 @@ class Store(object):
                     try:
                         txn = conn.cursor()
                         return func(
-                            LoggingTransaction(txn, desc, self.database_engine),
+                            LoggingTransaction(txn, desc, self.database_engine, []),
                             *args, **kwargs
                         )
                     except self.database_engine.module.DatabaseError as e:
@@ -156,31 +171,40 @@ class Porter(object):
     def setup_table(self, table):
         if table in APPEND_ONLY_TABLES:
             # It's safe to just carry on inserting.
-            next_chunk = yield self.postgres_store._simple_select_one_onecol(
+            row = yield self.postgres_store._simple_select_one(
                 table="port_from_sqlite3",
                 keyvalues={"table_name": table},
-                retcol="rowid",
+                retcols=("forward_rowid", "backward_rowid"),
                 allow_none=True,
             )
 
             total_to_port = None
-            if next_chunk is None:
+            if row is None:
                 if table == "sent_transactions":
-                    next_chunk, already_ported, total_to_port = (
+                    forward_chunk, already_ported, total_to_port = (
                         yield self._setup_sent_transactions()
                     )
+                    backward_chunk = 0
                 else:
                     yield self.postgres_store._simple_insert(
                         table="port_from_sqlite3",
-                        values={"table_name": table, "rowid": 1}
+                        values={
+                            "table_name": table,
+                            "forward_rowid": 1,
+                            "backward_rowid": 0,
+                        }
                     )
 
-                    next_chunk = 1
+                    forward_chunk = 1
+                    backward_chunk = 0
                     already_ported = 0
+            else:
+                forward_chunk = row["forward_rowid"]
+                backward_chunk = row["backward_rowid"]
 
             if total_to_port is None:
                 already_ported, total_to_port = yield self._get_total_count_to_port(
-                    table, next_chunk
+                    table, forward_chunk, backward_chunk
                 )
         else:
             def delete_all(txn):
@@ -194,42 +218,85 @@ class Porter(object):
 
             yield self.postgres_store._simple_insert(
                 table="port_from_sqlite3",
-                values={"table_name": table, "rowid": 0}
+                values={
+                    "table_name": table,
+                    "forward_rowid": 1,
+                    "backward_rowid": 0,
+                }
             )
 
-            next_chunk = 1
+            forward_chunk = 1
+            backward_chunk = 0
 
             already_ported, total_to_port = yield self._get_total_count_to_port(
-                table, next_chunk
+                table, forward_chunk, backward_chunk
             )
 
-        defer.returnValue((table, already_ported, total_to_port, next_chunk))
+        defer.returnValue(
+            (table, already_ported, total_to_port, forward_chunk, backward_chunk)
+        )
 
     @defer.inlineCallbacks
-    def handle_table(self, table, postgres_size, table_size, next_chunk):
+    def handle_table(self, table, postgres_size, table_size, forward_chunk,
+                     backward_chunk):
         if not table_size:
             return
 
         self.progress.add_table(table, postgres_size, table_size)
 
-        select = (
+        if table == "event_search":
+            yield self.handle_search_table(
+                postgres_size, table_size, forward_chunk, backward_chunk
+            )
+            return
+
+        forward_select = (
             "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?"
             % (table,)
         )
 
+        backward_select = (
+            "SELECT rowid, * FROM %s WHERE rowid <= ? ORDER BY rowid LIMIT ?"
+            % (table,)
+        )
+
+        do_forward = [True]
+        do_backward = [True]
+
         while True:
             def r(txn):
-                txn.execute(select, (next_chunk, self.batch_size,))
-                rows = txn.fetchall()
-                headers = [column[0] for column in txn.description]
+                forward_rows = []
+                backward_rows = []
+                if do_forward[0]:
+                    txn.execute(forward_select, (forward_chunk, self.batch_size,))
+                    forward_rows = txn.fetchall()
+                    if not forward_rows:
+                        do_forward[0] = False
 
-                return headers, rows
+                if do_backward[0]:
+                    txn.execute(backward_select, (backward_chunk, self.batch_size,))
+                    backward_rows = txn.fetchall()
+                    if not backward_rows:
+                        do_backward[0] = False
 
-            headers, rows = yield self.sqlite_store.runInteraction("select", r)
+                if forward_rows or backward_rows:
+                    headers = [column[0] for column in txn.description]
+                else:
+                    headers = None
 
-            if rows:
-                next_chunk = rows[-1][0] + 1
+                return headers, forward_rows, backward_rows
 
+            headers, frows, brows = yield self.sqlite_store.runInteraction(
+                "select", r
+            )
+
+            if frows or brows:
+                if frows:
+                    forward_chunk = max(row[0] for row in frows) + 1
+                if brows:
+                    backward_chunk = min(row[0] for row in brows) - 1
+
+                rows = frows + brows
                 self._convert_rows(table, headers, rows)
 
                 def insert(txn):
@@ -241,7 +308,10 @@ class Porter(object):
                         txn,
                         table="port_from_sqlite3",
                         keyvalues={"table_name": table},
-                        updatevalues={"rowid": next_chunk},
+                        updatevalues={
+                            "forward_rowid": forward_chunk,
+                            "backward_rowid": backward_chunk,
+                        },
                     )
 
                 yield self.postgres_store.execute(insert)
@@ -252,6 +322,76 @@ class Porter(object):
             else:
                 return
 
+    @defer.inlineCallbacks
+    def handle_search_table(self, postgres_size, table_size, forward_chunk,
+                            backward_chunk):
+        select = (
+            "SELECT es.rowid, es.*, e.origin_server_ts, e.stream_ordering"
+            " FROM event_search as es"
+            " INNER JOIN events AS e USING (event_id, room_id)"
+            " WHERE es.rowid >= ?"
+            " ORDER BY es.rowid LIMIT ?"
+        )
+
+        while True:
+            def r(txn):
+                txn.execute(select, (forward_chunk, self.batch_size,))
+                rows = txn.fetchall()
+                headers = [column[0] for column in txn.description]
+
+                return headers, rows
+
+            headers, rows = yield self.sqlite_store.runInteraction("select", r)
+
+            if rows:
+                forward_chunk = rows[-1][0] + 1
+
+                # We have to treat event_search differently since it has a
+                # different structure in the two different databases.
+                def insert(txn):
+                    sql = (
+                        "INSERT INTO event_search (event_id, room_id, key,"
+                        " sender, vector, origin_server_ts, stream_ordering)"
+                        " VALUES (?,?,?,?,to_tsvector('english', ?),?,?)"
+                    )
+
+                    rows_dict = [
+                        dict(zip(headers, row))
+                        for row in rows
+                    ]
+
+                    txn.executemany(sql, [
+                        (
+                            row["event_id"],
+                            row["room_id"],
+                            row["key"],
+                            row["sender"],
+                            row["value"],
+                            row["origin_server_ts"],
+                            row["stream_ordering"],
+                        )
+                        for row in rows_dict
+                    ])
+
+                    self.postgres_store._simple_update_one_txn(
+                        txn,
+                        table="port_from_sqlite3",
+                        keyvalues={"table_name": "event_search"},
+                        updatevalues={
+                            "forward_rowid": forward_chunk,
+                            "backward_rowid": backward_chunk,
+                        },
+                    )
+
+                yield self.postgres_store.execute(insert)
+
+                postgres_size += len(rows)
+
+                self.progress.update("event_search", postgres_size)
+
+            else:
+                return
+
     def setup_db(self, db_config, database_engine):
         db_conn = database_engine.module.connect(
             **{
@@ -260,7 +400,7 @@ class Porter(object):
             }
         )
 
-        database_engine.prepare_database(db_conn)
+        prepare_database(db_conn, database_engine, config=None)
 
         db_conn.commit()
 
@@ -277,8 +417,8 @@ class Porter(object):
                 **self.postgres_config["args"]
             )
 
-            sqlite_engine = create_engine("sqlite3")
-            postgres_engine = create_engine("psycopg2")
+            sqlite_engine = create_engine(sqlite_config)
+            postgres_engine = create_engine(postgres_config)
 
             self.sqlite_store = Store(sqlite_db_pool, sqlite_engine)
             self.postgres_store = Store(postgres_db_pool, postgres_engine)
@@ -322,10 +462,32 @@ class Porter(object):
                 txn.execute(
                     "CREATE TABLE port_from_sqlite3 ("
                     " table_name varchar(100) NOT NULL UNIQUE,"
-                    " rowid bigint NOT NULL"
+                    " forward_rowid bigint NOT NULL,"
+                    " backward_rowid bigint NOT NULL"
                     ")"
                 )
 
+            # The old port script created a table with just a "rowid" column.
+            # We want people to be able to rerun this script from an old port
+            # so that they can pick up any missing events that were not
+            # ported across.
+            def alter_table(txn):
+                txn.execute(
+                    "ALTER TABLE IF EXISTS port_from_sqlite3"
+                    " RENAME rowid TO forward_rowid"
+                )
+                txn.execute(
+                    "ALTER TABLE IF EXISTS port_from_sqlite3"
+                    " ADD backward_rowid bigint NOT NULL DEFAULT 0"
+                )
+
+            try:
+                yield self.postgres_store.runInteraction(
+                    "alter_table", alter_table
+                )
+            except Exception as e:
+                logger.info("Failed to create port table: %s", e)
+
             try:
                 yield self.postgres_store.runInteraction(
                     "create_port_table", create_port_table
@@ -377,9 +539,7 @@ class Porter(object):
 
         for i, row in enumerate(rows):
             rows[i] = tuple(
-                self.postgres_store.database_engine.encode_parameter(
-                    conv(j, col)
-                )
+                conv(j, col)
                 for j, col in enumerate(row)
                 if j > 0
             )
@@ -387,7 +547,7 @@ class Porter(object):
     @defer.inlineCallbacks
     def _setup_sent_transactions(self):
         # Only save things from the last day
-        yesterday = int(time.time()*1000) - 86400000
+        yesterday = int(time.time() * 1000) - 86400000
 
         # And save the max transaction id from each destination
         select = (
@@ -413,14 +573,17 @@ class Porter(object):
         self._convert_rows("sent_transactions", headers, rows)
 
         inserted_rows = len(rows)
-        max_inserted_rowid = max(r[0] for r in rows)
+        if inserted_rows:
+            max_inserted_rowid = max(r[0] for r in rows)
 
-        def insert(txn):
-            self.postgres_store.insert_many_txn(
-                txn, "sent_transactions", headers[1:], rows
-            )
+            def insert(txn):
+                self.postgres_store.insert_many_txn(
+                    txn, "sent_transactions", headers[1:], rows
+                )
 
-        yield self.postgres_store.execute(insert)
+            yield self.postgres_store.execute(insert)
+        else:
+            max_inserted_rowid = 0
 
         def get_start_id(txn):
             txn.execute(
@@ -440,7 +603,11 @@ class Porter(object):
 
         yield self.postgres_store._simple_insert(
             table="port_from_sqlite3",
-            values={"table_name": "sent_transactions", "rowid": next_chunk}
+            values={
+                "table_name": "sent_transactions",
+                "forward_rowid": next_chunk,
+                "backward_rowid": 0,
+            }
         )
 
         def get_sent_table_size(txn):
@@ -461,13 +628,18 @@ class Porter(object):
         defer.returnValue((next_chunk, inserted_rows, total_count))
 
     @defer.inlineCallbacks
-    def _get_remaining_count_to_port(self, table, next_chunk):
-        rows = yield self.sqlite_store.execute_sql(
+    def _get_remaining_count_to_port(self, table, forward_chunk, backward_chunk):
+        frows = yield self.sqlite_store.execute_sql(
             "SELECT count(*) FROM %s WHERE rowid >= ?" % (table,),
-            next_chunk,
+            forward_chunk,
         )
 
-        defer.returnValue(rows[0][0])
+        brows = yield self.sqlite_store.execute_sql(
+            "SELECT count(*) FROM %s WHERE rowid <= ?" % (table,),
+            backward_chunk,
+        )
+
+        defer.returnValue(frows[0][0] + brows[0][0])
 
     @defer.inlineCallbacks
     def _get_already_ported_count(self, table):
@@ -478,10 +650,10 @@ class Porter(object):
         defer.returnValue(rows[0][0])
 
     @defer.inlineCallbacks
-    def _get_total_count_to_port(self, table, next_chunk):
+    def _get_total_count_to_port(self, table, forward_chunk, backward_chunk):
         remaining, done = yield defer.gatherResults(
             [
-                self._get_remaining_count_to_port(table, next_chunk),
+                self._get_remaining_count_to_port(table, forward_chunk, backward_chunk),
                 self._get_already_ported_count(table),
             ],
             consumeErrors=True,
@@ -612,7 +784,7 @@ class CursesProgress(Progress):
             color = curses.color_pair(2) if perc == 100 else curses.color_pair(1)
 
             self.stdscr.addstr(
-                i+2, left_margin + max_len - len(table),
+                i + 2, left_margin + max_len - len(table),
                 table,
                 curses.A_BOLD | color,
             )
@@ -620,18 +792,18 @@ class CursesProgress(Progress):
             size = 20
 
             progress = "[%s%s]" % (
-                "#" * int(perc*size/100),
-                " " * (size - int(perc*size/100)),
+                "#" * int(perc * size / 100),
+                " " * (size - int(perc * size / 100)),
             )
 
             self.stdscr.addstr(
-                i+2, left_margin + max_len + middle_space,
+                i + 2, left_margin + max_len + middle_space,
                 "%s %3d%% (%d/%d)" % (progress, perc, data["num_done"], data["total"]),
             )
 
         if self.finished:
             self.stdscr.addstr(
-                rows-1, 0,
+                rows - 1, 0,
                 "Press any key to exit...",
             )
 
@@ -724,6 +896,9 @@ if __name__ == "__main__":
 
     postgres_config = yaml.safe_load(args.postgres_config)
 
+    if "database" in postgres_config:
+        postgres_config = postgres_config["database"]
+
     if "name" not in postgres_config:
         sys.stderr.write("Malformed database config: no 'name'")
         sys.exit(2)

scripts/upgrade_db_to_v0.6.0.py

@@ -1,331 +0,0 @@
-
-from synapse.storage import SCHEMA_VERSION, read_schema
-from synapse.storage._base import SQLBaseStore
-from synapse.storage.signatures import SignatureStore
-from synapse.storage.event_federation import EventFederationStore
-
-from syutil.base64util import encode_base64, decode_base64
-
-from synapse.crypto.event_signing import compute_event_signature
-
-from synapse.events.builder import EventBuilder
-from synapse.events.utils import prune_event
-
-from synapse.crypto.event_signing import check_event_content_hash
-
-from syutil.crypto.jsonsign import (
-    verify_signed_json, SignatureVerifyException,
-)
-from syutil.crypto.signing_key import decode_verify_key_bytes
-
-from syutil.jsonutil import encode_canonical_json
-
-import argparse
-# import dns.resolver
-import hashlib
-import httplib
-import json
-import sqlite3
-import syutil
-import urllib2
-
-
-delta_sql = """
-CREATE TABLE IF NOT EXISTS event_json(
-    event_id TEXT NOT NULL,
-    room_id TEXT NOT NULL,
-    internal_metadata NOT NULL,
-    json BLOB NOT NULL,
-    CONSTRAINT ev_j_uniq UNIQUE (event_id)
-);
-
-CREATE INDEX IF NOT EXISTS event_json_id ON event_json(event_id);
-CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id);
-
-PRAGMA user_version = 10;
-"""
-
-
-class Store(object):
-    _get_event_signatures_txn = SignatureStore.__dict__["_get_event_signatures_txn"]
-    _get_event_content_hashes_txn = SignatureStore.__dict__["_get_event_content_hashes_txn"]
-    _get_event_reference_hashes_txn = SignatureStore.__dict__["_get_event_reference_hashes_txn"]
-    _get_prev_event_hashes_txn = SignatureStore.__dict__["_get_prev_event_hashes_txn"]
-    _get_prev_events_and_state = EventFederationStore.__dict__["_get_prev_events_and_state"]
-    _get_auth_events = EventFederationStore.__dict__["_get_auth_events"]
-    cursor_to_dict = SQLBaseStore.__dict__["cursor_to_dict"]
-    _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"]
-    _simple_select_list_txn = SQLBaseStore.__dict__["_simple_select_list_txn"]
-    _simple_insert_txn = SQLBaseStore.__dict__["_simple_insert_txn"]
-
-    def _generate_event_json(self, txn, rows):
-        events = []
-        for row in rows:
-            d = dict(row)
-
-            d.pop("stream_ordering", None)
-            d.pop("topological_ordering", None)
-            d.pop("processed", None)
-
-            if "origin_server_ts" not in d:
-                d["origin_server_ts"] = d.pop("ts", 0)
-            else:
-                d.pop("ts", 0)
-
-            d.pop("prev_state", None)
-            d.update(json.loads(d.pop("unrecognized_keys")))
-
-            d["sender"] = d.pop("user_id")
-
-            d["content"] = json.loads(d["content"])
-
-            if "age_ts" not in d:
-                # For compatibility
-                d["age_ts"] = d.get("origin_server_ts", 0)
-
-            d.setdefault("unsigned", {})["age_ts"] = d.pop("age_ts")
-
-            outlier = d.pop("outlier", False)
-
-            # d.pop("membership", None)
-
-            d.pop("state_hash", None)
-
-            d.pop("replaces_state", None)
-
-            b = EventBuilder(d)
-            b.internal_metadata.outlier = outlier
-
-            events.append(b)
-
-        for i, ev in enumerate(events):
-            signatures = self._get_event_signatures_txn(
-                txn, ev.event_id,
-            )
-
-            ev.signatures = {
-                n: {
-                    k: encode_base64(v) for k, v in s.items()
-                }
-                for n, s in signatures.items()
-            }
-
-            hashes = self._get_event_content_hashes_txn(
-                txn, ev.event_id,
-            )
-
-            ev.hashes = {
-                k: encode_base64(v) for k, v in hashes.items()
-            }
-
-            prevs = self._get_prev_events_and_state(txn, ev.event_id)
-
-            ev.prev_events = [
-                (e_id, h)
-                for e_id, h, is_state in prevs
-                if is_state == 0
-            ]
-
-            # ev.auth_events = self._get_auth_events(txn, ev.event_id)
-
-            hashes = dict(ev.auth_events)
-
-            for e_id, hash in ev.prev_events:
-                if e_id in hashes and not hash:
-                    hash.update(hashes[e_id])
-            #
-            # if hasattr(ev, "state_key"):
-            #     ev.prev_state = [
-            #         (e_id, h)
-            #         for e_id, h, is_state in prevs
-            #         if is_state == 1
-            #     ]
-
-        return [e.build() for e in events]
-
-
-store = Store()
-
-
-# def get_key(server_name):
-#     print "Getting keys for: %s" % (server_name,)
-#     targets = []
-#     if ":" in server_name:
-#         target, port = server_name.split(":")
-#         targets.append((target, int(port)))
-#     try:
-#         answers = dns.resolver.query("_matrix._tcp." + server_name, "SRV")
-#         for srv in answers:
-#             targets.append((srv.target, srv.port))
-#     except dns.resolver.NXDOMAIN:
-#         targets.append((server_name, 8448))
-#     except:
-#         print "Failed to lookup keys for %s" % (server_name,)
-#         return {}
-#
-#     for target, port in targets:
-#         url = "https://%s:%i/_matrix/key/v1" % (target, port)
-#         try:
-#             keys = json.load(urllib2.urlopen(url, timeout=2))
-#             verify_keys = {}
-#             for key_id, key_base64 in keys["verify_keys"].items():
-#                 verify_key = decode_verify_key_bytes(
-#                     key_id, decode_base64(key_base64)
-#                 )
-#                 verify_signed_json(keys, server_name, verify_key)
-#                 verify_keys[key_id] = verify_key
-#             print "Got keys for: %s" % (server_name,)
-#             return verify_keys
-#         except urllib2.URLError:
-#             pass
-#         except urllib2.HTTPError:
-#             pass
-#         except httplib.HTTPException:
-#             pass
-#
-#     print "Failed to get keys for %s" % (server_name,)
-#     return {}
-
-
-def reinsert_events(cursor, server_name, signing_key):
-    print "Running delta: v10"
-
-    cursor.executescript(delta_sql)
-
-    cursor.execute(
-        "SELECT * FROM events ORDER BY rowid ASC"
-    )
-
-    print "Getting events..."
-
-    rows = store.cursor_to_dict(cursor)
-
-    events = store._generate_event_json(cursor, rows)
-
-    print "Got events from DB."
-
-    algorithms = {
-        "sha256": hashlib.sha256,
-    }
-
-    key_id = "%s:%s" % (signing_key.alg, signing_key.version)
-    verify_key = signing_key.verify_key
-    verify_key.alg = signing_key.alg
-    verify_key.version = signing_key.version
-
-    server_keys = {
-        server_name: {
-            key_id: verify_key
-        }
-    }
-
-    i = 0
-    N = len(events)
-
-    for event in events:
-        if i % 100 == 0:
-            print "Processed: %d/%d events" % (i,N,)
-        i += 1
-
-        # for alg_name in event.hashes:
-        #     if check_event_content_hash(event, algorithms[alg_name]):
-        #         pass
-        #     else:
-        #         pass
-        #         print "FAIL content hash %s %s" % (alg_name, event.event_id, )
-
-        have_own_correctly_signed = False
-        for host, sigs in event.signatures.items():
-            pruned = prune_event(event)
-
-            for key_id in sigs:
-                if host not in server_keys:
-                    server_keys[host] = {}  # get_key(host)
-                if key_id in server_keys[host]:
-                    try:
-                        verify_signed_json(
-                            pruned.get_pdu_json(),
-                            host,
-                            server_keys[host][key_id]
-                        )
-
-                        if host == server_name:
-                            have_own_correctly_signed = True
-                    except SignatureVerifyException:
-                        print "FAIL signature check %s %s" % (
-                            key_id, event.event_id
-                        )
-
-        # TODO: Re sign with our own server key
-        if not have_own_correctly_signed:
-            sigs = compute_event_signature(event, server_name, signing_key)
-            event.signatures.update(sigs)
-
-            pruned = prune_event(event)
-
-            for key_id in event.signatures[server_name]:
-                verify_signed_json(
-                    pruned.get_pdu_json(),
-                    server_name,
-                    server_keys[server_name][key_id]
-                )
-
-        event_json = encode_canonical_json(
-            event.get_dict()
-        ).decode("UTF-8")
-
-        metadata_json = encode_canonical_json(
-            event.internal_metadata.get_dict()
-        ).decode("UTF-8")
-
-        store._simple_insert_txn(
-            cursor,
-            table="event_json",
-            values={
-                "event_id": event.event_id,
-                "room_id": event.room_id,
-                "internal_metadata": metadata_json,
-                "json": event_json,
-            },
-            or_replace=True,
-        )
-
-
-def main(database, server_name, signing_key):
-    conn = sqlite3.connect(database)
-    cursor = conn.cursor()
-
-    # Do other deltas:
-    cursor.execute("PRAGMA user_version")
-    row = cursor.fetchone()
-
-    if row and row[0]:
-        user_version = row[0]
-        # Run every version since after the current version.
-        for v in range(user_version + 1, 10):
-            print "Running delta: %d" % (v,)
-            sql_script = read_schema("delta/v%d" % (v,))
-            cursor.executescript(sql_script)
-
-    reinsert_events(cursor, server_name, signing_key)
-
-    conn.commit()
-
-    print "Success!"
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser()
-
-    parser.add_argument("database")
-    parser.add_argument("server_name")
-    parser.add_argument(
-        "signing_key", type=argparse.FileType('r'),
-    )
-    args = parser.parse_args()
-
-    signing_key = syutil.crypto.signing_key.read_signing_keys(
-        args.signing_key
-    )
-
-    main(args.database, args.server_name, signing_key[0])

setup.cfg

@@ -3,9 +3,6 @@ source-dir = docs/sphinx
 build-dir  = docs/build
 all_files  = 1
 
-[aliases]
-test = trial
-
 [trial]
 test_suite = tests
 
@@ -16,3 +13,8 @@ ignore =
     docs/*
     pylint.cfg
     tox.ini
+
+[flake8]
+max-line-length = 90
+#  W503 requires that binary operators be at the end, not start, of lines. Erik doesn't like it.
+ignore = W503

setup.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,8 +14,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import glob
 import os
-from setuptools import setup, find_packages
+from setuptools import setup, find_packages, Command
+import sys
 
 
 here = os.path.abspath(os.path.dirname(__file__))
@@ -36,6 +38,39 @@ def exec_file(path_segments):
     exec(code, result)
     return result
 
+
+class Tox(Command):
+    user_options = [('tox-args=', 'a', "Arguments to pass to tox")]
+
+    def initialize_options(self):
+        self.tox_args = None
+
+    def finalize_options(self):
+        self.test_args = []
+        self.test_suite = True
+
+    def run(self):
+        #import here, cause outside the eggs aren't loaded
+        try:
+            import tox
+        except ImportError:
+            try:
+                self.distribution.fetch_build_eggs("tox")
+                import tox
+            except:
+                raise RuntimeError(
+                    "The tests need 'tox' to run. Please install 'tox'."
+                )
+        import shlex
+        args = self.tox_args
+        if args:
+            args = shlex.split(self.tox_args)
+        else:
+            args = []
+        errno = tox.cmdline(args=args)
+        sys.exit(errno)
+
+
 version = exec_file(("synapse", "__init__.py"))["__version__"]
 dependencies = exec_file(("synapse", "python_dependencies.py"))
 long_description = read_file(("README.rst",))
@@ -46,14 +81,10 @@ setup(
     packages=find_packages(exclude=["tests", "tests.*"]),
     description="Reference Synapse Home Server",
     install_requires=dependencies['requirements'](include_conditional=True).keys(),
-    setup_requires=[
-        "Twisted==14.0.2", # Here to override setuptools_trial's dependency on Twisted>=2.4.0
-        "setuptools_trial",
-        "mock"
-    ],
-    dependency_links=dependencies["DEPENDENCY_LINKS"],
+    dependency_links=dependencies["DEPENDENCY_LINKS"].values(),
     include_package_data=True,
     zip_safe=False,
     long_description=long_description,
-    scripts=["synctl", "register_new_matrix_user"],
+    scripts=["synctl"] + glob.glob("scripts/*"),
+    cmdclass={'test': Tox},
 )

synapse/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """
 
-__version__ = "0.8.1-r4"
+__version__ = "0.18.3"

synapse/api/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

synapse/api/constants.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -27,22 +27,11 @@ class Membership(object):
     LIST = (INVITE, JOIN, KNOCK, LEAVE, BAN)
 
 
-class Feedback(object):
-
-    """Represents the types of feedback a user can send in response to a
-    message."""
-
-    DELIVERED = u"delivered"
-    READ = u"read"
-    LIST = (DELIVERED, READ)
-
-
 class PresenceState(object):
     """Represents the presence state of a user."""
     OFFLINE = u"offline"
     UNAVAILABLE = u"unavailable"
     ONLINE = u"online"
-    FREE_FOR_CHAT = u"free_for_chat"
 
 
 class JoinRules(object):
@@ -73,7 +62,12 @@ class EventTypes(object):
     PowerLevels = "m.room.power_levels"
     Aliases = "m.room.aliases"
     Redaction = "m.room.redaction"
-    Feedback = "m.room.message.feedback"
+    ThirdPartyInvite = "m.room.third_party_invite"
+
+    RoomHistoryVisibility = "m.room.history_visibility"
+    CanonicalAlias = "m.room.canonical_alias"
+    RoomAvatar = "m.room.avatar"
+    GuestAccess = "m.room.guest_access"
 
     # These are used for validation
     Message = "m.room.message"
@@ -85,3 +79,14 @@ class RejectedReason(object):
     AUTH_ERROR = "auth_error"
     REPLACED = "replaced"
     NOT_ANCESTOR = "not_ancestor"
+
+
+class RoomCreationPreset(object):
+    PRIVATE_CHAT = "private_chat"
+    PUBLIC_CHAT = "public_chat"
+    TRUSTED_PRIVATE_CHAT = "trusted_private_chat"
+
+
+class ThirdPartyEntityKind(object):
+    USER = "user"
+    LOCATION = "location"

synapse/api/errors.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -29,10 +29,12 @@ class Codes(object):
     USER_IN_USE = "M_USER_IN_USE"
     ROOM_IN_USE = "M_ROOM_IN_USE"
     BAD_PAGINATION = "M_BAD_PAGINATION"
+    BAD_STATE = "M_BAD_STATE"
     UNKNOWN = "M_UNKNOWN"
     NOT_FOUND = "M_NOT_FOUND"
     MISSING_TOKEN = "M_MISSING_TOKEN"
     UNKNOWN_TOKEN = "M_UNKNOWN_TOKEN"
+    GUEST_ACCESS_FORBIDDEN = "M_GUEST_ACCESS_FORBIDDEN"
     LIMIT_EXCEEDED = "M_LIMIT_EXCEEDED"
     CAPTCHA_NEEDED = "M_CAPTCHA_NEEDED"
     CAPTCHA_INVALID = "M_CAPTCHA_INVALID"
@@ -40,13 +42,16 @@ class Codes(object):
     TOO_LARGE = "M_TOO_LARGE"
     EXCLUSIVE = "M_EXCLUSIVE"
     THREEPID_AUTH_FAILED = "M_THREEPID_AUTH_FAILED"
+    THREEPID_IN_USE = "M_THREEPID_IN_USE"
+    THREEPID_NOT_FOUND = "M_THREEPID_NOT_FOUND"
+    INVALID_USERNAME = "M_INVALID_USERNAME"
+    SERVER_NOT_TRUSTED = "M_SERVER_NOT_TRUSTED"
 
 
 class CodeMessageException(RuntimeError):
     """An exception with integer code and message string attributes."""
 
     def __init__(self, code, msg):
-        logger.info("%s: %s, %s", type(self).__name__, code, msg)
         super(CodeMessageException, self).__init__("%d: %s" % (code, msg))
         self.code = code
         self.msg = msg
@@ -76,11 +81,6 @@ class SynapseError(CodeMessageException):
         )
 
 
-class RoomError(SynapseError):
-    """An error raised when a room event fails."""
-    pass
-
-
 class RegistrationError(SynapseError):
     """An error raised when a registration event fails."""
     pass
@@ -124,6 +124,15 @@ class AuthError(SynapseError):
         super(AuthError, self).__init__(*args, **kwargs)
 
 
+class EventSizeError(SynapseError):
+    """An error raised when an event is too big."""
+
+    def __init__(self, *args, **kwargs):
+        if "errcode" not in kwargs:
+            kwargs["errcode"] = Codes.TOO_LARGE
+        super(EventSizeError, self).__init__(413, *args, **kwargs)
+
+
 class EventStreamError(SynapseError):
     """An error raised when there a problem with the event stream."""
     def __init__(self, *args, **kwargs):

synapse/api/filtering.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,6 +15,10 @@
 from synapse.api.errors import SynapseError
 from synapse.types import UserID, RoomID
 
+from twisted.internet import defer
+
+import ujson as json
+
 
 class Filtering(object):
 
@@ -22,20 +26,20 @@ class Filtering(object):
         super(Filtering, self).__init__()
         self.store = hs.get_datastore()
 
+    @defer.inlineCallbacks
     def get_user_filter(self, user_localpart, filter_id):
-        result = self.store.get_user_filter(user_localpart, filter_id)
-        result.addCallback(Filter)
-        return result
+        result = yield self.store.get_user_filter(user_localpart, filter_id)
+        defer.returnValue(FilterCollection(result))
 
     def add_user_filter(self, user_localpart, user_filter):
-        self._check_valid_filter(user_filter)
+        self.check_valid_filter(user_filter)
         return self.store.add_user_filter(user_localpart, user_filter)
 
     # TODO(paul): surely we should probably add a delete_user_filter or
     #   replace_user_filter at some point? There's no REST API specified for
     #   them however
 
-    def _check_valid_filter(self, user_filter_json):
+    def check_valid_filter(self, user_filter_json):
         """Check if the provided filter is valid.
 
         This inspects all definitions contained within the filter.
@@ -50,11 +54,11 @@ class Filtering(object):
         # many definitions.
 
         top_level_definitions = [
-            "public_user_data", "private_user_data", "server_data"
+            "presence", "account_data"
         ]
 
         room_level_definitions = [
-            "state", "events", "ephemeral"
+            "state", "timeline", "ephemeral", "account_data"
         ]
 
         for key in top_level_definitions:
@@ -62,10 +66,29 @@ class Filtering(object):
                 self._check_definition(user_filter_json[key])
 
         if "room" in user_filter_json:
+            self._check_definition_room_lists(user_filter_json["room"])
             for key in room_level_definitions:
                 if key in user_filter_json["room"]:
                     self._check_definition(user_filter_json["room"][key])
 
+    def _check_definition_room_lists(self, definition):
+        """Check that "rooms" and "not_rooms" are lists of room ids if they
+        are present
+
+        Args:
+            definition(dict): The filter definition
+        Raises:
+            SynapseError: If there was a problem with this definition.
+        """
+        # check rooms are valid room IDs
+        room_id_keys = ["rooms", "not_rooms"]
+        for key in room_id_keys:
+            if key in definition:
+                if type(definition[key]) != list:
+                    raise SynapseError(400, "Expected %s to be a list." % key)
+                for room_id in definition[key]:
+                    RoomID.from_string(room_id)
+
     def _check_definition(self, definition):
         """Check if the provided definition is valid.
 
@@ -85,14 +108,7 @@ class Filtering(object):
                 400, "Expected JSON object, not %s" % (definition,)
             )
 
-        # check rooms are valid room IDs
-        room_id_keys = ["rooms", "not_rooms"]
-        for key in room_id_keys:
-            if key in definition:
-                if type(definition[key]) != list:
-                    raise SynapseError(400, "Expected %s to be a list." % key)
-                for room_id in definition[key]:
-                    RoomID.from_string(room_id)
+        self._check_definition_room_lists(definition)
 
         # check senders are valid user IDs
         user_id_keys = ["senders", "not_senders"]
@@ -114,116 +130,162 @@ class Filtering(object):
                     if not isinstance(event_type, basestring):
                         raise SynapseError(400, "Event type should be a string")
 
-        if "format" in definition:
-            event_format = definition["format"]
-            if event_format not in ["federation", "events"]:
-                raise SynapseError(400, "Invalid format: %s" % (event_format,))
 
-        if "select" in definition:
-            event_select_list = definition["select"]
-            for select_key in event_select_list:
-                if select_key not in ["event_id", "origin_server_ts",
-                                      "thread_id", "content", "content.body"]:
-                    raise SynapseError(400, "Bad select: %s" % (select_key,))
+class FilterCollection(object):
+    def __init__(self, filter_json):
+        self._filter_json = filter_json
 
-        if ("bundle_updates" in definition and
-                type(definition["bundle_updates"]) != bool):
-            raise SynapseError(400, "Bad bundle_updates: expected bool.")
+        room_filter_json = self._filter_json.get("room", {})
+
+        self._room_filter = Filter({
+            k: v for k, v in room_filter_json.items()
+            if k in ("rooms", "not_rooms")
+        })
+
+        self._room_timeline_filter = Filter(room_filter_json.get("timeline", {}))
+        self._room_state_filter = Filter(room_filter_json.get("state", {}))
+        self._room_ephemeral_filter = Filter(room_filter_json.get("ephemeral", {}))
+        self._room_account_data = Filter(room_filter_json.get("account_data", {}))
+        self._presence_filter = Filter(filter_json.get("presence", {}))
+        self._account_data = Filter(filter_json.get("account_data", {}))
+
+        self.include_leave = filter_json.get("room", {}).get(
+            "include_leave", False
+        )
+
+    def __repr__(self):
+        return "<FilterCollection %s>" % (json.dumps(self._filter_json),)
+
+    def get_filter_json(self):
+        return self._filter_json
+
+    def timeline_limit(self):
+        return self._room_timeline_filter.limit()
+
+    def presence_limit(self):
+        return self._presence_filter.limit()
+
+    def ephemeral_limit(self):
+        return self._room_ephemeral_filter.limit()
+
+    def filter_presence(self, events):
+        return self._presence_filter.filter(events)
+
+    def filter_account_data(self, events):
+        return self._account_data.filter(events)
+
+    def filter_room_state(self, events):
+        return self._room_state_filter.filter(self._room_filter.filter(events))
+
+    def filter_room_timeline(self, events):
+        return self._room_timeline_filter.filter(self._room_filter.filter(events))
+
+    def filter_room_ephemeral(self, events):
+        return self._room_ephemeral_filter.filter(self._room_filter.filter(events))
+
+    def filter_room_account_data(self, events):
+        return self._room_account_data.filter(self._room_filter.filter(events))
 
 
 class Filter(object):
     def __init__(self, filter_json):
         self.filter_json = filter_json
 
-    def filter_public_user_data(self, events):
-        return self._filter_on_key(events, ["public_user_data"])
+        self.types = self.filter_json.get("types", None)
+        self.not_types = self.filter_json.get("not_types", [])
 
-    def filter_private_user_data(self, events):
-        return self._filter_on_key(events, ["private_user_data"])
+        self.rooms = self.filter_json.get("rooms", None)
+        self.not_rooms = self.filter_json.get("not_rooms", [])
 
-    def filter_room_state(self, events):
-        return self._filter_on_key(events, ["room", "state"])
+        self.senders = self.filter_json.get("senders", None)
+        self.not_senders = self.filter_json.get("not_senders", [])
 
-    def filter_room_events(self, events):
-        return self._filter_on_key(events, ["room", "events"])
+        self.contains_url = self.filter_json.get("contains_url", None)
 
-    def filter_room_ephemeral(self, events):
-        return self._filter_on_key(events, ["room", "ephemeral"])
+    def check(self, event):
+        """Checks whether the filter matches the given event.
 
-    def _filter_on_key(self, events, keys):
-        filter_json = self.filter_json
-        if not filter_json:
-            return events
-
-        try:
-            # extract the right definition from the filter
-            definition = filter_json
-            for key in keys:
-                definition = definition[key]
-            return self._filter_with_definition(events, definition)
-        except KeyError:
-            # return all events if definition isn't specified.
-            return events
-
-    def _filter_with_definition(self, events, definition):
-        return [e for e in events if self._passes_definition(definition, e)]
-
-    def _passes_definition(self, definition, event):
-        """Check if the event passes through the given definition.
-
-        Args:
-            definition(dict): The definition to check against.
-            event(Event): The event to check.
         Returns:
-            True if the event passes through the filter.
+            bool: True if the event matches
         """
-        # Algorithm notes:
-        # For each key in the definition, check the event meets the criteria:
-        #   * For types: Literal match or prefix match (if ends with wildcard)
-        #   * For senders/rooms: Literal match only
-        #   * "not_" checks take presedence (e.g. if "m.*" is in both 'types'
-        #     and 'not_types' then it is treated as only being in 'not_types')
+        sender = event.get("sender", None)
+        if not sender:
+            # Presence events have their 'sender' in content.user_id
+            content = event.get("content")
+            # account_data has been allowed to have non-dict content, so check type first
+            if isinstance(content, dict):
+                sender = content.get("user_id")
 
-        # room checks
-        if hasattr(event, "room_id"):
-            room_id = event.room_id
-            allow_rooms = definition.get("rooms", None)
-            reject_rooms = definition.get("not_rooms", None)
-            if reject_rooms and room_id in reject_rooms:
-                return False
-            if allow_rooms and room_id not in allow_rooms:
+        return self.check_fields(
+            event.get("room_id", None),
+            sender,
+            event.get("type", None),
+            "url" in event.get("content", {})
+        )
+
+    def check_fields(self, room_id, sender, event_type, contains_url):
+        """Checks whether the filter matches the given event fields.
+
+        Returns:
+            bool: True if the event fields match
+        """
+        literal_keys = {
+            "rooms": lambda v: room_id == v,
+            "senders": lambda v: sender == v,
+            "types": lambda v: _matches_wildcard(event_type, v)
+        }
+
+        for name, match_func in literal_keys.items():
+            not_name = "not_%s" % (name,)
+            disallowed_values = getattr(self, not_name)
+            if any(map(match_func, disallowed_values)):
                 return False
 
-        # sender checks
-        if hasattr(event, "sender"):
-            # Should we be including event.state_key for some event types?
-            sender = event.sender
-            allow_senders = definition.get("senders", None)
-            reject_senders = definition.get("not_senders", None)
-            if reject_senders and sender in reject_senders:
-                return False
-            if allow_senders and sender not in allow_senders:
-                return False
-
-        # type checks
-        if "not_types" in definition:
-            for def_type in definition["not_types"]:
-                if self._event_matches_type(event, def_type):
+            allowed_values = getattr(self, name)
+            if allowed_values is not None:
+                if not any(map(match_func, allowed_values)):
                     return False
-        if "types" in definition:
-            included = False
-            for def_type in definition["types"]:
-                if self._event_matches_type(event, def_type):
-                    included = True
-                    break
-            if not included:
+
+        contains_url_filter = self.filter_json.get("contains_url")
+        if contains_url_filter is not None:
+            if contains_url_filter != contains_url:
                 return False
 
         return True
 
-    def _event_matches_type(self, event, def_type):
-        if def_type.endswith("*"):
-            type_prefix = def_type[:-1]
-            return event.type.startswith(type_prefix)
-        else:
-            return event.type == def_type
+    def filter_rooms(self, room_ids):
+        """Apply the 'rooms' filter to a given list of rooms.
+
+        Args:
+            room_ids (list): A list of room_ids.
+
+        Returns:
+            list: A list of room_ids that match the filter
+        """
+        room_ids = set(room_ids)
+
+        disallowed_rooms = set(self.filter_json.get("not_rooms", []))
+        room_ids -= disallowed_rooms
+
+        allowed_rooms = self.filter_json.get("rooms", None)
+        if allowed_rooms is not None:
+            room_ids &= set(allowed_rooms)
+
+        return room_ids
+
+    def filter(self, events):
+        return filter(self.check, events)
+
+    def limit(self):
+        return self.filter_json.get("limit", 10)
+
+
+def _matches_wildcard(actual_value, filter_value):
+    if filter_value.endswith("*"):
+        type_prefix = filter_value[:-1]
+        return actual_value.startswith(type_prefix)
+    else:
+        return actual_value == filter_value
+
+
+DEFAULT_FILTER_COLLECTION = FilterCollection({})

synapse/api/ratelimiting.py

@@ -1,4 +1,4 @@
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -23,7 +23,7 @@ class Ratelimiter(object):
     def __init__(self):
         self.message_counts = collections.OrderedDict()
 
-    def send_message(self, user_id, time_now_s, msg_rate_hz, burst_count):
+    def send_message(self, user_id, time_now_s, msg_rate_hz, burst_count, update=True):
         """Can the user send a message?
         Args:
             user_id: The user sending a message.
@@ -32,12 +32,15 @@ class Ratelimiter(object):
                 second.
             burst_count: How many messages the user can send before being
                 limited.
+            update (bool): Whether to update the message rates or not. This is
+                useful to check if a message would be allowed to be sent before
+                its ready to be actually sent.
         Returns:
             A pair of a bool indicating if they can send a message now and a
                 time in seconds of when they can next send a message.
         """
         self.prune_message_counts(time_now_s)
-        message_count, time_start, _ignored = self.message_counts.pop(
+        message_count, time_start, _ignored = self.message_counts.get(
             user_id, (0., time_now_s, None),
         )
         time_delta = time_now_s - time_start
@@ -52,9 +55,10 @@ class Ratelimiter(object):
             allowed = True
             message_count += 1
 
-        self.message_counts[user_id] = (
-            message_count, time_start, msg_rate_hz
-        )
+        if update:
+            self.message_counts[user_id] = (
+                message_count, time_start, msg_rate_hz
+            )
 
         if msg_rate_hz > 0:
             time_allowed = (

synapse/api/urls.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -23,5 +23,5 @@ WEB_CLIENT_PREFIX = "/_matrix/client"
 CONTENT_REPO_PREFIX = "/_matrix/content"
 SERVER_KEY_PREFIX = "/_matrix/key/v1"
 SERVER_KEY_V2_PREFIX = "/_matrix/key/v2"
-MEDIA_PREFIX = "/_matrix/media/v1"
-APP_SERVICE_PREFIX = "/_matrix/appservice/v1"
+MEDIA_PREFIX = "/_matrix/media/r0"
+LEGACY_MEDIA_PREFIX = "/_matrix/media/v1"

synapse/app/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,3 +12,20 @@
 # 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 sys
+sys.dont_write_bytecode = True
+
+from synapse import python_dependencies   # noqa: E402
+
+try:
+    python_dependencies.check_requirements()
+except python_dependencies.MissingRequirementError as e:
+    message = "\n".join([
+        "Missing Requirement: %s" % (e.message,),
+        "To install run:",
+        "    pip install --upgrade --force \"%s\"" % (e.dependency,),
+        "",
+    ])
+    sys.stderr.writelines(message)
+    sys.exit(1)

synapse/app/appservice.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.server import HomeServer
+from synapse.config._base import ConfigError
+from synapse.config.logger import setup_logging
+from synapse.config.homeserver import HomeServerConfig
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.replication.slave.storage.directory import DirectoryStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.storage.engines import create_engine
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import gc
+
+logger = logging.getLogger("synapse.app.appservice")
+
+
+class AppserviceSlaveStore(
+    DirectoryStore, SlavedEventStore, SlavedApplicationServiceStore,
+    SlavedRegistrationStore,
+):
+    pass
+
+
+class AppserviceServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = AppserviceSlaveStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse appservice now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+        appservice_handler = self.get_application_service_handler()
+
+        @defer.inlineCallbacks
+        def replicate(results):
+            stream = results.get("events")
+            if stream:
+                max_stream_id = stream["position"]
+                yield appservice_handler.notify_interested_services(max_stream_id)
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+                replicate(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(30)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse appservice", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.appservice"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    database_engine = create_engine(config.database_config)
+
+    if config.notify_appservices:
+        sys.stderr.write(
+            "\nThe appservices must be disabled in the main synapse process"
+            "\nbefore they can be run in a separate worker."
+            "\nPlease add ``notify_appservices: false`` to the main config"
+            "\n"
+        )
+        sys.exit(1)
+
+    # Force the pushers to start since they will be disabled in the main config
+    config.notify_appservices = True
+
+    ps = AppserviceServer(
+        config.server_name,
+        db_config=config.database_config,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ps.setup()
+    ps.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ps.replicate()
+        ps.get_datastore().start_profiling()
+        ps.get_state_handler().start_caching()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-appservice",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])

synapse/app/client_reader.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.config._base import ConfigError
+from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
+from synapse.http.site import SynapseSite
+from synapse.http.server import JsonResource
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.keys import SlavedKeyStore
+from synapse.replication.slave.storage.room import RoomStore
+from synapse.replication.slave.storage.directory import DirectoryStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.rest.client.v1.room import PublicRoomListRestServlet
+from synapse.server import HomeServer
+from synapse.storage.client_ips import ClientIpStore
+from synapse.storage.engines import create_engine
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+from synapse.crypto import context_factory
+
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import gc
+
+logger = logging.getLogger("synapse.app.client_reader")
+
+
+class ClientReaderSlavedStore(
+    SlavedEventStore,
+    SlavedKeyStore,
+    RoomStore,
+    DirectoryStore,
+    SlavedApplicationServiceStore,
+    SlavedRegistrationStore,
+    BaseSlavedStore,
+    ClientIpStore,  # After BaseSlavedStore because the constructor is different
+):
+    pass
+
+
+class ClientReaderServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = ClientReaderSlavedStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+                elif name == "client":
+                    resource = JsonResource(self, canonical_json=False)
+                    PublicRoomListRestServlet(self).register(resource)
+                    resources.update({
+                        "/_matrix/client/r0": resource,
+                        "/_matrix/client/unstable": resource,
+                        "/_matrix/client/v2_alpha": resource,
+                        "/_matrix/client/api/v1": resource,
+                    })
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse client reader now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(5)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse client reader", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.client_reader"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    database_engine = create_engine(config.database_config)
+
+    tls_server_context_factory = context_factory.ServerContextFactory(config)
+
+    ss = ClientReaderServer(
+        config.server_name,
+        db_config=config.database_config,
+        tls_server_context_factory=tls_server_context_factory,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ss.setup()
+    ss.get_handlers()
+    ss.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ss.get_state_handler().start_caching()
+        ss.get_datastore().start_profiling()
+        ss.replicate()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-client-reader",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])

synapse/app/federation_reader.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.config._base import ConfigError
+from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.keys import SlavedKeyStore
+from synapse.replication.slave.storage.room import RoomStore
+from synapse.replication.slave.storage.transactions import TransactionStore
+from synapse.replication.slave.storage.directory import DirectoryStore
+from synapse.server import HomeServer
+from synapse.storage.engines import create_engine
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+from synapse.api.urls import FEDERATION_PREFIX
+from synapse.federation.transport.server import TransportLayerServer
+from synapse.crypto import context_factory
+
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import gc
+
+logger = logging.getLogger("synapse.app.federation_reader")
+
+
+class FederationReaderSlavedStore(
+    SlavedEventStore,
+    SlavedKeyStore,
+    RoomStore,
+    DirectoryStore,
+    TransactionStore,
+    BaseSlavedStore,
+):
+    pass
+
+
+class FederationReaderServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = FederationReaderSlavedStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+                elif name == "federation":
+                    resources.update({
+                        FEDERATION_PREFIX: TransportLayerServer(self),
+                    })
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse federation reader now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(5)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse federation reader", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.federation_reader"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    database_engine = create_engine(config.database_config)
+
+    tls_server_context_factory = context_factory.ServerContextFactory(config)
+
+    ss = FederationReaderServer(
+        config.server_name,
+        db_config=config.database_config,
+        tls_server_context_factory=tls_server_context_factory,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ss.setup()
+    ss.get_handlers()
+    ss.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ss.get_state_handler().start_caching()
+        ss.get_datastore().start_profiling()
+        ss.replicate()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-federation-reader",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])

synapse/app/homeserver.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,255 +14,208 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import synapse
+
+import gc
+import logging
+import os
 import sys
-sys.dont_write_bytecode = True
-from synapse.python_dependencies import check_requirements
+from synapse.config._base import ConfigError
 
-if __name__ == '__main__':
-    check_requirements()
-
-from synapse.storage.engines import create_engine, IncorrectDatabaseSetup
-from synapse.storage import (
-    are_all_users_on_domain, UpgradeDatabaseException,
+from synapse.python_dependencies import (
+    check_requirements, DEPENDENCY_LINKS
 )
 
+from synapse.rest import ClientRestResource
+from synapse.storage.engines import create_engine, IncorrectDatabaseSetup
+from synapse.storage import are_all_users_on_domain
+from synapse.storage.prepare_database import UpgradeDatabaseException, prepare_database
+
 from synapse.server import HomeServer
 
-
-from twisted.internet import reactor
+from twisted.internet import reactor, task, defer
 from twisted.application import service
-from twisted.enterprise import adbapi
-from twisted.web.resource import Resource
+from twisted.web.resource import Resource, EncodingResourceWrapper
 from twisted.web.static import File
-from twisted.web.server import Site
-from twisted.web.http import proxiedLogFormatter, combinedLogFormatter
-from synapse.http.server import JsonResource, RootRedirect
+from twisted.web.server import GzipEncoderFactory
+from synapse.http.server import RootRedirect
 from synapse.rest.media.v0.content_repository import ContentRepoResource
 from synapse.rest.media.v1.media_repository import MediaRepositoryResource
 from synapse.rest.key.v1.server_key_resource import LocalKey
 from synapse.rest.key.v2 import KeyApiV2Resource
-from synapse.http.matrixfederationclient import MatrixFederationHttpClient
 from synapse.api.urls import (
-    CLIENT_PREFIX, FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX,
-    SERVER_KEY_PREFIX, MEDIA_PREFIX, CLIENT_V2_ALPHA_PREFIX, STATIC_PREFIX,
+    FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX,
+    SERVER_KEY_PREFIX, LEGACY_MEDIA_PREFIX, MEDIA_PREFIX, STATIC_PREFIX,
     SERVER_KEY_V2_PREFIX,
 )
 from synapse.config.homeserver import HomeServerConfig
 from synapse.crypto import context_factory
 from synapse.util.logcontext import LoggingContext
-from synapse.rest.client.v1 import ClientV1RestResource
-from synapse.rest.client.v2_alpha import ClientV2AlphaRestResource
+from synapse.metrics import register_memory_metrics, get_metrics_for
 from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.replication.resource import ReplicationResource, REPLICATION_PREFIX
+from synapse.federation.transport.server import TransportLayerServer
+
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.manhole import manhole
+
+from synapse.http.site import SynapseSite
+
+from synapse import events
 
 from daemonize import Daemonize
-import twisted.manhole.telnet
-
-import synapse
-
-import logging
-import os
-import re
-import resource
-import subprocess
-
 
 logger = logging.getLogger("synapse.app.homeserver")
 
 
-class SynapseHomeServer(HomeServer):
+def gz_wrap(r):
+    return EncodingResourceWrapper(r, [GzipEncoderFactory()])
 
-    def build_http_client(self):
-        return MatrixFederationHttpClient(self)
 
-    def build_resource_for_client(self):
-        return ClientV1RestResource(self)
-
-    def build_resource_for_client_v2_alpha(self):
-        return ClientV2AlphaRestResource(self)
-
-    def build_resource_for_federation(self):
-        return JsonResource(self)
-
-    def build_resource_for_web_client(self):
-        import syweb
+def build_resource_for_web_client(hs):
+    webclient_path = hs.get_config().web_client_location
+    if not webclient_path:
+        try:
+            import syweb
+        except ImportError:
+            quit_with_error(
+                "Could not find a webclient.\n\n"
+                "Please either install the matrix-angular-sdk or configure\n"
+                "the location of the source to serve via the configuration\n"
+                "option `web_client_location`\n\n"
+                "To install the `matrix-angular-sdk` via pip, run:\n\n"
+                "    pip install '%(dep)s'\n"
+                "\n"
+                "You can also disable hosting of the webclient via the\n"
+                "configuration option `web_client`\n"
+                % {"dep": DEPENDENCY_LINKS["matrix-angular-sdk"]}
+            )
         syweb_path = os.path.dirname(syweb.__file__)
         webclient_path = os.path.join(syweb_path, "webclient")
-        return File(webclient_path)  # TODO configurable?
+    # GZip is disabled here due to
+    # https://twistedmatrix.com/trac/ticket/7678
+    # (It can stay enabled for the API resources: they call
+    # write() with the whole body and then finish() straight
+    # after and so do not trigger the bug.
+    # GzipFile was removed in commit 184ba09
+    # return GzipFile(webclient_path)  # TODO configurable?
+    return File(webclient_path)  # TODO configurable?
 
-    def build_resource_for_static_content(self):
-        return File("static")
 
-    def build_resource_for_content_repo(self):
-        return ContentRepoResource(
-            self, self.upload_dir, self.auth, self.content_addr
-        )
+class SynapseHomeServer(HomeServer):
+    def _listener_http(self, config, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        tls = listener_config.get("tls", False)
+        site_tag = listener_config.get("tag", port)
 
-    def build_resource_for_media_repository(self):
-        return MediaRepositoryResource(self)
+        if tls and config.no_tls:
+            return
 
-    def build_resource_for_server_key(self):
-        return LocalKey(self)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "client":
+                    client_resource = ClientRestResource(self)
+                    if res["compress"]:
+                        client_resource = gz_wrap(client_resource)
 
-    def build_resource_for_server_key_v2(self):
-        return KeyApiV2Resource(self)
+                    resources.update({
+                        "/_matrix/client/api/v1": client_resource,
+                        "/_matrix/client/r0": client_resource,
+                        "/_matrix/client/unstable": client_resource,
+                        "/_matrix/client/v2_alpha": client_resource,
+                        "/_matrix/client/versions": client_resource,
+                    })
 
-    def build_resource_for_metrics(self):
-        if self.get_config().enable_metrics:
-            return MetricsResource(self)
+                if name == "federation":
+                    resources.update({
+                        FEDERATION_PREFIX: TransportLayerServer(self),
+                    })
+
+                if name in ["static", "client"]:
+                    resources.update({
+                        STATIC_PREFIX: File(
+                            os.path.join(os.path.dirname(synapse.__file__), "static")
+                        ),
+                    })
+
+                if name in ["media", "federation", "client"]:
+                    media_repo = MediaRepositoryResource(self)
+                    resources.update({
+                        MEDIA_PREFIX: media_repo,
+                        LEGACY_MEDIA_PREFIX: media_repo,
+                        CONTENT_REPO_PREFIX: ContentRepoResource(
+                            self, self.config.uploads_path
+                        ),
+                    })
+
+                if name in ["keys", "federation"]:
+                    resources.update({
+                        SERVER_KEY_PREFIX: LocalKey(self),
+                        SERVER_KEY_V2_PREFIX: KeyApiV2Resource(self),
+                    })
+
+                if name == "webclient":
+                    resources[WEB_CLIENT_PREFIX] = build_resource_for_web_client(self)
+
+                if name == "metrics" and self.get_config().enable_metrics:
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+
+                if name == "replication":
+                    resources[REPLICATION_PREFIX] = ReplicationResource(self)
+
+        if WEB_CLIENT_PREFIX in resources:
+            root_resource = RootRedirect(WEB_CLIENT_PREFIX)
         else:
-            return None
+            root_resource = Resource()
 
-    def build_db_pool(self):
-        name = self.db_config["name"]
-
-        return adbapi.ConnectionPool(
-            name,
-            **self.db_config.get("args", {})
-        )
-
-    def create_resource_tree(self, redirect_root_to_web_client):
-        """Create the resource tree for this Home Server.
-
-        This in unduly complicated because Twisted does not support putting
-        child resources more than 1 level deep at a time.
-
-        Args:
-            web_client (bool): True to enable the web client.
-            redirect_root_to_web_client (bool): True to redirect '/' to the
-            location of the web client. This does nothing if web_client is not
-            True.
-        """
-        config = self.get_config()
-        web_client = config.web_client
-
-        # list containing (path_str, Resource) e.g:
-        # [ ("/aaa/bbb/cc", Resource1), ("/aaa/dummy", Resource2) ]
-        desired_tree = [
-            (CLIENT_PREFIX, self.get_resource_for_client()),
-            (CLIENT_V2_ALPHA_PREFIX, self.get_resource_for_client_v2_alpha()),
-            (FEDERATION_PREFIX, self.get_resource_for_federation()),
-            (CONTENT_REPO_PREFIX, self.get_resource_for_content_repo()),
-            (SERVER_KEY_PREFIX, self.get_resource_for_server_key()),
-            (SERVER_KEY_V2_PREFIX, self.get_resource_for_server_key_v2()),
-            (MEDIA_PREFIX, self.get_resource_for_media_repository()),
-            (STATIC_PREFIX, self.get_resource_for_static_content()),
-        ]
-
-        if web_client:
-            logger.info("Adding the web client.")
-            desired_tree.append((WEB_CLIENT_PREFIX,
-                                self.get_resource_for_web_client()))
-
-        if web_client and redirect_root_to_web_client:
-            self.root_resource = RootRedirect(WEB_CLIENT_PREFIX)
+        root_resource = create_resource_tree(resources, root_resource)
+        if tls:
+            reactor.listenSSL(
+                port,
+                SynapseSite(
+                    "synapse.access.https.%s" % (site_tag,),
+                    site_tag,
+                    listener_config,
+                    root_resource,
+                ),
+                self.tls_server_context_factory,
+                interface=bind_address
+            )
         else:
-            self.root_resource = Resource()
-
-        metrics_resource = self.get_resource_for_metrics()
-        if config.metrics_port is None and metrics_resource is not None:
-            desired_tree.append((METRICS_PREFIX, metrics_resource))
-
-        # ideally we'd just use getChild and putChild but getChild doesn't work
-        # unless you give it a Request object IN ADDITION to the name :/ So
-        # instead, we'll store a copy of this mapping so we can actually add
-        # extra resources to existing nodes. See self._resource_id for the key.
-        resource_mappings = {}
-        for full_path, res in desired_tree:
-            logger.info("Attaching %s to path %s", res, full_path)
-            last_resource = self.root_resource
-            for path_seg in full_path.split('/')[1:-1]:
-                if path_seg not in last_resource.listNames():
-                    # resource doesn't exist, so make a "dummy resource"
-                    child_resource = Resource()
-                    last_resource.putChild(path_seg, child_resource)
-                    res_id = self._resource_id(last_resource, path_seg)
-                    resource_mappings[res_id] = child_resource
-                    last_resource = child_resource
-                else:
-                    # we have an existing Resource, use that instead.
-                    res_id = self._resource_id(last_resource, path_seg)
-                    last_resource = resource_mappings[res_id]
-
-            # ===========================
-            # now attach the actual desired resource
-            last_path_seg = full_path.split('/')[-1]
-
-            # if there is already a resource here, thieve its children and
-            # replace it
-            res_id = self._resource_id(last_resource, last_path_seg)
-            if res_id in resource_mappings:
-                # there is a dummy resource at this path already, which needs
-                # to be replaced with the desired resource.
-                existing_dummy_resource = resource_mappings[res_id]
-                for child_name in existing_dummy_resource.listNames():
-                    child_res_id = self._resource_id(existing_dummy_resource,
-                                                     child_name)
-                    child_resource = resource_mappings[child_res_id]
-                    # steal the children
-                    res.putChild(child_name, child_resource)
-
-            # finally, insert the desired resource in the right place
-            last_resource.putChild(last_path_seg, res)
-            res_id = self._resource_id(last_resource, last_path_seg)
-            resource_mappings[res_id] = res
-
-        return self.root_resource
-
-    def _resource_id(self, resource, path_seg):
-        """Construct an arbitrary resource ID so you can retrieve the mapping
-        later.
-
-        If you want to represent resource A putChild resource B with path C,
-        the mapping should looks like _resource_id(A,C) = B.
-
-        Args:
-            resource (Resource): The *parent* Resource
-            path_seg (str): The name of the child Resource to be attached.
-        Returns:
-            str: A unique string which can be a key to the child Resource.
-        """
-        return "%s-%s" % (resource, path_seg)
+            reactor.listenTCP(
+                port,
+                SynapseSite(
+                    "synapse.access.http.%s" % (site_tag,),
+                    site_tag,
+                    listener_config,
+                    root_resource,
+                ),
+                interface=bind_address
+            )
+        logger.info("Synapse now listening on port %d", port)
 
     def start_listening(self):
         config = self.get_config()
 
-        if not config.no_tls and config.bind_port is not None:
-            reactor.listenSSL(
-                config.bind_port,
-                SynapseSite(
-                    "synapse.access.https",
-                    config,
-                    self.root_resource,
-                ),
-                self.tls_context_factory,
-                interface=config.bind_host
-            )
-            logger.info("Synapse now listening on port %d", config.bind_port)
-
-        if config.unsecure_port is not None:
-            reactor.listenTCP(
-                config.unsecure_port,
-                SynapseSite(
-                    "synapse.access.http",
-                    config,
-                    self.root_resource,
-                ),
-                interface=config.bind_host
-            )
-            logger.info("Synapse now listening on port %d", config.unsecure_port)
-
-        metrics_resource = self.get_resource_for_metrics()
-        if metrics_resource and config.metrics_port is not None:
-            reactor.listenTCP(
-                config.metrics_port,
-                SynapseSite(
-                    "synapse.access.metrics",
-                    config,
-                    metrics_resource,
-                ),
-                interface="127.0.0.1",
-            )
-            logger.info("Metrics now running on 127.0.0.1 port %d", config.metrics_port)
+        for listener in config.listeners:
+            if listener["type"] == "http":
+                self._listener_http(config, listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
 
     def run_startup_checks(self, db_conn, database_engine):
         all_users_native = are_all_users_on_domain(
@@ -280,161 +233,86 @@ class SynapseHomeServer(HomeServer):
         except IncorrectDatabaseSetup as e:
             quit_with_error(e.message)
 
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
 
 def quit_with_error(error_string):
     message_lines = error_string.split("\n")
-    line_length = max([len(l) for l in message_lines]) + 2
+    line_length = max([len(l) for l in message_lines if len(l) < 80]) + 2
     sys.stderr.write("*" * line_length + '\n')
     for line in message_lines:
-        if line.strip():
-            sys.stderr.write(" %s\n" % (line.strip(),))
+        sys.stderr.write(" %s\n" % (line.rstrip(),))
     sys.stderr.write("*" * line_length + '\n')
     sys.exit(1)
 
 
-def get_version_string():
-    try:
-        null = open(os.devnull, 'w')
-        cwd = os.path.dirname(os.path.abspath(__file__))
-        try:
-            git_branch = subprocess.check_output(
-                ['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
-                stderr=null,
-                cwd=cwd,
-            ).strip()
-            git_branch = "b=" + git_branch
-        except subprocess.CalledProcessError:
-            git_branch = ""
-
-        try:
-            git_tag = subprocess.check_output(
-                ['git', 'describe', '--exact-match'],
-                stderr=null,
-                cwd=cwd,
-            ).strip()
-            git_tag = "t=" + git_tag
-        except subprocess.CalledProcessError:
-            git_tag = ""
-
-        try:
-            git_commit = subprocess.check_output(
-                ['git', 'rev-parse', '--short', 'HEAD'],
-                stderr=null,
-                cwd=cwd,
-            ).strip()
-        except subprocess.CalledProcessError:
-            git_commit = ""
-
-        try:
-            dirty_string = "-this_is_a_dirty_checkout"
-            is_dirty = subprocess.check_output(
-                ['git', 'describe', '--dirty=' + dirty_string],
-                stderr=null,
-                cwd=cwd,
-            ).strip().endswith(dirty_string)
-
-            git_dirty = "dirty" if is_dirty else ""
-        except subprocess.CalledProcessError:
-            git_dirty = ""
-
-        if git_branch or git_tag or git_commit or git_dirty:
-            git_version = ",".join(
-                s for s in
-                (git_branch, git_tag, git_commit, git_dirty,)
-                if s
-            )
-
-            return (
-                "Synapse/%s (%s)" % (
-                    synapse.__version__, git_version,
-                )
-            ).encode("ascii")
-    except Exception as e:
-        logger.warn("Failed to check for git repository: %s", e)
-
-    return ("Synapse/%s" % (synapse.__version__,)).encode("ascii")
-
-
-def change_resource_limit(soft_file_no):
-    try:
-        soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
-
-        if not soft_file_no:
-            soft_file_no = hard
-
-        resource.setrlimit(resource.RLIMIT_NOFILE, (soft_file_no, hard))
-
-        logger.info("Set file limit to: %d", soft_file_no)
-    except (ValueError, resource.error) as e:
-        logger.warn("Failed to set file limit: %s", e)
-
-
 def setup(config_options):
     """
     Args:
         config_options_options: The options passed to Synapse. Usually
             `sys.argv[1:]`.
-        should_run (bool): Whether to start the reactor.
 
     Returns:
         HomeServer
     """
-    config = HomeServerConfig.load_config(
-        "Synapse Homeserver",
-        config_options,
-        generate_section="Homeserver"
-    )
+    try:
+        config = HomeServerConfig.load_or_generate_config(
+            "Synapse Homeserver",
+            config_options,
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    if not config:
+        # If a config isn't returned, and an exception isn't raised, we're just
+        # generating config files and shouldn't try to continue.
+        sys.exit(0)
 
     config.setup_logging()
 
     # check any extra requirements we have now we have a config
     check_requirements(config)
 
-    version_string = get_version_string()
+    version_string = "Synapse/" + get_version_string(synapse)
 
     logger.info("Server hostname: %s", config.server_name)
     logger.info("Server version: %s", version_string)
 
-    if re.search(":[0-9]+$", config.server_name):
-        domain_with_port = config.server_name
-    else:
-        domain_with_port = "%s:%s" % (config.server_name, config.bind_port)
+    events.USE_FROZEN_DICTS = config.use_frozen_dicts
 
-    tls_context_factory = context_factory.ServerContextFactory(config)
+    tls_server_context_factory = context_factory.ServerContextFactory(config)
 
-    database_engine = create_engine(config.database_config["name"])
+    database_engine = create_engine(config.database_config)
     config.database_config["args"]["cp_openfun"] = database_engine.on_new_connection
 
     hs = SynapseHomeServer(
         config.server_name,
-        domain_with_port=domain_with_port,
-        upload_dir=os.path.abspath("uploads"),
-        db_name=config.database_path,
         db_config=config.database_config,
-        tls_context_factory=tls_context_factory,
+        tls_server_context_factory=tls_server_context_factory,
         config=config,
-        content_addr=config.content_addr,
         version_string=version_string,
         database_engine=database_engine,
     )
 
-    hs.create_resource_tree(
-        redirect_root_to_web_client=True,
-    )
-
-    db_name = hs.get_db_name()
-
-    logger.info("Preparing database: %s...", db_name)
+    logger.info("Preparing database: %s...", config.database_config['name'])
 
     try:
-        db_conn = database_engine.module.connect(
-            **{
-                k: v for k, v in config.database_config.get("args", {}).items()
-                if not k.startswith("cp_")
-            }
-        )
+        db_conn = hs.get_db_conn(run_new_connection=False)
+        prepare_database(db_conn, database_engine, config=config)
+        database_engine.on_new_connection(db_conn)
 
-        database_engine.prepare_database(db_conn)
         hs.run_startup_checks(db_conn, database_engine)
 
         db_conn.commit()
@@ -446,21 +324,21 @@ def setup(config_options):
         )
         sys.exit(1)
 
-    logger.info("Database prepared in %s.", db_name)
-
-    if config.manhole:
-        f = twisted.manhole.telnet.ShellFactory()
-        f.username = "matrix"
-        f.password = "rabbithole"
-        f.namespace['hs'] = hs
-        reactor.listenTCP(config.manhole, f, interface='127.0.0.1')
+    logger.info("Database prepared in %s.", config.database_config['name'])
 
+    hs.setup()
     hs.start_listening()
 
-    hs.get_pusherpool().start()
-    hs.get_state_handler().start_caching()
-    hs.get_datastore().start_profiling()
-    hs.get_replication_layer().start_get_pdu_cache()
+    def start():
+        hs.get_pusherpool().start()
+        hs.get_state_handler().start_caching()
+        hs.get_datastore().start_profiling()
+        hs.get_datastore().start_doing_background_updates()
+        hs.get_replication_layer().start_get_pdu_cache()
+
+        register_memory_metrics(hs)
+
+    reactor.callWhenRunning(start)
 
     return hs
 
@@ -475,40 +353,111 @@ class SynapseService(service.Service):
     def startService(self):
         hs = setup(self.config)
         change_resource_limit(hs.config.soft_file_limit)
+        if hs.config.gc_thresholds:
+            gc.set_threshold(*hs.config.gc_thresholds)
 
     def stopService(self):
         return self._port.stopListening()
 
 
-class SynapseSite(Site):
-    """
-    Subclass of a twisted http Site that does access logging with python's
-    standard logging
-    """
-    def __init__(self, logger_name, config, resource, *args, **kwargs):
-        Site.__init__(self, resource, *args, **kwargs)
-        if config.captcha_ip_origin_is_x_forwarded:
-            self._log_formatter = proxiedLogFormatter
-        else:
-            self._log_formatter = combinedLogFormatter
-        self.access_logger = logging.getLogger(logger_name)
-
-    def log(self, request):
-        line = self._log_formatter(self._logDateTime, request)
-        self.access_logger.info(line)
-
-
 def run(hs):
+    PROFILE_SYNAPSE = False
+    if PROFILE_SYNAPSE:
+        def profile(func):
+            from cProfile import Profile
+            from threading import current_thread
+
+            def profiled(*args, **kargs):
+                profile = Profile()
+                profile.enable()
+                func(*args, **kargs)
+                profile.disable()
+                ident = current_thread().ident
+                profile.dump_stats("/tmp/%s.%s.%i.pstat" % (
+                    hs.hostname, func.__name__, ident
+                ))
+
+            return profiled
+
+        from twisted.python.threadpool import ThreadPool
+        ThreadPool._worker = profile(ThreadPool._worker)
+        reactor.run = profile(reactor.run)
+
+    start_time = hs.get_clock().time()
+
+    stats = {}
+
+    @defer.inlineCallbacks
+    def phone_stats_home():
+        logger.info("Gathering stats for reporting")
+        now = int(hs.get_clock().time())
+        uptime = int(now - start_time)
+        if uptime < 0:
+            uptime = 0
+
+        # If the stats directory is empty then this is the first time we've
+        # reported stats.
+        first_time = not stats
+
+        stats["homeserver"] = hs.config.server_name
+        stats["timestamp"] = now
+        stats["uptime_seconds"] = uptime
+        stats["total_users"] = yield hs.get_datastore().count_all_users()
+
+        room_count = yield hs.get_datastore().get_room_count()
+        stats["total_room_count"] = room_count
+
+        stats["daily_active_users"] = yield hs.get_datastore().count_daily_users()
+        daily_messages = yield hs.get_datastore().count_daily_messages()
+        if daily_messages is not None:
+            stats["daily_messages"] = daily_messages
+        else:
+            stats.pop("daily_messages", None)
+
+        if first_time:
+            # Add callbacks to report the synapse stats as metrics whenever
+            # prometheus requests them, typically every 30s.
+            # As some of the stats are expensive to calculate we only update
+            # them when synapse phones home to matrix.org every 24 hours.
+            metrics = get_metrics_for("synapse.usage")
+            metrics.add_callback("timestamp", lambda: stats["timestamp"])
+            metrics.add_callback("uptime_seconds", lambda: stats["uptime_seconds"])
+            metrics.add_callback("total_users", lambda: stats["total_users"])
+            metrics.add_callback("total_room_count", lambda: stats["total_room_count"])
+            metrics.add_callback(
+                "daily_active_users", lambda: stats["daily_active_users"]
+            )
+            metrics.add_callback(
+                "daily_messages", lambda: stats.get("daily_messages", 0)
+            )
+
+        logger.info("Reporting stats to matrix.org: %s" % (stats,))
+        try:
+            yield hs.get_simple_http_client().put_json(
+                "https://matrix.org/report-usage-stats/push",
+                stats
+            )
+        except Exception as e:
+            logger.warn("Error reporting stats: %s", e)
+
+    if hs.config.report_stats:
+        phone_home_task = task.LoopingCall(phone_stats_home)
+        logger.info("Scheduling stats reporting for 24 hour intervals")
+        phone_home_task.start(60 * 60 * 24, now=False)
 
     def in_thread():
+        # Uncomment to enable tracing of log context changes.
+        # sys.settrace(logcontext_tracer)
         with LoggingContext("run"):
             change_resource_limit(hs.config.soft_file_limit)
-
+            if hs.config.gc_thresholds:
+                gc.set_threshold(*hs.config.gc_thresholds)
             reactor.run()
 
     if hs.config.daemonize:
 
-        print hs.config.pid_file
+        if hs.config.print_pidfile:
+            print (hs.config.pid_file)
 
         daemon = Daemonize(
             app="synapse-homeserver",

synapse/app/media_repository.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.config._base import ConfigError
+from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.rest.media.v0.content_repository import ContentRepoResource
+from synapse.rest.media.v1.media_repository import MediaRepositoryResource
+from synapse.server import HomeServer
+from synapse.storage.client_ips import ClientIpStore
+from synapse.storage.engines import create_engine
+from synapse.storage.media_repository import MediaRepositoryStore
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+from synapse.api.urls import (
+    CONTENT_REPO_PREFIX, LEGACY_MEDIA_PREFIX, MEDIA_PREFIX
+)
+from synapse.crypto import context_factory
+
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import gc
+
+logger = logging.getLogger("synapse.app.media_repository")
+
+
+class MediaRepositorySlavedStore(
+    SlavedApplicationServiceStore,
+    SlavedRegistrationStore,
+    BaseSlavedStore,
+    MediaRepositoryStore,
+    ClientIpStore,
+):
+    pass
+
+
+class MediaRepositoryServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = MediaRepositorySlavedStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+                elif name == "media":
+                    media_repo = MediaRepositoryResource(self)
+                    resources.update({
+                        MEDIA_PREFIX: media_repo,
+                        LEGACY_MEDIA_PREFIX: media_repo,
+                        CONTENT_REPO_PREFIX: ContentRepoResource(
+                            self, self.config.uploads_path
+                        ),
+                    })
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse media repository now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(5)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse media repository", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.media_repository"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    database_engine = create_engine(config.database_config)
+
+    tls_server_context_factory = context_factory.ServerContextFactory(config)
+
+    ss = MediaRepositoryServer(
+        config.server_name,
+        db_config=config.database_config,
+        tls_server_context_factory=tls_server_context_factory,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ss.setup()
+    ss.get_handlers()
+    ss.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ss.get_state_handler().start_caching()
+        ss.get_datastore().start_profiling()
+        ss.replicate()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-media-repository",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])

synapse/app/pusher.py

@@ -0,0 +1,299 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.server import HomeServer
+from synapse.config._base import ConfigError
+from synapse.config.logger import setup_logging
+from synapse.config.homeserver import HomeServerConfig
+from synapse.http.site import SynapseSite
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.storage.roommember import RoomMemberStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.pushers import SlavedPusherStore
+from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
+from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
+from synapse.storage.engines import create_engine
+from synapse.storage import DataStore
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext, preserve_fn
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.versionstring import get_version_string
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import gc
+
+logger = logging.getLogger("synapse.app.pusher")
+
+
+class PusherSlaveStore(
+    SlavedEventStore, SlavedPusherStore, SlavedReceiptsStore,
+    SlavedAccountDataStore
+):
+    update_pusher_last_stream_ordering_and_success = (
+        DataStore.update_pusher_last_stream_ordering_and_success.__func__
+    )
+
+    update_pusher_failing_since = (
+        DataStore.update_pusher_failing_since.__func__
+    )
+
+    update_pusher_last_stream_ordering = (
+        DataStore.update_pusher_last_stream_ordering.__func__
+    )
+
+    get_throttle_params_by_room = (
+        DataStore.get_throttle_params_by_room.__func__
+    )
+
+    set_throttle_params = (
+        DataStore.set_throttle_params.__func__
+    )
+
+    get_time_of_last_push_action_before = (
+        DataStore.get_time_of_last_push_action_before.__func__
+    )
+
+    get_profile_displayname = (
+        DataStore.get_profile_displayname.__func__
+    )
+
+    who_forgot_in_room = (
+        RoomMemberStore.__dict__["who_forgot_in_room"]
+    )
+
+
+class PusherServer(HomeServer):
+
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = PusherSlaveStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def remove_pusher(self, app_id, push_key, user_id):
+        http_client = self.get_simple_http_client()
+        replication_url = self.config.worker_replication_url
+        url = replication_url + "/remove_pushers"
+        return http_client.post_json_get_json(url, {
+            "remove": [{
+                "app_id": app_id,
+                "push_key": push_key,
+                "user_id": user_id,
+            }]
+        })
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse pusher now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+        pusher_pool = self.get_pusherpool()
+
+        def stop_pusher(user_id, app_id, pushkey):
+            key = "%s:%s" % (app_id, pushkey)
+            pushers_for_user = pusher_pool.pushers.get(user_id, {})
+            pusher = pushers_for_user.pop(key, None)
+            if pusher is None:
+                return
+            logger.info("Stopping pusher %r / %r", user_id, key)
+            pusher.on_stop()
+
+        def start_pusher(user_id, app_id, pushkey):
+            key = "%s:%s" % (app_id, pushkey)
+            logger.info("Starting pusher %r / %r", user_id, key)
+            return pusher_pool._refresh_pusher(app_id, pushkey, user_id)
+
+        @defer.inlineCallbacks
+        def poke_pushers(results):
+            pushers_rows = set(
+                map(tuple, results.get("pushers", {}).get("rows", []))
+            )
+            deleted_pushers_rows = set(
+                map(tuple, results.get("deleted_pushers", {}).get("rows", []))
+            )
+            for row in sorted(pushers_rows | deleted_pushers_rows):
+                if row in deleted_pushers_rows:
+                    user_id, app_id, pushkey = row[1:4]
+                    stop_pusher(user_id, app_id, pushkey)
+                elif row in pushers_rows:
+                    user_id = row[1]
+                    app_id = row[5]
+                    pushkey = row[8]
+                    yield start_pusher(user_id, app_id, pushkey)
+
+            stream = results.get("events")
+            if stream and stream["rows"]:
+                min_stream_id = stream["rows"][0][0]
+                max_stream_id = stream["position"]
+                preserve_fn(pusher_pool.on_new_notifications)(
+                    min_stream_id, max_stream_id
+                )
+
+            stream = results.get("receipts")
+            if stream and stream["rows"]:
+                rows = stream["rows"]
+                affected_room_ids = set(row[1] for row in rows)
+                min_stream_id = rows[0][0]
+                max_stream_id = stream["position"]
+                preserve_fn(pusher_pool.on_new_receipts)(
+                    min_stream_id, max_stream_id, affected_room_ids
+                )
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+                poke_pushers(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(30)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse pusher", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.pusher"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    if config.start_pushers:
+        sys.stderr.write(
+            "\nThe pushers must be disabled in the main synapse process"
+            "\nbefore they can be run in a separate worker."
+            "\nPlease add ``start_pushers: false`` to the main config"
+            "\n"
+        )
+        sys.exit(1)
+
+    # Force the pushers to start since they will be disabled in the main config
+    config.start_pushers = True
+
+    database_engine = create_engine(config.database_config)
+
+    ps = PusherServer(
+        config.server_name,
+        db_config=config.database_config,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+    )
+
+    ps.setup()
+    ps.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ps.replicate()
+        ps.get_pusherpool().start()
+        ps.get_datastore().start_profiling()
+        ps.get_state_handler().start_caching()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-pusher",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        ps = start(sys.argv[1:])

synapse/app/synchrotron.py

@@ -0,0 +1,494 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 2016 OpenMarket 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.
+
+import synapse
+
+from synapse.api.constants import EventTypes, PresenceState
+from synapse.config._base import ConfigError
+from synapse.config.homeserver import HomeServerConfig
+from synapse.config.logger import setup_logging
+from synapse.events import FrozenEvent
+from synapse.handlers.presence import PresenceHandler
+from synapse.http.site import SynapseSite
+from synapse.http.server import JsonResource
+from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.rest.client.v2_alpha import sync
+from synapse.rest.client.v1 import events
+from synapse.rest.client.v1.room import RoomInitialSyncRestServlet
+from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
+from synapse.replication.slave.storage._base import BaseSlavedStore
+from synapse.replication.slave.storage.events import SlavedEventStore
+from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
+from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
+from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
+from synapse.replication.slave.storage.registration import SlavedRegistrationStore
+from synapse.replication.slave.storage.filtering import SlavedFilteringStore
+from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
+from synapse.replication.slave.storage.presence import SlavedPresenceStore
+from synapse.replication.slave.storage.deviceinbox import SlavedDeviceInboxStore
+from synapse.replication.slave.storage.room import RoomStore
+from synapse.server import HomeServer
+from synapse.storage.client_ips import ClientIpStore
+from synapse.storage.engines import create_engine
+from synapse.storage.presence import PresenceStore, UserPresenceState
+from synapse.storage.roommember import RoomMemberStore
+from synapse.util.async import sleep
+from synapse.util.httpresourcetree import create_resource_tree
+from synapse.util.logcontext import LoggingContext, preserve_fn
+from synapse.util.manhole import manhole
+from synapse.util.rlimit import change_resource_limit
+from synapse.util.stringutils import random_string
+from synapse.util.versionstring import get_version_string
+
+from twisted.internet import reactor, defer
+from twisted.web.resource import Resource
+
+from daemonize import Daemonize
+
+import sys
+import logging
+import contextlib
+import gc
+import ujson as json
+
+logger = logging.getLogger("synapse.app.synchrotron")
+
+
+class SynchrotronSlavedStore(
+    SlavedPushRuleStore,
+    SlavedEventStore,
+    SlavedReceiptsStore,
+    SlavedAccountDataStore,
+    SlavedApplicationServiceStore,
+    SlavedRegistrationStore,
+    SlavedFilteringStore,
+    SlavedPresenceStore,
+    SlavedDeviceInboxStore,
+    RoomStore,
+    BaseSlavedStore,
+    ClientIpStore,  # After BaseSlavedStore because the constructor is different
+):
+    who_forgot_in_room = (
+        RoomMemberStore.__dict__["who_forgot_in_room"]
+    )
+
+    # XXX: This is a bit broken because we don't persist the accepted list in a
+    # way that can be replicated. This means that we don't have a way to
+    # invalidate the cache correctly.
+    get_presence_list_accepted = PresenceStore.__dict__[
+        "get_presence_list_accepted"
+    ]
+    get_presence_list_observers_accepted = PresenceStore.__dict__[
+        "get_presence_list_observers_accepted"
+    ]
+
+
+UPDATE_SYNCING_USERS_MS = 10 * 1000
+
+
+class SynchrotronPresence(object):
+    def __init__(self, hs):
+        self.is_mine_id = hs.is_mine_id
+        self.http_client = hs.get_simple_http_client()
+        self.store = hs.get_datastore()
+        self.user_to_num_current_syncs = {}
+        self.syncing_users_url = hs.config.worker_replication_url + "/syncing_users"
+        self.clock = hs.get_clock()
+        self.notifier = hs.get_notifier()
+
+        active_presence = self.store.take_presence_startup_info()
+        self.user_to_current_state = {
+            state.user_id: state
+            for state in active_presence
+        }
+
+        self.process_id = random_string(16)
+        logger.info("Presence process_id is %r", self.process_id)
+
+        self._sending_sync = False
+        self._need_to_send_sync = False
+        self.clock.looping_call(
+            self._send_syncing_users_regularly,
+            UPDATE_SYNCING_USERS_MS,
+        )
+
+        reactor.addSystemEventTrigger("before", "shutdown", self._on_shutdown)
+
+    def set_state(self, user, state, ignore_status_msg=False):
+        # TODO Hows this supposed to work?
+        pass
+
+    get_states = PresenceHandler.get_states.__func__
+    get_state = PresenceHandler.get_state.__func__
+    _get_interested_parties = PresenceHandler._get_interested_parties.__func__
+    current_state_for_users = PresenceHandler.current_state_for_users.__func__
+
+    @defer.inlineCallbacks
+    def user_syncing(self, user_id, affect_presence):
+        if affect_presence:
+            curr_sync = self.user_to_num_current_syncs.get(user_id, 0)
+            self.user_to_num_current_syncs[user_id] = curr_sync + 1
+            prev_states = yield self.current_state_for_users([user_id])
+            if prev_states[user_id].state == PresenceState.OFFLINE:
+                # TODO: Don't block the sync request on this HTTP hit.
+                yield self._send_syncing_users_now()
+
+        def _end():
+            # We check that the user_id is in user_to_num_current_syncs because
+            # user_to_num_current_syncs may have been cleared if we are
+            # shutting down.
+            if affect_presence and user_id in self.user_to_num_current_syncs:
+                self.user_to_num_current_syncs[user_id] -= 1
+
+        @contextlib.contextmanager
+        def _user_syncing():
+            try:
+                yield
+            finally:
+                _end()
+
+        defer.returnValue(_user_syncing())
+
+    @defer.inlineCallbacks
+    def _on_shutdown(self):
+        # When the synchrotron is shutdown tell the master to clear the in
+        # progress syncs for this process
+        self.user_to_num_current_syncs.clear()
+        yield self._send_syncing_users_now()
+
+    def _send_syncing_users_regularly(self):
+        # Only send an update if we aren't in the middle of sending one.
+        if not self._sending_sync:
+            preserve_fn(self._send_syncing_users_now)()
+
+    @defer.inlineCallbacks
+    def _send_syncing_users_now(self):
+        if self._sending_sync:
+            # We don't want to race with sending another update.
+            # Instead we wait for that update to finish and send another
+            # update afterwards.
+            self._need_to_send_sync = True
+            return
+
+        # Flag that we are sending an update.
+        self._sending_sync = True
+
+        yield self.http_client.post_json_get_json(self.syncing_users_url, {
+            "process_id": self.process_id,
+            "syncing_users": [
+                user_id for user_id, count in self.user_to_num_current_syncs.items()
+                if count > 0
+            ],
+        })
+
+        # Unset the flag as we are no longer sending an update.
+        self._sending_sync = False
+        if self._need_to_send_sync:
+            # If something happened while we were sending the update then
+            # we might need to send another update.
+            # TODO: Check if the update that was sent matches the current state
+            # as we only need to send an update if they are different.
+            self._need_to_send_sync = False
+            yield self._send_syncing_users_now()
+
+    @defer.inlineCallbacks
+    def notify_from_replication(self, states, stream_id):
+        parties = yield self._get_interested_parties(
+            states, calculate_remote_hosts=False
+        )
+        room_ids_to_states, users_to_states, _ = parties
+
+        self.notifier.on_new_event(
+            "presence_key", stream_id, rooms=room_ids_to_states.keys(),
+            users=users_to_states.keys()
+        )
+
+    @defer.inlineCallbacks
+    def process_replication(self, result):
+        stream = result.get("presence", {"rows": []})
+        states = []
+        for row in stream["rows"]:
+            (
+                position, user_id, state, last_active_ts,
+                last_federation_update_ts, last_user_sync_ts, status_msg,
+                currently_active
+            ) = row
+            state = UserPresenceState(
+                user_id, state, last_active_ts,
+                last_federation_update_ts, last_user_sync_ts, status_msg,
+                currently_active
+            )
+            self.user_to_current_state[user_id] = state
+            states.append(state)
+
+        if states and "position" in stream:
+            stream_id = int(stream["position"])
+            yield self.notify_from_replication(states, stream_id)
+
+
+class SynchrotronTyping(object):
+    def __init__(self, hs):
+        self._latest_room_serial = 0
+        self._room_serials = {}
+        self._room_typing = {}
+
+    def stream_positions(self):
+        # We must update this typing token from the response of the previous
+        # sync. In particular, the stream id may "reset" back to zero/a low
+        # value which we *must* use for the next replication request.
+        return {"typing": self._latest_room_serial}
+
+    def process_replication(self, result):
+        stream = result.get("typing")
+        if stream:
+            self._latest_room_serial = int(stream["position"])
+
+            for row in stream["rows"]:
+                position, room_id, typing_json = row
+                typing = json.loads(typing_json)
+                self._room_serials[room_id] = position
+                self._room_typing[room_id] = typing
+
+
+class SynchrotronApplicationService(object):
+    def notify_interested_services(self, event):
+        pass
+
+
+class SynchrotronServer(HomeServer):
+    def get_db_conn(self, run_new_connection=True):
+        # Any param beginning with cp_ is a parameter for adbapi, and should
+        # not be passed to the database engine.
+        db_params = {
+            k: v for k, v in self.db_config.get("args", {}).items()
+            if not k.startswith("cp_")
+        }
+        db_conn = self.database_engine.module.connect(**db_params)
+
+        if run_new_connection:
+            self.database_engine.on_new_connection(db_conn)
+        return db_conn
+
+    def setup(self):
+        logger.info("Setting up.")
+        self.datastore = SynchrotronSlavedStore(self.get_db_conn(), self)
+        logger.info("Finished setting up.")
+
+    def _listen_http(self, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        site_tag = listener_config.get("tag", port)
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "metrics":
+                    resources[METRICS_PREFIX] = MetricsResource(self)
+                elif name == "client":
+                    resource = JsonResource(self, canonical_json=False)
+                    sync.register_servlets(self, resource)
+                    events.register_servlets(self, resource)
+                    InitialSyncRestServlet(self).register(resource)
+                    RoomInitialSyncRestServlet(self).register(resource)
+                    resources.update({
+                        "/_matrix/client/r0": resource,
+                        "/_matrix/client/unstable": resource,
+                        "/_matrix/client/v2_alpha": resource,
+                        "/_matrix/client/api/v1": resource,
+                    })
+
+        root_resource = create_resource_tree(resources, Resource())
+        reactor.listenTCP(
+            port,
+            SynapseSite(
+                "synapse.access.http.%s" % (site_tag,),
+                site_tag,
+                listener_config,
+                root_resource,
+            ),
+            interface=bind_address
+        )
+        logger.info("Synapse synchrotron now listening on port %d", port)
+
+    def start_listening(self, listeners):
+        for listener in listeners:
+            if listener["type"] == "http":
+                self._listen_http(listener)
+            elif listener["type"] == "manhole":
+                reactor.listenTCP(
+                    listener["port"],
+                    manhole(
+                        username="matrix",
+                        password="rabbithole",
+                        globals={"hs": self},
+                    ),
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
+
+    @defer.inlineCallbacks
+    def replicate(self):
+        http_client = self.get_simple_http_client()
+        store = self.get_datastore()
+        replication_url = self.config.worker_replication_url
+        notifier = self.get_notifier()
+        presence_handler = self.get_presence_handler()
+        typing_handler = self.get_typing_handler()
+
+        def notify_from_stream(
+            result, stream_name, stream_key, room=None, user=None
+        ):
+            stream = result.get(stream_name)
+            if stream:
+                position_index = stream["field_names"].index("position")
+                if room:
+                    room_index = stream["field_names"].index(room)
+                if user:
+                    user_index = stream["field_names"].index(user)
+
+                users = ()
+                rooms = ()
+                for row in stream["rows"]:
+                    position = row[position_index]
+
+                    if user:
+                        users = (row[user_index],)
+
+                    if room:
+                        rooms = (row[room_index],)
+
+                    notifier.on_new_event(
+                        stream_key, position, users=users, rooms=rooms
+                    )
+
+        def notify(result):
+            stream = result.get("events")
+            if stream:
+                max_position = stream["position"]
+                for row in stream["rows"]:
+                    position = row[0]
+                    internal = json.loads(row[1])
+                    event_json = json.loads(row[2])
+                    event = FrozenEvent(event_json, internal_metadata_dict=internal)
+                    extra_users = ()
+                    if event.type == EventTypes.Member:
+                        extra_users = (event.state_key,)
+                    notifier.on_new_room_event(
+                        event, position, max_position, extra_users
+                    )
+
+            notify_from_stream(
+                result, "push_rules", "push_rules_key", user="user_id"
+            )
+            notify_from_stream(
+                result, "user_account_data", "account_data_key", user="user_id"
+            )
+            notify_from_stream(
+                result, "room_account_data", "account_data_key", user="user_id"
+            )
+            notify_from_stream(
+                result, "tag_account_data", "account_data_key", user="user_id"
+            )
+            notify_from_stream(
+                result, "receipts", "receipt_key", room="room_id"
+            )
+            notify_from_stream(
+                result, "typing", "typing_key", room="room_id"
+            )
+            notify_from_stream(
+                result, "to_device", "to_device_key", user="user_id"
+            )
+
+        while True:
+            try:
+                args = store.stream_positions()
+                args.update(typing_handler.stream_positions())
+                args["timeout"] = 30000
+                result = yield http_client.get_json(replication_url, args=args)
+                yield store.process_replication(result)
+                typing_handler.process_replication(result)
+                yield presence_handler.process_replication(result)
+                notify(result)
+            except:
+                logger.exception("Error replicating from %r", replication_url)
+                yield sleep(5)
+
+    def build_presence_handler(self):
+        return SynchrotronPresence(self)
+
+    def build_typing_handler(self):
+        return SynchrotronTyping(self)
+
+
+def start(config_options):
+    try:
+        config = HomeServerConfig.load_config(
+            "Synapse synchrotron", config_options
+        )
+    except ConfigError as e:
+        sys.stderr.write("\n" + e.message + "\n")
+        sys.exit(1)
+
+    assert config.worker_app == "synapse.app.synchrotron"
+
+    setup_logging(config.worker_log_config, config.worker_log_file)
+
+    database_engine = create_engine(config.database_config)
+
+    ss = SynchrotronServer(
+        config.server_name,
+        db_config=config.database_config,
+        config=config,
+        version_string="Synapse/" + get_version_string(synapse),
+        database_engine=database_engine,
+        application_service_handler=SynchrotronApplicationService(),
+    )
+
+    ss.setup()
+    ss.start_listening(config.worker_listeners)
+
+    def run():
+        with LoggingContext("run"):
+            logger.info("Running")
+            change_resource_limit(config.soft_file_limit)
+            if config.gc_thresholds:
+                gc.set_threshold(*config.gc_thresholds)
+            reactor.run()
+
+    def start():
+        ss.get_datastore().start_profiling()
+        ss.replicate()
+        ss.get_state_handler().start_caching()
+
+    reactor.callWhenRunning(start)
+
+    if config.worker_daemonize:
+        daemon = Daemonize(
+            app="synapse-synchrotron",
+            pid=config.worker_pid_file,
+            action=run,
+            auto_close_fds=False,
+            verbose=True,
+            logger=logger,
+        )
+        daemon.start()
+    else:
+        run()
+
+
+if __name__ == '__main__':
+    with LoggingContext("main"):
+        start(sys.argv[1:])

synapse/app/synctl.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,57 +14,199 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sys
+import argparse
+import collections
+import glob
 import os
-import subprocess
+import os.path
 import signal
+import subprocess
+import sys
+import yaml
 
-SYNAPSE = ["python", "-B", "-m", "synapse.app.homeserver"]
-
-CONFIGFILE = "homeserver.yaml"
-PIDFILE = "homeserver.pid"
+SYNAPSE = [sys.executable, "-B", "-m", "synapse.app.homeserver"]
 
 GREEN = "\x1b[1;32m"
+RED = "\x1b[1;31m"
 NORMAL = "\x1b[m"
 
 
-def start():
-    if not os.path.exists(CONFIGFILE):
-        sys.stderr.write(
-            "No config file found\n"
-            "To generate a config file, run '%s -c %s --generate-config"
-            " --server-name=<server name>'\n" % (
-                " ".join(SYNAPSE), CONFIGFILE
-            )
-        )
-        sys.exit(1)
-    print "Starting ...",
+def write(message, colour=NORMAL, stream=sys.stdout):
+    if colour == NORMAL:
+        stream.write(message + "\n")
+    else:
+        stream.write(colour + message + NORMAL + "\n")
+
+
+def start(configfile):
+    write("Starting ...")
     args = SYNAPSE
-    args.extend(["--daemonize", "-c", CONFIGFILE, "--pid-file", PIDFILE])
-    subprocess.check_call(args)
-    print GREEN + "started" + NORMAL
+    args.extend(["--daemonize", "-c", configfile])
+
+    try:
+        subprocess.check_call(args)
+        write("started synapse.app.homeserver(%r)" % (configfile,), colour=GREEN)
+    except subprocess.CalledProcessError as e:
+        write(
+            "error starting (exit code: %d); see above for logs" % e.returncode,
+            colour=RED,
+        )
 
 
-def stop():
-    if os.path.exists(PIDFILE):
-        pid = int(open(PIDFILE).read())
+def start_worker(app, configfile, worker_configfile):
+    args = [
+        "python", "-B",
+        "-m", app,
+        "-c", configfile,
+        "-c", worker_configfile
+    ]
+
+    try:
+        subprocess.check_call(args)
+        write("started %s(%r)" % (app, worker_configfile), colour=GREEN)
+    except subprocess.CalledProcessError as e:
+        write(
+            "error starting %s(%r) (exit code: %d); see above for logs" % (
+                app, worker_configfile, e.returncode,
+            ),
+            colour=RED,
+        )
+
+
+def stop(pidfile, app):
+    if os.path.exists(pidfile):
+        pid = int(open(pidfile).read())
         os.kill(pid, signal.SIGTERM)
-        print GREEN + "stopped" + NORMAL
+        write("stopped %s" % (app,), colour=GREEN)
+
+
+Worker = collections.namedtuple("Worker", [
+    "app", "configfile", "pidfile", "cache_factor"
+])
 
 
 def main():
-    action = sys.argv[1] if sys.argv[1:] else "usage"
-    if action == "start":
-        start()
-    elif action == "stop":
-        stop()
-    elif action == "restart":
-        stop()
-        start()
-    else:
-        sys.stderr.write("Usage: %s [start|stop|restart]\n" % (sys.argv[0],))
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "action",
+        choices=["start", "stop", "restart"],
+        help="whether to start, stop or restart the synapse",
+    )
+    parser.add_argument(
+        "configfile",
+        nargs="?",
+        default="homeserver.yaml",
+        help="the homeserver config file, defaults to homserver.yaml",
+    )
+    parser.add_argument(
+        "-w", "--worker",
+        metavar="WORKERCONFIG",
+        help="start or stop a single worker",
+    )
+    parser.add_argument(
+        "-a", "--all-processes",
+        metavar="WORKERCONFIGDIR",
+        help="start or stop all the workers in the given directory"
+             " and the main synapse process",
+    )
+
+    options = parser.parse_args()
+
+    if options.worker and options.all_processes:
+        write(
+            'Cannot use "--worker" with "--all-processes"',
+            stream=sys.stderr
+        )
         sys.exit(1)
 
+    configfile = options.configfile
+
+    if not os.path.exists(configfile):
+        write(
+            "No config file found\n"
+            "To generate a config file, run '%s -c %s --generate-config"
+            " --server-name=<server name>'\n" % (
+                " ".join(SYNAPSE), options.configfile
+            ),
+            stream=sys.stderr,
+        )
+        sys.exit(1)
+
+    with open(configfile) as stream:
+        config = yaml.load(stream)
+
+    pidfile = config["pid_file"]
+    cache_factor = config.get("synctl_cache_factor")
+    start_stop_synapse = True
+
+    if cache_factor:
+        os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)
+
+    worker_configfiles = []
+    if options.worker:
+        start_stop_synapse = False
+        worker_configfile = options.worker
+        if not os.path.exists(worker_configfile):
+            write(
+                "No worker config found at %r" % (worker_configfile,),
+                stream=sys.stderr,
+            )
+            sys.exit(1)
+        worker_configfiles.append(worker_configfile)
+
+    if options.all_processes:
+        worker_configdir = options.all_processes
+        if not os.path.isdir(worker_configdir):
+            write(
+                "No worker config directory found at %r" % (worker_configdir,),
+                stream=sys.stderr,
+            )
+            sys.exit(1)
+        worker_configfiles.extend(sorted(glob.glob(
+            os.path.join(worker_configdir, "*.yaml")
+        )))
+
+    workers = []
+    for worker_configfile in worker_configfiles:
+        with open(worker_configfile) as stream:
+            worker_config = yaml.load(stream)
+        worker_app = worker_config["worker_app"]
+        worker_pidfile = worker_config["worker_pid_file"]
+        worker_daemonize = worker_config["worker_daemonize"]
+        assert worker_daemonize  # TODO print something more user friendly
+        worker_cache_factor = worker_config.get("synctl_cache_factor")
+        workers.append(Worker(
+            worker_app, worker_configfile, worker_pidfile, worker_cache_factor,
+        ))
+
+    action = options.action
+
+    if action == "stop" or action == "restart":
+        for worker in workers:
+            stop(worker.pidfile, worker.app)
+
+        if start_stop_synapse:
+            stop(pidfile, "synapse.app.homeserver")
+
+        # TODO: Wait for synapse to actually shutdown before starting it again
+
+    if action == "start" or action == "restart":
+        if start_stop_synapse:
+            start(configfile)
+
+        for worker in workers:
+            if worker.cache_factor:
+                os.environ["SYNAPSE_CACHE_FACTOR"] = str(worker.cache_factor)
+
+            start_worker(worker.app, configfile, worker.configfile)
+
+            if cache_factor:
+                os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)
+            else:
+                os.environ.pop("SYNAPSE_CACHE_FACTOR", None)
+
 
 if __name__ == "__main__":
     main()

synapse/appservice/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,6 +14,8 @@
 # limitations under the License.
 from synapse.api.constants import EventTypes
 
+from twisted.internet import defer
+
 import logging
 import re
 
@@ -79,7 +81,7 @@ class ApplicationService(object):
     NS_LIST = [NS_USERS, NS_ALIASES, NS_ROOMS]
 
     def __init__(self, token, url=None, namespaces=None, hs_token=None,
-                 sender=None, id=None):
+                 sender=None, id=None, protocols=None, rate_limited=True):
         self.token = token
         self.url = url
         self.hs_token = hs_token
@@ -87,6 +89,14 @@ class ApplicationService(object):
         self.namespaces = self._check_namespaces(namespaces)
         self.id = id
 
+        # .protocols is a publicly visible field
+        if protocols:
+            self.protocols = set(protocols)
+        else:
+            self.protocols = set()
+
+        self.rate_limited = rate_limited
+
     def _check_namespaces(self, namespaces):
         # Sanity check that it is of the form:
         # {
@@ -138,65 +148,66 @@ class ApplicationService(object):
             return regex_obj["exclusive"]
         return False
 
-    def _matches_user(self, event, member_list):
-        if (hasattr(event, "sender") and
-                self.is_interested_in_user(event.sender)):
-            return True
+    @defer.inlineCallbacks
+    def _matches_user(self, event, store):
+        if not event:
+            defer.returnValue(False)
+
+        if self.is_interested_in_user(event.sender):
+            defer.returnValue(True)
         # also check m.room.member state key
-        if (hasattr(event, "type") and event.type == EventTypes.Member
-                and hasattr(event, "state_key")
-                and self.is_interested_in_user(event.state_key)):
-            return True
+        if (event.type == EventTypes.Member and
+                self.is_interested_in_user(event.state_key)):
+            defer.returnValue(True)
+
+        if not store:
+            defer.returnValue(False)
+
+        member_list = yield store.get_users_in_room(event.room_id)
+
         # check joined member events
-        for member in member_list:
-            if self.is_interested_in_user(member.state_key):
-                return True
-        return False
+        for user_id in member_list:
+            if self.is_interested_in_user(user_id):
+                defer.returnValue(True)
+        defer.returnValue(False)
 
     def _matches_room_id(self, event):
         if hasattr(event, "room_id"):
             return self.is_interested_in_room(event.room_id)
         return False
 
-    def _matches_aliases(self, event, alias_list):
+    @defer.inlineCallbacks
+    def _matches_aliases(self, event, store):
+        if not store or not event:
+            defer.returnValue(False)
+
+        alias_list = yield store.get_aliases_for_room(event.room_id)
         for alias in alias_list:
             if self.is_interested_in_alias(alias):
-                return True
-        return False
+                defer.returnValue(True)
+        defer.returnValue(False)
 
-    def is_interested(self, event, restrict_to=None, aliases_for_event=None,
-                      member_list=None):
+    @defer.inlineCallbacks
+    def is_interested(self, event, store=None):
         """Check if this service is interested in this event.
 
         Args:
             event(Event): The event to check.
-            restrict_to(str): The namespace to restrict regex tests to.
-            aliases_for_event(list): A list of all the known room aliases for
-            this event.
-            member_list(list): A list of all joined room members in this room.
+            store(DataStore)
         Returns:
             bool: True if this service would like to know about this event.
         """
-        if aliases_for_event is None:
-            aliases_for_event = []
-        if member_list is None:
-            member_list = []
+        # Do cheap checks first
+        if self._matches_room_id(event):
+            defer.returnValue(True)
 
-        if restrict_to and restrict_to not in ApplicationService.NS_LIST:
-            # this is a programming error, so fail early and raise a general
-            # exception
-            raise Exception("Unexpected restrict_to value: %s". restrict_to)
+        if (yield self._matches_aliases(event, store)):
+            defer.returnValue(True)
 
-        if not restrict_to:
-            return (self._matches_user(event, member_list)
-                    or self._matches_aliases(event, aliases_for_event)
-                    or self._matches_room_id(event))
-        elif restrict_to == ApplicationService.NS_ALIASES:
-            return self._matches_aliases(event, aliases_for_event)
-        elif restrict_to == ApplicationService.NS_ROOMS:
-            return self._matches_room_id(event)
-        elif restrict_to == ApplicationService.NS_USERS:
-            return self._matches_user(event, member_list)
+        if (yield self._matches_user(event, store)):
+            defer.returnValue(True)
+
+        defer.returnValue(False)
 
     def is_interested_in_user(self, user_id):
         return (
@@ -216,11 +227,17 @@ class ApplicationService(object):
             or user_id == self.sender
         )
 
+    def is_interested_in_protocol(self, protocol):
+        return protocol in self.protocols
+
     def is_exclusive_alias(self, alias):
         return self._is_exclusive(ApplicationService.NS_ALIASES, alias)
 
     def is_exclusive_room(self, room_id):
         return self._is_exclusive(ApplicationService.NS_ROOMS, room_id)
 
+    def is_rate_limited(self):
+        return self.rate_limited
+
     def __str__(self):
         return "ApplicationService: %s" % (self.__dict__,)

synapse/appservice/api.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,9 +14,11 @@
 # limitations under the License.
 from twisted.internet import defer
 
+from synapse.api.constants import ThirdPartyEntityKind
 from synapse.api.errors import CodeMessageException
 from synapse.http.client import SimpleHttpClient
 from synapse.events.utils import serialize_event
+from synapse.util.caches.response_cache import ResponseCache
 
 import logging
 import urllib
@@ -24,17 +26,57 @@ import urllib
 logger = logging.getLogger(__name__)
 
 
+HOUR_IN_MS = 60 * 60 * 1000
+
+
+APP_SERVICE_PREFIX = "/_matrix/app/unstable"
+
+
+def _is_valid_3pe_metadata(info):
+    if "instances" not in info:
+        return False
+    if not isinstance(info["instances"], list):
+        return False
+    return True
+
+
+def _is_valid_3pe_result(r, field):
+    if not isinstance(r, dict):
+        return False
+
+    for k in (field, "protocol"):
+        if k not in r:
+            return False
+        if not isinstance(r[k], str):
+            return False
+
+    if "fields" not in r:
+        return False
+    fields = r["fields"]
+    if not isinstance(fields, dict):
+        return False
+    for k in fields.keys():
+        if not isinstance(fields[k], str):
+            return False
+
+    return True
+
+
 class ApplicationServiceApi(SimpleHttpClient):
     """This class manages HS -> AS communications, including querying and
     pushing.
     """
 
-    def __init__(self,  hs):
+    def __init__(self, hs):
         super(ApplicationServiceApi, self).__init__(hs)
         self.clock = hs.get_clock()
 
+        self.protocol_meta_cache = ResponseCache(hs, timeout_ms=HOUR_IN_MS)
+
     @defer.inlineCallbacks
     def query_user(self, service, user_id):
+        if service.url is None:
+            defer.returnValue(False)
         uri = service.url + ("/users/%s" % urllib.quote(user_id))
         response = None
         try:
@@ -54,6 +96,8 @@ class ApplicationServiceApi(SimpleHttpClient):
 
     @defer.inlineCallbacks
     def query_alias(self, service, alias):
+        if service.url is None:
+            defer.returnValue(False)
         uri = service.url + ("/rooms/%s" % urllib.quote(alias))
         response = None
         try:
@@ -71,8 +115,84 @@ class ApplicationServiceApi(SimpleHttpClient):
             logger.warning("query_alias to %s threw exception %s", uri, ex)
         defer.returnValue(False)
 
+    @defer.inlineCallbacks
+    def query_3pe(self, service, kind, protocol, fields):
+        if kind == ThirdPartyEntityKind.USER:
+            required_field = "userid"
+        elif kind == ThirdPartyEntityKind.LOCATION:
+            required_field = "alias"
+        else:
+            raise ValueError(
+                "Unrecognised 'kind' argument %r to query_3pe()", kind
+            )
+        if service.url is None:
+            defer.returnValue([])
+
+        uri = "%s%s/thirdparty/%s/%s" % (
+            service.url,
+            APP_SERVICE_PREFIX,
+            kind,
+            urllib.quote(protocol)
+        )
+        try:
+            response = yield self.get_json(uri, fields)
+            if not isinstance(response, list):
+                logger.warning(
+                    "query_3pe to %s returned an invalid response %r",
+                    uri, response
+                )
+                defer.returnValue([])
+
+            ret = []
+            for r in response:
+                if _is_valid_3pe_result(r, field=required_field):
+                    ret.append(r)
+                else:
+                    logger.warning(
+                        "query_3pe to %s returned an invalid result %r",
+                        uri, r
+                    )
+
+            defer.returnValue(ret)
+        except Exception as ex:
+            logger.warning("query_3pe to %s threw exception %s", uri, ex)
+            defer.returnValue([])
+
+    def get_3pe_protocol(self, service, protocol):
+        if service.url is None:
+            defer.returnValue({})
+
+        @defer.inlineCallbacks
+        def _get():
+            uri = "%s%s/thirdparty/protocol/%s" % (
+                service.url,
+                APP_SERVICE_PREFIX,
+                urllib.quote(protocol)
+            )
+            try:
+                info = yield self.get_json(uri, {})
+
+                if not _is_valid_3pe_metadata(info):
+                    logger.warning("query_3pe_protocol to %s did not return a"
+                                   " valid result", uri)
+                    defer.returnValue(None)
+
+                defer.returnValue(info)
+            except Exception as ex:
+                logger.warning("query_3pe_protocol to %s threw exception %s",
+                               uri, ex)
+                defer.returnValue(None)
+
+        key = (service.id, protocol)
+        return self.protocol_meta_cache.get(key) or (
+            self.protocol_meta_cache.set(key, _get())
+        )
+
     @defer.inlineCallbacks
     def push_bulk(self, service, events, txn_id=None):
+        if service.url is None:
+            defer.returnValue(True)
+
         events = self._serialize(events)
 
         if txn_id is None:
@@ -100,11 +220,6 @@ class ApplicationServiceApi(SimpleHttpClient):
             logger.warning("push_bulk to %s threw exception %s", uri, ex)
         defer.returnValue(False)
 
-    @defer.inlineCallbacks
-    def push(self, service, event, txn_id=None):
-        response = yield self.push_bulk(service, [event], txn_id)
-        defer.returnValue(response)
-
     def _serialize(self, events):
         time_now = self.clock.time_msec()
         return [

synapse/appservice/scheduler.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -48,32 +48,35 @@ UP & quit           +---------- YES                       SUCCESS
 This is all tied together by the AppServiceScheduler which DIs the required
 components.
 """
+from twisted.internet import defer
 
 from synapse.appservice import ApplicationServiceState
-from twisted.internet import defer
+from synapse.util.logcontext import preserve_fn
+from synapse.util.metrics import Measure
+
 import logging
 
 logger = logging.getLogger(__name__)
 
 
-class AppServiceScheduler(object):
+class ApplicationServiceScheduler(object):
     """ Public facing API for this module. Does the required DI to tie the
     components together. This also serves as the "event_pool", which in this
     case is a simple array.
     """
 
-    def __init__(self, clock, store, as_api):
-        self.clock = clock
-        self.store = store
-        self.as_api = as_api
+    def __init__(self, hs):
+        self.clock = hs.get_clock()
+        self.store = hs.get_datastore()
+        self.as_api = hs.get_application_service_api()
 
         def create_recoverer(service, callback):
-            return _Recoverer(clock, store, as_api, service, callback)
+            return _Recoverer(self.clock, self.store, self.as_api, service, callback)
 
         self.txn_ctrl = _TransactionController(
-            clock, store, as_api, create_recoverer
+            self.clock, self.store, self.as_api, create_recoverer
         )
-        self.queuer = _ServiceQueuer(self.txn_ctrl)
+        self.queuer = _ServiceQueuer(self.txn_ctrl, self.clock)
 
     @defer.inlineCallbacks
     def start(self):
@@ -94,38 +97,36 @@ class _ServiceQueuer(object):
     this schedules any other events in the queue to run.
     """
 
-    def __init__(self, txn_ctrl):
+    def __init__(self, txn_ctrl, clock):
         self.queued_events = {}  # dict of {service_id: [events]}
-        self.pending_requests = {}  # dict of {service_id: Deferred}
+        self.requests_in_flight = set()
         self.txn_ctrl = txn_ctrl
+        self.clock = clock
 
     def enqueue(self, service, event):
         # if this service isn't being sent something
-        if not self.pending_requests.get(service.id):
-            self._send_request(service, [event])
-        else:
-            # add to queue for this service
-            if service.id not in self.queued_events:
-                self.queued_events[service.id] = []
-            self.queued_events[service.id].append(event)
+        self.queued_events.setdefault(service.id, []).append(event)
+        preserve_fn(self._send_request)(service)
 
-    def _send_request(self, service, events):
-        # send request and add callbacks
-        d = self.txn_ctrl.send(service, events)
-        d.addBoth(self._on_request_finish)
-        d.addErrback(self._on_request_fail)
-        self.pending_requests[service.id] = d
+    @defer.inlineCallbacks
+    def _send_request(self, service):
+        if service.id in self.requests_in_flight:
+            return
 
-    def _on_request_finish(self, service):
-        self.pending_requests[service.id] = None
-        # if there are queued events, then send them.
-        if (service.id in self.queued_events
-                and len(self.queued_events[service.id]) > 0):
-            self._send_request(service, self.queued_events[service.id])
-            self.queued_events[service.id] = []
+        self.requests_in_flight.add(service.id)
+        try:
+            while True:
+                events = self.queued_events.pop(service.id, [])
+                if not events:
+                    return
 
-    def _on_request_fail(self, err):
-        logger.error("AS request failed: %s", err)
+                with Measure(self.clock, "servicequeuer.send"):
+                    try:
+                        yield self.txn_ctrl.send(service, events)
+                    except:
+                        logger.exception("AS request failed")
+        finally:
+            self.requests_in_flight.discard(service.id)
 
 
 class _TransactionController(object):
@@ -149,14 +150,12 @@ class _TransactionController(object):
             if service_is_up:
                 sent = yield txn.send(self.as_api)
                 if sent:
-                    txn.complete(self.store)
+                    yield txn.complete(self.store)
                 else:
-                    self._start_recoverer(service)
+                    preserve_fn(self._start_recoverer)(service)
         except Exception as e:
             logger.exception(e)
-            self._start_recoverer(service)
-        # request has finished
-        defer.returnValue(service)
+            preserve_fn(self._start_recoverer)(service)
 
     @defer.inlineCallbacks
     def on_recovered(self, recoverer):
@@ -224,8 +223,8 @@ class _Recoverer(object):
         self.clock.call_later((2 ** self.backoff_counter), self.retry)
 
     def _backoff(self):
-        # cap the backoff to be around 18h => (2^16) = 65536 secs
-        if self.backoff_counter < 16:
+        # cap the backoff to be around 8.5min => (2^9) = 512 secs
+        if self.backoff_counter < 9:
             self.backoff_counter += 1
         self.recover()
 

synapse/config/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

synapse/config/__main__.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 OpenMarket 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.
+from synapse.config._base import ConfigError
+
+if __name__ == "__main__":
+    import sys
+    from homeserver import HomeServerConfig
+
+    action = sys.argv[1]
+
+    if action == "read":
+        key = sys.argv[2]
+        try:
+            config = HomeServerConfig.load_config("", sys.argv[3:])
+        except ConfigError as e:
+            sys.stderr.write("\n" + e.message + "\n")
+            sys.exit(1)
+
+        print (getattr(config, key))
+        sys.exit(0)
+    else:
+        sys.stderr.write("Unknown command %r\n" % (action,))
+        sys.exit(1)

synapse/config/_base.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,28 +14,67 @@
 # limitations under the License.
 
 import argparse
-import sys
+import errno
 import os
 import yaml
+from textwrap import dedent
 
 
 class ConfigError(Exception):
     pass
 
 
-class Config(object):
-    def __init__(self, args):
-        pass
+# We split these messages out to allow packages to override with package
+# specific instructions.
+MISSING_REPORT_STATS_CONFIG_INSTRUCTIONS = """\
+Please opt in or out of reporting anonymized homeserver usage statistics, by
+setting the `report_stats` key in your config file to either True or False.
+"""
 
+MISSING_REPORT_STATS_SPIEL = """\
+We would really appreciate it if you could help our project out by reporting
+anonymized usage statistics from your homeserver. Only very basic aggregate
+data (e.g. number of users) will be reported, but it helps us to track the
+growth of the Matrix community, and helps us to make Matrix a success, as well
+as to convince other networks that they should peer with us.
+
+Thank you.
+"""
+
+MISSING_SERVER_NAME = """\
+Missing mandatory `server_name` config option.
+"""
+
+
+class Config(object):
     @staticmethod
-    def parse_size(string):
+    def parse_size(value):
+        if isinstance(value, int) or isinstance(value, long):
+            return value
         sizes = {"K": 1024, "M": 1024 * 1024}
         size = 1
-        suffix = string[-1]
+        suffix = value[-1]
         if suffix in sizes:
-            string = string[:-1]
+            value = value[:-1]
             size = sizes[suffix]
-        return int(string) * size
+        return int(value) * size
+
+    @staticmethod
+    def parse_duration(value):
+        if isinstance(value, int) or isinstance(value, long):
+            return value
+        second = 1000
+        hour = 60 * 60 * second
+        day = 24 * hour
+        week = 7 * day
+        year = 365 * day
+        sizes = {"s": second, "h": hour, "d": day, "w": week, "y": year}
+        size = 1
+        suffix = value[-1]
+        if suffix in sizes:
+            value = value[:-1]
+            size = sizes[suffix]
+        return int(value) * size
 
     @staticmethod
     def abspath(file_path):
@@ -63,8 +102,11 @@ class Config(object):
     @classmethod
     def ensure_directory(cls, dir_path):
         dir_path = cls.abspath(dir_path)
-        if not os.path.exists(dir_path):
+        try:
             os.makedirs(dir_path)
+        except OSError as e:
+            if e.errno != errno.EEXIST:
+                raise
         if not os.path.isdir(dir_path):
             raise ConfigError(
                 "%s is not a directory" % (dir_path,)
@@ -77,17 +119,6 @@ class Config(object):
         with open(file_path) as file_stream:
             return file_stream.read()
 
-    @classmethod
-    def read_yaml_file(cls, file_path, config_name):
-        cls.check_file(file_path, config_name)
-        with open(file_path) as file_stream:
-            try:
-                return yaml.load(file_stream)
-            except:
-                raise ConfigError(
-                    "Error parsing yaml in file %r" % (file_path,)
-                )
-
     @staticmethod
     def default_path(name):
         return os.path.abspath(os.path.join(os.path.curdir, name))
@@ -97,84 +128,272 @@ class Config(object):
         with open(file_path) as file_stream:
             return yaml.load(file_stream)
 
-    @classmethod
-    def add_arguments(cls, parser):
-        pass
+    def invoke_all(self, name, *args, **kargs):
+        results = []
+        for cls in type(self).mro():
+            if name in cls.__dict__:
+                results.append(getattr(cls, name)(self, *args, **kargs))
+        return results
+
+    def generate_config(
+            self,
+            config_dir_path,
+            server_name,
+            is_generating_file,
+            report_stats=None,
+    ):
+        default_config = "# vim:ft=yaml\n"
+
+        default_config += "\n\n".join(dedent(conf) for conf in self.invoke_all(
+            "default_config",
+            config_dir_path=config_dir_path,
+            server_name=server_name,
+            is_generating_file=is_generating_file,
+            report_stats=report_stats,
+        ))
+
+        config = yaml.load(default_config)
+
+        return default_config, config
 
     @classmethod
-    def generate_config(cls, args, config_dir_path):
-        pass
+    def load_config(cls, description, argv):
+        config_parser = argparse.ArgumentParser(
+            description=description,
+        )
+        config_parser.add_argument(
+            "-c", "--config-path",
+            action="append",
+            metavar="CONFIG_FILE",
+            help="Specify config file. Can be given multiple times and"
+                 " may specify directories containing *.yaml files."
+        )
+
+        config_parser.add_argument(
+            "--keys-directory",
+            metavar="DIRECTORY",
+            help="Where files such as certs and signing keys are stored when"
+                 " their location is given explicitly in the config."
+                 " Defaults to the directory containing the last config file",
+        )
+
+        config_args = config_parser.parse_args(argv)
+
+        config_files = find_config_files(search_paths=config_args.config_path)
+
+        obj = cls()
+        obj.read_config_files(
+            config_files,
+            keys_directory=config_args.keys_directory,
+            generate_keys=False,
+        )
+        return obj
 
     @classmethod
-    def load_config(cls, description, argv, generate_section=None):
+    def load_or_generate_config(cls, description, argv):
         config_parser = argparse.ArgumentParser(add_help=False)
         config_parser.add_argument(
             "-c", "--config-path",
+            action="append",
             metavar="CONFIG_FILE",
-            help="Specify config file"
+            help="Specify config file. Can be given multiple times and"
+                 " may specify directories containing *.yaml files."
         )
         config_parser.add_argument(
             "--generate-config",
             action="store_true",
-            help="Generate config file"
+            help="Generate a config file for the server name"
+        )
+        config_parser.add_argument(
+            "--report-stats",
+            action="store",
+            help="Whether the generated config reports anonymized usage statistics",
+            choices=["yes", "no"]
+        )
+        config_parser.add_argument(
+            "--generate-keys",
+            action="store_true",
+            help="Generate any missing key files then exit"
+        )
+        config_parser.add_argument(
+            "--keys-directory",
+            metavar="DIRECTORY",
+            help="Used with 'generate-*' options to specify where files such as"
+                 " certs and signing keys should be stored in, unless explicitly"
+                 " specified in the config."
+        )
+        config_parser.add_argument(
+            "-H", "--server-name",
+            help="The server name to generate a config file for"
         )
         config_args, remaining_args = config_parser.parse_known_args(argv)
 
+        config_files = find_config_files(search_paths=config_args.config_path)
+
+        generate_keys = config_args.generate_keys
+
+        obj = cls()
+
         if config_args.generate_config:
-            if not config_args.config_path:
+            if config_args.report_stats is None:
                 config_parser.error(
-                    "Must specify where to generate the config file"
+                    "Please specify either --report-stats=yes or --report-stats=no\n\n" +
+                    MISSING_REPORT_STATS_SPIEL
                 )
-            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)
+            if not config_files:
+                config_parser.error(
+                    "Must supply a config file.\nA config file can be automatically"
+                    " generated using \"--generate-config -H SERVER_NAME"
+                    " -c CONFIG-FILE\""
+                )
+            (config_path,) = config_files
+            if not os.path.exists(config_path):
+                if config_args.keys_directory:
+                    config_dir_path = config_args.keys_directory
+                else:
+                    config_dir_path = os.path.dirname(config_path)
+                config_dir_path = os.path.abspath(config_dir_path)
+
+                server_name = config_args.server_name
+                if not server_name:
+                    raise ConfigError(
+                        "Must specify a server_name to a generate config for."
+                        " Pass -H server.name."
+                    )
+                if not os.path.exists(config_dir_path):
+                    os.makedirs(config_dir_path)
+                with open(config_path, "wb") as config_file:
+                    config_bytes, config = obj.generate_config(
+                        config_dir_path=config_dir_path,
+                        server_name=server_name,
+                        report_stats=(config_args.report_stats == "yes"),
+                        is_generating_file=True
+                    )
+                    obj.invoke_all("generate_files", config)
+                    config_file.write(config_bytes)
+                print (
+                    "A config file has been generated in %r for server name"
+                    " %r with corresponding SSL keys and self-signed"
+                    " certificates. Please review this file and customise it"
+                    " to your needs."
+                ) % (config_path, server_name)
+                print (
+                    "If this server name is incorrect, you will need to"
+                    " regenerate the SSL certificates"
+                )
+                return
             else:
-                defaults = {}
-        else:
-            if config_args.config_path:
-                defaults = cls.read_config_file(config_args.config_path)
-            else:
-                defaults = {}
+                print (
+                    "Config file %r already exists. Generating any missing key"
+                    " files."
+                ) % (config_path,)
+                generate_keys = True
 
         parser = argparse.ArgumentParser(
             parents=[config_parser],
             description=description,
             formatter_class=argparse.RawDescriptionHelpFormatter,
         )
-        cls.add_arguments(parser)
-        parser.set_defaults(**defaults)
 
+        obj.invoke_all("add_arguments", parser)
         args = parser.parse_args(remaining_args)
 
-        if config_args.generate_config:
-            config_dir_path = os.path.dirname(config_args.config_path)
-            config_dir_path = os.path.abspath(config_dir_path)
-            if not os.path.exists(config_dir_path):
-                os.makedirs(config_dir_path)
-            cls.generate_config(args, config_dir_path)
-            config = {}
-            for key, value in vars(args).items():
-                if (key not in set(["config_path", "generate_config"])
-                        and value is not None):
-                    config[key] = value
-            with open(config_args.config_path, "w") as config_file:
-                # TODO(mark/paul) We might want to output emacs-style mode
-                # markers as well as vim-style mode markers into the file,
-                # to further hint to people this is a YAML file.
-                config_file.write("# vim:ft=yaml\n")
-                yaml.dump(config, config_file, default_flow_style=False)
-            print (
-                "A config file has been generated in %s for server name"
-                " '%s' with corresponding SSL keys and self-signed"
-                " certificates. Please review this file and customise it to"
-                " your needs."
-            ) % (
-                config_args.config_path, config['server_name']
+        if not config_files:
+            config_parser.error(
+                "Must supply a config file.\nA config file can be automatically"
+                " generated using \"--generate-config -H SERVER_NAME"
+                " -c CONFIG-FILE\""
             )
-            print (
-                "If this server name is incorrect, you will need to regenerate"
-                " the SSL certificates"
-            )
-            sys.exit(0)
 
-        return cls(args)
+        obj.read_config_files(
+            config_files,
+            keys_directory=config_args.keys_directory,
+            generate_keys=generate_keys,
+        )
+
+        if generate_keys:
+            return None
+
+        obj.invoke_all("read_arguments", args)
+
+        return obj
+
+    def read_config_files(self, config_files, keys_directory=None,
+                          generate_keys=False):
+        if not keys_directory:
+            keys_directory = os.path.dirname(config_files[-1])
+
+        config_dir_path = os.path.abspath(keys_directory)
+
+        specified_config = {}
+        for config_file in config_files:
+            yaml_config = self.read_config_file(config_file)
+            specified_config.update(yaml_config)
+
+        if "server_name" not in specified_config:
+            raise ConfigError(MISSING_SERVER_NAME)
+
+        server_name = specified_config["server_name"]
+        _, config = self.generate_config(
+            config_dir_path=config_dir_path,
+            server_name=server_name,
+            is_generating_file=False,
+        )
+        config.pop("log_config")
+        config.update(specified_config)
+
+        if "report_stats" not in config:
+            raise ConfigError(
+                MISSING_REPORT_STATS_CONFIG_INSTRUCTIONS + "\n" +
+                MISSING_REPORT_STATS_SPIEL
+            )
+
+        if generate_keys:
+            self.invoke_all("generate_files", config)
+            return
+
+        self.invoke_all("read_config", config)
+
+
+def find_config_files(search_paths):
+    """Finds config files using a list of search paths. If a path is a file
+    then that file path is added to the list. If a search path is a directory
+    then all the "*.yaml" files in that directory are added to the list in
+    sorted order.
+
+    Args:
+        search_paths(list(str)): A list of paths to search.
+
+    Returns:
+        list(str): A list of file paths.
+    """
+
+    config_files = []
+    if search_paths:
+        for config_path in search_paths:
+            if os.path.isdir(config_path):
+                # We accept specifying directories as config paths, we search
+                # inside that directory for all files matching *.yaml, and then
+                # we apply them in *sorted* order.
+                files = []
+                for entry in os.listdir(config_path):
+                    entry_path = os.path.join(config_path, entry)
+                    if not os.path.isfile(entry_path):
+                        print (
+                            "Found subdirectory in config directory: %r. IGNORING."
+                        ) % (entry_path, )
+                        continue
+
+                    if not entry.endswith(".yaml"):
+                        print (
+                            "Found file in config directory that does not"
+                            " end in '.yaml': %r. IGNORING."
+                        ) % (entry_path, )
+                        continue
+
+                    files.append(entry_path)
+
+                config_files.extend(sorted(files))
+            else:
+                config_files.append(config_path)
+    return config_files

synapse/config/api.py

@@ -0,0 +1,40 @@
+# Copyright 2015, 2016 OpenMarket 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.
+
+from ._base import Config
+
+from synapse.api.constants import EventTypes
+
+
+class ApiConfig(Config):
+
+    def read_config(self, config):
+        self.room_invite_state_types = config.get("room_invite_state_types", [
+            EventTypes.JoinRules,
+            EventTypes.CanonicalAlias,
+            EventTypes.RoomAvatar,
+            EventTypes.Name,
+        ])
+
+    def default_config(cls, **kwargs):
+        return """\
+        ## API Configuration ##
+
+        # A list of event types that will be included in the room_invite_state
+        room_invite_state_types:
+            - "{JoinRules}"
+            - "{CanonicalAlias}"
+            - "{RoomAvatar}"
+            - "{Name}"
+        """.format(**vars(EventTypes))

synapse/config/appservice.py

@@ -1,4 +1,4 @@
-# Copyright 2015 OpenMarket Ltd
+# Copyright 2015, 2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,20 +12,153 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ._base import Config
+from ._base import Config, ConfigError
+
+from synapse.appservice import ApplicationService
+from synapse.types import UserID
+
+import urllib
+import yaml
+import logging
+
+logger = logging.getLogger(__name__)
 
 
 class AppServiceConfig(Config):
 
-    def __init__(self, args):
-        super(AppServiceConfig, self).__init__(args)
-        self.app_service_config_files = args.app_service_config_files
+    def read_config(self, config):
+        self.app_service_config_files = config.get("app_service_config_files", [])
+        self.notify_appservices = config.get("notify_appservices", True)
 
-    @classmethod
-    def add_arguments(cls, parser):
-        super(AppServiceConfig, cls).add_arguments(parser)
-        group = parser.add_argument_group("appservice")
-        group.add_argument(
-            "--app-service-config-files", type=str, nargs='+',
-            help="A list of application service config files to use."
+    def default_config(cls, **kwargs):
+        return """\
+        # A list of application service config file to use
+        app_service_config_files: []
+        """
+
+
+def load_appservices(hostname, config_files):
+    """Returns a list of Application Services from the config files."""
+    if not isinstance(config_files, list):
+        logger.warning(
+            "Expected %s to be a list of AS config files.", config_files
         )
+        return []
+
+    # Dicts of value -> filename
+    seen_as_tokens = {}
+    seen_ids = {}
+
+    appservices = []
+
+    for config_file in config_files:
+        try:
+            with open(config_file, 'r') as f:
+                appservice = _load_appservice(
+                    hostname, yaml.load(f), config_file
+                )
+                if appservice.id in seen_ids:
+                    raise ConfigError(
+                        "Cannot reuse ID across application services: "
+                        "%s (files: %s, %s)" % (
+                            appservice.id, config_file, seen_ids[appservice.id],
+                        )
+                    )
+                seen_ids[appservice.id] = config_file
+                if appservice.token in seen_as_tokens:
+                    raise ConfigError(
+                        "Cannot reuse as_token across application services: "
+                        "%s (files: %s, %s)" % (
+                            appservice.token,
+                            config_file,
+                            seen_as_tokens[appservice.token],
+                        )
+                    )
+                seen_as_tokens[appservice.token] = config_file
+                logger.info("Loaded application service: %s", appservice)
+                appservices.append(appservice)
+        except Exception as e:
+            logger.error("Failed to load appservice from '%s'", config_file)
+            logger.exception(e)
+            raise
+    return appservices
+
+
+def _load_appservice(hostname, as_info, config_filename):
+    required_string_fields = [
+        "id", "as_token", "hs_token", "sender_localpart"
+    ]
+    for field in required_string_fields:
+        if not isinstance(as_info.get(field), basestring):
+            raise KeyError("Required string field: '%s' (%s)" % (
+                field, config_filename,
+            ))
+
+    # 'url' must either be a string or explicitly null, not missing
+    # to avoid accidentally turning off push for ASes.
+    if (not isinstance(as_info.get("url"), basestring) and
+            as_info.get("url", "") is not None):
+        raise KeyError(
+            "Required string field or explicit null: 'url' (%s)" % (config_filename,)
+        )
+
+    localpart = as_info["sender_localpart"]
+    if urllib.quote(localpart) != localpart:
+        raise ValueError(
+            "sender_localpart needs characters which are not URL encoded."
+        )
+    user = UserID(localpart, hostname)
+    user_id = user.to_string()
+
+    # Rate limiting for users of this AS is on by default (excludes sender)
+    rate_limited = True
+    if isinstance(as_info.get("rate_limited"), bool):
+        rate_limited = as_info.get("rate_limited")
+
+    # namespace checks
+    if not isinstance(as_info.get("namespaces"), dict):
+        raise KeyError("Requires 'namespaces' object.")
+    for ns in ApplicationService.NS_LIST:
+        # specific namespaces are optional
+        if ns in as_info["namespaces"]:
+            # expect a list of dicts with exclusive and regex keys
+            for regex_obj in as_info["namespaces"][ns]:
+                if not isinstance(regex_obj, dict):
+                    raise ValueError(
+                        "Expected namespace entry in %s to be an object,"
+                        " but got %s", ns, regex_obj
+                    )
+                if not isinstance(regex_obj.get("regex"), basestring):
+                    raise ValueError(
+                        "Missing/bad type 'regex' key in %s", regex_obj
+                    )
+                if not isinstance(regex_obj.get("exclusive"), bool):
+                    raise ValueError(
+                        "Missing/bad type 'exclusive' key in %s", regex_obj
+                    )
+    # protocols check
+    protocols = as_info.get("protocols")
+    if protocols:
+        # Because strings are lists in python
+        if isinstance(protocols, str) or not isinstance(protocols, list):
+            raise KeyError("Optional 'protocols' must be a list if present.")
+        for p in protocols:
+            if not isinstance(p, str):
+                raise KeyError("Bad value for 'protocols' item")
+
+    if as_info["url"] is None:
+        logger.info(
+            "(%s) Explicitly empty 'url' provided. This application service"
+            " will not receive events or queries.",
+            config_filename,
+        )
+    return ApplicationService(
+        token=as_info["as_token"],
+        url=as_info["url"],
+        namespaces=as_info["namespaces"],
+        hs_token=as_info["hs_token"],
+        sender=user_id,
+        id=as_info["id"],
+        protocols=protocols,
+        rate_limited=rate_limited
+    )

synapse/config/captcha.py

@@ -1,4 +1,4 @@
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,42 +17,32 @@ from ._base import Config
 
 class CaptchaConfig(Config):
 
-    def __init__(self, args):
-        super(CaptchaConfig, self).__init__(args)
-        self.recaptcha_private_key = args.recaptcha_private_key
-        self.recaptcha_public_key = args.recaptcha_public_key
-        self.enable_registration_captcha = args.enable_registration_captcha
+    def read_config(self, config):
+        self.recaptcha_private_key = config["recaptcha_private_key"]
+        self.recaptcha_public_key = config["recaptcha_public_key"]
+        self.enable_registration_captcha = config["enable_registration_captcha"]
+        self.captcha_bypass_secret = config.get("captcha_bypass_secret")
+        self.recaptcha_siteverify_api = config["recaptcha_siteverify_api"]
 
-        # XXX: This is used for more than just captcha
-        self.captcha_ip_origin_is_x_forwarded = (
-            args.captcha_ip_origin_is_x_forwarded
-        )
-        self.captcha_bypass_secret = args.captcha_bypass_secret
+    def default_config(self, **kwargs):
+        return """\
+        ## Captcha ##
+        # See docs/CAPTCHA_SETUP for full details of configuring this.
 
-    @classmethod
-    def add_arguments(cls, parser):
-        super(CaptchaConfig, cls).add_arguments(parser)
-        group = parser.add_argument_group("recaptcha")
-        group.add_argument(
-            "--recaptcha-public-key", type=str, default="YOUR_PUBLIC_KEY",
-            help="This Home Server's ReCAPTCHA public key."
-        )
-        group.add_argument(
-            "--recaptcha-private-key", type=str, default="YOUR_PRIVATE_KEY",
-            help="This Home Server's ReCAPTCHA private key."
-        )
-        group.add_argument(
-            "--enable-registration-captcha", type=bool, default=False,
-            help="Enables ReCaptcha checks when registering, preventing signup"
-            + " unless a captcha is answered. Requires a valid ReCaptcha "
-            + "public/private key."
-        )
-        group.add_argument(
-            "--captcha_ip_origin_is_x_forwarded", type=bool, default=False,
-            help="When checking captchas, use the X-Forwarded-For (XFF) header"
-            + " as the client IP and not the actual client IP."
-        )
-        group.add_argument(
-            "--captcha_bypass_secret", type=str,
-            help="A secret key used to bypass the captcha test entirely."
-        )
+        # This Home Server's ReCAPTCHA public key.
+        recaptcha_public_key: "YOUR_PUBLIC_KEY"
+
+        # This Home Server's ReCAPTCHA private key.
+        recaptcha_private_key: "YOUR_PRIVATE_KEY"
+
+        # Enables ReCaptcha checks when registering, preventing signup
+        # unless a captcha is answered. Requires a valid ReCaptcha
+        # public/private key.
+        enable_registration_captcha: False
+
+        # A secret key used to bypass the captcha test entirely.
+        #captcha_bypass_secret: "YOUR_SECRET_HERE"
+
+        # The API endpoint to use for verifying m.login.recaptcha responses.
+        recaptcha_siteverify_api: "https://www.google.com/recaptcha/api/siteverify"
+        """

synapse/config/cas.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 OpenMarket 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.
+
+from ._base import Config
+
+
+class CasConfig(Config):
+    """Cas Configuration
+
+    cas_server_url: URL of CAS server
+    """
+
+    def read_config(self, config):
+        cas_config = config.get("cas_config", None)
+        if cas_config:
+            self.cas_enabled = cas_config.get("enabled", True)
+            self.cas_server_url = cas_config["server_url"]
+            self.cas_service_url = cas_config["service_url"]
+            self.cas_required_attributes = cas_config.get("required_attributes", {})
+        else:
+            self.cas_enabled = False
+            self.cas_server_url = None
+            self.cas_service_url = None
+            self.cas_required_attributes = {}
+
+    def default_config(self, config_dir_path, server_name, **kwargs):
+        return """
+        # Enable CAS for registration and login.
+        #cas_config:
+        #   enabled: true
+        #   server_url: "https://cas-server.com"
+        #   service_url: "https://homesever.domain.com:8448"
+        #   #required_attributes:
+        #   #    name: value
+        """

synapse/config/database.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014, 2015 OpenMarket Ltd
+# Copyright 2014-2016 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,28 +14,21 @@
 # limitations under the License.
 
 from ._base import Config
-import os
-import yaml
 
 
 class DatabaseConfig(Config):
-    def __init__(self, args):
-        super(DatabaseConfig, self).__init__(args)
-        if args.database_path == ":memory:":
-            self.database_path = ":memory:"
-        else:
-            self.database_path = self.abspath(args.database_path)
-        self.event_cache_size = self.parse_size(args.event_cache_size)
 
-        if args.database_config:
-            with open(args.database_config) as f:
-                self.database_config = yaml.safe_load(f)
-        else:
+    def read_config(self, config):
+        self.event_cache_size = self.parse_size(
+            config.get("event_cache_size", "10K")
+        )
+
+        self.database_config = config.get("database")
+
+        if self.database_config is None:
             self.database_config = {
                 "name": "sqlite3",
-                "args": {
-                    "database": self.database_path,
-                },
+                "args": {},
             }
 
         name = self.database_config.get("name", None)
@@ -50,24 +43,37 @@ class DatabaseConfig(Config):
         else:
             raise RuntimeError("Unsupported database type '%s'" % (name,))
 
-    @classmethod
-    def add_arguments(cls, parser):
-        super(DatabaseConfig, cls).add_arguments(parser)
+        self.set_databasepath(config.get("database_path"))
+
+    def default_config(self, **kwargs):
+        database_path = self.abspath("homeserver.db")
+        return """\
+        # Database configuration
+        database:
+          # The database engine name
+          name: "sqlite3"
+          # Arguments to pass to the engine
+          args:
+            # Path to the database
+            database: "%(database_path)s"
+
+        # Number of events to cache in memory.
+        event_cache_size: "10K"
+        """ % locals()
+
+    def read_arguments(self, args):
+        self.set_databasepath(args.database_path)
+
+    def set_databasepath(self, database_path):
+        if database_path != ":memory:":
+            database_path = self.abspath(database_path)
+        if self.database_config.get("name", None) == "sqlite3":
+            if database_path is not None:
+                self.database_config["args"]["database"] = database_path
+
+    def add_arguments(self, parser):
         db_group = parser.add_argument_group("database")
         db_group.add_argument(
-            "-d", "--database-path", default="homeserver.db",
-            metavar="SQLITE_DATABASE_PATH", help="The database name."
+            "-d", "--database-path", metavar="SQLITE_DATABASE_PATH",
+            help="The path to a sqlite database to use."
         )
-        db_group.add_argument(
-            "--event-cache-size", default="100K",
-            help="Number of events to cache in memory."
-        )
-        db_group.add_argument(
-            "--database-config", default=None,
-            help="Location of the database configuration file."
-        )
-
-    @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)

synapse/config/emailconfig.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 OpenMarket 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.
+
+# This file can't be called email.py because if it is, we cannot:
+import email.utils
+
+from ._base import Config
+
+
+class EmailConfig(Config):
+    def read_config(self, config):
+        self.email_enable_notifs = False
+
+        email_config = config.get("email", {})
+        self.email_enable_notifs = email_config.get("enable_notifs", False)
+
+        if self.email_enable_notifs:
+            # make sure we can import the required deps
+            import jinja2
+            import bleach
+            # prevent unused warnings
+            jinja2
+            bleach
+
+            required = [
+                "smtp_host",
+                "smtp_port",
+                "notif_from",
+                "template_dir",
+                "notif_template_html",
+                "notif_template_text",
+            ]
+
+            missing = []
+            for k in required:
+                if k not in email_config:
+                    missing.append(k)
+
+            if (len(missing) > 0):
+                raise RuntimeError(
+                    "email.enable_notifs is True but required keys are missing: %s" %
+                    (", ".join(["email." + k for k in missing]),)
+                )
+
+            if config.get("public_baseurl") is None:
+                raise RuntimeError(
+                    "email.enable_notifs is True but no public_baseurl is set"
+                )
+
+            self.email_smtp_host = email_config["smtp_host"]
+            self.email_smtp_port = email_config["smtp_port"]
+            self.email_notif_from = email_config["notif_from"]
+            self.email_template_dir = email_config["template_dir"]
+            self.email_notif_template_html = email_config["notif_template_html"]
+            self.email_notif_template_text = email_config["notif_template_text"]
+            self.email_notif_for_new_users = email_config.get(
+                "notif_for_new_users", True
+            )
+            if "app_name" in email_config:
+                self.email_app_name = email_config["app_name"]
+            else:
+                self.email_app_name = "Matrix"
+
+            # make sure it's valid
+            parsed = email.utils.parseaddr(self.email_notif_from)
+            if parsed[1] == '':
+                raise RuntimeError("Invalid notif_from address")
+        else:
+            self.email_enable_notifs = False
+            # Not much point setting defaults for the rest: it would be an
+            # error for them to be used.
+
+    def default_config(self, config_dir_path, server_name, **kwargs):
+        return """
+        # Enable sending emails for notification events
+        #email:
+        #   enable_notifs: false
+        #   smtp_host: "localhost"
+        #   smtp_port: 25
+        #   notif_from: "Your Friendly %(app)s Home Server <noreply@example.com>"
+        #   app_name: Matrix
+        #   template_dir: res/templates
+        #   notif_template_html: notif_mail.html
+        #   notif_template_text: notif_mail.txt
+        #   notif_for_new_users: True
+        """