878ced7f-3b32-4aaf-9280-63d2666620db

Committed 17 Oct 2025 03:56AM UTC coverage: 92.855% (-0.002%) from 92.857%

Build # 878ced7f-3b32-4aaf-9280-63d2666620db

Build Type

push

circleci

Committed by

snarfed

Commit Message

fix integration test test_nostr_follow_activitypub_bot_user_invalid_nip05

Run Details

5939 of 6396 relevant lines covered (92.85%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.21

/dms.py

"""Protocol-independent code for sending and receiving DMs aka chat messages."""
from datetime import timedelta
import logging

from granary import as1, source
from oauth_dropins.webutil import util

from collections import namedtuple
import common
from common import create_task, DOMAINS
import ids
import memcache
import models
from models import Object, PROTOCOLS, User
import protocol

logger = logging.getLogger(__name__)

REQUESTS_LIMIT_EXPIRE = timedelta(days=1)
REQUESTS_LIMIT_USER = 10

# populated by the command() decorator
_commands = {}


def command(names, arg=False, user_bridged=None, handle_bridged=None):
    """Function decorator. Defines and registers a DM command.

    Args:
      names (sequence of str): the command strings that trigger this command, or
        ``None`` if this command has no command string
      arg: whether this command takes an argument. ``False`` for no, ``True``
        for yes, anything, ``'handle_or_id'`` for yes, a handle or ID in the bot
        account's protocol for a user that must not already be bridged.
      user_bridged (bool): whether the user sending the DM should be
        bridged. ``True`` for yes, ``False`` for no, ``None` for either.
      handle_bridged (bool): whether the handle arg should be bridged. ``True``
        for yes, ``False`` for no, ``None` for either, ``'eligible'`` for
        not bridged but eligible.

    The decorated function should have the signature:
      (from_user, to_proto, arg=None, to_user=None) => (str, None)

    If it returns a string, that text is sent to the user as a reply to their DM.

    Args for the decorated function:
      from_user (models.User): the user who sent the DM
      to_proto (protocol.Protocol): the protocol bot account they sent it to
      arg (str or None): the argument to the command, if any
      to_user (models.User or None): the user for the argument, if it's a handle

    The decorated function returns:
      str: text to reply to the user in a DM, if any
    """
    assert arg in (False, True, 'handle_or_id'), arg
    if handle_bridged is not None:
        assert arg == 'handle_or_id', arg

    def decorator(fn):
        def wrapped(from_user, to_proto, cmd, cmd_arg, dm_as1):
            def reply(text, type=None):
                maybe_send(from_=to_proto, to_user=from_user, text=text,
                           type=type, in_reply_to=dm_as1.get('id'))
                return 'OK', 200

            if arg and not cmd_arg:
                return reply(f'{cmd} command needs an argument<br><br>{help_text(from_user, to_proto)}')

            to_user = None

            if arg == 'handle_or_id':
                if to_proto.owns_id(cmd_arg) is not False:
                    if not (to_user := to_proto.get_or_create(cmd_arg, allow_opt_out=True)):
                        # Skip trying as a handle, assuming that a valid ID
                        # never happens to be a valid handle of another user
                        # at the same time in any supported protocol.
                        return reply(f"Couldn't find user {cmd_arg} on {to_proto.PHRASE}")
                else:
                    logging.info(f"doesn't look like an ID, trying as a handle")

                    if not to_proto.owns_handle(cmd_arg) and cmd_arg.startswith('@'):
                        logging.info(f"doesn't look like a handle, trying without leading @")
                        cmd_arg = cmd_arg.removeprefix('@')

                    if not (to_user := load_user_by_handle(to_proto, cmd_arg)):
                        return reply(f"Couldn't find user {cmd_arg} on {to_proto.PHRASE}")

            if to_user:
                from_proto = from_user.__class__
                enabled = to_user.is_enabled(from_proto)
                if handle_bridged is True and not enabled:
                    return reply(f'{to_user.user_link(proto=from_proto)} is not bridged into {from_proto.PHRASE}.')
                elif handle_bridged in (False, 'eligible') and enabled:
                    return reply(f'{to_user.user_link(proto=from_proto)} is already bridged into {from_proto.PHRASE}.')
                elif handle_bridged == 'eligible' and to_user.status:
                    to_user.reload_profile()
                    if to_user.status:
                        because = ''
                        if desc := models.USER_STATUS_DESCRIPTIONS.get(to_user.status):
                            because = f' because their {desc}'
                        return reply(f"{to_user.user_link()} on {to_proto.PHRASE} isn't eligible for bridging into {from_proto.PHRASE}{because}.")

            from_user_enabled = from_user.is_enabled(to_proto)
            if user_bridged is True and not from_user_enabled:
                return reply(f"Looks like you're not bridged to {to_proto.PHRASE} yet! Please bridge your account first by following this account.")
            elif user_bridged is False and from_user_enabled:
                return reply(f"Looks like you're already bridged to {to_proto.PHRASE}!")
            # dispatch!
            kwargs = {}
            if arg and cmd_arg:
                kwargs['arg'] = cmd_arg
            if arg == 'handle_or_id':
                kwargs['to_user'] = to_user
            reply_text = fn(from_user, to_proto, **kwargs)
            if reply_text:
                reply(reply_text)

            return 'OK', 200

        if names is None:
            assert None not in _commands
            _commands[None] = wrapped
        else:
            assert isinstance(names, (tuple, list))
            for name in names:
                _commands[name] = wrapped

        return wrapped

    return decorator


def help_text(from_user, to_proto):
    extra = ''
    if to_proto.LABEL == 'atproto':
        extra = """<li><em>did</em>: get your bridged Bluesky account's <a href="https://atproto.com/guides/identity#identifiers">DID</a>"""

    text = f"""\
<p>Hi! I'm a friendly bot that can help you bridge your account into {to_proto.PHRASE}. Here are some commands I respond to:</p>
<ul>
<li><em>start</em>: enable bridging for your account
<li><em>stop</em>: disable bridging for your account
<li><em>notify</em>: enable notifications when someone who's not bridged replies to you, quotes you, or @-mentions you
<li><em>mute</em>: disable notifications
<li><em>username [domain]</em>: set a custom domain username (handle)
<li><em>[handle or ID]</em>: ask me to DM a user on {to_proto.PHRASE} to request that they bridge their account into {from_user.PHRASE}
<li><em>block [handle or ID]</em>: block a user on {to_proto.PHRASE} who's not bridged here
<li><em>unblock [handle or ID]</em>: unblock a user on {to_proto.PHRASE} who's not bridged here
{extra}
<li><em>help</em>: print this message
</ul>"""
# <li><em>migrate-to [handle]</em>: migrate your bridged account on {to_proto.PHRASE} out of Bridgy Fed to a native account

    if from_user.LABEL == 'atproto':
        text = source.html_to_text(text, ignore_emphasis=True)

    return text

@command(['?', 'help', 'commands', 'info', 'hi', 'hello'])
def help(from_user, to_proto):
    return help_text(from_user, to_proto)


@command(['yes', 'ok', 'start'], user_bridged=False)
def start(from_user, to_proto):
    from_user.enable_protocol(to_proto)
    to_proto.bot_maybe_follow_back(from_user)


@command(['no', 'stop'])
def stop(from_user, to_proto, user_bridged=True):
    from_user.delete(to_proto)
    from_user.disable_protocol(to_proto)


@command(['notify'], user_bridged=True)
def notify(from_user, to_proto):
    from_user.send_notifs = 'all'
    from_user.put()
    return f"Notifications enabled! You'll now receive batched notifications via DM when someone on {to_proto.PHRASE} who's not bridged replies to you, quotes you, or @-mentions you. To disable, reply with the text 'mute'."


@command(['mute'], user_bridged=True)
def mute(from_user, to_proto):
    from_user.send_notifs = 'none'
    from_user.put()
    return f"Notifications disabled. You won't receive DM notifications when someone on {to_proto.PHRASE} who's not bridged replies to you, quotes you, or @-mentions you. To re-enable, reply with the text 'notify'."


@command(['did'], user_bridged=True)
def did(from_user, to_proto):
    if to_proto.LABEL == 'atproto':
        return f'Your DID is <code>{from_user.get_copy(PROTOCOLS["atproto"])}</code>'


@command(['username', 'handle'], arg=True, user_bridged=True)
def username(from_user, to_proto, arg):
    try:
        to_proto.set_username(from_user, arg)
    except NotImplementedError:
        return f"Sorry, Bridgy Fed doesn't support custom usernames for {to_proto.PHRASE} yet."
    except (ValueError, RuntimeError) as e:
        return str(e)

    return f"Your username in {to_proto.PHRASE} has been set to {from_user.user_link(proto=to_proto, name=False, handle=True)}. It should appear soon!"


@command(['block'], arg='handle_or_id', user_bridged=True)
def block(from_user, to_proto, arg, to_user):
    id = f'{from_user.key.id()}#bridgy-fed-block-{util.now().isoformat()}'
    obj = Object(id=id, source_protocol=from_user.LABEL, our_as1={
        'objectType': 'activity',
        'verb': 'block',
        'id': id,
        'actor': from_user.key.id(),
        'object': to_user.key.id(),
    })
    obj.put()
    from_user.deliver(obj, from_user=from_user)
    return f"""OK, you're now blocking {to_user.user_link()} on {to_proto.PHRASE}."""


@command(['unblock'], arg='handle_or_id', user_bridged=True)
def unblock(from_user, to_proto, arg, to_user):
    id = f'{from_user.key.id()}#bridgy-fed-unblock-{util.now().isoformat()}'
    obj = Object(id=id, source_protocol=from_user.LABEL, our_as1={
        'objectType': 'activity',
        'verb': 'undo',
        'id': id,
        'actor': from_user.key.id(),
        'object': {
            'objectType': 'activity',
            'verb': 'block',
            'actor': from_user.key.id(),
            'object': to_user.key.id(),
        },
    })
    obj.put()
    from_user.deliver(obj, from_user=from_user)
    return f"""OK, you're not blocking {to_user.user_link()} on {to_proto.PHRASE}."""


@command(['migrate-to'], arg='handle_or_id', user_bridged=True)
def migrate_to(from_user, to_proto, arg, to_user):
    try:
        to_proto.migrate_out(from_user, to_user.key.id())
    except ValueError as e:
        return str(e)

    return f"OK, we'll migrate your bridged account on {to_proto.PHRASE} to {to_user.user_link()}."


@command(None, arg='handle_or_id', user_bridged=True, handle_bridged='eligible')
def prompt(from_user, to_proto, arg, to_user):
    """Prompt a non-bridged user to bridge. No command, just the handle, alone."""
    from_proto = from_user.__class__
    try:
        ids.translate_handle(handle=arg, from_=to_proto, to=from_user)
    except ValueError as e:
        logger.warning(e)
        return f"Sorry, Bridgy Fed doesn't yet support bridging handle {arg} from {to_proto.PHRASE} to {from_proto.PHRASE}."

    if (models.DM(protocol=from_proto.LABEL, type='request_bridging')
          in to_user.sent_dms):
        # already requested
        return f"We've already sent {to_user.user_link()} a DM. Fingers crossed!"

    # check and update rate limits
    attempts_key = f'dm-user-requests-{from_user.LABEL}-{from_user.key.id()}'
    # incr leaves existing expiration as is, doesn't change it
    # https://stackoverflow.com/a/4084043/186123
    attempts = memcache.memcache.incr(attempts_key, 1)
    if not attempts:
        memcache.memcache.add(
            attempts_key, 1,
            expire=int(REQUESTS_LIMIT_EXPIRE.total_seconds()))
    elif attempts > REQUESTS_LIMIT_USER:
        return f"Sorry, you've hit your limit of {REQUESTS_LIMIT_USER} requests per day. Try again tomorrow!"

    # send the DM request!
    maybe_send(from_=from_proto, to_user=to_user, type='request_bridging', text=f"""\
<p>Hi! {from_user.user_link(proto=to_proto, proto_fallback=True)} is using Bridgy Fed to bridge their account from {from_proto.PHRASE} into {to_proto.PHRASE}, and they'd like to follow you. You can bridge your account into {from_proto.PHRASE} by following this account. <a href="https://fed.brid.gy/docs">See the docs</a> for more information.
<p>If you do nothing, your account won't be bridged, and users on {from_proto.PHRASE} won't be able to see or interact with you.
<p>Bridgy Fed will only send you this message once.""")
    return f"Got it! We'll send {to_user.user_link()} a message and say that you hope they'll enable the bridge. Fingers crossed!"


def maybe_send(*, from_, to_user, text, type=None, in_reply_to=None):
    """Sends a DM.

    Creates a task to send the DM asynchronously.

    If ``type`` is provided, and we've already sent this user a DM of this type
    from this protocol, does nothing.

    Args:
      from_ (protocol.Protocol or models.User)
      to_user (models.User)
      text (str): message content. May be HTML.
      type (str): optional, one of DM.TYPES
      in_reply_to (str): optional, ``id`` of a DM to reply to
    """
    if not to_user.SUPPORTS_DMS:
        return

    from_proto = from_
    if not isinstance(from_, User):
        assert issubclass(from_, protocol.Protocol)
        from web import Web
        if not (from_ := Web.get_by_id(from_.bot_user_id())):
            logger.info(f'not sending DM, {from_proto.LABEL} has no bot user')
            return

    if type:
        dm = models.DM(protocol=from_proto.LABEL, type=type)
        if dm in to_user.sent_dms:
            return

    logger.info(f'Sending DM from {from_.key.id()} to {to_user.key.id()} : {text}')

    if not to_user.obj or not to_user.obj.as1:
        logger.info("  can't send DM, recipient has no profile obj")
        return

    now = util.now().isoformat()
    dm_id = f'{from_.profile_id()}#bridgy-fed-dm-{type or "?"}-{to_user.key.id()}-{now}'
    dm_as1 = {
        'objectType': 'note',
        'id': dm_id,
        'author': from_.key.id(),
        'content': text,
        'inReplyTo': in_reply_to,
        'tags': [{
            'objectType': 'mention',
            'url': to_user.key.id(),
        }],
        'published': now,
        'to': [to_user.key.id()],
    }
    Object(id=dm_id, our_as1=dm_as1).put()

    create_id = f'{dm_id}-create'
    create_as1 = {
        'objectType': 'activity',
        'verb': 'post',
        'id': create_id,
        'actor': from_.key.id(),
        'object': dm_as1,
        'published': now,
        'to': [to_user.key.id()],
    }

    target_uri = to_user.target_for(to_user.obj, shared=False)
    target = models.Target(protocol=to_user.LABEL, uri=target_uri)
    create_task(queue='send', id=create_id, our_as1=create_as1, source_protocol='web',
                protocol=to_user.LABEL, url=target.uri, user=from_.key.urlsafe())

    if type:
        to_user.sent_dms.append(dm)
        to_user.put()


def receive(*, from_user, obj):
    """Handles a DM that a user sent to one of our protocol bot users.

    Args:
      from_user (models.User)
      obj (Object): DM

    Returns:
      (str, int) tuple: (response body, HTTP status code) Flask response
    """
    recip = as1.recipient_if_dm(obj.as1)
    assert recip

    to_proto = protocol.Protocol.for_bridgy_subdomain(recip)
    assert to_proto  # already checked in check_supported call in Protocol.receive

    inner_as1 = (as1.get_object(obj.as1) if as1.object_type(obj.as1) == 'post'
                 else obj.as1)
    logger.info(f'got DM from {from_user.key.id()} to {to_proto.LABEL}: {inner_as1.get("content")}')

    # parse message
    text = util.remove_invisible_chars(source.html_to_text(inner_as1.get('content', '')))
    tokens = text.strip().lower().split()
    logger.info(f'  tokens: {tokens}')

    # remove @-mention of bot, if any
    bot_handles = (DOMAINS + ids.BOT_ACTOR_AP_IDS
                   + tuple(h.lstrip('@') for h in ids.BOT_ACTOR_AP_HANDLES))
    if tokens and tokens[0].lstrip('@') in bot_handles:
        logger.debug(f'  first token is bot mention, removing')
        tokens = tokens[1:]

    if not tokens or len(tokens) > 2:
        return r'¯\_(ツ)_/¯', 204

    if fn := _commands.get(tokens[0]):
        return fn(from_user, to_proto, dm_as1=inner_as1,
                  cmd=tokens[0], cmd_arg=tokens[1] if len(tokens) == 2 else None)
    elif len(tokens) == 1:
        fn = _commands.get(None)
        assert fn, tokens[0]
        return fn(from_user, to_proto, dm_as1=inner_as1, cmd=None, cmd_arg=tokens[0])

    return r'¯\_(ツ)_/¯', 204


def load_user_by_handle(proto, handle):
    """
    Args:
      proto (protocol.Protocol)
      handle (str)

    Returns:
      models.User or None
    """
    if proto.owns_handle(handle) is False:
        return None

    if id := proto.handle_to_id(handle):
        if user := proto.get_or_create(id, allow_opt_out=True):
            if user.obj and user.obj.as1:
                return user

1	"""Protocol-independent code for sending and receiving DMs aka chat messages."""
2	from datetime import timedelta	1✔
3	import logging	1✔
4
5	from granary import as1, source	1✔
6	from oauth_dropins.webutil import util	1✔
7
8	from collections import namedtuple	1✔
9	import common	1✔
10	from common import create_task, DOMAINS	1✔
11	import ids	1✔
12	import memcache	1✔
13	import models	1✔
14	from models import Object, PROTOCOLS, User	1✔
15	import protocol	1✔
16
17	logger = logging.getLogger(__name__)	1✔
18
19	REQUESTS_LIMIT_EXPIRE = timedelta(days=1)	1✔
20	REQUESTS_LIMIT_USER = 10	1✔
21
22	# populated by the command() decorator
23	_commands = {}	1✔
24
25
26	def command(names, arg=False, user_bridged=None, handle_bridged=None):	1✔
27	"""Function decorator. Defines and registers a DM command.
28
29	Args:
30	names (sequence of str): the command strings that trigger this command, or
31	``None`` if this command has no command string
32	arg: whether this command takes an argument. ``False`` for no, ``True``
33	for yes, anything, ``'handle_or_id'`` for yes, a handle or ID in the bot
34	account's protocol for a user that must not already be bridged.
35	user_bridged (bool): whether the user sending the DM should be
36	bridged. ``True`` for yes, ``False`` for no, ``None` for either.
37	handle_bridged (bool): whether the handle arg should be bridged. ``True``
38	for yes, ``False`` for no, ``None` for either, ``'eligible'`` for
39	not bridged but eligible.
40
41	The decorated function should have the signature:
42	(from_user, to_proto, arg=None, to_user=None) => (str, None)
43
44	If it returns a string, that text is sent to the user as a reply to their DM.
45
46	Args for the decorated function:
47	from_user (models.User): the user who sent the DM
48	to_proto (protocol.Protocol): the protocol bot account they sent it to
49	arg (str or None): the argument to the command, if any
50	to_user (models.User or None): the user for the argument, if it's a handle
51
52	The decorated function returns:
53	str: text to reply to the user in a DM, if any
54	"""
55	assert arg in (False, True, 'handle_or_id'), arg	1✔
56	if handle_bridged is not None:	1✔
57	assert arg == 'handle_or_id', arg	1✔
58
59	def decorator(fn):	1✔
60	def wrapped(from_user, to_proto, cmd, cmd_arg, dm_as1):	1✔
61	def reply(text, type=None):	1✔
62	maybe_send(from_=to_proto, to_user=from_user, text=text,	1✔
63	type=type, in_reply_to=dm_as1.get('id'))
64	return 'OK', 200	1✔
65
66	if arg and not cmd_arg:	1✔
67	return reply(f'{cmd} command needs an argument<br><br>{help_text(from_user, to_proto)}')	1✔
68
69	to_user = None	1✔
70
71	if arg == 'handle_or_id':	1✔
72	if to_proto.owns_id(cmd_arg) is not False:	1✔
73	if not (to_user := to_proto.get_or_create(cmd_arg, allow_opt_out=True)):	1✔
74	# Skip trying as a handle, assuming that a valid ID
75	# never happens to be a valid handle of another user
76	# at the same time in any supported protocol.
77	return reply(f"Couldn't find user {cmd_arg} on {to_proto.PHRASE}")	×
78	else:
79	logging.info(f"doesn't look like an ID, trying as a handle")	1✔
80
81	if not to_proto.owns_handle(cmd_arg) and cmd_arg.startswith('@'):	1✔
82	logging.info(f"doesn't look like a handle, trying without leading @")	1✔
83	cmd_arg = cmd_arg.removeprefix('@')	1✔
84
85	if not (to_user := load_user_by_handle(to_proto, cmd_arg)):	1✔
86	return reply(f"Couldn't find user {cmd_arg} on {to_proto.PHRASE}")	1✔
87
88	if to_user:	1✔
89	from_proto = from_user.__class__	1✔
90	enabled = to_user.is_enabled(from_proto)	1✔
91	if handle_bridged is True and not enabled:	1✔
92	return reply(f'{to_user.user_link(proto=from_proto)} is not bridged into {from_proto.PHRASE}.')	×
93	elif handle_bridged in (False, 'eligible') and enabled:	1✔
94	return reply(f'{to_user.user_link(proto=from_proto)} is already bridged into {from_proto.PHRASE}.')	1✔
95	elif handle_bridged == 'eligible' and to_user.status:	1✔
96	to_user.reload_profile()	1✔
97	if to_user.status:	1✔
98	because = ''	1✔
99	if desc := models.USER_STATUS_DESCRIPTIONS.get(to_user.status):	1✔
100	because = f' because their {desc}'	1✔
101	return reply(f"{to_user.user_link()} on {to_proto.PHRASE} isn't eligible for bridging into {from_proto.PHRASE}{because}.")	1✔
102
103	from_user_enabled = from_user.is_enabled(to_proto)	1✔
104	if user_bridged is True and not from_user_enabled:	1✔
105	return reply(f"Looks like you're not bridged to {to_proto.PHRASE} yet! Please bridge your account first by following this account.")	1✔
106	elif user_bridged is False and from_user_enabled:	1✔
107	return reply(f"Looks like you're already bridged to {to_proto.PHRASE}!")	1✔
108	# dispatch!
109	kwargs = {}	1✔
110	if arg and cmd_arg:	1✔
111	kwargs['arg'] = cmd_arg	1✔
112	if arg == 'handle_or_id':	1✔
113	kwargs['to_user'] = to_user	1✔
114	reply_text = fn(from_user, to_proto, **kwargs)	1✔
115	if reply_text:	1✔
116	reply(reply_text)	1✔
117
118	return 'OK', 200	1✔
119
120	if names is None:	1✔
121	assert None not in _commands	1✔
122	_commands[None] = wrapped	1✔
123	else:
124	assert isinstance(names, (tuple, list))	1✔
125	for name in names:	1✔
126	_commands[name] = wrapped	1✔
127
128	return wrapped	1✔
129
130	return decorator	1✔
131
132
133	def help_text(from_user, to_proto):	1✔
134	extra = ''	1✔
135	if to_proto.LABEL == 'atproto':	1✔
136	extra = """<li><em>did</em>: get your bridged Bluesky account's <a href="https://atproto.com/guides/identity#identifiers">DID</a>"""	×
137
138	text = f"""\	1✔
139	<p>Hi! I'm a friendly bot that can help you bridge your account into {to_proto.PHRASE}. Here are some commands I respond to:</p>
140	<ul>
141	<li><em>start</em>: enable bridging for your account
142	<li><em>stop</em>: disable bridging for your account
143	<li><em>notify</em>: enable notifications when someone who's not bridged replies to you, quotes you, or @-mentions you
144	<li><em>mute</em>: disable notifications
145	<li><em>username [domain]</em>: set a custom domain username (handle)
146	<li><em>[handle or ID]</em>: ask me to DM a user on {to_proto.PHRASE} to request that they bridge their account into {from_user.PHRASE}
147	<li><em>block [handle or ID]</em>: block a user on {to_proto.PHRASE} who's not bridged here
148	<li><em>unblock [handle or ID]</em>: unblock a user on {to_proto.PHRASE} who's not bridged here
149	{extra}
150	<li><em>help</em>: print this message
151	</ul>"""
152	# <li><em>migrate-to [handle]</em>: migrate your bridged account on {to_proto.PHRASE} out of Bridgy Fed to a native account
153
154	if from_user.LABEL == 'atproto':	1✔
155	text = source.html_to_text(text, ignore_emphasis=True)	1✔
156
157	return text	1✔
158
159	@command(['?', 'help', 'commands', 'info', 'hi', 'hello'])	1✔
160	def help(from_user, to_proto):	1✔
161	return help_text(from_user, to_proto)	1✔
162
163
164	@command(['yes', 'ok', 'start'], user_bridged=False)	1✔
165	def start(from_user, to_proto):	1✔
166	from_user.enable_protocol(to_proto)	1✔
167	to_proto.bot_maybe_follow_back(from_user)	1✔
168
169
170	@command(['no', 'stop'])	1✔
171	def stop(from_user, to_proto, user_bridged=True):	1✔
172	from_user.delete(to_proto)	1✔
173	from_user.disable_protocol(to_proto)	1✔
174
175
176	@command(['notify'], user_bridged=True)	1✔
177	def notify(from_user, to_proto):	1✔
178	from_user.send_notifs = 'all'	1✔
179	from_user.put()	1✔
180	return f"Notifications enabled! You'll now receive batched notifications via DM when someone on {to_proto.PHRASE} who's not bridged replies to you, quotes you, or @-mentions you. To disable, reply with the text 'mute'."	1✔
181
182
183	@command(['mute'], user_bridged=True)	1✔
184	def mute(from_user, to_proto):	1✔
185	from_user.send_notifs = 'none'	1✔
186	from_user.put()	1✔
187	return f"Notifications disabled. You won't receive DM notifications when someone on {to_proto.PHRASE} who's not bridged replies to you, quotes you, or @-mentions you. To re-enable, reply with the text 'notify'."	1✔
188
189
190	@command(['did'], user_bridged=True)	1✔
191	def did(from_user, to_proto):	1✔
192	if to_proto.LABEL == 'atproto':	1✔
193	return f'Your DID is <code>{from_user.get_copy(PROTOCOLS["atproto"])}</code>'	1✔
194
195
196	@command(['username', 'handle'], arg=True, user_bridged=True)	1✔
197	def username(from_user, to_proto, arg):	1✔
198	try:	1✔
199	to_proto.set_username(from_user, arg)	1✔
200	except NotImplementedError:	1✔
201	return f"Sorry, Bridgy Fed doesn't support custom usernames for {to_proto.PHRASE} yet."	1✔
202	except (ValueError, RuntimeError) as e:	1✔
203	return str(e)	1✔
204
205	return f"Your username in {to_proto.PHRASE} has been set to {from_user.user_link(proto=to_proto, name=False, handle=True)}. It should appear soon!"	1✔
206
207
208	@command(['block'], arg='handle_or_id', user_bridged=True)	1✔
209	def block(from_user, to_proto, arg, to_user):	1✔
210	id = f'{from_user.key.id()}#bridgy-fed-block-{util.now().isoformat()}'	1✔
211	obj = Object(id=id, source_protocol=from_user.LABEL, our_as1={	1✔
212	'objectType': 'activity',
213	'verb': 'block',
214	'id': id,
215	'actor': from_user.key.id(),
216	'object': to_user.key.id(),
217	})
218	obj.put()	1✔
219	from_user.deliver(obj, from_user=from_user)	1✔
220	return f"""OK, you're now blocking {to_user.user_link()} on {to_proto.PHRASE}."""	1✔
221
222
223	@command(['unblock'], arg='handle_or_id', user_bridged=True)	1✔
224	def unblock(from_user, to_proto, arg, to_user):	1✔
225	id = f'{from_user.key.id()}#bridgy-fed-unblock-{util.now().isoformat()}'	1✔
226	obj = Object(id=id, source_protocol=from_user.LABEL, our_as1={	1✔
227	'objectType': 'activity',
228	'verb': 'undo',
229	'id': id,
230	'actor': from_user.key.id(),
231	'object': {
232	'objectType': 'activity',
233	'verb': 'block',
234	'actor': from_user.key.id(),
235	'object': to_user.key.id(),
236	},
237	})
238	obj.put()	1✔
239	from_user.deliver(obj, from_user=from_user)	1✔
240	return f"""OK, you're not blocking {to_user.user_link()} on {to_proto.PHRASE}."""	1✔
241
242
243	@command(['migrate-to'], arg='handle_or_id', user_bridged=True)	1✔
244	def migrate_to(from_user, to_proto, arg, to_user):	1✔
245	try:	1✔
246	to_proto.migrate_out(from_user, to_user.key.id())	1✔
247	except ValueError as e:	×
248	return str(e)	×
249
250	return f"OK, we'll migrate your bridged account on {to_proto.PHRASE} to {to_user.user_link()}."	1✔
251
252
253	@command(None, arg='handle_or_id', user_bridged=True, handle_bridged='eligible')	1✔
254	def prompt(from_user, to_proto, arg, to_user):	1✔
255	"""Prompt a non-bridged user to bridge. No command, just the handle, alone."""
256	from_proto = from_user.__class__	1✔
257	try:	1✔
258	ids.translate_handle(handle=arg, from_=to_proto, to=from_user)	1✔
259	except ValueError as e:	1✔
260	logger.warning(e)	1✔
261	return f"Sorry, Bridgy Fed doesn't yet support bridging handle {arg} from {to_proto.PHRASE} to {from_proto.PHRASE}."	1✔
262
263	if (models.DM(protocol=from_proto.LABEL, type='request_bridging')	1✔
264	in to_user.sent_dms):
265	# already requested
266	return f"We've already sent {to_user.user_link()} a DM. Fingers crossed!"	1✔
267
268	# check and update rate limits
269	attempts_key = f'dm-user-requests-{from_user.LABEL}-{from_user.key.id()}'	1✔
270	# incr leaves existing expiration as is, doesn't change it
271	# https://stackoverflow.com/a/4084043/186123
272	attempts = memcache.memcache.incr(attempts_key, 1)	1✔
273	if not attempts:	1✔
274	memcache.memcache.add(	1✔
275	attempts_key, 1,
276	expire=int(REQUESTS_LIMIT_EXPIRE.total_seconds()))
277	elif attempts > REQUESTS_LIMIT_USER:	1✔
278	return f"Sorry, you've hit your limit of {REQUESTS_LIMIT_USER} requests per day. Try again tomorrow!"	1✔
279
280	# send the DM request!
281	maybe_send(from_=from_proto, to_user=to_user, type='request_bridging', text=f"""\	1✔
282	<p>Hi! {from_user.user_link(proto=to_proto, proto_fallback=True)} is using Bridgy Fed to bridge their account from {from_proto.PHRASE} into {to_proto.PHRASE}, and they'd like to follow you. You can bridge your account into {from_proto.PHRASE} by following this account. <a href="https://fed.brid.gy/docs">See the docs</a> for more information.
283	<p>If you do nothing, your account won't be bridged, and users on {from_proto.PHRASE} won't be able to see or interact with you.
284	<p>Bridgy Fed will only send you this message once.""")
285	return f"Got it! We'll send {to_user.user_link()} a message and say that you hope they'll enable the bridge. Fingers crossed!"	1✔
286
287
288	def maybe_send(*, from_, to_user, text, type=None, in_reply_to=None):	1✔
289	"""Sends a DM.
290
291	Creates a task to send the DM asynchronously.
292
293	If ``type`` is provided, and we've already sent this user a DM of this type
294	from this protocol, does nothing.
295
296	Args:
297	from_ (protocol.Protocol or models.User)
298	to_user (models.User)
299	text (str): message content. May be HTML.
300	type (str): optional, one of DM.TYPES
301	in_reply_to (str): optional, ``id`` of a DM to reply to
302	"""
303	if not to_user.SUPPORTS_DMS:	1✔
304	return	1✔
305
306	from_proto = from_	1✔
307	if not isinstance(from_, User):	1✔
308	assert issubclass(from_, protocol.Protocol)	1✔
309	from web import Web	1✔
310	if not (from_ := Web.get_by_id(from_.bot_user_id())):	1✔
311	logger.info(f'not sending DM, {from_proto.LABEL} has no bot user')	1✔
312	return	1✔
313
314	if type:	1✔
315	dm = models.DM(protocol=from_proto.LABEL, type=type)	1✔
316	if dm in to_user.sent_dms:	1✔
317	return	1✔
318
319	logger.info(f'Sending DM from {from_.key.id()} to {to_user.key.id()} : {text}')	1✔
320
321	if not to_user.obj or not to_user.obj.as1:	1✔
322	logger.info(" can't send DM, recipient has no profile obj")	1✔
323	return	1✔
324
325	now = util.now().isoformat()	1✔
326	dm_id = f'{from_.profile_id()}#bridgy-fed-dm-{type or "?"}-{to_user.key.id()}-{now}'	1✔
327	dm_as1 = {	1✔
328	'objectType': 'note',
329	'id': dm_id,
330	'author': from_.key.id(),
331	'content': text,
332	'inReplyTo': in_reply_to,
333	'tags': [{
334	'objectType': 'mention',
335	'url': to_user.key.id(),
336	}],
337	'published': now,
338	'to': [to_user.key.id()],
339	}
340	Object(id=dm_id, our_as1=dm_as1).put()	1✔
341
342	create_id = f'{dm_id}-create'	1✔
343	create_as1 = {	1✔
344	'objectType': 'activity',
345	'verb': 'post',
346	'id': create_id,
347	'actor': from_.key.id(),
348	'object': dm_as1,
349	'published': now,
350	'to': [to_user.key.id()],
351	}
352
353	target_uri = to_user.target_for(to_user.obj, shared=False)	1✔
354	target = models.Target(protocol=to_user.LABEL, uri=target_uri)	1✔
355	create_task(queue='send', id=create_id, our_as1=create_as1, source_protocol='web',	1✔
356	protocol=to_user.LABEL, url=target.uri, user=from_.key.urlsafe())
357
358	if type:	1✔
359	to_user.sent_dms.append(dm)	1✔
360	to_user.put()	1✔
361
362
363	def receive(*, from_user, obj):	1✔
364	"""Handles a DM that a user sent to one of our protocol bot users.
365
366	Args:
367	from_user (models.User)
368	obj (Object): DM
369
370	Returns:
371	(str, int) tuple: (response body, HTTP status code) Flask response
372	"""
373	recip = as1.recipient_if_dm(obj.as1)	1✔
374	assert recip	1✔
375
376	to_proto = protocol.Protocol.for_bridgy_subdomain(recip)	1✔
377	assert to_proto # already checked in check_supported call in Protocol.receive	1✔
378
379	inner_as1 = (as1.get_object(obj.as1) if as1.object_type(obj.as1) == 'post'	1✔
380	else obj.as1)
381	logger.info(f'got DM from {from_user.key.id()} to {to_proto.LABEL}: {inner_as1.get("content")}')	1✔
382
383	# parse message
384	text = util.remove_invisible_chars(source.html_to_text(inner_as1.get('content', '')))	1✔
385	tokens = text.strip().lower().split()	1✔
386	logger.info(f' tokens: {tokens}')	1✔
387
388	# remove @-mention of bot, if any
389	bot_handles = (DOMAINS + ids.BOT_ACTOR_AP_IDS	1✔
390	+ tuple(h.lstrip('@') for h in ids.BOT_ACTOR_AP_HANDLES))
391	if tokens and tokens[0].lstrip('@') in bot_handles:	1✔
392	logger.debug(f' first token is bot mention, removing')	1✔
393	tokens = tokens[1:]	1✔
394
395	if not tokens or len(tokens) > 2:	1✔
396	return r'¯\_(ツ)_/¯', 204	1✔
397
398	if fn := _commands.get(tokens[0]):	1✔
399	return fn(from_user, to_proto, dm_as1=inner_as1,	1✔
400	cmd=tokens[0], cmd_arg=tokens[1] if len(tokens) == 2 else None)
401	elif len(tokens) == 1:	1✔
402	fn = _commands.get(None)	1✔
403	assert fn, tokens[0]	1✔
404	return fn(from_user, to_proto, dm_as1=inner_as1, cmd=None, cmd_arg=tokens[0])	1✔
405
406	return r'¯\_(ツ)_/¯', 204	×
407
408
409	def load_user_by_handle(proto, handle):	1✔
410	"""
411	Args:
412	proto (protocol.Protocol)
413	handle (str)
414
415	Returns:
416	models.User or None
417	"""
418	if proto.owns_handle(handle) is False:	1✔
419	return None	1✔
420
421	if id := proto.handle_to_id(handle):	1✔
422	if user := proto.get_or_create(id, allow_opt_out=True):	1✔
423	if user.obj and user.obj.as1:	1✔
424	return user	1✔

snarfed / bridgy-fed / 878ced7f-3b32-4aaf-9280-63d2666620db

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous