Matrix-Mirrors/synapse-product
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse-product.git synced 2025-02-07 23:55:18 -05:00
Code Issues Packages Projects Releases Wiki Activity

apply changes as a result of PR review

Browse Source
This commit is contained in:
Hubert Chathi 2019-10-30 10:01:53 -04:00
parent c40d7244f8
commit d78b1e339d
2 changed files with 46 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
synapse/handlers/e2e_keys.py
View File

@ -1072,7 +1072,7 @@ class SignatureListItem:
class SigningKeyEduUpdater(object): class SigningKeyEduUpdater(object):
"Handles incoming signing key updates from federation and updates the DB" """Handles incoming signing key updates from federation and updates the DB"""
def __init__(self, hs, e2e_keys_handler): def __init__(self, hs, e2e_keys_handler):
self.store = hs.get_datastore() self.store = hs.get_datastore()
@ -1111,7 +1111,6 @@ class SigningKeyEduUpdater(object):
self_signing_key = edu_content.pop("self_signing_key", None) self_signing_key = edu_content.pop("self_signing_key", None)
if get_domain_from_id(user_id) != origin: if get_domain_from_id(user_id) != origin:
# TODO: Raise?
logger.warning("Got signing key update edu for %r from %r", user_id, origin) logger.warning("Got signing key update edu for %r from %r", user_id, origin)
return return
@ -1122,7 +1121,7 @@ class SigningKeyEduUpdater(object):
return return
self._pending_updates.setdefault(user_id, []).append( self._pending_updates.setdefault(user_id, []).append(
(master_key, self_signing_key, edu_content) (master_key, self_signing_key)
) )
yield self._handle_signing_key_updates(user_id) yield self._handle_signing_key_updates(user_id)
@ -1147,22 +1146,21 @@ class SigningKeyEduUpdater(object):
logger.info("pending updates: %r", pending_updates) logger.info("pending updates: %r", pending_updates)
for master_key, self_signing_key, edu_content in pending_updates: for master_key, self_signing_key in pending_updates:
if master_key: if master_key:
yield self.store.set_e2e_cross_signing_key( yield self.store.set_e2e_cross_signing_key(
user_id, "master", master_key user_id, "master", master_key
) )
device_id = get_verify_key_from_cross_signing_key(master_key)[ _, verify_key = get_verify_key_from_cross_signing_key(master_key)
1 # verify_key is a VerifyKey from signedjson, which uses
].version # .version to denote the portion of the key ID after the
device_ids.append(device_id) # algorithm and colon, which is the device ID
device_ids.append(verify_key.version)
if self_signing_key: if self_signing_key:
yield self.store.set_e2e_cross_signing_key( yield self.store.set_e2e_cross_signing_key(
user_id, "self_signing", self_signing_key user_id, "self_signing", self_signing_key
) )
device_id = get_verify_key_from_cross_signing_key(self_signing_key)[ _, verify_key = get_verify_key_from_cross_signing_key(self_signing_key)
1 device_ids.append(verify_key.version)
].version
device_ids.append(device_id)
yield device_handler.notify_device_update(user_id, device_ids) yield device_handler.notify_device_update(user_id, device_ids)

79
synapse/storage/data_stores/main/devices.py
View File

