bridgy-fed/pages.py

"""UI pages."""
from functools import wraps
import html
import itertools
import logging
import re
import time

from flask import render_template, request
from google.cloud.ndb import tasklets
from google.cloud.ndb.key import Key
from google.cloud.ndb.query import OR
from google.cloud.ndb.model import get_multi, Model
from granary import as1, as2, atom, microformats2, rss
import oauth_dropins
from oauth_dropins.webutil import flask_util, logs, util
from oauth_dropins.webutil.flask_util import (
    canonicalize_request_domain,
    error,
    flash,
    get_flashed_messages,
    get_required_param,
    Found,
    MovedPermanently,
)
from oauth_dropins.webutil.util import json_loads, json_dumps
import requests
import werkzeug.exceptions
from werkzeug.exceptions import NotFound

import activitypub
from activitypub import ActivityPub
import atproto
from atproto import ATProto, BlueskyOAuthStart
import common
from common import CACHE_CONTROL, DOMAIN_RE, ErrorButDoNotRetryTask, PROTOCOL_DOMAINS
from flask_app import app
from flask import redirect, session
import ids
import memcache
import models
from models import (
    fetch_objects,
    fetch_page,
    Follower,
    Object,
    PAGE_SIZE,
    PROTOCOLS,
    USER_STATUS_DESCRIPTIONS,
)
from nostr import Nostr
from protocol import Protocol
from web import Web
import webfinger

# precompute this because we get a ton of requests for non-existing users
# from weird open redirect referrers:
# https://github.com/snarfed/bridgy-fed/issues/422
with app.test_request_context('/'):
    USER_NOT_FOUND_HTML = render_template('user_not_found.html')

logger = logging.getLogger(__name__)

BLOG_REDIRECT_DOMAINS = (
    'blog.anew.social',
    'snarfed.org',
)

TEMPLATE_VARS = {
    'ActivityPub': ActivityPub,
    'as1': as1,
    'as2': as2,
    'ATProto': ATProto,
    'ids': ids,
    'isinstance': isinstance,
    'logs': logs,
    'Nostr': Nostr,
    'PROTOCOLS': PROTOCOLS,
    'set': set,
    'util': util,
    'Web': Web,
}


def load_user(protocol, id):
    """Loads and returns the current request's user.

    Args:
      protocol (str):
      id (str):

    Returns:
      models.User:

    Raises:
      :class:`werkzeug.exceptions.HTTPException` on error or redirect
    """
    assert id

    if id in PROTOCOL_DOMAINS:
        error(f'{protocol} user {id} not found', status=404)

    cls = PROTOCOLS[protocol]

    if cls.ABBREV == 'ap' and not id.startswith('@'):
        id = '@' + id
    elif cls.ABBREV == 'bsky':
        id = id.removeprefix('@')

    filters = [cls.key == cls(id=id).key]
    if cls.ABBREV != 'web':
        # also query by handle, except for web. Web.handle is custom username, which
        # isn't unique
        filters.append(cls.handle == id)

    redirect_user = None
    for user in cls.query(OR(*filters)):
        if user.use_instead:
            if not (user := user.use_instead.get()):
                continue

        if id not in (user.key.id(), user.handle):
            # keep looking for an exact match. if we don't find one, we'll redirect
            # to this one later
            redirect_user = user
            continue
        elif not user.status and (user.enabled_protocols
                                  or user.DEFAULT_SERVE_USER_PAGES):
            assert not user.use_instead
            return user

    if redirect_user:
        error('', status=302, location=user.user_page_path())

    # TODO: switch back to USER_NOT_FOUND_HTML
    # not easy via exception/abort because this uses Werkzeug's built in
    # NotFound exception subclass, and we'd need to make it implement
    # get_body to return arbitrary HTML.
    error(f'{protocol} user {id} not found', status=404)


def require_login(fn):
    """Decorator that requires and loads the current request's logged in user.

    Passes the userin the ``user`` kwarg, as a :class:`models.User`.

    HTTP POST params:
      key (str): url-safe ndb key

    Raises:
      :class:`werkzeug.exceptions.HTTPException` on error or redirect
    """
    @wraps(fn)
    def wrapper(*args, **kwargs):
        key = Key(urlsafe=get_required_param('key'))
        if key not in [login_to_user_key(l) for l in get_logins()]:
            logger.warning(f'not logged in for {key}')
            raise Found(location='/login')
        elif not (user := key.get()):
            raise Found(location='/login')

        return fn(*args, user=user, **kwargs)

    return wrapper


