# 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 assoc: AuthAssocOnClient

73

74 if token == '':

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

76 uid = -self._kedama_uid

77 self._kedama_uid += 1

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

79 assoc = {

80 'uid': uid,

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

82 'kedama': True,

83 'permissions': set(),

84 }

85 u._[self] = assoc

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

87 else:

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

89

90 return

91

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

93 query($token: String) {

94 player(token: $token) {

95 id

96 user {

97 isActive

98 userPermissions {

99 codename

100 }

101 groups {

102 permissions {

103 codename

104 }

105 }

106 }

107 name

108 }

109 }

110 ''', token=token)

111

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

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

114 return

115

116 rst = rst['player']

117

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

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

120 else:

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

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

123 assoc = {

124 'uid': uid,

125 'name': rst['name'],

126 'kedama': False,

127 'permissions': set(

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

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

130 ),

131 }

132 u._[self] = assoc

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

134

135 # ----- Public Methods -----

136 def allow_kedama(self) -> None:

137 self._allow_kedama = True

138

139 def deny_kedama(self) -> None:

140 self._allow_kedama = False

141

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

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

144

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

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

147

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

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

150

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

152 assoc: AuthAssocOnClient = {

153 'uid': uid,

154 'name': name,

155 'kedama': kedama,

156 'permissions': set(permissions),

157 }

158 u._[self] = assoc