Rewrite the KeyRing

[erikjohnston]

Sorry for a bit of a mammoth PR, but this is a total rewrite of the KeyRing class to use BatchingQueue rather than a mess of deferreds. The idea here is that we write the functions out in a way to handle each server key request separately, and then rely on BatchingQueue to merge requests that happen at the same when querying the keys from the database or perspective key servers.

It's probably best to review commit by commit, and for the first to ignore the diff of the KeyRing class and instead look at the end result by itself:

synapse/synapse/crypto/keyring.py

Lines 281 to 447 in f941830

	async def process_request(self, verify_request: VerifyJsonRequest) -> None:
	"""Processes the `VerifyJsonRequest`. Raises if the object is not signed
	by the server, the signatures don't match or we failed to fetch the
	necessary keys.
	"""
	if not verify_request.key_ids:
	raise SynapseError(
	400,
	f"Not signed by {verify_request.server_name}",
	Codes.UNAUTHORIZED,
	)
	# Add they keys we need to verify to the queue for retrieval. We queue
	# up requests for the same server so we don't end up with many in flight
	# requests for the same keys.
	found_keys_by_server = await self._server_queue.add_to_queue(
	verify_request.to_fetch_key_request(), key=verify_request.server_name
	)
	# Since we batch up requests the returned set of keys may contain keys
	# from other servers, so we pull out only the ones we care about.s
	found_keys = found_keys_by_server.get(verify_request.server_name, {})
	# For each signature to check we ensure we have fetched the necessary
	# keys and the signature matches.
	for key_id in verify_request.key_ids:
	key_result = found_keys.get(key_id)
	if not key_result:
	raise SynapseError(
	401,
	f"Missing keys for {verify_request.server_name}: {key_id}",
	Codes.UNAUTHORIZED,
	)
	if key_result.valid_until_ts < verify_request.minimum_valid_until_ts:
	raise SynapseError(
	401,
	f"Failed to find key with recent enough `valid_until_ts` for {verify_request.server_name}: {key_id}",
	Codes.UNAUTHORIZED,
	)
	verify_key = key_result.verify_key
	try:
	json_object = verify_request.get_json_object()
	verify_signed_json(
	json_object,
	verify_request.server_name,
	verify_key,
	)
	except SignatureVerifyException as e:
	logger.debug(
	"Error verifying signature for %s:%s:%s with key %s: %s",
	verify_request.server_name,
	verify_key.alg,
	verify_key.version,
	encode_verify_key_base64(verify_key),
	str(e),
	)
	raise SynapseError(
	401,
	"Invalid signature for server %s with key %s:%s: %s"
	% (
	verify_request.server_name,
	verify_key.alg,
	verify_key.version,
	str(e),
	),
	Codes.UNAUTHORIZED,
	)
	async def _inner_fetch_key_requests(
	self, requests: List[_FetchKeyRequest]
	) -> Dict[str, Dict[str, FetchKeyResult]]:
	"""Processing function for the queue of `_FetchKeyRequest`."""
	logger.debug("Starting fetch for %s", requests)
	# First we need to deduplicate requests for the same key. We do this by
	# taking the *maximum* requested `minimum_valid_until_ts` for each pair
	# of server name/key ID.
	server_to_key_to_ts = {} # type: Dict[str, Dict[str, int]]
	for request in requests:
	by_server = server_to_key_to_ts.setdefault(request.server_name, {})
	for key_id in request.key_ids:
	existing_ts = by_server.get(key_id)
	if existing_ts:
	by_server[key_id] = max(request.minimum_valid_until_ts, existing_ts)
	else:
	by_server[key_id] = request.minimum_valid_until_ts
	deduped_requests = [
	_FetchKeyRequest(server_name, minimum_valid_ts, [key_id])
	for server_name, by_server in server_to_key_to_ts.items()
	for key_id, minimum_valid_ts in by_server.items()
	]
	logger.debug("Deduplicated key requests to %s", deduped_requests)
	# For each key we call `_inner_verify_request` which will handle
	# fetching each key. Note these shouldn't throw if we fail to contact
	# other servers etc.
	results_per_request = await yieldable_gather_results(
	self._inner_fetch_key_request,
	deduped_requests,
	)
	# We now convert the returned list of results into a map from server
	# name to key ID to FetchKeyResult, to return.
	to_return = {} # type: Dict[str, Dict[str, FetchKeyResult]]
	for (request, results) in zip(deduped_requests, results_per_request):
	to_return_by_server = to_return.setdefault(request.server_name, {})
	for key_id, key_result in results.items():
	existing = to_return_by_server.get(key_id)
	if not existing or existing.valid_until_ts < key_result.valid_until_ts:
	to_return_by_server[key_id] = key_result
	return to_return
	async def _inner_fetch_key_request(
	self, verify_request: _FetchKeyRequest
	) -> Dict[str, FetchKeyResult]:
	"""Attempt to fetch the given key by calling each key fetcher one by
	one.
	"""
	logger.debug("Starting fetch for %s", verify_request)
	found_keys: Dict[str, FetchKeyResult] = {}
	missing_key_ids = set(verify_request.key_ids)
	for fetcher in self._key_fetchers:
	if not missing_key_ids:
	break
	logger.debug("Getting keys from %s for %s", fetcher, verify_request)
	keys = await fetcher.get_keys(
	verify_request.server_name,
	list(missing_key_ids),
	verify_request.minimum_valid_until_ts,
	)
	for key_id, key in keys.items():
	if not key:
	continue
	# If we already have a result for the given key ID we keep the
	# one with the highest `valid_until_ts`.
	existing_key = found_keys.get(key_id)
	if existing_key:
	if key.valid_until_ts <= existing_key.valid_until_ts:
	continue
	# We always store the returned key even if it doesn't the
	# `minimum_valid_until_ts` requirement, as some verification
	# requests may still be able to be satisfied by it.
	#
	# We still keep looking for the key from other fetchers in that
	# case though.
	found_keys[key_id] = key
	if key.valid_until_ts < verify_request.minimum_valid_until_ts:
	continue
	missing_key_ids.discard(key_id)
	return found_keys

Fixes #3825

[richvdh]

well, it's certainly much clearer now.

I've got a few concerns though. The big one is about what performance of concurrent fetches of keys when one key is quick and another is slow.

I'm also a bit concerned about dropping the event id from the logs when verification fails.

[richvdh]

give these some docstrings? In particular, is key_ids disjunctive (any one key id will do) or conjunctive (we want all of them?)

[richvdh]

(aside: do we define anywhere what the semantics should be if a JSON object is signed by multiple keys from the same server? Do they all have to be verified, or will one do? It looks like currently we require them all, but semantically that feels wrong to me.)

[erikjohnston]

This does feel wrong to me, and re-reading the previous code I think we actually did only require one signature to be valid?

[erikjohnston]

Could you have a quick look to see if you agree that that's the case, and then I'll fix this PR to do the same (and document it at the same time)

[richvdh]

yes, looks like we will run the verification with the first key we find.

[erikjohnston]

Pushed the change. The code works slightly differently now where we try and find all the keys before running the verifications, and also bails if either a) any of the keys we find fail verification or b) we don't find any keys.

I don't think the change matters much, as I think its pretty rare for a server to sign the same event multiple times?

[richvdh]

lgtm other than the nits below