|
|
  ws_handler
1 # coding: utf-8
2 import struct
3 import time
4 import weakref
5 from hashlib import md5
6
7 from tornado.websocket import WebSocketHandler, WebSocketClosedError
8
9 from core.player_mgr import player_mgr
10 from settings import crypt, update_header, sign
11 from core.session_mgr import session_mgr
12 from core.logger import logger
13 from core.blacklist import blacklist
14
15
16 # noinspection PyAbstractClass
17 class WSHandler(WebSocketHandler):
18
19 uuid = None
20 app_id = None
21 game_id = None
22 ver = None
23 timestamp = None
24 key = None
25 last_time = 0
26 stick_package_stack = None
27 token = None
28 player = None
29 receive_id = 0
30 send_id = 0
31
32 def bind(self, player):
33 if player.session:
34 player.session.close(2)
35 player.logger("player {0} cancel old session".format(self.uuid))
36 self.player = weakref.proxy(player)
37 self.player.session = self
38 player_mgr.add_player(player)
39
40 def check_origin(self, origin):
41 return True
42
43 @staticmethod
44 def str2bytes(string):
45 return bytes(string.encode("utf8"))
46
47 def xor(self, string):
48 if not crypt:
49 return string
50 value = bytearray()
51 step = len(self.token)
52 for i in range(0, len(string), step):
53 for j in range(step):
54 if i + j >= len(string):
55 break
56 value.append(string[i + j] ^ self.token[j])
57 return bytes(value)
58
59 def encrypt(self, ostream):
60 return self.xor(ostream)
61
62 def decrypt(self, istream):
63 return self.xor(istream)
64
65 def verify_args(self):
66 try:
67 #self.app_id = int(self.get_argument("app_id"))
68 #self.game_id = int(self.get_argument("game_id"))
69 # self.uuid = self.get_argument("uuid")
70 # self.ver = [int(i) for i in self.get_argument("version").split(".")]
71 # self.timestamp = int(self.get_argument("timestamp"))
72 self.key = self.get_argument("key")
73 except Exception as e:
74 logger.error("session argument type incorrect {0} {1}".format(self.request.uri, e))
75 return False
76 try:
77 assert self.key #self.app_id and self.game_id and self.uuid and self.ver and self.timestamp and self.key
78 return True
79 except AssertionError:
80 logger.error("session miss arguments {0}".format(self.request.uri))
81 return False
82
83 def auth(self):
84 uri = self.request.uri
85 last_index = len(uri)#uri.rindex("&")
86 body = uri[uri.index("?")+1: last_index]
87 key = uri[last_index+5:]
88 self.timestamp = 0
89 t = abs((time.time() * 1000 - self.timestamp) / 1000)
90 if t > 30 and self.timestamp:
91 logger.info("session close: auth failed, timestamp expired {0}".format(t))
92 blacklist(self.request.host_name, self.request.remote_ip, 2)
93 # return False
94 md5_key = md5((body+"52LYD50YDs57jmy52Bc49").encode("utf-8")).hexdigest()
95 # print("key", key, "md5", md5_key)
96 #if key == md5_key:
97 if uri.find(md5_key):
98 return True
99 else:
100 logger.info("session close: auth failed, key incorrect")
101 blacklist(self.request.host_name, self.request.remote_ip, 3)
102 return False
103
104 def open(self, *args, **kwargs):
105 if not self.verify_args():
106 blacklist(self.request.host_name, self.request.remote_ip, 1)
107 self.close(8)
108 return
109 if not self.auth():
110 self.close(7)
111 return
112 session_mgr.add(self)
113 logger.debug("proxy {0} remote ip {1}".format(self.request.host_name, self.request.remote_ip))
114
115 def close(self, code=None, reason=None):
116 #主动断开
117 super().close(code, reason)
118 if self.player:
119 self.player.session = None
120 session_mgr.delete(self)
121 self.player.online_state = 0
122
123 def on_close(self):
124 #收到客户端断开连接触发事件
125 session_mgr.delete(self)
126 logger.debug("session {0} close {1}".format(self.uuid, self.close_code))
127 try:
128 if self.player and self.player.session:
129 self.player.online_state = 0
130 self.player.offline()
131 self.player.session = None
132 self.player = None
133 except (ReferenceError, AttributeError):
134 pass
135 if isinstance(self.close_code, int) and self.close_code >= 1000:
136 logger.warn("{0} close code {1} bad".format(self.uuid, self.close_code))
137
138 def unpack_header(self, raw):
139 from protocol.deserialize import parse
140 if update_header:
141 size, seq, app_id, game_id, t, v1, v2, v3, sign1, sign2, sign3, sign4, sign5, sign6, sign7, sign8, sign9, cmd, = struct.unpack('>iihhibbbbbbbbbbbbb', raw[:29])
142 string, = struct.unpack('>{0}s'.format(size - 29), raw[29:size])
143 # print("receive_id", seq, self.receive_id)
144 client_sign = chr(sign1) + chr(sign2) + chr(sign3) + chr(sign4) + chr(sign5) + chr(sign6) + chr(sign7) + chr(sign8) + chr(sign9)
145 if seq == self.receive_id and client_sign == sign:
146 self.receive_id += 1
147 parse(cmd, string, self)
148 else:
149 self.close(3)
150 else:
151 size, = struct.unpack('>H',raw[0:2])
152 cmd, = struct.unpack('>H', raw[2:4])
153 string, = struct.unpack('>{0}s'.format(size - 4), raw[4:size])
154
155 # size, seq, self.game_id, cmd, = struct.unpack('>ihh', raw[:0])
156 # string, = struct.unpack('>{0}s'.format(size - 1), raw[1:size])
157 # size = len(raw)
158 # cmd = int(raw[0])#struct.unpack('>ihh', raw[:0])
159 # string = raw[2:size]#struct.unpack('>{0}s'.format(size - 1), raw[1:size])
160
161 #协议处理
162 parse(cmd, string, self)
163
164 def send(self, cmd, proto, serialize=True):
165 if serialize:
166 result = proto.SerializeToString()
167 else:
168 result = proto
169 result = self.encrypt(result)
170 if update_header:
171 fmt = ">iihhibbbb{0}s".format(len(result))
172 args = (len(result) + 20, self.send_id, self.app_id, self.game_id, int(time.time()), *self.ver, cmd, result)
173 #else:
174 # fmt = ">ii{0}s".format(len(result))
175 # args = (len(result) + 8, cmd, result)
176 fmt = ">HH{0}s".format(len(result))
177 args = (len(result), cmd,result)
178 data = struct.pack(fmt, *args)
179 # data = struct.pack('>BB{0}s',len(result),cmd,len(result),result)
180 try:
181 self.write_message(data, True)
182 if len(data) > 1024:
183 logger.debug("{0} send message size {1}".format(self.uuid, len(data)))
184 except WebSocketClosedError:
185 pass
186 self.send_id += 1
187
188 def on_message(self, buf):
189 self.last_time = time.time()
190 """解析数据流（半包，全包，粘包），并将命令和协议数据解开"""
191 if isinstance(buf, str):
192 buf = buf.encode(encoding='utf-8')
193 self.unpack_header(buf)
194 return
195 #buf = buf.encode(encoding='utf-8')
196 #self.unpack_header(buf)
197
198 try:
199 if len(buf) > 1024:
200 logger.debug("{0} receive message size {1}".format(self.uuid, len(buf)))
201 if self.stick_package_stack:
202 buf = self.stick_package_stack + buf
203 self.stick_package_stack = None
204 #size, = struct.unpack('>i', buf[:4])
205 size, = struct.unpack('>H',buf[0:2])
206 #size = buf[0]
207 raw_size = len(buf)
208 if raw_size > size: # 至少有一个包
209 self.unpack_header(buf)
210 rest = buf[size:]
211 self.on_message(rest)
212 elif size > raw_size: # 半包
213 self.stick_package_stack = buf
214 logger.warn("{0} stick package {1}".format(self.uuid, len(buf)))
215 else:
216 self.unpack_header(buf)
217 except Exception as e:
218 logger.exception("message parse failed {0}".format(str(e)))
219 封装微博socket操作，包括解包、发包等，这里采用谷歌protobuf协议格式，具体的解包、组包格式在另外一篇中介绍
|