erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

component AP Gateway: implement HTTP GET signature.

Browse source
This commit is contained in:
Goffi 2024-06-05 22:34:09 +02:00
commit b402be2a7a
5 changed files with 615 additions and 263 deletions
Download patch file Download diff file Expand all files Collapse all files

443
libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py
View file

@ -43,9 +43,7 @@ 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
@ -141,6 +139,7 @@ class APGateway:
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"]
@ -261,6 +260,9 @@ class APGateway:
'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
@ -362,15 +364,20 @@ class APGateway:
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:
@ -380,18 +387,21 @@ class APGateway:
except (KeyError, TypeError):
return False
async def ap_get(self, url: str) -> dict:
"""Retrieve AP JSON from given URL
async def ap_get(self, url: str, requestor_actor_id: str) -> dict:
"""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(
url,
headers = {
"Accept": [MEDIA_TYPE_AP],
}
)
if self.http_sign_get:
headers = self._generate_signed_headers(url, requestor_actor_id, method="get")
else:
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:
@ -407,21 +417,91 @@ class APGateway:
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
@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: dict, key: str) -> Optional[dict]:
async def ap_get_object(
self, requestor_actor_id: str, data: dict, key: str
) -> dict|None:
...
@overload
async def ap_get_object(
self, data: Union[str, dict], key: None = None
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
@ -440,7 +520,7 @@ class APGateway:
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}}"
@ -517,6 +597,7 @@ class APGateway:
async def ap_get_list(
self,
requestor_actor_id: str,
data: dict,
key: str,
only_ids: bool = False
@ -526,6 +607,7 @@ class APGateway:
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
@ -538,7 +620,7 @@ class APGateway:
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):
@ -549,16 +631,18 @@ class APGateway:
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
@ -591,29 +675,34 @@ class APGateway:
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"'
@ -708,7 +797,7 @@ class APGateway:
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.
@ -819,13 +908,17 @@ class APGateway:
)
)
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)
@ -834,10 +927,10 @@ class APGateway:
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)
@ -895,22 +988,24 @@ class APGateway:
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"]
@ -969,6 +1064,7 @@ class APGateway:
async def check_signature(
self,
requestor_actor_id: str,
signature: str,
key_id: str,
headers: Dict[str, str]
@ -977,6 +1073,7 @@ class APGateway:
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
@ -991,7 +1088,10 @@ class APGateway:
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")
@ -1066,7 +1166,11 @@ class APGateway:
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({
@ -1146,7 +1250,7 @@ class APGateway:
)
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,
@ -1179,7 +1283,11 @@ class APGateway:
)
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"]
@ -1203,7 +1311,9 @@ class APGateway:
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
@ -1227,7 +1337,9 @@ class APGateway:
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:
@ -1254,7 +1366,7 @@ class APGateway:
)
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
@ -1265,7 +1377,7 @@ class APGateway:
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", []))
@ -1290,7 +1402,7 @@ class APGateway:
)
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:
@ -1304,7 +1416,7 @@ class APGateway:
)
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")
@ -1315,7 +1427,7 @@ class APGateway:
)
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
@ -1330,56 +1442,6 @@ class APGateway:
[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)
@ -1424,21 +1486,32 @@ class APGateway:
)
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"]
@ -1447,10 +1520,11 @@ class APGateway:
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
"""
@ -1474,7 +1548,7 @@ class APGateway:
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(
@ -1496,17 +1570,20 @@ class APGateway:
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
@ -1569,7 +1646,7 @@ class APGateway:
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}")
@ -1604,7 +1681,7 @@ class APGateway:
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)
@ -1660,9 +1737,14 @@ class APGateway:
return items, rsm.RSMResponse(**rsm_resp)
async def ap_item_2_mb_data_and_elt(self, ap_item: dict) -> Tuple[dict, domish.Element]:
"""Convert AP item to parsed microblog data and corresponding item element"""
mb_data = await self.ap_item_2_mb_data(ap_item)
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.
"""
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
)
@ -1672,31 +1754,38 @@ class APGateway:
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:
"""Convert AP item to XMPP item element"""
__, item_elt = await self.ap_item_2_mb_data_and_elt(ap_item)
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
"""
__, 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, []
@ -1709,7 +1798,7 @@ class APGateway:
# 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
@ -1717,6 +1806,7 @@ class APGateway:
async def get_comments_nodes(
self,
requestor_actor_id: str,
item_id: str,
parent_id: Optional[str]
) -> Tuple[Optional[str], Optional[str]]:
@ -1724,6 +1814,7 @@ class APGateway:
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:
@ -1741,7 +1832,7 @@ class APGateway:
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:
@ -1759,9 +1850,10 @@ class APGateway:
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
@ -1771,7 +1863,7 @@ class APGateway:
"""
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
@ -1831,9 +1923,9 @@ class APGateway:
# 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
@ -1864,7 +1956,9 @@ class APGateway:
# 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,
@ -2131,7 +2225,10 @@ class APGateway:
)
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
@ -2192,7 +2289,7 @@ class APGateway:
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,
@ -2266,11 +2363,11 @@ class APGateway:
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
@ -2288,6 +2385,7 @@ class APGateway:
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:
@ -2298,7 +2396,9 @@ class APGateway:
)
# 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()))
@ -2334,7 +2434,7 @@ class APGateway:
})
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(
@ -2357,6 +2457,10 @@ class APGateway:
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
@ -2365,11 +2469,11 @@ class APGateway:
)
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(
@ -2455,9 +2559,10 @@ class APGateway:
"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
@ -2512,7 +2617,14 @@ class APGateway:
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"]
@ -2548,7 +2660,7 @@ class APGateway:
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)
@ -2600,12 +2712,38 @@ class APGateway:
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]],
mentions: List[Dict[str, str]],
destinee: Optional[jid.JID],
targets: dict[str, Set[str]],
mentions: list[Dict[str, str]],
destinee: jid.JID|None,
item: dict,
) -> None:
"""Parse and deliver direct AP items translating to XMPP messages
@ -2614,14 +2752,15 @@ class APGateway:
@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)
for t_set in targets.values()
for t in t_set
await self.get_jid_from_id(requestor_actor_id, url)
for url in targets_urls
}
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"]
}
@ -2643,8 +2782,8 @@ class APGateway:
async def notify_mentions(
self,
targets: Dict[str, Set[str]],
mentions: List[Dict[str, str]],
targets: dict[str, set[str]],
mentions: list[dict[str, str]],
service: jid.JID,
node: str,
item_id: str,
@ -2657,13 +2796,15 @@ class APGateway:
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,
@ -2672,7 +2813,7 @@ class APGateway:
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
@ -2686,9 +2827,9 @@ class APGateway:
async def handle_pubsub_ap_item(
self,
client: SatXMPPEntity,
targets: Dict[str, Set[str]],
mentions: List[Dict[str, str]],
destinee: Optional[jid.JID],
targets: dict[str, set[str]],
mentions: list[dict[str, str]],
destinee: jid.JID|None,
node: str,
item: dict,
public: bool
@ -2702,6 +2843,8 @@ class APGateway:
@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")
@ -2715,7 +2858,11 @@ class APGateway:
# 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,
@ -2735,9 +2882,15 @@ class APGateway:
)
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(
requestor_actor_id,
item
)
else:
data, item_elt = await self.ap_item_2_mb_data_and_elt(item)
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,

40
libervia/backend/plugins/plugin_comp_ap_gateway/events.py
View file

@ -104,7 +104,7 @@ class APEvents:
"""Convert event data to AP activity
@param event_data: event data as used in [plugin_exp_events]
@param author_jid: jid of the published of the event
@param author_jid: jid of the publisher of the event
@param is_new: if True, the item is a new one (no instance has been found in
cache).
If True, a "Create" activity will be generated, otherwise an "Update" one will
@ -236,9 +236,14 @@ class APEvents:
activity["@context"].append(AP_EVENTS_CONTEXT)
return activity
async def ap_item_2_event_data(self, ap_item: dict) -> dict:
async def ap_item_2_event_data(
self,
requestor_actor_id: str,
ap_item: dict
) -> dict:
"""Convert AP activity or object to event data
@param requestor_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 event data
@ -246,7 +251,7 @@ class APEvents:
"""
is_activity = self.apg.is_activity(ap_item)
if is_activity:
ap_object = await self.apg.ap_get_object(ap_item, "object")
ap_object = await self.apg.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
@ -268,9 +273,9 @@ class APEvents:
raise exceptions.DataError("AP Object is not an event")
# author
actor = await self.apg.ap_get_sender_actor(ap_object)
actor = await self.apg.ap_get_sender_actor(requestor_actor_id, ap_object)
account = await self.apg.get_ap_account_from_id(actor)
account = await self.apg.get_ap_account_from_id(requestor_actor_id, actor)
author_jid = self.apg.get_local_jid_from_account(account).full()
# name, start, end
@ -370,7 +375,11 @@ class APEvents:
# comments
if ap_object.get("commentsEnabled"):
__, comments_node = await self.apg.get_comments_nodes(object_id, None)
__, comments_node = await self.apg.get_comments_nodes(
requestor_actor_id,
object_id,
None
)
event_data["comments"] = {
"service": author_jid,
"node": comments_node,
@ -391,17 +400,28 @@ class APEvents:
async def ap_item_2_event_data_and_elt(
self,
requestor_actor_id: str,
ap_item: dict
) -> Tuple[dict, domish.Element]:
) -> tuple[dict, domish.Element]:
"""Convert AP item to parsed event data and corresponding item element"""
event_data = await self.ap_item_2_event_data(ap_item)
event_data = await self.ap_item_2_event_data(
requestor_actor_id,
ap_item
)
event_elt = self._events.event_data_2_event_elt(event_data)
item_elt = domish.Element((None, "item"))
item_elt["id"] = event_data["id"]
item_elt.addChild(event_elt)
return event_data, item_elt
async def ap_item_2_event_elt(self, ap_item: dict) -> domish.Element:
async def ap_item_2_event_elt(
self,
requestor_actor_id: str,
ap_item: dict
) -> domish.Element:
"""Convert AP item to XMPP item element"""
__, item_elt = await self.ap_item_2_event_data_and_elt(ap_item)
__, item_elt = await self.ap_item_2_event_data_and_elt(
requestor_actor_id,
ap_item
)
return item_elt

