libervia-backend: libervia/backend/plugins/plugin_comp_ap_gateway/__init_

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4259:49019947cc76

component AP Gateway: implement HTTP GET signature.

author	Goffi <goffi@goffi.org>
date	Wed, 05 Jun 2024 22:34:09 +0200
parents	6784d07b99c8
children	d366d90a71aa

comparison

equal deleted inserted replaced

-:ba28ca268f4a
+:49019947cc76
 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
-from dateutil.parser import parserinfo
 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 error, jid
 def __init__(self, host):
 self.host = host
 self.initialised = False
 self.client = None
+self.http_sign_get = True
 self._p = host.plugins["XEP-0060"]
 self._a = host.plugins["XEP-0084"]
 self._e = host.plugins["XEP-0106"]
 self._m = host.plugins["XEP-0277"]
 self._v = host.plugins["XEP-0292"]
 if connection_type not in ('http', 'https'):
 raise exceptions.ConfigError(
 'bad ap-gateay http_connection_type, you must use one of "http" or '
 '"https"'
 )
+self.http_sign_get = C.bool(
+self.host.memory.config_get(CONF_SECTION, "http_sign_get", C.BOOL_TRUE)
+)
 self.max_items = int(self.host.memory.config_get(
 CONF_SECTION, 'new_node_max_items', 50
 ))
 self.comments_max_depth = int(self.host.memory.config_get(
 await self.convert_and_post_items(
 client, ap_account, itemsEvent.sender, itemsEvent.nodeIdentifier,
 itemsEvent.items
 )
-async def get_virtual_client(self, actor_id: str) -> SatXMPPEntity:
+async def get_virtual_client(
+self,
+requestor_actor_id: str,
+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 requestor_actor_id: originating actor ID (URL)
 @param actor_id: ID of the actor
 @return: virtual client
 """
-local_jid = await self.get_jid_from_id(actor_id)
+local_jid = await self.get_jid_from_id(requestor_actor_id, actor_id)
 return self.client.get_virtual_client(local_jid)
 def is_activity(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 ap_get(self, url: str) -> dict:
+async def ap_get(self, url: str, requestor_actor_id: str) -> dict:
-"""Retrieve AP JSON from given URL
+"""Retrieve AP JSON from given URL with HTTP Signature
+@param url: AP server endpoint
+@param requestor_actor_id: originating actor ID (URL)
 @raise error.StanzaError: "service-unavailable" is sent when something went wrong
 with AP server
 """
-resp = await treq.get(
+if self.http_sign_get:
-url,
+headers = self._generate_signed_headers(url, requestor_actor_id, method="get")
-headers = {
+else:
-"Accept": [MEDIA_TYPE_AP],
+headers = {}
-}
+headers["Accept"] = MEDIA_TYPE_AP
-)
+resp = await treq.get(url, headers=headers)
 if resp.code >= 300:
 text = await resp.text()
 if resp.code == 404:
 raise exceptions.NotFound(f"Can't find resource at {url}")
 else:
 raise error.StanzaError(
 "service-unavailable",
 text=f"Can't get AP data at {url}: {e}"
 )
-@overload
+async def ap_post(self, url: str, requestor_actor_id: str, doc: dict) -> TReqResponse:
-async def ap_get_object(self, data: dict, key: str) -> Optional[dict]:
+"""Sign a document and post it to AP server
-...
+@param url: AP server endpoint
+@param requestor_actor_id: originating actor ID (URL)
+@param doc: document to send
+"""
+if self.verbose:
+__, actor_args = self.parse_apurl(requestor_actor_id)
+actor_account = actor_args[0]
+to_log = [
+"",
+f">>> {actor_account} is signing and posting to {url}:\n{pformat(doc)}"
+]
+body = json.dumps(doc).encode()
+headers = self._generate_signed_headers(url, requestor_actor_id, method="post", body=body)
+headers["Content-Type"] = MEDIA_TYPE_AP
+if self.verbose:
+if self.verbose >= 3:
+h_to_log = "\n".join(f"    {k}: {v}" for k, v in headers.items())
+to_log.append(f"  headers:\n{h_to_log}")
+to_log.append("---")
+log.info("\n".join(to_log))
+resp = await treq.post(
+url,
+body,
+headers=headers
+)
+if resp.code >= 300:
+text = await resp.text()
+log.warning(f"POST request to {url} failed [{resp.code}]: {text}")
+elif self.verbose:
+log.info(f"==> response code: {resp.code}")
+return resp
+def _generate_signed_headers(
+self,
+url: str,
+actor_id: str,
+method: str,
+body: bytes|None = None
+) -> dict[str, str]:
+"""Generate HTTP headers with signature for a given request
+@param url: AP server endpoint
+@param actor_id: originating actor ID (URL)
+@param method: HTTP method (e.g., 'get', 'post')
+@param body: request body if any
+@return: signed headers
+"""
+p_url = parse.urlparse(url)
+headers = {
+"(request-target)": f"{method} {p_url.path}",
+"Host": p_url.hostname,
+"Date": http.datetimeToString().decode()
+}
+if body:
+digest_algo, digest_hash = self.get_digest(body)
+headers["Digest"] = f"{digest_algo}={digest_hash}"
+headers, __ = self.get_signature_data(self.get_key_id(actor_id), headers)
+return headers
 @overload
 async def ap_get_object(
-self, data: Union[str, dict], key: None = None
+self, requestor_actor_id: str, data: dict, key: str
+) -> dict|None:
+...
+@overload
+async def ap_get_object(
+self, requestor_actor_id: str, data: Union[str, dict], key: None = None
 ) -> dict:
 ...
-async def ap_get_object(self, data, key = None):
+async def ap_get_object(self, requestor_actor_id: str, data, key = None) -> dict|None:
 """Retrieve an AP object, dereferencing when necessary
 This method is to be used with attributes marked as "Functional" in
 https://www.w3.org/TR/activitystreams-vocabulary
+@param requestor_actor_id: ID of the actor doing the request.
 @param data: AP object where an other object is looked for, or the object itself
 @param key: name of the object to look for, or None if data is the object directly
 @return: found object if any
 """
 if key is not None:
 return value
 elif isinstance(value, str):
 if self.is_local_url(value):
 return await self.ap_get_local_object(value)
 else:
-return await self.ap_get(value)
+return await self.ap_get(value, requestor_actor_id)
 else:
 raise NotImplementedError(
 "was expecting a string or a dict, got {type(value)}: {value!r}}"
 )
 'only object from "item" URLs can be retrieved for now'
 )
 async def ap_get_list(
 self,
+requestor_actor_id: str,
 data: dict,
 key: str,
 only_ids: bool = False
 ) -> Optional[List[Dict[str, Any]]]:
 """Retrieve a list of objects from AP data, dereferencing when necessary
 This method is to be used with non functional vocabularies. Use ``ap_get_object``
 otherwise.
 If the value is a dictionary, it will be wrapped in a list
+@param requestor_actor_id: ID of the actor doing the request.
 @param data: AP object where a list of objects is looked for
 @param key: key of the list to look for
 @param only_ids: if Trye, only items IDs are retrieved
 @return: list of objects, or None if the key is not present
 """
 return None
 elif isinstance(value, str):
 if self.is_local_url(value):
 value = await self.ap_get_local_object(value)
 else:
-value = await self.ap_get(value)
+value = await self.ap_get(value, requestor_actor_id)
 if isinstance(value, dict):
 return [value]
 if not isinstance(value, list):
 raise ValueError(f"A list was expected, got {type(value)}: {value!r}")
 if only_ids:
 return [
 {"id": v["id"]} if isinstance(v, dict) else {"id": v}
 for v in value
 ]
 else:
-return [await self.ap_get_object(i) for i in value]
+return [await self.ap_get_object(requestor_actor_id, i) for i in value]
 async def ap_get_actors(
 self,
+requestor_actor_id: str,
 data: dict,
 key: str,
 as_account: bool = True
 ) -> List[str]:
 """Retrieve AP actors from data
+@param requestor_actor_id: ID of the actor doing the request.
 @param data: AP object containing a field with actors
 @param key: field to use to retrieve actors
 @param as_account: if True returns account handles, otherwise will return actor
 IDs
 @raise exceptions.DataError: there is not actor data or it is invalid
 if not value:
 raise exceptions.DataError(
 f"list of actors is empty"
 )
 if as_account:
-return [await self.get_ap_account_from_id(actor_id) for actor_id in value]
+return [
+await self.get_ap_account_from_id(requestor_actor_id, actor_id)
+for actor_id in value
+]
 else:
 return value
 async def ap_get_sender_actor(
 self,
+requestor_actor_id: str,
 data: dict,
 ) -> str:
 """Retrieve actor who sent data
 This is done by checking "actor" field first, then "attributedTo" field.
 Only the first found actor is taken into account
+@param requestor_actor_id: ID of the actor doing the request.
 @param data: AP object
 @return: actor id of the sender
 @raise exceptions.NotFound: no actor has been found in data
 """
 try:
-actors = await self.ap_get_actors(data, "actor", as_account=False)
+actors = await self.ap_get_actors(requestor_actor_id, data, "actor", as_account=False)
 except exceptions.DataError:
 actors = None
 if not actors:
 try:
-actors = await self.ap_get_actors(data, "attributedTo", as_account=False)
+actors = await self.ap_get_actors(requestor_actor_id, data, "attributedTo", as_account=False)
 except exceptions.DataError:
 raise exceptions.NotFound(
 'actor not specified in "actor" or "attributedTo"'
 )
 try:
 return (
 ("pubsub", "service") in host_disco.identities
 and not ("pubsub", "pep") in host_disco.identities
 )
-async def get_jid_and_node(self, ap_account: str) -> Tuple[jid.JID, Optional[str]]:
+async def get_jid_and_node(self, ap_account: str) -> tuple[jid.JID, str|None]:
 """Decode raw AP account handle to get XMPP JID and Pubsub Node
 Username are case insensitive.
 By default, the username correspond to local username (i.e. username from
 self.client.jid.host,
 None
 )
 )
-async def get_jid_from_id(self, actor_id: str) -> jid.JID:
+async def get_jid_from_id(self, requestor_actor_id: str, actor_id: str) -> jid.JID:
 """Compute JID linking to an AP Actor ID
 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
 If the actor_id comes from local server (checked with self.public_url), it means
 that we have an XMPP entity, and the original JID is returned
+@param requestor_actor_id: ID of the actor doing the request.
+@param actor_id: ID of the actor to generate JID from.
+@return: generated JID.
 """
 if self.is_local_url(actor_id):
 request_type, extra_args = self.parse_apurl(actor_id)
 if request_type != TYPE_ACTOR or len(extra_args) != 1:
 raise ValueError(f"invalid actor id: {actor_id!r}")
 actor_jid, __ = await self.get_jid_and_node(extra_args[0])
 return actor_jid
-account = await self.get_ap_account_from_id(actor_id)
+account = await self.get_ap_account_from_id(requestor_actor_id, actor_id)
 return self.get_local_jid_from_account(account)
-def parse_apurl(self, url: str) -> Tuple[str, List[str]]:
+def parse_apurl(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 extra arguments
 """
 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 get_actor_data(self, actor_id) -> dict:
+async def get_actor_data(self, requestor_actor_id: str, actor_id: str) -> dict:
 """Retrieve actor data with LRU cache"""
-return await self.ap_get(actor_id)
+return await self.ap_get(actor_id, requestor_actor_id)
 @async_lru(maxsize=LRU_MAX_SIZE)
 async def get_actor_pub_key_data(
 self,
+requestor_actor_id: str,
 actor_id: str
 ) -> Tuple[str, str, rsa.RSAPublicKey]:
 """Retrieve Public Key data from actor ID
+@param requestor_actor_id: ID of the actor doing the request.
 @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.get_actor_data(actor_id)
+actor_data = await self.get_actor_data(requestor_actor_id, 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())
 """Get local key ID from actor ID"""
 return f"{actor_id}#main-key"
 async def check_signature(
 self,
+requestor_actor_id: str,
 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 requestor_actor_id: ID of the actor doing the request.
 @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
 actor = key_id[5:]
 actor_id = await self.get_ap_actor_id_from_account(actor)
 else:
 actor_id = key_id.split("#", 1)[0]
-pub_key_id, pub_key_owner, pub_key = await self.get_actor_pub_key_data(actor_id)
+pub_key_id, pub_key_owner, pub_key = await self.get_actor_pub_key_data(
+requestor_actor_id,
+actor_id
+)
 if pub_key_id != key_id or pub_key_owner != actor_id:
 raise exceptions.EncryptionError("Public Key mismatch")
 try:
 pub_key.verify(
 @param node: (virtual) node corresponding where the item has been published
 @param subscribe_extra_nodes: if True, extra data nodes will be automatically
 subscribed, that is comment nodes if present and attachments nodes.
 """
 actor_id = await self.get_ap_actor_id_from_account(ap_account)
-inbox = await self.get_ap_inbox_from_id(actor_id)
+requestor_actor_id = self.build_apurl(
+TYPE_ACTOR,
+await self.get_ap_account_from_jid_and_node(service, node)
+)
+inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id)
 for item in items:
 if item.name == "item":
 cached_item = await self.host.memory.storage.search_pubsub_items({
 "profiles": [self.client.profile],
 "services": [service],
 url_actor, ap_item = await self.ap_delete_item(
 client.jid, node, item["id"]
 )
 else:
 raise exceptions.InternalError(f"unexpected element: {item.toXml()}")
-await self.sign_and_post(inbox, url_actor, ap_item)
+await self.ap_post(inbox, url_actor, ap_item)
 async def convert_and_post_attachments(
 self,
 client: SatXMPPEntity,
 ap_account: str,
 "we should get exactly one attachment item for an entity, got "
 f"{len(items)})"
 )
 actor_id = await self.get_ap_actor_id_from_account(ap_account)
-inbox = await self.get_ap_inbox_from_id(actor_id)
+requestor_actor_id = self.build_apurl(
+TYPE_ACTOR,
+await self.get_ap_account_from_jid_and_node(service, node)
+)
+inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id)
 item_elt = items[0]
 item_id = item_elt["id"]
 if publisher is None:
 if publisher is not None:
 item_elt["publisher"] = publisher.userhost()
 item_service, item_node, item_id = self._pa.attachment_node_2_item(node)
-item_account = await self.get_ap_account_from_jid_and_node(item_service, item_node)
+item_account = await self.get_ap_account_from_jid_and_node(
+item_service, item_node
+)
 if self.is_virtual_jid(item_service):
 # it's a virtual JID mapping to an external AP actor, we can use the
 # item_id directly
 item_url = item_id
 if not item_url.startswith("https:"):
 })
 if not old_attachment_pubsub_items:
 old_attachment = {}
 else:
 old_attachment_items = [i.data for i in old_attachment_pubsub_items]
-old_attachments = self._pa.items_2_attachment_data(client, old_attachment_items)
+old_attachments = self._pa.items_2_attachment_data(
+client, old_attachment_items
+)
 try:
 old_attachment = old_attachments[0]
 except IndexError:
 # no known element was present in attachments
 old_attachment = {}
 activity = self.create_activity(
 TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
 )
 activity["to"] = [ap_account]
 activity["cc"] = [NS_AP_PUBLIC]
-await self.sign_and_post(inbox, publisher_actor_id, activity)
+await self.ap_post(inbox, publisher_actor_id, activity)
 else:
 if "noticed" in old_attachment:
 # "noticed" attachment has been removed, we undo the "Like" activity
 activity_id = self.build_apurl("like", item_account, item_id)
 activity = self.create_activity(
 TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
 )
 activity["to"] = [ap_account]
 activity["cc"] = [NS_AP_PUBLIC]
 undo = self.create_activity("Undo", publisher_actor_id, activity)
-await self.sign_and_post(inbox, publisher_actor_id, undo)
+await self.ap_post(inbox, publisher_actor_id, undo)
 # reactions
 new_reactions = set(attachments.get("reactions", {}).get("reactions", []))
 old_reactions = set(old_attachment.get("reactions", {}).get("reactions", []))
 reactions_remove = old_reactions - new_reactions
 activy = self.create_activity(
 "Undo", publisher_actor_id, reaction_activity
 )
 else:
 activy = reaction_activity
-await self.sign_and_post(inbox, publisher_actor_id, activy)
+await self.ap_post(inbox, publisher_actor_id, activy)
 # RSVP
 if "rsvp" in attachments:
 attending = attachments["rsvp"].get("attending", "no")
 old_attending = old_attachment.get("rsvp", {}).get("attending", "no")
 activity = self.create_activity(
 activity_type, publisher_actor_id, item_url, activity_id=activity_id
 )
 activity["to"] = [ap_account]
 activity["cc"] = [NS_AP_PUBLIC]
-await self.sign_and_post(inbox, publisher_actor_id, activity)
+await self.ap_post(inbox, publisher_actor_id, activity)
 else:
 if "rsvp" in old_attachment:
 old_attending = old_attachment.get("rsvp", {}).get("attending", "no")
 if old_attending == "yes":
 activity_id = self.build_apurl(TYPE_LEAVE.lower(), item_account, item_id)
 activity = self.create_activity(
 TYPE_LEAVE, publisher_actor_id, item_url, activity_id=activity_id
 )
 activity["to"] = [ap_account]
 activity["cc"] = [NS_AP_PUBLIC]
-await self.sign_and_post(inbox, publisher_actor_id, activity)
+await self.ap_post(inbox, publisher_actor_id, activity)
 if service.user and self.is_virtual_jid(service):
 # the item is on a virtual service, we need to store it in cache
 log.debug("storing attachments item in cache")
 cached_node = await self.host.memory.storage.get_pubsub_node(
 self.client,
 cached_node,
 [item_elt],
 [attachments]
 )
-async def sign_and_post(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
-"""
-if self.verbose:
-__, actor_args = self.parse_apurl(actor_id)
-actor_account = actor_args[0]
-to_log = [
-"",
-f">>> {actor_account} is signing and posting to {url}:\n{pformat(doc)}"
-]
-p_url = parse.urlparse(url)
-body = json.dumps(doc).encode()
-digest_algo, digest_hash = self.get_digest(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["Content-Type"] = (
-MEDIA_TYPE_AP
-)
-headers, __ = self.get_signature_data(self.get_key_id(actor_id), headers)
-if self.verbose:
-if self.verbose>=3:
-h_to_log = "\n".join(f"    {k}: {v}" for k,v in headers.items())
-to_log.append(f"  headers:\n{h_to_log}")
-to_log.append("---")
-log.info("\n".join(to_log))
-resp = await treq.post(
-url,
-body,
-headers=headers,
-)
-if resp.code >= 300:
-text = await resp.text()
-log.warning(f"POST request to {url} failed [{resp.code}]: {text}")
-elif self.verbose:
-log.info(f"==> response code: {resp.code}")
-return resp
 def _publish_message(self, mess_data_s: str, service_s: str, profile: str):
 mess_data: dict = data_format.deserialise(mess_data_s) # type: ignore
 service = jid.JID(service_s)
 client = self.host.get_client(profile)
 raise ValueError(
 f"No ActivityPub link found for {account!r}"
 )
 return href
-async def get_ap_actor_data_from_account(self, account: str) -> dict:
+async def get_ap_actor_data_from_account(
+self,
+requestor_actor_id: str,
+account: str
+) -> dict:
 """Retrieve ActivityPub Actor data
 @param account: ActivityPub Actor identifier
 """
 href = await self.get_ap_actor_id_from_account(account)
-return await self.ap_get(href)
+return await self.ap_get(href, requestor_actor_id)
-async def get_ap_inbox_from_id(self, actor_id: str, use_shared: bool = True) -> str:
+async def get_ap_inbox_from_id(
+self,
+requestor_actor_id: str,
+actor_id: str,
+use_shared: bool = True
+) -> str:
 """Retrieve inbox of an actor_id
+@param requestor_actor_id: ID of the actor doing the request.
+@param actor_id: ID of the actor from whom Inbox must be retrieved.
 @param use_shared: if True, and a shared inbox exists, it will be used instead of
 the user inbox
 """
-data = await self.get_actor_data(actor_id)
+data = await self.get_actor_data(requestor_actor_id, actor_id)
 if use_shared:
 try:
 return data["endpoints"]["sharedInbox"]
 except KeyError:
 pass
 return data["inbox"]
 @async_lru(maxsize=LRU_MAX_SIZE)
-async def get_ap_account_from_id(self, actor_id: str) -> str:
+async def get_ap_account_from_id(self, requestor_actor_id: str, actor_id: str) -> str:
 """Retrieve AP account from the ID URL
 Works with external or local actor IDs.
+@param requestor_actor_id: ID of the actor doing the request.
 @param actor_id: AP ID of the actor (URL to the actor data)
 @return: AP handle
 """
 if self.is_local_url(actor_id):
 url_type, url_args = self.parse_apurl(actor_id)
 f"{account!r} is not a valid local account (from {actor_id})"
 )
 return account
 url_parsed = parse.urlparse(actor_id)
-actor_data = await self.get_actor_data(actor_id)
+actor_data = await self.get_actor_data(requestor_actor_id, actor_id)
 username = actor_data.get("preferredUsername")
 if not username:
 raise exceptions.DataError(
 'No "preferredUsername" field found, can\'t retrieve actor account'
 )
 raise exceptions.DataError(msg)
 return account
 async def get_ap_items(
 self,
+actor_id: str,
 collection: dict,
 max_items: Optional[int] = None,
 chronological_pagination: bool = True,
 after_id: Optional[str] = None,
 start_index: Optional[int] = None,
-parser: Optional[Callable[[dict], Awaitable[domish.Element]]] = None,
+parser: Optional[Callable[[str, dict], Awaitable[domish.Element]]] = None,
 only_ids: bool = False,
 ) -> Tuple[List[domish.Element], rsm.RSMResponse]:
 """Retrieve AP items and convert them to XMPP items
-@param account: AP account handle to get items from
+@param actor_id: ID of the actor doing the request.
+@param collection: AP collection data.
+Items will be retrieved from this collection.
 @param max_items: maximum number of items to retrieve
 retrieve all items by default
 @param chronological_pagination: get pages in chronological order
 AP use reversed chronological order for pagination, "first" page returns more
 recent items. If "chronological_pagination" is True, "last" AP page will be
 previous_index = start_index - 1
 retrieved_items = 0
 current_page = collection["last"]
 while retrieved_items < count:
 page_data, items = await self.parse_ap_page(
-current_page, parser, only_ids
+actor_id, current_page, parser, only_ids
 )
 if not items:
 log.warning(f"found an empty AP page at {current_page}")
 return [], rsm_resp
 page_start_idx = retrieved_items
 page_items = []
 retrieved_items = 0
 found_after_id = False
 while retrieved_items < count:
-__, page_items = await self.parse_ap_page(page, parser, only_ids)
+__, page_items = await self.parse_ap_page(actor_id, page, parser, only_ids)
 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
 "last": items[-1]["id"]
 })
 return items, rsm.RSMResponse(**rsm_resp)
-async def ap_item_2_mb_data_and_elt(self, ap_item: dict) -> Tuple[dict, domish.Element]:
+async def ap_item_2_mb_data_and_elt(self, requestor_actor_id: str, ap_item: dict) -> tuple[dict, domish.Element]:
-"""Convert AP item to parsed microblog data and corresponding item element"""
+"""Convert AP item to parsed microblog data and corresponding item element
-mb_data = await self.ap_item_2_mb_data(ap_item)
+@param requestor_actor_id: ID of the actor requesting the conversion.
+@param ap_item: AP item to convert.
+@return: microblog and correspondign <item> element.
+"""
+mb_data = await self.ap_item_2_mb_data(requestor_actor_id, ap_item)
 item_elt = await self._m.mb_data_2_entry_elt(
 self.client, mb_data, mb_data["id"], None, self._m.namespace
 )
 if "repeated" in mb_data["extra"]:
 item_elt["publisher"] = mb_data["extra"]["repeated"]["by"]
 else:
 item_elt["publisher"] = mb_data["author_jid"]
 return mb_data, item_elt
-async def ap_item_2_mb_elt(self, ap_item: dict) -> domish.Element:
+async def ap_item_2_mb_elt(self, requestor_actor_id: str, ap_item: dict) -> domish.Element:
-"""Convert AP item to XMPP item element"""
+"""Convert AP item to XMPP item element
-__, item_elt = await self.ap_item_2_mb_data_and_elt(ap_item)
+@param requestor_actor_id: ID of the actor requesting the conversion.
+@param ap_item: AP item to convert.
+@return: <item> element
+"""
+__, item_elt = await self.ap_item_2_mb_data_and_elt(requestor_actor_id, ap_item)
 return item_elt
 async def parse_ap_page(
 self,
+requestor_actor_id: str,
 page: Union[str, dict],
-parser: Callable[[dict], Awaitable[domish.Element]],
+parser: Callable[[str, dict], Awaitable[domish.Element]],
 only_ids: bool = False
 ) -> Tuple[dict, List[domish.Element]]:
 """Convert AP objects from an AP page to XMPP items
+@param requestor_actor_id: ID of the actor doing the request.
 @param page: Can be either url linking and AP page, or the page data directly
 @param parser: method to use to parse AP items and get XMPP item elements
 @param only_ids: if True, only retrieve items IDs
 @return: page data, pubsub items
 """
-page_data = await self.ap_get_object(page)
+page_data = await self.ap_get_object(requestor_actor_id, page)
 if page_data is None:
 log.warning('No data found in collection')
 return {}, []
-ap_items = await self.ap_get_list(page_data, "orderedItems", only_ids=only_ids)
+ap_items = await self.ap_get_list(requestor_actor_id, page_data, "orderedItems", only_ids=only_ids)
 if ap_items is None:
-ap_items = await self.ap_get_list(page_data, "items", only_ids=only_ids)
+ap_items = await self.ap_get_list(requestor_actor_id, page_data, "items", only_ids=only_ids)
 if not ap_items:
 log.warning(f'No item field found in collection: {page_data!r}')
 return page_data, []
 else:
 log.warning(
 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:
-items.append(await parser(ap_item))
+items.append(await parser(requestor_actor_id, ap_item))
 except (exceptions.DataError, NotImplementedError, error.StanzaError):
 continue
 return page_data, items
 async def get_comments_nodes(
 self,
+requestor_actor_id: str,
 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 requestor_actor_id: ID of the actor doing the request.
 @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.
 self._m.get_comments_node(item_id)
 )
 parent_url = parent_id
 parents = []
 for __ in range(COMMENTS_MAX_PARENTS):
-parent_item = await self.ap_get(parent_url)
+parent_item = await self.ap_get(parent_url, requestor_actor_id)
 parents.insert(0, parent_item)
 parent_url = parent_item.get("inReplyTo")
 if parent_url is None:
 break
 parent_limit = self.comments_max_depth-1
 return (
 self._m.get_comments_node(last_level_item["id"]),
 None
 )
-async def ap_item_2_mb_data(self, ap_item: dict) -> dict:
+async def ap_item_2_mb_data(self, requestor_actor_id: str, ap_item: dict) -> dict:
 """Convert AP activity or object to microblog data
+@param actor_id: ID of the actor doing the request.
 @param ap_item: ActivityPub item to convert
 Can be either an activity of an object
 @return: AP Item's Object and microblog data
 @raise exceptions.DataError: something is invalid in the AP item
 @raise NotImplementedError: some AP data is not handled yet
 @raise error.StanzaError: error while contacting the AP server
 """
 is_activity = self.is_activity(ap_item)
 if is_activity:
-ap_object = await self.ap_get_object(ap_item, "object")
+ap_object = await self.ap_get_object(requestor_actor_id, 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
 attachment[key] = value
 attachments.append(attachment)
 # author
 if is_activity:
-authors = await self.ap_get_actors(ap_item, "actor")
+authors = await self.ap_get_actors(requestor_actor_id, ap_item, "actor")
 else:
-authors = await self.ap_get_actors(ap_object, "attributedTo")
+authors = await self.ap_get_actors(requestor_actor_id, 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")
 if "_repeated" in ap_item:
 mb_data["extra"]["repeated"] = ap_item["_repeated"]
 # comments
 in_reply_to = ap_object.get("inReplyTo")
-__, comments_node = await self.get_comments_nodes(item_id, in_reply_to)
+__, comments_node = await self.get_comments_nodes(
+requestor_actor_id, item_id, in_reply_to
+)
 if comments_node is not None:
 comments_data = {
 "service": author_jid,
 "node": comments_node,
 "uri": uri.build_xmpp_uri(
 target_ap_account = await self.get_ap_account_from_jid_and_node(
 service, node
 )
 if self.is_virtual_jid(service):
 # service is a proxy JID for AP account
-actor_data = await self.get_ap_actor_data_from_account(target_ap_account)
+actor_data = await self.get_ap_actor_data_from_account(
+url_actor,
+target_ap_account
+)
 followers = actor_data.get("followers")
 else:
 # service is a real XMPP entity
 followers = self.build_apurl(TYPE_FOLLOWERS, target_ap_account)
 if followers:
 except KeyError:
 raise exceptions.DataError("Can't get ActivityPub actor inbox")
 item_data = await self.mb_data_2_ap_item(client, mess_data)
 url_actor = item_data["actor"]
-resp = await self.sign_and_post(inbox_url, url_actor, item_data)
+await self.ap_post(inbox_url, url_actor, item_data)
 async def ap_delete_item(
 self,
 jid_: jid.JID,
 node: Optional[str],
 """add the gateway workflow on post treatment"""
 if self.client is None:
 log.debug(f"no client set, ignoring message: {message_elt.toXml()}")
 return True
 post_treat.addCallback(
-lambda mess_data: defer.ensureDeferred(self.onMessage(client, mess_data))
+lambda mess_data: defer.ensureDeferred(self.on_message(client, mess_data))
 )
 return True
-async def onMessage(self, client: SatXMPPEntity, mess_data: dict) -> dict:
+async def on_message(self, client: SatXMPPEntity, mess_data: dict) -> dict:
 """Called once message has been parsed
 this method handle the conversion to AP items and posting
 """
 if client != self.client:
 if not mess_data["to"].user:
 log.warning(
 f"ignoring message addressed to gateway itself: {mess_data}"
 )
 return mess_data
+requestor_actor_id = self.build_apurl(TYPE_ACTOR, mess_data["from"].userhost())
 actor_account = self._e.unescape(mess_data["to"].user)
 try:
 actor_id = await self.get_ap_actor_id_from_account(actor_account)
 except Exception as e:
 log.warning(
 f"Can't retrieve data on actor {actor_account}: {e}"
 )
 # TODO: send an error <message>
 return mess_data
-inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+inbox = await self.get_ap_inbox_from_id(
+requestor_actor_id, actor_id, use_shared=False
+)
 try:
 language, message = next(iter(mess_data["message"].items()))
 except (KeyError, StopIteration):
 log.warning(f"ignoring empty message: {mess_data}")
 "href": actor_id,
 "name": f"@{actor_account}",
 })
 try:
-await self.sign_and_post(inbox, ap_item["actor"], ap_item)
+await self.ap_post(inbox, ap_item["actor"], ap_item)
 except Exception as e:
 # TODO: send an error <message>
 log.warning(
 f"Can't send message to {inbox}: {e}"
 )
 if not self.is_local(from_jid):
 log.debug(
 f"ignoring retract request from non local jid {from_jid}"
 )
 return False
+requestor_actor_id = self.build_apurl(
+TYPE_ACTOR,
+from_jid.userhost()
+)
 to_jid = jid.JID(message_elt["to"])
 if (to_jid.host != self.client.jid.full() or not to_jid.user):
 # to_jid should be a virtual JID from this gateway
 raise exceptions.InternalError(
 f"Invalid destinee's JID: {to_jid.full()}"
 )
 ap_account = self._e.unescape(to_jid.user)
 actor_id = await self.get_ap_actor_id_from_account(ap_account)
-inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id, use_shared=False)
 url_actor, ap_item = await self.ap_delete_item(
 from_jid.userhostJID(), None, retract_elt["id"], public=False
 )
-resp = await self.sign_and_post(inbox, url_actor, ap_item)
+resp = await self.ap_post(inbox, url_actor, ap_item)
 return False
 async def _on_reference_received(
 self,
 client: SatXMPPEntity,
 "type": TYPE_MENTION,
 "href": actor_id,
 "name": ap_account,
 })
-inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+requestor_actor_id = ap_item["actor"]
+inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id, use_shared=False)
-resp = await self.sign_and_post(inbox, ap_item["actor"], ap_item)
+await self.ap_post(inbox, requestor_actor_id, ap_item)
 return False
 async def new_reply_to_xmpp_item(
 self,
 # we don't have a comment node set for this item
 from libervia.backend.tools.xml_tools import pp_elt
 log.info(f"{pp_elt(parent_item_elt.toXml())}")
 raise NotImplementedError()
 else:
-__, item_elt = await self.ap_item_2_mb_data_and_elt(ap_item)
+requestor_actor_id = self.build_apurl(
+TYPE_ACTOR,
+await self.get_ap_account_from_jid_and_node(comment_service, comment_node)
+)
+__, item_elt = await self.ap_item_2_mb_data_and_elt(
+requestor_actor_id,
+ap_item
+)
 await self._p.publish(client, comment_service, comment_node, [item_elt])
 await self.notify_mentions(
 targets, mentions, comment_service, comment_node, item_elt["id"]
 )
 continue
 if not self.is_local_url(value):
 continue
 target_type = self.parse_apurl(value)[0]
 if target_type != TYPE_ACTOR:
-log.debug(f"ignoring non actor type as a target: {href}")
+log.debug(f"ignoring non actor type as a target: {value}")
 else:
 targets.setdefault(target_type, set()).add(value)
 mentions = []
 tags = item.get("tag")
 else:
 await self.handle_pubsub_ap_item(
 client, targets, mentions, destinee, node, item, is_public
 )
+def get_requestor_actor_id_from_targets(
+self,
+targets: set[str]
+) -> str:
+"""Find local actor to use as requestor_actor_id from request targets.
+A local actor must be used to sign HTTP request, notably HTTP GET request for AP
+instance checking signature, such as Mastodon when set in "secure mode".
+This method check a set of targets and use the first local one.
+If none match, a generic local actor is used.
+@param targets: set of actor IDs to which the current request is sent.
+@return: local actor ID to use as requestor_actor_id.
+"""
+try:
+return next(t for t in targets if self.is_local_url(t))
+except StopIteration:
+log.warning(
+f"Can't find local target to use as requestor ID: {targets!r}"
+)
+return self.build_apurl(
+TYPE_ACTOR, f"libervia@{self.public_url}"
+)
 async def handle_message_ap_item(
 self,
 client: SatXMPPEntity,
-targets: Dict[str, Set[str]],
+targets: dict[str, Set[str]],
-mentions: List[Dict[str, str]],
+mentions: list[Dict[str, str]],
-destinee: Optional[jid.JID],
+destinee: jid.JID|None,
 item: dict,
 ) -> None:
 """Parse and deliver direct AP items translating to XMPP messages
 @param targets: actors where the item must be delivered
 @param destinee: jid of the destinee,
 @param item: AP object payload
 """
+targets_urls = {t for t_set in targets.values() for t in t_set}
+requestor_actor_id = self.get_requestor_actor_id_from_targets(targets_urls)
 targets_jids = {
-await self.get_jid_from_id(t)
+await self.get_jid_from_id(requestor_actor_id, url)
-for t_set in targets.values()
+for url in targets_urls
-for t in t_set
 }
 if destinee is not None:
 targets_jids.add(destinee)
-mb_data = await self.ap_item_2_mb_data(item)
+mb_data = await self.ap_item_2_mb_data(requestor_actor_id, item)
 extra = {
 "origin_id": mb_data["id"]
 }
 attachments = mb_data["extra"].get(C.KEY_ATTACHMENTS)
 if attachments:
 )
 await defer.DeferredList(defer_l)
 async def notify_mentions(
 self,
-targets: Dict[str, Set[str]],
+targets: dict[str, set[str]],
-mentions: List[Dict[str, str]],
+mentions: list[dict[str, str]],
 service: jid.JID,
 node: str,
 item_id: str,
 ) -> None:
 """Send mention notifications to recipients and mentioned entities
 Mentions are also sent to recipients as they are primary audience (see
 https://www.w3.org/TR/activitystreams-vocabulary/#microsyntaxes).
 """
+targets_urls = {t for t_set in targets.values() for t in t_set}
+requestor_actor_id = self.get_requestor_actor_id_from_targets(targets_urls)
 anchor = uri.build_xmpp_uri("pubsub", path=service.full(), node=node, item=item_id)
 seen = set()
 # we start with explicit mentions because mentions' content will be used in the
 # future to fill "begin" and "end" reference attributes (we can't do it at the
 # moment as there is no way to specify the XML element to use in the blog item).
 for mention in mentions:
-mentioned_jid = await self.get_jid_from_id(mention["uri"])
+mentioned_jid = await self.get_jid_from_id(requestor_actor_id, mention["uri"])
 self._refs.send_reference(
 self.client,
 to_jid=mentioned_jid,
 anchor=anchor
 )
 seen.add(mentioned_jid)
 remaining = {
-await self.get_jid_from_id(t)
+await self.get_jid_from_id(requestor_actor_id, t)
 for t_set in targets.values()
 for t in t_set
 } - seen
 for target in remaining:
 self._refs.send_reference(
 )
 async def handle_pubsub_ap_item(
 self,
 client: SatXMPPEntity,
-targets: Dict[str, Set[str]],
+targets: dict[str, set[str]],
-mentions: List[Dict[str, str]],
+mentions: list[dict[str, str]],
-destinee: Optional[jid.JID],
+destinee: jid.JID|None,
 node: str,
 item: dict,
 public: bool
 ) -> None:
 """Analyse, cache and deliver AP items translating to Pubsub
 @param node: XMPP pubsub node
 @param item: AP object payload
 @param public: True if the item is public
 """
 # XXX: "public" is not used for now
+targets_urls = {t for t_set in targets.values() for t in t_set}
+requestor_actor_id = self.get_requestor_actor_id_from_targets(targets_urls)
 service = client.jid
 in_reply_to = item.get("inReplyTo")
 if in_reply_to and isinstance(in_reply_to, list):
 in_reply_to = in_reply_to[0]
 await self.new_reply_to_xmpp_item(client, item, targets, mentions)
 return
 # this item is a reply to an AP item, we use or create a corresponding node
 # for comments
-parent_node, __ = await self.get_comments_nodes(item["id"], in_reply_to)
+parent_node, __ = await self.get_comments_nodes(
+requestor_actor_id,
+item["id"],
+in_reply_to
+)
 node = parent_node or node
 cached_node = await self.host.memory.storage.get_pubsub_node(
 client, service, node, with_subscriptions=True, create=True,
 create_kwargs={"subscribed": True}
 )
 f"due to a cache purge. We synchronise the node\n{item}"
 )
 return
 if item.get("type") == TYPE_EVENT:
-data, item_elt = await self.ap_events.ap_item_2_event_data_and_elt(item)
+data, item_elt = await self.ap_events.ap_item_2_event_data_and_elt(
-else:
+requestor_actor_id,
-data, item_elt = await self.ap_item_2_mb_data_and_elt(item)
+item
+)
+else:
+data, item_elt = await self.ap_item_2_mb_data_and_elt(
+requestor_actor_id,
+item
+)
 await self.host.memory.storage.cache_pubsub_items(
 client,
 cached_node,
 [item_elt],
 [data]

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py @ 4259:49019947cc76

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4259:49019947cc76