comp AP gateway: signature checking, caching and threads management:

- HTTP signature is checked for incoming messages
- AP actor can now be followed using pubsub subscription. When following is accepted, the
  node is cached
- replies to posts are put in cached pubsub comment nodes, with a `comments_max_depth`
  option to limit the number of comment nodes for a root message (documentation will come
  to explain this).

ticket 364

sat/core/exceptions.py

@@ -123,6 +123,11 @@ class ParsingError(Exception):
     pass
 
 
+class EncryptionError(Exception):
+    """Invalid encryption"""
+    pass
+
+
 # Something which need to be done is not available yet
 class NotReady(Exception):
     pass

sat/plugins/plugin_comp_ap_gateway/__init__.py

@@ -17,25 +17,28 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import base64
+import calendar
 import hashlib
 import json
 from pathlib import Path
-from typing import Optional, Dict, Tuple, List, Union
+from pprint import pformat
-from urllib import parse
-import calendar
 import re
+from typing import Any, Dict, List, Optional, Tuple, Union, overload
+from urllib import parse
 
-import dateutil
+from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.asymmetric import padding
+import dateutil
 import shortuuid
+from sqlalchemy.exc import IntegrityError
 import treq
 from treq.response import _Response as TReqResponse
 from twisted.internet import defer, reactor, threads
 from twisted.web import http
-from twisted.words.protocols.jabber import jid, error
+from twisted.words.protocols.jabber import error, jid
 from twisted.words.xish import domish
 from wokkel import rsm
 
@@ -44,12 +47,25 @@ from sat.core.constants import Const as C
 from sat.core.core_types import SatXMPPEntity
 from sat.core.i18n import _
 from sat.core.log import getLogger
+from sat.memory.sqla_mapping import PubsubSub, SubscriptionState
 from sat.tools import utils
-from sat.tools.common import data_format, tls
+from sat.tools.common import data_format, tls, uri
 from sat.tools.common.async_utils import async_lru
 
-from .constants import (IMPORT_NAME, CONF_SECTION, TYPE_ACTOR, TYPE_ITEM, MEDIA_TYPE_AP,
+from .constants import (
-                       AP_MB_MAP, LRU_MAX_SIZE)
+    ACTIVITY_OBJECT_MANDATORY,
+    ACTIVITY_TARGET_MANDATORY,
+    ACTIVITY_TYPES,
+    ACTIVITY_TYPES_LOWER,
+    AP_MB_MAP,
+    COMMENTS_MAX_PARENTS,
+    CONF_SECTION,
+    IMPORT_NAME,
+    LRU_MAX_SIZE,
+    MEDIA_TYPE_AP,
+    TYPE_ACTOR,
+    TYPE_ITEM,
+)
 from .http_server import HTTPServer
 from .pubsub_service import APPubsubService
 
@@ -64,7 +80,7 @@ PLUGIN_INFO = {
     C.PI_MODES: [C.PLUG_MODE_COMPONENT],
     C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT,
     C.PI_PROTOCOLS: [],
-    C.PI_DEPENDENCIES: ["XEP-0106", "XEP-0277", "XEP-0060"],
+    C.PI_DEPENDENCIES: ["XEP-0106", "XEP-0277", "XEP-0060", "PUBSUB_CACHE"],
     C.PI_RECOMMENDATIONS: [],
     C.PI_MAIN: "APGateway",
     C.PI_HANDLER: C.BOOL_TRUE,
@@ -86,6 +102,9 @@ class APGateway:
         self.initialised = False
         self._m = host.plugins["XEP-0277"]
         self._p = host.plugins["XEP-0060"]
+        self._e = host.plugins["XEP-0106"]
+        self._c = host.plugins["PUBSUB_CACHE"]
+        self.pubsub_service = APPubsubService(self)
 
         host.bridge.addMethod(
             "APSend",
@@ -97,7 +116,7 @@ class APGateway:
         )
 
     def getHandler(self, __):
-        return APPubsubService(self)
+        return self.pubsub_service
 
     async def init(self, client):
         if self.initialised:
@@ -165,10 +184,13 @@ class APGateway:
                 'bad ap-gateay http_connection_type, you must use one of "http" or '
                 '"https"'
             )
-        self.max_items = self.host.memory.getConfig(
+        self.max_items = int(self.host.memory.getConfig(
             CONF_SECTION, 'new_node_max_items', 50
 
-        )
+        ))
+        self.comments_max_depth = int(self.host.memory.getConfig(
+            CONF_SECTION, 'comments_max_depth', 0
+        ))
         self.ap_path = self.host.memory.getConfig(CONF_SECTION, 'ap_path', '_ap')
         self.base_ap_url = parse.urljoin(f"https://{self.public_url}", f"{self.ap_path}/")
         # True (default) if we provide gateway only to entities/services from our server
@@ -191,6 +213,25 @@ class APGateway:
         self.client = client
         await self.init(client)
 
+    async def getVirtualClient(self, actor_id: str) -> SatXMPPEntity:
+        """Get client for this component with a specified jid
+
+        This is needed to perform operations with the virtual JID corresponding to the AP
+        actor instead of the JID of the gateway itself.
+        @param actor_id: ID of the actor
+        @return: virtual client
+        """
+        account = await self.getAPAccountFromId(actor_id)
+        local_jid = self.getLocalJIDFromAccount(account)
+        return self.client.getVirtualClient(local_jid)
+
+    def isActivity(self, data: dict) -> bool:
+        """Return True if the data has an activity type"""
+        try:
+            return (data.get("type") or "").lower() in ACTIVITY_TYPES_LOWER
+        except (KeyError, TypeError):
+            return False
+
     async def apGet(self, url: str) -> dict:
         """Retrieve AP JSON from given URL
 
@@ -514,15 +555,30 @@ class APGateway:
 
         return jid_, node
 
-    def parseAPURL(self, url: str) -> Tuple[str, str]:
+    def getLocalJIDFromAccount(self, account: str) -> jid.JID:
+        """Compute JID linking to an AP account
+
+        The local jid is computer by escaping AP actor handle and using it as local part
+        of JID, where domain part is this gateway own JID
+        """
+        return jid.JID(
+            None,
+            (
+                self._e.escape(account),
+                self.client.jid.host,
+                None
+            )
+        )
+
+    def parseAPURL(self, url: str) -> Tuple[str, List[str]]:
         """Parse an URL leading to an AP endpoint
 
         @param url: URL to parse (schema is not mandatory)
-        @return: endpoint type and AP account
+        @return: endpoint type and extra arguments
         """
         path = parse.urlparse(url).path.lstrip("/")
-        type_, account = path[len(self.ap_path):].lstrip("/").split("/", 1)
+        type_, *extra_args = path[len(self.ap_path):].lstrip("/").split("/", 1)
-        return type_, parse.unquote(account)
+        return type_, [parse.unquote(a) for a in extra_args]
 
     def buildAPURL(self, type_:str , *args: str) -> str:
         """Build an AP endpoint URL
@@ -535,42 +591,199 @@ class APGateway:
             str(Path(type_).joinpath(*(parse.quote_plus(a) for a in args)))
         )
 
