erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

component AP gateway: AP <=> XMPP reactions conversions:

Browse source 
- Pubsub Attachments plugin has been renamed to XEP-0470 following publication
- XEP-0470 has been updated to follow 0.2 changes
- AP reactions (as implemented in Pleroma) are converted to XEP-0470
- XEP-0470 events are converted to AP reactions (again, using "EmojiReact" from Pleroma)
- AP activities related to attachments (like/reactions) are cached in Libervia because
  it's not possible to retrieve them from Pleroma instances once they have been emitted
  (doing an HTTP get on their ID returns a 404). For now those cache are not flushed, this
  should be improved in the future.
- `sharedInbox` is used when available. Pleroma returns a 500 HTTP error when ``to`` or
  ``cc`` are used in a direct inbox.
- reactions and like are not currently used for direct messages, because they can't be
  emitted from Pleroma in this case, thus there is no point in implementing them for the
  moment.

rel 371
This commit is contained in:
Goffi 2022-08-31 17:07:03 +02:00
commit 5e3d4ca581
4 changed files with 175 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

61
sat/plugins/plugin_comp_ap_gateway/__init__.py
View file

@ -71,6 +71,7 @@ from .constants import (
TYPE_TOMBSTONE,
TYPE_MENTION,
TYPE_LIKE,
TYPE_REACTION,
NS_AP,
NS_AP_PUBLIC,
PUBLIC_TUPLE
@ -92,8 +93,8 @@ PLUGIN_INFO = {
C.PI_PROTOCOLS: [],
C.PI_DEPENDENCIES: [
"XEP-0054", "XEP-0060", "XEP-0084", "XEP-0106", "XEP-0277", "XEP-0292",
"XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "PUBSUB_CACHE", "TEXT_SYNTAXES",
"IDENTITY", "PUBSUB_ATTACHMENTS"
"XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "XEP-0470", "PUBSUB_CACHE",
"TEXT_SYNTAXES", "IDENTITY"
],
C.PI_RECOMMENDATIONS: [],
C.PI_MAIN: "APGateway",
@ -132,7 +133,7 @@ class APGateway:
self._c = host.plugins["PUBSUB_CACHE"]
self._t = host.plugins["TEXT_SYNTAXES"]
self._i = host.plugins["IDENTITY"]
self._pa = host.plugins["PUBSUB_ATTACHMENTS"]
self._pa = host.plugins["XEP-0470"]
self._p.addManagedNode(
"",
items_cb=self._itemsReceived,
@ -335,7 +336,7 @@ class APGateway:
if resp.code >= 300:
text = await resp.text()
if resp.code == 404:
raise exceptions.NotFound()
raise exceptions.NotFound(f"Can't find resource at {url}")
else:
msg = f"HTTP error {resp.code}: {text}"
raise exceptions.ExternalRequestError(msg)
@ -1104,6 +1105,7 @@ class APGateway:
old_attachment_pubsub_items = await self.host.memory.storage.searchPubsubItems({
"profiles": [self.client.profile],
"services": [service],
"nodes": [node],
"names": [item_elt["id"]]
})
if not old_attachment_pubsub_items:
@ -1127,6 +1129,7 @@ class APGateway:
# no known element was present in attachments
attachments = {}
# noticed
if "noticed" in attachments:
if not "noticed" in old_attachment:
# new "noticed" attachment, we translate to "Like" activity
@ -1134,7 +1137,8 @@ class APGateway:
like = self.createActivity(
TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
)
like["to"] = [NS_AP_PUBLIC]
like["to"] = [ap_account]
like["cc"] = [NS_AP_PUBLIC]
await self.signAndPost(inbox, publisher_actor_id, like)
else:
if "noticed" in old_attachment:
@ -1143,10 +1147,36 @@ class APGateway:
like = self.createActivity(
TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
)
like["to"] = [NS_AP_PUBLIC]
like["to"] = [ap_account]
like["cc"] = [NS_AP_PUBLIC]
undo = self.createActivity("Undo", publisher_actor_id, like)
await self.signAndPost(inbox, publisher_actor_id, undo)
# reactions
new_reactions = set(attachments.get("reactions", {}).get("reactions", []))
old_reactions = set(old_attachment.get("reactions", {}).get("reactions", []))
reactions_remove = old_reactions - new_reactions
reactions_add = new_reactions - old_reactions
for reactions, undo in ((reactions_remove, True), (reactions_add, False)):
for reaction in reactions:
activity_id = self.buildAPURL(
"reaction", item_account, item_id, reaction.encode().hex()
)
reaction_activity = self.createActivity(
TYPE_REACTION, publisher_actor_id, item_url,
activity_id=activity_id
)
reaction_activity["content"] = reaction
reaction_activity["to"] = [ap_account]
reaction_activity["cc"] = [NS_AP_PUBLIC]
if undo:
activy = self.createActivity(
"Undo", publisher_actor_id, reaction_activity
)
else:
activy = reaction_activity
await self.signAndPost(inbox, publisher_actor_id, activy)
if service.user and self.isVirtualJID(service):
# the item is on a virtual service, we need to store it in cache
log.debug("storing attachments item in cache")
@ -1262,9 +1292,18 @@ class APGateway:
href = await self.getAPActorIdFromAccount(account)
return await self.apGet(href)
async def getAPInboxFromId(self, actor_id: str) -> str:
"""Retrieve inbox of an actor_id"""
async def getAPInboxFromId(self, actor_id: str, use_shared: bool = True) -> str:
"""Retrieve inbox of an actor_id
@param use_shared: if True, and a shared inbox exists, it will be used instead of
the user inbox
"""
data = await self.getActorData(actor_id)
if use_shared:
try:
return data["endpoints"]["sharedInbox"]
except KeyError:
pass
return data["inbox"]
@async_lru(maxsize=LRU_MAX_SIZE)
@ -2039,7 +2078,7 @@ class APGateway:
actor_account = self._e.unescape(mess_data["to"].user)
actor_id = await self.getAPActorIdFromAccount(actor_account)
inbox = await self.getAPInboxFromId(actor_id)
inbox = await self.getAPInboxFromId(actor_id, use_shared=False)
try:
language, message = next(iter(mess_data["message"].items()))
@ -2085,7 +2124,7 @@ class APGateway:
)
ap_account = self._e.unescape(to_jid.user)
actor_id = await self.getAPActorIdFromAccount(ap_account)
inbox = await self.getAPInboxFromId(actor_id)
inbox = await self.getAPInboxFromId(actor_id, use_shared=False)
url_actor, ap_item = await self.apDeleteItem(
from_jid.userhostJID(), None, fastened_elts.id, public=False
)
@ -2175,7 +2214,7 @@ class APGateway:
"name": ap_account,
})
inbox = await self.getAPInboxFromId(actor_id)
inbox = await self.getAPInboxFromId(actor_id, use_shared=False)
resp = await self.signAndPost(inbox, ap_item["actor"], ap_item)

