Mercurial > libervia-backend
view sat/plugins/plugin_comp_ap_gateway/http_server.py @ 3836:e183f1fbfa8d
doc (cli/pubsub): add doc for ``reference`` subcommand:
rel 369
author | Goffi <goffi@goffi.org> |
---|---|
date | Sun, 10 Jul 2022 16:15:09 +0200 |
parents | 381340b9a9ee |
children | 65e5718e7710 |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia ActivityPub Gateway # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import time import html from typing import Optional, Dict, List, Any import json from urllib import parse from collections import deque import unicodedata from twisted.web import http, resource as web_resource, server from twisted.web import static from twisted.python import failure from twisted.internet import defer from twisted.words.protocols.jabber import jid, error from wokkel import pubsub, rsm from sat.core import exceptions from sat.core.constants import Const as C from sat.core.i18n import _ from sat.core.log import getLogger from sat.tools.common import date_utils from sat.memory.sqla_mapping import SubscriptionState from .constants import ( CONTENT_TYPE_AP, TYPE_ACTOR, TYPE_INBOX, TYPE_SHARED_INBOX, TYPE_OUTBOX, AP_REQUEST_TYPES, PAGE_SIZE, ACTIVITY_TYPES_LOWER, ACTIVIY_NO_ACCOUNT_ALLOWED, SIGN_HEADERS, HS2019, SIGN_EXP, TYPE_FOLLOWERS, TYPE_FOLLOWING ) from .regex import RE_SIG_PARAM log = getLogger(__name__) VERSION = unicodedata.normalize( 'NFKD', f"{C.APP_NAME} ActivityPub Gateway {C.APP_VERSION}" ) class HTTPAPGServer(web_resource.Resource): """HTTP Server handling ActivityPub S2S protocol""" isLeaf = True def __init__(self, ap_gateway): self.apg = ap_gateway self._seen_digest = deque(maxlen=50) super().__init__() def responseCode( self, request: "HTTPRequest", http_code: int, msg: Optional[str] = None ) -> None: """Log and set HTTP return code and associated message""" if msg is not None: log.warning(msg) request.setResponseCode(http_code, None if msg is None else msg.encode()) def _onRequestError(self, failure_: failure.Failure, request: "HTTPRequest") -> None: log.error(f"Internal error: {failure_.value}") self.responseCode( request, http.INTERNAL_SERVER_ERROR, f"internal error: {failure_.value}" ) request.finish() async def webfinger(self, request): url_parsed = parse.urlparse(request.uri.decode()) query = parse.parse_qs(url_parsed.query) resource = query.get("resource", [""])[0] account = resource[5:].strip() if not resource.startswith("acct:") or not account: return web_resource.ErrorPage( http.BAD_REQUEST, "Bad Request" , "Invalid webfinger resource" ).render(request) actor_url = self.apg.buildAPURL(TYPE_ACTOR, account) resp = { "subject": resource, "links": [ { "rel": "self", "type": "application/activity+json", "href": actor_url } ] } request.setHeader("content-type", CONTENT_TYPE_AP) request.write(json.dumps(resp).encode()) request.finish() async def handleUndoActivity( self, request: "HTTPRequest", data: dict, account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, signing_actor: str ) -> None: if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) objects = await self.apg.apGetList(data, "object") for obj in objects: type_ = obj.get("type") actor = await self.apg.apGetSenderActor(obj) if actor != signing_actor: log.warning(f"ignoring object not attributed to signing actor: {data}") continue try: target_account = obj["object"] except KeyError: log.warning(f'ignoring invalid object, missing "object" key: {data}') continue if not self.apg.isLocalURL(target_account): log.warning(f"ignoring unfollow request to non local actor: {data}") continue if type_ == "Follow": await self.apg._p.unsubscribe( client, account_jid, node, sender=client.jid, ) else: log.warning(f"Unmanaged undo type: {type_!r}") async def handleFollowActivity( self, request: "HTTPRequest", data: dict, account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, signing_actor: str ) -> None: if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) try: subscription = await self.apg._p.subscribe( client, account_jid, node, # subscriptions from AP are always public options=self.apg._pps.setPublicOpt() ) except pubsub.SubscriptionPending: log.info(f"subscription to node {node!r} of {account_jid} is pending") # TODO: manage SubscriptionUnconfigured else: if subscription.state != "subscribed": # other states should raise an Exception raise exceptions.InternalError('"subscribed" state was expected') inbox = await self.apg.getAPInboxFromId(signing_actor) actor_id = self.apg.buildAPURL(TYPE_ACTOR, ap_account) accept_data = self.apg.createActivity( "Accept", actor_id, object_=data ) await self.apg.signAndPost(inbox, actor_id, accept_data) await self.apg._c.synchronise(client, account_jid, node, resync=False) async def handleAcceptActivity( self, request: "HTTPRequest", data: dict, account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, signing_actor: str ) -> None: if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) objects = await self.apg.apGetList(data, "object") for obj in objects: type_ = obj.get("type") if type_ == "Follow": follow_node = await self.apg.host.memory.storage.getPubsubNode( client, client.jid, node, with_subscriptions=True ) if follow_node is None: log.warning( f"Received a follow accept on an unknown node: {node!r} at " f"{client.jid}. Ignoring it" ) continue try: sub = next( s for s in follow_node.subscriptions if s.subscriber==account_jid ) except StopIteration: log.warning( "Received a follow accept on a node without subscription: " f"{node!r} at {client.jid}. Ignoring it" ) else: if sub.state == SubscriptionState.SUBSCRIBED: log.warning(f"Already subscribed to {node!r} at {client.jid}") elif sub.state == SubscriptionState.PENDING: follow_node.subscribed = True sub.state = SubscriptionState.SUBSCRIBED await self.apg.host.memory.storage.add(follow_node) else: raise exceptions.InternalError( f"Unhandled subscription state {sub.state!r}" ) else: log.warning(f"Unmanaged accept type: {type_!r}") async def handleDeleteActivity( self, request: "HTTPRequest", data: dict, account_jid: Optional[jid.JID], node: Optional[str], ap_account: Optional[str], ap_url: str, signing_actor: str ): digest = request.getHeader("digest") if digest in self._seen_digest: log.debug(f"Ignoring duplicated request (digest: {digest!r})") return self._seen_digest.append(digest) if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) objects = await self.apg.apGetList(data, "object") for obj in objects: await self.apg.newAPDeleteItem(client, account_jid, node, obj) async def handleCreateActivity( self, request: "HTTPRequest", data: dict, account_jid: Optional[jid.JID], node: Optional[str], ap_account: Optional[str], ap_url: str, signing_actor: str ): digest = request.getHeader("digest") if digest in self._seen_digest: log.debug(f"Ignoring duplicated request (digest: {digest!r})") return self._seen_digest.append(digest) if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) objects = await self.apg.apGetList(data, "object") for obj in objects: sender = await self.apg.apGetSenderActor(obj) if sender != signing_actor: log.warning( "Ignoring object not attributed to signing actor: {obj}" ) else: await self.apg.newAPItem(client, account_jid, node, obj) async def APActorRequest( self, request: "HTTPRequest", account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, signing_actor: Optional[str] ) -> dict: inbox = self.apg.buildAPURL(TYPE_INBOX, ap_account) shared_inbox = self.apg.buildAPURL(TYPE_SHARED_INBOX) outbox = self.apg.buildAPURL(TYPE_OUTBOX, ap_account) followers = self.apg.buildAPURL(TYPE_FOLLOWERS, ap_account) following = self.apg.buildAPURL(TYPE_FOLLOWING, ap_account) # we have to use AP account as preferredUsername because it is used to retrieve # actor handle (see https://socialhub.activitypub.rocks/t/how-to-retrieve-user-server-tld-handle-from-actors-url/2196) preferred_username = ap_account.split("@", 1)[0] identity_data = await self.apg._i.getIdentity(self.apg.client, account_jid) actor_data = { "@context": [ "https://www.w3.org/ns/activitystreams", "https://w3id.org/security/v1" ], "id": ap_url, "type": "Person", "preferredUsername": preferred_username, "inbox": inbox, "outbox": outbox, "followers": followers, "following": following, "publicKey": { "id": f"{ap_url}#main-key", "owner": ap_url, "publicKeyPem": self.apg.public_key_pem }, "endpoints": { "sharedInbox": shared_inbox }, } if identity_data.get("nicknames"): actor_data["name"] = identity_data["nicknames"][0] if identity_data.get("description"): # description is plain text while summary expects HTML actor_data["summary"] = html.escape(identity_data["description"]) if identity_data.get("avatar"): avatar_data = identity_data["avatar"] try: filename = avatar_data["filename"] media_type = avatar_data["media_type"] except KeyError: log.error(f"incomplete avatar data: {identity_data!r}") else: avatar_url = self.apg.buildAPURL("avatar", filename) actor_data["icon"] = { "type": "Image", "url": avatar_url, "mediaType": media_type } return actor_data def getCanonicalURL(self, request: "HTTPRequest") -> str: return parse.urljoin( f"https://{self.apg.public_url}", request.path.decode().rstrip("/") ) def queryData2RSMRequest( self, query_data: Dict[str, List[str]] ) -> rsm.RSMRequest: """Get RSM kwargs to use with RSMRequest from query data""" page = query_data.get("page") if page == ["first"]: return rsm.RSMRequest(max_=PAGE_SIZE, before="") elif page == ["last"]: return rsm.RSMRequest(max_=PAGE_SIZE) else: for query_key in ("index", "before", "after"): try: kwargs={query_key: query_data[query_key][0], "max_": PAGE_SIZE} except (KeyError, IndexError, ValueError): pass else: return rsm.RSMRequest(**kwargs) raise ValueError(f"Invalid query data: {query_data!r}") async def APOutboxPageRequest( self, request: "HTTPRequest", account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, query_data: Dict[str, List[str]] ) -> dict: if node is None: node = self.apg._m.namespace # we only keep useful keys, and sort to have consistent URL which can # be used as ID url_keys = sorted(set(query_data) & {"page", "index", "before", "after"}) query_data = {k: query_data[k] for k in url_keys} try: items, metadata = await self.apg._p.getItems( client=self.apg.client, service=account_jid, node=node, rsm_request=self.queryData2RSMRequest(query_data), extra = {C.KEY_USE_CACHE: False} ) except error.StanzaError as e: log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") return {} base_url = self.getCanonicalURL(request) url = f"{base_url}?{parse.urlencode(query_data, True)}" data = { "@context": "https://www.w3.org/ns/activitystreams", "id": url, "type": "OrderedCollectionPage", "partOf": base_url, "orderedItems" : [ await self.apg.mbdata2APitem( self.apg.client, await self.apg._m.item2mbdata( self.apg.client, item, account_jid, node ) ) for item in reversed(items) ] } # AP OrderedCollection must be in reversed chronological order, thus the opposite # of what we get with RSM (at least with Libervia Pubsub) if not metadata["complete"]: try: last= metadata["rsm"]["last"] except KeyError: last = None data["prev"] = f"{base_url}?{parse.urlencode({'after': last})}" if metadata["rsm"]["index"] != 0: try: first= metadata["rsm"]["first"] except KeyError: first = None data["next"] = f"{base_url}?{parse.urlencode({'before': first})}" return data async def APOutboxRequest( self, request: "HTTPRequest", account_jid: jid.JID, node: Optional[str], ap_account: str, ap_url: str, signing_actor: Optional[str] ) -> dict: if node is None: node = self.apg._m.namespace parsed_url = parse.urlparse(request.uri.decode()) query_data = parse.parse_qs(parsed_url.query) if query_data: return await self.APOutboxPageRequest( request, account_jid, node, ap_account, ap_url, query_data ) # XXX: we can't use disco#info here because this request won't work on a bare jid # due to security considerations of XEP-0030 (we don't have presence # subscription). # The current workaround is to do a request as if RSM was available, and actually # check its availability according to result. try: __, metadata = await self.apg._p.getItems( client=self.apg.client, service=account_jid, node=node, max_items=0, rsm_request=rsm.RSMRequest(max_=0), extra = {C.KEY_USE_CACHE: False} ) except error.StanzaError as e: log.warning(f"Can't get data from pubsub node {node} at {account_jid}: {e}") return {} try: items_count = metadata["rsm"]["count"] except KeyError: log.warning( f"No RSM metadata found when requesting pubsub node {node} at " f"{account_jid}, defaulting to items_count=20" ) items_count = 20 url = self.getCanonicalURL(request) url_first_page = f"{url}?{parse.urlencode({'page': 'first'})}" url_last_page = f"{url}?{parse.urlencode({'page': 'last'})}" return { "@context": "https://www.w3.org/ns/activitystreams", "id": url, "totalItems": items_count, "type": "OrderedCollection", "first": url_first_page, "last": url_last_page, } async def APInboxRequest( self, request: "HTTPRequest", account_jid: Optional[jid.JID], node: Optional[str], ap_account: Optional[str], ap_url: str, signing_actor: Optional[str] ) -> None: if signing_actor is None: raise exceptions.InternalError("signing_actor must be set for inbox requests") if node is None: node = self.apg._m.namespace try: data = json.load(request.content) if not isinstance(data, dict): raise ValueError("data should be an object") except (json.JSONDecodeError, ValueError) as e: return self.responseCode( request, http.BAD_REQUEST, f"invalid json in inbox request: {e}" ) await self.checkSigningActor(data, signing_actor) activity_type = (data.get("type") or "").lower() if not activity_type in ACTIVITY_TYPES_LOWER: return self.responseCode( request, http.UNSUPPORTED_MEDIA_TYPE, f"request is not an activity, ignoring" ) if account_jid is None and activity_type not in ACTIVIY_NO_ACCOUNT_ALLOWED: return self.responseCode( request, http.UNSUPPORTED_MEDIA_TYPE, f"{activity_type.title()!r} activity must target an account" ) try: method = getattr(self, f"handle{activity_type.title()}Activity") except AttributeError: return self.responseCode( request, http.UNSUPPORTED_MEDIA_TYPE, f"{activity_type.title()} activity is not yet supported" ) else: await method( request, data, account_jid, node, ap_account, ap_url, signing_actor ) async def APFollowersRequest( self, request: "HTTPRequest", account_jid: jid.JID, node: Optional[str], ap_account: Optional[str], ap_url: str, signing_actor: Optional[str] ) -> dict: if node is None: node = self.apg._m.namespace client = self.apg.client subscribers = await self.apg._pps.getPublicNodeSubscriptions( client, account_jid, node ) followers = [] for subscriber in subscribers.keys(): if subscriber.host == self.apg.client.jid.userhost(): # the subscriber is an AP user subscribed with this gateway ap_account = self.apg._e.unescape(subscriber.user) else: # regular XMPP user ap_account = await self.apg.getAPAccountFromJidAndNode(subscriber, node) followers.append(ap_account) url = self.getCanonicalURL(request) return { "@context": "https://www.w3.org/ns/activitystreams", "type": "OrderedCollection", "id": url, "totalItems": len(subscribers), "first": { "type": "OrderedCollectionPage", "id": url, "orderedItems": followers } } async def APFollowingRequest( self, request: "HTTPRequest", account_jid: jid.JID, node: Optional[str], ap_account: Optional[str], ap_url: str, signing_actor: Optional[str] ) -> dict[str, Any]: client = self.apg.client subscriptions = await self.apg._pps.subscriptions( client, account_jid, node ) following = [] for sub_dict in subscriptions: service = jid.JID(sub_dict["service"]) if service.host == self.apg.client.jid.userhost(): # the subscription is to an AP actor with this gateway ap_account = self.apg._e.unescape(service.user) else: # regular XMPP user ap_account = await self.apg.getAPAccountFromJidAndNode( service, sub_dict["node"] ) following.append(ap_account) url = self.getCanonicalURL(request) return { "@context": "https://www.w3.org/ns/activitystreams", "type": "OrderedCollection", "id": url, "totalItems": len(subscriptions), "first": { "type": "OrderedCollectionPage", "id": url, "orderedItems": following } } async def APRequest( self, request: "HTTPRequest", signing_actor: Optional[str] = None ) -> None: path = request.path.decode() ap_url = parse.urljoin( f"https://{self.apg.public_url}", path ) request_type, extra_args = self.apg.parseAPURL(ap_url) if len(extra_args) == 0: if request_type != "shared_inbox": raise exceptions.DataError(f"Invalid request type: {request_type!r}") ret_data = await self.APInboxRequest( request, None, None, None, ap_url, signing_actor ) elif request_type == "avatar": if len(extra_args) != 1: raise exceptions.DataError("avatar argument expected in URL") avatar_filename = extra_args[0] avatar_path = self.apg.host.common_cache.getPath(avatar_filename) return static.File(str(avatar_path)).render(request) else: if len(extra_args) > 1: log.warning(f"unexpected extra arguments: {extra_args!r}") ap_account = extra_args[0] account_jid, node = await self.apg.getJIDAndNode(ap_account) if request_type not in AP_REQUEST_TYPES.get( request.method.decode().upper(), [] ): raise exceptions.DataError(f"Invalid request type: {request_type!r}") method = getattr(self, f"AP{request_type.title()}Request") ret_data = await method( request, account_jid, node, ap_account, ap_url, signing_actor ) if ret_data is not None: request.setHeader("content-type", CONTENT_TYPE_AP) request.write(json.dumps(ret_data).encode()) request.finish() async def APPostRequest(self, request: "HTTPRequest"): try: signing_actor = await self.checkSignature(request) except exceptions.EncryptionError as e: self.responseCode( request, http.FORBIDDEN, f"invalid signature: {e}" ) request.finish() return # default response code, may be changed, e.g. in case of exception request.setResponseCode(http.ACCEPTED) try: return await self.APRequest(request, signing_actor) except Exception as e: self._onRequestError(failure.Failure(e), request) async def checkSigningActor(self, data: dict, signing_actor: str) -> None: """That that signing actor correspond to actor declared in data @param data: request payload @param signing_actor: actor ID of the signing entity, as returned by checkSignature @raise exceptions.NotFound: no actor found in data @raise exceptions.EncryptionError: signing actor doesn't match actor in data """ actor = await self.apg.apGetSenderActor(data) if signing_actor != actor: raise exceptions.EncryptionError( f"signing actor ({signing_actor}) doesn't match actor in data ({actor})" ) async def checkSignature(self, request: "HTTPRequest") -> str: """Check and validate HTTP signature @return: id of the signing actor @raise exceptions.EncryptionError: signature is not present or doesn't match """ signature = request.getHeader("Signature") if signature is None: raise exceptions.EncryptionError("No signature found") sign_data = { m["key"]: m["uq_value"] or m["quoted_value"][1:-1] for m in RE_SIG_PARAM.finditer(signature) } try: key_id = sign_data["keyId"] except KeyError: raise exceptions.EncryptionError('"keyId" is missing from signature') algorithm = sign_data.get("algorithm", HS2019) signed_headers = sign_data.get( "headers", "(created)" if algorithm==HS2019 else "date" ).lower().split() try: headers_to_check = SIGN_HEADERS[None] + SIGN_HEADERS[request.method] except KeyError: raise exceptions.InternalError( f"there should be a list of headers for {request.method} method" ) if not headers_to_check: raise exceptions.InternalError("headers_to_check must not be empty") for header in headers_to_check: if isinstance(header, tuple): if len(set(header).intersection(signed_headers)) == 0: raise exceptions.EncryptionError( f"at least one of following header must be signed: {header}" ) elif header not in signed_headers: raise exceptions.EncryptionError( f"the {header!r} header must be signed" ) body = request.content.read() request.content.seek(0) headers = {} for to_sign in signed_headers: if to_sign == "(request-target)": method = request.method.decode().lower() uri = parse.unquote(request.uri.decode()) headers[to_sign] = f"{method} /{uri.lstrip('/')}" elif to_sign in ("(created)", "(expires)"): if algorithm != HS2019: raise exceptions.EncryptionError( f"{to_sign!r} pseudo-header can only be used with {HS2019} " "algorithm" ) key = to_sign[1:-1] value = sign_data.get(key) if not value: raise exceptions.EncryptionError( "{key!r} parameter is missing from signature" ) try: if float(value) < 0: raise ValueError except ValueError: raise exceptions.EncryptionError( f"{to_sign} must be a Unix timestamp" ) headers[to_sign] = value else: value = request.getHeader(to_sign) if not value: raise exceptions.EncryptionError( f"value of header {to_sign!r} is missing!" ) elif to_sign == "host": # we check Forwarded/X-Forwarded-Host headers # as we need original host if a proxy has modified the header forwarded = request.getHeader("forwarded") if forwarded is not None: try: host = [ f[5:] for f in forwarded.split(";") if f.startswith("host=") ][0] or None except IndexError: host = None else: host = None if host is None: host = request.getHeader("x-forwarded-host") if host: value = host elif to_sign == "digest": hashes = { algo.lower(): hash_ for algo, hash_ in ( digest.split("=", 1) for digest in value.split(",") ) } try: given_digest = hashes["sha-256"] except KeyError: raise exceptions.EncryptionError( "Only SHA-256 algorithm is currently supported for digest" ) __, computed_digest = self.apg.getDigest(body) if given_digest != computed_digest: raise exceptions.EncryptionError( f"SHA-256 given and computed digest differ:\n" f"given: {given_digest!r}\ncomputed: {computed_digest!r}" ) headers[to_sign] = value # date check limit_ts = time.time() + SIGN_EXP if "(created)" in headers: created = float(headers["created"]) else: created = date_utils.date_parse(headers["date"]) try: expires = float(headers["expires"]) except KeyError: pass else: if expires < created: log.warning( f"(expires) [{expires}] set in the past of (created) [{created}] " "ignoring it according to specs" ) else: limit_ts = min(limit_ts, expires) if created > limit_ts: raise exceptions.EncryptionError("Signature has expired") return await self.apg.checkSignature( sign_data["signature"], key_id, headers ) def render(self, request): request.setHeader("server", VERSION) return super().render(request) def render_GET(self, request): path = request.path.decode().lstrip("/") if path.startswith(".well-known/webfinger"): defer.ensureDeferred(self.webfinger(request)) return server.NOT_DONE_YET elif path.startswith(self.apg.ap_path): d = defer.ensureDeferred(self.APRequest(request)) d.addErrback(self._onRequestError, request) return server.NOT_DONE_YET return web_resource.NoResource().render(request) def render_POST(self, request): path = request.path.decode().lstrip("/") if not path.startswith(self.apg.ap_path): return web_resource.NoResource().render(request) defer.ensureDeferred(self.APPostRequest(request)) return server.NOT_DONE_YET class HTTPRequest(server.Request): pass class HTTPServer(server.Site): requestFactory = HTTPRequest def __init__(self, ap_gateway): super().__init__(HTTPAPGServer(ap_gateway))