mirrors/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2024-12-22 04:34:28 +03:00
Code Issues Projects Releases Packages Wiki Activity

Trace e2e

Browse source
This commit is contained in:
Jorik Schellekens 2019-06-28 10:34:53 +01:00
parent 823c34a940
commit 565544b603
5 changed files with 115 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
synapse/handlers/e2e_keys.py
View file

@ -15,6 +15,7 @@
# limitations under the License.
import logging
from synapse.util.tracerutils import TracerUtil, trace_defered_function
from six import iteritems
@ -45,6 +46,7 @@ class E2eKeysHandler(object):
"client_keys", self.on_federation_query_client_keys
)
@trace_defered_function
@defer.inlineCallbacks
def query_devices(self, query_body, timeout):
""" Handle a device key query from a client
@ -65,6 +67,7 @@ class E2eKeysHandler(object):
}
}
"""
device_keys_query = query_body.get("device_keys", {})
# separate users by domain.
@ -79,6 +82,9 @@ class E2eKeysHandler(object):
else:
remote_queries[user_id] = device_ids
TracerUtil.set_tag("local_key_query", local_query)
TracerUtil.set_tag("remote_key_query", remote_queries)
# First get local devices.
failures = {}
results = {}
@ -119,9 +125,12 @@ class E2eKeysHandler(object):
r[user_id] = remote_queries[user_id]
# Now fetch any devices that we don't have in our cache
@trace_defered_function
@defer.inlineCallbacks
def do_remote_query(destination):
destination_query = remote_queries_not_in_cache[destination]
TracerUtil.set_tag("key_query", destination_query)
try:
remote_result = yield self.federation.query_client_keys(
destination, {"device_keys": destination_query}, timeout=timeout
@ -132,7 +141,10 @@ class E2eKeysHandler(object):
results[user_id] = keys
except Exception as e:
failures[destination] = _exception_to_failure(e)
failure = _exception_to_failure(e)
failures[destination] = failure
TracerUtil.set_tag("error", True)
TracerUtil.set_tag("reason", failure)
yield make_deferred_yieldable(
defer.gatherResults(
@ -146,6 +158,7 @@ class E2eKeysHandler(object):
return {"device_keys": results, "failures": failures}
@trace_defered_function
@defer.inlineCallbacks
def query_local_devices(self, query):
"""Get E2E device keys for local users
@ -158,6 +171,7 @@ class E2eKeysHandler(object):
defer.Deferred: (resolves to dict[string, dict[string, dict]]):
map from user_id -> device_id -> device details
"""
TracerUtil.set_tag("local_query", query)
local_query = []
result_dict = {}
@ -165,6 +179,14 @@ class E2eKeysHandler(object):
# we use UserID.from_string to catch invalid user ids
if not self.is_mine(UserID.from_string(user_id)):
logger.warning("Request for keys for non-local user %s", user_id)
TracerUtil.log_kv(
{
"message": "Requested a local key for a user which"
+ " was not local to the homeserver",
"user_id": user_id,
}
)
TracerUtil.set_tag("error", True)
raise SynapseError(400, "Not a user here")
if not device_ids:
@ -189,6 +211,7 @@ class E2eKeysHandler(object):
r["unsigned"]["device_display_name"] = display_name
result_dict[user_id][device_id] = r
TracerUtil.log_kv(results)
return result_dict
@defer.inlineCallbacks
@ -199,6 +222,7 @@ class E2eKeysHandler(object):
res = yield self.query_local_devices(device_keys_query)
return {"device_keys": res}
@trace_defered_function
@defer.inlineCallbacks
def claim_one_time_keys(self, query, timeout):
local_query = []
@ -213,6 +237,9 @@ class E2eKeysHandler(object):
domain = get_domain_from_id(user_id)
remote_queries.setdefault(domain, {})[user_id] = device_keys
TracerUtil.set_tag("local_key_query", local_query)
TracerUtil.set_tag("remote_key_query", remote_queries)
results = yield self.store.claim_e2e_one_time_keys(local_query)
json_result = {}
@ -224,8 +251,10 @@ class E2eKeysHandler(object):
key_id: json.loads(json_bytes)
}
@trace_defered_function
@defer.inlineCallbacks
def claim_client_keys(destination):
TracerUtil.set_tag("destination", destination)
device_keys = remote_queries[destination]
try:
remote_result = yield self.federation.claim_client_keys(
@ -234,8 +263,12 @@ class E2eKeysHandler(object):
for user_id, keys in remote_result["one_time_keys"].items():
if user_id in device_keys:
json_result[user_id] = keys
except Exception as e:
failures[destination] = _exception_to_failure(e)
failure = _exception_to_failure(e)
failures[destination] = failure
TracerUtil.set_tag("error", True)
TracerUtil.set_tag("reason", failure)
yield make_deferred_yieldable(
defer.gatherResults(
@ -259,14 +292,21 @@ class E2eKeysHandler(object):
),
)
TracerUtil.log_kv({"one_time_keys": json_result, "failures": failures})
return {"one_time_keys": json_result, "failures": failures}
@trace_defered_function
@defer.inlineCallbacks
def upload_keys_for_user(self, user_id, device_id, keys):
TracerUtil.set_tag("user_id", user_id)
TracerUtil.set_tag("device_id", device_id)
TracerUtil.set_tag("keys", keys)
time_now = self.clock.time_msec()
# TODO: Validate the JSON to make sure it has the right keys.
device_keys = keys.get("device_keys", None)
TracerUtil.set_tag("device_keys", device_keys)
if device_keys:
logger.info(
"Updating device_keys for device %r for user %s at %d",
@ -287,6 +327,10 @@ class E2eKeysHandler(object):
yield self._upload_one_time_keys_for_user(
user_id, device_id, time_now, one_time_keys
)
else:
TracerUtil.log_kv(
{"event": "did not upload one_time_keys", "reason": "no keys given"}
)
# the device should have been registered already, but it may have been
# deleted due to a race with a DELETE request. Or we may be using an
@ -297,8 +341,10 @@ class E2eKeysHandler(object):
result = yield self.store.count_e2e_one_time_keys(user_id, device_id)
TracerUtil.set_tag("one_time_key_counts", result)
return {"one_time_key_counts": result}
@trace_defered_function
@defer.inlineCallbacks
def _upload_one_time_keys_for_user(
self, user_id, device_id, time_now, one_time_keys

10
synapse/handlers/e2e_room_keys.py
View file

@ -27,6 +27,7 @@ from synapse.api.errors import (
SynapseError,
)
from synapse.util.async_helpers import Linearizer
from synapse.util.tracerutils import TracerUtil, trace_defered_function
logger = logging.getLogger(__name__)
@ -49,6 +50,7 @@ class E2eRoomKeysHandler(object):
# changed.
self._upload_linearizer = Linearizer("upload_room_keys_lock")
@trace_defered_function
@defer.inlineCallbacks
def get_room_keys(self, user_id, version, room_id=None, session_id=None):
"""Bulk get the E2E room keys for a given backup, optionally filtered to a given
@ -84,8 +86,10 @@ class E2eRoomKeysHandler(object):
user_id, version, room_id, session_id
)
TracerUtil.log_kv(results)
return results
@trace_defered_function
@defer.inlineCallbacks
def delete_room_keys(self, user_id, version, room_id=None, session_id=None):
"""Bulk delete the E2E room keys for a given backup, optionally filtered to a given
@ -107,6 +111,7 @@ class E2eRoomKeysHandler(object):
with (yield self._upload_linearizer.queue(user_id)):
yield self.store.delete_e2e_room_keys(user_id, version, room_id, session_id)
@trace_defered_function
@defer.inlineCallbacks
def upload_room_keys(self, user_id, version, room_keys):
"""Bulk upload a list of room keys into a given backup version, asserting
@ -174,6 +179,7 @@ class E2eRoomKeysHandler(object):
user_id, version, room_id, session_id, session
)
@trace_defered_function
@defer.inlineCallbacks
def _upload_room_key(self, user_id, version, room_id, session_id, room_key):
"""Upload a given room_key for a given room and session into a given
@ -236,6 +242,7 @@ class E2eRoomKeysHandler(object):
return False
return True
@trace_defered_function
@defer.inlineCallbacks
def create_version(self, user_id, version_info):
"""Create a new backup version. This automatically becomes the new
@ -264,6 +271,7 @@ class E2eRoomKeysHandler(object):
)
return new_version
@trace_defered_function
@defer.inlineCallbacks
def get_version_info(self, user_id, version=None):
"""Get the info about a given version of the user's backup
@ -294,6 +302,7 @@ class E2eRoomKeysHandler(object):
raise
return res
@trace_defered_function
@defer.inlineCallbacks
def delete_version(self, user_id, version=None):
"""Deletes a given version of the user's e2e_room_keys backup
@ -314,6 +323,7 @@ class E2eRoomKeysHandler(object):
else:
raise
@trace_defered_function
@defer.inlineCallbacks
def update_version(self, user_id, version, version_info):
"""Update the info about a given version of the user's backup

13
synapse/rest/client/v2_alpha/keys.py
View file

@ -25,6 +25,10 @@ from synapse.http.servlet import (
parse_string,
)
from synapse.types import StreamToken
from synapse.util.tracerutils import (
TracerUtil,
trace_defered_function_using_operation_name,
)
from ._base import client_patterns
@ -68,6 +72,7 @@ class KeyUploadServlet(RestServlet):
self.auth = hs.get_auth()
self.e2e_keys_handler = hs.get_e2e_keys_handler()
@trace_defered_function_using_operation_name("upload_keys")
@defer.inlineCallbacks
def on_POST(self, request, device_id):
requester = yield self.auth.get_user_by_req(request, allow_guest=True)
@ -78,6 +83,14 @@ class KeyUploadServlet(RestServlet):
# passing the device_id here is deprecated; however, we allow it
# for now for compatibility with older clients.
if requester.device_id is not None and device_id != requester.device_id:
TracerUtil.set_tag("error", True)
TracerUtil.log_kv(
{
"message": "Client uploading keys for a different device",
"logged_in_id": requester.device_id,
"key_being_uploaded": device_id,
}
)
logger.warning(
"Client uploading keys for a different device "
"(logged in as %s, uploading for %s)",

9
synapse/storage/e2e_room_keys.py
View file

@ -18,11 +18,13 @@ import json
from twisted.internet import defer
from synapse.api.errors import StoreError
from synapse.util.tracerutils import trace_defered_function, trace_function
from ._base import SQLBaseStore
class EndToEndRoomKeyStore(SQLBaseStore):
@trace_defered_function
@defer.inlineCallbacks
def get_e2e_room_key(self, user_id, version, room_id, session_id):
"""Get the encrypted E2E room key for a given session from a given
@ -63,6 +65,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
return row
@trace_defered_function
@defer.inlineCallbacks
def set_e2e_room_key(self, user_id, version, room_id, session_id, room_key):
"""Replaces or inserts the encrypted E2E room key for a given session in
@ -95,6 +98,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
lock=False,
)
@trace_defered_function
@defer.inlineCallbacks
def get_e2e_room_keys(self, user_id, version, room_id=None, session_id=None):
"""Bulk get the E2E room keys for a given backup, optionally filtered to a given
@ -153,6 +157,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
return sessions
@trace_defered_function
@defer.inlineCallbacks
def delete_e2e_room_keys(self, user_id, version, room_id=None, session_id=None):
"""Bulk delete the E2E room keys for a given backup, optionally filtered to a given
@ -194,6 +199,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
raise StoreError(404, "No current backup version")
return row[0]
@trace_function
def get_e2e_room_keys_version_info(self, user_id, version=None):
"""Get info metadata about a version of our room_keys backup.
@ -236,6 +242,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
"get_e2e_room_keys_version_info", _get_e2e_room_keys_version_info_txn
)
@trace_function
def create_e2e_room_keys_version(self, user_id, info):
"""Atomically creates a new version of this user's e2e_room_keys store
with the given version info.
@ -276,6 +283,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
"create_e2e_room_keys_version_txn", _create_e2e_room_keys_version_txn
)
@trace_function
def update_e2e_room_keys_version(self, user_id, version, info):
"""Update a given backup version
@ -292,6 +300,7 @@ class EndToEndRoomKeyStore(SQLBaseStore):
desc="update_e2e_room_keys_version",
)
@trace_function
def delete_e2e_room_keys_version(self, user_id, version=None):
"""Delete a given backup version of the user's room keys.
Doesn't delete their actual key data.

35
synapse/storage/end_to_end_keys.py
View file

@ -22,8 +22,11 @@ from synapse.util.caches.descriptors import cached
from ._base import SQLBaseStore, db_to_json
from synapse.util.tracerutils import TracerUtil, trace_defered_function, trace_function
class EndToEndKeyWorkerStore(SQLBaseStore):
@trace_defered_function
@defer.inlineCallbacks
def get_e2e_device_keys(
self, query_list, include_all_devices=False, include_deleted_devices=False
@ -40,6 +43,7 @@ class EndToEndKeyWorkerStore(SQLBaseStore):
Dict mapping from user-id to dict mapping from device_id to
dict containing "key_json", "device_display_name".
"""
TracerUtil.set_tag("query_list", query_list)
if not query_list:
return {}
@ -57,9 +61,13 @@ class EndToEndKeyWorkerStore(SQLBaseStore):
return results
@trace_function
def _get_e2e_device_keys_txn(
self, txn, query_list, include_all_devices=False, include_deleted_devices=False
):
TracerUtil.set_tag("include_all_devices", include_all_devices)
TracerUtil.set_tag("include_deleted_devices", include_deleted_devices)
query_clauses = []
query_params = []
@ -104,8 +112,10 @@ class EndToEndKeyWorkerStore(SQLBaseStore):
for user_id, device_id in deleted_devices:
result.setdefault(user_id, {})[device_id] = None
TracerUtil.log_kv(result)
return result
@trace_defered_function
@defer.inlineCallbacks
def get_e2e_one_time_keys(self, user_id, device_id, key_ids):
"""Retrieve a number of one-time keys for a user
@ -121,6 +131,10 @@ class EndToEndKeyWorkerStore(SQLBaseStore):
key_id) to json string for key
"""
TracerUtil.set_tag("user_id", user_id)
TracerUtil.set_tag("device_id", device_id)
TracerUtil.set_tag("key_ids", key_ids)
rows = yield self._simple_select_many_batch(
table="e2e_one_time_keys_json",
column="key_id",
@ -145,7 +159,11 @@ class EndToEndKeyWorkerStore(SQLBaseStore):
(algorithm, key_id, key json)
"""
@trace_function
def _add_e2e_one_time_keys(txn):
TracerUtil.set_tag("user_id", user_id)
TracerUtil.set_tag("device_id", device_id)
TracerUtil.set_tag("new_keys", new_keys)
# We are protected from race between lookup and insertion due to
# a unique constraint. If there is a race of two calls to
# `add_e2e_one_time_keys` then they'll conflict and we will only
@ -201,7 +219,13 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
or the keys were already in the database.
"""
@trace_function
def _set_e2e_device_keys_txn(txn):
TracerUtil.set_tag("user_id", user_id)
TracerUtil.set_tag("device_id", device_id)
TracerUtil.set_tag("time_now", time_now)
TracerUtil.set_tag("device_keys", device_keys)
old_key_json = self._simple_select_one_onecol_txn(
txn,
table="e2e_device_keys_json",
@ -215,6 +239,7 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
new_key_json = encode_canonical_json(device_keys).decode("utf-8")
if old_key_json == new_key_json:
TracerUtil.set_tag("error", True)
return False
self._simple_upsert_txn(
@ -231,6 +256,7 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
def claim_e2e_one_time_keys(self, query_list):
"""Take a list of one time keys out of the database"""
@trace_function
def _claim_e2e_one_time_keys(txn):
sql = (
"SELECT key_id, key_json FROM e2e_one_time_keys_json"
@ -252,7 +278,13 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
" AND key_id = ?"
)
for user_id, device_id, algorithm, key_id in delete:
TracerUtil.log_kv(
{"message": "executing claim transaction on database"}
)
txn.execute(sql, (user_id, device_id, algorithm, key_id))
TracerUtil.log_kv(
{"message": "finished executing and invalidating cache"}
)
self._invalidate_cache_and_stream(
txn, self.count_e2e_one_time_keys, (user_id, device_id)
)
@ -261,7 +293,10 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
return self.runInteraction("claim_e2e_one_time_keys", _claim_e2e_one_time_keys)
def delete_e2e_keys_by_device(self, user_id, device_id):
@trace_function
def delete_e2e_keys_by_device_txn(txn):
TracerUtil.set_tag("user_id", user_id)
TracerUtil.set_tag("device_id", device_id)
self._simple_delete_txn(
txn,
table="e2e_device_keys_json",