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

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	d366d90a71aa
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 TYPE_LIKE,
 TYPE_MENTION,
 TYPE_REACTION,
 TYPE_TOMBSTONE,
 TYPE_JOIN,
-TYPE_LEAVE
+TYPE_LEAVE,
 )
 from .http_server import HTTPServer
 from .pubsub_service import APPubsubService
 from .regex import RE_MENTION
 C.PI_IMPORT_NAME: IMPORT_NAME,
 C.PI_MODES: [C.PLUG_MODE_COMPONENT],
 C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT,
 C.PI_PROTOCOLS: [],
 C.PI_DEPENDENCIES: [
-"XEP-0050", "XEP-0054", "XEP-0060", "XEP-0084", "XEP-0106", "XEP-0277",
+"XEP-0050",
-"XEP-0292", "XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "XEP-0470",
+"XEP-0054",
-"XEP-0447", "XEP-0471", "PUBSUB_CACHE", "TEXT_SYNTAXES", "IDENTITY"
+"XEP-0060",
+"XEP-0084",
+"XEP-0106",
+"XEP-0277",
+"XEP-0292",
+"XEP-0329",
+"XEP-0372",
+"XEP-0424",
+"XEP-0465",
+"XEP-0470",
+"XEP-0447",
+"XEP-0471",
+"PUBSUB_CACHE",
+"TEXT_SYNTAXES",
+"IDENTITY",
 ],
 C.PI_RECOMMENDATIONS: [],
 C.PI_MAIN: "APGateway",
 C.PI_HANDLER: C.BOOL_TRUE,
 C.PI_DESCRIPTION: _(
 "",
 items_cb=self._items_received,
 # we want to be sure that the callbacks are launched before pubsub cache's
 # one, as we need to inspect items before they are actually removed from cache
 # or updated
-priority=1000
+priority=1000,
 )
 self.pubsub_service = APPubsubService(self)
 self.ad_hoc = APAdHocService(self)
 self.ap_events = APEvents(self)
 host.trigger.add_with_check(
-"message_received",
+"message_received", self, self._message_received_trigger, priority=-1000
-self,
-self._message_received_trigger,
-priority=-1000
 )
 host.trigger.add_with_check(
-"XEP-0424_retract_received",
+"XEP-0424_retract_received", self, self._on_message_retract
-self,
-self._on_message_retract
 )
 host.trigger.add_with_check(
-"XEP-0372_ref_received",
+"XEP-0372_ref_received", self, self._on_reference_received
-self,
-self._on_reference_received
 )
 host.bridge.add_method(
 "ap_send",
 ".plugin",
 key_size=4096,
 )
 private_key_pem = self.private_key.private_bytes(
 encoding=serialization.Encoding.PEM,
 format=serialization.PrivateFormat.PKCS8,
-encryption_algorithm=serialization.NoEncryption()
+encryption_algorithm=serialization.NoEncryption(),
 ).decode()
 await self.host.memory.storage.set_private_value(
 IMPORT_NAME, "rsa_key", private_key_pem, profile=client.profile
 )
 else:
 password=None,
 )
 self.public_key = self.private_key.public_key()
 self.public_key_pem = self.public_key.public_bytes(
 encoding=serialization.Encoding.PEM,
-format=serialization.PublicFormat.SubjectPublicKeyInfo
+format=serialization.PublicFormat.SubjectPublicKeyInfo,
 ).decode()
 # params
 # URL and port
 self.public_url = self.host.memory.config_get(
 CONF_SECTION, "public_url"
-) or self.host.memory.config_get(
+) or self.host.memory.config_get(CONF_SECTION, "xmpp_domain")
-CONF_SECTION, "xmpp_domain"
-)
 if self.public_url is None:
 log.error(
 '"public_url" not set in configuration, this is mandatory to have'
 "ActivityPub Gateway running. Please set this option it to public facing "
 f"url in {CONF_SECTION!r} configuration section."
 )
 return
 if parse.urlparse(self.public_url).scheme:
 log.error(
-"Scheme must not be specified in \"public_url\", please remove it from "
+'Scheme must not be specified in "public_url", please remove it from '
-"\"public_url\" configuration option. ActivityPub Gateway won't be run."
+'"public_url" configuration option. ActivityPub Gateway won\'t be run.'
 )
 return
