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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
#--require-config tls-certificate
import urllib.parse
from src import ModuleManager, utils
HOSTMETA = "https://%s/.well-known/host-meta"
WEBFINGER_DEFAULT = "https://%s/.well-known/webfinger?resource={uri}"
WEBFINGER_HEADERS = {"Accept": "application/jrd+json"}
ACTIVITY_TYPE = "application/activity+json"
ACTIVITY_HEADERS = {"Accept": ("application/ld+json; "
'profile="https://www.w3.org/ns/activitystreams"')}
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"
@utils.hook("received.command.fediverse")
@utils.hook("received.command.fedi", alias_of="fediverse")
@utils.kwarg("help", "Get someone's latest toot")
@utils.kwarg("usage", "@<user>@<instance>")
def fedi(self, event):
account = None
if not event["args"]:
account = event["user"].get_setting("fediverse", None)
elif not "@" in event["args"]:
target = event["args_split"][0]
if event["server"].has_user_id(target):
target_user = event["server"].get_user(target)
account = target_user.get_setting("fediverse", None)
else:
account = event["args_split"][0]
username = None
instance = None
if account:
username, instance = _parse_username(account)
if not username or not instance:
raise utils.EventError("Please provide @<user>@<instance>")
hostmeta = utils.http.request(HOSTMETA % instance,
soup=True, check_content_type=False)
webfinger_url = None
for item in hostmeta.data.find_all("link"):
if item["rel"] and item["rel"][0] == "lrdd":
webfinger_url = item["template"]
break
if webfinger_url == None:
self.log.debug("host-meta lookup failed for %s" % instance)
webfinger_url = WEBFINGER_DEFAULT % instance
webfinger_url = webfinger_url.replace("{uri}",
"acct:%s@%s" % (username, instance))
webfinger = utils.http.request(webfinger_url,
headers=WEBFINGER_HEADERS, json=True)
activity_url = None
for link in webfinger.data["links"]:
if link["type"] == ACTIVITY_TYPE:
activity_url = link["href"]
break
if not activity_url:
raise utils.EventError("Failed to find user activity feed")
activity = utils.http.request(activity_url,
headers=ACTIVITY_HEADERS, json=True)
preferred_username = activity.data["preferredUsername"]
outbox_url = activity.data["outbox"]
outbox = utils.http.request(outbox_url, headers=ACTIVITY_HEADERS,
json=True)
items = None
if "first" in outbox.data:
if type(outbox.data["first"]) == dict:
# pleroma
items = outbox.data["first"]["orderedItems"]
else:
# mastodon
first = utils.http.request(outbox.data["first"],
headers=ACTIVITY_HEADERS, json=True)
items = first.data["orderedItems"]
else:
items = outbox.data["orderedItems"]
if not items:
raise utils.EventError("No toots found")
first_item = items[0]
if first_item["type"] == "Announce":
retoot_url = first_item["object"]
retoot_instance = urllib.parse.urlparse(retoot_url).hostname
retoot = utils.http.request(retoot_url,
headers=ACTIVITY_HEADERS, json=True)
original_tooter_url = retoot.data["attributedTo"]
original_tooter = utils.http.request(original_tooter_url,
headers=ACTIVITY_HEADERS, json=True)
retooted_user = "@%s@%s" % (
original_tooter.data["preferredUsername"],
retoot_instance)
shorturl = self.exports.get_one("shorturl")(
event["server"], retoot_url)
retoot_content = utils.http.strip_html(
retoot.data["content"])
event["stdout"].write("%s (boost %s): %s - %s" % (
preferred_username, retooted_user, retoot_content,
shorturl))
elif first_item["type"] == "Create":
content = utils.http.strip_html(
first_item["object"]["content"])
url = first_item["object"]["id"]
shorturl = self.exports.get_one("shorturl")(
event["server"], url)
event["stdout"].write("%s: %s - %s" % (preferred_username,
content, shorturl))
|
