slidge.core.gateway.base

This module extends slixmpp.ComponentXMPP to make writing new LegacyClients easier

Module Contents

Classes

BaseGateway

Must be subclassed by a plugin to set up various aspects of the XMPP

Attributes

KICKABLE_ERRORS
SLIXMPP_PLUGINS
LOG_STRIP_ELEMENTS
log

class slidge.core.gateway.base.BaseGateway

Bases: slidge.core.gateway.presence.PresenceHandlerMixin, slixmpp.ComponentXMPP, slidge.core.mixins.MessageMixin

Must be subclassed by a plugin to set up various aspects of the XMPP component behaviour, such as its display name or its registration process.

On slidge launch, a singleton is instantiated, and it will be made available to public classes such LegacyContact or BaseSession as the .xmpp attribute. Since it inherits from slixmpp.componentxmpp.ComponentXMPP, this gives you a hand on low-level XMPP interactions via slixmpp plugins, e.g.:

self.send_presence(
    pfrom="somebody@component.example.com",
    pto="someonwelse@anotherexample.com",
)

However, you should not need to do so often since the classes of the plugin API provides higher level abstractions around most commonly needed use-cases, such as sending messages, or displaying a custom status.

property jid

Override to avoid slixmpp deprecation warnings.

REGISTRATION_FIELDS: Collection[slidge.core.command.base.FormField]

Iterable of fields presented to the gateway user when registering using XEP-0077 extended by XEP-0004.

REGISTRATION_INSTRUCTIONS: str = 'Enter your credentials'

The text presented to a user that wants to register (or modify) their legacy account configuration.

REGISTRATION_TYPE

SINGLE_STEP_FORM: 1 step, 1 form, compatible with XEP-0077 (in-band registration)

QRCODE: The registration requires flashing a QR code in an official client. See BaseGateway.()

TWO_FACTOR_CODE: The registration requires confirming login with a 2FA code

COMPONENT_NAME: str

Name of the component, as seen in service discovery by XMPP clients

COMPONENT_TYPE: str | None = ''

Type of the gateway, should ideally follow https://xmpp.org/registrar/disco-categories.html

COMPONENT_AVATAR: slidge.util.types.AvatarType | None

Path, bytes or URL used by the component as an avatar.

ROSTER_GROUP: str = 'slidge'

Roster entries added by the plugin in the user’s roster will be part of the group specified here.

SEARCH_FIELDS: Sequence[slidge.core.command.base.FormField]

Fields used for searching items via the component, through XEP-0055 (jabber search). A common use case is to allow users to search for legacy contacts by something else than their usernames, eg their phone number.

Plugins should implement search by overriding BaseSession.search(), effectively restricting search to registered users by default.

If there is only one field, it can also be used via the jabber:iq:gateway protocol described in XEP-0100. Limitation: this only works if the search request returns one result item, and if this item has a ‘jid’ var.

SEARCH_TITLE: str = 'Search for legacy contacts'

Title of the search form.

SEARCH_INSTRUCTIONS: str = ''

Instructions of the search form.

WELCOME_MESSAGE = "Thank you for registering. Type 'help' to list the available commands, or just start messaging away!"

A welcome message displayed to users on registration. This is useful notably for clients that don’t consider component JIDs as a valid recipient in their UI, yet still open a functional chat window on incoming messages from components.

MARK_ALL_MESSAGES = False

Set this to True for legacy networks that expects read marks for all messages and not just the latest one that was read (as most XMPP clients will only send a read mark for the latest msg).

REGISTRATION_2FA_TITLE = 'Enter your 2FA code'

REGISTRATION_2FA_INSTRUCTIONS = 'You should have received something via email or SMS, or something'

REGISTRATION_QR_INSTRUCTIONS = 'Flash this code or follow this link'

PROPER_RECEIPTS = False

Set this to True if the legacy service provides a real equivalent of message delivery receipts (XEP-0184), meaning that there is an event thrown when the actual device of a contact receives a message. Make sure to call Contact.received() adequately if this is set to True.

GROUPS = False

mtype: slixmpp.types.MessageTypes = 'chat'

is_group = False

_can_send_carbon = False

__register_commands()

__exception_handler(loop, context)

Called when a task created by loop.create_task() raises an Exception

Parameters:

• loop (asyncio.AbstractEventLoop) –

• context –

Returns:

__register_slixmpp_events()

async __on_group_chat_error(msg)

Parameters:

msg (slixmpp.Message) –

async __on_session_start(event)

async __login_wrap(session)

Parameters:

session (slidge.core.session.BaseSession) –

_send(stanza, **send_kwargs)

Parameters:

stanza (Union[slixmpp.Message, slixmpp.Presence]) –

async _on_user_register(iq)

Parameters:

