Add 0.6.5 system & game message support

Also add a test for sys info
Stills needs more tests. At least one game message.
This commit is contained in:
ChillerDragon 2023-05-24 09:28:11 +02:00
parent 01331e8141
commit 1fa282b192
3 changed files with 97 additions and 6 deletions

View file

@ -0,0 +1,68 @@
from twnet_parser.packet import parse6, TwPacket, PacketHeader6
from twnet_parser.messages6.system.info import MsgInfo
def test_unpack_info():
"""
Sent by a vanilla 0.6.5 client
copied from a real traffic capture
https://chillerdragon.github.io/teeworlds-protocol/06/system_messages.html#NETMSG_INFO
"""
data = \
b'\x08\x00\x01' \
b'\x9c\x6b\xcb\xda' \
b'\x41\x07\x01\x03\x30\x2e\x36' \
b'\x20\x36\x32\x36\x66\x63\x65' \
b'\x39\x61\x37\x37\x38\x64\x66' \
b'\x34\x64\x34\x00\x00'
packet = parse6(data)
assert packet.version == '0.6'
assert packet.header.ack == 0
assert packet.header.num_chunks == 1
assert packet.header.flags.token is True
assert packet.header.flags.control is False
assert packet.header.flags.connless is False
assert packet.header.flags.resend is False
assert packet.header.flags.compression is False
assert len(packet.messages) == 1
msg: MsgInfo = packet.messages[0]
assert msg.system_message is True
assert msg.message_name == 'info'
assert msg.version == '0.6 626fce9a778df4d4'
assert msg.password == ''
repack = packet.pack()
assert repack == data
def test_pack_info():
packet: TwPacket = TwPacket()
packet.header = PacketHeader6()
msg: MsgInfo = MsgInfo()
msg.version = '0.6 626fce9a778df4d4'
msg.password = ''
packet.messages.append(msg)
data: bytes = packet.pack()
packet2: TwPacket = parse6(data)
assert packet2.version == '0.6'
assert packet2.header.ack == 0
assert packet2.header.num_chunks == 1
assert packet2.header.flags.token is False
assert packet2.header.flags.control is False
assert packet2.header.flags.connless is False
assert packet2.header.flags.resend is False
assert packet2.header.flags.compression is False
assert len(packet.messages) == 1
msg2: MsgInfo = packet2.messages[0]
assert msg2.system_message is True
assert msg2.message_name == 'info'
assert msg2.version == '0.6 626fce9a778df4d4'
assert msg2.password == ''
assert packet2.pack() == data

View file

@ -1,15 +1,34 @@
from typing import Literal
from twnet_parser.net_message import NetMessage
from twnet_parser.msg_matcher.game7 import match_game7
from twnet_parser.msg_matcher.system7 import match_system7
from twnet_parser.msg_matcher.game6 import match_game6
from twnet_parser.msg_matcher.system6 import match_system6
# could also be named ChunkParser
class MessageParser():
# the first byte of data has to be the
# first byte of a message PAYLOAD
# NOT the whole packet with packet header
# and NOT the whole message with chunk header
def parse_game_message(self, msg_id: int, data: bytes) -> NetMessage:
def parse_game_message(
self,
version: Literal['0.6', '0.7'],
msg_id: int,
data: bytes
) -> NetMessage:
if version == '0.6':
return match_game6(msg_id, data)
return match_game7(msg_id, data)
def parse_sys_message(self, msg_id: int, data: bytes) -> NetMessage:
def parse_sys_message(
self,
version: Literal['0.6', '0.7'],
msg_id: int,
data: bytes
) -> NetMessage:
if version == '0.6':
return match_system6(msg_id, data)
return match_system7(msg_id, data)

View file

@ -1,6 +1,6 @@
#!/usr/bin/env python
from typing import Union, cast, Optional
from typing import Union, cast, Optional, Literal
from twnet_parser.packer import Unpacker, pack_int
from twnet_parser.pretty_print import PrettyPrint
@ -340,6 +340,9 @@ class ChunkHeaderParser:
return header
class PacketParser():
def __init__(self) -> None:
self.version: Literal['0.6', '0.7'] = '0.7'
# the first byte of data has to be the
# first byte of a message chunk
# NOT the whole packet with packet header
@ -371,15 +374,16 @@ class PacketParser():
msg_id >>= 1
msg: NetMessage
if sys:
msg = MessageParser().parse_sys_message(msg_id, unpacker.get_raw())
msg = MessageParser().parse_sys_message(self.version, msg_id, unpacker.get_raw())
else:
msg = MessageParser().parse_game_message(msg_id, unpacker.get_raw())
msg = MessageParser().parse_game_message(self.version, msg_id, unpacker.get_raw())
msg.header = chunk_header
return msg
def parse6(self, data: bytes, client: bool) -> TwPacket:
pck = TwPacket()
pck.version = '0.6'
self.version = '0.6'
pck.version = self.version
# TODO: what is the most performant way in python to do this?
# heap allocating a PacketHeaderParser7 just to bundle a bunch of
# methods that do not share state seems like a waste of performance