Limnoria/plugins/Fediverse/plugin.py

408 lines
14 KiB
Python
Raw Normal View History

2020-05-09 19:14:56 +02:00
###
# Copyright (c) 2020, Valentin Lorentz
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# * Redistributions of source code must retain the above copyright notice,
# this list of conditions, and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions, and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of the author of this software nor the name of
# contributors to this software may be used to endorse or promote products
# derived from this software without specific prior written consent.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
###
import re
import json
import importlib
import urllib.parse
2020-05-09 20:29:24 +02:00
from supybot import utils, callbacks, httpserver
2020-05-10 14:52:55 +02:00
from supybot.commands import urlSnarfer, wrap
2020-05-09 19:14:56 +02:00
from supybot.i18n import PluginInternationalization
from . import activitypub as ap
2020-05-09 20:29:24 +02:00
2020-05-09 19:14:56 +02:00
importlib.reload(ap)
2020-05-09 20:29:24 +02:00
_ = PluginInternationalization("Fediverse")
2020-05-09 20:47:11 +02:00
_username_regexp = re.compile("@(?P<localuser>[^@ ]+)@(?P<hostname>[^@ ]+)")
2020-05-09 19:14:56 +02:00
class FediverseHttp(httpserver.SupyHTTPServerCallback):
name = "minimal ActivityPub server"
defaultResponse = _(
"""
You shouldn't be here, this subfolder is not for you. Go back to the
index and try out other plugins (if any)."""
)
def doGetOrHead(self, handler, path, write_content):
if path == "/instance_actor":
self.instance_actor(write_content)
else:
assert False, repr(path)
def doWellKnown(self, handler, path):
actor_url = ap.get_instance_actor_url()
instance_hostname = urllib.parse.urlsplit(actor_url).hostname
instance_account = "acct:%s@%s" % (
instance_hostname,
instance_hostname,
)
if path == "/webfinger?resource=%s" % instance_account:
headers = {"Content-Type": "application/jrd+json"}
content = {
"subject": instance_account,
"links": [
{
"rel": "self",
"type": "application/activity+json",
"href": actor_url,
}
],
}
return (200, headers, json.dumps(content).encode())
else:
return None
def instance_actor(self, write_content):
self.send_response(200)
self.send_header("Content-type", ap.ACTIVITY_MIMETYPE)
self.end_headers()
if not write_content:
return
pem = ap.get_public_key_pem()
actor_url = ap.get_instance_actor_url()
2020-05-09 20:29:24 +02:00
hostname = urllib.parse.urlparse(actor_url).hostname
2020-05-09 19:14:56 +02:00
actor = {
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1",
],
"id": actor_url,
"preferredUsername": hostname,
2020-05-10 14:52:55 +02:00
"type": "Service",
2020-05-09 19:14:56 +02:00
"publicKey": {
"id": actor_url + "#main-key",
"owner": actor_url,
"publicKeyPem": pem.decode(),
},
"inbox": actor_url + "/inbox",
}
self.wfile.write(json.dumps(actor).encode())
2020-05-09 20:47:11 +02:00
class Fediverse(callbacks.PluginRegexp):
2020-05-09 19:14:56 +02:00
"""Fetches information from ActivityPub servers."""
threaded = True
2020-05-10 14:52:55 +02:00
regexps = ["usernameSnarfer", "urlSnarfer_"]
callBefore = ("Web",)
2020-05-09 19:14:56 +02:00
def __init__(self, irc):
super().__init__(irc)
self._startHttp()
self._actor_cache = utils.structures.TimeoutDict(timeout=600)
2020-05-09 19:14:56 +02:00
# Used when snarfing, to cheaply avoid querying non-ActivityPub
# servers.
# Is also written to when using commands that successfully find
# ActivityPub data.
self._webfinger_support_cache = utils.structures.TimeoutDict(
timeout=60 * 60 * 24
)
2020-05-09 19:14:56 +02:00
def _startHttp(self):
callback = FediverseHttp()
callback._plugin = self
httpserver.hook("fediverse", callback)
def die(self):
self._stopHttp()
super().die()
def _stopHttp(self):
httpserver.unhook("fediverse")
def _has_webfinger_support(self, hostname):
if hostname not in self._webfinger_support_cache:
self._webfinger_support_cache[hostname] = ap.has_webfinger_support(
hostname
)
return self._webfinger_support_cache[hostname]
2020-05-09 20:29:24 +02:00
def _get_actor(self, irc, username):
if username in self._actor_cache:
return self._actor_cache[username]
2020-05-09 20:47:11 +02:00
match = _username_regexp.match(username)
2020-05-09 21:39:58 +02:00
if match:
localuser = match.group("localuser")
hostname = match.group("hostname")
try:
actor = ap.get_actor(localuser, hostname)
except ap.ActorNotFound:
irc.error("Unknown user %s." % username, Raise=True)
else:
match = utils.web.urlRe.match(username)
if match:
# TODO: error handling
2020-05-10 14:52:55 +02:00
actor = ap.get_resource_from_url(match.group(0))
try:
hostname = urllib.parse.urlparse(actor.get("id")).hostname
username = "@%s@%s" % (
hostname,
actor.get["preferredUsername"],
)
except Exception:
username = None
2020-05-09 21:39:58 +02:00
else:
irc.errorInvalid("fediverse username", username)
2020-05-09 19:14:56 +02:00
if username:
self._actor_cache[username] = actor
self._webfinger_support_cache[hostname] = True
2020-05-09 20:29:24 +02:00
self._actor_cache[actor["id"]] = actor
return actor
def _format_actor_fullname(self, actor):
try:
hostname = urllib.parse.urlparse(actor.get("id")).hostname
except Exception:
hostname = "<unknown>"
username = actor.get("preferredUsername", "<unknown>")
name = actor.get("name", username)
return "\x02%s\x02 (@%s@%s)" % (name, username, hostname)
2020-05-09 20:29:24 +02:00
def _format_status(self, irc, msg, status):
2020-05-09 21:39:58 +02:00
if status["type"] == "Create":
return self._format_status(irc, msg, status["object"])
2020-05-09 21:39:58 +02:00
elif status["type"] == "Note":
author_url = status["attributedTo"]
try:
author = self._get_actor(irc, author_url)
except ap.ActivityPubError as e:
author_fullname = _("<error: %s>") % str(e)
else:
author_fullname = self._format_actor_fullname(author)
2020-05-10 13:04:01 +02:00
cw = status.get("summary")
if cw:
if self.registryValue(
"format.statuses.showContentWithCW",
msg.channel,
irc.network,
):
# show CW and content
2020-05-15 18:51:19 +02:00
res = [_("%s: \x02[CW %s]\x02 %s") % (
author_fullname,
cw,
utils.web.htmlToText(status["content"]),
2020-05-15 18:51:19 +02:00
)]
else:
# show CW but not content
2020-05-15 18:51:19 +02:00
res = [_("%s: CW %s") % (author_fullname, cw)]
2020-05-10 13:04:01 +02:00
else:
# no CW, show content
2020-05-15 18:51:19 +02:00
res = [_("%s: %s") % (
author_fullname,
2020-05-10 13:04:01 +02:00
utils.web.htmlToText(status["content"]),
2020-05-15 18:51:19 +02:00
)]
for attachment in status.get("attachment", []):
res.append(utils.str.url(attachment.get("url")))
return " ".join(res)
2020-05-09 21:39:58 +02:00
elif status["type"] == "Announce":
# aka boost; let's go fetch the original status
try:
content = ap.signed_request(
status["object"], headers={"Accept": ap.ACTIVITY_MIMETYPE}
)
status = json.loads(content.decode())
return self._format_status(irc, msg, status)
2020-05-09 21:39:58 +02:00
except ap.ActivityPubProtocolError as e:
return "<Could not fetch status: %s>" % e.args[0]
else:
assert False, "Unknown status type %s: %r" % (
status["type"],
status,
)
2020-05-09 20:29:24 +02:00
@wrap(["somethingWithoutSpaces"])
def profile(self, irc, msg, args, username):
"""<@user@instance>
Returns generic information on the account @user@instance."""
actor = self._get_actor(irc, username)
2020-05-09 19:14:56 +02:00
irc.reply(
_("%s: %s")
2020-05-09 19:14:56 +02:00
% (
self._format_actor_fullname(actor),
2020-05-09 20:29:24 +02:00
utils.web.htmlToText(actor["summary"]),
2020-05-09 19:14:56 +02:00
)
)
2020-05-10 14:52:55 +02:00
def _format_profile(self, irc, msg, actor):
return _("%s: %s") % (
self._format_actor_fullname(actor),
2020-05-10 14:52:55 +02:00
utils.web.htmlToText(actor["summary"]),
)
2020-05-09 20:47:11 +02:00
def usernameSnarfer(self, irc, msg, match):
2020-05-09 21:39:58 +02:00
if callbacks.addressed(irc, msg):
return
2020-05-10 14:52:55 +02:00
if not self.registryValue(
"snarfers.username", msg.channel, irc.network
):
return
if not self._has_webfinger_support(match.group("hostname")):
self.log.debug(
"Not snarfing, host doesn't have Webfinger support."
)
return
2020-05-09 20:47:11 +02:00
try:
actor = self._get_actor(irc, match.group(0))
except ap.ActivityPubError as e:
self.log.info("Could not fetch %s: %s", match.group(0), e)
2020-05-09 20:47:11 +02:00
# Be silent on errors
return
2020-05-10 14:52:55 +02:00
irc.reply(self._format_profile(irc, msg, actor))
2020-05-09 20:55:28 +02:00
2020-05-09 20:47:11 +02:00
usernameSnarfer.__doc__ = _username_regexp.pattern
2020-05-10 14:52:55 +02:00
@urlSnarfer
def urlSnarfer_(self, irc, msg, match):
channel = msg.channel
network = irc.network
url = match.group(0)
if not channel:
return
if callbacks.addressed(irc, msg):
return
snarf_profile = self.registryValue(
"snarfers.profile", channel, network
)
snarf_status = self.registryValue("snarfers.status", channel, network)
if not snarf_profile and not snarf_status:
return
hostname = urllib.parse.urlparse(url).hostname
if not self._has_webfinger_support(hostname):
self.log.debug(
"Not snarfing, host doesn't have Webfinger support."
)
return
2020-05-10 14:52:55 +02:00
try:
resource = ap.get_resource_from_url(url)
except ap.ActivityPubError:
return
try:
if snarf_profile and resource["type"] in ("Person", "Service"):
irc.reply(self._format_profile(irc, msg, resource))
elif snarf_status and resource["type"] in (
"Create",
"Note",
"Announce",
):
irc.reply(self._format_status(irc, msg, resource))
except ap.ActivityPubError:
return
urlSnarfer_.__doc__ = utils.web._httpUrlRe
2020-05-09 20:29:24 +02:00
@wrap(["somethingWithoutSpaces"])
def featured(self, irc, msg, args, username):
"""<@user@instance>
2020-05-09 21:39:58 +02:00
Returnes the featured statuses of @user@instance (aka. pinned toots).
2020-05-09 20:29:24 +02:00
"""
actor = self._get_actor(irc, username)
if "featured" not in actor:
2020-05-10 10:29:01 +02:00
irc.reply(_("No featured statuses."))
return
statuses = json.loads(
ap.signed_request(actor["featured"]).decode()
).get("orderedItems", [])
2020-05-10 10:29:01 +02:00
if not statuses:
irc.reply(_("No featured statuses."))
return
2020-05-09 21:39:58 +02:00
irc.replies(
filter(
bool,
(self._format_status(irc, msg, status) for status in statuses),
2020-05-09 21:39:58 +02:00
)
)
@wrap(["somethingWithoutSpaces"])
def statuses(self, irc, msg, args, username):
"""<@user@instance>
Returned the last statuses of @user@instance.
"""
actor = self._get_actor(irc, username)
if "outbox" not in actor:
irc.error(_("No status."), Raise=True)
outbox = json.loads(ap.signed_request(actor["outbox"]).decode())
2020-05-09 21:39:58 +02:00
# Fetches the first page of the outbox. This should be a good-enough
# approximation of the number of statuses to show.
statuses = json.loads(ap.signed_request(outbox["first"]).decode()).get(
2020-05-09 21:39:58 +02:00
"orderedItems", []
)
irc.replies(
filter(
bool,
(self._format_status(irc, msg, status) for status in statuses),
2020-05-09 21:39:58 +02:00
)
)
2020-05-09 20:29:24 +02:00
2020-05-10 21:12:05 +02:00
@wrap(["url"])
def status(self, irc, msg, args, url):
"""<url>
Shows the content of the status at <url>.
"""
try:
status = ap.get_resource_from_url(url)
except ap.ActivityPubError as e:
irc.error(_("Could not get status: %s") % e.args[0], Raise=True)
else:
hostname = urllib.parse.urlparse(url).hostname
self._webfinger_support_cache[hostname] = True
2020-05-10 21:12:05 +02:00
irc.reply(self._format_status(irc, msg, status))
2020-05-09 19:14:56 +02:00
Class = Fediverse
# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: