re-merge fediverse an fediverse_server, so they can share utils

2 files changed, 0 insertions, 258 deletions

diff --git a/modules/fediverse_server/__init__.py b/modules/fediverse_server/__init__.py
deleted file mode 100644
index 82f03fc8..00000000
--- a/modules/fediverse_server/__init__.py
+++ /dev/null
@@ -1,226 +0,0 @@
-#--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
-
diff --git a/modules/fediverse_server/security.py b/modules/fediverse_server/security.py
deleted file mode 100644
index 6ae75cd3..00000000
--- a/modules/fediverse_server/security.py
+++ /dev/null
@@ -1,32 +0,0 @@
-import base64, typing
-from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.primitives.asymmetric import padding, rsa
-from cryptography.hazmat.backends import default_backend
-
-SIGNATURE_FORMAT = (
-    "keyId=\"%s\",headers=\"%s\",signature=\"%s\",algorithm=\"rsa-sha256\"")
-
-
-def private_key(key_filename: str) -> rsa.RSAPrivateKey:
-    with open(key_filename) as key_file:
-        return serialization.load_pem_private_key(
-            key_file.read(), password=None, backend=default_backend())
-
-def signature(key: rsa.RSAPrivateKey, key_id: str,
-        headers: typing.List[typing.Tuple[str, str]]) -> str:
-    private_key = _private_key(key_filename)
-    sign_header_keys = " ".join(h[0] for h in headers)
-
-    sign_string_parts = ["%s: %s" % (k, v) for k, v in headers]
-    sign_string = "\n".join(sign_string_parts)
-
-    signature = private_key.sign(
-        sign_string.encode("utf8"),
-        padding.PSS(
-            mgf=padding.MGF1(hashes.SHA256()),
-            salt_length=padding.PSS.MAX_LENGTH),
-        hashes.SHA256()
-    )
-
-    signature = base64.b64encode(signature).decode("ascii")
-    return SIGNATURE_FORMAT % (key_id, sign_header_keys, signature)