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

73 

74 if token == '': 

75 if self._allow_kedama: 75 ↛ 88line 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 ↛ 113line 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 ↛ 119line 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