forked from snarfed/bridgy-fed
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathredirect.py
144 lines (122 loc) · 4.87 KB
/
redirect.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
"""Simple conneg endpoint that serves AS2 or redirects to to the original post.
Only for :class:`web.Web` users. Other protocols (including :class:`web.Web`
sometimes) use ``/convert/`` in convert.py instead.
Serves ``/r/https://foo.com/bar`` URL paths, where ``https://foo.com/bar`` is a
original post for a :class:`Web` user. Needed for Mastodon interop, they require
that AS2 object ids and urls are on the same domain that serves them.
Background:
* https://github.com/snarfed/bridgy-fed/issues/16#issuecomment-424799599
* https://github.com/tootsuite/mastodon/pull/6219#issuecomment-429142747
The conneg makes these ``/r/`` URLs searchable in Mastodon:
https://github.com/snarfed/bridgy-fed/issues/352
"""
import logging
import re
import urllib.parse
from flask import redirect, request
from granary import as2
from negotiator import ContentNegotiator, AcceptParameters, ContentType
from oauth_dropins.webutil import flask_util, util
from oauth_dropins.webutil.flask_util import error
from oauth_dropins.webutil.util import json_dumps, json_loads
from activitypub import ActivityPub
from common import CACHE_CONTROL, CONTENT_TYPE_HTML
from flask_app import app
from protocol import Protocol
from web import Web
logger = logging.getLogger(__name__)
_negotiator = ContentNegotiator(acceptable=[
AcceptParameters(ContentType(CONTENT_TYPE_HTML)),
AcceptParameters(ContentType(as2.CONTENT_TYPE)),
AcceptParameters(ContentType(as2.CONTENT_TYPE_LD)),
])
DOMAIN_ALLOWLIST = frozenset((
'bsky.app',
))
HEADERS = {
'Vary': 'Accept',
**CACHE_CONTROL,
}
@app.get(r'/r/<path:to>')
@flask_util.headers(HEADERS)
def redir(to):
"""Either redirect to a given URL or convert it to another format.
E.g. redirects ``/r/https://foo.com/bar?baz`` to
``https://foo.com/bar?baz``, or if it's requested with AS2 conneg in the
``Accept`` header, fetches and converts and serves it as AS2.
"""
to = to.strip()
if request.args:
to += '?' + urllib.parse.urlencode(request.args)
# some browsers collapse repeated /s in the path down to a single slash.
# if that happened to this URL, expand it back to two /s.
to = re.sub(r'^(https?:/)([^/])', r'\1/\2', to)
if not util.is_web(to):
error(f'Expected fully qualified URL; got {to}')
try:
to_domain = urllib.parse.urlparse(to).hostname
except ValueError as e:
error(f'Invalid URL {to} : {e}')
# check conneg
accept_as2 = False
accept = request.headers.get('Accept')
if accept:
try:
negotiated = _negotiator.negotiate(accept)
except ValueError:
# work around https://github.com/CottageLabs/negotiator/issues/6
negotiated = None
if negotiated:
accept_type = str(negotiated.content_type)
if accept_type in (as2.CONTENT_TYPE, as2.CONTENT_TYPE_LD):
accept_as2 = True
# check that we've seen this domain before so we're not an open redirect
domains = set((util.domain_from_link(to, minimize=True),
util.domain_from_link(to, minimize=False),
to_domain))
web_user = None
for domain in domains:
if domain:
if domain in DOMAIN_ALLOWLIST:
break
if web_user := Web.get_by_id(domain):
logger.info(f'Found web user for domain {domain}')
break
else:
if not accept_as2:
return f'No web user found for any of {domains}', 404
if not accept_as2:
# redirect. include rel-alternate link to make posts discoverable by entering
# https://fed.brid.gy/r/[URL] in a fediverse instance's search.
logger.info(f'redirecting to {to}')
return f"""\
<!doctype html>
<html>
<head>
<link href="{request.url}" rel="alternate" type="application/activity+json">
</head>
<title>Redirecting...</title>
<h1>Redirecting...</h1>
<p>You should be redirected automatically to the target URL: <a href="{to}">{to}</a>. If not, click the link.
</html>
""", 301, {'Location': to}
# AS2 requested, fetch and convert and serve
proto = Protocol.for_id(to)
if not proto:
return f"Couldn't determine protocol for {to}", 404
obj = proto.load(to)
if not obj or obj.deleted:
return f'Object not found: {to}', 404
# TODO: do this for other protocols too?
if proto == Web and not web_user:
web_user = Web.get_or_create(util.domain_from_link(to), direct=False, obj=obj)
if not web_user:
return f'Object not found: {to}', 404
ret = ActivityPub.convert(obj, from_user=web_user)
# logger.info(f'Returning: {json_dumps(ret, indent=2)}')
return ret, {
'Content-Type': (as2.CONTENT_TYPE_LD_PROFILE
if accept_type == as2.CONTENT_TYPE_LD
else accept_type),
'Access-Control-Allow-Origin': '*',
}