-    async def signAndPost(self, url: str, url_actor: str, doc: dict) -> TReqResponse:
+    def buildSignatureHeader(self, values: Dict[str, str]) -> str:
-        """Sign a documentent and post it to AP server
+        """Build key="<value>" signature header from signature data"""
+        fields = []
+        for key, value in values.items():
+            if key not in ("(created)", "(expired)"):
+                if '"' in value:
+                    raise NotImplementedError(
+                        "string escaping is not implemented, double-quote can't be used "
+                        f"in {value!r}"
+                    )
+                value = f'"{value}"'
+            fields.append(f"{key}={value}")
 
-        @param url: AP server endpoint
+        return ",".join(fields)
-        @param url_actor: URL generated by this gateway for local actor
+
-        @param doc: document to send
+    def getDigest(self, body: bytes, algo="SHA-256") -> Tuple[str, str]:
+        """Get digest data to use in header and signature
+
+        @param body: body of the request
+        @return: hash name and digest
         """
-        p_url = parse.urlparse(url)
+        if algo != "SHA-256":
-        date = http.datetimeToString().decode()
+            raise NotImplementedError("only SHA-256 is implemented for now")
-        body = json.dumps(doc).encode()
+        return algo, base64.b64encode(hashlib.sha256(body).digest()).decode()
-        digest_hash = base64.b64encode(hashlib.sha256(body).digest()).decode()
+
-        digest = f"sha-256={digest_hash}"
+    @async_lru(maxsize=LRU_MAX_SIZE)
-        to_sign = (
+    async def getActorPubKeyData(
-            f"(request-target): post {p_url.path}\nhost: {p_url.hostname}\n"
+        self,
-            f"date: {date}\ndigest: {digest}"
+        actor_id: str
-        )
+    ) -> Tuple[str, str, rsa.RSAPublicKey]:
+        """Retrieve Public Key data from actor ID
+
+        @param actor_id: actor ID (url)
+        @return: key_id, owner and public_key
+        @raise KeyError: publicKey is missing from actor data
+        """
+        actor_data = await self.apGet(actor_id)
+        pub_key_data = actor_data["publicKey"]
+        key_id = pub_key_data["id"]
+        owner = pub_key_data["owner"]
+        pub_key_pem = pub_key_data["publicKeyPem"]
+        pub_key = serialization.load_pem_public_key(pub_key_pem.encode())
+        return key_id, owner, pub_key
+
+    def createActivity(
+        self,
+        activity: str,
+        actor_id: str,
+        object_: Optional[Union[str, dict]] = None,
+        target: Optional[Union[str, dict]] = None,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        """Generate base data for an activity
+
+        @param activity: one of ACTIVITY_TYPES
+        """
+        if activity not in ACTIVITY_TYPES:
+            raise exceptions.InternalError(f"invalid activity: {activity!r}")
+        if object_ is None and activity in ACTIVITY_OBJECT_MANDATORY:
+            raise exceptions.InternalError(
+                f'"object_" is mandatory for activity {activity!r}'
+            )
+        if target is None and activity in ACTIVITY_TARGET_MANDATORY:
+            raise exceptions.InternalError(
+                f'"target" is mandatory for activity {activity!r}'
+            )
+        activity_id = f"{actor_id}#{activity.lower()}_{shortuuid.uuid()}"
+        data: Dict[str, Any] = {
+            "@context": "https://www.w3.org/ns/activitystreams",
+            "actor": actor_id,
+            "id": activity_id,
+            "type": activity,
+        }
+        data.update(kwargs)
+        if object_ is not None:
+            data["object"] = object_
+        if target is not None:
+            data["target"] = target
+
+        return data
+
+    def getKeyId(self, actor_id: str) -> str:
+        """Get local key ID from actor ID"""
+        return f"{actor_id}#main-key"
+
+    async def checkSignature(
+        self,
+        signature: str,
+        key_id: str,
+        headers: Dict[str, str]
+    ) -> str:
+        """Verify that signature matches given headers
+
+        see https://datatracker.ietf.org/doc/html/draft-cavage-http-signatures-06#section-3.1.2
+
+        @param signature: Base64 encoded signature
+        @param key_id: ID of the key used to sign the data
+        @param headers: headers and their values, including pseudo-headers
+        @return: id of the signing actor
+
+        @raise InvalidSignature: signature doesn't match headers
+        """
+        to_sign = "\n".join(f"{k.lower()}: {v}" for k,v in headers.items())
+        if key_id.startswith("acct:"):
+            actor = key_id[5:]
+            actor_id = await self.getAPActorIdFromAccount(actor)
+        else:
+            actor_id = key_id.split("#", 1)[0]
+
+        pub_key_id, pub_key_owner, pub_key = await self.getActorPubKeyData(actor_id)
+        if pub_key_id != key_id or pub_key_owner != actor_id:
+            raise exceptions.EncryptionError("Public Key mismatch")
+
+        try:
+            pub_key.verify(
+                base64.b64decode(signature),
+                to_sign.encode(),
+                # we have to use PKCS1v15 padding to be compatible with Mastodon
+                padding.PKCS1v15(),  # type: ignore
+                hashes.SHA256()  # type: ignore
+            )
+        except InvalidSignature:
+            raise exceptions.EncryptionError("Invalid signature (using PKC0S1 v1.5 and SHA-256)")
+
+        return actor_id
+
+    def getSignatureData(
+            self,
+            key_id: str,
+            headers: Dict[str, str]
+    ) -> Tuple[Dict[str, str], Dict[str, str]]:
+        """Generate and return signature and corresponding headers
+
+        @param parsed_url: URL where the request is sent/has been received
+        @param key_id: ID of the key (URL linking to the data with public key)
+        @param date: HTTP datetime string of signature generation
+        @param body: body of the HTTP request
+        @param headers: headers to sign and their value:
+            default value will be used if not specified
+
+        @return: headers and signature data
+            ``headers`` is an updated copy of ``headers`` arguments, with pseudo-headers
+            removed, and ``Signature`` added.
+        """
+        to_sign = "\n".join(f"{k.lower()}: {v}" for k,v in headers.items())
         signature = base64.b64encode(self.private_key.sign(
             to_sign.encode(),
             # we have to use PKCS1v15 padding to be compatible with Mastodon
-            padding.PKCS1v15(),
+            padding.PKCS1v15(),  # type: ignore
-            hashes.SHA256()
+            hashes.SHA256()  # type: ignore
         )).decode()
-        h_signature = (
+        sign_data = {
-            f'keyId="{url_actor}",headers="(request-target) host date digest",'
+            "keyId": key_id,
-            f'signature="{signature}"'
+            "Algorithm": "rsa-sha256",
+            "headers": " ".join(headers.keys()),
+            "signature": signature
+        }
+        new_headers = {k: v for k,v in headers.items() if not k.startswith("(")}
+        new_headers["Signature"] = self.buildSignatureHeader(sign_data)
+        return new_headers, sign_data
+
+    async def signAndPost(self, url: str, actor_id: str, doc: dict) -> TReqResponse:
+        """Sign a documentent and post it to AP server
+
+        @param url: AP server endpoint
+        @param actor_id: originating actor ID (URL)
+        @param doc: document to send
+        """
+        p_url = parse.urlparse(url)
+        body = json.dumps(doc).encode()
+        digest_algo, digest_hash = self.getDigest(body)
+        digest = f"{digest_algo}={digest_hash}"
+
+        headers = {
+            "(request-target)": f"post {p_url.path}",
+            "Host": p_url.hostname,
+            "Date": http.datetimeToString().decode(),
+            "Digest": digest
+        }
+        headers, __ = self.getSignatureData(self.getKeyId(actor_id), headers)
+
+        headers["Content-Type"] = (
+            'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'
         )
-        return await treq.post(
+        resp = await treq.post(
             url,
             body,
-            headers={
+            headers=headers,
-                "Host": [p_url.hostname],
-                "Date": [date],
-                "Digest": [digest],
-                "Signature": [h_signature],
-            }
         )
+        if resp.code >= 400:
+            text = await resp.text()
+            log.warning(f"POST request to {url} failed: {text}")
+        return resp
 
     def _publishMessage(self, mess_data_s: str, service_s: str, profile: str):
         mess_data: dict = data_format.deserialise(mess_data_s) # type: ignore
@@ -578,6 +791,7 @@ class APGateway:
         client = self.host.getClient(profile)
         return defer.ensureDeferred(self.publishMessage(client, mess_data, service))
 
+    @async_lru(maxsize=LRU_MAX_SIZE)
     async def getAPActorIdFromAccount(self, account: str) -> str:
         """Retrieve account ID from it's handle using WebFinger
 
