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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Sequence, Set, TYPE_CHECKING, Tuple 

6import logging 

7 

8# -- third party -- 

9from mypy_extensions import TypedDict 

10 

11# -- own -- 

12from server.endpoint import Client 

13from server.utils import command 

14import wire 

15 

16# -- typing -- 

17if TYPE_CHECKING: 

18 from server.core import Core # noqa: F401 

19 

20 

21# -- code -- 

22log = logging.getLogger('Auth') 

23 

24 

25class AuthAssocOnClient(TypedDict): 

26 uid: int 

27 name: str 

28 kedama: bool 

29 permissions: Set[str] 

30 

31 

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

33 return v._[self] 

34 

35 

36class 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 ↛ 87line 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 ↛ 112line 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 ↛ 118line 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