def get_logins():
    """Returns the user's current logged in sessions:

    Returns:
      list of :class:`oauth_dropins.models.BaseAuth`
    """
    logins = [l for l in get_multi(oauth_dropins.get_logins()) if l]
    return sorted(logins, key=lambda l: (l.key.kind(), l.user_display_name()))


def login_to_user_key(login):
    """"Converts an oauth-dropins auth entity to a :model:`User` key.

    Args:
      login (oauth_dropins.models.BaseAuth)

    Returns:
      ndb.key.Key:
    """
    match login.site_name():
        case 'Bluesky':
            return ATProto(id=login.key.id()).key
        case 'Mastodon':
            if login.user_json and (id := json_loads(login.user_json).get('uri')):
                return ActivityPub(id=id).key
            logger.warning(f'Mastodon auth entity {login.key.id()} has no user_json or uri')
            return None
        case 'Pixelfed':
            user, server = login.key.id().strip('@').split('@')
            return ActivityPub(id=f'https://{server}/users/{user}').key
        case 'Threads':
            username = json_loads(login.user_json).get('username')
            handle = f'@{username}@threads.net'
            if user := ActivityPub.query(ActivityPub.handle == handle).get():
                return user.key
            if not (actor_id := webfinger.fetch_actor_url(handle)):
                for msg in get_flashed_messages:
                    if 'HTTP 404' in msg:
                        flash('You need to <a href="https://help.instagram.com/169559812696339">turn on fediverse sharing</a> first.')
                return None
            return ActivityPub(id=actor_id).key
        case _:
            assert False, repr(login)


def render(template, **vars):
    """Renders a Jinja2 template and adds our standard template variables.

    Args:
      template (str): file name
    """
    return render_template(template, **TEMPLATE_VARS, logins=get_logins(), **vars)


@app.route('/')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
@flask_util.headers(CACHE_CONTROL)
def front_page():
    """View for the front page."""
    return render('index.html')


@app.route('/docs')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
@flask_util.headers(CACHE_CONTROL)
def docs():
    """View for the docs page."""
    return render('docs.html')


@app.route('/login')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
@flask_util.headers(CACHE_CONTROL)
def login():
    """View for the front page."""
    return render('login.html')


@app.post('/logout')
def logout():
    """Logs the user out of all current login sessions."""
    oauth_dropins.logout()
    flash(f"OK, you're now logged out.")
    return redirect('/', code=302)


@app.route('/settings')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def settings():
    """User settings page. Requires logged in session."""
    auth_entity = request.args.get('auth_entity')
    logged_in_as = Key(urlsafe=auth_entity) if auth_entity else None

    def site_logo(login):
        return f'/oauth_dropins_static/{login.site_name().lower()}_icon.png'

    users = []
    logins_and_user_keys = []
    for login in get_logins():
        if user_key := login_to_user_key(login):
            if login.key == logged_in_as:
                cls = Model._lookup_model(user_key.kind())
                user = cls.get_or_create(id=user_key.id(), allow_opt_out=True,
                                         reload=True)
                user.logo = site_logo(login)
                users.append(user)
            else:
                logins_and_user_keys.append((login, user_key))

    loaded = get_multi(key for _, key in logins_and_user_keys)
    for (login, _), user in zip(logins_and_user_keys, loaded):
        if user:
            user.logo = site_logo(login)
            users.append(user)

    if not users:
        return redirect('/login', code=302)

    return render(
        'settings.html',
        **locals(),
        USER_STATUS_DESCRIPTIONS=USER_STATUS_DESCRIPTIONS,
    )


@app.post('/settings/enable')
@require_login
def enable(user=None):
    """Enables bridging for a given account.

    Args:
      user (models.User)
    """
    enabled = []

    for proto in set(PROTOCOLS.values()):
        if (proto and not isinstance(user, proto)
                and proto.LABEL not in ('ui', 'web')
                and not user.is_enabled(proto)):
            try:
                user.enable_protocol(proto)
            except ErrorButDoNotRetryTask as e:
                msg = str(e)
                if resp := e.get_response():
                    if resp.is_json:
                        msg = resp.json['error']
                flash(f"Couldn't enable bridging to {proto.PHRASE}: {msg}")
                return redirect('/settings', code=302)

            proto.bot_follow(user)
            enabled.append(proto)

    if enabled:
        flash(f'Now bridging {user.handle_or_id()} to {",".join(p.PHRASE for p in enabled)}.')
    else:
        flash(f'{user.handle_or_id()} is already bridging.')

    return redirect('/settings', code=302)


