diff --git a/sat/plugins/plugin_comp_ap_gateway.py b/sat/plugins/plugin_comp_ap_gateway/__init__.py similarity index 70% rename from sat/plugins/plugin_comp_ap_gateway.py rename to sat/plugins/plugin_comp_ap_gateway/__init__.py index abc9de50..3b270498 100644 --- a/sat/plugins/plugin_comp_ap_gateway.py +++ b/sat/plugins/plugin_comp_ap_gateway/__init__.py @@ -20,12 +20,10 @@ import base64 import hashlib import json from pathlib import Path -import time from typing import Optional, Dict, Tuple, List, Union from urllib import parse import calendar import re -import unicodedata import dateutil from cryptography.hazmat.primitives import serialization @@ -36,10 +34,10 @@ import shortuuid import treq from treq.response import _Response as TReqResponse from twisted.internet import defer, reactor, threads -from twisted.web import http, resource as web_resource, server +from twisted.web import http from twisted.words.protocols.jabber import jid, error from twisted.words.xish import domish -from wokkel import pubsub, rsm +from wokkel import rsm from sat.core import exceptions from sat.core.constants import Const as C @@ -49,7 +47,11 @@ from sat.core.log import getLogger from sat.tools import utils from sat.tools.common import data_format, tls from sat.tools.common.async_utils import async_lru -from sat.tools.utils import ensure_deferred + +from .constants import (IMPORT_NAME, CONF_SECTION, TYPE_ACTOR, TYPE_ITEM, MEDIA_TYPE_AP, + AP_MB_MAP, LRU_MAX_SIZE) +from .http_server import HTTPServer +from .pubsub_service import APPubsubService log = getLogger(__name__) @@ -71,285 +73,10 @@ PLUGIN_INFO = { ), } -VERSION = unicodedata.normalize( - 'NFKD', - f"{C.APP_NAME} ActivityPub Gateway {C.APP_VERSION}" -) HEXA_ENC = r"(?P[0-9a-fA-f]{2})" RE_PERIOD_ENC = re.compile(f"\\.{HEXA_ENC}") RE_PERCENT_ENC = re.compile(f"%{HEXA_ENC}") RE_ALLOWED_UNQUOTED = re.compile(r"^[a-zA-Z0-9_-]+$") -CONF_SECTION = f"component {IMPORT_NAME}" -CONTENT_TYPE_AP = "application/activity+json; charset=utf-8" -TYPE_ACTOR = "actor" -TYPE_INBOX = "inbox" -TYPE_OUTBOX = "outbox" -TYPE_ITEM = "item" -MEDIA_TYPE_AP = "application/activity+json" -# mapping from AP metadata to microblog data -AP_MB_MAP = { - "content": "content_xhtml", - -} -AP_REQUEST_TYPES = {"actor", "outbox"} -PAGE_SIZE = 10 - -LRU_MAX_SIZE = 200 - - -class HTTPAPGServer(web_resource.Resource): - """HTTP Server handling ActivityPub S2S protocol""" - isLeaf = True - - def __init__(self, ap_gateway): - self.apg = ap_gateway - super().__init__() - - async def webfinger(self, request): - url_parsed = parse.urlparse(request.uri.decode()) - query = parse.parse_qs(url_parsed.query) - resource = query.get("resource", [""])[0] - account = resource[5:].strip() - if not resource.startswith("acct:") or not account: - return web_resource.ErrorPage( - http.BAD_REQUEST, "Bad Request" , "Invalid webfinger resource" - ).render(request) - - actor_url = self.apg.buildAPURL(TYPE_ACTOR, account) - - resp = { - "subject": resource, - "links": [ - { - "rel": "self", - "type": "application/activity+json", - "href": actor_url - } - ] - } - request.setHeader("content-type", CONTENT_TYPE_AP) - request.write(json.dumps(resp).encode()) - request.finish() - - async def APActorRequest( - self, - request: "HTTPRequest", - account_jid: jid.JID, - node: Optional[str], - ap_account: str, - actor_url: str - ) -> dict: - inbox_url = self.apg.buildAPURL(TYPE_INBOX, ap_account) - outbox_url = self.apg.buildAPURL(TYPE_OUTBOX, ap_account) - - # we have to use AP account as preferredUsername because it is used to retrieve - # actor handle (see https://socialhub.activitypub.rocks/t/how-to-retrieve-user-server-tld-handle-from-actors-url/2196) - preferred_username = ap_account.split("@", 1)[0] - return { - "@context": [ - "https://www.w3.org/ns/activitystreams", - "https://w3id.org/security/v1" - ], - - "id": actor_url, - "type": "Person", - "preferredUsername": preferred_username, - "inbox": inbox_url, - "outbox": outbox_url, - "publicKey": { - "id": f"{actor_url}#main-key", - "owner": actor_url, - "publicKeyPem": self.apg.public_key_pem - } - } - - def getCanonicalURL(self, request: "HTTPRequest") -> str: - return parse.urljoin( - f"https://{self.apg.public_url}", - request.path.decode().rstrip("/") - ) - - def queryData2RSMRequest( - self, - query_data: Dict[str, List[str]] - ) -> rsm.RSMRequest: - """Get RSM kwargs to use with RSMRequest from query data""" - page = query_data.get("page") - - if page == ["first"]: - return rsm.RSMRequest(max_=PAGE_SIZE, before="") - elif page == ["last"]: - return rsm.RSMRequest(max_=PAGE_SIZE) - else: - for query_key in ("index", "before", "after"): - try: - kwargs={query_key: query_data[query_key][0], "max_": PAGE_SIZE} - except (KeyError, IndexError, ValueError): - pass - else: - return rsm.RSMRequest(**kwargs) - raise ValueError(f"Invalid query data: {query_data!r}") - - async def APOutboxPageRequest( - self, - request: "HTTPRequest", - account_jid: jid.JID, - node: Optional[str], - ap_account: str, - actor_url: str, - query_data: Dict[str, List[str]] - ) -> dict: - # we only keep useful keys, and sort to have consistent URL which can - # be used as ID - url_keys = sorted(set(query_data) & {"page", "index", "before", "after"}) - query_data = {k: query_data[k] for k in url_keys} - rsm_kwargs = self.queryData2RSMRequest(query_data) - try: - items, metadata = await self.apg._p.getItems( - client=self.apg.client, - service=account_jid, - node=node, - rsm_request=self.queryData2RSMRequest(query_data), - extra = {C.KEY_USE_CACHE: False} - ) - except error.StanzaError as e: - log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") - return {} - - base_url = self.getCanonicalURL(request) - url = f"{base_url}?{parse.urlencode(query_data, True)}" - data = { - "@context": "https://www.w3.org/ns/activitystreams", - "id": url, - "type": "OrderedCollectionPage", - "partOf": base_url, - "orderedItems" : [ - await self.apg.mbdata2APitem( - self.apg.client, - await self.apg._m.item2mbdata( - self.apg.client, - item, - account_jid, - node - ) - ) - for item in reversed(items) - ] - } - - # AP OrderedCollection must be in reversed chronological order, thus the opposite - # of what we get with RSM (at least with Libervia Pubsub) - if not metadata["complete"]: - try: - last= metadata["rsm"]["last"] - except KeyError: - last = None - data["prev"] = f"{base_url}?{parse.urlencode({'after': last})}" - if metadata["rsm"]["index"] != 0: - try: - first= metadata["rsm"]["first"] - except KeyError: - first = None - data["next"] = f"{base_url}?{parse.urlencode({'before': first})}" - - return data - - async def APOutboxRequest( - self, - request: "HTTPRequest", - account_jid: jid.JID, - node: Optional[str], - ap_account: str, - actor_url: str - ) -> dict: - if node is None: - node = self.apg._m.namespace - - parsed_url = parse.urlparse(request.uri.decode()) - query_data = parse.parse_qs(parsed_url.query) - if query_data: - return await self.APOutboxPageRequest( - request, account_jid, node, ap_account, actor_url, query_data - ) - - # XXX: we can't use disco#info here because this request won't work on a bare jid - # due to security considerations of XEP-0030 (we don't have presence - # subscription). - # The current workaround is to do a request as if RSM was available, and actually - # check its availability according to result. - try: - __, metadata = await self.apg._p.getItems( - client=self.apg.client, - service=account_jid, - node=node, - max_items=0, - rsm_request=rsm.RSMRequest(max_=0) - ) - except error.StanzaError as e: - log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") - return {} - try: - items_count = metadata["rsm"]["count"] - except KeyError: - log.warning( - f"No RSM metadata found when requesting pubsub node {node} at " - f"{account_jid}, defaulting to items_count=20" - ) - items_count = 20 - - url = self.getCanonicalURL(request) - url_first_page = f"{url}?{parse.urlencode({'page': 'first'})}" - url_last_page = f"{url}?{parse.urlencode({'page': 'last'})}" - return { - "@context": "https://www.w3.org/ns/activitystreams", - "id": url, - "totalItems": items_count, - "type": "OrderedCollection", - "first": url_first_page, - "last": url_last_page, - } - - async def APRequest(self, request): - path = request.path.decode() - actor_url = parse.urljoin( - f"https://{self.apg.public_url}", - path - ) - request_type, ap_account = self.apg.parseAPURL(actor_url) - account_jid, node = await self.apg.getJIDAndNode(ap_account) - if request_type not in AP_REQUEST_TYPES: - raise exceptions.DataError(f"Invalid request type: {request_type!r}") - method = getattr(self, f"AP{request_type.title()}Request") - ret_data = await method(request, account_jid, node, ap_account, actor_url) - request.setHeader("content-type", CONTENT_TYPE_AP) - request.write(json.dumps(ret_data).encode()) - request.finish() - - def render(self, request): - request.setHeader("server", VERSION) - return super().render(request) - - def render_GET(self, request): - path = request.path.decode().lstrip("/") - if path.startswith(".well-known/webfinger"): - defer.ensureDeferred(self.webfinger(request)) - return server.NOT_DONE_YET - elif path.startswith(self.apg.ap_path): - defer.ensureDeferred(self.APRequest(request)) - return server.NOT_DONE_YET - - return web_resource.NoResource().render(request) - - -class HTTPRequest(server.Request): - pass - - -class HTTPServer(server.Site): - requestFactory = HTTPRequest - - def __init__(self, ap_gateway): - super().__init__(HTTPAPGServer(ap_gateway)) class APGateway: @@ -519,7 +246,7 @@ class APGateway: node = None if node and not jid_.user and not self.mustEncode(node): - is_pubsub = self.isPubsub(jid_) + is_pubsub = await self.isPubsub(jid_) # when we have a pubsub service, the user part can be used to set the node # this produces more user-friendly AP accounts if is_pubsub: @@ -582,18 +309,19 @@ class APGateway: This horror is necessary due to limitation in some AP implementation (notably Mastodon), cf. https://github.com/mastodon/mastodon/issues/17222 - examples:: + examples: ``toto@example.org`` => JID = toto@example.org, node = None - ``___toto.40example.net@example.org`` => JID = toto@example.net (this one is a non-local JID, and will work only if setings ``local_only`` is False), node = None + ``___toto.40example.net@example.org`` => JID = toto@example.net (this one is a + non-local JID, and will work only if setings ``local_only`` is False), node = None ``toto@pubsub.example.org`` (with pubsub.example.org being a pubsub service) => JID = pubsub.example.org, node = toto ``tata---toto@example.org`` => JID = toto@example.org, node = tata - ``___urn.3axmpp.3a.microblog.3a0@pubsub.example.org`` (with pubsub.example.org + ``___urn.3axmpp.3amicroblog.3a0@pubsub.example.org`` (with pubsub.example.org being a pubsub service) ==> JID = pubsub.example.org, node = urn:xmpp:microblog:0 @param ap_account: ActivityPub account handle (``username@domain.tld``) @@ -799,7 +527,7 @@ class APGateway: ) -> Tuple[List[domish.Element], rsm.RSMResponse]: """Retrieve AP items and convert them to XMPP items - @param account: AP account to get items from + @param account: AP account handle to get items from @param max_items: maximum number of items to retrieve retrieve all items by default @param chronological_pagination: get pages in chronological order @@ -893,19 +621,20 @@ class APGateway: if after_id is not None and not found_after_id: # if we have an after_id, we ignore all items until the requested one is # found - limit_idx = [i["id"] for i in page_items].index(after_id) - if limit_idx == -1: + try: + limit_idx = [i["id"] for i in page_items].index(after_id) + except ValueError: # if "after_id" is not found, we don't add any item from this page log.debug(f"{after_id!r} not found at {page}, skipping") else: found_after_id = True if chronological_pagination: - page_items = page_items[limit_idx+1:] start_index = retrieved_items - len(page_items) + limit_idx + 1 + page_items = page_items[limit_idx+1:] else: - page_items = page_items[:limit_idx] start_index = count - (retrieved_items - len(page_items) + limit_idx + 1) + page_items = page_items[:limit_idx] items.extend(page_items) else: items.extend(page_items) @@ -1122,102 +851,3 @@ class APGateway: resp = await self.signAndPost(inbox_url, url_actor, item_data) if resp.code != 202: raise exceptions.NetworkError(f"unexpected return code: {resp.code}") - - -class APPubsubService(rsm.PubSubService): - """Pubsub service for XMPP requests""" - - def __init__(self, apg): - super(APPubsubService, self).__init__() - self.host = apg.host - self.apg = apg - self.discoIdentity = { - "category": "pubsub", - "type": "service", - "name": "Libervia ActivityPub Gateway", - } - - @ensure_deferred - async def publish(self, requestor, service, nodeIdentifier, items): - raise NotImplementedError - - @ensure_deferred - async def items( - self, - requestor: jid.JID, - service: jid.JID, - node: str, - maxItems: Optional[int], - itemIdentifiers: Optional[List[str]], - rsm_req: Optional[rsm.RSMRequest] - ) -> List[domish.Element]: - if not service.user: - return [] - ap_account = self.host.plugins["XEP-0106"].unescape(service.user) - if ap_account.count("@") != 1: - log.warning(f"Invalid AP account used by {requestor}: {ap_account!r}") - return [] - if node != self.apg._m.namespace: - raise error.StanzaError( - "feature-not-implemented", - text=f"{VERSION} only supports {self.apg._m.namespace} " - "node for now" - ) - if rsm_req is None: - if maxItems is None: - maxItems = 20 - kwargs = { - "max_items": maxItems, - "chronological_pagination": False, - } - else: - if len( - [v for v in (rsm_req.after, rsm_req.before, rsm_req.index) - if v is not None] - ) > 1: - raise error.StanzaError( - "bad-request", - text="You can't use after, before and index at the same time" - ) - kwargs = {"max_items": rsm_req.max} - if rsm_req.after is not None: - kwargs["after_id"] = rsm_req.after - elif rsm_req.before is not None: - kwargs["chronological_pagination"] = False - if rsm_req.before != "": - kwargs["after_id"] = rsm_req.before - elif rsm_req.index is not None: - kwargs["start_index"] = rsm_req.index - - log.info( - f"No cache found for node {node} at {service} (AP account {ap_account}), " - "using Collection Paging to RSM translation" - ) - return await self.apg.getAPItems(ap_account, **kwargs) - - @ensure_deferred - async def retract(self, requestor, service, nodeIdentifier, itemIdentifiers): - raise NotImplementedError - - def getNodeInfo( - self, - requestor: jid.JID, - service: jid.JID, - nodeIdentifier: str, - pep: bool = False, - recipient: Optional[jid.JID] = None - ) -> Optional[dict]: - if not nodeIdentifier: - return None - info = { - "type": "leaf", - "meta-data": [ - {"var": "pubsub#persist_items", "type": "boolean", "value": True}, - {"var": "pubsub#max_items", "value": "max"}, - {"var": "pubsub#access_model", "type": "list-single", "value": "open"}, - {"var": "pubsub#publish_model", "type": "list-single", "value": "open"}, - - ] - - } - return info diff --git a/sat/plugins/plugin_comp_ap_gateway/constants.py b/sat/plugins/plugin_comp_ap_gateway/constants.py new file mode 100644 index 00000000..575b40bc --- /dev/null +++ b/sat/plugins/plugin_comp_ap_gateway/constants.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 + +# Libervia ActivityPub Gateway +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + + +IMPORT_NAME = "ap-gateway" +CONF_SECTION = f"component {IMPORT_NAME}" +CONTENT_TYPE_AP = "application/activity+json; charset=utf-8" +TYPE_ACTOR = "actor" +TYPE_INBOX = "inbox" +TYPE_OUTBOX = "outbox" +TYPE_ITEM = "item" +MEDIA_TYPE_AP = "application/activity+json" +# mapping from AP metadata to microblog data +AP_MB_MAP = { + "content": "content_xhtml", + +} +AP_REQUEST_TYPES = {"actor", "outbox"} +PAGE_SIZE = 10 + +LRU_MAX_SIZE = 200 diff --git a/sat/plugins/plugin_comp_ap_gateway/http_server.py b/sat/plugins/plugin_comp_ap_gateway/http_server.py new file mode 100644 index 00000000..771acc21 --- /dev/null +++ b/sat/plugins/plugin_comp_ap_gateway/http_server.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 + +# Libervia ActivityPub Gateway +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from typing import Optional, Dict, List +import json +from urllib import parse +import re +import unicodedata + +from twisted.web import http, resource as web_resource, server +from twisted.internet import defer +from twisted.words.protocols.jabber import jid, error +from wokkel import pubsub, rsm + +from sat.core import exceptions +from sat.core.constants import Const as C +from sat.core.i18n import _ +from sat.core.log import getLogger + +from .constants import (CONTENT_TYPE_AP, TYPE_ACTOR, TYPE_INBOX, TYPE_OUTBOX, + AP_REQUEST_TYPES, PAGE_SIZE) + + +log = getLogger(__name__) + +VERSION = unicodedata.normalize( + 'NFKD', + f"{C.APP_NAME} ActivityPub Gateway {C.APP_VERSION}" +) + + +class HTTPAPGServer(web_resource.Resource): + """HTTP Server handling ActivityPub S2S protocol""" + isLeaf = True + + def __init__(self, ap_gateway): + self.apg = ap_gateway + super().__init__() + + async def webfinger(self, request): + url_parsed = parse.urlparse(request.uri.decode()) + query = parse.parse_qs(url_parsed.query) + resource = query.get("resource", [""])[0] + account = resource[5:].strip() + if not resource.startswith("acct:") or not account: + return web_resource.ErrorPage( + http.BAD_REQUEST, "Bad Request" , "Invalid webfinger resource" + ).render(request) + + actor_url = self.apg.buildAPURL(TYPE_ACTOR, account) + + resp = { + "subject": resource, + "links": [ + { + "rel": "self", + "type": "application/activity+json", + "href": actor_url + } + ] + } + request.setHeader("content-type", CONTENT_TYPE_AP) + request.write(json.dumps(resp).encode()) + request.finish() + + async def APActorRequest( + self, + request: "HTTPRequest", + account_jid: jid.JID, + node: Optional[str], + ap_account: str, + actor_url: str + ) -> dict: + inbox_url = self.apg.buildAPURL(TYPE_INBOX, ap_account) + outbox_url = self.apg.buildAPURL(TYPE_OUTBOX, ap_account) + + # we have to use AP account as preferredUsername because it is used to retrieve + # actor handle (see https://socialhub.activitypub.rocks/t/how-to-retrieve-user-server-tld-handle-from-actors-url/2196) + preferred_username = ap_account.split("@", 1)[0] + return { + "@context": [ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1" + ], + + "id": actor_url, + "type": "Person", + "preferredUsername": preferred_username, + "inbox": inbox_url, + "outbox": outbox_url, + "publicKey": { + "id": f"{actor_url}#main-key", + "owner": actor_url, + "publicKeyPem": self.apg.public_key_pem + } + } + + def getCanonicalURL(self, request: "HTTPRequest") -> str: + return parse.urljoin( + f"https://{self.apg.public_url}", + request.path.decode().rstrip("/") + ) + + def queryData2RSMRequest( + self, + query_data: Dict[str, List[str]] + ) -> rsm.RSMRequest: + """Get RSM kwargs to use with RSMRequest from query data""" + page = query_data.get("page") + + if page == ["first"]: + return rsm.RSMRequest(max_=PAGE_SIZE, before="") + elif page == ["last"]: + return rsm.RSMRequest(max_=PAGE_SIZE) + else: + for query_key in ("index", "before", "after"): + try: + kwargs={query_key: query_data[query_key][0], "max_": PAGE_SIZE} + except (KeyError, IndexError, ValueError): + pass + else: + return rsm.RSMRequest(**kwargs) + raise ValueError(f"Invalid query data: {query_data!r}") + + async def APOutboxPageRequest( + self, + request: "HTTPRequest", + account_jid: jid.JID, + node: Optional[str], + ap_account: str, + ap_url: str, + query_data: Dict[str, List[str]] + ) -> dict: + # we only keep useful keys, and sort to have consistent URL which can + # be used as ID + url_keys = sorted(set(query_data) & {"page", "index", "before", "after"}) + query_data = {k: query_data[k] for k in url_keys} + try: + items, metadata = await self.apg._p.getItems( + client=self.apg.client, + service=account_jid, + node=node, + rsm_request=self.queryData2RSMRequest(query_data), + extra = {C.KEY_USE_CACHE: False} + ) + except error.StanzaError as e: + log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") + return {} + + base_url = self.getCanonicalURL(request) + url = f"{base_url}?{parse.urlencode(query_data, True)}" + data = { + "@context": "https://www.w3.org/ns/activitystreams", + "id": url, + "type": "OrderedCollectionPage", + "partOf": base_url, + "orderedItems" : [ + await self.apg.mbdata2APitem( + self.apg.client, + await self.apg._m.item2mbdata( + self.apg.client, + item, + account_jid, + node + ) + ) + for item in reversed(items) + ] + } + + # AP OrderedCollection must be in reversed chronological order, thus the opposite + # of what we get with RSM (at least with Libervia Pubsub) + if not metadata["complete"]: + try: + last= metadata["rsm"]["last"] + except KeyError: + last = None + data["prev"] = f"{base_url}?{parse.urlencode({'after': last})}" + if metadata["rsm"]["index"] != 0: + try: + first= metadata["rsm"]["first"] + except KeyError: + first = None + data["next"] = f"{base_url}?{parse.urlencode({'before': first})}" + + return data + + async def APOutboxRequest( + self, + request: "HTTPRequest", + account_jid: jid.JID, + node: Optional[str], + ap_account: str, + ap_url: str + ) -> dict: + if node is None: + node = self.apg._m.namespace + + parsed_url = parse.urlparse(request.uri.decode()) + query_data = parse.parse_qs(parsed_url.query) + if query_data: + return await self.APOutboxPageRequest( + request, account_jid, node, ap_account, ap_url, query_data + ) + + # XXX: we can't use disco#info here because this request won't work on a bare jid + # due to security considerations of XEP-0030 (we don't have presence + # subscription). + # The current workaround is to do a request as if RSM was available, and actually + # check its availability according to result. + try: + __, metadata = await self.apg._p.getItems( + client=self.apg.client, + service=account_jid, + node=node, + max_items=0, + rsm_request=rsm.RSMRequest(max_=0) + ) + except error.StanzaError as e: + log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") + return {} + try: + items_count = metadata["rsm"]["count"] + except KeyError: + log.warning( + f"No RSM metadata found when requesting pubsub node {node} at " + f"{account_jid}, defaulting to items_count=20" + ) + items_count = 20 + + url = self.getCanonicalURL(request) + url_first_page = f"{url}?{parse.urlencode({'page': 'first'})}" + url_last_page = f"{url}?{parse.urlencode({'page': 'last'})}" + return { + "@context": "https://www.w3.org/ns/activitystreams", + "id": url, + "totalItems": items_count, + "type": "OrderedCollection", + "first": url_first_page, + "last": url_last_page, + } + + async def APRequest(self, request): + path = request.path.decode() + ap_url = parse.urljoin( + f"https://{self.apg.public_url}", + path + ) + request_type, ap_account = self.apg.parseAPURL(ap_url) + account_jid, node = await self.apg.getJIDAndNode(ap_account) + if request_type not in AP_REQUEST_TYPES: + raise exceptions.DataError(f"Invalid request type: {request_type!r}") + method = getattr(self, f"AP{request_type.title()}Request") + ret_data = await method(request, account_jid, node, ap_account, ap_url) + request.setHeader("content-type", CONTENT_TYPE_AP) + request.write(json.dumps(ret_data).encode()) + request.finish() + + def render(self, request): + request.setHeader("server", VERSION) + return super().render(request) + + def render_GET(self, request): + path = request.path.decode().lstrip("/") + if path.startswith(".well-known/webfinger"): + defer.ensureDeferred(self.webfinger(request)) + return server.NOT_DONE_YET + elif path.startswith(self.apg.ap_path): + defer.ensureDeferred(self.APRequest(request)) + return server.NOT_DONE_YET + + return web_resource.NoResource().render(request) + + +class HTTPRequest(server.Request): + pass + + +class HTTPServer(server.Site): + requestFactory = HTTPRequest + + def __init__(self, ap_gateway): + super().__init__(HTTPAPGServer(ap_gateway)) diff --git a/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py b/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py new file mode 100644 index 00000000..cbc16a02 --- /dev/null +++ b/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 + +# Libervia ActivityPub Gateway +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +from typing import Optional, List + +from twisted.words.protocols.jabber import jid, error +from twisted.words.xish import domish +from wokkel import rsm + +from sat.core.i18n import _ +from sat.core.log import getLogger +from sat.tools.utils import ensure_deferred + + +log = getLogger(__name__) + + +class APPubsubService(rsm.PubSubService): + """Pubsub service for XMPP requests""" + + def __init__(self, apg): + super(APPubsubService, self).__init__() + self.host = apg.host + self.apg = apg + self.discoIdentity = { + "category": "pubsub", + "type": "service", + "name": "Libervia ActivityPub Gateway", + } + + @ensure_deferred + async def publish(self, requestor, service, nodeIdentifier, items): + raise NotImplementedError + + @ensure_deferred + async def items( + self, + requestor: jid.JID, + service: jid.JID, + node: str, + maxItems: Optional[int], + itemIdentifiers: Optional[List[str]], + rsm_req: Optional[rsm.RSMRequest] + ) -> List[domish.Element]: + if not service.user: + return [] + ap_account = self.host.plugins["XEP-0106"].unescape(service.user) + if ap_account.count("@") != 1: + log.warning(f"Invalid AP account used by {requestor}: {ap_account!r}") + return [] + if node != self.apg._m.namespace: + raise error.StanzaError( + "feature-not-implemented", + text=f"{VERSION} only supports {self.apg._m.namespace} " + "node for now" + ) + if rsm_req is None: + if maxItems is None: + maxItems = 20 + kwargs = { + "max_items": maxItems, + "chronological_pagination": False, + } + else: + if len( + [v for v in (rsm_req.after, rsm_req.before, rsm_req.index) + if v is not None] + ) > 1: + raise error.StanzaError( + "bad-request", + text="You can't use after, before and index at the same time" + ) + kwargs = {"max_items": rsm_req.max} + if rsm_req.after is not None: + kwargs["after_id"] = rsm_req.after + elif rsm_req.before is not None: + kwargs["chronological_pagination"] = False + if rsm_req.before != "": + kwargs["after_id"] = rsm_req.before + elif rsm_req.index is not None: + kwargs["start_index"] = rsm_req.index + + log.info( + f"No cache found for node {node} at {service} (AP account {ap_account}), " + "using Collection Paging to RSM translation" + ) + return await self.apg.getAPItems(ap_account, **kwargs) + + @ensure_deferred + async def retract(self, requestor, service, nodeIdentifier, itemIdentifiers): + raise NotImplementedError + + def getNodeInfo( + self, + requestor: jid.JID, + service: jid.JID, + nodeIdentifier: str, + pep: bool = False, + recipient: Optional[jid.JID] = None + ) -> Optional[dict]: + if not nodeIdentifier: + return None + info = { + "type": "leaf", + "meta-data": [ + {"var": "pubsub#persist_items", "type": "boolean", "value": True}, + {"var": "pubsub#max_items", "value": "max"}, + {"var": "pubsub#access_model", "type": "list-single", "value": "open"}, + {"var": "pubsub#publish_model", "type": "list-single", "value": "open"}, + + ] + + } + return info