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