Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return []
attrs = xml_children_as_dict(element)
transformed = transform_attributes(attrs, cls)
if hasattr(cls, "fill_extra_attributes"):
transformed = cls.fill_extra_attributes(transformed)
entity = cls(**transformed)
# Add protocol name
entity._source_protocol = "diaspora"
# Save element object to entity for possible later use
entity._source_object = etree.tostring(element)
# Save receivers on the entity
if user:
# Single receiver
entity._receivers = [UserType(id=user.id, receiver_variant=ReceiverVariant.ACTOR)]
else:
# Followers
entity._receivers = [UserType(id=sender, receiver_variant=ReceiverVariant.FOLLOWERS)]
if issubclass(cls, DiasporaRelayableMixin):
# If relayable, fetch sender key for validation
entity._xml_tags = get_element_child_info(element, "tag")
if sender_key_fetcher:
entity._sender_key = sender_key_fetcher(entity.actor_id)
else:
profile = retrieve_and_parse_profile(entity.handle)
if profile:
entity._sender_key = profile.public_key
else:
# If not relayable, ensure handles match
if not check_sender_and_entity_handle_match(sender, entity.handle):
return
# Check for this being a list reference to followers of an actor?
# TODO: terrible hack! the way some platforms deliver to sharedInbox using just
# the followers collection as a target is annoying to us since we would have to
# store the followers collection references on application side, which we don't
# want to do since it would make application development another step more complex.
# So for now we're going to do a terrible assumption that
# 1) if "followers" in ID and
# 2) if ID starts with actor ID
# then; assume this is the followers collection of said actor ID.
# When we have a caching system, just fetch each receiver and check what it is.
# Without caching this would be too expensive to do.
elif receiver.find("followers") > -1 and receiver.startswith(actor):
return UserType(id=actor, receiver_variant=ReceiverVariant.FOLLOWERS)
# Assume actor ID
return UserType(id=receiver, receiver_variant=ReceiverVariant.ACTOR)
transformed = transform_attributes(attrs, cls)
if hasattr(cls, "fill_extra_attributes"):
transformed = cls.fill_extra_attributes(transformed)
entity = cls(**transformed)
# Add protocol name
entity._source_protocol = "diaspora"
# Save element object to entity for possible later use
entity._source_object = etree.tostring(element)
# Save receivers on the entity
if user:
# Single receiver
entity._receivers = [UserType(id=user.id, receiver_variant=ReceiverVariant.ACTOR)]
else:
# Followers
entity._receivers = [UserType(id=sender, receiver_variant=ReceiverVariant.FOLLOWERS)]
if issubclass(cls, DiasporaRelayableMixin):
# If relayable, fetch sender key for validation
entity._xml_tags = get_element_child_info(element, "tag")
if sender_key_fetcher:
entity._sender_key = sender_key_fetcher(entity.actor_id)
else:
profile = retrieve_and_parse_profile(entity.handle)
if profile:
entity._sender_key = profile.public_key
else:
# If not relayable, ensure handles match
if not check_sender_and_entity_handle_match(sender, entity.handle):
return []
try:
entity.validate()
if receiver == NAMESPACE_PUBLIC:
# Ignore since we already store "public" as a boolean on the entity
return
# Check for this being a list reference to followers of an actor?
# TODO: terrible hack! the way some platforms deliver to sharedInbox using just
# the followers collection as a target is annoying to us since we would have to
# store the followers collection references on application side, which we don't
# want to do since it would make application development another step more complex.
# So for now we're going to do a terrible assumption that
# 1) if "followers" in ID and
# 2) if ID starts with actor ID
# then; assume this is the followers collection of said actor ID.
# When we have a caching system, just fetch each receiver and check what it is.
# Without caching this would be too expensive to do.
elif receiver.find("followers") > -1 and receiver.startswith(actor):
return UserType(id=actor, receiver_variant=ReceiverVariant.FOLLOWERS)
# Assume actor ID
return UserType(id=receiver, receiver_variant=ReceiverVariant.ACTOR)
activity_id=f"{self.target_id}#accept-{uuid.uuid4()}",
actor_id=self.target_id,
target_id=self.activity_id,
object=self.to_as2(),
)
try:
profile = retrieve_and_parse_profile(self.actor_id)
except Exception:
profile = None
if not profile:
logger.warning("ActivitypubFollow.post_receive - Failed to fetch remote profile for sending back Accept")
return
try:
handle_send(
accept,
UserType(id=self.target_id, private_key=key),
recipients=[{
"endpoint": profile.inboxes["private"],
"fid": self.actor_id,
"protocol": "activitypub",
"public": False,
}],
)
except Exception:
logger.exception("ActivitypubFollow.post_receive - Failed to send Accept back")