-self.http_port = int(self.host.memory.config_get(
+self.http_port = int(self.host.memory.config_get(CONF_SECTION, "http_port", 8123))
-CONF_SECTION, 'http_port', 8123))
 connection_type = self.host.memory.config_get(
-CONF_SECTION, 'http_connection_type', 'https')
+CONF_SECTION, "http_connection_type", "https"
-if connection_type not in ('http', 'https'):
+)
+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(
+self.max_items = int(
-CONF_SECTION, 'new_node_max_items', 50
+self.host.memory.config_get(CONF_SECTION, "new_node_max_items", 50)
+)
-))
+self.comments_max_depth = int(
-self.comments_max_depth = int(self.host.memory.config_get(
+self.host.memory.config_get(CONF_SECTION, "comments_max_depth", 0)
-CONF_SECTION, 'comments_max_depth', 0
+)
-))
+self.ap_path = self.host.memory.config_get(CONF_SECTION, "ap_path", "_ap")
-self.ap_path = self.host.memory.config_get(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
 self.local_only = C.bool(
-self.host.memory.config_get(CONF_SECTION, 'local_only', C.BOOL_TRUE)
+self.host.memory.config_get(CONF_SECTION, "local_only", C.BOOL_TRUE)
 )
 # if True (default), mention will be parsed in non-private content coming from
 # XMPP. This is necessary as XEP-0372 are coming separately from item where the
 # mention is done, which is hard to impossible to translate to ActivityPub (where
 # mention specified inside the item directly). See documentation for details.
 self.auto_mentions = C.bool(
 self.host.memory.config_get(CONF_SECTION, "auto_mentions", C.BOOL_TRUE)
 )
 html_redirect: Dict[str, Union[str, dict]] = self.host.memory.config_get(
-CONF_SECTION, 'html_redirect_dict', {}
+CONF_SECTION, "html_redirect_dict", {}
 )
 self.html_redirect: Dict[str, List[dict]] = {}
 for url_type, target in html_redirect.items():
 if isinstance(target, str):
 target = {"url": target}
 filters["node"] = node_filter
 self.html_redirect.setdefault(url_type, []).append(target)
 # HTTP server launch
 self.server = HTTPServer(self)
-if connection_type == 'http':
+if connection_type == "http":
 reactor.listenTCP(self.http_port, self.server)
 else:
-options = tls.get_options_from_config(
+options = tls.get_options_from_config(self.host.memory.config, CONF_SECTION)
-self.host.memory.config, CONF_SECTION)
 tls.tls_options_check(options)
 context_factory = tls.get_tls_context_factory(options)
 reactor.listenSSL(self.http_port, self.server, context_factory)
 async def profile_connecting(self, client):
 self.client = client
 client.sendHistory = True
 client._ap_storage = persistent.LazyPersistentBinaryDict(
-IMPORT_NAME,
+IMPORT_NAME, client.profile
-client.profile
 )
 await self.init(client)
 def profile_connected(self, client):
 self.ad_hoc.init(client)
 async def _items_received(
-self,
+self, client: SatXMPPEntity, itemsEvent: pubsub.ItemsEvent
-client: SatXMPPEntity,
-itemsEvent: pubsub.ItemsEvent
 ) -> None:
 """Callback called when pubsub items are received
 if the items are adressed to a JID corresponding to an AP actor, they are
 converted to AP items and sent to the corresponding AP server.
 ap_account = self._e.unescape(recipient.user)
 if self._pa.is_attachment_node(itemsEvent.nodeIdentifier):
 await self.convert_and_post_attachments(
-client, ap_account, itemsEvent.sender, itemsEvent.nodeIdentifier,
+client,
-itemsEvent.items
+ap_account,
+itemsEvent.sender,
+itemsEvent.nodeIdentifier,
+itemsEvent.items,
 )
 else:
 await self.convert_and_post_items(
-client, ap_account, itemsEvent.sender, itemsEvent.nodeIdentifier,
+client,
-itemsEvent.items
+ap_account,
+itemsEvent.sender,
+itemsEvent.nodeIdentifier,
+itemsEvent.items,
 )
 async def get_virtual_client(
-self,
+self, requestor_actor_id: str, actor_id: str
-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.
 raise exceptions.ExternalRequestError(msg)
 try:
 return await treq.json_content(resp)
 except Exception as e:
 raise error.StanzaError(
-"service-unavailable",
+"service-unavailable", text=f"Can't get AP data at {url}: {e}"
-text=f"Can't get AP data at {url}: {e}"
 )
 async def ap_post(self, url: str, requestor_actor_id: str, doc: dict) -> TReqResponse:
 """Sign a document and post it to AP server
 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)}"
+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 = 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(
+resp = await treq.post(url, body, headers=headers)
-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,
+self, url: str, actor_id: str, method: str, body: bytes | None = None
-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)
 """
 p_url = parse.urlparse(url)
 headers = {
 "(request-target)": f"{method} {p_url.path}",
 "Host": p_url.hostname,
-"Date": http.datetimeToString().decode()
+"Date": http.datetimeToString().decode(),
 }
 if body:
 digest_algo, digest_hash = self.get_digest(body)
 headers["Digest"] = f"{digest_algo}={digest_hash}"
 return headers
 @overload
 async def ap_get_object(
 self, requestor_actor_id: str, data: dict, key: str
-) -> dict|None:
+) -> dict | None: ...
-...
 @overload
 async def ap_get_object(
 self, requestor_actor_id: str, data: Union[str, dict], key: None = None
-) -> dict:
+) -> dict: ...
-...
+async def ap_get_object(self, requestor_actor_id: str, data, key=None) -> dict | 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.
 else:
 raise NotImplementedError(
 "was expecting a string or a dict, got {type(value)}: {value!r}}"
 )
-async def ap_get_local_object(
+async def ap_get_local_object(self, url: str) -> dict:
-self,
-url: str
-) -> dict:
 """Retrieve or generate local object
 for now, only handle XMPP items to convert to AP
 """
 url_type, url_args = self.parse_apurl(url)
 raise NotImplementedError(
 'only object from "item" URLs can be retrieved for now'
 )
 async def ap_get_list(
-self,
+self, requestor_actor_id: str, data: dict, key: str, only_ids: bool = False
-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 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 [
+return [{"id": v["id"]} if isinstance(v, dict) else {"id": v} for v in value]
-{"id": v["id"]} if isinstance(v, dict) else {"id": v}
-for v in value
-]
 else:
 return [await self.ap_get_object(requestor_actor_id, i) for i in value]
 async def ap_get_actors(
-self,
+self, requestor_actor_id: str, data: dict, key: str, as_account: bool = True
-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
 except (TypeError, KeyError):
 raise exceptions.DataError(
 f"invalid actors list to object {data.get('id')!r}: {value!r}"
 )
 if not value:
-raise exceptions.DataError(
+raise exceptions.DataError(f"list of actors is empty")
-f"list of actors is empty"
-)
 if as_account:
 return [
 await self.get_ap_account_from_id(requestor_actor_id, actor_id)
 for actor_id in value
 ]
 @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(requestor_actor_id, 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(requestor_actor_id, 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:
 raise exceptions.NotFound("list of actors is empty")
 def must_encode(self, text: str) -> bool:
 """Indicate if a text must be period encoded"""
 return (
-not RE_ALLOWED_UNQUOTED.match(text)
+not RE_ALLOWED_UNQUOTED.match(text) or text.startswith("___") or "---" in text
-or text.startswith("___")
-or "---" in text
 )
 def period_encode(self, text: str) -> str:
 """Period encode a text
 .replace("~", "%7e")
 .replace("%", ".")
 )
 async def get_ap_account_from_jid_and_node(
-self,
+self, jid_: jid.JID, node: Optional[str]
-jid_: jid.JID,
-node: Optional[str]
 ) -> str:
 """Construct AP account from JID and node
 The account construction will use escaping when necessary
 """
 raise MissingLocalPartError("there should be a local part")
 if node:
 account_elts.extend((node, "---"))