169
libervia/backend/plugins/plugin_comp_ap_gateway/http_server.py
View file

@ -127,6 +127,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_undo_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: jid.JID,
@ -137,7 +138,7 @@ class HTTPAPGServer(web_resource.Resource):
) -> None:
if node is None:
node = self.apg._m.namespace
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
object_ = data.get("object")
if isinstance(object_, str):
# we check first if it's not a cached object
@ -150,10 +151,10 @@ class HTTPAPGServer(web_resource.Resource):
# because we'll undo the activity, we can remove it from cache
await self.apg.client._ap_storage.remove(ap_cache_key)
else:
objects = await self.apg.ap_get_list(data, "object")
objects = await self.apg.ap_get_list(requestor_actor_id, data, "object")
for obj in objects:
type_ = obj.get("type")
actor = await self.apg.ap_get_sender_actor(obj)
actor = await self.apg.ap_get_sender_actor(requestor_actor_id, obj)
if actor != signing_actor:
log.warning(f"ignoring object not attributed to signing actor: {data}")
continue
@ -188,6 +189,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_follow_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: jid.JID,
@ -198,7 +200,7 @@ class HTTPAPGServer(web_resource.Resource):
) -> None:
if node is None:
node = self.apg._m.namespace
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
try:
subscription = await self.apg._p.subscribe(
client,
@ -224,6 +226,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_accept_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: jid.JID,
@ -234,8 +237,8 @@ class HTTPAPGServer(web_resource.Resource):
) -> None:
if node is None:
node = self.apg._m.namespace
client = await self.apg.get_virtual_client(signing_actor)
objects = await self.apg.ap_get_list(data, "object")
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
objects = await self.apg.ap_get_list(requestor_actor_id, data, "object")
for obj in objects:
type_ = obj.get("type")
if type_ == "Follow":
@ -273,6 +276,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_delete_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -283,13 +287,14 @@ class HTTPAPGServer(web_resource.Resource):
):
if node is None:
node = self.apg._m.namespace
client = await self.apg.get_virtual_client(signing_actor)
objects = await self.apg.ap_get_list(data, "object")
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
objects = await self.apg.ap_get_list(requestor_actor_id, data, "object")
for obj in objects:
await self.apg.new_ap_delete_item(client, account_jid, node, obj)
async def handle_new_ap_items(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -308,20 +313,26 @@ class HTTPAPGServer(web_resource.Resource):
f"happen. Ignoring object from {signing_actor}\n{data}"
)
raise exceptions.DataError("unexpected field in item")
client = await self.apg.get_virtual_client(signing_actor)
objects = await self.apg.ap_get_list(data, "object")
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
objects = await self.apg.ap_get_list(requestor_actor_id, data, "object")
for obj in objects:
if node is None:
if obj.get("type") == TYPE_EVENT:
node = self.apg._events.namespace
else:
node = self.apg._m.namespace
sender = await self.apg.ap_get_sender_actor(obj)
sender = await self.apg.ap_get_sender_actor(requestor_actor_id, obj)
if repeated:
# we don't check sender when item is repeated, as it should be different
# from post author in this case
sender_jid = await self.apg.get_jid_from_id(sender)
repeater_jid = await self.apg.get_jid_from_id(signing_actor)
sender_jid = await self.apg.get_jid_from_id(
requestor_actor_id,
sender
)
repeater_jid = await self.apg.get_jid_from_id(
requestor_actor_id,
signing_actor
)
repeated_item_id = obj["id"]
if self.apg.is_local_url(repeated_item_id):
# the repeated object is from XMPP, we need to parse the URL to find
@ -374,6 +385,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_create_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -382,10 +394,13 @@ class HTTPAPGServer(web_resource.Resource):
ap_url: str,
signing_actor: str
):
await self.handle_new_ap_items(request, data, account_jid, node, signing_actor)
await self.handle_new_ap_items(
requestor_actor_id, request, data, account_jid, node, signing_actor
)
async def handle_update_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -396,10 +411,13 @@ class HTTPAPGServer(web_resource.Resource):
):
# Update is the same as create: the item ID stays the same, thus the item will be
# overwritten
await self.handle_new_ap_items(request, data, account_jid, node, signing_actor)
await self.handle_new_ap_items(
requestor_actor_id, request, data, account_jid, node, signing_actor
)
async def handle_announce_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -410,6 +428,7 @@ class HTTPAPGServer(web_resource.Resource):
):
# we create a new item
await self.handle_new_ap_items(
requestor_actor_id,
request,
data,
account_jid,
@ -516,6 +535,7 @@ class HTTPAPGServer(web_resource.Resource):
async def handle_like_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -524,11 +544,12 @@ class HTTPAPGServer(web_resource.Resource):
ap_url: str,
signing_actor: str
) -> None:
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
await self.handle_attachment_item(client, data, {"noticed": True})
async def handle_emojireact_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -537,13 +558,14 @@ class HTTPAPGServer(web_resource.Resource):
ap_url: str,
signing_actor: str
) -> None:
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
await self.handle_attachment_item(client, data, {
"reactions": {"operation": "update", "add": [data["content"]]}
})
async def handle_join_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -552,11 +574,12 @@ class HTTPAPGServer(web_resource.Resource):
ap_url: str,
signing_actor: str
) -> None:
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
await self.handle_attachment_item(client, data, {"rsvp": {"attending": "yes"}})
async def handle_leave_activity(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: dict,
account_jid: Optional[jid.JID],
@ -565,7 +588,7 @@ class HTTPAPGServer(web_resource.Resource):
ap_url: str,
signing_actor: str
) -> None:
client = await self.apg.get_virtual_client(signing_actor)
client = await self.apg.get_virtual_client(requestor_actor_id, signing_actor)
await self.handle_attachment_item(client, data, {"rsvp": {"attending": "no"}})
async def ap_actor_request(
@ -814,6 +837,7 @@ class HTTPAPGServer(web_resource.Resource):
async def ap_inbox_request(
self,
requestor_actor_id: str,
request: "HTTPRequest",
data: Optional[dict],
account_jid: Optional[jid.JID],
@ -825,7 +849,7 @@ class HTTPAPGServer(web_resource.Resource):
assert data is not None
if signing_actor is None:
raise exceptions.InternalError("signing_actor must be set for inbox requests")
await self.check_signing_actor(data, signing_actor)
await self.check_signing_actor(requestor_actor_id, data, signing_actor)
activity_type = (data.get("type") or "").lower()
if not activity_type in ACTIVITY_TYPES_LOWER:
return self.response_code(
@ -851,7 +875,8 @@ class HTTPAPGServer(web_resource.Resource):
)
else:
await method(
request, data, account_jid, node, ap_account, ap_url, signing_actor
requestor_actor_id, request, data, account_jid, node, ap_account, ap_url,
signing_actor
)
async def ap_followers_request(
@ -954,11 +979,73 @@ class HTTPAPGServer(web_resource.Resource):
to_log.append(f" headers:\n{headers}")
return to_log
def get_requestor_actor_id(
self,
data: dict|None = None,
uri_extra_args: list[str]|None = None
) -> str:
"""Find the actor ID of the requestor.
The requestor here is actually the local actor which will do the requests to
achieve the task (e.g. retrieve external actor data), not the requestor of the
received AP request.
It will notably be used as requestor actor ID to sign HTTP requests. We need to
sign GET request too to access instance checking HTTP GET signature (e.g. Mastodon
instances set in "secure mode").
We look for the destinee of the request and check if it's a local actor, and
default to a generic one if we can't find it.
Destinee is first checked in data if any, otherwise in request URI.
@param data: parsed JSON data of original AP request, if any.
@param uri_extra_args: arguments of the AP request as returned by
[self.apg.parse_apurl]. It is most of time the destinee of the request.
@return: requestor_actor_id to use to sign HTTP request needed to answer the
original request.
"""
# We first check for destinee in data.
if data:
try:
for to_ in data["to"]:
if self.apg.is_local_url(to_):
url_type, url_args = self.apg.parse_apurl(to_)
if url_type != TYPE_ACTOR or not url_args:
continue
ap_account = url_args[0]
if (
not ap_account.endswith(f"@{self.apg.public_url}")
or ap_account.count("@") != 1
):
continue
return to_
except KeyError:
pass
# If nothing relevant, we try URI arguments.
if uri_extra_args:
ap_account = uri_extra_args[0]
if (
ap_account.endswith(f"@{self.apg.public_url}")
and ap_account.count("@") == 1
):
return self.apg.build_apurl(TYPE_ACTOR, ap_account)
# Still nothing, we'll have to use a generic actor.
log.warning(
"Can't find destinee in \"to\" field, using generic requestor for signature."
)
return self.apg.build_apurl(
TYPE_ACTOR, f"libervia@{self.apg.public_url}"
)
async def ap_request(
self,
request: "HTTPRequest",
data: Optional[dict] = None,
signing_actor: Optional[str] = None
data: dict|None = None,
signing_actor: str|None = None,
requestor_actor_id: str|None = None,
) -> None:
if self.apg.verbose:
to_log = self._get_to_log(request, data)
@ -969,6 +1056,7 @@ class HTTPAPGServer(web_resource.Resource):
path
)
request_type, extra_args = self.apg.parse_apurl(ap_url)
header_accept = request.getHeader("accept") or ""
if ((MEDIA_TYPE_AP not in header_accept
and MEDIA_TYPE_AP_ALT not in header_accept
@ -1007,11 +1095,15 @@ class HTTPAPGServer(web_resource.Resource):
request.finish()
return
if requestor_actor_id is None:
requestor_actor_id = self.get_requestor_actor_id(
data, extra_args
)
if len(extra_args) == 0:
if request_type != "shared_inbox":
raise exceptions.DataError(f"Invalid request type: {request_type!r}")
ret_data = await self.ap_inbox_request(
request, data, None, None, None, ap_url, signing_actor
requestor_actor_id, request, data, None, None, None, ap_url, signing_actor
)
elif request_type == "avatar":
if len(extra_args) != 1:
@ -1034,7 +1126,7 @@ class HTTPAPGServer(web_resource.Resource):
raise exceptions.DataError(f"Invalid request type: {request_type!r}")
method = getattr(self, f"ap_{request_type}_request")
ret_data = await method(
request, data, account_jid, node, ap_account, ap_url, signing_actor
requestor_actor_id, request, data, account_jid, node, ap_account, ap_url, signing_actor
)
if ret_data is not None:
request.setHeader("content-type", CONTENT_TYPE_AP)
@ -1072,6 +1164,8 @@ class HTTPAPGServer(web_resource.Resource):
else:
request.content.seek(0)
requestor_actor_id = self.get_requestor_actor_id(data)
try:
if data["type"] == "Delete" and data["actor"] == data["object"]:
# we don't handle actor deletion
@ -1084,7 +1178,7 @@ class HTTPAPGServer(web_resource.Resource):
pass
try:
signing_actor = await self.check_signature(request)
signing_actor = await self.check_signature(requestor_actor_id, request)
except exceptions.EncryptionError as e:
if self.apg.verbose:
to_log = self._get_to_log(request)
@ -1118,29 +1212,42 @@ class HTTPAPGServer(web_resource.Resource):
# default response code, may be changed, e.g. in case of exception
try:
return await self.ap_request(request, data, signing_actor)
return await self.ap_request(
request, data, signing_actor, requestor_actor_id=requestor_actor_id
)
except Exception as e:
self._on_request_error(failure.Failure(e), request)
async def check_signing_actor(self, data: dict, signing_actor: str) -> None:
async def check_signing_actor(
self,
requestor_actor_id: str,
data: dict,
signing_actor: str
) -> None:
"""That that signing actor correspond to actor declared in data
@param requestor_actor_id: ID of the actor doing the request.
@param data: request payload
@param signing_actor: actor ID of the signing entity, as returned by
check_signature
@raise exceptions.NotFound: no actor found in data
@raise exceptions.EncryptionError: signing actor doesn't match actor in data
"""
actor = await self.apg.ap_get_sender_actor(data)
actor = await self.apg.ap_get_sender_actor(requestor_actor_id, data)
if signing_actor != actor:
raise exceptions.EncryptionError(
f"signing actor ({signing_actor}) doesn't match actor in data ({actor})"
)
async def check_signature(self, request: "HTTPRequest") -> str:
async def check_signature(
self,
requestor_actor_id: str,
request: "HTTPRequest"
) -> str:
"""Check and validate HTTP signature
@param requestor_actor_id: ID of the actor doing the request.
@return: id of the signing actor
@raise exceptions.EncryptionError: signature is not present or doesn't match
@ -1279,6 +1386,7 @@ class HTTPAPGServer(web_resource.Resource):
try:
return await self.apg.check_signature(
requestor_actor_id,
sign_data["signature"],
key_id,
headers
@ -1291,6 +1399,7 @@ class HTTPAPGServer(web_resource.Resource):
"see https://github.com/mastodon/mastodon/issues/18871"
)
return await self.apg.check_signature(
requestor_actor_id,
sign_data["signature"],
key_id,
headers

106
libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py
View file

@ -144,27 +144,51 @@ class APPubsubService(rsm.PubSubService):
[(subscription.subscriber, None, items)]
)
async def ap_following_2_elt(self, ap_item: dict) -> domish.Element:
"""Convert actor ID from following collection to XMPP item"""
async def ap_following_2_elt(self, requestor_actor_id: str, ap_item: dict) -> domish.Element:
"""Convert actor ID from following collection to XMPP item
@param requestor_actor_id: ID of the actor doing the request.
@param ap_item: AP item from which actor ID must be extracted.
"""
actor_id = ap_item["id"]
actor_jid = await self.apg.get_jid_from_id(actor_id)
actor_jid = await self.apg.get_jid_from_id(requestor_actor_id, actor_id)
subscription_elt = self.apg._pps.build_subscription_elt(
self.apg._m.namespace, actor_jid
)
item_elt = pubsub.Item(id=actor_id, payload=subscription_elt)
return item_elt
async def ap_follower_2_elt(self, ap_item: dict) -> domish.Element:
"""Convert actor ID from followers collection to XMPP item"""
async def ap_follower_2_elt(
self,
requestor_actor_id: str,
ap_item: dict
) -> domish.Element:
"""Convert actor ID from followers collection to XMPP item
@param requestor_actor_id: ID of the actor doing the request.
@param ap_item: AP item from which actor ID must be extracted.
"""
actor_id = ap_item["id"]
actor_jid = await self.apg.get_jid_from_id(actor_id)
actor_jid = await self.apg.get_jid_from_id(requestor_actor_id, actor_id)
subscriber_elt = self.apg._pps.build_subscriber_elt(actor_jid)
item_elt = pubsub.Item(id=actor_id, payload=subscriber_elt)
return item_elt
async def generate_v_card(self, ap_account: str) -> domish.Element:
"""Generate vCard4 (XEP-0292) item element from ap_account's metadata"""
actor_data = await self.apg.get_ap_actor_data_from_account(ap_account)
async def generate_v_card(
self,
requestor_actor_id: str,
ap_account: str
) -> domish.Element:
"""Generate vCard4 (XEP-0292) item element from ap_account's metadata
@param requestor_actor_id: ID of the actor doing the request.
@param ap_account: AP account from where the vcard must be retrieved.
@return: <item> with the <vcard> element
"""
actor_data = await self.apg.get_ap_actor_data_from_account(
requestor_actor_id,
ap_account
)
identity_data = {}
summary = actor_data.get("summary")
@ -190,16 +214,21 @@ class APPubsubService(rsm.PubSubService):
async def get_avatar_data(
self,
client: SatXMPPEntity,
requestor_actor_id: str,
ap_account: str
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Retrieve actor's avatar if any, cache it and file actor_data
``cache_uid``, `path``` and ``media_type`` keys are always files
``base64`` key is only filled if the file was not already in cache
@param client: client to use for the request.
@param requestor_actor_id: ID of the actor doing the request.
@param ap_account: AP account from where the avatar data must be retrieved.
@return: Avatar data.
``cache_uid``, `path``` and ``media_type`` keys are always filed.
``base64`` key is only filled if the file was not already in cache.
"""
actor_data = await self.apg.get_ap_actor_data_from_account(ap_account)
for icon in await self.apg.ap_get_list(actor_data, "icon"):
for icon in await self.apg.ap_get_list(requestor_actor_id, actor_data, "icon"):
url = icon.get("url")
if icon["type"] != "Image" or not url:
continue
@ -249,14 +278,16 @@ class APPubsubService(rsm.PubSubService):
async def generate_avatar_metadata(
self,
client: SatXMPPEntity,
requestor_actor_id: str,
ap_account: str
) -> domish.Element:
"""Generate the metadata element for user avatar
@param requestor_actor_id: ID of the actor doing the request.
@raise StanzaError("item-not-found"): no avatar is present in actor data (in
``icon`` field)
"""
avatar_data = await self.get_avatar_data(client, ap_account)
avatar_data = await self.get_avatar_data(client, requestor_actor_id, ap_account)
return self.apg._a.build_item_metadata_elt(avatar_data)
def _blocking_b_6_4_encode_avatar(self, avatar_data: Dict[str, Any]) -> None:
@ -266,17 +297,26 @@ class APPubsubService(rsm.PubSubService):
async def generate_avatar_data(
self,
client: SatXMPPEntity,
requestor_actor_id: str,
ap_account: str,
itemIdentifiers: Optional[List[str]],
) -> domish.Element:
"""Generate the data element for user avatar
@param requestor_actor_id: ID of the actor doing the request.
@raise StanzaError("item-not-found"): no avatar cached with requested ID
"""
if not itemIdentifiers:
avatar_data = await self.get_avatar_data(client, ap_account)
avatar_data = await self.get_avatar_data(
client,
requestor_actor_id,
ap_account
)
if "base64" not in avatar_data:
await threads.deferToThread(self._blocking_b_6_4_encode_avatar, avatar_data)
await threads.deferToThread(
self._blocking_b_6_4_encode_avatar,
avatar_data
)
else:
if len(itemIdentifiers) > 1:
# only a single item ID is supported
@ -312,6 +352,11 @@ class APPubsubService(rsm.PubSubService):
log.warning(f"Invalid AP account used by {requestor}: {ap_account!r}")
return [], None
requestor_actor_id = self.apg.build_apurl(
TYPE_ACTOR,
await self.apg.get_ap_account_from_jid_and_node(service, node)
)
# cached_node may be pre-filled with some nodes (e.g. attachments nodes),
# otherwise it is filled when suitable
cached_node = None
@ -330,14 +375,21 @@ class APPubsubService(rsm.PubSubService):
use_cache = False
elif node == self.apg._v.node:
# vCard4 request
item_elt = await self.generate_v_card(ap_account)
item_elt = await self.generate_v_card(requestor_actor_id, ap_account)
return [item_elt], None
elif node == self.apg._a.namespace_metadata:
item_elt = await self.generate_avatar_metadata(self.apg.client, ap_account)
item_elt = await self.generate_avatar_metadata(
self.apg.client,
requestor_actor_id,
ap_account
)
return [item_elt], None
elif node == self.apg._a.namespace_data:
item_elt = await self.generate_avatar_data(
self.apg.client, ap_account, itemIdentifiers
self.apg.client,
requestor_actor_id,
ap_account,
itemIdentifiers
)
return [item_elt], None
elif self.apg._pa.is_attachment_node(node):
@ -384,8 +436,8 @@ class APPubsubService(rsm.PubSubService):
if itemIdentifiers:
items = []
for item_id in itemIdentifiers:
item_data = await self.apg.ap_get(item_id)
item_elt = await parser(item_data)
item_data = await self.apg.ap_get(item_id, requestor_actor_id)
item_elt = await parser(requestor_actor_id, item_data)
items.append(item_elt)
return items, None
else:
@ -422,8 +474,9 @@ class APPubsubService(rsm.PubSubService):
if self.apg._m.is_comment_node(node):
parent_item = self.apg._m.get_parent_item(node)
try:
parent_data = await self.apg.ap_get(parent_item)
parent_data = await self.apg.ap_get(parent_item, requestor_actor_id)
collection = await self.apg.ap_get_object(
requestor_actor_id,
parent_data.get("object", {}),
"replies"
)
@ -433,8 +486,11 @@ class APPubsubService(rsm.PubSubService):
text=str(e)
)
else:
actor_data = await self.apg.get_ap_actor_data_from_account(ap_account)
collection = await self.apg.ap_get_object(actor_data, collection_name)
actor_data = await self.apg.get_ap_actor_data_from_account(
requestor_actor_id,
ap_account
)
collection = await self.apg.ap_get_object(requestor_actor_id, actor_data, collection_name)
if not collection:
raise error.StanzaError(
"item-not-found",
@ -442,7 +498,7 @@ class APPubsubService(rsm.PubSubService):
)
kwargs["parser"] = parser
return await self.apg.get_ap_items(collection, **kwargs)
return await self.apg.get_ap_items(requestor_actor_id, collection, **kwargs)
@ensure_deferred
async def retract(self, requestor, service, nodeIdentifier, itemIdentifiers):

120
tests/unit/test_ap-gateway.py
View file

@ -339,7 +339,11 @@ for item in TEST_AP_ITEMS:
AP_REQUESTS[item["id"]] = item
async def mock_ap_get(url):
async def mock_treq_get(url):
return deepcopy(AP_REQUESTS[url])
async def mock_ap_get(url, requestor_actor_id):
return deepcopy(AP_REQUESTS[url])
@ -514,13 +518,15 @@ class TestActivityPubGateway:
@ed
async def test_ap_to_pubsub_conversion(self, ap_gateway, monkeypatch):
"""AP requests are converted to pubsub"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
actor_data = await ap_gateway.get_ap_actor_data_from_account(TEST_AP_ACCOUNT)
outbox = await ap_gateway.ap_get_object(actor_data, "outbox")
items, rsm_resp = await ap_gateway.get_ap_items(outbox, 2)
actor_data = await ap_gateway.get_ap_actor_data_from_account(
TEST_AP_ACTOR_ID, TEST_AP_ACCOUNT
)
outbox = await ap_gateway.ap_get_object(TEST_AP_ACTOR_ID, actor_data, "outbox")
items, rsm_resp = await ap_gateway.get_ap_items(TEST_AP_ACTOR_ID, outbox, 2)
assert rsm_resp.count == 4
assert rsm_resp.index == 0
@ -552,6 +558,7 @@ class TestActivityPubGateway:
assert str(items[1].entry.published) == "2021-12-16T17:26:03Z"
items, rsm_resp = await ap_gateway.get_ap_items(
TEST_AP_ACTOR_ID,
outbox,
max_items=2,
after_id="https://example.org/users/test_user/statuses/3",
@ -586,7 +593,9 @@ class TestActivityPubGateway:
assert author_uri == "xmpp:test_user\\40example.org@ap.test.example"
assert str(items[1].entry.published) == "2021-12-16T17:28:03Z"
items, rsm_resp = await ap_gateway.get_ap_items(outbox, max_items=1, start_index=2)
items, rsm_resp = await ap_gateway.get_ap_items(
TEST_AP_ACTOR_ID, outbox, max_items=1, start_index=2
)
assert rsm_resp.count == 4
assert rsm_resp.index == 2
@ -603,7 +612,7 @@ class TestActivityPubGateway:
assert str(items[0].entry.published) == "2021-12-16T17:27:03Z"
items, rsm_resp = await ap_gateway.get_ap_items(
outbox, max_items=3, chronological_pagination=False
TEST_AP_ACTOR_ID, outbox, max_items=3, chronological_pagination=False
)
assert rsm_resp.count == 4
assert rsm_resp.index == 1
@ -670,6 +679,8 @@ class TestActivityPubGateway:
"ap_url": ap_url,
"signing_actor": signing_actor,
}
if type_ not in (ap_const.TYPE_ACTOR, "outbox", "following", "followers"):
kwargs["requestor_actor_id"] = TEST_AP_ACTOR_ID
if type_ == "outbox" and query_data:
kwargs["query_data"] = query_data
# signing_actor is not used for page requests
@ -722,7 +733,7 @@ class TestActivityPubGateway:
@ed
async def test_following_to_pps(self, ap_gateway, monkeypatch):
"""AP following items are converted to Public Pubsub Subscription subscriptions"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -746,7 +757,7 @@ class TestActivityPubGateway:
@ed
async def test_followers_to_pps(self, ap_gateway, monkeypatch):
"""AP followers items are converted to Public Pubsub Subscription subscribers"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -846,7 +857,7 @@ class TestActivityPubGateway:
@ed
async def test_xmpp_message_to_ap_direct_message(self, ap_gateway, monkeypatch):
"""XMPP message are sent as AP direct message"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
mess_data = {
@ -856,9 +867,9 @@ class TestActivityPubGateway:
"message": {"": "This is a test message."},
"extra": {"origin-id": "123"},
}
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
await ap_gateway.onMessage(ap_gateway.client, mess_data)
url, actor_id, doc = sign_and_post.call_args[0]
with patch.object(ap_gateway, "ap_post") as ap_post:
await ap_gateway.on_message(ap_gateway.client, mess_data)
url, actor_id, doc = ap_post.call_args[0]
assert url == "https://example.org/users/test_user/inbox"
assert actor_id == "https://test.example/_ap/actor/some_user@test.example"
obj = doc["object"]
@ -880,7 +891,7 @@ class TestActivityPubGateway:
@ed
async def test_ap_direct_message_to_xmpp_message(self, ap_gateway, monkeypatch):
"""AP direct message are sent as XMPP message (not Pubsub)"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
# we have to patch DeferredList to not wait forever
@ -914,7 +925,7 @@ class TestActivityPubGateway:
@ed
async def test_pubsub_retract_to_ap_delete(self, ap_gateway, monkeypatch):
"""Pubsub retract requests are converted to AP delete activity"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
retract_id = "retract_123"
@ -927,11 +938,11 @@ class TestActivityPubGateway:
items=[retract_elt],
headers={},
)
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value = FakeTReqPostResponse()
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value = FakeTReqPostResponse()
# we simulate the reception of a retract event
await ap_gateway._items_received(ap_gateway.client, items_event)
url, actor_id, doc = sign_and_post.call_args[0]
url, actor_id, doc = ap_post.call_args[0]
jid_account = await ap_gateway.get_ap_account_from_jid_and_node(TEST_JID, None)
jid_actor_id = ap_gateway.build_apurl(ap_const.TYPE_ACTOR, jid_account)
assert url == f"{TEST_BASE_URL}/inbox"
@ -1004,7 +1015,7 @@ class TestActivityPubGateway:
@ed
async def test_message_retract_to_ap_delete(self, ap_gateway, monkeypatch):
"""Message retract requests are converted to AP delete activity"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
# origin ID is the ID of the message to retract
@ -1021,8 +1032,8 @@ class TestActivityPubGateway:
message_retract_elt = fake_client.send.call_args.args[0]
retract_elt = next(message_retract_elt.elements(NS_MESSAGE_RETRACT, "retract"))
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value = FakeTReqPostResponse()
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value = FakeTReqPostResponse()
fake_fastened_elts = MagicMock()
fake_fastened_elts.id = origin_id
# we simulate the reception of a retract event using the message element that
@ -1030,9 +1041,9 @@ class TestActivityPubGateway:
await ap_gateway._on_message_retract(
ap_gateway.client, message_retract_elt, retract_elt, fake_fastened_elts
)
url, actor_id, doc = sign_and_post.call_args[0]
url, actor_id, doc = ap_post.call_args[0]
# the AP delete activity must have been sent through sign_and_post
# the AP delete activity must have been sent through ap_post
# we check its values
jid_account = await ap_gateway.get_ap_account_from_jid_and_node(TEST_JID, None)
jid_actor_id = ap_gateway.build_apurl(ap_const.TYPE_ACTOR, jid_account)
@ -1101,7 +1112,7 @@ class TestActivityPubGateway:
@ed
async def test_ap_actor_metadata_to_vcard(self, ap_gateway, monkeypatch):
"""AP actor metadata are converted to XMPP/vCard4"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1141,7 +1152,7 @@ class TestActivityPubGateway:
@ed
async def test_direct_addressing_mention_to_reference(self, ap_gateway, monkeypatch):
"""AP mentions by direct addressing are converted to XEP-0372 references"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1182,7 +1193,7 @@ class TestActivityPubGateway:
@ed
async def test_tag_mention_to_reference(self, ap_gateway, monkeypatch):
"""AP mentions in "tag" field are converted to XEP-0372 references"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1224,7 +1235,7 @@ class TestActivityPubGateway:
@ed
async def test_auto_mentions(self, ap_gateway, monkeypatch):
"""Check that mentions in body are converted to AP mentions"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1252,7 +1263,7 @@ class TestActivityPubGateway:
"""Check that no mention is send when the message is not public"""
# this is the same test as test_auto_mentions above, except that public is not set
# in mb_data_2_ap_item
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1273,7 +1284,7 @@ class TestActivityPubGateway:
@ed
async def test_xmpp_reference_to_ap_mention(self, ap_gateway, monkeypatch):
"""Check that XEP-0372 references are converted to AP mention"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1312,16 +1323,16 @@ class TestActivityPubGateway:
mock_pubsub_item = MagicMock
mock_pubsub_item.data = item_elt
get_items.return_value = ([mock_pubsub_item], {})
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value.code = 202
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value.code = 202
await ap_gateway._on_reference_received(
ap_gateway.client, message_elt, ref_data
)
# when reference is received, the referencing item must be sent to referenced
# actor, and they must be in "to" field and in "tag"
assert sign_and_post.call_count == 1
send_ap_item = sign_and_post.call_args.args[-1]
assert ap_post.call_count == 1
send_ap_item = ap_post.call_args.args[-1]
ap_object = send_ap_item["object"]
assert TEST_AP_ACTOR_ID in ap_object["to"]
expected_mention = {
@ -1336,7 +1347,7 @@ class TestActivityPubGateway:
@ed
async def test_xmpp_repeat_to_ap_announce(self, ap_gateway, monkeypatch):
"""XEP-0272 post repeat is converted to AP Announce activity"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1374,8 +1385,8 @@ class TestActivityPubGateway:
)
item_elt.uri = pubsub.NS_PUBSUB_EVENT
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value.code = 202
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value.code = 202
await ap_gateway.convert_and_post_items(
ap_gateway.client,
TEST_AP_ACCOUNT,
@ -1384,8 +1395,8 @@ class TestActivityPubGateway:
[item_elt],
)
assert sign_and_post.called
url, actor_id, doc = sign_and_post.call_args.args
assert ap_post.called
url, actor_id, doc = ap_post.call_args.args
assert url == TEST_USER_DATA["endpoints"]["sharedInbox"]
assert actor_id == ap_gateway.build_apurl(ap_const.TYPE_ACTOR, TEST_JID.userhost())
assert doc["type"] == "Announce"
@ -1395,7 +1406,7 @@ class TestActivityPubGateway:
@ed
async def test_ap_announce_to_xmpp_repeat(self, ap_gateway, monkeypatch):
"""AP Announce activity is converted to XEP-0272 post repeat"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1423,7 +1434,8 @@ class TestActivityPubGateway:
ap_gateway.host.memory.storage, "cache_pubsub_items"
) as cache_pubsub_items:
await ap_gateway.server.resource.handle_announce_activity(
Request(MagicMock()), announce, None, None, None, "", TEST_AP_ACTOR_ID
TEST_AP_ACTOR_ID, Request(MagicMock()), announce, None, None, None,
"", TEST_AP_ACTOR_ID
)
assert cache_pubsub_items.called
@ -1444,7 +1456,7 @@ class TestActivityPubGateway:
@ed
async def test_xmpp_attachment_noticed_to_ap_like(self, ap_gateway, monkeypatch):
"""Pubsub-attachments ``noticed`` is converted to AP Like activity"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1468,12 +1480,12 @@ class TestActivityPubGateway:
TEST_JID, recipient_jid, attachment_node, [item_elt], {}
)
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value.code = 202
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value.code = 202
await ap_gateway._items_received(ap_gateway.client, items_event)
assert sign_and_post.called
url, actor_id, doc = sign_and_post.call_args.args
assert ap_post.called
url, actor_id, doc = ap_post.call_args.args
assert url == TEST_USER_DATA["endpoints"]["sharedInbox"]
assert actor_id == ap_gateway.build_apurl(ap_const.TYPE_ACTOR, TEST_JID.userhost())
assert doc["type"] == "Like"
@ -1483,7 +1495,7 @@ class TestActivityPubGateway:
@ed
async def test_ap_like_to_xmpp_noticed_attachment(self, ap_gateway, monkeypatch):
"""AP Like activity is converted to ``noticed`` attachment"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1527,7 +1539,7 @@ class TestActivityPubGateway:
@ed
async def test_xmpp_pubsub_reactions_to_ap(self, ap_gateway, monkeypatch):
"""Pubsub-attachments ``reactions`` is converted to AP EmojiReact activity"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1562,12 +1574,12 @@ class TestActivityPubGateway:
TEST_JID, recipient_jid, attachment_node, [item_elt], {}
)
with patch.object(ap_gateway, "sign_and_post") as sign_and_post:
sign_and_post.return_value.code = 202
with patch.object(ap_gateway, "ap_post") as ap_post:
ap_post.return_value.code = 202
await ap_gateway._items_received(ap_gateway.client, items_event)
assert sign_and_post.call_count == 3
for idx, call_args in enumerate(sign_and_post.call_args_list):
assert ap_post.call_count == 3
for idx, call_args in enumerate(ap_post.call_args_list):
url, actor_id, doc = call_args.args
assert url == TEST_USER_DATA["endpoints"]["sharedInbox"]
assert actor_id == ap_gateway.build_apurl(
@ -1586,7 +1598,7 @@ class TestActivityPubGateway:
@ed
async def test_ap_reactions_to_xmpp(self, ap_gateway, monkeypatch):
"""AP EmojiReact activity is converted to ``reactions`` attachment"""
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_ap_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "get", mock_treq_get)
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
@ -1725,7 +1737,9 @@ class TestActivityPubGateway:
"url": f"https://test.example/_ap/item/{test_actor}/event_123",
}
event_data = await ap_gateway.ap_events.ap_item_2_event_data(ap_object)
event_data = await ap_gateway.ap_events.ap_item_2_event_data(
TEST_AP_ACTOR_ID, ap_object
)
assert event_data["id"] == ap_object["id"]
assert event_data["name"] == {"": "test event"}