bridgy-fed/redirect.py at main · snarfed/bridgy-fed · GitHub

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
"""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
"""
from datetime import timedelta
import logging
import re
import urllib.parse

from flask import redirect, request
from granary import as1
from webutil import flask_util, util
from webutil.flask_util import error
from webutil.util import json_dumps, json_loads

from activitypub import ActivityPub, as2_request_type
from common import (
    CACHE_CONTROL_VARY_ACCEPT,
    CONTENT_TYPE_HTML,
    SUPERDOMAIN,
)
from flask_app import app
import memcache
from protocol import Protocol
from web import Web

logger = logging.getLogger(__name__)

DOMAIN_ALLOWLIST = frozenset((
    'bsky.app',
    'farcaster.xyz',
))


@app.get(r'/r/<path:to>')
@memcache.memoize(expire=timedelta(hours=1),
                  key=lambda **kwargs: (kwargs, request.args, as2_request_type()))
@flask_util.headers(CACHE_CONTROL_VARY_ACCEPT)
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}')

    if to_domain and to_domain.endswith(SUPERDOMAIN):
        return redirect(to, code=301)

    # check conneg
    as2_request = as2_request_type()

    # 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.debug(f'Found web user for domain {domain}')
                break
    else:
        if not as2_request:
            error(f'No web user found for any of {domains}', status=404)

    if not as2_request:
        # redirect. include rel-alternate link to make posts discoverable by entering
        # https://fed.brid.gy/r/[URL] in a fediverse instance's search.
        logger.debug(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>
    </html>
    """, 301, {'Location': to}

    # AS2 requested, fetch and convert and serve
    proto = Protocol.for_id(to)
    if not proto:
        error(f"Couldn't determine protocol for {to}", status=404)

    obj = proto.load(to)
    if not obj or obj.deleted:
        error(f'Object not found: {to}', status=404)

    if proto == Web:
        if not web_user:
            error(f'Object not found: {to}', status=404)
    else:
        if obj.type in as1.ACTOR_TYPES:
            user = proto.query(proto.obj_key == obj.key).get()
            if not user or not user.is_enabled(ActivityPub):
                error(f'Object not found: {to}', status=404)

    if not (ret := ActivityPub.convert(obj, from_user=web_user)):
        error(f'Object not found: {to}', status=404)

    # logger.info(f'Returning: {json_dumps(ret, indent=2)}')
    return ret, {
        'Content-Type': as2_request,
        'Access-Control-Allow-Origin': '*',
    }