@app.post('/settings/disable')
@require_login
def disable(user=None):
    """Disables bridging for a given account.

    Args:
      user (models.User)
    """
    if not user.enabled_protocols:
        flash(f'{user.handle_or_id()} is not currently bridging.')
        return redirect('/settings', code=302)

    enabled = list(user.enabled_protocols)
    for proto in user.enabled_protocols:
        user.delete(PROTOCOLS[proto])
        user.disable_protocol(PROTOCOLS[proto])

    flash(f'Disabled bridging {user.handle_or_id()} to {",".join(PROTOCOLS[p].PHRASE for p in enabled)}.')
    return redirect('/settings', code=302)


@app.post('/settings/set-username')
@require_login
def set_username(user=None):
    """Enables bridging for a given account.

    Args:
      user (models.User)

    Query params:
      protocol (str)
      username (str)
    """
    proto = PROTOCOLS[flask_util.get_required_param('protocol')]
    username = flask_util.get_required_param('username')

    try:
        proto.set_username(user, username)
        flash(f"Setting username on {proto.PHRASE} to {username}...")
    except NotImplementedError:
        flash(f"Custom usernames aren't supported on {proto.PHRASE}.")
    except (ValueError, RuntimeError) as e:
        flash(f"Couldn't set username on {proto.PHRASE} to {username}: {e}")

    return redirect('/settings', code=302)