@@ -611,7 +825,7 @@ class APGateway:
             )
         return href
 
-    async def getAPActorDataFromId(self, account: str) -> dict:
+    async def getAPActorDataFromAccount(self, account: str) -> dict:
         """Retrieve ActivityPub Actor data
 
         @param account: ActivityPub Actor identifier
@@ -620,7 +834,13 @@ class APGateway:
         return await self.apGet(href)
 
     @async_lru(maxsize=LRU_MAX_SIZE)
-    async def getAPAccountFromId(self, actor_id: str):
+    async def getAPInboxFromId(self, actor_id: str) -> str:
+        """Retrieve inbox of an actor_id"""
+        data = await self.apGet(actor_id)
+        return data["inbox"]
+
+    @async_lru(maxsize=LRU_MAX_SIZE)
+    async def getAPAccountFromId(self, actor_id: str) -> str:
         """Retrieve AP account from the ID URL
 
         @param actor_id: AP ID of the actor (URL to the actor data)
@@ -648,7 +868,7 @@ class APGateway:
 
     async def getAPItems(
         self,
-        account: str,
+        collection: dict,
         max_items: Optional[int] = None,
         chronological_pagination: bool = True,
         after_id: Optional[str] = None,
@@ -675,21 +895,18 @@ class APGateway:
         @return: XMPP Pubsub items and corresponding RSM Response
             Items are always returned in chronological order in the result
         """
-        actor_data = await self.getAPActorDataFromId(account)
-        outbox = actor_data.get("outbox")
         rsm_resp: Dict[str, Union[bool, int]] = {}
-        if not outbox:
-            raise exceptions.DataError(f"No outbox found for actor {account}")
-        outbox_data = await self.apGet(outbox)
         try:
-            count = outbox_data["totalItems"]
+            count = collection["totalItems"]
         except KeyError:
             log.warning(
-                f'"totalItems" not found in outbox of {account}, defaulting to 20'
+                f'"totalItems" not found in collection {collection.get("id")}, '
+                "defaulting to 20"
             )
             count = 20
         else:
-            log.info(f"{account}'s outbox has {count} item(s)")
+            log.info(f"{collection.get('id')} has {count} item(s)")
+
             rsm_resp["count"] = count
 
         if start_index is not None:
@@ -710,7 +927,7 @@ class APGateway:
                 # before "start_index"
                 previous_index = start_index - 1
                 retrieved_items = 0
-                current_page = outbox_data["last"]
+                current_page = collection["last"]
                 while retrieved_items < count:
                     page_data, items = await self.parseAPPage(current_page)
                     if not items:
@@ -734,10 +951,11 @@ class APGateway:
                         )
 
         init_page = "last" if chronological_pagination else "first"
-        page = outbox_data.get(init_page)
+        page = collection.get(init_page)
         if not page:
             raise exceptions.DataError(
-                f"Initial page {init_page!r} not found for outbox {outbox}"
+                f"Initial page {init_page!r} not found for collection "
+                f"{collection.get('id')})"
             )
         items = []
         page_items = []
@@ -746,6 +964,8 @@ class APGateway:
 
         while retrieved_items < count:
             __, page_items = await self.parseAPPage(page)
+            if not page_items:
+                break
             retrieved_items += len(page_items)
             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
@@ -754,7 +974,8 @@ class APGateway:
                     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")
+                    page_id = page.get("id") if isinstance(page, dict) else page
+                    log.debug(f"{after_id!r} not found at {page_id}, skipping")
                 else:
                     found_after_id = True
                     if chronological_pagination:
@@ -773,25 +994,24 @@ class APGateway:
                 else:
                     items = items[-max_items:]
                 break
-            page = outbox_data.get("prev" if chronological_pagination else "next")
+            page = collection.get("prev" if chronological_pagination else "next")
             if not page:
                 break
 
         if after_id is not None and not found_after_id:
             raise error.StanzaError("item-not-found")
 
-        if after_id is None:
-            rsm_resp["index"] = 0 if chronological_pagination else count - len(items)
-
-        if start_index is not None:
-            rsm_resp["index"] = start_index
-        elif after_id is not None:
-            log.warning("Can't determine index of first element")
-        elif chronological_pagination:
-            rsm_resp["index"] = 0
-        else:
-            rsm_resp["index"] = count - len(items)
         if items:
+            if after_id is None:
+                rsm_resp["index"] = 0 if chronological_pagination else count - len(items)
+            if start_index is not None:
+                rsm_resp["index"] = start_index
+            elif after_id is not None:
+                log.warning("Can't determine index of first element")
+            elif chronological_pagination:
+                rsm_resp["index"] = 0
+            else:
+                rsm_resp["index"] = count - len(items)
             rsm_resp.update({
                 "first": items[0]["id"],
                 "last": items[-1]["id"]
@@ -799,55 +1019,119 @@ class APGateway:
 
         return items, rsm.RSMResponse(**rsm_resp)
 
-    async def parseAPPage(self, url: str) -> Tuple[dict, List[domish.Element]]:
+    async def parseAPPage(
+        self,
+        page: Union[str, dict]
+    ) -> Tuple[dict, List[domish.Element]]:
         """Convert AP objects from an AP page to XMPP items
 
-        @param url: url linking and AP page
+        @param page: Can be either url linking and AP page, or the page data directly
         @return: page data, pubsub items
         """
-        page_data = await self.apGet(url)
+        page_data = await self.apGetObject(page)
-        ap_items = page_data.get("orderedItems")
+        if page_data is None:
-        if not ap_items:
+            log.warning('No data found in collection')
-            log.warning('No "orderedItems" collection found')
+            return {}, []
-            return page_data, []
+        ap_items = await self.apGetList(page_data, "orderedItems")
+        if ap_items is None:
+            ap_items = await self.apGetList(page_data, "items")
+            if not ap_items:
+                log.warning(f'No item field found in collection: {page_data!r}')
+                return page_data, []
+            else:
+                log.warning(
+                    "Items are not ordered, this is not spec compliant"
+                )
         items = []
         # AP Collections are in antichronological order, but we expect chronological in
         # Pubsub, thus we reverse it
         for ap_item in reversed(ap_items):
             try:
-                ap_object, mb_data = await self.apItem2MBdata(ap_item)
+                items.append(await self.apItem2Elt(ap_item))
             except (exceptions.DataError, NotImplementedError, error.StanzaError):
                 continue
 
-            item_elt = await self._m.data2entry(
-                self.client, mb_data, ap_object["id"], None, self._m.namespace
-            )
-            item_elt["publisher"] = mb_data["author_jid"].full()
-            items.append(item_elt)
-
         return page_data, items
 
-    async def apItem2MBdata(self, ap_item: dict) -> Tuple[dict, dict]:
+    async def apItem2MbDataAndElt(self, ap_item: dict) -> Tuple[dict, domish.Element]:
-        """Convert AP item to microblog data
+        """Convert AP item to parsed microblog data and corresponding item element"""
+        mb_data = await self.apItem2MBdata(ap_item)
+        item_elt = await self._m.data2entry(
+            self.client, mb_data, mb_data["id"], None, self._m.namespace
+        )
+        item_elt["publisher"] = mb_data["author_jid"]
+        return mb_data, item_elt
+
+    async def apItem2Elt(self, ap_item: dict) -> domish.Element:
+        """Convert AP item to XMPP item element"""
+        __, item_elt = await self.apItem2MbDataAndElt(ap_item)
+        return item_elt
+
+    async def getCommentsNodes(
+        self,
+        item_id: str,
+        parent_id: Optional[str]
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """Get node where this item is and node to use for comments
+
+        if config option "comments_max_depth" is set, a common node will be used below the
+        given depth
+        @param item_id: ID of the reference item
+        @param parent_id: ID of the parent item if any (the ID set in "inReplyTo")
+        @return: a tuple with parent_node_id, comments_node_id:
+            - parent_node_id is the ID of the node where reference item must be. None is
+              returned when the root node (i.e. not a comments node) must be used.
+            - comments_node_id: is the ID of the node to use for comments. None is
+              returned when no comment node must be used (happens when we have reached
+              "comments_max_depth")
+        """
+        if parent_id is None or not self.comments_max_depth:
+            return (
+                self._m.getCommentsNode(parent_id) if parent_id is not None else None,
+                self._m.getCommentsNode(item_id)
+            )
+        parent_url = parent_id
+        parents = []
+        for __ in range(COMMENTS_MAX_PARENTS):
+            parent_item = await self.apGet(parent_url)
+            parents.insert(0, parent_item)
+            parent_url = parent_item.get("inReplyTo")
+            if parent_url is None:
+                break
+        parent_limit = self.comments_max_depth-1
+        if len(parents) <= parent_limit:
+            return (
+                self._m.getCommentsNode(parents[-1]["id"]),
+                self._m.getCommentsNode(item_id)
+            )
+        else:
+            last_level_item = parents[parent_limit]
+            return (
+                self._m.getCommentsNode(last_level_item["id"]),
+                None
+            )
+
+    async def apItem2MBdata(self, ap_item: dict) -> dict:
+        """Convert AP activity or object to microblog data
 
         @return: AP Item's Object and microblog data
         @raise exceptions.DataError: something is invalid in the AP item
         @raise NotImplemented: some AP data is not handled yet
         @raise error.StanzaError: error while contacting the AP server
         """
-        ap_object = ap_item.get("object")
+        is_activity = self.isActivity(ap_item)
-        if not ap_object:
+        if is_activity:
-            log.warning(f'No "object" found in AP item {ap_item!r}')
+            ap_object = await self.apGetObject(ap_item, "object")
+            if not ap_object:
+                log.warning(f'No "object" found in AP item {ap_item!r}')
+                raise exceptions.DataError
+        else:
+            ap_object = ap_item
+        item_id = ap_object.get("id")
+        if not item_id:
+            log.warning(f'No "id" found in AP item: {ap_object!r}')
             raise exceptions.DataError
-        if isinstance(ap_object, str):
+        mb_data = {"id": item_id}
-            ap_object = await self.apGet(ap_object)
-        obj_id = ap_object.get("id")
-        if not obj_id:
-            log.warning(f'No "id" found in AP object: {ap_object!r}')
-            raise exceptions.DataError
-        if ap_object.get("inReplyTo") is not None:
-            raise NotImplementedError
-        mb_data = {}
         for ap_key, mb_key in AP_MB_MAP.items():
             data = ap_object.get(ap_key)
             if data is None:
@@ -868,37 +1152,20 @@ class APGateway:
             mb_data["content_xhtml"] = content_xhtml
 
         # author
-        actor = ap_item.get("actor")
+        if is_activity:
-        if not actor:
+            authors = await self.apGetActors(ap_item, "actor")
-            log.warning(f"no actor associated to object id {obj_id!r}")
-            raise exceptions.DataError
-        elif isinstance(actor, list):
-            # we only keep first item of list as author
-            # TODO: handle multiple actors
-            if len(actor) > 1:
-                log.warning("multiple actors are not managed")
-            actor = actor[0]
-
-        if isinstance(actor, dict):
-            actor = actor.get("id")
-            if not actor:
-                log.warning(f"no actor id found: {actor!r}")
-                raise exceptions.DataError
-
-        if isinstance(actor, str):
-            account = await self.getAPAccountFromId(actor)
-            mb_data["author"] = account.split("@", 1)[0]
-            author_jid = mb_data["author_jid"] = jid.JID(
-                None,
-                (
-                    self.host.plugins["XEP-0106"].escape(account),
-                    self.client.jid.host,
-                    None
-                )
-            )
         else:
-            log.warning(f"unknown actor type found: {actor!r}")
+            authors = await self.apGetActors(ap_object, "attributedTo")
-            raise exceptions.DataError
+        if len(authors) > 1:
+            # we only keep first item as author
+            # TODO: handle multiple actors
+            log.warning("multiple actors are not managed")
+
+        account = authors[0]
+        author_jid = self.getLocalJIDFromAccount(account).full()
+
+        mb_data["author"] = account.split("@", 1)[0]
+        mb_data["author_jid"] = author_jid
 
         # published/updated
         for field in ("published", "updated"):
@@ -912,14 +1179,30 @@ class APGateway:
                     )
                 except dateutil.parser.ParserError as e:
                     log.warning(f"Can't parse {field!r} field: {e}")
-        return ap_object, mb_data
+
+        # comments
+        in_reply_to = ap_object.get("inReplyTo")
+        __, comments_node = await self.getCommentsNodes(item_id, in_reply_to)
+        if comments_node is not None:
+            comments_data = {
+                "service": author_jid,
+                "node": comments_node,
+                "uri": uri.buildXMPPUri(
+                    "pubsub",
+                    path=author_jid,
+                    node=comments_node
+                )
+            }
+            mb_data["comments"] = [comments_data]
+
+        return mb_data
 
     async def mbdata2APitem(self, client: SatXMPPEntity, mb_data: dict) -> dict:
         """Convert Libervia Microblog Data to ActivityPub item"""
         if not mb_data.get("id"):
             mb_data["id"] = shortuuid.uuid()
         if not mb_data.get("author_jid"):
-            mb_data["author_jid"] = client.jid
+            mb_data["author_jid"] = client.jid.full()
         ap_account = await self.getAPAccountFromJidAndNode(
             jid.JID(mb_data["author_jid"]),
             None
@@ -967,8 +1250,8 @@ class APGateway:
         """
         if not service.user:
             raise ValueError("service must have a local part")
-        account = self.host.plugins["XEP-0106"].unescape(service.user)
+        account = self._e.unescape(service.user)
-        ap_actor_data = await self.getAPActorDataFromId(account)
+        ap_actor_data = await self.getAPActorDataFromAccount(account)
 
         try:
             inbox_url = ap_actor_data["endpoints"]["sharedInbox"]
@@ -980,3 +1263,72 @@ 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}")
+
+    async def newAPItem(
+        self,
+        client: SatXMPPEntity,
+        destinee: Optional[jid.JID],
+        node: str,
+        item: dict,
+    ) -> None:
+        """Analyse, cache and send notification for received AP item
+
+        @param destinee: jid of the destinee,
+        @param node: XMPP pubsub node
+        @param item: AP object payload
+        """
+        service = client.jid
+        in_reply_to = item.get("inReplyTo")
+        if in_reply_to and isinstance(in_reply_to, str):
+            # this item is a reply, we use or create a corresponding node for comments
+            parent_node, __ = await self.getCommentsNodes(item["id"], in_reply_to)
+            node = parent_node or node
+            cached_node = await self.host.memory.storage.getPubsubNode(
+                client, service, node, with_subscriptions=True
+            )
+            if cached_node is None:
+                try:
+                    cached_node = await self.host.memory.storage.setPubsubNode(
+                        client,
+                        service,
+                        node,
+                        subscribed=True
+                    )
+                except IntegrityError as e:
+                    if "unique" in str(e.orig).lower():
+                        # the node may already exist, if it has been created just after
+                        # getPubsubNode above
+                        log.debug("ignoring UNIQUE constraint error")
+                        cached_node = await self.host.memory.storage.getPubsubNode(
+                            client, service, node, with_subscriptions=True
+                        )
+                    else:
+                        raise e
+
+        else:
+            # it is a root item (i.e. not a reply to an other item)
+            cached_node = await self.host.memory.storage.getPubsubNode(
+                client, service, node, with_subscriptions=True
+            )
+            if cached_node is None:
+                log.warning(
+                    f"Received item in unknown node {node!r} at {service}\n{item}"
+
+                )
+                return
+        mb_data, item_elt = await self.apItem2MbDataAndElt(item)
+        await self.host.memory.storage.cachePubsubItems(
+            client,
+            cached_node,
+            [item_elt],
+            [mb_data]
+        )
+
+        for subscription in cached_node.subscriptions:
+            if subscription.state != SubscriptionState.SUBSCRIBED:
+                continue
+            self.pubsub_service.notifyPublish(
+                service,
+                node,
+                [(subscription.subscriber, None, [item_elt])]
+            )

sat/plugins/plugin_comp_ap_gateway/constants.py

@@ -22,6 +22,7 @@ CONF_SECTION = f"component {IMPORT_NAME}"
 CONTENT_TYPE_AP = "application/activity+json; charset=utf-8"
 TYPE_ACTOR = "actor"
 TYPE_INBOX = "inbox"
+TYPE_SHARED_INBOX = "shared_inbox"
 TYPE_OUTBOX = "outbox"
 TYPE_ITEM = "item"
 MEDIA_TYPE_AP = "application/activity+json"
@@ -30,7 +31,39 @@ AP_MB_MAP = {
     "content": "content_xhtml",
 
 }
-AP_REQUEST_TYPES = {"actor", "outbox"}
+AP_REQUEST_TYPES = {
+    "GET": {"actor", "outbox"},
+    "POST": {"inbox"},
+}
+# headers to check for signature
+SIGN_HEADERS = {
+    # headers needed for all HTTP methods
+    None: [
+        # tuples are equivalent headers/pseudo headers, one of them must be present
+        ("date", "(created)"),
+        ("digest", "(request-target)"),
+    ],
+    b"GET": ["host"],
+    b"POST": ["digest"]
+}
 PAGE_SIZE = 10
+HS2019 = "hs2019"
+# delay after which a signed request is not accepted anymore
+SIGN_EXP = 12*60*60  # 12 hours (same value as for Mastodon)
 
 LRU_MAX_SIZE = 200
+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"
+)
+ACTIVITY_TYPES_LOWER = [a.lower() for a in ACTIVITY_TYPES]
+ACTIVITY_OBJECT_MANDATORY = (
+    "Create", "Update", "Delete", "Follow", "Add", "Remove", "Like", "Block", "Undo"
+)
+ACTIVITY_TARGET_MANDATORY = ("Add", "Remove")
+# activities which can be used with Shared Inbox (i.e. with not account specified)
+ACTIVIY_NO_ACCOUNT_ALLOWED = ("create",)
+# maximum number of parents to retrieve when comments_max_depth option is set
+COMMENTS_MAX_PARENTS = 100

sat/plugins/plugin_comp_ap_gateway/http_server.py

@@ -16,14 +16,16 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import time
 from typing import Optional, Dict, List
 import json
 from urllib import parse
-import re
+from collections import deque
 import unicodedata
+from pprint import pformat
 
 from twisted.web import http, resource as web_resource, server
-from twisted.internet import defer
+from twisted.internet import reactor, defer
 from twisted.words.protocols.jabber import jid, error
 from wokkel import pubsub, rsm
 
@@ -31,9 +33,15 @@ 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 sat.tools.common import date_utils
+from sat.memory.sqla_mapping import SubscriptionState
 
-from .constants import (CONTENT_TYPE_AP, TYPE_ACTOR, TYPE_INBOX, TYPE_OUTBOX,
+from .constants import (
-                       AP_REQUEST_TYPES, PAGE_SIZE)
+    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
+)
+from .regex import RE_SIG_PARAM
 
 
 log = getLogger(__name__)
@@ -50,8 +58,19 @@ class HTTPAPGServer(web_resource.Resource):
 
     def __init__(self, ap_gateway):
         self.apg = ap_gateway
+        self._seen_digest = deque(maxlen=50)
         super().__init__()
 
+    def responseCode(
+        self,
+        request: "HTTPRequest",
+        http_code: int,
+        msg: Optional[str] = None
+    ) -> None:
+        """Log and set HTTP return code and associated message"""
+        log.warning(msg)
+        request.setResponseCode(http_code, None if msg is None else msg.encode())
+
     async def webfinger(self, request):
         url_parsed = parse.urlparse(request.uri.decode())
         query = parse.parse_qs(url_parsed.query)
@@ -78,15 +97,127 @@ class HTTPAPGServer(web_resource.Resource):
         request.write(json.dumps(resp).encode())
         request.finish()
 
+    async def handleFollowActivity(
+        self,
+        request: "HTTPRequest",
+        data: dict,
+        account_jid: jid.JID,
+        node: Optional[str],
+        ap_account: str,
+        ap_url: str,
+        signing_actor: str
+    ) -> None:
+        if node is None:
+            node = self.apg._m.namespace
+        client = await self.apg.getVirtualClient(signing_actor)
+        try:
+            subscription = await self.apg._p.subscribe(
+                client,
+                account_jid,
+                node
+            )
+        except pubsub.SubscriptionPending:
+            log.info(f"subscription to node {node!r} of {account_jid} is pending")
+        # TODO: manage SubscriptionUnconfigured
+        else:
+            if subscription.state != "subscribed":
+                # other states should raise an Exception
+                raise exceptions.InternalError('"subscribed" state was expected')
+            inbox = await self.apg.getAPInboxFromId(signing_actor)
+            actor_id = self.apg.buildAPURL(TYPE_ACTOR, ap_account)
+            accept_data = self.apg.createActivity(
+                "Accept", actor_id, object_=data
+            )
+            await self.apg.signAndPost(inbox, actor_id, accept_data)
+
+    async def handleAcceptActivity(
+        self,
+        request: "HTTPRequest",
+        data: dict,
+        account_jid: jid.JID,
+        node: Optional[str],
+        ap_account: str,
+        ap_url: str,
+        signing_actor: str
+    ) -> None:
+        if node is None:
+            node = self.apg._m.namespace
+        client = await self.apg.getVirtualClient(signing_actor)
+        objects = await self.apg.apGetList(data, "object")
+        for obj in objects:
+            type_ = obj.get("type")
+            if type_ == "Follow":
+                follow_node = await self.apg.host.memory.storage.getPubsubNode(
+                    client, client.jid, node, with_subscriptions=True
+                )
+                if follow_node is None:
+                    log.warning(
+                        f"Received a follow accept on an unknown node: {node!r} at "
+                        f"{client.jid}. Ignoring it"
+                    )
+                    continue
+                try:
+                    sub = next(
+                        s for s in follow_node.subscriptions if s.subscriber==account_jid
+                    )
+                except StopIteration:
+                    log.warning(
+                        "Received a follow accept on a node without subscription: "
+                        f"{node!r} at {client.jid}. Ignoring it"
+                    )
+                else:
+                    if sub.state == SubscriptionState.SUBSCRIBED:
+                        log.warning(f"Already subscribed to {node!r} at {client.jid}")
+                    elif sub.state == SubscriptionState.PENDING:
+                        follow_node.subscribed = True
+                        sub.state = SubscriptionState.SUBSCRIBED
+                        await self.apg.host.memory.storage.add(follow_node)
+                    else:
+                        raise exceptions.InternalError(
+                            f"Unhandled subscription state {sub.state!r}"
+                        )
+            else:
+                log.warning(f"Unmanaged accept type: {type_!r}")
+
+    async def handleCreateActivity(
+        self,
+        request: "HTTPRequest",
+        data: dict,
+        account_jid: Optional[jid.JID],
+        node: Optional[str],
+        ap_account: Optional[str],
+        ap_url: str,
+        signing_actor: str
+    ):
+        digest = request.getHeader("digest")
+        if digest in self._seen_digest:
+            log.debug(f"Ignoring duplicated request (digest: {digest!r})")
+            return
+        self._seen_digest.append(digest)
+        if node is None:
+            node = self.apg._m.namespace
+        client = await self.apg.getVirtualClient(signing_actor)
+        objects = await self.apg.apGetList(data, "object")
+        for obj in objects:
+            sender = await self.apg.apGetSenderActor(obj)
+            if sender != signing_actor:
+                log.warning(
+                    "Ignoring object not attributed to signing actor: {obj}"
+                )
+            else:
+                await self.apg.newAPItem(client, account_jid, node, obj)
+
     async def APActorRequest(
         self,
         request: "HTTPRequest",
         account_jid: jid.JID,
         node: Optional[str],
         ap_account: str,
-        actor_url: str
+        actor_url: str,
+        signing_actor: Optional[str]
     ) -> dict:
         inbox_url = self.apg.buildAPURL(TYPE_INBOX, ap_account)
+        shared_inbox = self.apg.buildAPURL(TYPE_SHARED_INBOX)
         outbox_url = self.apg.buildAPURL(TYPE_OUTBOX, ap_account)
 
         # we have to use AP account as preferredUsername because it is used to retrieve
@@ -107,7 +238,10 @@ class HTTPAPGServer(web_resource.Resource):
                 "id": f"{actor_url}#main-key",
                 "owner": actor_url,
                 "publicKeyPem": self.apg.public_key_pem
-            }
+            },
+            "endpoints": {
+                "sharedInbox": shared_inbox
+            },
         }
 
     def getCanonicalURL(self, request: "HTTPRequest") -> str:
@@ -206,7 +340,8 @@ class HTTPAPGServer(web_resource.Resource):
         account_jid: jid.JID,
         node: Optional[str],
         ap_account: str,
-        ap_url: str
+        ap_url: str,
+        signing_actor: Optional[str]
     ) -> dict:
         if node is None:
             node = self.apg._m.namespace
@@ -229,7 +364,8 @@ class HTTPAPGServer(web_resource.Resource):
                 service=account_jid,
                 node=node,
                 max_items=0,
-                rsm_request=rsm.RSMRequest(max_=0)
+                rsm_request=rsm.RSMRequest(max_=0),
+                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}")
@@ -255,22 +391,268 @@ class HTTPAPGServer(web_resource.Resource):
             "last": url_last_page,
         }
 
-    async def APRequest(self, request):
+    async def APInboxRequest(
+        self,
+        request: "HTTPRequest",
+        account_jid: Optional[jid.JID],
+        node: Optional[str],
+        ap_account: Optional[str],
+        ap_url: str,
+        signing_actor: Optional[str]
+    ) -> None:
+        if signing_actor is None:
+            raise exceptions.InternalError("signing_actor must be set for inbox requests")
+        if node is None:
+            node = self.apg._m.namespace
+        try:
+            data = json.load(request.content)
+            if not isinstance(data, dict):
+                raise ValueError("data should be an object")
+        except (json.JSONDecodeError, ValueError) as e:
+            return self.responseCode(
+                request,
+                http.BAD_REQUEST,
+                f"invalid json in inbox request: {e}"
+            )
+        await self.checkSigningActor(data, signing_actor)
+        activity_type = (data.get("type") or "").lower()
+        if not activity_type in ACTIVITY_TYPES_LOWER:
+            return self.responseCode(
+                request,
+                http.UNSUPPORTED_MEDIA_TYPE,
+                f"request is not an activity, ignoring"
+            )
+
+        if account_jid is None and activity_type not in ACTIVIY_NO_ACCOUNT_ALLOWED:
+            return self.responseCode(
+                request,
+                http.UNSUPPORTED_MEDIA_TYPE,
+                f"{activity_type.title()!r} activity must target an account"
+            )
+
+        try:
+            method = getattr(self, f"handle{activity_type.title()}Activity")
+        except AttributeError:
+            return self.responseCode(
+                request,
+                http.UNSUPPORTED_MEDIA_TYPE,
+                f"{activity_type.title()} activity is not yet supported"
+            )
+        else:
+            await method(
+                request, data, account_jid, node, ap_account, ap_url, signing_actor
+            )
+
+    async def APRequest(
+        self,
+        request: "HTTPRequest",
+        signing_actor: Optional[str] = None
+    ) -> None:
         path = request.path.decode()
         ap_url = parse.urljoin(
             f"https://{self.apg.public_url}",
             path
         )
-        request_type, ap_account = self.apg.parseAPURL(ap_url)
+        request_type, extra_args = self.apg.parseAPURL(ap_url)
-        account_jid, node = await self.apg.getJIDAndNode(ap_account)
+        if len(extra_args) == 0:
-        if request_type not in AP_REQUEST_TYPES:
+            if request_type != "shared_inbox":
-            raise exceptions.DataError(f"Invalid request type: {request_type!r}")
+                raise exceptions.DataError(f"Invalid request type: {request_type!r}")
-        method = getattr(self, f"AP{request_type.title()}Request")
+            ret_data = await self.APInboxRequest(
-        ret_data = await method(request, account_jid, node, ap_account, ap_url)
+                request, None, None, None, ap_url, signing_actor
-        request.setHeader("content-type", CONTENT_TYPE_AP)
+            )
-        request.write(json.dumps(ret_data).encode())
+        else:
+            if len(extra_args) > 1:
+                log.warning(f"unexpected extra arguments: {extra_args!r}")
+            ap_account = extra_args[0]
+            account_jid, node = await self.apg.getJIDAndNode(ap_account)
+            if request_type not in AP_REQUEST_TYPES.get(
+                    request.method.decode().upper(), []
+            ):
+                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, signing_actor
+            )
+        if ret_data is not None:
+            request.setHeader("content-type", CONTENT_TYPE_AP)
+            request.write(json.dumps(ret_data).encode())
         request.finish()
 
+    async def APPostRequest(self, request: "HTTPRequest"):
+        try:
+            signing_actor = await self.checkSignature(request)
+        except exceptions.EncryptionError as e:
+            self.responseCode(
+                request,
+                http.FORBIDDEN,
+                f"invalid signature: {e}"
+            )
+            request.finish()
+            return
+
+        return await self.APRequest(request, signing_actor)
+
+    async def checkSigningActor(self, data: dict, signing_actor: str) -> None:
+        """That that signing actor correspond to actor declared in data
+
+        @param data: request payload
+        @param signing_actor: actor ID of the signing entity, as returned by
+            checkSignature
+        @raise exceptions.NotFound: no actor found in data
+        @raise exceptions.EncryptionError: signing actor doesn't match actor in data
+        """
+        actor = await self.apg.apGetSenderActor(data)
+
+        if signing_actor != actor:
+            raise exceptions.EncryptionError(
+                f"signing actor ({signing_actor}) doesn't match actor in data ({actor})"
+            )
+
+    async def checkSignature(self, request: "HTTPRequest") -> str:
+        """Check and validate HTTP signature
+
+        @return: id of the signing actor
+
+        @raise exceptions.EncryptionError: signature is not present or doesn't match
+        """
+        signature = request.getHeader("Signature")
+        if signature is None:
+            raise exceptions.EncryptionError("No signature found")
+        sign_data = {
+            m["key"]: m["uq_value"] or m["quoted_value"][1:-1]
+            for m in RE_SIG_PARAM.finditer(signature)
+        }
+        try:
+            key_id = sign_data["keyId"]
+        except KeyError:
+            raise exceptions.EncryptionError('"keyId" is missing from signature')
+        algorithm = sign_data.get("algorithm", HS2019)
+        signed_headers = sign_data.get(
+            "headers",
+            "(created)" if algorithm==HS2019 else "date"
+        ).lower().split()
+        try:
+            headers_to_check = SIGN_HEADERS[None] + SIGN_HEADERS[request.method]
+        except KeyError:
+            raise exceptions.InternalError(
+                f"there should be a list of headers for {request.method} method"
+            )
+        if not headers_to_check:
+            raise exceptions.InternalError("headers_to_check must not be empty")
+
+        for header in headers_to_check:
+            if isinstance(header, tuple):
+                if len(set(header).intersection(signed_headers)) == 0:
+                    raise exceptions.EncryptionError(
+                        f"at least one of following header must be signed: {header}"
+                    )
+            elif header not in signed_headers:
+                raise exceptions.EncryptionError(
+                    f"the {header!r} header must be signed"
+                )
+
+        body = request.content.read()
+        request.content.seek(0)
+        headers = {}
+        for to_sign in signed_headers:
+            if to_sign == "(request-target)":
+                method = request.method.decode().lower()
+                uri = parse.unquote(request.uri.decode())
+                headers[to_sign] = f"{method} /{uri.lstrip('/')}"
+            elif to_sign in ("(created)", "(expires)"):
+                if algorithm != HS2019:
+                    raise exceptions.EncryptionError(
+                        f"{to_sign!r} pseudo-header can only be used with {HS2019} "
+                        "algorithm"
+                    )
+                key = to_sign[1:-1]
+                value = sign_data.get(key)
+                if not value:
+                    raise exceptions.EncryptionError(
+                        "{key!r} parameter is missing from signature"
+                    )
+                try:
+                    if float(value) < 0:
+                        raise ValueError
+                except ValueError:
+                    raise exceptions.EncryptionError(
+                        f"{to_sign} must be a Unix timestamp"
+                    )
+                headers[to_sign] = value
+            else:
+                value = request.getHeader(to_sign)
+                if not value:
+                    raise exceptions.EncryptionError(
+                        f"value of header {to_sign!r} is missing!"
+                    )
+                elif to_sign == "host":
+                    # we check Forwarded/X-Forwarded-Host headers
+                    # as we need original host if a proxy has modified the header
+                    forwarded = request.getHeader("forwarded")
+                    if forwarded is not None:
+                        try:
+                            host = [
+                                f[5:] for f in forwarded.split(";")
+                                if f.startswith("host=")
+                            ][0] or None
+                        except IndexError:
+                            host = None
+                    else:
+                        host = None
+                    if host is None:
+                        host = request.getHeader("x-forwarded-host")
+                    if host:
+                        value = host
+                elif to_sign == "digest":
+                    hashes = {
+                        algo.lower(): hash_ for algo, hash_ in (
+                            digest.split("=", 1) for digest in value.split(",")
+                        )
+                    }
+                    try:
+                        given_digest = hashes["sha-256"]
+                    except KeyError:
+                        raise exceptions.EncryptionError(
+                            "Only SHA-256 algorithm is currently supported for digest"
+                        )
+                    __, computed_digest = self.apg.getDigest(body)
+                    if given_digest != computed_digest:
+                        raise exceptions.EncryptionError(
+                            f"SHA-256 given and computed digest differ:\n"
+                            f"given: {given_digest!r}\ncomputed: {computed_digest!r}"
+                        )
+                headers[to_sign] = value
+
+        # date check
+        limit_ts = time.time() + SIGN_EXP
+        if "(created)" in headers:
+            created = float(headers["created"])
+        else:
+            created = date_utils.date_parse(headers["date"])
+
+
+        try:
+            expires = float(headers["expires"])
+        except KeyError:
+            pass
+        else:
+            if expires < created:
+                log.warning(
+                    f"(expires) [{expires}] set in the past of (created) [{created}] "
+                    "ignoring it according to specs"
+                )
+            else:
+                limit_ts = min(limit_ts, expires)
+
+        if created > limit_ts:
+            raise exceptions.EncryptionError("Signature has expired")
+
+        return await self.apg.checkSignature(
+            sign_data["signature"],
+            key_id,
+            headers
+        )
+
     def render(self, request):
         request.setHeader("server", VERSION)
         return super().render(request)
@@ -286,6 +668,13 @@ class HTTPAPGServer(web_resource.Resource):
 
         return web_resource.NoResource().render(request)
 
+    def render_POST(self, request):
+        path = request.path.decode().lstrip("/")
+        if not path.startswith(self.apg.ap_path):
+            return web_resource.NoResource().render(request)
+        defer.ensureDeferred(self.APPostRequest(request))
+        return server.NOT_DONE_YET
+
 
 class HTTPRequest(server.Request):
     pass

sat/plugins/plugin_comp_ap_gateway/pubsub_service.py

@@ -16,19 +16,41 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import Optional, List
+from typing import Optional, Tuple, List, Dict, Any
 
+from twisted.internet import defer
 from twisted.words.protocols.jabber import jid, error
 from twisted.words.xish import domish
-from wokkel import rsm
+from wokkel import rsm, pubsub, data_form
 
 from sat.core.i18n import _
+from sat.core import exceptions
 from sat.core.log import getLogger
+from sat.core.constants import Const as C
 from sat.tools.utils import ensure_deferred
+from sat.memory.sqla_mapping import PubsubSub, SubscriptionState
+
+from .constants import (
+    TYPE_ACTOR,
+)
 
 
 log = getLogger(__name__)
 
+# all nodes have the same config
+NODE_CONFIG = [
+    {"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"},
+
+]
+
+NODE_CONFIG_VALUES = {c["var"]: c["value"] for c in NODE_CONFIG}
+NODE_OPTIONS = {c["var"]: {} for c in NODE_CONFIG}
+for c in NODE_CONFIG:
+    NODE_OPTIONS[c["var"]].update({k:v for k,v in c.items() if k not in ("var", "value")})
+
 
 class APPubsubService(rsm.PubSubService):
     """Pubsub service for XMPP requests"""
@@ -43,6 +65,31 @@ class APPubsubService(rsm.PubSubService):
             "name": "Libervia ActivityPub Gateway",
         }
 
+    async def getAPActorIdsAndInbox(
+        self,
+        requestor: jid.JID,
+        recipient: jid.JID,
+    ) -> Tuple[str, str, str]:
+        """Get AP actor IDs from requestor and destinee JIDs
+
+        @param requestor: XMPP entity doing a request to an AP actor via the gateway
+        @param recipient: JID mapping an AP actor via the gateway
+        @return: requestor actor ID, recipient actor ID and recipient inbox
+        @raise error.StanzaError: "item-not-found" is raised if not user part is specified
+            in requestor
+        """
+        if not recipient.user:
+            raise error.StanzaError(
+                "item-not-found",
+                text="No user part specified"
+            )
+        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)
+        return requestor_actor_id, recipient_actor_id, inbox
+
+
     @ensure_deferred
     async def publish(self, requestor, service, nodeIdentifier, items):
         raise NotImplementedError
@@ -56,55 +103,193 @@ class APPubsubService(rsm.PubSubService):
         maxItems: Optional[int],
         itemIdentifiers: Optional[List[str]],
         rsm_req: Optional[rsm.RSMRequest]
-    ) -> List[domish.Element]:
+    ) -> Tuple[List[domish.Element], Optional[rsm.RSMResponse]]:
         if not service.user:
-            return []
+            return [], None
         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 []
+            return [], None
-        if node != self.apg._m.namespace:
+        if not node.startswith(self.apg._m.namespace):
             raise error.StanzaError(
                 "feature-not-implemented",
-                text=f"{VERSION} only supports {self.apg._m.namespace} "
+                text=f"AP Gateway {C.APP_VERSION} only supports {self.apg._m.namespace} "
                 "node for now"
             )
-        if rsm_req is None:
+        client = self.apg.client
-            if maxItems is None:
+        cached_node = await self.host.memory.storage.getPubsubNode(
-                maxItems = 20
+            client, service, node
-            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)
+        # TODO: check if node is synchronised
+        if cached_node is not None:
+            # the node is cached, we return items from cache
+            log.debug(f"node {node!r} from {service} is in cache")
+            pubsub_items, metadata = await self.apg._c.getItemsFromCache(
+                client, cached_node, maxItems, itemIdentifiers, rsm_request=rsm_req
+            )
+            try:
+                rsm_resp = rsm.RSMResponse(**metadata["rsm"])
+            except KeyError:
+                rsm_resp = None
+            return [i.data for i in pubsub_items], rsm_resp
+
+        if itemIdentifiers:
+            items = []
+            for item_id in itemIdentifiers:
+                item_data = await self.apg.apGet(item_id)
+                item_elt = await self.apg.apItem2Elt(item_data)
+                items.append(item_elt)
+            return items, None
+        else:
+            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"
+            )
+            if self.apg._m.isCommentsNode(node):
+                parent_node = self.apg._m.getParentNode(node)
+                try:
+                    parent_data = await self.apg.apGet(parent_node)
+                    collection = await self.apg.apGetObject(
+                        parent_data.get("object", {}),
+                        "replies"
+                    )
+                except Exception as e:
+                    raise error.StanzaError(
+                        "item-not-found",
+                        text=e
+                    )
+            else:
+                actor_data = await self.apg.getAPActorDataFromAccount(ap_account)
+                collection = await self.apg.apGetObject(actor_data, "outbox")
+            if not collection:
+                raise error.StanzaError(
+                    "item-not-found",
+                    text=f"No collection found for node {node!r} (account: {ap_account})"
+                )
+            return await self.apg.getAPItems(collection, **kwargs)
 
     @ensure_deferred
     async def retract(self, requestor, service, nodeIdentifier, itemIdentifiers):
         raise NotImplementedError
 
+    @ensure_deferred
+    async def subscribe(self, requestor, service, nodeIdentifier, subscriber):
+        # TODO: handle comments nodes
+        client = self.apg.client
+        node = await self.host.memory.storage.getPubsubNode(
+            client, service, nodeIdentifier, with_subscriptions=True
+        )
+        if node is None:
+            node = await self.host.memory.storage.setPubsubNode(
+                client,
+                service,
+                nodeIdentifier,
+            )
+            subscription = None
+        else:
+            try:
+                subscription = next(
+                    s for s in node.subscriptions
+                    if s.subscriber == requestor.userhostJID()
+                )
+            except StopIteration:
+                subscription = None
+
+        if subscription is None:
+            subscription = PubsubSub(
+                subscriber=requestor.userhostJID(),
+                state=SubscriptionState.PENDING
+            )
+            node.subscriptions.append(subscription)
+            await self.host.memory.storage.add(node)
+        else:
+            if subscription.state is None:
+                subscription.state = SubscriptionState.PENDING
+                await self.host.memory.storage.add(node)
+            elif subscription.state == SubscriptionState.SUBSCRIBED:
+                log.info(
+                    f"{requestor.userhostJID()} has already a subscription to {node!r} "
+                    f"at {service}. Doing the request anyway."
+                )
+            elif subscription.state == SubscriptionState.PENDING:
+                log.info(
+                    f"{requestor.userhostJID()} has already a pending subscription to "
+                    f"{node!r} at {service}. Doing the request anyway."
+                )
+            else:
+                raise exceptions.InternalError(
+                    f"unmanaged subscription state: {subscription.state}"
+                )
+
+        req_actor_id, recip_actor_id, inbox = await self.getAPActorIdsAndInbox(
+            requestor, service
+        )
+
+        data = self.apg.createActivity("Follow", req_actor_id, recip_actor_id)
+
+        resp = await self.apg.signAndPost(inbox, req_actor_id, data)
+        if resp.code >= 400:
+            text = await resp.text()
+            raise error.StanzaError("service-unavailable", text=text)
+        return pubsub.Subscription(nodeIdentifier, requestor, "subscribed")
+
+    @ensure_deferred
+    async def unsubscribe(self, requestor, service, nodeIdentifier, subscriber):
+        req_actor_id, recip_actor_id, inbox = await self.getAPActorIdsAndInbox(
+            requestor, service
+        )
+        data = self.apg.createActivity(
+            "Undo",
+            req_actor_id,
+            self.apg.createActivity(
+                "Follow",
+                req_actor_id,
+                recip_actor_id
+            )
+        )
+
+        resp = await self.apg.signAndPost(inbox, req_actor_id, data)
+        if resp.code >= 400:
+            text = await resp.text()
+            raise error.StanzaError("service-unavailable", text=text)
+
+    def getConfigurationOptions(self):
+        return NODE_OPTIONS
+
+    def getConfiguration(
+        self,
+        requestor: jid.JID,
+        service: jid.JID,
+        nodeIdentifier: str
+    ) -> defer.Deferred:
+        return defer.succeed(NODE_CONFIG_VALUES)
+
     def getNodeInfo(
         self,
         requestor: jid.JID,
@@ -117,13 +302,6 @@ class APPubsubService(rsm.PubSubService):
             return None
         info = {
             "type": "leaf",
-            "meta-data": [
+            "meta-data": NODE_CONFIG
-                {"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