diff --git a/libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py b/libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py index e07bb268..bfb977e2 100644 --- a/libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py +++ b/libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py @@ -305,9 +305,7 @@ class APGateway: if not recipient.user: log.debug("ignoring items event without local part specified") return - ap_account = self._e.unescape(recipient.user) - if self._pa.is_attachment_node(itemsEvent.nodeIdentifier): await self.convert_and_post_attachments( client, @@ -323,6 +321,7 @@ class APGateway: itemsEvent.sender, itemsEvent.nodeIdentifier, itemsEvent.items, + itemsEvent.sender, ) async def get_virtual_client( @@ -1078,6 +1077,7 @@ class APGateway: service: jid.JID, node: str, items: List[domish.Element], + publisher: jid.JID, subscribe_extra_nodes: bool = True, ) -> None: """Convert XMPP items to AP items and post them to actor inbox @@ -1086,27 +1086,27 @@ class APGateway: @param service: JID of the (virtual) pubsub service where the item has been published @param node: (virtual) node corresponding where the item has been published + @param publisher: JID of the item publisher @param subscribe_extra_nodes: if True, extra data nodes will be automatically subscribed, that is comment nodes if present and attachments nodes. """ - actor_id = await self.get_ap_actor_id_from_account(ap_account) - requestor_actor_id = self.build_apurl( - TYPE_ACTOR, await self.get_ap_account_from_jid_and_node(service, node) - ) - inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id) + # The AP request is signed with the service/node's actor (the collection owner); + # the publisher is used for attributedTo (the item author). + try: + actor_id = await self.get_ap_actor_id_from_account(ap_account) + requestor_actor_id = self.build_apurl( + TYPE_ACTOR, + await self.get_ap_account_from_jid_and_node(service, node), + ) + inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id) + except exceptions.DataError as e: + log.warning(f"Can't resolve AP account {ap_account!r}, skipping items: {e}") + return for item in items: if item.name == "item": - cached_item = await self.host.memory.storage.search_pubsub_items( - { - "profiles": [self.client.profile], - "services": [service], - "nodes": [node], - "names": [item["id"]], - } - ) - is_new = not bool(cached_item) if node.startswith(self._events.namespace): # event item + # FIXME: detect event edits event_data = self._events.event_elt_2_event_data(item) try: author_jid = jid.JID(item["publisher"]).userhostJID() @@ -1136,24 +1136,26 @@ class APGateway: f"on {node!r} at {service}" ) ap_item = await self.ap_events.event_data_2_ap_item( - event_data, author_jid, is_new=is_new + event_data, author_jid ) else: # blog item mb_data = await MbData.from_element(client, item, service, node) + is_update = mb_data.is_edit + mb_data.author_jid = publisher.userhostJID() author_jid = mb_data.author_jid if subscribe_extra_nodes and not self.is_virtual_jid(author_jid): # we subscribe automatically to comment nodes if any recipient_jid = self.get_local_jid_from_account(ap_account) recipient_client = self.client.get_virtual_client(recipient_jid) - for comment_data in mb_data.get("comments", []): - comment_service = jid.JID(comment_data["service"]) - if self.is_virtual_jid(comment_service): - log.debug( - f"ignoring virtual comment service: {comment_data}" - ) + for comment in mb_data.comments: + comment_service = comment.service + if comment_service is None: continue - comment_node = comment_data["node"] + if self.is_virtual_jid(comment_service): + log.debug(f"ignoring virtual comment service: {comment}") + continue + comment_node = comment.node await self._p.subscribe( recipient_client, comment_service, comment_node ) @@ -1166,7 +1168,9 @@ class APGateway: f"no attachment node found for item {mb_data.id!r} on " f"{node!r} at {service}" ) - ap_item = await self.mb_data_2_ap_item(client, mb_data, is_new=is_new) + ap_item = await self.mb_data_2_ap_item( + client, mb_data, is_update=is_update + ) url_actor = ap_item["actor"] elif item.name == "retract": @@ -1207,12 +1211,6 @@ class APGateway: f"{len(items)})" ) - actor_id = await self.get_ap_actor_id_from_account(ap_account) - requestor_actor_id = self.build_apurl( - TYPE_ACTOR, await self.get_ap_account_from_jid_and_node(service, node) - ) - inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id) - item_elt = items[0] item_id = item_elt["id"] @@ -1234,6 +1232,12 @@ class APGateway: if publisher is not None: item_elt["publisher"] = publisher.userhost() + actor_id = await self.get_ap_actor_id_from_account(ap_account) + requestor_actor_id = self.build_apurl( + TYPE_ACTOR, await self.get_ap_account_from_jid_and_node(publisher, None) + ) + inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id) + item_service, item_node, item_id = self._pa.attachment_node_2_item(node) item_account = await self.get_ap_account_from_jid_and_node( item_service, item_node @@ -1877,6 +1881,16 @@ class APGateway: by=jid.JID(ap_item["_repeated"]["by"]), uri=ap_item["_repeated"]["uri"] ) + # tags + if "tag" in ap_object: + hashtags = [ + tag["name"].lstrip("#") + for tag in ap_object["tag"] + if tag.get("type") == "Hashtag" and "name" in tag + ] + if hashtags: + mb_data.tags = hashtags + # comments in_reply_to = ap_object.get("inReplyTo") __, comments_node = await self.get_comments_nodes( @@ -2011,7 +2025,7 @@ class APGateway: client: SatXMPPEntity, mb_data: MbData, public: bool = True, - is_new: bool = True, + is_update: bool = False, ) -> dict: """Convert Libervia Microblog Data to ActivityPub item @@ -2024,10 +2038,8 @@ class APGateway: ``inReplyTo`` will also be set if suitable if False, no destinee will be set (i.e., no ``to`` or ``cc`` or public flag). This is usually used for direct messages. - @param is_new: if True, the item is a new one (no instance has been found in - cache). - If True, a "Create" activity will be generated, otherwise an "Update" one will - be. + @param is_update: if True, generate an "Update" activity, otherwise a + "Create" one. @return: Activity item """ if mb_data.extra.repeated: @@ -2047,6 +2059,9 @@ class APGateway: "content": mb_data.content_xhtml or mb_data.content, } + if mb_data.is_edit: + ap_object["updated"] = utils.xmpp_date(mb_data.updated) + if mb_data.language: ap_object["contentMap"] = {mb_data.language: ap_object["content"]} @@ -2069,6 +2084,11 @@ class APGateway: ap_attachment[ap_key] = value ap_attachments.append(ap_attachment) + if mb_data.tags: + ap_object["tag"] = [ + {"type": "Hashtag", "name": f"#{tag}"} for tag in mb_data.tags + ] + if public: ap_object["to"] = [NS_AP_PUBLIC] if self.auto_mentions: @@ -2141,8 +2161,9 @@ class APGateway: client, ap_account, parent_item, mb_data ) + activity_type = "Update" if is_update else "Create" return self.create_activity( - "Create" if is_new else "Update", url_actor, ap_object, activity_id=url_item + activity_type, url_actor, ap_object, activity_id=url_item ) async def publish_message( @@ -2168,7 +2189,10 @@ class APGateway: if not service.user: raise ValueError("service must have a local part") account = self._e.unescape(service.user) - ap_actor_data = await self.get_ap_actor_data_from_account(account) + requestor_actor_id = self.build_apurl(TYPE_ACTOR, client.jid.userhost()) + ap_actor_data = await self.get_ap_actor_data_from_account( + requestor_actor_id, account + ) try: inbox_url = ap_actor_data["endpoints"]["sharedInbox"] @@ -2350,6 +2374,8 @@ class APGateway: message_elt: domish.Element, reference_data: Dict[str, Union[str, int]], ) -> bool: + if client != self.client: + return True parsed_uri: dict = reference_data.get("parsed_uri") if not parsed_uri: log.warning(f"no parsed URI available in reference {reference_data}") @@ -2462,6 +2488,11 @@ class APGateway: f"({in_reply_to!r}):\n{pformat(ap_item)}" ) return + # If the item ID is a URL-encoded full URI, extract the local ID (last path + # segment) so that get_items can find it in the pubsub node. + decoded_id = parse.unquote(parent_item_id) + if decoded_id.startswith("http://") or decoded_id.startswith("https://"): + parent_item_id = decoded_id.rsplit("/", 1)[-1] parent_item_service, parent_item_node = await self.get_jid_and_node( parent_item_account ) @@ -2701,11 +2732,10 @@ class APGateway: @param public: True if the item is public """ # XXX: "public" is not used for now + service = client.jid targets_urls = {t for t_set in targets.values() for t in t_set} requestor_actor_id = self.get_requestor_actor_id_from_targets(targets_urls) - service = client.jid in_reply_to = item.get("inReplyTo") - if in_reply_to and isinstance(in_reply_to, list): in_reply_to = in_reply_to[0] if in_reply_to and isinstance(in_reply_to, str): @@ -2730,9 +2760,8 @@ class APGateway: ) else: # it is a root item (i.e. not a reply to another item) - create = node == self._events.namespace cached_node = await self.host.memory.storage.get_pubsub_node( - client, service, node, with_subscriptions=True, create=create + client, service, node, with_subscriptions=True, create=True ) if cached_node is None: log.warning( @@ -2748,6 +2777,7 @@ class APGateway: data, item_elt = await self.ap_item_2_mb_data_and_elt( requestor_actor_id, item ) + data = data.model_dump(mode="json", exclude_none=True) await self.host.memory.storage.cache_pubsub_items( client, cached_node, [item_elt], [data] ) diff --git a/libervia/backend/plugins/plugin_comp_ap_gateway/http_server.py b/libervia/backend/plugins/plugin_comp_ap_gateway/http_server.py index 6b36e176..d3a9f27d 100644 --- a/libervia/backend/plugins/plugin_comp_ap_gateway/http_server.py +++ b/libervia/backend/plugins/plugin_comp_ap_gateway/http_server.py @@ -229,10 +229,12 @@ 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.get_ap_inbox_from_id(signing_actor, use_shared=False) actor_id = self.apg.build_apurl(TYPE_ACTOR, ap_account) + inbox = await self.apg.get_ap_inbox_from_id( + actor_id, signing_actor, use_shared=False + ) accept_data = self.apg.create_activity("Accept", actor_id, object_=data) - await self.apg.sign_and_post(inbox, actor_id, accept_data) + await self.apg.ap_post(inbox, actor_id, accept_data) await self.apg._c.synchronise(client, account_jid, node, resync=False) async def handle_accept_activity( @@ -588,6 +590,7 @@ class HTTPAPGServer(web_resource.Resource): async def ap_actor_request( self, + requestor_actor_id: str, request: "HTTPRequest", data: Optional[dict], account_jid: jid.JID, @@ -615,14 +618,15 @@ class HTTPAPGServer(web_resource.Resource): ) events = self.apg.build_apurl(TYPE_OUTBOX, events_account) + # XXX: Mastodon doesn't like percent-encoded at sign, so we have to unescape it + actor_id = ap_url.replace("%40", "@") + actor_data = { "@context": [ "https://www.w3.org/ns/activitystreams", "https://w3id.org/security/v1", ], - # XXX: Mastodon doesn't like percent-encode arobas, so we have to unescape it - # if it is escaped - "id": ap_url.replace("%40", "@"), + "id": actor_id, "type": "Person", "preferredUsername": preferred_username, "inbox": inbox, @@ -631,8 +635,8 @@ class HTTPAPGServer(web_resource.Resource): "followers": followers, "following": following, "publicKey": { - "id": f"{ap_url}#main-key", - "owner": ap_url, + "id": f"{actor_id}#main-key", + "owner": actor_id, "publicKeyPem": self.apg.public_key_pem, }, "endpoints": { @@ -641,17 +645,17 @@ class HTTPAPGServer(web_resource.Resource): }, } - if identity_data.get("nicknames"): - actor_data["name"] = identity_data["nicknames"][0] - if identity_data.get("description"): + if identity_data.nicknames: + actor_data["name"] = identity_data.nicknames[0] + if identity_data.description: # description is plain text while summary expects HTML - actor_data["summary"] = html.escape(identity_data["description"]) - if identity_data.get("avatar"): - avatar_data = identity_data["avatar"] + actor_data["summary"] = html.escape(identity_data.description) + if identity_data.avatar: + avatar_data = identity_data.avatar try: filename = avatar_data["filename"] media_type = avatar_data["media_type"] - except KeyError: + except (KeyError, TypeError): log.error(f"incomplete avatar data: {identity_data!r}") else: avatar_url = self.apg.build_apurl("avatar", filename) @@ -766,6 +770,7 @@ class HTTPAPGServer(web_resource.Resource): async def ap_outbox_request( self, + requestor_actor_id: str, request: "HTTPRequest", data: Optional[dict], account_jid: jid.JID, @@ -874,6 +879,7 @@ class HTTPAPGServer(web_resource.Resource): async def ap_followers_request( self, + requestor_actor_id: str, request: "HTTPRequest", data: Optional[dict], account_jid: jid.JID, @@ -915,6 +921,7 @@ class HTTPAPGServer(web_resource.Resource): async def ap_following_request( self, + requestor_actor_id: str, request: "HTTPRequest", data: Optional[dict], account_jid: jid.JID, diff --git a/libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py b/libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py index 43fcf66f..028c4ba2 100644 --- a/libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py +++ b/libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py @@ -16,7 +16,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from typing import Optional, Tuple, List, Dict, Any +from typing import TYPE_CHECKING, Optional, Tuple, List, Dict, Any from urllib.parse import urlparse from pathlib import Path from base64 import b64encode @@ -39,6 +39,9 @@ from libervia.backend.memory.sqla_mapping import PubsubSub, SubscriptionState from .constants import TYPE_ACTOR, ST_AVATAR, MAX_AVATAR_SIZE +if TYPE_CHECKING: + from . import APGateway + log = getLogger(__name__) # all nodes have the same config @@ -60,7 +63,7 @@ for c in NODE_CONFIG: class APPubsubService(rsm.PubSubService): """Pubsub service for XMPP requests""" - def __init__(self, apg): + def __init__(self, apg: "APGateway"): super(APPubsubService, self).__init__() self.host = apg.host self.apg = apg @@ -90,7 +93,9 @@ class APPubsubService(rsm.PubSubService): recipient_actor_id = await self.apg.get_ap_actor_id_from_account( recipient_account ) - inbox = await self.apg.get_ap_inbox_from_id(recipient_actor_id, use_shared=False) + inbox = await self.apg.get_ap_inbox_from_id( + requestor_actor_id, recipient_actor_id, use_shared=False + ) return requestor_actor_id, recipient_actor_id, inbox @ensure_deferred @@ -117,7 +122,12 @@ class APPubsubService(rsm.PubSubService): ) else: await self.apg.convert_and_post_items( - client, ap_account, service, nodeIdentifier, items + client, + ap_account, + service, + nodeIdentifier, + items, + publisher=requestor, ) cached_node = await self.host.memory.storage.get_pubsub_node( client, service, nodeIdentifier, with_subscriptions=True, create=True @@ -319,7 +329,7 @@ class APPubsubService(rsm.PubSubService): return [], None requestor_actor_id = self.apg.build_apurl( - TYPE_ACTOR, await self.apg.get_ap_account_from_jid_and_node(service, node) + TYPE_ACTOR, await self.apg.get_ap_account_from_jid_and_node(requestor, None) ) # cached_node may be pre-filled with some nodes (e.g. attachments nodes), @@ -534,7 +544,7 @@ class APPubsubService(rsm.PubSubService): data = self.apg.create_activity("Follow", req_actor_id, recip_actor_id) - resp = await self.apg.sign_and_post(inbox, req_actor_id, data) + resp = await self.apg.ap_post(inbox, req_actor_id, data) if resp.code >= 300: text = await resp.text() raise error.StanzaError("service-unavailable", text=text) @@ -542,6 +552,22 @@ class APPubsubService(rsm.PubSubService): @ensure_deferred async def unsubscribe(self, requestor, service, nodeIdentifier, subscriber): + client = self.apg.client + node = await self.host.memory.storage.get_pubsub_node( + client, service, nodeIdentifier, with_subscriptions=True + ) + if node is not None: + subscription = next( + ( + s + for s in node.subscriptions + if s.subscriber == requestor.userhostJID() + ), + None, + ) + if subscription is not None: + await self.host.memory.storage.delete(subscription) + req_actor_id, recip_actor_id, inbox = await self.get_ap_actor_ids_and_inbox( requestor, service ) @@ -551,7 +577,7 @@ class APPubsubService(rsm.PubSubService): self.apg.create_activity("Follow", req_actor_id, recip_actor_id), ) - resp = await self.apg.sign_and_post(inbox, req_actor_id, data) + resp = await self.apg.ap_post(inbox, req_actor_id, data) if resp.code >= 300: text = await resp.text() raise error.StanzaError("service-unavailable", text=text) diff --git a/libervia/backend/plugins/plugin_misc_identity.py b/libervia/backend/plugins/plugin_misc_identity.py index 2a512f0d..7cfba0bb 100644 --- a/libervia/backend/plugins/plugin_misc_identity.py +++ b/libervia/backend/plugins/plugin_misc_identity.py @@ -88,6 +88,7 @@ class IdentityData(BaseModel): nicknames: list[str] = Field( default_factory=list, description="Known nicknames of the entity." ) + description: str | None = Field(None, description="Description of the entity.") class Identities(RootModel): @@ -788,6 +789,8 @@ class IdentityPlugin: identity.avatar = value elif metadata_name == "nicknames": identity.nicknames = value or [] + elif metadata_name == "description": + identity.description = value if entity is not None: try: diff --git a/libervia/backend/plugins/plugin_xep_0277.py b/libervia/backend/plugins/plugin_xep_0277.py index 0c17f9a0..eef075d3 100644 --- a/libervia/backend/plugins/plugin_xep_0277.py +++ b/libervia/backend/plugins/plugin_xep_0277.py @@ -174,6 +174,15 @@ class MbData(BaseModel): extra: MbExtra = Field(default_factory=MbExtra) + @property + def is_edit(self) -> bool: + """Return True if the item has been edited.""" + return ( + self.updated is not None + and self.published is not None + and self.updated != self.published + ) + @classmethod async def from_element( cls, diff --git a/tests/unit/test_ap-gateway.py b/tests/unit/test_ap-gateway.py index e9d6cbd4..7058efe8 100644 --- a/tests/unit/test_ap-gateway.py +++ b/tests/unit/test_ap-gateway.py @@ -42,6 +42,7 @@ from libervia.backend.plugins import plugin_comp_ap_gateway from libervia.backend.plugins.plugin_comp_ap_gateway import constants as ap_const from libervia.backend.plugins.plugin_comp_ap_gateway import TYPE_ACTOR from libervia.backend.plugins.plugin_comp_ap_gateway.http_server import HTTPServer +from libervia.backend.plugins.plugin_misc_identity import IdentityData from libervia.backend.plugins.plugin_xep_0277 import NS_ATOM, MbData from libervia.backend.plugins.plugin_xep_0424 import NS_MESSAGE_RETRACT from libervia.backend.plugins.plugin_xep_0465 import NS_PPS @@ -671,6 +672,7 @@ class TestActivityPubGateway: ap_url = parse.urljoin(f"https://{ap_gateway.public_url}", path) kwargs = { + "requestor_actor_id": test_jid.full(), "request": request, "data": data, "account_jid": test_jid, @@ -701,8 +703,11 @@ class TestActivityPubGateway: assert outbox["first"] assert outbox["last"] + page_params = self.ap_request_params(ap_gateway, url=outbox["first"]) + # the page request doesn't take the requestor actor ID + page_params.pop("requestor_actor_id") first_page = await ap_gateway.server.resource.ap_outbox_page_request( - **self.ap_request_params(ap_gateway, url=outbox["first"]) + **page_params ) assert first_page["@context"] == ["https://www.w3.org/ns/activitystreams"] assert ( @@ -1134,10 +1139,10 @@ class TestActivityPubGateway: # XXX: XMPP identity is normally an amalgam of metadata from several # XEPs/locations (vCard4, vcard-tmp, etc) with patch.object(ap_gateway._i, "get_identity") as get_identity: - get_identity.return_value = { - "nicknames": ["nick1", "nick2"], - "description": "test description", - } + get_identity.return_value = IdentityData( + nicknames=["nick1", "nick2"], + description="test description", + ) actor_data = await ap_gateway.server.resource.ap_actor_request( **self.ap_request_params(ap_gateway, ap_const.TYPE_ACTOR) ) @@ -1388,6 +1393,8 @@ class TestActivityPubGateway: TEST_JID, ap_gateway._m.namespace, [item_elt], + TEST_JID, + subscribe_extra_nodes=False, ) assert ap_post.called @@ -1444,17 +1451,19 @@ class TestActivityPubGateway: assert cache_pubsub_items.called # the microblog data put in cache correspond to the item sent to subscribers __, __, __, [mb_data] = cache_pubsub_items.call_args.args - extra = mb_data.extra - repeated = extra.repeated + repeated = mb_data["extra"]["repeated"] assert repeated is not None - assert repeated.by == ap_gateway.get_local_jid_from_account(TEST_AP_ACCOUNT) + assert ( + repeated["by"] + == ap_gateway.get_local_jid_from_account(TEST_AP_ACCOUNT).full() + ) xmpp_item_xmpp_url = xmpp_uri.build_xmpp_uri( "pubsub", path=TEST_JID.full(), node=ap_gateway._m.namespace, item=xmpp_item["id"], ) - assert repeated.uri == xmpp_item_xmpp_url + assert repeated["uri"] == xmpp_item_xmpp_url @ed async def test_xmpp_attachment_noticed_to_ap_like(self, ap_gateway, monkeypatch):