implement /room_keys/version too (untested)

synapse/api/errors.py

@@ -56,6 +56,7 @@ class Codes(object):
     CONSENT_NOT_GIVEN = "M_CONSENT_NOT_GIVEN"
     CANNOT_LEAVE_SERVER_NOTICE_ROOM = "M_CANNOT_LEAVE_SERVER_NOTICE_ROOM"
     MAU_LIMIT_EXCEEDED = "M_MAU_LIMIT_EXCEEDED"
+    WRONG_ROOM_KEYS_VERSION = "M_WRONG_ROOM_KEYS_VERSION"
 
 
 class CodeMessageException(RuntimeError):
@@ -285,6 +286,30 @@ class LimitExceededError(SynapseError):
         )
 
 
+class RoomKeysVersionError(SynapseError):
+    """A client has tried to upload to a non-current version of the room_keys store
+    """
+    def __init__(self, code=403, msg="Wrong room_keys version", current_version=None,
+                 errcode=Codes.WRONG_ROOM_KEYS_VERSION):
+        super(RoomKeysVersionError, self).__init__(code, msg, errcode)
+        self.current_version = current_version
+
+    def error_dict(self):
+        return cs_error(
+            self.msg,
+            self.errcode,
+            current_version=self.current_version,
+        )
+
+
+def cs_exception(exception):
+    if isinstance(exception, CodeMessageException):
+        return exception.error_dict()
+    else:
+        logger.error("Unknown exception type: %s", type(exception))
+        return {}
+
+
 def cs_error(msg, code=Codes.UNKNOWN, **kwargs):
     """ Utility method for constructing an error response for client-server
     interactions.

synapse/handlers/e2e_room_keys.py

@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.api.errors import StoreError
+from synapse.api.errors import StoreError, SynapseError, RoomKeysVersionError
 from synapse.util.async import Linearizer
 
 logger = logging.getLogger(__name__)
@@ -30,10 +30,13 @@ class E2eRoomKeysHandler(object):
 
     @defer.inlineCallbacks
     def get_room_keys(self, user_id, version, room_id, session_id):
-        results = yield self.store.get_e2e_room_keys(
+        # we deliberately take the lock to get keys so that changing the version
-            user_id, version, room_id, session_id
+        # works atomically
-        )
+        with (yield self._upload_linearizer.queue(user_id)):
-        defer.returnValue(results)
+            results = yield self.store.get_e2e_room_keys(
+                user_id, version, room_id, session_id
+            )
+            defer.returnValue(results)
 
     @defer.inlineCallbacks
     def delete_room_keys(self, user_id, version, room_id, session_id):
@@ -44,6 +47,16 @@ class E2eRoomKeysHandler(object):
 
         # TODO: Validate the JSON to make sure it has the right keys.
 
+        # Check that the version we're trying to upload is the current version
+        try:
+            version_info = yield self.get_version_info(user_id, version)
+        except StoreError as e:
+            if e.code == 404:
+                raise SynapseError(404, "Version '%d' not found" % (version,))
+
+        if version_info.version != version:
+            raise RoomKeysVersionError(current_version=version_info.version)
+
         # XXX: perhaps we should use a finer grained lock here?
         with (yield self._upload_linearizer.queue(user_id)):
 
@@ -91,3 +104,27 @@ class E2eRoomKeysHandler(object):
             yield self.store.set_e2e_room_key(
                 user_id, version, room_id, session_id, room_key
             )
+
+    @defer.inlineCallbacks
+    def create_version(self, user_id, version, version_info):
+
+        # TODO: Validate the JSON to make sure it has the right keys.
+
+        # lock everyone out until we've switched version
+        with (yield self._upload_linearizer.queue(user_id)):
+            yield self.store.create_version(
+                user_id, version, version_info
+            )
+
+    @defer.inlineCallbacks
+    def get_version_info(self, user_id, version):
+        with (yield self._upload_linearizer.queue(user_id)):
+            results = yield self.store.get_e2e_room_key_version(
+                user_id, version
+            )
+            defer.returnValue(results)
+
+    @defer.inlineCallbacks
+    def delete_version(self, user_id, version):
+        with (yield self._upload_linearizer.queue(user_id)):
+            yield self.store.delete_e2e_room_key_version(user_id, version)

synapse/rest/client/v2_alpha/room_keys.py

@@ -221,5 +221,52 @@ class RoomKeysServlet(RestServlet):
         defer.returnValue((200, {}))
 
 
+class RoomKeysVersionServlet(RestServlet):
+    PATTERNS = client_v2_patterns(
+        "/room_keys/version(/(?P<version>[^/]+))?$"
+    )
+
+    def __init__(self, hs):
+        """
+        Args:
+            hs (synapse.server.HomeServer): server
+        """
+        super(RoomKeysVersionServlet, self).__init__()
+        self.auth = hs.get_auth()
+        self.e2e_room_keys_handler = hs.get_e2e_room_keys_handler()
+
+    @defer.inlineCallbacks
+    def on_POST(self, request, version):
+        requester = yield self.auth.get_user_by_req(request, allow_guest=False)
+        user_id = requester.user.to_string()
+        info = parse_json_object_from_request(request)
+
+        new_version = yield self.e2e_room_keys_handler.create_version(
+            user_id, version, info
+        )
+        defer.returnValue((200, {"version": new_version}))
+
+    @defer.inlineCallbacks
+    def on_GET(self, request, version):
+        requester = yield self.auth.get_user_by_req(request, allow_guest=False)
+        user_id = requester.user.to_string()
+
+        info = yield self.e2e_room_keys_handler.get_version_info(
+            user_id, version
+        )
+        defer.returnValue((200, info))
+
+    @defer.inlineCallbacks
+    def on_DELETE(self, request, version):
+        requester = yield self.auth.get_user_by_req(request, allow_guest=False)
+        user_id = requester.user.to_string()
+
+        yield self.e2e_room_keys_handler.delete_version(
+            user_id, version
+        )
+        defer.returnValue((200, {}))
+
+
 def register_servlets(hs, http_server):
     RoomKeysServlet(hs).register(http_server)
+    RoomKeysVersionServlet(hs).register(http_server)

synapse/storage/e2e_room_keys.py

@@ -170,3 +170,59 @@ class EndToEndRoomKeyStore(SQLBaseStore):
             keyvalues=keyvalues,
             desc="delete_e2e_room_keys",
         )
+
+    @defer.inlineCallbacks
+    def get_e2e_room_key_version(self, user_id, version):
+
+        row = yield self._simple_select_one(
+            table="e2e_room_key_versions",
+            keyvalues={
+                "user_id": user_id,
+                "version": version,
+            },
+            retcols=(
+                "user_id",
+                "version",
+                "algorithm",
+                "auth_data",
+            ),
+            desc="get_e2e_room_key_version_info",
+        )
+
+        defer.returnValue(row)
+
+    def create_e2e_room_key_version(self, user_id, version, info):
+
+        def _create_e2e_room_key_version_txn(txn):
+
+            self._simple_insert_txn(
+                txn,
+                table="e2e_room_key_versions",
+                values={
+                    "user_id": user_id,
+                    "version": version,
+                    "algorithm": info["algorithm"],
+                    "auth_data": info["auth_data"],
+                },
+                lock=False,
+            )
+
+            return True
+
+        return self.runInteraction(
+            "create_e2e_room_key_version_txn", _create_e2e_room_key_version_txn
+        )
+
+    @defer.inlineCallbacks
+    def delete_e2e_room_key_version(self, user_id, version):
+
+        keyvalues = {
+            "user_id": user_id,
+            "version": version,
+        }
+
+        yield self._simple_delete(
+            table="e2e_room_key_versions",
+            keyvalues=keyvalues,
+            desc="delete_e2e_room_key_version",
+        )

synapse/storage/schema/delta/46/e2e_room_keys.sql

@@ -34,7 +34,7 @@ CREATE TABLE e2e_room_key_versions (
     user_id TEXT NOT NULL,
     version INT NOT NULL,
     algorithm TEXT NOT NULL,
-    dummy_session_data TEXT NOT NULL
+    auth_data TEXT NOT NULL
 );
 
 CREATE UNIQUE INDEX e2e_room_key_user_idx ON e2e_room_keys(user_id);