-account_elts.extend((
+account_elts.extend(
-user, "@", jid_.host if is_local else self.client.jid.userhost()
+(user, "@", jid_.host if is_local else self.client.jid.userhost())
-))
+)
 return "".join(account_elts)
 def is_local(self, jid_: jid.JID) -> bool:
 """Returns True if jid_ use a domain or subdomain of gateway's host"""
 local_host = self.client.host.split(".")
 assert local_host
-return jid_.host.split(".")[-len(local_host):] == local_host
+return jid_.host.split(".")[-len(local_host) :] == local_host
 async def is_pubsub(self, jid_: jid.JID) -> bool:
 """Indicate if a JID is a Pubsub service"""
 host_disco = await self.host.get_disco_infos(self.client, jid_)
-return (
+return ("pubsub", "service") in host_disco.identities and not (
-("pubsub", "service") in host_disco.identities
+"pubsub",
-and not ("pubsub", "pep") in host_disco.identities
+"pep",
-)
+) in host_disco.identities
-async def get_jid_and_node(self, ap_account: str) -> tuple[jid.JID, str|None]:
+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
 else:
 node = None
 if encoded:
 username = parse.unquote(
-RE_PERIOD_ENC.sub(r"%\g<hex>", username),
+RE_PERIOD_ENC.sub(r"%\g<hex>", username), errors="strict"
-errors="strict"
 )
 if node:
 node = parse.unquote(
-RE_PERIOD_ENC.sub(r"%\g<hex>", node),
+RE_PERIOD_ENC.sub(r"%\g<hex>", node), errors="strict"
-errors="strict"
 )
 if "@" in username:
 username, domain = username.rsplit("@", 1)
 """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(
+return jid.JID(None, (self._e.escape(account), self.client.jid.host, None))
-None,
-(
-self._e.escape(account),
-self.client.jid.host,
-None
-)
-)
 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
 @param url: URL to parse (schema is not mandatory)
 @return: endpoint type and extra arguments
 """
 path = parse.urlparse(url).path.lstrip("/")
-type_, *extra_args = path[len(self.ap_path):].lstrip("/").split("/")
+type_, *extra_args = path[len(self.ap_path) :].lstrip("/").split("/")
 return type_, [parse.unquote(a) for a in extra_args]
-def build_apurl(self, type_:str , *args: str) -> str:
+def build_apurl(self, type_: str, *args: str) -> str:
 """Build an AP endpoint URL
 @param type_: type of AP endpoing
 @param arg: endpoint dependant arguments
 """
 return parse.urljoin(
 self.base_ap_url,
-str(Path(type_).joinpath(*(parse.quote_plus(a, safe="@") for a in args)))
+str(Path(type_).joinpath(*(parse.quote_plus(a, safe="@") for a in args))),
 )
 def is_local_url(self, url: str) -> bool:
 """Tells if an URL link to this component
 """Retrieve actor data with LRU cache"""
 return await self.ap_get(actor_id, requestor_actor_id)
 @async_lru(maxsize=LRU_MAX_SIZE)
 async def get_actor_pub_key_data(
-self,
+self, requestor_actor_id: str, actor_id: str
-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)
 async def check_signature(
 self,
 requestor_actor_id: str,
 signature: str,
 key_id: str,
-headers: Dict[str, 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 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())
+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.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(
-requestor_actor_id,
+requestor_actor_id, 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(
 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
+hashes.SHA256(),  # type: ignore
 )
 except InvalidSignature:
 raise exceptions.EncryptionError(
 "Invalid signature (using PKC0S1 v1.5 and SHA-256)"
 )
 return actor_id
 def get_signature_data(
-self,
+self, key_id: str, headers: Dict[str, str]
-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)
 ``headers`` is an updated copy of ``headers`` arguments, with pseudo-headers
 removed, and ``Signature`` added.
 """
 # headers must be lower case
 l_headers: Dict[str, str] = {k.lower(): v for k, v in headers.items()}
-to_sign = "\n".join(f"{k}: {v}" for k,v in l_headers.items())
+to_sign = "\n".join(f"{k}: {v}" for k, v in l_headers.items())
-signature = base64.b64encode(self.private_key.sign(
+signature = base64.b64encode(
-to_sign.encode(),
+self.private_key.sign(
-# we have to use PKCS1v15 padding to be compatible with Mastodon
+to_sign.encode(),
-padding.PKCS1v15(),  # type: ignore
+# we have to use PKCS1v15 padding to be compatible with Mastodon
-hashes.SHA256()  # type: ignore
+padding.PKCS1v15(),  # type: ignore
-)).decode()
+hashes.SHA256(),  # type: ignore
+)
+).decode()
 sign_data = {
 "keyId": key_id,
 "Algorithm": "rsa-sha256",
 "headers": " ".join(l_headers.keys()),
-"signature": signature
+"signature": signature,
 }
