diff options
Diffstat (limited to 'modules/fediverse_server/__init__.py')
| -rw-r--r-- | modules/fediverse_server/__init__.py | 226 |
1 files changed, 226 insertions, 0 deletions
diff --git a/modules/fediverse_server/__init__.py b/modules/fediverse_server/__init__.py new file mode 100644 index 00000000..82f03fc8 --- /dev/null +++ b/modules/fediverse_server/__init__.py @@ -0,0 +1,226 @@ +#--require-config tls-certificate + +import base64, binascii, os, urllib.parse +from src import ModuleManager, utils + +from cryptography.hazmat.primitives import serialization, hashes +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.backends import default_backend + +LD_TYPE = ("application/ld+json; " + "profile=\"https://www.w3.org/ns/activitystreams\"") +JRD_TYPE = "application/jrd+json" +ACTIVITY_TYPE = "application/activity+json" + +ACTIVITY_SETTING_PREFIX = "ap-activity-" + +def _parse_username(s): + username, _, instance = s.rpartition("@") + if username.startswith("@"): + username = username[1:] + if username and instance: + return username, instance + return None, None +def _format_username(username, instance): + return "@%s@%s" % (username, instance) +def _setting_parse(s): + username, instance = _parse_username(s) + if username and instance: + return _format_username(username, instance) + return None + +@utils.export("botset", utils.FunctionSetting(_setting_parse, "fediverse", + help="Set the bot's fediverse server account", + example="@gargron@mastodon.social")) +class Module(ModuleManager.BaseModule): + _name = "Fedi" + + def _random_id(self): + return binascii.hexlify(os.urandom(3)).decode("ascii") + + def _get_activities(self): + activities = [] + for setting, (content, timestamp) in self.bot.find_settings_prefix( + ACTIVITY_SETTING_PREFIX): + activity_id = setting.replace(ACTIVITY_SETTING_PREFIX, "", 1) + activities.append([activity_id, content, timestamp]) + return activities + def _make_activity(self, content): + timestamp = utils.iso8601_format_now() + activity_id = self._random_id() + self.bot.set_setting("ap-activity-%s" % activity_id, + [content, timestamp]) + return activity_id + + @utils.hook("received.command.toot") + @utils.kwarg("min_args", 1) + @utils.kwarg("permission", "toot") + def toot(self, event): + activity_id = self._make_activity(event["args"]) + event["stdout"].write("Sent toot %s" % activity_id) + + def _federate_activity(self, activity_id, content, timestamp): + + message = { + "@context": "https://www.w3.org/ns/activitystreams", + "type": "Announce", + "to": [], + "actor": "", + "object": "" + } + + + def _federate(self, data): + our_username, our_instance = self._ap_self() + key_id = self._ap_keyid_url(url_for, our_username) + now = email.utils.formatdate(timeval=None, localtime=False, usegmt=True) + url_for = self.exports.get_one("url-for") + + key = security.private_key(self.bot.config["tls-certificate"]) + + for inbox in self._get_inboxes(): + parts = urllib.parse.urlparse(inbox) + headers = [ + ["host", parts.netloc], + ["date", now] + ] + sign_headers = headers[:] + sign_headers.insert(0, ["(request-target)", "post %s" % parts.path]) + + signature = security.signature(key, key_id, sign_headers) + data = "" + request = utils.http.Request(inbox, data=data, headers=headers, + content_type=ACTIVITY_TYPE, useragent="BitBot Fediverse") + utils.http.request() + + def _ap_self(self): + our_username = self.bot.get_setting("fediverse", None) + return _parse_username(our_username) + + def _ap_url(self, url_for, fragment, kwargs): + return "https://%s" % url_for("api", fragment, kwargs) + def _ap_self_url(self, url_for, our_username): + return self._ap_url(url_for, "ap-user", {"u": our_username}) + def _ap_inbox_url(self, url_for, our_username): + return self._ap_url(url_for, "ap-inbox", {"u": our_username}) + def _ap_outbox_url(self, url_for, our_username): + return self._ap_url(url_for, "ap-outbox", {"u": our_username}) + def _ap_activity_url(self, url_for, activity_id): + return self._ap_url(url_for, "ap-activity", {"a": activity_id}) + def _ap_keyid_url(self, url_for, our_username): + return "%s#key" % self._ap_self_url(url_for, our_username) + + @utils.hook("api.get.ap-webfinger") + @utils.kwarg("authenticated", False) + def ap_webfinger(self, event): + our_username, our_instance = self._ap_self() + + resource = event["params"].get("resource", None) + if resource.startswith("acct:"): + resource = resource.split(":", 1)[1] + + if resource: + requested_username, requested_instance = _parse_username(resource) + + if (requested_username == our_username and + requested_instance == our_instance): + + self_id = self._ap_self_url(event["url_for"], our_username) + + event["response"].content_type = JRD_TYPE + event["response"].write_json({ + "aliases": [self_id], + "links": [{ + "href": self_id, + "rel": "self", + "type": ACTIVITY_TYPE + }], + "subject": "acct:%s" % resource + }) + else: + event["response"].code = 404 + else: + event["response"].code = 400 + + @utils.hook("api.get.ap-user") + @utils.kwarg("authenticated", False) + def ap_user(self, event): + our_username, our_instance = self._ap_self() + username = event["params"].get("u", None) + + if username and username == our_username: + self_id = self._ap_self_url(event["url_for"], our_username) + inbox = self._ap_inbox_url(event["url_for"], our_username) + outbox = self._ap_outbox_url(event["url_for"], our_username) + + cert_filename = self.bot.config["tls-certificate"] + with open(cert_filename) as cert_file: + cert = cert_file.read().strip() + + event["response"].content_type = LD_TYPE + event["response"].write_json({ + "@context": "https://www.w3.org/ns/activitystreams", + "id": self_id, "url": self_id, + "type": "Person", + "summary": "beep boop", + "preferredUsername": our_username, "name": our_username, + "inbox": inbox, + "outbox": outbox, + "publicKey": { + "id": "%s#key" % self_id, + "owner": self_id, + "publicKeyPem": cert + } + }) + else: + event["response"].code = 404 + + def _prepare_activity(self, url_for, self_id, activity_id, content, + timestamp): + activity_url = self._ap_activity_url(url_for, activity_id) + context = "data:%s" % activity_id + return activity_url, { + "attributedTo": self_id, + "content": content, + "conversation": context, "context": context, + "id": activity_url, "url": activity_url, + "published": timestamp, + "summary": "", # content warning here + "to": "https://www.w3.org/ns/activitystreams#Public", + "type": "Note", + } + + @utils.hook("api.get.ap-outbox") + @utils.kwarg("authenticated", False) + def ap_outbox(self, event): + our_username, our_instance = self._ap_self() + username = event["params"].get("u", None) + if username and username == our_username: + self_id = self._ap_self_url(event["url_for"], our_username) + outbox = self._ap_outbox_url(event["url_for"], our_username) + + activities = [] + for activity_id, content, timestamp in self._get_activities(): + activity_url, activity_object = self._prepare_activity( + event["url_for"], self_id, activity_id, content, timestamp) + activities.append({ + "actor": self_id, + "id": activity_url, + "object": activity_object, + "published": timestamp, + "to": "https://www.w3.org/ns/activitystreams#Public", + "type": "Create" + }) + + event["response"].content_type = LD_TYPE + event["response"].write_json({ + "@context": "https://www.w3.org/ns/activitystreams", + "id": outbox, + "orderedItems": activities, + "totalItems": len(activities), + "type": "OrderedCollection" + }) + + else: + event["response"].code = 404 + |