make it work and fix pep8

7 files changed, 134 insertions, 82 deletions

diff --git a/synapse/handlers/e2e_room_keys.py b/synapse/handlers/e2e_room_keys.py
index 15e3beb5ed..93f4ad5194 100644
--- a/synapse/handlers/e2e_room_keys.py
+++ b/synapse/handlers/e2e_room_keys.py
@@ -13,15 +13,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import ujson as json
 import logging
 
-from canonicaljson import encode_canonical_json
 from twisted.internet import defer
 
-from synapse.api.errors import SynapseError, CodeMessageException
+from synapse.api.errors import StoreError
 from synapse.util.async import Linearizer
-from synapse.util.retryutils import NotRetryingDestination
 
 logger = logging.getLogger(__name__)
 
@@ -29,11 +26,13 @@ logger = logging.getLogger(__name__)
 class E2eRoomKeysHandler(object):
     def __init__(self, hs):
         self.store = hs.get_datastore()
-        self._upload_linearizer = async.Linearizer("upload_room_keys_lock")
+        self._upload_linearizer = Linearizer("upload_room_keys_lock")
 
     @defer.inlineCallbacks
     def get_room_keys(self, user_id, version, room_id, session_id):
-        results = yield self.store.get_e2e_room_keys(user_id, version, room_id, session_id)
+        results = yield self.store.get_e2e_room_keys(
+            user_id, version, room_id, session_id
+        )
         defer.returnValue(results)
 
     @defer.inlineCallbacks
@@ -46,31 +45,49 @@ class E2eRoomKeysHandler(object):
         # TODO: Validate the JSON to make sure it has the right keys.
 
         # XXX: perhaps we should use a finer grained lock here?
-        with (yield self._upload_linearizer.queue(user_id):
+        with (yield self._upload_linearizer.queue(user_id)):
 
             # go through the room_keys
             for room_id in room_keys['rooms']:
                 for session_id in room_keys['rooms'][room_id]['sessions']:
                     room_key = room_keys['rooms'][room_id]['sessions'][session_id]
 
-                    # get the room_key for this particular row
-                    current_room_key = yield self.store.get_e2e_room_key(
-                        user_id, version, room_id, session_id
+                    yield self._upload_room_key(
+                        user_id, version, room_id, session_id, room_key
                     )
 
-                    # check whether we merge or not. spelling it out with if/elifs rather than
-                    # lots of booleans for legibility.
-                    replace = False
-                    if current_room_key:
-                        if room_key['is_verified'] and not current_room_key['is_verified']:
-                            replace = True
-                        elif room_key['first_message_index'] < current_room_key['first_message_index']:
-                            replace = True
-                        elif room_key['forwarded_count'] < room_key['forwarded_count']:
-                            replace = True
-
-                    # if so, we set the new room_key
-                    if replace:
-                        yield self.store.set_e2e_room_key(
-                            user_id, version, room_id, session_id, room_key
-                        )
+    @defer.inlineCallbacks
+    def _upload_room_key(self, user_id, version, room_id, session_id, room_key):
+        # get the room_key for this particular row
+        current_room_key = None
+        try:
+            current_room_key = yield self.store.get_e2e_room_key(
+                user_id, version, room_id, session_id
+            )
+        except StoreError as e:
+            if e.code == 404:
+                pass
+            else:
+                raise
+
+        # check whether we merge or not. spelling it out with if/elifs rather
+        # than lots of booleans for legibility.
+        upsert = True
+        if current_room_key:
+            if room_key['is_verified'] and not current_room_key['is_verified']:
+                pass
+            elif (
+                room_key['first_message_index'] <
+                current_room_key['first_message_index']
+            ):
+                pass
+            elif room_key['forwarded_count'] < room_key['forwarded_count']:
+                pass
+            else:
+                upsert = False
+
+        # if so, we set the new room_key
+        if upsert:
+            yield self.store.set_e2e_room_key(
+                user_id, version, room_id, session_id, room_key
+            )
diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py
index 3418f06fd6..4856822a5d 100644
--- a/synapse/rest/__init__.py
+++ b/synapse/rest/__init__.py
@@ -46,6 +46,7 @@ from synapse.rest.client.v2_alpha import (
     receipts,
     register,
     report_event,
+    room_keys,
     sendtodevice,
     sync,
     tags,
@@ -102,6 +103,7 @@ class ClientRestResource(JsonResource):
         auth.register_servlets(hs, client_resource)
         receipts.register_servlets(hs, client_resource)
         read_marker.register_servlets(hs, client_resource)
+        room_keys.register_servlets(hs, client_resource)
         keys.register_servlets(hs, client_resource)
         tokenrefresh.register_servlets(hs, client_resource)
         tags.register_servlets(hs, client_resource)
diff --git a/synapse/rest/client/v2_alpha/room_keys.py b/synapse/rest/client/v2_alpha/room_keys.py
index 7291018a48..010aed98f9 100644
--- a/synapse/rest/client/v2_alpha/room_keys.py
+++ b/synapse/rest/client/v2_alpha/room_keys.py
@@ -17,26 +17,25 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.api.errors import SynapseError
 from synapse.http.servlet import (
-    RestServlet, parse_json_object_from_request, parse_integer
+    RestServlet, parse_json_object_from_request
 )
-from synapse.http.servlet import parse_string
-from synapse.types import StreamToken
 from ._base import client_v2_patterns
 
 logger = logging.getLogger(__name__)
 
 
 class RoomKeysServlet(RestServlet):
-    PATTERNS = client_v2_patterns("/room_keys/keys(/(?P<room_id>[^/]+))?(/(?P<session_id>[^/]+))?$")
+    PATTERNS = client_v2_patterns(
+        "/room_keys/keys(/(?P<room_id>[^/]+))?(/(?P<session_id>[^/]+))?$"
+    )
 
     def __init__(self, hs):
         """
         Args:
             hs (synapse.server.HomeServer): server
         """
-        super(RoomKeysUploadServlet, self).__init__()
+        super(RoomKeysServlet, self).__init__()
         self.auth = hs.get_auth()
         self.e2e_room_keys_handler = hs.get_e2e_room_keys_handler()
 
@@ -45,24 +44,32 @@ class RoomKeysServlet(RestServlet):
         requester = yield self.auth.get_user_by_req(request, allow_guest=False)
         user_id = requester.user.to_string()
         body = parse_json_object_from_request(request)
-        version = request.args.get("version", None)
+        version = request.args.get("version")[0]
 
         if session_id:
-            body = { "sessions": { session_id : body } }
+            body = {
+                "sessions": {
+                    session_id: body
+                }
+            }
 
         if room_id:
-            body = { "rooms": { room_id : body } }
+            body = {
+                "rooms": {
+                    room_id: body
+                }
+            }
 
-        result = yield self.e2e_room_keys_handler.upload_room_keys(
+        yield self.e2e_room_keys_handler.upload_room_keys(
             user_id, version, body
         )
-        defer.returnValue((200, result))
+        defer.returnValue((200, {}))
 
     @defer.inlineCallbacks
     def on_GET(self, request, room_id, session_id):
         requester = yield self.auth.get_user_by_req(request, allow_guest=False)
         user_id = requester.user.to_string()
-        version = request.args.get("version", None)
+        version = request.args.get("version")[0]
 
         room_keys = yield self.e2e_room_keys_handler.get_room_keys(
             user_id, version, room_id, session_id
@@ -73,7 +80,7 @@ class RoomKeysServlet(RestServlet):
     def on_DELETE(self, request, room_id, session_id):
         requester = yield self.auth.get_user_by_req(request, allow_guest=False)
         user_id = requester.user.to_string()
-        version = request.args.get("version", None)
+        version = request.args.get("version")[0]
 
         yield self.e2e_room_keys_handler.delete_room_keys(
             user_id, version, room_id, session_id
diff --git a/synapse/server.py b/synapse/server.py
index 140be9ebe8..706cb1361f 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -49,6 +49,7 @@ from synapse.handlers.deactivate_account import DeactivateAccountHandler
 from synapse.handlers.device import DeviceHandler
 from synapse.handlers.devicemessage import DeviceMessageHandler
 from synapse.handlers.e2e_keys import E2eKeysHandler
+from synapse.handlers.e2e_room_keys import E2eRoomKeysHandler
 from synapse.handlers.events import EventHandler, EventStreamHandler
 from synapse.handlers.groups_local import GroupsLocalHandler
 from synapse.handlers.initial_sync import InitialSyncHandler
@@ -127,6 +128,7 @@ class HomeServer(object):
         'auth_handler',
         'device_handler',
         'e2e_keys_handler',
+        'e2e_room_keys_handler',
         'event_handler',
         'event_stream_handler',
         'initial_sync_handler',
@@ -288,6 +290,9 @@ class HomeServer(object):
     def build_e2e_keys_handler(self):
         return E2eKeysHandler(self)
 
+    def build_e2e_room_keys_handler(self):
+        return E2eRoomKeysHandler(self)
+
     def build_application_service_api(self):
         return ApplicationServiceApi(self)
 
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 134e4a80f1..69cb28268a 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -30,6 +30,7 @@ from .appservice import ApplicationServiceStore, ApplicationServiceTransactionSt
 from .client_ips import ClientIpStore
 from .deviceinbox import DeviceInboxStore
 from .directory import DirectoryStore
+from .e2e_room_keys import EndToEndRoomKeyStore
 from .end_to_end_keys import EndToEndKeyStore
 from .engines import PostgresEngine
 from .event_federation import EventFederationStore
@@ -76,6 +77,7 @@ class DataStore(RoomMemberStore, RoomStore,
                 ApplicationServiceTransactionStore,
                 ReceiptsStore,
                 EndToEndKeyStore,
+                EndToEndRoomKeyStore,
                 SearchStore,
                 TagsStore,
                 AccountDataStore,
diff --git a/synapse/storage/e2e_room_keys.py b/synapse/storage/e2e_room_keys.py
index 903dc083f8..5982710bd5 100644
--- a/synapse/storage/e2e_room_keys.py
+++ b/synapse/storage/e2e_room_keys.py
@@ -15,11 +15,6 @@
 
 from twisted.internet import defer
 
-from synapse.util.caches.descriptors import cached
-
-from canonicaljson import encode_canonical_json
-import ujson as json
-
 from ._base import SQLBaseStore
 
 
@@ -45,29 +40,27 @@ class EndToEndRoomKeyStore(SQLBaseStore):
             desc="get_e2e_room_key",
         )
 
-        defer.returnValue(row);
+        defer.returnValue(row)
 
     def set_e2e_room_key(self, user_id, version, room_id, session_id, room_key):
 
         def _set_e2e_room_key_txn(txn):
 
-            self._simple_upsert(
+            self._simple_upsert_txn(
                 txn,
                 table="e2e_room_keys",
                 keyvalues={
                     "user_id": user_id,
                     "room_id": room_id,
-                    "session_id": session_id,   
-                }
-                values=[
-                    {
-                        "version": version,
-                        "first_message_index": room_key['first_message_index'],
-                        "forwarded_count": room_key['forwarded_count'],
-                        "is_verified": room_key['is_verified'],
-                        "session_data": room_key['session_data'],
-                    }
-                ],
+                    "session_id": session_id,
+                },
+                values={
+                    "version": version,
+                    "first_message_index": room_key['first_message_index'],
+                    "forwarded_count": room_key['forwarded_count'],
+                    "is_verified": room_key['is_verified'],
+                    "session_data": room_key['session_data'],
+                },
                 lock=False,
             )
 
@@ -77,7 +70,6 @@ class EndToEndRoomKeyStore(SQLBaseStore):
             "set_e2e_room_key", _set_e2e_room_key_txn
         )
 
-
     # XXX: this isn't currently used and isn't tested anywhere
     # it could be used in future for bulk-uploading new versions of room_keys
     # for a user or something though.
@@ -85,23 +77,27 @@ class EndToEndRoomKeyStore(SQLBaseStore):
 
         def _set_e2e_room_keys_txn(txn):
 
+            values = []
+            for room_id in room_keys['rooms']:
+                for session_id in room_keys['rooms'][room_id]['sessions']:
+                    session = room_keys['rooms'][room_id]['sessions'][session_id]
+                    values.append(
+                        {
+                            "user_id": user_id,
+                            "room_id": room_id,
+                            "session_id": session_id,
+                            "version": version,
+                            "first_message_index": session['first_message_index'],
+                            "forwarded_count": session['forwarded_count'],
+                            "is_verified": session['is_verified'],
+                            "session_data": session['session_data'],
+                        }
+                    )
+
             self._simple_insert_many_txn(
                 txn,
                 table="e2e_room_keys",
-                values=[
-                    {
-                        "user_id": user_id,
-                        "room_id": room_id,
-                        "session_id": session_id,
-                        "version": version,
-                        "first_message_index": room_keys['rooms'][room_id]['sessions'][session_id]['first_message_index'],
-                        "forwarded_count": room_keys['rooms'][room_id]['sessions'][session_id]['forwarded_count'],
-                        "is_verified": room_keys['rooms'][room_id]['sessions'][session_id]['is_verified'],
-                        "session_data": room_keys['rooms'][room_id]['sessions'][session_id]['session_data'],
-                    }
-                    for session_id in room_keys['rooms'][room_id]['sessions']
-                    for room_id in room_keys['rooms']
-                ]
+                values=values
             )
 
             return True
@@ -113,17 +109,22 @@ class EndToEndRoomKeyStore(SQLBaseStore):
     @defer.inlineCallbacks
     def get_e2e_room_keys(self, user_id, version, room_id, session_id):
 
-        keyvalues={
+        keyvalues = {
             "user_id": user_id,
             "version": version,
         }
-        if room_id: keyvalues['room_id'] = room_id
-        if session_id: keyvalues['session_id'] = session_id
+        if room_id:
+            keyvalues['room_id'] = room_id
+        if session_id:
+            keyvalues['session_id'] = session_id
 
         rows = yield self._simple_select_list(
             table="e2e_room_keys",
             keyvalues=keyvalues,
             retcols=(
+                "user_id",
+                "room_id",
+                "session_id",
                 "first_message_index",
                 "forwarded_count",
                 "is_verified",
@@ -132,19 +133,37 @@ class EndToEndRoomKeyStore(SQLBaseStore):
             desc="get_e2e_room_keys",
         )
 
-        sessions = {}
-        sessions['rooms'][roomId]['sessions'][session_id] = row for row in rows;
-        defer.returnValue(sessions);
+        # perlesque autovivification from https://stackoverflow.com/a/19829714/6764493
+        class AutoVivification(dict):
+            def __getitem__(self, item):
+                try:
+                    return dict.__getitem__(self, item)
+                except KeyError:
+                    value = self[item] = type(self)()
+                    return value
+
+        sessions = AutoVivification()
+        for row in rows:
+            sessions['rooms'][row['room_id']]['sessions'][row['session_id']] = {
+                "first_message_index": row["first_message_index"],
+                "forwarded_count": row["forwarded_count"],
+                "is_verified": row["is_verified"],
+                "session_data": row["session_data"],
+            }
+
+        defer.returnValue(sessions)
 
     @defer.inlineCallbacks
     def delete_e2e_room_keys(self, user_id, version, room_id, session_id):
 
-        keyvalues={
+        keyvalues = {
             "user_id": user_id,
             "version": version,
         }
-        if room_id: keyvalues['room_id'] = room_id
-        if session_id: keyvalues['session_id'] = session_id
+        if room_id:
+            keyvalues['room_id'] = room_id
+        if session_id:
+            keyvalues['session_id'] = session_id
 
         yield self._simple_delete(
             table="e2e_room_keys",
diff --git a/synapse/storage/schema/delta/46/e2e_room_keys.sql b/synapse/storage/schema/delta/46/e2e_room_keys.sql
index 51b826e8b3..6b344c5ad7 100644
--- a/synapse/storage/schema/delta/46/e2e_room_keys.sql
+++ b/synapse/storage/schema/delta/46/e2e_room_keys.sql
@@ -29,7 +29,7 @@ CREATE UNIQUE INDEX e2e_room_keys_user_idx ON e2e_room_keys(user_id);
 CREATE UNIQUE INDEX e2e_room_keys_room_idx ON e2e_room_keys(room_id);
 CREATE UNIQUE INDEX e2e_room_keys_session_idx ON e2e_room_keys(session_id);
 
--- the versioning metadata about versions of users' encrypted e2e session backups
+-- the metadata for each generation of encrypted e2e session backups
 CREATE TABLE e2e_room_key_versions (
     user_id TEXT NOT NULL,
     version INT NOT NULL,