10
sat/plugins/plugin_comp_ap_gateway/constants.py
View file

@ -30,6 +30,7 @@ TYPE_ITEM = "item"
TYPE_TOMBSTONE = "Tombstone"
TYPE_MENTION = "Mention"
TYPE_LIKE = "Like"
TYPE_REACTION = "EmojiReact"
MEDIA_TYPE_AP = "application/activity+json"
NS_AP = "https://www.w3.org/ns/activitystreams"
NS_AP_PUBLIC = f"{NS_AP}#Public"
@ -60,7 +61,9 @@ ACTIVITY_TYPES = (
"Accept", "Add", "Announce", "Arrive", "Block", "Create", "Delete", "Dislike", "Flag",
"Follow", "Ignore", "Invite", "Join", "Leave", "Like", "Listen", "Move", "Offer",
"Question", "Reject", "Read", "Remove", "TentativeReject", "TentativeAccept",
"Travel", "Undo", "Update", "View"
"Travel", "Undo", "Update", "View",
# non-standard activities
"EmojiReact"
)
ACTIVITY_TYPES_LOWER = [a.lower() for a in ACTIVITY_TYPES]
ACTIVITY_OBJECT_MANDATORY = (
@ -69,7 +72,9 @@ ACTIVITY_OBJECT_MANDATORY = (
ACTIVITY_TARGET_MANDATORY = ("Add", "Remove")
# activities which can be used with Shared Inbox (i.e. with no account specified)
# must be lowercase
ACTIVIY_NO_ACCOUNT_ALLOWED = ("create", "delete", "announce", "undo")
ACTIVIY_NO_ACCOUNT_ALLOWED = (
"create", "delete", "announce", "undo", "like", "emojireact"
)
# maximum number of parents to retrieve when comments_max_depth option is set
COMMENTS_MAX_PARENTS = 100
# maximum size of avatar, in bytes
@ -77,3 +82,4 @@ MAX_AVATAR_SIZE = 1024 * 1024 * 5
# storage prefixes
ST_AVATAR = "[avatar]"
ST_AP_CACHE = "[AP_item_cache]"

95
sat/plugins/plugin_comp_ap_gateway/http_server.py
View file

@ -42,7 +42,8 @@ from sat.memory.sqla_mapping import SubscriptionState
from .constants import (
NS_AP, CONTENT_TYPE_AP, TYPE_ACTOR, TYPE_INBOX, TYPE_SHARED_INBOX, TYPE_OUTBOX,
AP_REQUEST_TYPES, PAGE_SIZE, ACTIVITY_TYPES_LOWER, ACTIVIY_NO_ACCOUNT_ALLOWED,
SIGN_HEADERS, HS2019, SIGN_EXP, TYPE_FOLLOWERS, TYPE_FOLLOWING, TYPE_ITEM, TYPE_LIKE
SIGN_HEADERS, HS2019, SIGN_EXP, TYPE_FOLLOWERS, TYPE_FOLLOWING, TYPE_ITEM, TYPE_LIKE,
TYPE_REACTION, ST_AP_CACHE
)
from .regex import RE_SIG_PARAM
@ -124,7 +125,19 @@ class HTTPAPGServer(web_resource.Resource):
if node is None:
node = self.apg._m.namespace
client = await self.apg.getVirtualClient(signing_actor)
objects = await self.apg.apGetList(data, "object")
object_ = data.get("object")
if isinstance(object_, str):
# we check first if it's not a cached object
ap_cache_key = f"{ST_AP_CACHE}{object_}"
value = await self.apg.client._ap_storage.get(ap_cache_key)
else:
value = None
if value is not None:
objects = [value]
# because we'll undo the activity, we can remove it from cache
await self.apg.client._ap_storage.remove(ap_cache_key)
else:
objects = await self.apg.apGetList(data, "object")
for obj in objects:
type_ = obj.get("type")
actor = await self.apg.apGetSenderActor(obj)
@ -152,7 +165,11 @@ class HTTPAPGServer(web_resource.Resource):
# needed
await self.apg.newAPDeleteItem(client, None, node, obj)
elif type_ == TYPE_LIKE:
await self.handleNewLikeItem(client, obj, True)
await self.handleAttachmentItem(client, obj, {"noticed": False})
elif type_ == TYPE_REACTION:
await self.handleAttachmentItem(client, obj, {
"reactions": {"operation": "update", "remove": [obj["content"]]}
})
else:
log.warning(f"Unmanaged undo type: {type_!r}")
@ -184,7 +201,7 @@ class HTTPAPGServer(web_resource.Resource):
if subscription.state != "subscribed":
# other states should raise an Exception
raise exceptions.InternalError('"subscribed" state was expected')
inbox = await self.apg.getAPInboxFromId(signing_actor)
inbox = await self.apg.getAPInboxFromId(signing_actor, use_shared=False)
actor_id = self.apg.buildAPURL(TYPE_ACTOR, ap_account)
accept_data = self.apg.createActivity(
"Accept", actor_id, object_=data
@ -370,40 +387,47 @@ class HTTPAPGServer(web_resource.Resource):
repeated=True
)
async def handleNewLikeItem(
async def handleAttachmentItem(
self,
client: SatXMPPEntity,
data: dict,
undo: bool = False,
attachment_data: dict
) -> None:
liked_ids = data.get("object")
if not liked_ids:
target_ids = data.get("object")
if not target_ids:
raise exceptions.DataError("object should be set")
elif isinstance(liked_ids, list):
elif isinstance(target_ids, list):
try:
liked_ids = [o["id"] for o in liked_ids]
target_ids = [o["id"] for o in target_ids]
except (KeyError, TypeError):
raise exceptions.DataError(f"invalid object: {liked_ids!r}")
elif isinstance(liked_ids, dict):
obj_id = liked_ids.get("id")
raise exceptions.DataError(f"invalid object: {target_ids!r}")
elif isinstance(target_ids, dict):
obj_id = target_ids.get("id")
if not obj_id or not isinstance(obj_id, str):
raise exceptions.DataError(f"invalid object: {liked_ids!r}")
liked_ids = [obj_id]
elif isinstance(liked_ids, str):
liked_ids = [liked_ids]
raise exceptions.DataError(f"invalid object: {target_ids!r}")
target_ids = [obj_id]
elif isinstance(target_ids, str):
target_ids = [target_ids]
for liked_id in liked_ids:
if not self.apg.isLocalURL(liked_id):
log.debug(f"ignoring non local liked ID: {liked_id}")
# XXX: we have to cache AP items because some implementation (Pleroma notably)
# don't keep object accessible, and we need to be able to retrieve them for
# UNDO. Current implementation will grow, we need to add a way to flush it after
# a while.
# TODO: add a way to flush old cached AP items.
await client._ap_storage.aset(f"{ST_AP_CACHE}{data['id']}", data)
for target_id in target_ids:
if not self.apg.isLocalURL(target_id):
log.debug(f"ignoring non local target ID: {target_id}")
continue
url_type, url_args = self.apg.parseAPURL(liked_id)
url_type, url_args = self.apg.parseAPURL(target_id)
if url_type != TYPE_ITEM:
log.warning(f"unexpected local URL for liked item: {liked_id}")
log.warning(f"unexpected local URL for attachment on item {target_id}")
continue
try:
account, item_id = url_args
except ValueError:
raise ValueError(f"invalid URL: {liked_id}")
raise ValueError(f"invalid URL: {target_id}")
author_jid, item_node = await self.apg.getJIDAndNode(account)
if item_node is None:
item_node = self.apg._m.namespace
@ -418,7 +442,7 @@ class HTTPAPGServer(web_resource.Resource):
create=True
)
found_items, __ = await self.apg.host.memory.storage.getItems(
cached_node, item_ids=[item_id]
cached_node, item_ids=[client.jid.userhost()]
)
if not found_items:
old_item_elt = None
@ -428,9 +452,9 @@ class HTTPAPGServer(web_resource.Resource):
item_elt = self.apg._pa.applySetHandler(
client,
{"extra": {"noticed": not undo}},
{"extra": attachment_data},
old_item_elt,
[("noticed", self.apg._pa.namespace)]
None
)
# we reparse the element, as there can be other attachments
attachments_data = self.apg._pa.items2attachmentData(client, [item_elt])
@ -468,9 +492,24 @@ class HTTPAPGServer(web_resource.Resource):
ap_account: Optional[str],
ap_url: str,
signing_actor: str
):
) -> None:
client = await self.apg.getVirtualClient(signing_actor)
await self.handleNewLikeItem(client, data)
await self.handleAttachmentItem(client, data, {"noticed": True})
async def handleEmojireactActivity(
self,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
node: Optional[str],
ap_account: Optional[str],
ap_url: str,
signing_actor: str
) -> None:
client = await self.apg.getVirtualClient(signing_actor)
await self.handleAttachmentItem(client, data, {
"reactions": {"operation": "update", "add": [data["content"]]}
})
async def APActorRequest(
self,

2
sat/plugins/plugin_comp_ap_gateway/pubsub_service.py
View file

@ -95,7 +95,7 @@ class APPubsubService(rsm.PubSubService):
requestor_actor_id = self.apg.buildAPURL(TYPE_ACTOR, requestor.userhost())
recipient_account = self.apg._e.unescape(recipient.user)
recipient_actor_id = await self.apg.getAPActorIdFromAccount(recipient_account)
inbox = await self.apg.getAPInboxFromId(recipient_actor_id)
inbox = await self.apg.getAPInboxFromId(recipient_actor_id, use_shared=False)
return requestor_actor_id, recipient_actor_id, inbox