# Coverage for server/parts/auth.py : 89%

## 70 statements 64 run 6 missing 4 excluded 3 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

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

2 from __future__ import annotations

3

4# -- stdlib --

5 from typing import Sequence, Set, TYPE_CHECKING, Tuple

6 import logging

7

8# -- third party --

9 from mypy_extensions import TypedDict

10

11# -- own --

12 from server.endpoint import Client

13 from server.utils import command

14 import wire

15

16# -- typing --

17 if TYPE_CHECKING:

18 from server.core import Core # noqa: F401

19

20

21# -- code --

22 log = logging.getLogger('Auth')

23

24

25 class AuthAssocOnClient(TypedDict):

26 uid: int

27 name: str

28 kedama: bool

29 permissions: Set[str]

30

31

32 def A(self: Auth, v: Client) -> AuthAssocOnClient:

33 return v._[self]

34

35

36 class Auth(object):

37 def __init__(self, core: Core):

38 self.core = core

39 self._kedama_uid = 10032

40 self._allow_kedama = True

41

42 core.events.user_state_transition += self.handle_user_state_transition

43 D = core.events.client_command

44 D[wire.Auth] += self._auth

45

46 def __repr__(self) -> str:

47 return self.__class__.__name__

48

49 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]:

50 u, f, t = ev

51

52 if (f, t) == ('initial', 'connected'):

53 from settings import VERSION

54 core = self.core

55 u.write(wire.Greeting(node=core.options.node, version=VERSION))

56 assoc: AuthAssocOnClient = {

57 'uid': 0,

58 'name': '',

59 'kedama': False,

60 'permissions': set(),

61 }

62 u._[self] = assoc

63

64 return ev

65

66 # ----- Command -----

67

68 @command('connected')

69 def _auth(self, u: Client, m: wire.Auth) -> None:

70 core = self.core

71 token = m.token

72

73 if token == '':

74 if self._allow_kedama: 74 ↛ 87 line 74 didn't jump to line 87, because the condition on line 74 was never false

75 uid = -self._kedama_uid

76 self._kedama_uid += 1

77 u.write(wire.AuthSuccess(uid))

78 assoc: AuthAssocOnClient = {

79 'uid': uid,

80 'name': f'毛玉{-uid}',

81 'kedama': True,

82 'permissions': set(),

83 }

84 u._[self] = assoc

85 core.lobby.state_of(u).transit('authed')

86 else:

87 u.write(wire.AuthError('not_available'))

88

89 return

90

91 rst = core.backend.query('''

92 query($token: String) {

93 player(token: $token) {

94 id

95 user {

96 isActive

97 userPermissions {

98 codename

99 }

100 groups {

101 permissions {

102 codename

103 }

104 }

105 }

106 name

107 }

108 }

109 ''', token=token)

110

111 if not rst or not rst['player']: 111 ↛ 112 line 111 didn't jump to line 112, because the condition on line 111 was never true

112 u.write(wire.AuthError('invalid_credentials'))

113 return

114

115 rst = rst['player']

116

117 if not rst['user']['isActive']: 117 ↛ 118 line 117 didn't jump to line 118, because the condition on line 117 was never true

118 u.write(wire.AuthError('not_available'))

119 else:

120 uid = int(rst['id'])

121 u.write(wire.AuthSuccess(uid))

122 assoc: AuthAssocOnClient = {

123 'uid': uid,

124 'name': rst['name'],

125 'kedama': False,

126 'permissions': set(

127 [i['codename'] for i in rst['user']['userPermissions']] +

128 [i['codename'] for i in rst['user']['groups']['permissions']]

129 ),

130 }

131 u._[self] = assoc

132 core.lobby.state_of(u).transit('authed')

133

134 # ----- Public Methods -----

135 def allow_kedama(self) -> None:

136 self._allow_kedama = True

137

138 def deny_kedama(self) -> None:

139 self._allow_kedama = False

140

141 def uid_of(self, u: Client) -> int:

142 return A(self, u)['uid']

143

144 def name_of(self, u: Client) -> str:

145 return A(self, u)['name']

146

147 def is_kedama(self, u: Client) -> bool:

148 return A(self, u)['kedama']

149

150 def set_auth(self, u: Client, uid: int = 1, name: str = 'Foo', kedama: bool = False, permissions: Sequence[str] = []) -> None:

151 assoc: AuthAssocOnClient = {

152 'uid': uid,

153 'name': name,

154 'kedama': kedama,

155 'permissions': set(permissions),

156 }

157 u._[self] = assoc