Limnoria/plugins/Fediverse/plugin.py

345 lines
12 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
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")
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))
2020-05-09 21:39:58 +02:00
username = self._format_actor_username(actor)
else:
irc.errorInvalid("fediverse username", username)
2020-05-09 19:14:56 +02:00
2020-05-09 20:29:24 +02:00
self._actor_cache[username] = actor
self._actor_cache[actor["id"]] = actor
return actor
def _format_actor_username(self, actor):
hostname = urllib.parse.urlparse(actor["id"]).hostname
return "@%s@%s" % (actor["preferredUsername"], hostname)
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"]
author = self._get_actor(irc, author_url)
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
return _("\x02%s (%s)\x02: \x02[CW %s]\x02 %s") % (
author["name"],
self._format_actor_username(author),
cw,
utils.web.htmlToText(status["content"]),
)
else:
# show CW but not content
return _("\x02%s (%s)\x02: CW %s") % (
author["name"],
self._format_actor_username(author),
cw,
)
2020-05-10 13:04:01 +02:00
else:
# no CW, show content
2020-05-10 13:04:01 +02:00
return _("\x02%s (%s)\x02: %s") % (
author["name"],
self._format_actor_username(author),
utils.web.htmlToText(status["content"]),
)
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(
2020-05-09 20:29:24 +02:00
_("\x02%s\x02 (%s): %s")
2020-05-09 19:14:56 +02:00
% (
actor["name"],
2020-05-09 20:29:24 +02:00
self._format_actor_username(actor),
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 _("\x02%s\x02 (%s): %s") % (
actor["name"],
self._format_actor_username(actor),
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
2020-05-09 20:47:11 +02:00
try:
actor = self._get_actor(irc, match.group(0))
except ap.ActivityPubError:
# 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
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-09 19:14:56 +02:00
Class = Fediverse
# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: