in kinto-remote-settings/src/kinto_remote_settings/signer/listeners.py [0:0]
def prevent_collection_delete(event, resources):
request = event.request
bid = event.payload["bucket_id"]
for impacted in event.impacted_objects:
cid = impacted["old"]["id"]
# Locate any collections that imply usage of this collection.
# If there's some path s -> p -> d for which this collection
# corresponds to p or d, we forbid deletion of this collection
# (it's "in use").
in_use = None
# The most obvious path is if there is a signer that mentions
# this collection explicitly in p or d.
specific_signers = [
v
for v in resources.values()
if v["source"]["collection"] is not None
and signer_impacts_resource(v, bid, cid)
]
if specific_signers:
assert len(specific_signers) == 1, (
f"Inconsistent signers: multiple signers touch {bid} and {cid}"
)
in_use = specific_signers[0]
if not in_use:
# We identify bucket-wide signers for which p or d matches
# this collection -- in this case, editing the collection of
# the same name in s could trigger writes to p or d.
bucket_signers = [
v
for v in resources.values()
if v["source"]["collection"] is None
and signer_impacts_resource(v, bid, cid)
]
if bucket_signers:
assert len(bucket_signers) == 1, (
f"Inconsistent signers: multiple signers touch {bid}"
)
in_use = bucket_signers[0]
if in_use:
# See if this bucket-wide signer is superseded by any
# specific-collection signers. A specific-collection
# signer counts as superseding a bucket-wide signer if
# the specific collection is in the same bucket as the
# bucket-wide signer, and the specific-collection
# signer has the same collection ID as the collection
# being deleted. In this case, we can ignore the
# bucket-wide s -> p -> d because the
# collection-specific signer specifies a different
# workflow for the collection that we thought to
# impact this one.
#
# Specific-collection signers that point *from* other
# collections to this one are handled explicitly, above.
#
# N.B. We can't use signer_impacts_resource here
# because we want to detect a signer for a
# specific source collection, regardless of whether it
# impacts the collection to be deleted or not. A good
# example where this comes up is where a
# specific-collection signer disables preview. We want
# to find this signer even though the preview
# collection is no longer being impacted.
for signer in resources.values():
same_bucket = (
signer["source"]["bucket"] == in_use["source"]["bucket"]
)
this_collection = signer["source"]["collection"] == cid
if same_bucket and this_collection:
# Clear the bucket-wide signer.
# This signer either named this collection
# explicitly (in which case it was handled
# above), or it didn't (in which case the
# collection is safe to be deleted).
in_use = None
break
if in_use is None:
# Can delete!
continue
source_bucket_uri = instance_uri(
event.request, "bucket", id=in_use["source"]["bucket"]
)
source_collection_id = in_use["source"]["collection"] or cid
try:
request.registry.storage.get(
resource_name="collection",
parent_id=source_bucket_uri,
object_id=source_collection_id,
)
raise_forbidden(message="Collection is in use.")
except ObjectNotFoundError:
# Do not prevent delete of preview/destination if source does not exist.
pass