1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
#--require-config tls-certificate
import urllib.parse
from src import ModuleManager, utils
ACTIVITY_TYPE = "application/activity+json"
ACTIVITY_TYPE = ("application/ld+json; "
"profile=\"https://www.w3.org/ns/activitystreams\"")
WEBFINGER_TYPE = "application/jrd+json"
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("set", utils.FunctionSetting(_setting_parse, "fediverse",
help="Set your fediverse account", example="@gargron@mastodon.social"))
class Module(ModuleManager.BaseModule):
_name = "Fedi"
def _ap_self(self):
our_username = self.bot.get_setting("fediverse", None)
return _parse_username(our_username)
def _ap_self_id(self, url_for, our_username):
return "https://%s" % url_for("api", "ap-user", {"u": 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_id(event["url_for"], our_username)
event["response"].content_type = WEBFINGER_TYPE
event["response"].write_json({
"aliases": [self_id],
"links": [{
"href": self_id,
"rel": "self",
"type": ACTIVITY_TYPE
}],
"subject": resource
})
@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_id(event["url_for"], our_username)
inbox = event["url_for"]("api", "ap-inbox", {"u": 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 = ACTIVITY_TYPE
event["response"].write_json({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": self_id,
"type": "Person",
"preferredUsername": our_username,
"inbox": inbox,
"publicKey": {
"id": "%s#key" % self_id,
"owner": self_id,
"publicKeyPem": cert
}
})
|
