Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self.guid:
as2["@context"].append(CONTEXT_DIASPORA)
as2["object"]["diaspora:guid"] = self.guid
return as2
class ActivitypubComment(ActivitypubNoteMixin, Comment):
def to_as2(self) -> Dict:
as2 = super().to_as2()
as2["object"]["inReplyTo"] = self.target_id
return as2
class ActivitypubFollow(ActivitypubEntityMixin, Follow):
_type = ActivityType.FOLLOW.value
def post_receive(self) -> None:
"""
Post receive hook - send back follow ack.
"""
super().post_receive()
if not self.following:
return
from federation.utils.activitypub import retrieve_and_parse_profile # Circulars
try:
from federation.utils.django import get_function_from_config
except ImportError:
logger.warning("ActivitypubFollow.post_receive - Unable to send automatic Accept back, only supported on "
"Django currently")
return
def to_as2(self) -> Dict:
if self.following:
as2 = {
"@context": CONTEXTS_DEFAULT,
"id": self.activity_id,
"type": self._type,
"actor": self.actor_id,
"object": self.target_id,
}
else:
as2 = {
"@context": CONTEXTS_DEFAULT,
"id": self.activity_id,
"type": ActivityType.UNDO.value,
"actor": self.actor_id,
"object": {
"id": f"{self.actor_id}#follow-{uuid.uuid4()}",
"type": self._type,
"actor": self.actor_id,
"object": self.target_id,
},
}
return as2
def remove_tag_links(attrs, new=False):
rel = (None, "rel")
if attrs.get(rel) == "tag":
return
return attrs
self.raw_content = bleach.linkify(
self.raw_content,
callbacks=[remove_tag_links],
parse_email=False,
skip_tags=["code", "pre"],
)
class ActivitypubAccept(ActivitypubEntityMixin, Accept):
_type = ActivityType.ACCEPT.value
object: Dict = None
def to_as2(self) -> Dict:
as2 = {
"@context": CONTEXTS_DEFAULT,
"id": self.activity_id,
"type": self._type,
"actor": self.actor_id,
"object": self.object,
}
return as2
class ActivitypubNoteMixin(AttachImagesMixin, CleanContentMixin, PublicMixin, ActivitypubEntityMixin):
_type = ObjectType.NOTE.value
def resolve_type(self):
return {
"Comment": ActivityType.DELETE.value,
"Post": ActivityType.DELETE.value,
"Share": ActivityType.UNDO.value,
}.get(self.entity_type)
def resolve_object_type(self):
return {
"Comment": ObjectType.TOMBSTONE.value,
"Post": ObjectType.TOMBSTONE.value,
"Share": ActivityType.ANNOUNCE.value,
}.get(self.entity_type)
self.media_type = self.get_media_type()
def get_media_type(self) -> str:
media_type = fetch_content_type(self.url)
if media_type in self._valid_media_types:
return media_type
return ""
class Comment(RawContentMixin, ParticipationMixin, CreatedAtMixin, RootTargetIDMixin, BaseEntity):
"""Represents a comment, linked to another object."""
participation = "comment"
url = ""
_allowed_children = (Image,)
_default_activity = ActivityType.CREATE
class Follow(CreatedAtMixin, TargetIDMixin, BaseEntity):
"""Represents a handle following or unfollowing another handle."""
following = True
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._required += ["following"]
# ID not required for follow
self._required.remove('id')
class Post(RawContentMixin, PublicMixin, CreatedAtMixin, ProviderDisplayNameMixin, BaseEntity):
"""Reflects a post, status message, etc, which will be composed from the message or to the message."""
location = ""