Coverage for slidge / core / dispatcher / registration.py: 71%

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING 

5 

6from slixmpp import JID, Iq 

7from slixmpp.exceptions import XMPPError 

8 

9from ...db import GatewayUser 

10from .. import config 

11from .util import DispatcherMixin 

12 

13if TYPE_CHECKING: 

14 from slidge.core.gateway import BaseGateway 

15 

16 

17class RegistrationMixin(DispatcherMixin): 

18 __slots__: list[str] = [] 

19 

20 def __init__(self, xmpp: BaseGateway) -> None: 

21 super().__init__(xmpp) 

22 xmpp["xep_0077"].api.register( 

23 self.xmpp.make_registration_form, "make_registration_form" 

24 ) 

25 xmpp["xep_0077"].api.register(self._user_get, "user_get") 

26 xmpp["xep_0077"].api.register(self._user_validate, "user_validate") 

27 xmpp["xep_0077"].api.register(self._user_modify, "user_modify") 

28 # kept for slixmpp internal API compat 

29 # TODO: either fully use slixmpp internal API or rewrite registration without it at all 

30 xmpp["xep_0077"].api.register(lambda *a: None, "user_remove") 

31 

32 xmpp.add_event_handler("user_register", self._on_user_register) 

33 xmpp.add_event_handler("user_unregister", self._on_user_unregister) 

34 

35 def get_user(self, jid: JID) -> GatewayUser | None: 

36 session = self.xmpp.get_session_from_jid(jid) 

37 if session is None: 

38 return None 

39 return session.user 

40 

41 async def _user_get( 

42 self, 

43 _gateway_jid: JID, 

44 _node: str, 

45 ifrom: JID, 

46 iq: Iq, 

47 ) -> GatewayUser | None: 

48 if ifrom is None: 

49 ifrom = iq.get_from() 

50 return self.get_user(ifrom) 

51 

52 async def _user_validate( 

53 self, 

54 _gateway_jid: JID, 

55 _node: str, 

56 ifrom: JID, 

57 iq: Iq, 

58 ) -> None: 

59 xmpp = self.xmpp 

60 log.debug("User validate: %s", ifrom.bare) 

61 form_dict = {f.var: iq.get(f.var) for f in xmpp.REGISTRATION_FIELDS} 

62 xmpp.raise_if_not_allowed_jid(ifrom) 

63 legacy_module_data = await xmpp.user_prevalidate(ifrom, form_dict) 

64 if legacy_module_data is None: 

65 legacy_module_data = form_dict 

66 with self.xmpp.store.session() as orm: 

67 user = GatewayUser( 

68 jid=ifrom.bare, 

69 legacy_module_data=legacy_module_data, 

70 ) 

71 orm.add(user) 

72 orm.commit() 

73 log.info("New user: %s", user) 

74 

75 async def _user_modify( 

76 self, 

77 _gateway_jid: JID, 

78 _node: str, 

79 ifrom: JID, 

80 form_dict: dict[str, str | None], 

81 ) -> None: 

82 await self.xmpp.user_prevalidate(ifrom, form_dict) 

83 log.debug("Modify user: %s", ifrom) 

84 with self.xmpp.store.session() as orm: 

85 user = orm.query(GatewayUser).one_or_none() 

86 if user is None: 

87 raise XMPPError("internal-server-error", "User not found") 

88 user.legacy_module_data.update(form_dict) 

89 self.xmpp.store.users.update(user) 

90 

91 async def _on_user_register(self, iq: Iq) -> None: 

92 session = await self._get_session(iq, wait_for_ready=False) 

93 for jid in config.ADMINS: 

94 self.xmpp.send_message( 

95 mto=jid, 

96 mbody=f"{iq.get_from()} has registered", 

97 mtype="chat", 

98 mfrom=self.xmpp.boundjid.bare, 

99 ) 

100 session.send_gateway_message(self.xmpp.WELCOME_MESSAGE) 

101 await self.xmpp.login_wrap(session) 

102 

103 async def _on_user_unregister(self, iq: Iq) -> None: 

104 await self.xmpp.kill_session(iq.get_from()) 

105 

106 

107log = logging.getLogger(__name__)