diff --git a/sat/core/exceptions.py b/sat/core/exceptions.py
index 7646f4fd..9b5ac4dc 100644
--- a/sat/core/exceptions.py
+++ b/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
diff --git a/sat/plugins/plugin_comp_ap_gateway/__init__.py b/sat/plugins/plugin_comp_ap_gateway/__init__.py
index 2c170f1b..7dfdfde5 100644
--- a/sat/plugins/plugin_comp_ap_gateway/__init__.py
+++ b/sat/plugins/plugin_comp_ap_gateway/__init__.py
@@ -17,25 +17,28 @@
# along with this program. If not, see .
import base64
+import calendar
import hashlib
import json
from pathlib import Path
-from typing import Optional, Dict, Tuple, List, Union
-from urllib import parse
-import calendar
+from pprint import pformat
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,
- AP_MB_MAP, LRU_MAX_SIZE)
+from .constants import (
+ 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)
- return type_, parse.unquote(account)
+ type_, *extra_args = path[len(self.ap_path):].lstrip("/").split("/", 1)
+ 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:
- """Sign a documentent and post it to AP server
+ def buildSignatureHeader(self, values: Dict[str, str]) -> str:
+ """Build key="" 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
- @param url_actor: URL generated by this gateway for local actor
- @param doc: document to send
+ return ",".join(fields)
+
+ 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)
- date = http.datetimeToString().decode()
- body = json.dumps(doc).encode()
- digest_hash = base64.b64encode(hashlib.sha256(body).digest()).decode()
- digest = f"sha-256={digest_hash}"
- to_sign = (
- f"(request-target): post {p_url.path}\nhost: {p_url.hostname}\n"
- f"date: {date}\ndigest: {digest}"
- )
+ if algo != "SHA-256":
+ raise NotImplementedError("only SHA-256 is implemented for now")
+ return algo, base64.b64encode(hashlib.sha256(body).digest()).decode()
+
+ @async_lru(maxsize=LRU_MAX_SIZE)
+ async def getActorPubKeyData(
+ self,
+ 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(),
- hashes.SHA256()
+ padding.PKCS1v15(), # type: ignore
+ hashes.SHA256() # type: ignore
)).decode()
- h_signature = (
- f'keyId="{url_actor}",headers="(request-target) host date digest",'
- f'signature="{signature}"'
+ sign_data = {
+ "keyId": key_id,
+ "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={
- "Host": [p_url.hostname],
- "Date": [date],
- "Digest": [digest],
- "Signature": [h_signature],
- }
+ headers=headers,
)
+ 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)
- ap_items = page_data.get("orderedItems")
- if not ap_items:
- log.warning('No "orderedItems" collection found')
- return page_data, []
+ page_data = await self.apGetObject(page)
+ if page_data is None:
+ log.warning('No data found in collection')
+ return {}, []
+ 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]:
- """Convert AP item to microblog data
+ async def apItem2MbDataAndElt(self, ap_item: dict) -> Tuple[dict, domish.Element]:
+ """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")
- if not ap_object:
- log.warning(f'No "object" found in AP item {ap_item!r}')
+ is_activity = self.isActivity(ap_item)
+ if is_activity:
+ 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):
- 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 = {}
+ mb_data = {"id": item_id}
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 not 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
- )
- )
+ if is_activity:
+ authors = await self.apGetActors(ap_item, "actor")
else:
- log.warning(f"unknown actor type found: {actor!r}")
- raise exceptions.DataError
+ authors = await self.apGetActors(ap_object, "attributedTo")
+ 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)
- ap_actor_data = await self.getAPActorDataFromId(account)
+ account = self._e.unescape(service.user)
+ 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])]
+ )
diff --git a/sat/plugins/plugin_comp_ap_gateway/constants.py b/sat/plugins/plugin_comp_ap_gateway/constants.py
index 575b40bc..3cc32193 100644
--- a/sat/plugins/plugin_comp_ap_gateway/constants.py
+++ b/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
diff --git a/sat/plugins/plugin_comp_ap_gateway/http_server.py b/sat/plugins/plugin_comp_ap_gateway/http_server.py
index 771acc21..cf41d31c 100644
--- a/sat/plugins/plugin_comp_ap_gateway/http_server.py
+++ b/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 .
+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,
- AP_REQUEST_TYPES, PAGE_SIZE)
+from .constants import (
+ 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)
- 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_type, extra_args = self.apg.parseAPURL(ap_url)
+ if len(extra_args) == 0:
+ if request_type != "shared_inbox":
+ raise exceptions.DataError(f"Invalid request type: {request_type!r}")
+ ret_data = await self.APInboxRequest(
+ request, None, None, None, ap_url, signing_actor
+ )
+ 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
diff --git a/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py b/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py
index cbc16a02..39b87d26 100644
--- a/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py
+++ b/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 .
-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 []
- if node != self.apg._m.namespace:
+ return [], None
+ 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:
- 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"
+ client = self.apg.client
+ cached_node = await self.host.memory.storage.getPubsubNode(
+ client, service, node
)
- 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": [
- {"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"},
-
- ]
-
+ "meta-data": NODE_CONFIG
}
return info