-
Notifications
You must be signed in to change notification settings - Fork 43
/
webfinger.py
301 lines (245 loc) · 10.2 KB
/
webfinger.py
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
"""Handles requests for WebFinger endpoints.
* https://webfinger.net/
* https://tools.ietf.org/html/rfc7033
"""
from datetime import timedelta
import logging
from urllib.parse import urljoin, urlparse
from flask import render_template, request
from granary import as2
from oauth_dropins.webutil import flask_util, util
from oauth_dropins.webutil.flask_util import error, flash, Found
from oauth_dropins.webutil.util import json_dumps, json_loads
import activitypub
import common
from common import (
CACHE_CONTROL,
LOCAL_DOMAINS,
PRIMARY_DOMAIN,
PROTOCOL_DOMAINS,
SUPERDOMAIN,
)
from flask_app import app
from protocol import Protocol
from web import Web
SUBSCRIBE_LINK_REL = 'http://ostatus.org/schema/1.0/subscribe'
logger = logging.getLogger(__name__)
class Webfinger(flask_util.XrdOrJrd):
"""Serves a user's WebFinger profile.
Supports both JRD and XRD; defaults to JRD.
https://tools.ietf.org/html/rfc7033#section-4
"""
@flask_util.headers(CACHE_CONTROL)
def dispatch_request(self, *args, **kwargs):
return super().dispatch_request(*args, **kwargs)
def template_prefix(self):
return 'webfinger_user'
def template_vars(self):
# logger.debug(f'Headers: {list(request.headers.items())}')
resource = flask_util.get_required_param('resource').strip()
resource = resource.removeprefix(common.host_url())
# handle Bridgy Fed actor URLs, eg https://fed.brid.gy/snarfed.org
host = util.domain_from_link(common.host_url())
if resource in ('', '/', f'acct:{host}', f'acct:@{host}'):
error('Expected other domain, not *.brid.gy')
allow_indirect = False
cls = None
try:
username, server = util.parse_acct_uri(resource)
id = server
cls = Protocol.for_bridgy_subdomain(id, fed='web')
if cls:
id = username
allow_indirect = True
except ValueError:
id = username = server = urlparse(resource).netloc or resource
if id == PRIMARY_DOMAIN or id in PROTOCOL_DOMAINS:
cls = Web
allow_indirect = True
elif not cls:
cls = Protocol.for_request(fed='web')
if not cls:
error(f"Couldn't determine protocol for f{resource}")
# is this a handle?
if cls.owns_id(id) is False:
logger.info(f'{id} is not a {cls.LABEL} id')
handle = id
id = None
if cls.owns_handle(handle) is not False:
logger.info(' ...might be a handle, trying to resolve')
id = cls.handle_to_id(handle)
if not id:
error(f'{resource} is not a valid handle for a {cls.LABEL} user',
status=404)
logger.info(f'Protocol {cls.LABEL}, user id {id}')
# only allow indirect/no-redirects users if this id is "on" a brid.gy
# subdomain, eg [email protected] but not [email protected]
if allow_indirect:
user = cls.get_or_create(id)
else:
user = cls.get_by_id(id)
if user and not user.direct:
error(f"{user.key.id()} hasn't signed up yet", status=404)
if (not user
or not user.is_enabled(activitypub.ActivityPub)
or (cls == Web and username not in (user.key.id(), user.username()))
or (cls == Web and not allow_indirect and not user.has_redirects)):
error(f'No {cls.LABEL} user found for {id}', status=404)
ap_handle = user.handle_as('activitypub')
if not ap_handle:
error(f'{cls.LABEL} user {id} has no handle', status=404)
# backward compatibility for initial Web users whose AP actor ids are on
# fed.brid.gy, not web.brid.gy
subdomain = request.host.split('.')[0]
if (user.LABEL == 'web'
and subdomain not in (LOCAL_DOMAINS + (user.ap_subdomain,))):
url = urljoin(f'https://{user.ap_subdomain}{common.SUPERDOMAIN}/',
request.full_path)
raise Found(location=url)
actor = user.obj.as1 if user.obj and user.obj.as1 else {}
logger.info(f'Generating WebFinger data for {user.key.id()}')
actor_id = user.id_as(activitypub.ActivityPub)
logger.info(f'AS1 actor: {actor_id}')
urls = util.dedupe_urls(util.get_list(actor, 'urls') +
util.get_list(actor, 'url') +
[user.web_url()])
logger.info(f'URLs: {urls}')
canonical_url = urls[0]
# generate webfinger content
data = util.trim_nulls({
'subject': 'acct:' + ap_handle.lstrip('@'),
'aliases': urls,
'links':
[{
'rel': 'http://webfinger.net/rel/profile-page',
'type': 'text/html',
'href': url,
} for url in urls if util.is_web(url)] +
[{
'rel': 'http://webfinger.net/rel/avatar',
'href': url,
} for url in util.get_urls(actor, 'image')] +
[{
'rel': 'canonical_uri',
'type': 'text/html',
'href': canonical_url,
},
# ActivityPub
#
# include two self links, one for each AP content type, since some
# fediverse servers (eg Pleroma, Akkoma) are unnecessarily picky
# about which one they use from XRD vs JRD.
# https://github.com/snarfed/bridgy-fed/issues/995
{
'rel': 'self',
'type': as2.CONTENT_TYPE_LD_PROFILE,
'href': actor_id,
}, {
'rel': 'self',
'type': as2.CONTENT_TYPE,
'href': actor_id,
}, {
# AP reads this and sharedInbox from the AS2 actor, not
# webfinger, so strictly speaking, it's probably not needed here.
'rel': 'inbox',
'type': as2.CONTENT_TYPE_LD_PROFILE,
'href': actor_id + '/inbox',
}, {
# https://www.w3.org/TR/activitypub/#sharedInbox
'rel': 'sharedInbox',
'type': as2.CONTENT_TYPE_LD_PROFILE,
'href': common.subdomain_wrap(cls, '/ap/sharedInbox'),
},
# remote follow
# https://socialhub.activitypub.rocks/t/what-is-the-current-spec-for-remote-follow/2020/11?u=snarfed
# https://github.com/snarfed/bridgy-fed/issues/60#issuecomment-1325589750
{
'rel': 'http://ostatus.org/schema/1.0/subscribe',
# always use fed.brid.gy for UI pages, not protocol subdomain
# TODO: switch to:
# 'template': common.host_url(user.user_page_path('?url={uri}')),
# the problem is that user_page_path() uses handle_or_id, which uses
# custom username instead of domain, which may not be unique
'template': f'https://{common.PRIMARY_DOMAIN}' +
user.user_page_path('?url={uri}'),
}]
})
# logger.info(f'Returning WebFinger data: {json_dumps(data, indent=2)}')
return data
class HostMeta(flask_util.XrdOrJrd):
"""Renders and serves the ``/.well-known/host-meta`` file.
Supports both JRD and XRD; defaults to XRD.
https://tools.ietf.org/html/rfc6415#section-3
"""
DEFAULT_TYPE = flask_util.XrdOrJrd.XRD
def template_prefix(self):
return 'host-meta'
def template_vars(self):
return {'host_uri': common.host_url()}
@app.get('/.well-known/host-meta.xrds')
@flask_util.headers(CACHE_CONTROL)
def host_meta_xrds():
"""Renders and serves the ``/.well-known/host-meta.xrds`` XRDS-Simple file."""
return render_template('host-meta.xrds', host_uri=common.host_url()), {
'Content-Type': 'application/xrds+xml',
}
def fetch(addr):
"""Fetches and returns an address's WebFinger data.
On failure, flashes a message and returns None.
TODO: switch to raising exceptions instead of flashing messages and
returning None
Args:
addr (str): a Webfinger-compatible address, eg ``@x@y``, ``acct:x@y``, or
``https://x/y``
Returns:
dict: fetched WebFinger data, or None on error
"""
addr = addr.strip().strip('@')
split = addr.split('@')
if len(split) == 2:
addr_domain = split[1]
resource = f'acct:{addr}'
elif util.is_web(addr):
addr_domain = util.domain_from_link(addr, minimize=False)
resource = addr
else:
flash('Enter a fediverse address in @[email protected] format')
return None
try:
resp = util.requests_get(
f'https://{addr_domain}/.well-known/webfinger?resource={resource}')
except BaseException as e:
if util.is_connection_failure(e):
flash(f"Couldn't connect to {addr_domain}")
return None
raise
if not resp.ok:
flash(f'WebFinger on {addr_domain} returned HTTP {resp.status_code}')
return None
try:
data = resp.json()
except ValueError as e:
logger.warning(f'Got {e}', exc_info=True)
flash(f'WebFinger on {addr_domain} returned non-JSON')
return None
logger.info(f'Got WebFinger for {addr}')
return data
def fetch_actor_url(addr):
"""Fetches and returns a WebFinger address's ActivityPub actor URL.
On failure, flashes a message and returns None.
Args:
addr (str): a Webfinger-compatible address, eg ``@x@y``, ``acct:x@y``, or
``https://x/y``
Returns:
str: ActivityPub actor URL, or None on error or not fouund
"""
data = fetch(addr)
if not data:
return None
for link in data.get('links', []):
type = link.get('type', '').split(';')[0]
if link.get('rel') == 'self' and type in as2.CONTENT_TYPES:
return link.get('href')
app.add_url_rule('/.well-known/webfinger', view_func=Webfinger.as_view('webfinger'))
app.add_url_rule('/.well-known/host-meta', view_func=HostMeta.as_view('hostmeta'))
app.add_url_rule('/.well-known/host-meta.json', view_func=HostMeta.as_view('hostmeta-json'))