#!/usr/bin/env python3 import socket import sys import time from signal import signal, SIGINT from typing import Optional import random from twnet_parser.packet import TwPacket, parse7 from twnet_parser.messages7.control.token import CtrlToken from twnet_parser.messages7.control.connect import CtrlConnect from twnet_parser.messages7.control.close import CtrlClose from twnet_parser.messages7.control.keep_alive import CtrlKeepAlive from twnet_parser.messages7.system.info import MsgInfo from twnet_parser.messages7.system.map_change import MsgMapChange from twnet_parser.messages7.system.ready import MsgReady from twnet_parser.messages7.game.cl_start_info import MsgClStartInfo from twnet_parser.messages7.system.enter_game import MsgEnterGame from twnet_parser.messages7.game.cl_call_vote import MsgClCallVote from twnet_parser.messages7.game.cl_kill import MsgClKill from twnet_parser.messages7.game.cl_emoticon import MsgClEmoticon from twnet_parser.messages7.game.cl_say import MsgClSay from twnet_parser.constants import NET_MAX_PACKETSIZE, NET_MAX_SEQUENCE user_list = {} bot_info = {"id": "None"} class TeeworldsClient(): def __init__(self, name, clan, host, port): self.name = name self.clan = clan self.host = host self.port = port self.dest_srv = (self.host, self.port) self.log(f"Connecting to {host}:{port} ...") self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(('', 0)) self.outfile = None self.downloaded_chunks = 0 self.downloaded_bytes = 0 self.sent_vital = 0 self.got_vital = 0 self.got_seqs = set() self.map_info: Optional[MsgMapChange] = None self.my_token = b'\xff\xaa\xbb\xee' self.srv_token = b'\xff\xff\xff\xff' self.last_send_time = time.time() # TODO: we should be able to set this # ctrl_token = CtrlToken(we_are_a_client = True) ctrl_token = CtrlToken() ctrl_token.response_token = self.my_token self.send_msg(ctrl_token) def log(self, message): print(f"[{self.name}] {message}") def send_msg(self, messages): if not isinstance(messages, list): messages = [messages] packet = TwPacket() packet.header.token = self.srv_token for msg in messages: if hasattr(msg, 'header'): if msg.header.flags.vital: self.sent_vital += 1 msg.header.seq = self.sent_vital packet.messages.append(msg) self.last_send_time = time.time() packet.header.ack = self.got_vital self.sock.sendto(packet.pack(), self.dest_srv) def tick(self): data, addr = self.sock.recvfrom(NET_MAX_PACKETSIZE) packet = parse7(data) for msg in packet.messages: if msg.message_type == "game" and msg.message_name == "sv_client_info" and msg.client_id != -1: user_list[msg.client_id] = msg if hasattr(msg, 'header'): if msg.header.flags.vital and not msg.header.flags.resend: self.got_vital = (self.got_vital + 1) % NET_MAX_SEQUENCE if msg.header.seq in self.got_seqs: continue self.got_seqs.add(msg.header.seq) if msg.message_name == 'sv_client_info': if msg.local: self.log(f"got client id: {msg.client_id}") bot_info["id"] = msg.client_id elif msg.message_name == 'token': self.srv_token = msg.response_token ctrl_connect = CtrlConnect() ctrl_connect.response_token = self.my_token self.send_msg(ctrl_connect) elif msg.message_name == 'accept': info = MsgInfo() info.header.flags.vital = True self.send_msg(info) elif msg.message_name == 'map_change' or msg.message_name == 'map_data': ready = MsgReady() ready.header.flags.vital = True self.send_msg(ready) elif msg.message_name == 'con_ready': self.log("sending info") startinfo = MsgClStartInfo(clan = self.clan, name=self.name) startinfo.header.flags.vital = True self.send_msg(startinfo) enter_game = MsgEnterGame() enter_game.header.flags.vital = True self.send_msg(enter_game) elif msg.message_name == 'close': self.log(f"disconnected reason='{msg.reason}'") exit(1) for msg in packet.messages: if msg.message_name != 'sv_chat': continue if msg.client_id == -1: continue if msg.client_id == bot_info['id']: continue if msg.message == "Do you know someone who uses a bot? Please report them to the moderators.": continue print(msg) if msg.message.find('!test') != -1: print(1234) msg = MsgClSay(message = msg.message, target = -1) msg.header.flags.vital = True msg.header.size = None self.send_msg(msg) elif msg.message.find('!kill') != -1: kill = MsgClKill() kill.header.flags.vital = True self.send_msg(kill) elif msg.message.find('!crash') != -1: close = CtrlClose() client.log("sending disconnect") client.send_msg(close) exit(0) #msgg = MsgClSay(message = f"{bot_info['id']}") #msgg.header.flags.vital = True #self.send_msg(msgg) #kill = MsgClKill() #kill.header.flags.vital = True #self.send_msg(kill) #time.sleep(0.5) #msgg = MsgClSay(message = '12345') #msgg.header.flags.vital = True #self.send_msg(msgg) if (time.time() - self.last_send_time) > 0.1: self.send_msg(CtrlKeepAlive()) num_clients = 1 if 3 > 3: num_clients = int(1) clients = [] for i in range(0, num_clients): client = TeeworldsClient(name = "Mыш", clan="1", host = "127.0.0.1", port = int(8303)) clients.append(client) def handler(signal_received, frame): global clients print('SIGINT or CTRL-C detected. Exiting gracefully') for client in clients: close = CtrlClose() client.log("sending disconnect") client.send_msg(close) exit(0) signal(SIGINT, handler) while True: time.sleep(0.1) for client in clients: client.tick()