@ -92,8 +92,12 @@ class DeviceWorkerStore(SQLBaseStore):
@trace @trace
@defer.inlineCallbacks @defer.inlineCallbacks
def get_devices_by_remote(self, destination, from_stream_id, limit): def get_devices_by_remote(self, destination, from_stream_id, limit):
"""Get stream of updates to send to remote servers """Get a stream of device updates to send to the given remote server.
Args:
destination (str): The host the device updates are intended for
from_stream_id (int): The minimum stream_id to filter updates by, exclusive
limit (int): Maximum number of device updates to return
Returns: Returns:
Deferred[tuple[int, list[tuple[string,dict]]]]: Deferred[tuple[int, list[tuple[string,dict]]]]:
current stream id (ie, the stream id of the last update included in the current stream id (ie, the stream id of the last update included in the
@ -131,7 +135,8 @@ class DeviceWorkerStore(SQLBaseStore):
if not updates: if not updates:
return now_stream_id, [] return now_stream_id, []
# get the cross-signing keys of the users the list # get the cross-signing keys of the users in the list, so that we can
# determine which of the device changes were cross-signing keys
users = set(r[0] for r in updates) users = set(r[0] for r in updates)
master_key_by_user = {} master_key_by_user = {}
self_signing_key_by_user = {} self_signing_key_by_user = {}
@ -141,9 +146,12 @@ class DeviceWorkerStore(SQLBaseStore):
key_id, verify_key = get_verify_key_from_cross_signing_key( key_id, verify_key = get_verify_key_from_cross_signing_key(
cross_signing_key cross_signing_key
) )
# verify_key is a VerifyKey from signedjson, which uses
# .version to denote the portion of the key ID after the
# algorithm and colon, which is the device ID
master_key_by_user[user] = { master_key_by_user[user] = {
"key_info": cross_signing_key, "key_info": cross_signing_key,
"pubkey": verify_key.version, "device_id": verify_key.version,
} }
cross_signing_key = yield self.get_e2e_cross_signing_key( cross_signing_key = yield self.get_e2e_cross_signing_key(
@ -155,7 +163,7 @@ class DeviceWorkerStore(SQLBaseStore):
) )
self_signing_key_by_user[user] = { self_signing_key_by_user[user] = {
"key_info": cross_signing_key, "key_info": cross_signing_key,
"pubkey": verify_key.version, "device_id": verify_key.version,
} }
# if we have exceeded the limit, we need to exclude any results with the # if we have exceeded the limit, we need to exclude any results with the
@ -182,69 +190,54 @@ class DeviceWorkerStore(SQLBaseStore):
# context which created the Edu. # context which created the Edu.
query_map = {} query_map = {}
for update in updates: cross_signing_keys_by_user = {}
if stream_id_cutoff is not None and update[2] >= stream_id_cutoff: for user_id, device_id, update_stream_id, update_context in updates:
if stream_id_cutoff is not None and update_stream_id >= stream_id_cutoff:
# Stop processing updates # Stop processing updates
break break
# skip over cross-signing keys
if ( if (
update[0] in master_key_by_user user_id in master_key_by_user
and update[1] == master_key_by_user[update[0]]["pubkey"] and device_id == master_key_by_user[user_id]["device_id"]
) or (
update[0] in master_key_by_user
and update[1] == self_signing_key_by_user[update[0]]["pubkey"]
): ):
continue
key = (update[0], update[1])
update_context = update[3]
update_stream_id = update[2]
previous_update_stream_id, _ = query_map.get(key, (0, None))
if update_stream_id > previous_update_stream_id:
query_map[key] = (update_stream_id, update_context)
# If we didn't find any updates with a stream_id lower than the cutoff, it
# means that there are more than limit updates all of which have the same
# steam_id.
# figure out which cross-signing keys were changed by intersecting the
# update list with the master/self-signing key by user maps
cross_signing_keys_by_user = {}
for user_id, device_id, stream, _opentracing_context in updates:
if device_id == master_key_by_user.get(user_id, {}).get("pubkey", None):
result = cross_signing_keys_by_user.setdefault(user_id, {}) result = cross_signing_keys_by_user.setdefault(user_id, {})
result["master_key"] = master_key_by_user[user_id]["key_info"] result["master_key"] = master_key_by_user[user_id]["key_info"]
elif device_id == self_signing_key_by_user.get(user_id, {}).get( elif (
"pubkey", None user_id in master_key_by_user
and device_id == self_signing_key_by_user[user_id]["device_id"]
): ):
result = cross_signing_keys_by_user.setdefault(user_id, {}) result = cross_signing_keys_by_user.setdefault(user_id, {})
result["self_signing_key"] = self_signing_key_by_user[user_id][ result["self_signing_key"] = self_signing_key_by_user[user_id][
"key_info" "key_info"
] ]
else:
key = (user_id, device_id)
cross_signing_results = [] previous_update_stream_id, _ = query_map.get(key, (0, None))
# add the updated cross-signing keys to the results list if update_stream_id > previous_update_stream_id:
for user_id, result in iteritems(cross_signing_keys_by_user): query_map[key] = (update_stream_id, update_context)
result["user_id"] = user_id
# FIXME: switch to m.signing_key_update when MSC1756 is merged into the spec # If we didn't find any updates with a stream_id lower than the cutoff, it
cross_signing_results.append(("org.matrix.signing_key_update", result)) # means that there are more than limit updates all of which have the same
# steam_id.
# That should only happen if a client is spamming the server with new # That should only happen if a client is spamming the server with new
# devices, in which case E2E isn't going to work well anyway. We'll just # devices, in which case E2E isn't going to work well anyway. We'll just
# skip that stream_id and return an empty list, and continue with the next # skip that stream_id and return an empty list, and continue with the next
# stream_id next time. # stream_id next time.
if not query_map and not cross_signing_results: if not query_map and not cross_signing_keys_by_user:
return stream_id_cutoff, [] return stream_id_cutoff, []
results = yield self._get_device_update_edus_by_remote( results = yield self._get_device_update_edus_by_remote(
destination, from_stream_id, query_map destination, from_stream_id, query_map
) )
results.extend(cross_signing_results)
# add the updated cross-signing keys to the results list
for user_id, result in iteritems(cross_signing_keys_by_user):
result["user_id"] = user_id
# FIXME: switch to m.signing_key_update when MSC1756 is merged into the spec
results.append(("org.matrix.signing_key_update", result))
return now_stream_id, results return now_stream_id, results