diff --git a/Client.py b/Client.py index 927502e6..dcb60863 100644 --- a/Client.py +++ b/Client.py @@ -153,6 +153,20 @@ def set_msg_id(self, msg): self.msg_id = '#%s ' % test return (' '.join(msg.split(' ')[1:])) + def with_msg_id(self, callback): + # a reply sent after the handler returns, from a DB callback or the login queue, still + # answers the command that started it. Capture that command's id now and put it back + # on the client only while the callback runs. + msg_id = self.msg_id + def run(*args, **kwargs): + previous = self.msg_id + self.msg_id = msg_id + try: + return callback(*args, **kwargs) + finally: + self.msg_id = previous + return run + def setFlagByIP(self, ip, force=True): cc = ip2country.lookup(ip) if force or cc != '??': diff --git a/DataHandler.py b/DataHandler.py index 5c462a2d..dcd1e11c 100644 --- a/DataHandler.py +++ b/DataHandler.py @@ -210,7 +210,7 @@ def __init__(self): # simultaneously backed up on their write buffers (the 2.3 producer signal), new # logins are queued FIFO and drained by drain_login_queue() (a 1s LoopingCall) once # the backpressure clears. Under normal load the queue stays empty. - self.login_queue = collections.deque() # (client, login_args) awaiting login under backpressure + self.login_queue = collections.deque() # (client, login_now carrying the LOGIN's msg_id, login_args) awaiting login under backpressure self.login_backpressure_limit = 50 # paused-producer count above which login admission pauses # rate limits @@ -807,11 +807,11 @@ def drain_login_queue(self): # shares login_queue with in_LOGIN without locking. Each login gets its own session # commit/rollback/close, mirroring the per-request guards in dataReceived. while self.login_queue and not self.login_backpressured(): - client, args = self.login_queue.popleft() + client, login_now, args = self.login_queue.popleft() if client.session_id not in self.clients: continue # client disconnected while queued try: - self.protocol.login_now(client, *args) + login_now(client, *args) self.session_manager.commit_guard() except: logging.error(traceback.format_exc()) diff --git a/PROTOCOL.md b/PROTOCOL.md index 6dfa6f68..4b3139d1 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -74,9 +74,13 @@ cannot drift. **[GAP]** the `@emits` convention does not exist yet — see argument may legitimately contain spaces (e.g. chat text, topics). This is the "sentence argument" behaviour. (`Protocol.py` `get_function_args`.) - **Message IDs.** A client may prefix a command with `# ` to correlate a request - with the server's reply; the server echoes the id back on responses generated in that - command's handling. (`Client.py`.) **[GAP]** document exact echo semantics and which - responses carry the id vs which do not. + with the server's reply. The server puts the same `# ` on every line it sends that + client as the answer to that command, including refusals (`SERVERMSG`, `FAILED`) and + replies that arrive later because the command waited on the database or the login queue. + Lines that are not an answer to the command carry no id, even when they arrive before + the answer: chat from other players, broadcasts, status changes, and offline messages + delivered after `LOGIN`. Lines sent to other clients as a result of the command carry no + id either. (`Client.py` `set_msg_id`, `with_msg_id`.) - **Tab-separated payloads.** Some structured fields (notably battle `script_tags`) use tab-separated `key=value` pairs. **[GAP]** enumerate every command that uses tab separation and the exact field grammar. diff --git a/protocol/Protocol.py b/protocol/Protocol.py index 99919a87..8e9a06a5 100755 --- a/protocol/Protocol.py +++ b/protocol/Protocol.py @@ -395,11 +395,17 @@ def get_function_args(self, client, command, function, numspaces, args): def _handle(self, client, msg): assert(type(msg) == str) - # client.Send() prepends client.msg_id if the current thread - # is the same thread as the client's handler. - # this works because handling is done in order for each ClientHandler thread - # so we can be sure client.Send() was performed in the client's own handling code. + # client.Send() prepends client.msg_id, so the id is set only while this command is + # handled. Restore rather than clear: ChanServ runs a mod command typed into #moderator + # through here while the same client's SAY is still being handled. + previous_msg_id = client.msg_id msg = client.set_msg_id(msg) + try: + return self._dispatch(client, msg) + finally: + client.msg_id = previous_msg_id + + def _dispatch(self, client, msg): numspaces = msg.count(' ') if (numspaces > 0): @@ -1019,8 +1025,8 @@ def in_REGISTER(self, client, username, password, email = ''): # the INSERT continue in _register_checked, preserving the original denial ordering. email = email.lower() d = self._root.defer_db(self.userdb.check_register_user, username, email, client.ip_address) - d.addCallback(self._register_checked, client, username, password, email) - d.addErrback(self._register_failed, client, username) + d.addCallback(client.with_msg_id(self._register_checked), client, username, password, email) + d.addErrback(client.with_msg_id(self._register_failed), client, username) def _register_checked(self, verdict, client, username, password, email): # reactor-thread callback after the uniqueness/ban check. Applies the same memory-only @@ -1054,8 +1060,8 @@ def _register_checked(self, verdict, client, username, password, email): # hop 2: INSERT off the reactor; the worker catches the unique-constraint race. d = self._root.defer_db(self.userdb.do_register_insert, username, password, client.ip_address, email) - d.addCallback(self._register_inserted, client, username, email) - d.addErrback(self._register_failed, client, username) + d.addCallback(client.with_msg_id(self._register_inserted), client, username, email) + d.addErrback(client.with_msg_id(self._register_failed), client, username) def _register_inserted(self, verdict, client, username, email): # reactor-thread callback after the INSERT. @@ -1152,7 +1158,7 @@ def in_LOGIN(self, client, username, password, cpu='0', local_ip='', sentence_ar keeps the same signature so command dispatch (which reflects on it) is unchanged. ''' if self._root.login_queue or self._root.login_backpressured(): - self._root.login_queue.append((client, (username, password, cpu, local_ip, sentence_args))) + self._root.login_queue.append((client, client.with_msg_id(self.login_now), (username, password, cpu, local_ip, sentence_args))) client.Send('SERVERMSG The server is busy; you are number %d in the login queue, please wait...' % len(self._root.login_queue)) return self.login_now(client, username, password, cpu, local_ip, sentence_args) @@ -1193,8 +1199,8 @@ def login_now(self, client, username, password, cpu='0', local_ip='', sentence_a # _login_checked once the DB worker returns. The remaining memory-only checks and # the denial ordering are applied in the callback, preserving the original flow. d = self._root.defer_db(self.userdb.precheck_login, username, password, client.ip_address) - d.addCallback(self._login_checked, client, username, local_ip, sentence_args) - d.addErrback(self._login_failed, client, username) + d.addCallback(client.with_msg_id(self._login_checked), client, username, local_ip, sentence_args) + d.addErrback(client.with_msg_id(self._login_failed), client, username) def _login_checked(self, result, client, username, local_ip, sentence_args): # reactor-thread callback after precheck_login (result is plain data). Applies the @@ -1239,8 +1245,8 @@ def _login_checked(self, result, client, username, local_ip, sentence_args): # login checks complete; write the login record off the reactor thread. d = self._root.defer_db(self.userdb.do_login, username, client.ip_address, agent, last_sys_id, last_mac_id, local_ip, client.country_code) - d.addCallback(self._login_finish, client, username, agent, local_ip) - d.addErrback(self._login_failed, client, username) + d.addCallback(client.with_msg_id(self._login_finish), client, username, agent, local_ip) + d.addErrback(client.with_msg_id(self._login_failed), client, username) def _login_finish(self, result, client, username, agent, local_ip): # reactor-thread callback after do_login. result is (snapshot, ignored_user_ids): @@ -1466,8 +1472,8 @@ def in_CONFIRMAGREEMENT(self, client, verification_code = ""): # dump. _SendLoginInfo touches the DB on the reactor (get_ignored_user_ids), so the # callback brackets it with the session guards exactly like _login_finish does. d = self._root.defer_db(self.userdb.do_confirm_agreement, client.username) - d.addCallback(self._confirmagreement_done, client) - d.addErrback(self._confirmagreement_failed, client) + d.addCallback(client.with_msg_id(self._confirmagreement_done), client) + d.addErrback(client.with_msg_id(self._confirmagreement_failed), client) def _confirmagreement_done(self, uid, client): if client.session_id not in self._root.clients: @@ -1675,8 +1681,8 @@ def _enqueue_offline_message(self, client, user, msg, ex_msg): # never echo a message that was refused (unknown user, bot target). cmd = 'SAYPRIVATEEX' if ex_msg else 'SAYPRIVATE' d = self._root.defer_db(self.userdb.do_enqueue_offline_message, client.user_id, user, msg, ex_msg) - d.addCallback(self._offline_message_queued, client, user, msg, cmd) - d.addErrback(self._offline_message_failed, client, user, cmd) + d.addCallback(client.with_msg_id(self._offline_message_queued), client, user, msg, cmd) + d.addErrback(client.with_msg_id(self._offline_message_failed), client, user, cmd) def _offline_message_queued(self, result, client, user, msg, cmd): # reactor-thread callback: pure send, no DB -> no commit/rollback/close bracket. @@ -1937,8 +1943,8 @@ def in_IGNORE(self, client, tags): # target's access; the already-ignored/list-full checks and the client.ignored # mutation stay on the reactor thread (live memory) - see the callbacks below. d = self._root.defer_db(self.userdb._user_access_from_username, username) - d.addCallback(self._ignore_resolved, client, username, reason) - d.addErrback(self._ignore_op_failed, client, "IGNORE") + d.addCallback(client.with_msg_id(self._ignore_resolved), client, username, reason) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "IGNORE") def _ignore_resolved(self, resolved, client, username, reason): # reactor-thread callback: resolved is (target_id, access) or None. No reactor-side DB @@ -1964,8 +1970,8 @@ def _ignore_resolved(self, resolved, client, username, reason): return # checks passed: do the bare INSERT off the reactor, then sync memory + reply. d = self._root.defer_db(self.userdb.do_ignore_insert, client.user_id, target_id, reason) - d.addCallback(self._ignore_stored, client, username, target_id, reason) - d.addErrback(self._ignore_op_failed, client, "IGNORE") + d.addCallback(client.with_msg_id(self._ignore_stored), client, username, target_id, reason) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "IGNORE") def _ignore_stored(self, result, client, username, target_id, reason): # reactor-thread callback: the row is committed; sync in-memory state + reply. @@ -1992,8 +1998,8 @@ def in_UNIGNORE(self, client, tags): # 3.1: resolve off the reactor (reuse the IGNORE resolver; access is unused here). # The not-ignored check reads live memory and the DELETE runs off the reactor. d = self._root.defer_db(self.userdb._user_access_from_username, username) - d.addCallback(self._unignore_resolved, client, username) - d.addErrback(self._ignore_op_failed, client, "UNIGNORE") + d.addCallback(client.with_msg_id(self._unignore_resolved), client, username) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "UNIGNORE") def _unignore_resolved(self, resolved, client, username): # reactor-thread callback: only memory + a second deferred, so no session bracket. @@ -2007,8 +2013,8 @@ def _unignore_resolved(self, resolved, client, username): self.out_SERVERMSG(client, "User is not ignored.") return d = self._root.defer_db(self.userdb.do_unignore_delete, client.user_id, target_id) - d.addCallback(self._unignore_removed, client, username, target_id) - d.addErrback(self._ignore_op_failed, client, "UNIGNORE") + d.addCallback(client.with_msg_id(self._unignore_removed), client, username, target_id) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "UNIGNORE") def _unignore_removed(self, result, client, username, target_id): # reactor-thread callback: the row(s) are deleted; sync in-memory state + reply. @@ -2029,8 +2035,8 @@ def in_IGNORELIST(self, client): # 3.1: read the ignore list off the reactor thread, then format + send in # _ignorelist_send once the worker returns plain [(ignored_user_id, reason), ...]. d = self._root.defer_db(self.userdb.get_ignore_list, client.user_id) - d.addCallback(self._ignorelist_send, client) - d.addErrback(self._social_list_failed, client, "IGNORELIST") + d.addCallback(client.with_msg_id(self._ignorelist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "IGNORELIST") def _ignorelist_send(self, entries, client): # reactor-thread callback. clientFromID may do a reactor-side DB read on an @@ -2078,8 +2084,8 @@ def in_FRIENDREQUEST(self, client, tags): # online ignore-set is kept in sync with the ignores table, so the db check is # correct either way. Notify an online target in the callback. d = self._root.defer_db(self.userdb.do_friend_request, client.user_id, username, msg) - d.addCallback(self._friendrequest_done, client, username, msg) - d.addErrback(self._friend_op_failed, client, "FRIENDREQUEST") + d.addCallback(client.with_msg_id(self._friendrequest_done), client, username, msg) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "FRIENDREQUEST") def _friendrequest_done(self, verdict, client, username, msg): # reactor-thread callback: no reactor-side DB (clientFromID is memory-only here), @@ -2116,8 +2122,8 @@ def in_ACCEPTFRIENDREQUEST(self, client, tags): # 3.1: verify the request exists + create the friendship + delete the request as ONE # atomic DB unit off the reactor; notify an online requester in the callback. d = self._root.defer_db(self.userdb.do_accept_friend_request, client.user_id, username) - d.addCallback(self._acceptfriend_done, client, username) - d.addErrback(self._friend_op_failed, client, "ACCEPTFRIENDREQUEST") + d.addCallback(client.with_msg_id(self._acceptfriend_done), client, username) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "ACCEPTFRIENDREQUEST") def _acceptfriend_done(self, verdict, client, username): if client.session_id not in self._root.clients: @@ -2141,8 +2147,8 @@ def in_DECLINEFRIENDREQUEST(self, client, tags): return # 3.1: verify + delete the request as ONE atomic DB unit off the reactor. d = self._root.defer_db(self.userdb.do_decline_friend_request, client.user_id, username) - d.addCallback(self._declinefriend_done, client) - d.addErrback(self._friend_op_failed, client, "DECLINEFRIENDREQUEST") + d.addCallback(client.with_msg_id(self._declinefriend_done), client) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "DECLINEFRIENDREQUEST") def _declinefriend_done(self, verdict, client): if client.session_id not in self._root.clients: @@ -2160,8 +2166,8 @@ def in_UNFRIEND(self, client, tags): # 3.1: resolve + delete the friendship (both directions) as ONE atomic DB unit off # the reactor; notify an online ex-friend in the callback. d = self._root.defer_db(self.userdb.do_unfriend, client.user_id, username) - d.addCallback(self._unfriend_done, client, username) - d.addErrback(self._friend_op_failed, client, "UNFRIEND") + d.addCallback(client.with_msg_id(self._unfriend_done), client, username) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "UNFRIEND") def _unfriend_done(self, verdict, client, username): if client.session_id not in self._root.clients: @@ -2188,8 +2194,8 @@ def _friend_op_failed(self, failure, client, op): def in_FRIENDREQUESTLIST(self, client): # 3.1: read off the reactor thread; worker returns plain [(user_id, msg), ...]. d = self._root.defer_db(self.userdb.get_friend_request_list, client.user_id) - d.addCallback(self._friendrequestlist_send, client) - d.addErrback(self._social_list_failed, client, "FRIENDREQUESTLIST") + d.addCallback(client.with_msg_id(self._friendrequestlist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "FRIENDREQUESTLIST") def _friendrequestlist_send(self, entries, client): # reactor-thread callback; bracket the session because clientFromID may read the DB. @@ -2215,8 +2221,8 @@ def _friendrequestlist_send(self, entries, client): def in_FRIENDLIST(self, client): # 3.1: read off the reactor thread; worker returns plain [user_id, ...]. d = self._root.defer_db(self.userdb.get_friend_user_ids, client.user_id) - d.addCallback(self._friendlist_send, client) - d.addErrback(self._social_list_failed, client, "FRIENDLIST") + d.addCallback(client.with_msg_id(self._friendlist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "FRIENDLIST") def _friendlist_send(self, userIds, client): # reactor-thread callback; bracket the session because clientFromID may read the DB. @@ -2817,8 +2823,8 @@ def in_GETCHANNELMESSAGES(self, client, chan, last_msg_id): # query (joins to User/BridgedUser), so the callback does NO reactor-side DB and # needs no session bracket; it just formats + sends. A pure read, so no new races. d = self._root.defer_db(self.userdb.get_channel_messages, client.user_id, channel.id, last_msg_id, self._channel_history_max_fetch) - d.addCallback(self._channelmessages_send, client, chan) - d.addErrback(self._channelmessages_failed, client) + d.addCallback(client.with_msg_id(self._channelmessages_send), client, chan) + d.addErrback(client.with_msg_id(self._channelmessages_failed), client) def _channelmessages_send(self, result, client, chan): # reactor-thread callback: pure send, no DB -> no commit/rollback/close bracket. @@ -3196,8 +3202,8 @@ def in_FINDIP(self, client, address): # plain (username, last_login) tuples. The online/offline check reads shared memory, so # it stays in the reactor callback. d = self._root.defer_db(self.userdb.do_find_ip, address) - d.addCallback(self._findip_done, client, address) - d.addErrback(self._findip_failed, client, address) + d.addCallback(client.with_msg_id(self._findip_done), client, address) + d.addErrback(client.with_msg_id(self._findip_failed), client, address) def _findip_done(self, results, client, address): if client.session_id not in self._root.clients: @@ -3231,8 +3237,8 @@ def in_GETIP(self, client, username): # 3.1: only the offline branch hits the DB (get_ip returns a plain string); defer it. # The online check above is pure memory and stays on the reactor. d = self._root.defer_db(self.userdb.get_ip, username) - d.addCallback(self._getip_done, client, username) - d.addErrback(self._getip_failed, client, username) + d.addCallback(client.with_msg_id(self._getip_done), client, username) + d.addErrback(client.with_msg_id(self._getip_failed), client, username) def _getip_done(self, ip, client, username): if client.session_id not in self._root.clients: @@ -3274,8 +3280,8 @@ def in_RENAMEACCOUNT(self, client, newname): # the old and new name, replies, and disconnects on success; it does no reactor-side DB, # so it needs no session bracket. d = self._root.defer_db(self.userdb.do_rename_account, client.username, newname) - d.addCallback(self._renameaccount_done, client, newname) - d.addErrback(self._renameaccount_failed, client, newname) + d.addCallback(client.with_msg_id(self._renameaccount_done), client, newname) + d.addErrback(client.with_msg_id(self._renameaccount_failed), client, newname) def _renameaccount_done(self, verdict, client, newname): if client.session_id not in self._root.clients: @@ -3315,8 +3321,8 @@ def in_CHANGEPASSWORD(self, client, cur_password, new_password): # as one atomic DB unit. The callback (on the reactor) invalidates the 1.2 user # cache and replies; it does no reactor-side DB, so it needs no session bracket. d = self._root.defer_db(self.userdb.do_change_password, client.username, cur_password, new_password) - d.addCallback(self._changepassword_done, client) - d.addErrback(self._changepassword_failed, client) + d.addCallback(client.with_msg_id(self._changepassword_done), client) + d.addErrback(client.with_msg_id(self._changepassword_failed), client) def _changepassword_done(self, verdict, client): if client.session_id not in self._root.clients: @@ -3628,8 +3634,8 @@ def in_LISTBANS(self, client): # 3.1: list_bans() returns a plain list of dicts, so it defers cleanly; the callback only # Sends, so it needs no session bracket. d = self._root.defer_db(self.bandb.list_bans) - d.addCallback(self._listbans_done, client) - d.addErrback(self._listbans_failed, client) + d.addCallback(client.with_msg_id(self._listbans_done), client) + d.addErrback(client.with_msg_id(self._listbans_failed), client) def _listbans_done(self, banlist, client): if client.session_id not in self._root.clients: @@ -3651,8 +3657,8 @@ def in_LISTBLACKLIST(self, client): # send the blacklist of domains for email verification # 3.1: list_blacklist() returns a plain list of dicts; defer it, callback only Sends. d = self._root.defer_db(self.bandb.list_blacklist) - d.addCallback(self._listblacklist_done, client) - d.addErrback(self._listblacklist_failed, client) + d.addCallback(client.with_msg_id(self._listblacklist_done), client) + d.addErrback(client.with_msg_id(self._listblacklist_failed), client) def _listblacklist_done(self, blacklist, client): if client.session_id not in self._root.clients: @@ -3707,8 +3713,8 @@ def in_LISTMODS(self, client): # 3.1: the access guard stays on the reactor; list_mods() returns plain (admins, mods) # strings, so the query defers and the callback only Sends. d = self._root.defer_db(self.userdb.list_mods) - d.addCallback(self._listmods_done, client) - d.addErrback(self._listmods_failed, client) + d.addCallback(client.with_msg_id(self._listmods_done), client) + d.addErrback(client.with_msg_id(self._listmods_failed), client) def _listmods_done(self, mods_tuple, client): if client.session_id not in self._root.clients: @@ -4025,8 +4031,8 @@ def in_CHANGEEMAIL(self, client, newmail, verification_code=""): # one atomic DB unit; the callback (on the reactor) sets the new email in memory, invalidates # the 1.2 cache and replies. It does no reactor-side DB, so it needs no session bracket. d = self._root.defer_db(self.userdb.do_change_email, client.username, newmail) - d.addCallback(self._changeemail_done, client, newmail) - d.addErrback(self._changeemail_failed, client, newmail) + d.addCallback(client.with_msg_id(self._changeemail_done), client, newmail) + d.addErrback(client.with_msg_id(self._changeemail_failed), client, newmail) def _changeemail_done(self, verdict, client, newmail): if client.session_id not in self._root.clients: @@ -4092,8 +4098,8 @@ def in_RESETPASSWORD(self, client, email, verification_code): uid = recover_client.user_id raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_set_password, uid, hashed) - d.addCallback(self._resetpassword_done, client, raw, uid) - d.addErrback(self._resetpassword_failed, client) + d.addCallback(client.with_msg_id(self._resetpassword_done), client, raw, uid) + d.addErrback(client.with_msg_id(self._resetpassword_failed), client) def _resetpassword_done(self, verdict, client, raw_password, uid): if client.session_id not in self._root.clients: @@ -4151,8 +4157,8 @@ def in_RESETUSERPASSWORD(self, client, username, newmail=None): uid = recover_client.user_id raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_set_password, uid, hashed, add_email) - d.addCallback(self._resetuserpassword_done, client, raw, uid) - d.addErrback(self._resetuserpassword_failed, client) + d.addCallback(client.with_msg_id(self._resetuserpassword_done), client, raw, uid) + d.addErrback(client.with_msg_id(self._resetuserpassword_failed), client) def _resetuserpassword_done(self, verdict, client, raw_password, uid): if client.session_id not in self._root.clients: @@ -4195,8 +4201,8 @@ def in_DELETEACCOUNT(self, client, username): # snapshot. The callback only invalidates the cache + Sends, so it needs no session bracket. _raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_scrub_account, delete_client.user_id, hashed) - d.addCallback(self._deleteaccount_done, client, delete_client.username) - d.addErrback(self._deleteaccount_failed, client, delete_client.username) + d.addCallback(client.with_msg_id(self._deleteaccount_done), client, delete_client.username) + d.addErrback(client.with_msg_id(self._deleteaccount_failed), client, delete_client.username) def _deleteaccount_done(self, verdict, client, username): if client.session_id not in self._root.clients: diff --git a/tests/integration/msgidtest.py b/tests/integration/msgidtest.py new file mode 100644 index 00000000..b4338efc --- /dev/null +++ b/tests/integration/msgidtest.py @@ -0,0 +1,138 @@ +""" +End-to-end tier: a command's #id is on its reply and on nothing else (issue #60). + +The id was stored on the Client and never cleared, so every later line to that client carried +it until the next command. Replies sent from a deferred DB callback had the opposite problem, +carrying the id of whatever command the client had sent since, or none. + + unrelated line after '#5 PING', a SAIDPRIVATE from another player carries no id + deferred reply '#7 GETIP ' then '#8 PING' in one write: the GETIP reply is #7 + deferred, no next '#9 LISTBANS' then a bare 'PING' in one write: the LISTBANS reply is #9 + errors and refusals '#10 NOTACOMMAND' is answered with #10 on both lines + +Run: activate the venv, then python3 tests/integration/msgidtest.py +""" +import os as _os, sys as _sys +_sys.path[:0] = [_os.path.join(_os.path.dirname(__file__), _os.pardir, _os.pardir), + _os.path.join(_os.path.dirname(__file__), _os.pardir)] +from testenv import DB_KWARGS, HOST, PORT +import socket, hashlib, base64, time, sys +import pymysql + +TAG = sys.argv[1] if len(sys.argv) > 1 else "m1" +PW = base64.b64encode(hashlib.md5(b"secretpw").digest()).decode() +MOD = "mid_%s_mod" % TAG +TALKER = "mid_%s_talk" % TAG +OFFLINE = "mid_%s_off" % TAG + +errors = [] +def check(cond, label): + if not cond: errors.append(label) + +def recv_until(s, substr, timeout=8): + s.settimeout(timeout); buf = b"" + try: + while substr.encode() not in buf: + chunk = s.recv(8192) + if not chunk: break + buf += chunk + except socket.timeout: pass + return buf.decode(errors="replace") + +def read_for(s, seconds): + s.settimeout(seconds); buf = b"" + try: + while True: + chunk = s.recv(8192) + if not chunk: break + buf += chunk + except socket.timeout: pass + return [ln for ln in buf.decode(errors="replace").splitlines() if ln.strip()] + +def connect(): + s = socket.create_connection((HOST, PORT)); recv_until(s, "\n"); return s + +def register_and_confirm(user): + s = connect() + s.sendall(("REGISTER %s %s\n" % (user, PW)).encode()); recv_until(s, "\n") + time.sleep(2.5) + s.sendall(("LOGIN %s %s 0 * TestClient\n" % (user, PW)).encode()); recv_until(s, "AGREEMENTEND") + s.sendall(b"CONFIRMAGREEMENT\n") + ok = "LOGININFOEND" in recv_until(s, "LOGININFOEND"); s.close(); return ok + +def login_keep(user): + s = connect() + s.sendall(("LOGIN %s %s 0 * TestClient\n" % (user, PW)).encode()) + resp = recv_until(s, "LOGININFOEND") + if "ACCEPTED" not in resp: + s.close(); return None + read_for(s, 0.5) # drain the rest of the login burst + return s + +def exchange(s, lines, seconds=1.5): + """Send several commands in one write, so the server reads them back to back.""" + s.sendall(("".join(ln + "\n" for ln in lines)).encode()) + got = read_for(s, seconds) + print("%-44r -> %r" % (lines, got)) + return got + + +for u in (MOD, TALKER, OFFLINE): + if not register_and_confirm(u): + print("ABORT: could not register %s" % u); sys.exit(1) + +conn = pymysql.connect(**DB_KWARGS); cur = conn.cursor() +cur.execute("UPDATE users SET access='mod' WHERE username=%s", (MOD,)) +cur.execute("SELECT last_ip FROM users WHERE username=%s", (OFFLINE,)); OFFLINE_IP = cur.fetchone()[0] +conn.close() + +mod = login_keep(MOD) +talker = login_keep(TALKER) +if not (mod and talker): + print("ABORT: test users could not log in"); sys.exit(1) +read_for(mod, 0.5) # the ADDUSER for TALKER, who logged in after MOD + +# the reply to an id'd command carries the id +got = exchange(mod, ["#5 PING"]) +check(got == ["#5 PONG"], "'#5 PING' should be answered '#5 PONG', got %r" % got) + +# a line that has nothing to do with that command does not +talker.sendall(("SAYPRIVATE %s hello there\n" % MOD).encode()) +got = read_for(mod, 1.0) +print("%-44r -> %r" % ("(SAYPRIVATE from %s)" % TALKER, got)) +check("SAIDPRIVATE %s hello there" % TALKER in got, + "a private message from another player should arrive with no id, got %r" % got) + +# a deferred reply carries the id of the command that asked, not of a later one +got = exchange(mod, ["#7 GETIP %s" % OFFLINE, "#8 PING"]) +check("#8 PONG" in got, "'#8 PING' should be answered '#8 PONG', got %r" % got) +check("#7 SERVERMSG <%s> was recently bound to %s" % (OFFLINE, OFFLINE_IP) in got, + "the deferred GETIP reply should carry #7, got %r" % got) + +# and still carries it when the next command had no id at all +got = exchange(mod, ["#9 LISTBANS", "PING"]) +check("PONG" in got, "a bare PING should be answered with a bare PONG, got %r" % got) +check("#9 SERVERMSG Banlist is empty" in got, + "the deferred LISTBANS reply should carry #9, got %r" % got) + +# refusals from _handle itself carry the id on both lines +got = exchange(mod, ["#10 NOTACOMMAND"]) +check(len(got) == 2 and all(ln.startswith("#10 ") for ln in got), + "both lines refusing '#10 NOTACOMMAND' should carry #10, got %r" % got) + +# after all of that, an unrelated line is still clean +talker.sendall(("SAYPRIVATE %s second\n" % MOD).encode()) +got = read_for(mod, 1.0) +print("%-44r -> %r" % ("(SAYPRIVATE from %s)" % TALKER, got)) +check("SAIDPRIVATE %s second" % TALKER in got, + "a later private message should still arrive with no id, got %r" % got) + +mod.close(); talker.close() + + +if errors: + print("FAIL (%d):" % len(errors)) + for e in errors: print(" -", e) + sys.exit(1) +print("RESULT: PASS") +sys.exit(0)