-new_headers = {k: v for k,v in headers.items() if not k.startswith("(")}
+new_headers = {k: v for k, v in headers.items() if not k.startswith("(")}
 new_headers["Signature"] = self.build_signature_header(sign_data)
 return new_headers, sign_data
 async def convert_and_post_items(
 self,
 @param subscribe_extra_nodes: if True, extra data nodes will be automatically
 subscribed, that is comment nodes if present and attachments nodes.
 """
 actor_id = await self.get_ap_actor_id_from_account(ap_account)
 requestor_actor_id = self.build_apurl(
-TYPE_ACTOR,
+TYPE_ACTOR, await self.get_ap_account_from_jid_and_node(service, node)
-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({
+cached_item = await self.host.memory.storage.search_pubsub_items(
-"profiles": [self.client.profile],
+{
-"services": [service],
+"profiles": [self.client.profile],
-"nodes": [node],
+"services": [service],
-"names": [item["id"]]
+"nodes": [node],
-})
+"names": [item["id"]],
+}
+)
 is_new = not bool(cached_item)
 if node.startswith(self._events.namespace):
 # event item
 event_data = self._events.event_elt_2_event_data(item)
 try:
 client: SatXMPPEntity,
 ap_account: str,
 service: jid.JID,
 node: str,
 items: List[domish.Element],
-publisher: Optional[jid.JID] = None
+publisher: Optional[jid.JID] = None,
 ) -> None:
 """Convert XMPP item attachments to AP activities and post them to actor inbox
 @param ap_account: account of ActivityPub actor receiving the item
 @param service: JID of the (virtual) pubsub service where the item has been
 f"{len(items)})"
 )
 actor_id = await self.get_ap_actor_id_from_account(ap_account)
 requestor_actor_id = self.build_apurl(
-TYPE_ACTOR,
+TYPE_ACTOR, await self.get_ap_account_from_jid_and_node(service, node)
-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"]
 )
 return
 else:
 item_url = self.build_apurl(TYPE_ITEM, item_account, item_id)
-old_attachment_pubsub_items = await self.host.memory.storage.search_pubsub_items({
+old_attachment_pubsub_items = await self.host.memory.storage.search_pubsub_items(
-"profiles": [self.client.profile],
+{
-"services": [service],
+"profiles": [self.client.profile],
-"nodes": [node],
+"services": [service],
-"names": [item_elt["id"]]
+"nodes": [node],
-})
+"names": [item_elt["id"]],
+}
+)
 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(
 try:
 old_attachment = old_attachments[0]
 except IndexError:
 # no known element was present in attachments
 old_attachment = {}
-publisher_account = await self.get_ap_account_from_jid_and_node(
+publisher_account = await self.get_ap_account_from_jid_and_node(publisher, None)
-publisher,
-None
-)
 publisher_actor_id = self.build_apurl(TYPE_ACTOR, publisher_account)
 try:
 attachments = self._pa.items_2_attachment_data(client, [item_elt])[0]
 except IndexError:
 # no known element was present in attachments
 for reaction in reactions:
 activity_id = self.build_apurl(
 "reaction", item_account, item_id, reaction.encode().hex()
 )
 reaction_activity = self.create_activity(
-TYPE_REACTION, publisher_actor_id, item_url,
+TYPE_REACTION, publisher_actor_id, item_url, activity_id=activity_id
-activity_id=activity_id
 )
 reaction_activity["content"] = reaction
 reaction_activity["to"] = [ap_account]
 reaction_activity["cc"] = [NS_AP_PUBLIC]
 if undo:
 if "rsvp" in attachments:
 attending = attachments["rsvp"].get("attending", "no")
 old_attending = old_attachment.get("rsvp", {}).get("attending", "no")
 if attending != old_attending:
 activity_type = TYPE_JOIN if attending == "yes" else TYPE_LEAVE
-activity_id = self.build_apurl(activity_type.lower(), item_account, item_id)
+activity_id = self.build_apurl(
+activity_type.lower(), item_account, item_id
+)
 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.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_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]
 log.debug("storing attachments item in cache")
 cached_node = await self.host.memory.storage.get_pubsub_node(
 client, service, node, with_subscriptions=True, create=True
 )
 await self.host.memory.storage.cache_pubsub_items(
-self.client,
+self.client, cached_node, [item_elt], [attachments]
-cached_node,
-[item_elt],
-[attachments]
 )
 def _publish_message(self, mess_data_s: str, service_s: str, profile: str):
-mess_data: dict = data_format.deserialise(mess_data_s) # type: ignore
+mess_data: dict = data_format.deserialise(mess_data_s)  # type: ignore
 service = jid.JID(service_s)
 client = self.host.get_client(profile)
 return defer.ensureDeferred(self.publish_message(client, mess_data, service))
 @async_lru(maxsize=LRU_MAX_SIZE)
 """
 if account.count("@") != 1 or "/" in account:
 raise ValueError(f"Invalid account: {account!r}")
 host = account.split("@")[1]
 try:
-finger_data = await treq.json_content(await treq.get(
+finger_data = await treq.json_content(
-f"https://{host}/.well-known/webfinger?"
+await treq.get(
-f"resource=acct:{parse.quote_plus(account)}",
+f"https://{host}/.well-known/webfinger?"
-))
+f"resource=acct:{parse.quote_plus(account)}",
+)
+)
 except Exception as e:
 raise exceptions.DataError(f"Can't get webfinger data for {account!r}: {e}")
 for link in finger_data.get("links", []):
 if (
 link.get("type") == "application/activity+json"
 raise ValueError(
 f"Invalid webfinger data for {account:r}: missing href"
 )
 break
 else:
-raise ValueError(
+raise ValueError(f"No ActivityPub link found for {account!r}")
-f"No ActivityPub link found for {account!r}"
-)
 return href
 async def get_ap_actor_data_from_account(
-self,
+self, requestor_actor_id: str, account: str
-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, requestor_actor_id)
 async def get_ap_inbox_from_id(
-self,
+self, requestor_actor_id: str, actor_id: str, use_shared: bool = True
-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.
 @return: AP handle
 """
 if self.is_local_url(actor_id):
 url_type, url_args = self.parse_apurl(actor_id)
 if url_type != "actor" or not url_args:
-raise exceptions.DataError(
+raise exceptions.DataError(f"invalid local actor ID: {actor_id}")
-f"invalid local actor ID: {actor_id}"
-)
 account = url_args[0]
 try:
-account_user, account_host = account.split('@')
+account_user, account_host = account.split("@")
 except ValueError:
