1# -*- coding: utf-8 -*- 

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import TYPE_CHECKING, Union 

6import logging 

7 

8# -- third party -- 

9# -- own -- 

10from server.endpoint import Client 

11from server.utils import command 

12from utils.events import EventHub 

13import wire 

14 

15# -- typing -- 

16if TYPE_CHECKING: 

17 from server.core import Core # noqa: F401 

18 

19 

20# -- code -- 

21log = logging.getLogger('server.parts.kedama') 

22 

23 

24class Kedama(object): 

25 def __init__(self, core: Core): 

26 self.core = core 

27 D = core.events.client_command 

28 D[wire.CreateRoom].subscribe(self._room_create_limit, -5) 

29 D[wire.JoinRoom].subscribe(self._room_join_limit, -5) 

30 D[wire.Invite].subscribe(self._invite_limit, -5) 

31 

32 def __repr__(self) -> str: 

33 return self.__class__.__name__ 

34 

35 # ----- Commands ----- 

36 @command('*') 

37 def _room_create_limit(self, u: Client, ev: wire.CreateRoom) -> Union[wire.CreateRoom, EventHub.StopPropagation]: 

38 core = self.core 

39 from thb import modes_kedama 

40 if core.auth.is_kedama(u) and ev.mode not in modes_kedama: 

41 u.write(wire.Error('kedama_limitation')) 

42 return EventHub.STOP_PROPAGATION 

43 

44 return ev 

45 

46 @command('*') 

47 def _room_join_limit(self, u: Client, ev: wire.JoinRoom) -> Union[wire.JoinRoom, EventHub.StopPropagation]: 

48 core = self.core 

49 g = core.room.get(ev.gid) 

50 if not g: 50 ↛ 51line 50 didn't jump to line 51, because the condition on line 50 was never true

51 return ev 

52 

53 from thb import modes_kedama 

54 if core.auth.is_kedama(u) and g.__class__.__name__ not in modes_kedama: 

55 u.write(wire.Error('kedama_limitation')) 

56 return EventHub.STOP_PROPAGATION 

57 

58 return ev 

59 

60 @command('*') 

61 def _invite_limit(self, c: Client, ev: wire.Invite) -> Union[wire.Invite, EventHub.StopPropagation]: 

62 core = self.core 

63 uid = core.auth.uid_of(c) 

64 if uid <= 0: 

65 c.write(wire.Error('kedama_limitation')) 

66 return EventHub.STOP_PROPAGATION 

67 

68 return ev