iq (slixmpp.Iq) –

async _on_user_unregister(iq)

Parameters:

iq (slixmpp.Iq) –

raise_if_not_allowed_jid(jid)

Parameters:

jid (slixmpp.JID) –

send_raw(data)

Send raw data across the stream.

Parameters:

data (string) – Any bytes or utf-8 string value.

get_session_from_jid(j)

Parameters:

j (slixmpp.JID) –

async get_muc_from_stanza(iq)

Parameters:

iq (Union[slixmpp.Iq, slixmpp.Message]) –

Return type:

slidge.core.muc.room.LegacyMUC

exception(exception)

Called when a task created by slixmpp’s internal (eg, on slix events) raises an Exception.

Stop the event loop and exit on unhandled exception.

The default slixmpp.basexmpp.BaseXMPP behaviour is just to log the exception, but we want to avoid undefined behaviour.

Parameters:

exception (Exception) – An unhandled Exception object.

re_login(session)

Parameters:

session (slidge.core.session.BaseSession) –

async make_registration_form(_jid, _node, _ifrom, iq)

Parameters:

iq (slixmpp.Iq) –

async user_prevalidate(ifrom, form_dict)

Pre validate a registration form using the content of self.REGISTRATION_FIELDS before passing it to the plugin custom validation logic.

Parameters:

• ifrom (slixmpp.JID) –

• form_dict (dict[str, Optional[str]]) –

async validate(user_jid, registration_form)

Validate a registration form from a user.

Since XEP-0077 is pretty limited in terms of validation, it is OK to validate anything that looks good here and continue the legacy auth process via direct messages to the user (using BaseGateway.input() for instance).

Parameters:

• user_jid (slixmpp.JID) – JID of the user that has just registered

• registration_form (dict[str, Optional[str]]) – A dict where keys are the FormField.var attributes of the BaseGateway.REGISTRATION_FIELDS iterable

async unregister_user(user)

Parameters:

user (slidge.util.db.GatewayUser) –

async unregister(user)

Optionally override this if you need to clean additional stuff after a user has been removed from the permanent user_store.

By default, this just calls session.logout()

Parameters:

user (slidge.util.db.GatewayUser) –

async input(jid, text=None, mtype='chat', **msg_kwargs)

Request arbitrary user input using a simple chat message, and await the result.

You shouldn’t need to call directly bust instead use BaseSession.input() to directly target a user.

NB: When using this, the next message that the user sent to the component will not be transmitted to BaseGateway.on_gateway_message(), but rather intercepted. Await the coroutine to get its content.

Parameters:

• jid (slixmpp.JID) – The JID we want input from

• text – A prompt to display for the user

• mtype (slixmpp.types.MessageTypes) – Message type

Returns:

The user’s reply

Return type:

str

async send_qr(text, **msg_kwargs)

Sends a QR Code to a JID

Parameters:

• text (str) – The text that will be converted to a QR Code

• msg_kwargs – Optional additional arguments to pass to BaseGateway.send_file(), such as the recipient of the QR code.

abstract async validate_two_factor_code(user, code)

Called when the user enters their 2FA code.

Should raise the appropriate XMPPError if the login fails

Parameters:

• user (slidge.util.db.GatewayUser) – The gateway user whose registration is pending Use their .bare_jid and/or``.registration_form`` attributes to get what you need

• code (str) – The code they entered, either via “chatbot” message or adhoc command

abstract async get_qr_text(user)

Plugins should call this to complete registration with QR codes

Parameters:

user (slidge.util.db.GatewayUser) – The not-yet-fully-registered GatewayUser. Use its .bare_jid and/or``.registration_form`` attributes to get what you need

Return type:

str

async confirm_qr(user_bare_jid, exception=None)

Plugins should call this to complete registration with QR codes

Parameters:

• user_bare_jid (str) – The not-yet-fully-registered GatewayUser instance Use their .bare_jid and/or``.registration_form`` attributes to get what you need

• exception (Optional[Exception]) – Optionally, an XMPPError to be raised to not confirm QR code flashing.

shutdown()

Called by the slidge entrypoint on normal exit.

Sends offline presences from all contacts of all user sessions and from the gateway component itself. No need to call this manually, slidge.__main__.main() should take care of it.

slidge.core.gateway.base.KICKABLE_ERRORS = ['gone', 'internal-server-error', 'item-not-found', 'jid-malformed', 'recipient-unavailable',...

slidge.core.gateway.base.SLIXMPP_PLUGINS = ['xep_0030', 'xep_0045', 'xep_0050', 'xep_0054', 'xep_0055', 'xep_0059', 'xep_0066', 'xep_0077',...

slidge.core.gateway.base.LOG_STRIP_ELEMENTS = ['data', 'binval']

slidge.core.gateway.base.log