@app.get(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>')
# WARNING: this overrides the /ap/... actor URL route in activitypub.py, *only*
# for handles with leading @ character. be careful when changing this route!
@app.get(f'/ap/@<id>', defaults={'protocol': 'ap'})
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def profile(protocol, id):
    user = load_user(protocol, id)
    query = Object.query(Object.users == user.key)
    objects, before, after = fetch_objects(query, by=Object.updated, user=user)
    num_followers, num_following = user.count_followers()
    return render('profile.html', **locals())


@app.get(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>/home')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def home(protocol, id):
    user = load_user(protocol, id)
    query = Object.query(Object.feed == user.key)
    objects, before, after = fetch_objects(query, by=Object.created, user=user)

    # this calls Object.actor_link serially for each object, which loads the
    # actor from the datastore if necessary. TODO: parallelize those fetches
    return render('home.html', **locals())


@app.get(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>/notifications')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def notifications(protocol, id):
    user = load_user(protocol, id)

    query = Object.query(Object.notify == user.key)
    objects, before, after = fetch_objects(query, by=Object.updated, user=user)

    format = request.args.get('format')
    if format:
        return serve_feed(objects=objects, format=format, as_snippets=True,
                          user=user, title=f'Bridgy Fed notifications for {id}',
                          quiet=request.args.get('quiet'))

    # notifications tab UI page
    return render('notifications.html', **locals())


@app.get(f'/user-page')
@flask_util.headers(CACHE_CONTROL)
def find_user_page_form():
    return render('find_user_page.html')


@app.post(f'/user-page')
def find_user_page():
    id = request.form['id']

    proto = Protocol.for_id(id)

    resolved_id = None
    if not proto:
        proto, resolved_id = Protocol.for_handle(id)
        if not proto:
            flash(f"Couldn't determine network for {html.escape(id)}.")
            return render('find_user_page.html'), 404

    try:
        user = load_user(proto.LABEL, resolved_id or id)
    except NotFound:
        flash(f"User {html.escape(id)} on {proto.PHRASE} isn't signed up.")
        return render('find_user_page.html'), 404

    return redirect(user.user_page_path(), code=302)


@app.post(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>/update-profile')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def update_profile(protocol, id):
    user = load_user(protocol, id)
    link = f'<a href="{user.web_url()}">{user.handle_or_id()}</a>'

    try:
        user.reload_profile()
    except (requests.RequestException, werkzeug.exceptions.HTTPException) as e:
        _, msg = util.interpret_http_exception(e)
        flash(f"Couldn't update profile for {link}: {msg}")
        return redirect(user.user_page_path(), code=302)

    if not user.obj:
        flash(f"Couldn't update profile for {link}")
        return redirect(user.user_page_path(), code=302)

    common.create_task(queue='receive', obj_id=user.obj_key.id(),
                       authed_as=user.key.id())
    flash(f'Updating profile from {link}...')

    if user.LABEL == 'web':
        if user.status:
            logger.info(f'Disabling web user {user.key.id()}')
            user.delete()
        else:
            for label in list(user.DEFAULT_ENABLED_PROTOCOLS) + user.enabled_protocols:
                try:
                    PROTOCOLS[label].set_username(user, id)
                except (AssertionError, ValueError, RuntimeError, NotImplementedError):
                    pass

    return redirect(user.user_page_path(), code=302)


@app.get(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>/<any(followers,following):collection>')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
def followers_or_following(protocol, id, collection):
    user = load_user(protocol, id)
    id = user.key.id()
    handle = user.handle

    followers, before, after = Follower.fetch_page(collection, user)
    num_followers, num_following = user.count_followers()
    return render(
        f'{collection}.html',
        address=request.args.get('address'),
        follow_url=request.values.get('url'),
        **locals(),
    )


@app.get(f'/<any({",".join(PROTOCOLS)}):protocol>/<id>/feed')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
@flask_util.headers(CACHE_CONTROL)
def feed(protocol, id):
    user = load_user(protocol, id)
    query = Object.query(Object.feed == user.key)
    objects, _, _ = fetch_objects(query, by=Object.created, user=user)
    return serve_feed(objects=objects, format=request.args.get('format', 'html'),
                      user=user, title=f'Bridgy Fed feed for {id}')


def serve_feed(*, objects, format, user, title, as_snippets=False, quiet=False):
    """Generates a feed based on :class:`Object` s.

    Args:
      objects (sequence of models.Object)
      format (str): ``html``, ``atom``, or ``rss``
      user (models.User)
      title (str)
      as_snippets (bool): if True, render short snippets for objects instead of
        full contents
      quiet (bool): if True, exclude follows, unfollows, likes, and reposts

    Returns:
      str or (str, dict) tuple: Flask response
    """
    if format not in ('html', 'atom', 'rss'):
        error(f'format {format} not supported; expected html, atom, or rss')

    objects = [obj for obj in objects if not obj.deleted]
    if quiet:
        objects = [obj for obj in objects if obj.type not in
                   ('delete', 'follow', 'stop-following', 'like', 'share',
                    'undo', 'update')]

    if as_snippets:
        activities = [{
            'objectType': 'note',
            'id': obj.key.id(),
            'content': f'{obj.actor_link(image=False, user=user)} {obj.phrase} {obj.content}',
            'updated': obj.updated.isoformat(),
            'url': as1.get_url(obj.as1) or as1.get_url(as1.get_object(obj.as1)),
        } for obj in objects]
    else:
        activities = [obj.as1 for obj in objects]

    # hydrate authors, actors, objects from stored Objects
    futures = []
    for a in activities:
        futures.extend(models.hydrate(a))
    tasklets.wait_all(futures)

    actor = (user.obj.as1 if user.obj and user.obj.as1
             else {'displayName': user.handle, 'url': user.web_url()})

    # TODO: inject/merge common.pretty_link into microformats2.render_content
    # (specifically into hcard_to_html) somehow to convert Mastodon URLs to @-@
    # syntax. maybe a fediverse kwarg down through the call chain?
    if format == 'html':
        entries = [microformats2.object_to_html(a) for a in activities]
        return render('feed.html', **locals())

    elif format == 'atom':
        body = atom.activities_to_atom(activities, actor=actor, title=title,
                                       request_url=request.url)
        return body, {'Content-Type': atom.CONTENT_TYPE}

    elif format == 'rss':
        # RSS requires email to generate an author element, so fill in blank one
        # where necessary
        for a in activities:
            for field in ('actor', 'author', 'object'):
                if val := as1.get_object(a, field):
                    if as1.object_type(val) in as1.ACTOR_TYPES:
                        val.setdefault('email', '_@_._')

        body = rss.from_activities(activities, actor=actor, title=title,
                                   feed_url=request.url)
        return body, {'Content-Type': rss.CONTENT_TYPE}


@app.get('/log')
@canonicalize_request_domain(common.PROTOCOL_DOMAINS, common.PRIMARY_DOMAIN)
@flask_util.headers(CACHE_CONTROL)
def log():
    return logs.log()


@app.get(f'/internal/<any({",".join(BLOG_REDIRECT_DOMAINS)}):host>/<path:path>')
@flask_util.headers(CACHE_CONTROL)
def blog_redirect(host, path):
    return MovedPermanently(location=f'https://{host}/{path}')


@app.post('/admin/memcache-evict')
def memcache_evict():
    if request.headers.get('Authorization') != app.config['SECRET_KEY']:
        return '', 401

    key = Key(urlsafe=flask_util.get_required_param('key'))
    memcache.evict(key)

    return ''