Coverage for slidge/command/register.py: 52%
106 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1"""
2This module handles the registration :term:`Command`, which is a necessary
3step for a JID to become a slidge :term:`User`.
4"""
6from __future__ import annotations
8import asyncio
9import functools
10from collections.abc import Mapping, Sequence
11from enum import IntEnum
12from typing import TYPE_CHECKING, Any, NamedTuple
14from slixmpp import JID, Iq
15from slixmpp.exceptions import XMPPError
17from ..core import config
18from ..db import GatewayUser
19from ..util.types import AnySession, RegistrationValidationCoroutine, UserPreferences
20from .base import Command, CommandAccess, Form, FormField, FormSession, FormValues
21from .user import Preferences
23if TYPE_CHECKING:
24 from slidge import BaseGateway
27class AltRegistrationFlow[
28 RegistrationFormT,
29 ModuleDataT: Mapping[str, object],
30](NamedTuple):
31 """
32 Named tuple defining an alternative registration flow.
33 """
35 type: RegistrationType
36 """
37 Type of the alternative registration flow.
38 """
39 title: str
40 """
41 Name of that registration flow (presented to the candidate user).
42 """
43 instructions: str
44 """
45 Instructions of that registration flow.
46 """
47 fields: tuple[FormField, ...] = ()
48 """
49 Fields defining the form of this registration flow.
50 """
51 validate: RegistrationValidationCoroutine[RegistrationFormT, ModuleDataT] | None = (
52 None
53 )
54 """
55 A coroutine awaited with the content of the registration form, similar to
56 :meth:`BaseGatway.validate`. If `None`, then :meth:`BaseGatway.validate` is
57 used and you have to handle different form contents in there.
58 """
61class RegistrationType(IntEnum):
62 """
63 An :class:`Enum` to define the registration flow.
64 """
66 SINGLE_STEP_FORM = 0
67 """
68 1 step, 1 form, the only flow compatible with :xep:`0077`.
69 Using this, the whole flow is defined
70 by :attr:`slidge.BaseGateway.REGISTRATION_FIELDS` and
71 :attr:`.REGISTRATION_INSTRUCTIONS`.
72 """
74 QRCODE = 10
75 """
76 The registration requires flashing a QR code in an official client.
77 See :meth:`slidge.BaseGateway.send_qr`, :meth:`.get_qr_text`
78 and :meth:`.confirm_qr`.
79 """
81 TWO_FACTOR_CODE = 20
82 """
83 The registration requires confirming login with a 2FA code,
84 eg something received by email or SMS to finalize the authentication.
85 See :meth:`.validate_two_factor_code`.
86 """
89class TwoFactorNotRequired(Exception):
90 """
91 Should be raised in :meth:`slidge.BaseGateway.validate` if the code is not
92 required after all. This can happen for a :term:`Legacy Network` where 2FA
93 is optional.
94 """
97class Register[RegistrationFormT, ModuleDataT: Mapping[str, object]](
98 Command[AnySession]
99):
100 NAME = "📝 Register to the gateway"
101 HELP = "Link your JID to this gateway"
102 NODE = "jabber:iq:register"
103 CHAT_COMMAND = "register"
104 ACCESS = CommandAccess.NON_USER
106 SUCCESS_MESSAGE = "Success, welcome!"
108 _fields: tuple[FormField, ...] | None = None
109 _type: RegistrationType | None = None
110 _validate: RegistrationValidationCoroutine | None = None
111 _instructions: str | None = None
113 xmpp: BaseGateway[Any, RegistrationFormT, ModuleDataT]
115 @property
116 def type(self) -> RegistrationType:
117 return self._type or self.xmpp.REGISTRATION_TYPE
119 @property
120 def fields(self) -> Sequence[FormField]:
121 if self._fields is None:
122 return self.xmpp.REGISTRATION_FIELDS
123 return self._fields
125 def _finalize(
126 self,
127 form_values: UserPreferences,
128 _session: None,
129 ifrom: JID,
130 user: GatewayUser,
131 *_: Any, # noqa:ANN401
132 ) -> str:
133 user.preferences = form_values # type: ignore
134 self.xmpp.store.users.update(user)
135 self.xmpp.event("user_register", Iq(sfrom=ifrom.bare))
136 return self.SUCCESS_MESSAGE
138 async def run(
139 self,
140 _session: AnySession | None,
141 ifrom: JID,
142 *_: str,
143 ) -> FormSession[AnySession]:
144 self.xmpp.raise_if_not_allowed_jid(ifrom)
145 return FormSession(
146 title=f"Registration to '{self.xmpp.COMPONENT_NAME}'",
147 instructions=self._instructions or self.xmpp.REGISTRATION_INSTRUCTIONS,
148 fields=self.fields,
149 handler=self.register,
150 )
152 async def register(
153 self,
154 form_values: RegistrationFormT,
155 _session: None,
156 ifrom: JID,
157 ) -> Form | None:
158 two_fa_needed = True
159 try:
160 data = await self.xmpp.user_prevalidate(
161 ifrom, form_values, self.fields, self._validate
162 )
163 except ValueError as e:
164 raise XMPPError("bad-request", str(e))
165 except TwoFactorNotRequired:
166 data = None
167 if self.type == RegistrationType.TWO_FACTOR_CODE:
168 two_fa_needed = False
169 else:
170 raise
171 except Exception as e: # noqa: BLE001
172 raise XMPPError("internal-server-error", str(e))
174 user = GatewayUser(
175 jid=JID(ifrom.bare),
176 legacy_module_data=form_values if data is None else data,
177 )
179 if self.type == RegistrationType.SINGLE_STEP_FORM or (
180 self.type == RegistrationType.TWO_FACTOR_CODE and not two_fa_needed
181 ):
182 return await self.preferences(user)
184 if self.type == RegistrationType.TWO_FACTOR_CODE:
185 return Form(
186 title=self.xmpp.REGISTRATION_2FA_TITLE,
187 instructions=self.xmpp.REGISTRATION_2FA_INSTRUCTIONS,
188 fields=[FormField("code", label="Code", required=True)],
189 handler=functools.partial(self.two_fa, user=user),
190 )
192 elif self.type == RegistrationType.QRCODE:
193 self.xmpp.qr_pending_registrations[user.jid.bare] = (
194 self.xmpp.loop.create_future()
195 )
196 qr_text = await self.xmpp.get_qr_text(user)
197 img_url = await self.xmpp.send_qr(qr_text, mto=ifrom)
198 if img_url is None:
199 raise XMPPError(
200 "internal-server-error", "Slidge cannot send attachments"
201 )
202 self.xmpp.send_text(qr_text, mto=ifrom)
203 return Form(
204 title="Flash this",
205 instructions="Flash this QR in the appropriate place",
206 fields=[
207 FormField(
208 "qr_img",
209 type="fixed",
210 value=qr_text,
211 image_url=img_url,
212 ),
213 FormField(
214 "qr_text",
215 type="fixed",
216 value=qr_text,
217 label="Text encoded in the QR code",
218 ),
219 FormField(
220 "qr_img_url",
221 type="fixed",
222 value=img_url,
223 label="URL of the QR code image",
224 ),
225 ],
226 handler=functools.partial(self.qr, user=user),
227 )
229 async def two_fa(
230 self,
231 form_values: FormValues,
232 _session: None,
233 _ifrom: JID,
234 user: GatewayUser,
235 ) -> Form:
236 assert isinstance(form_values["code"], str)
237 data = await self.xmpp.validate_two_factor_code(user, form_values["code"])
238 if data is not None:
239 user.legacy_module_data.update(data) # type:ignore[arg-type]
240 return await self.preferences(user)
242 async def qr(
243 self,
244 _form_values: FormValues,
245 _session: None,
246 _ifrom: JID,
247 user: GatewayUser,
248 ) -> Form:
249 try:
250 data = await asyncio.wait_for(
251 self.xmpp.qr_pending_registrations[user.jid.bare],
252 config.QR_TIMEOUT,
253 )
254 except TimeoutError:
255 raise XMPPError(
256 "remote-server-timeout",
257 (
258 "It does not seem that the QR code was correctly used, "
259 "or you took too much time"
260 ),
261 )
262 if data is not None:
263 user.legacy_module_data.update(data) # type:ignore[arg-type]
264 return await self.preferences(user)
266 async def preferences(self, user: GatewayUser) -> Form:
267 return Form(
268 title="Preferences",
269 instructions=Preferences.HELP,
270 fields=self.xmpp.PREFERENCES,
271 handler=functools.partial(self._finalize, user=user),
272 timeout_handler=functools.partial(self._preferences_timeout, user=user),
273 )
275 def _preferences_timeout(self, user: GatewayUser) -> None:
276 self.xmpp.event("user_register", Iq(sfrom=user.jid.bare))
277 self.xmpp.store.users.update(user)
278 self.xmpp.send_message(
279 mfrom=self.xmpp.boundjid.bare,
280 mto=user.jid,
281 mbody="You did not choose your preferences in time, falling back to defaults. "
282 "You can change preferences later with the 'preferences' command.\n"
283 + self.SUCCESS_MESSAGE,
284 )