-raise exceptions.DataError(
+raise exceptions.DataError(f"invalid account from url: {actor_id}")
-f"invalid account from url: {actor_id}"
-)
 if not account_user or account_host != self.public_url:
 raise exceptions.DataError(
 f"{account!r} is not a valid local account (from {actor_id})"
 )
 return account
 pass
 elif start_index > 5000:
 raise error.StanzaError(
 "feature-not-implemented",
 text="Maximum limit for previous_index has been reached, this limit"
-"is set to avoid DoS"
+"is set to avoid DoS",
 )
 else:
 # we'll convert "start_index" to "after_id", thus we need the item just
 # before "start_index"
 previous_index = start_index - 1
 f"missing previous page link at {current_page}: {page_data!r}"
 )
 raise error.StanzaError(
 "service-unavailable",
 "Error while retrieving previous page from AP service at "
-f"{current_page}"
+f"{current_page}",
 )
 init_page = "last" if chronological_pagination else "first"
 page = collection.get(init_page)
 if not page:
 log.debug(f"{after_id!r} not found at {page_id}, skipping")
 else:
 found_after_id = True
 if chronological_pagination:
 start_index = retrieved_items - len(page_items) + limit_idx + 1
-page_items = page_items[limit_idx+1:]
+page_items = page_items[limit_idx + 1 :]
 else:
-start_index = count - (retrieved_items - len(page_items) +
+start_index = count - (
-limit_idx + 1)
+retrieved_items - len(page_items) + limit_idx + 1
+)
 page_items = page_items[:limit_idx]
 items.extend(page_items)
 else:
 items.extend(page_items)
 if max_items is not None and len(items) >= max_items:
 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({
+rsm_resp.update({"first": items[0]["id"], "last": items[-1]["id"]})
-"first": items[0]["id"],
-"last": items[-1]["id"]
-})
 return items, rsm.RSMResponse(**rsm_resp)
