3cd1d0d6-6477-432e-a032-fafe4361bf08

Committed 03 Nov 2025 08:23PM UTC coverage: 92.903% (+0.03%) from 92.878%

Build # 3cd1d0d6-6477-432e-a032-fafe4361bf08

Build Type

push

circleci

Committed by

snarfed

Commit Message

extend the `block` DM command to allow blocking a list

for #1632

Run Details

10 of 10 new or added lines in 2 files covered. (100.0%)

20 existing lines in 2 files now uncovered.

6022 of 6482 relevant lines covered (92.9%)

0.93 hits per line

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

97.74

/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 werkzeug.exceptions import BadRequest

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':
                cmd_arg = cmd_arg.removeprefix('@')
                if not (to_user := _load_user(cmd_arg, to_proto)):
                    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 or list URL]</em>: block a user who's not bridged here, or a list, on {to_proto.PHRASE}
<li><em>unblock [handle or ID or list URL]</em>: unblock a user who's not bridged here, or a list, on {to_proto.PHRASE}
{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=True, user_bridged=True)
def block(from_user, to_proto, arg):
    try:
        # first, try interpreting as a user handle or id
        blockee = _load_user(arg, to_proto)
    except BadRequest:
        # may not be a user, see if it's a list
        blockee = to_proto.load(arg)
        if not blockee or blockee.type != 'collection':
            return f"{arg} doesn't look like a user or list on {to_proto.PHRASE}"

    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': blockee.key.id(),
    })
    obj.put()
    from_user.deliver(obj, from_user=from_user)

    link = (blockee.user_link() if isinstance(blockee, User)
            else util.pretty_link(blockee.as1.get('url') or '',
                                  text=blockee.as1.get('displayName')))

    return f"""OK, you're now blocking {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(handle_or_id, proto):
    """Loads the user with the given handle or id.

    Args:
      handle_or_id (str)
      protocol (Protocol subclass)

    Returns:
      modelsUser or None:
    """
    if proto.owns_id(handle_or_id) is not False:
        return proto.get_or_create(handle_or_id, allow_opt_out=True)

    logging.info(f"doesn't look like an ID, trying as a handle")

    handle_or_id = handle_or_id.removeprefix('@')
    if proto.owns_handle(handle_or_id) is False:
        return None

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

snarfed / bridgy-fed / 3cd1d0d6-6477-432e-a032-fafe4361bf08

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