-async def ap_item_2_mb_data_and_elt(self, requestor_actor_id: str, 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
 @param requestor_actor_id: ID of the actor requesting the conversion.
 @param ap_item: AP item to convert.
 @return: microblog and correspondign <item> element.
 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, requestor_actor_id: str, 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
 @param requestor_actor_id: ID of the actor requesting the conversion.
 @param ap_item: AP item to convert.
 @return: <item> element
 async def parse_ap_page(
 self,
 requestor_actor_id: str,
 page: Union[str, dict],
 parser: Callable[[str, dict], Awaitable[domish.Element]],
-only_ids: bool = False
+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 only_ids: if True, only retrieve items IDs
 @return: page data, pubsub items
 """
 page_data = await self.ap_get_object(requestor_actor_id, page)
 if page_data is None:
-log.warning('No data found in collection')
+log.warning("No data found in collection")
 return {}, []
-ap_items = await self.ap_get_list(requestor_actor_id, 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(requestor_actor_id, 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}')
+log.warning(f"No item field found in collection: {page_data!r}")
 return page_data, []
 else:
-log.warning(
+log.warning("Items are not ordered, this is not spec compliant")
-"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:
 continue
 return page_data, items
 async def get_comments_nodes(
-self,
+self, requestor_actor_id: str, item_id: str, parent_id: Optional[str]
-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
 "comments_max_depth")
 """
 if parent_id is None or not self.comments_max_depth:
 return (
 self._m.get_comments_node(parent_id) if parent_id is not None else None,
-self._m.get_comments_node(item_id)
+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, 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
+parent_limit = self.comments_max_depth - 1
 if len(parents) <= parent_limit:
 return (
 self._m.get_comments_node(parents[-1]["id"]),
-self._m.get_comments_node(item_id)
+self._m.get_comments_node(item_id),
 )
 else:
 last_level_item = parents[parent_limit]
-return (
+return (self._m.get_comments_node(last_level_item["id"]), None)
-self._m.get_comments_node(last_level_item["id"]),
-None
-)
 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.
 # author
 if is_activity:
 authors = await self.ap_get_actors(requestor_actor_id, ap_item, "actor")
 else:
-authors = await self.ap_get_actors(requestor_actor_id, 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 comments_node is not None:
 comments_data = {
 "service": author_jid,
 "node": comments_node,
-"uri": uri.build_xmpp_uri(
+"uri": uri.build_xmpp_uri("pubsub", path=author_jid, node=comments_node),
-"pubsub",
-path=author_jid,
-node=comments_node
-)
 }
 mb_data["comments"] = [comments_data]
 return mb_data
 async def get_reply_to_id_from_xmpp_node(
-self,
+self, client: SatXMPPEntity, ap_account: str, parent_item: str, mb_data: dict
-client: SatXMPPEntity,
-ap_account: str,
-parent_item: str,
-mb_data: dict
 ) -> str:
 """Get URL to use for ``inReplyTo`` field in AP item.
 There is currently no way to know the parent service of a comment with XEP-0277.
 To work around that, we try to check if we have this item in the cache (we
 @param mb_data: microblog data of the publication
 @return: URL to use in ``inReplyTo`` field
 """
 # FIXME: propose a protoXEP to properly get parent item, node and service
-found_items = await self.host.memory.storage.search_pubsub_items({
+found_items = await self.host.memory.storage.search_pubsub_items(
-"profiles": [client.profile],
+{"profiles": [client.profile], "names": [parent_item]}
-"names": [parent_item]
+)
-})
 if not found_items:
 log.warning(f"parent item {parent_item!r} not found in cache")
 parent_ap_account = ap_account
 elif len(found_items) == 1:
 cached_node = found_items[0].node
 parent_ap_account = await self.get_ap_account_from_jid_and_node(
-cached_node.service,
+cached_node.service, cached_node.name
-cached_node.name
 )
 else:
 # we found several cached item with given ID, we check if there is one
 # corresponding to this author
 try:
 author = jid.JID(mb_data["author_jid"]).userhostJID()
 cached_item = next(
-i for i in found_items
+i
-if jid.JID(i.data["publisher"]).userhostJID()
+for i in found_items
-== author
+if jid.JID(i.data["publisher"]).userhostJID() == author
 )
 except StopIteration:
 # no item corresponding to this author, we use ap_account
 log.warning(
-"Can't find a single cached item for parent item "
+"Can't find a single cached item for parent item " f"{parent_item!r}"
-f"{parent_item!r}"
 )
 parent_ap_account = ap_account
 else:
 cached_node = cached_item.node
 parent_ap_account = await self.get_ap_account_from_jid_and_node(
-cached_node.service,
+cached_node.service, cached_node.name
-cached_node.name
+)
-)
+return self.build_apurl(TYPE_ITEM, parent_ap_account, parent_item)
-return self.build_apurl(
-TYPE_ITEM, parent_ap_account, parent_item
+async def repeated_mb_2_ap_item(self, mb_data: dict) -> dict:
-)
-async def repeated_mb_2_ap_item(
-self,
-mb_data: dict
-) -> dict:
 """Convert repeated blog item to suitable AP Announce activity
 @param mb_data: microblog metadata of an item repeating an other blog post
 @return: Announce activity linking to the repeated item
 """
 repeated = mb_data["extra"]["repeated"]
 repeater = jid.JID(repeated["by"])
-repeater_account = await self.get_ap_account_from_jid_and_node(
+repeater_account = await self.get_ap_account_from_jid_and_node(repeater, None)
-repeater,
-None
-)
 repeater_id = self.build_apurl(TYPE_ACTOR, repeater_account)
 repeated_uri = repeated["uri"]
 if not repeated_uri.startswith("xmpp:"):
 log.warning(
 "Announce", repeater_id, announced_uri, activity_id=activity_id
 )
 announce["to"] = [NS_AP_PUBLIC]
 announce["cc"] = [
 self.build_apurl(TYPE_FOLLOWERS, repeater_account),
-await self.get_ap_actor_id_from_account(repeated_account)
+await self.get_ap_actor_id_from_account(repeated_account),
 ]
 return announce
 async def mb_data_2_ap_item(
 self,
 client: SatXMPPEntity,
 mb_data: dict,
-public: bool =True,
+public: bool = True,
 is_new: bool = True,
 ) -> dict:
 """Convert Libervia Microblog Data to ActivityPub item
 @param mb_data: microblog data (as used in plugin XEP-0277) to convert
 if not mb_data.get("id"):
 mb_data["id"] = shortuuid.uuid()
 if not mb_data.get("author_jid"):
 mb_data["author_jid"] = client.jid.userhost()
 ap_account = await self.get_ap_account_from_jid_and_node(
-jid.JID(mb_data["author_jid"]),
+jid.JID(mb_data["author_jid"]), None
-None
 )
 url_actor = self.build_apurl(TYPE_ACTOR, ap_account)
 url_item = self.build_apurl(TYPE_ITEM, ap_account, mb_data["id"])
 ap_object = {
 "id": url_item,
 attachments = extra.get(C.KEY_ATTACHMENTS)
 if attachments:
 ap_attachments = ap_object["attachment"] = []
 for attachment in attachments:
 try:
-url = next(
+url = next(s["url"] for s in attachment["sources"] if "url" in s)
-s['url'] for s in attachment["sources"] if 'url' in s
-)
 except (StopIteration, KeyError):
-log.warning(
+log.warning(f"Ignoring attachment without URL: {attachment}")
-f"Ignoring attachment without URL: {attachment}"
-)
 continue
-ap_attachment = {
+ap_attachment = {"url": url}
-"url": url
-}
 for key, ap_key in (
 ("media_type", "mediaType"),
 # XXX: yes "name", cf. [ap_item_2_mb_data]
 ("desc", "name"),
 ):
 mentioned_id = await self.get_ap_actor_id_from_account(mentioned)
 except Exception as e:
 log.warning(f"Can't add mention to {mentioned!r}: {e}")
 else:
 ap_object["to"].append(mentioned_id)
-ap_object.setdefault("tag", []).append({
+ap_object.setdefault("tag", []).append(
-"type": TYPE_MENTION,
+{
-"href": mentioned_id,
+"type": TYPE_MENTION,
-"name": mention,
+"href": mentioned_id,
-})
+"name": mention,
+}
+)
 try:
 node = mb_data["node"]
 service = jid.JID(mb_data["service"])
 except KeyError:
 # node and service must always be specified when this method is used
-raise exceptions.InternalError(
+raise exceptions.InternalError("node or service is missing in mb_data")
-"node or service is missing in mb_data"
-)
 try:
 target_ap_account = await self.get_ap_account_from_jid_and_node(
 service, node
 )
 except MissingLocalPartError:
 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(
-url_actor,
+url_actor, target_ap_account
-target_ap_account
 )
 followers = actor_data.get("followers")
 else:
 # service is a real XMPP entity
 followers = self.build_apurl(TYPE_FOLLOWERS, target_ap_account)
 # this gateway and linking to an ActivityPub actor)
 ap_object["inReplyTo"] = parent_item
 else:
 # the publication is from a followed real XMPP node
 ap_object["inReplyTo"] = await self.get_reply_to_id_from_xmpp_node(
-client,
+client, ap_account, parent_item, mb_data
-ap_account,
-parent_item,
-mb_data
 )
 return self.create_activity(
 "Create" if is_new else "Update", url_actor, ap_object, activity_id=url_item
 )
 async def publish_message(
-self,
+self, client: SatXMPPEntity, mess_data: dict, service: jid.JID
-client: SatXMPPEntity,
-mess_data: dict,
-service: jid.JID
 ) -> None:
 """Send an AP message
 .. note::
 item_data = await self.mb_data_2_ap_item(client, mess_data)
 url_actor = item_data["actor"]
 await self.ap_post(inbox_url, url_actor, item_data)
 async def ap_delete_item(
-self,
+self, jid_: jid.JID, node: Optional[str], item_id: str, public: bool = True
-jid_: jid.JID,
-node: Optional[str],
-item_id: str,
-public: bool = True
 ) -> Tuple[str, Dict[str, Any]]:
 """Build activity to delete an AP item
 @param jid_: JID of the entity deleting an item
 @param node: node where the item is deleted
 node = self._m.namespace
 author_account = await self.get_ap_account_from_jid_and_node(jid_, node)
 author_actor_id = self.build_apurl(TYPE_ACTOR, author_account)
-items = await self.host.memory.storage.search_pubsub_items({
+items = await self.host.memory.storage.search_pubsub_items(
-"profiles": [self.client.profile],
+{"profiles": [self.client.profile], "services": [jid_], "names": [item_id]}
-"services": [jid_],
+)
-"names": [item_id]
-})
 if not items:
 log.warning(
 f"Deleting an unknown item at service {jid_}, node {node} and id "
 f"{item_id}"
 )
 else:
 try:
-mb_data = await self._m.item_2_mb_data(self.client, items[0].data, jid_, node)
+mb_data = await self._m.item_2_mb_data(
+self.client, items[0].data, jid_, node
+)
 if "repeated" in mb_data["extra"]:
 # we are deleting a repeated item, we must translate this to an
 # "Undo" of the "Announce" activity instead of a "Delete" one
 announce = await self.repeated_mb_2_ap_item(mb_data)
 undo = self.create_activity("Undo", author_actor_id, announce)
 f"{items[0].toXml()}"
 )
 url_item = self.build_apurl(TYPE_ITEM, author_account, item_id)
 ap_item = self.create_activity(
-"Delete",
+"Delete", author_actor_id, {"id": url_item, "type": TYPE_TOMBSTONE}
-author_actor_id,
-{
-"id": url_item,
-"type": TYPE_TOMBSTONE
-}
 )
 if public:
 ap_item["to"] = [NS_AP_PUBLIC]
 return author_actor_id, ap_item
 def _message_received_trigger(
 self,
 client: SatXMPPEntity,
 message_elt: domish.Element,
-post_treat: defer.Deferred
+post_treat: defer.Deferred,
 ) -> bool:
 """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
 return mess_data
 if not self.is_local(mess_data["from"]):
 log.warning(f"ignoring non local message: {mess_data}")
 return mess_data
 if not mess_data["to"].user:
-log.warning(
+log.warning(f"ignoring message addressed to gateway itself: {mess_data}")
-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(
+log.warning(f"Can't retrieve data on actor {actor_account}: {e}")
-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(
 requestor_actor_id, actor_id, use_shared=False
 )
 if origin_id:
 # we need to use origin ID when present to be able to retract the message
 mb_data["id"] = origin_id
 attachments = mess_data["extra"].get(C.KEY_ATTACHMENTS)
 if attachments:
-mb_data["extra"] = {
+mb_data["extra"] = {C.KEY_ATTACHMENTS: attachments}
-C.KEY_ATTACHMENTS: attachments
-}
 client = self.client.get_virtual_client(mess_data["from"])
 ap_item = await self.mb_data_2_ap_item(client, mb_data, public=False)
 ap_object = ap_item["object"]
 ap_object["to"] = ap_item["to"] = [actor_id]
 # we add a mention to direct message, otherwise peer is not notified in some AP
 # implementations (notably Mastodon), and the message may be missed easily.
-ap_object.setdefault("tag", []).append({
+ap_object.setdefault("tag", []).append(
-"type": TYPE_MENTION,
+{
-"href": actor_id,
+"type": TYPE_MENTION,
-"name": f"@{actor_account}",
+"href": actor_id,
-})
+"name": f"@{actor_account}",
+}
+)
 try:
 await self.ap_post(inbox, ap_item["actor"], ap_item)
 except Exception as e:
 # TODO: send an error <message>
-log.warning(
+log.warning(f"Can't send message to {inbox}: {e}")
-f"Can't send message to {inbox}: {e}"
-)
 return mess_data
 async def _on_message_retract(
 self,
 client: SatXMPPEntity,
 message_elt: domish.Element,
 retract_elt: domish.Element,
-history: History
+history: History,
 ) -> bool:
 if client != self.client:
 return True
 from_jid = jid.JID(message_elt["from"])
 if not self.is_local(from_jid):
-log.debug(
+log.debug(f"ignoring retract request from non local jid {from_jid}")
-f"ignoring retract request from non local jid {from_jid}"
-)
 return False
-requestor_actor_id = self.build_apurl(
+requestor_actor_id = self.build_apurl(TYPE_ACTOR, from_jid.userhost())
-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):
+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(
+raise exceptions.InternalError(f"Invalid destinee's JID: {to_jid.full()}")
-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(requestor_actor_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.ap_post(inbox, url_actor, ap_item)
 return False
 async def _on_reference_received(
 self,
 client: SatXMPPEntity,
 message_elt: domish.Element,
-reference_data: Dict[str, Union[str, int]]
+reference_data: Dict[str, Union[str, int]],
 ) -> bool:
 parsed_uri: dict = reference_data.get("parsed_uri")
 if not parsed_uri:
 log.warning(f"no parsed URI available in reference {reference_data}")
 return False
 client, cached_item.data, pubsub_service, pubsub_node
 )
 ap_item = await self.mb_data_2_ap_item(client, mb_data)
 ap_object = ap_item["object"]
 ap_object["to"] = [actor_id]
-ap_object.setdefault("tag", []).append({
+ap_object.setdefault("tag", []).append(
-"type": TYPE_MENTION,
+{
-"href": actor_id,
+"type": TYPE_MENTION,
-"name": ap_account,
+"href": actor_id,
-})
+"name": ap_account,
+}
+)
 requestor_actor_id = ap_item["actor"]
-inbox = await self.get_ap_inbox_from_id(requestor_actor_id, actor_id, use_shared=False)
+inbox = await self.get_ap_inbox_from_id(
+requestor_actor_id, actor_id, use_shared=False
+)
 await self.ap_post(inbox, requestor_actor_id, ap_item)
 return False
 "Ignoring AP item replying to an XMPP item with an unexpected URL "
 f"type({url_type!r}):\n{pformat(ap_item)}"
 )
 return
 try:
-parent_item_account, parent_item_id = url_args[0], '/'.join(url_args[1:])
+parent_item_account, parent_item_id = url_args[0], "/".join(url_args[1:])
 except (IndexError, ValueError):
 log.warning(
 "Ignoring AP item replying to an XMPP item with invalid inReplyTo URL "
 f"({in_reply_to!r}):\n{pformat(ap_item)}"
 )
 try:
 parent_item_elt = items[0]
 except IndexError:
 log.warning(
 f"Can't find parent item at {parent_item_service} (node "
-f"{parent_item_node!r})\n{pformat(ap_item)}")
+f"{parent_item_node!r})\n{pformat(ap_item)}"
+)
 return
 parent_item_parsed = await self._m.item_2_mb_data(
 client, parent_item_elt, parent_item_service, parent_item_node
 )
 try:
 comment_service = jid.JID(parent_item_parsed["comments"][0]["service"])
 comment_node = parent_item_parsed["comments"][0]["node"]
 except (KeyError, IndexError):
 # 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:
 requestor_actor_id = self.build_apurl(
 TYPE_ACTOR,
-await self.get_ap_account_from_jid_and_node(comment_service, comment_node)
+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,
+requestor_actor_id, ap_item
-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"]
 )
 def get_ap_item_targets(
-self,
+self, item: Dict[str, Any]
-item: Dict[str, Any]
 ) -> Tuple[bool, Dict[str, Set[str]], List[Dict[str, str]]]:
 """Retrieve targets of an AP item, and indicate if it's a public one
 @param item: AP object payload
 @return: Are returned:
 @param item: AP object payload
 """
 is_public, targets, mentions = self.get_ap_item_targets(item)
 if not is_public and targets.keys() == {TYPE_ACTOR}:
 # this is a direct message
-await self.handle_message_ap_item(
+await self.handle_message_ap_item(client, targets, mentions, destinee, item)
-client, targets, mentions, destinee, item
-)
 else:
 await self.handle_pubsub_ap_item(
 client, targets, mentions, destinee, node, item, is_public
 )
-def get_requestor_actor_id_from_targets(
+def get_requestor_actor_id_from_targets(self, targets: set[str]) -> str:
-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".
 @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(
+log.warning(f"Can't find local target to use as requestor ID: {targets!r}")
-f"Can't find local target to use as requestor ID: {targets!r}"
+return self.build_apurl(TYPE_ACTOR, f"libervia@{self.public_url}")
-)
-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]],
 mentions: list[Dict[str, str]],
-destinee: jid.JID|None,
+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 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(requestor_actor_id, url)
+await self.get_jid_from_id(requestor_actor_id, url) for url in targets_urls
-for url in targets_urls
 }
 if destinee is not None:
 targets_jids.add(destinee)
 mb_data = await self.ap_item_2_mb_data(requestor_actor_id, item)
-extra = {
+extra = {"origin_id": mb_data["id"]}
-"origin_id": mb_data["id"]
-}
 attachments = mb_data["extra"].get(C.KEY_ATTACHMENTS)
 if attachments:
 extra[C.KEY_ATTACHMENTS] = attachments
 defer_l = []
 for target_jid in targets_jids:
 defer_l.append(
 client.sendMessage(
 target_jid,
-{'': mb_data.get("content", "")},
+{"": mb_data.get("content", "")},
 mb_data.get("title"),
-extra=extra
+extra=extra,
 )
 )
 await defer.DeferredList(defer_l)
 async def notify_mentions(
 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)
+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(requestor_actor_id, mention["uri"])
-self._refs.send_reference(
+self._refs.send_reference(self.client, to_jid=mentioned_jid, anchor=anchor)
-self.client,
-to_jid=mentioned_jid,
-anchor=anchor
-)
 seen.add(mentioned_jid)
 remaining = {
 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(
+self._refs.send_reference(self.client, to_jid=target, anchor=anchor)
-self.client,
-to_jid=target,
-anchor=anchor
-)
 async def handle_pubsub_ap_item(
 self,
 client: SatXMPPEntity,
 targets: dict[str, set[str]],
 mentions: list[dict[str, str]],
-destinee: jid.JID|None,
+destinee: jid.JID | None,
 node: str,
 item: dict,
-public: bool
+public: bool,
 ) -> None:
 """Analyse, cache and deliver AP items translating to Pubsub
 @param targets: actors/collections where the item must be delivered
 @param destinee: jid of the destinee,
 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(
-requestor_actor_id,
+requestor_actor_id, item["id"], in_reply_to
-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,
+client,
-create_kwargs={"subscribed": True}
+service,
+node,
+with_subscriptions=True,
+create=True,
+create_kwargs={"subscribed": True},
 )
 else:
 # it is a root item (i.e. not a reply to an other item)
 create = node == self._events.namespace
 cached_node = await self.host.memory.storage.get_pubsub_node(
 )
 if cached_node is None:
 log.warning(
 f"Received item in unknown node {node!r} at {service}. This may be "
 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(
-requestor_actor_id,
+requestor_actor_id, item
-item
 )
 else:
 data, item_elt = await self.ap_item_2_mb_data_and_elt(
-requestor_actor_id,
+requestor_actor_id, item
-item
 )
 await self.host.memory.storage.cache_pubsub_items(
-client,
+client, cached_node, [item_elt], [data]
-cached_node,
-[item_elt],
-[data]
 )
 for subscription in cached_node.subscriptions:
 if subscription.state != SubscriptionState.SUBSCRIBED:
 continue
 self.pubsub_service.notifyPublish(
-service,
+service, node, [(subscription.subscriber, None, [item_elt])]
-node,
-[(subscription.subscriber, None, [item_elt])]
 )
 await self.notify_mentions(targets, mentions, service, node, item_elt["id"])
 async def new_ap_delete_item(
 history = await self.host.memory.storage.get(
 client,
 History,
 History.origin_id,
 item_id,
-(History.messages, History.subjects)
+(History.messages, History.subjects),
 )
 if history is not None:
 # it's a direct message
 if history.source_jid != client.jid:
 item_elt["id"] = item_id
 for subscription in cached_node.subscriptions:
 if subscription.state != SubscriptionState.SUBSCRIBED:
 continue
 self.pubsub_service.notifyRetract(
-client.jid,
+client.jid, node, [(subscription.subscriber, None, [item_elt])]
-node,
+)
-[(subscription.subscriber, None, [item_elt])]
-)

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py @ 4270:0d7bb4df2343

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4270:0d7bb4df2343