Source code for bravo.beta.packets

from collections import namedtuple

from construct import Struct, Container, Embed, Enum, MetaField
from construct import MetaArray, If, Switch, Const, Peek, Magic
from construct import OptionalGreedyRange, RepeatUntil
from construct import Flag, PascalString, Adapter
from construct import UBInt8, UBInt16, UBInt32, UBInt64
from construct import SBInt8, SBInt16, SBInt32
from construct import BFloat32, BFloat64
from construct import BitStruct, BitField
from construct import StringAdapter, LengthValueAdapter, Sequence
from construct import ConstructError

[docs]def IPacket(object): """ Interface for packets. """ def parse(buf, offset): """ Parse a packet out of the given buffer, starting at the given offset. If the parse is successful, returns a tuple of the parsed packet and the next packet offset in the buffer. If the parse fails due to insufficient data, returns a tuple of None and the amount of data required before the parse can be retried. Exceptions may be raised if the parser finds invalid data. """
[docs]def simple(name, fmt, *args): """ Make a customized namedtuple representing a simple, primitive packet. """ from struct import Struct s = Struct(fmt) @classmethod def parse(cls, buf, offset): if len(buf) >= s.size + offset: unpacked = s.unpack_from(buf, offset) return cls(*unpacked), s.size + offset else: return None, s.size - len(buf) def build(self): return s.pack(*self) methods = { "parse": parse, "build": build, } return type(name, (namedtuple(name, *args),), methods)
DUMP_ALL_PACKETS = False # Strings. # This one is a UCS2 string, which effectively decodes single writeChar() # invocations. We need to import the encoding for it first, though. from bravo.encodings import ucs2 from codecs import register register(ucs2) class DoubleAdapter(LengthValueAdapter): def _encode(self, obj, context): return len(obj) / 2, obj def AlphaString(name): return StringAdapter( DoubleAdapter( Sequence(name, UBInt16("length"), MetaField("data", lambda ctx: ctx["length"] * 2), ) ), encoding="ucs2", ) # Boolean converter. def Bool(*args, **kwargs): return Flag(*args, default=True, **kwargs) # Flying, position, and orientation, reused in several places. grounded = Struct("grounded", UBInt8("grounded")) position = Struct("position", BFloat64("x"), BFloat64("y"), BFloat64("stance"), BFloat64("z") ) orientation = Struct("orientation", BFloat32("rotation"), BFloat32("pitch")) # TODO: this must be replaced with 'slot' (see below) # Notchian item packing (slot data) items = Struct("items", SBInt16("primary"), If(lambda context: context["primary"] >= 0, Embed(Struct("item_information", UBInt8("count"), UBInt16("secondary"), Magic("\xff\xff"), )), ), ) Speed = namedtuple('speed', 'x y z') class Slot(object): def __init__(self, item_id=-1, count=1, damage=0, nbt=None): self.item_id = item_id self.count = count self.damage = damage # TODO: Implement packing/unpacking of gzipped NBT data self.nbt = nbt @classmethod def fromItem(cls, item, count): return cls(item[0], count, item[1]) @property def is_empty(self): return self.item_id == -1 def __len__(self): return 0 if self.nbt is None else len(self.nbt) def __repr__(self): from bravo.blocks import items if self.is_empty: return 'Slot()' elif len(self): return 'Slot(%s, count=%d, damage=%d, +nbt:%dB)' % ( str(items[self.item_id]), self.count, self.damage, len(self) ) else: return 'Slot(%s, count=%d, damage=%d)' % ( str(items[self.item_id]), self.count, self.damage ) def __eq__(self, other): return (self.item_id == other.item_id and self.count == other.count and self.damage == self.damage and self.nbt == self.nbt) class SlotAdapter(Adapter): def _decode(self, obj, context): if obj.item_id == -1: s = Slot(obj.item_id) else: s = Slot(obj.item_id, obj.count, obj.damage, obj.nbt) return s def _encode(self, obj, context): if not isinstance(obj, Slot): raise ConstructError('Slot object expected') if obj.is_empty: return Container(item_id=-1) else: return Container(item_id=obj.item_id, count=obj.count, damage=obj.damage, nbt_len=len(obj) if len(obj) else -1, nbt=obj.nbt) slot = SlotAdapter( Struct("slot", SBInt16("item_id"), If(lambda context: context["item_id"] >= 0, Embed(Struct("item_information", UBInt8("count"), UBInt16("damage"), SBInt16("nbt_len"), If(lambda context: context["nbt_len"] >= 0, MetaField("nbt", lambda ctx: ctx["nbt_len"]) ) )), ) ) ) Metadata = namedtuple("Metadata", "type value") metadata_types = ["byte", "short", "int", "float", "string", "slot", "coords"] # Metadata adaptor. class MetadataAdapter(Adapter): def _decode(self, obj, context): d = {} for m in d[] = Metadata(metadata_types[], m.value) return d def _encode(self, obj, context): c = Container(data=[], terminator=None) for k, v in obj.iteritems(): t, value = v d = Container( id=Container(type=metadata_types.index(t), key=k), value=value, peeked=None) if[-1].peeked = 127 else:, second=0), value=0, peeked=127)) return c # Metadata inner container. metadata_switch = { 0: UBInt8("value"), 1: UBInt16("value"), 2: UBInt32("value"), 3: BFloat32("value"), 4: AlphaString("value"), 5: slot, 6: Struct("coords", UBInt32("x"), UBInt32("y"), UBInt32("z"), ), } # Metadata subconstruct. metadata = MetadataAdapter( Struct("metadata", RepeatUntil(lambda obj, context: obj["peeked"] == 0x7f, Struct("data", BitStruct("id", BitField("type", 3), BitField("key", 5), ), Switch("value", lambda context: context["id"]["type"], metadata_switch), Peek(UBInt8("peeked")), ), ), Const(UBInt8("terminator"), 0x7f), ), ) # Build faces, used during dig and build. faces = { "noop": -1, "-y": 0, "+y": 1, "-z": 2, "+z": 3, "-x": 4, "+x": 5, } face = Enum(SBInt8("face"), **faces) # World dimension. dimensions = { "earth": 0, "sky": 1, "nether": 255, } dimension = Enum(UBInt8("dimension"), **dimensions) # Difficulty levels difficulties = { "peaceful": 0, "easy": 1, "normal": 2, "hard": 3, } difficulty = Enum(UBInt8("difficulty"), **difficulties) modes = { "survival": 0, "creative": 1, "adventure": 2, } mode = Enum(UBInt8("mode"), **modes) # Possible effects. # XXX these names aren't really canonized yet effect = Enum(UBInt8("effect"), move_fast=1, move_slow=2, dig_fast=3, dig_slow=4, damage_boost=5, heal=6, harm=7, jump=8, confusion=9, regenerate=10, resistance=11, fire_resistance=12, water_resistance=13, invisibility=14, blindness=15, night_vision=16, hunger=17, weakness=18, poison=19, wither=20, ) # The actual packet list. packets = { 0x00: Struct("ping", UBInt32("pid"), ), 0x01: Struct("login", # Player Entity ID (random number generated by the server) UBInt32("eid"), # default, flat, largeBiomes AlphaString("leveltype"), mode, dimension, difficulty, UBInt8("unused"), UBInt8("maxplayers"), ), 0x02: Struct("handshake", UBInt8("protocol"), AlphaString("username"), AlphaString("host"), UBInt32("port"), ), 0x03: Struct("chat", AlphaString("data"), ), 0x04: Struct("time", # Total Ticks UBInt64("timestamp"), # Time of day UBInt64("time"), ), 0x05: Struct("entity-equipment", UBInt32("eid"), UBInt16("slot"), Embed(items), ), 0x06: Struct("spawn", SBInt32("x"), SBInt32("y"), SBInt32("z"), ), 0x07: Struct("use", UBInt32("eid"), UBInt32("target"), UBInt8("button"), ), 0x08: Struct("health", BFloat32("hp"), UBInt16("fp"), BFloat32("saturation"), ), 0x09: Struct("respawn", dimension, difficulty, mode, UBInt16("height"), AlphaString("leveltype"), ), 0x0a: grounded, 0x0b: Struct("position", position, grounded ), 0x0c: Struct("orientation", orientation, grounded ), # TODO: Differ between client and server 'position' 0x0d: Struct("location", position, orientation, grounded ), 0x0e: Struct("digging", Enum(UBInt8("state"), started=0, cancelled=1, stopped=2, checked=3, dropped=4, # Also eating shooting=5, ), SBInt32("x"), UBInt8("y"), SBInt32("z"), face, ), 0x0f: Struct("build", SBInt32("x"), UBInt8("y"), SBInt32("z"), face, Embed(items), UBInt8("cursorx"), UBInt8("cursory"), UBInt8("cursorz"), ), # Hold Item Change 0x10: Struct("equip", # Only 0-8 UBInt16("slot"), ), 0x11: Struct("bed", UBInt32("eid"), UBInt8("unknown"), SBInt32("x"), UBInt8("y"), SBInt32("z"), ), 0x12: Struct("animate", UBInt32("eid"), Enum(UBInt8("animation"), noop=0, arm=1, hit=2, leave_bed=3, eat=5, unknown=102, crouch=104, uncrouch=105, ), ), 0x13: Struct("action", UBInt32("eid"), Enum(UBInt8("action"), crouch=1, uncrouch=2, leave_bed=3, start_sprint=4, stop_sprint=5, ), UBInt32("unknown"), ), 0x14: Struct("player", UBInt32("eid"), AlphaString("username"), SBInt32("x"), SBInt32("y"), SBInt32("z"), UBInt8("yaw"), UBInt8("pitch"), # 0 For none, unlike other packets # -1 crashes clients SBInt16("item"), metadata, ), 0x16: Struct("collect", UBInt32("eid"), UBInt32("destination"), ), # Object/Vehicle 0x17: Struct("object", # XXX: was 'vehicle'! UBInt32("eid"), Enum(UBInt8("type"), # See boat=1, item_stack=2, minecart=10, storage_cart=11, powered_cart=12, tnt=50, ender_crystal=51, arrow=60, snowball=61, egg=62, thrown_enderpearl=65, wither_skull=66, falling_block=70, frames=71, ender_eye=72, thrown_potion=73, dragon_egg=74, thrown_xp_bottle=75, fishing_float=90, ), SBInt32("x"), SBInt32("y"), SBInt32("z"), UBInt8("pitch"), UBInt8("yaw"), SBInt32("data"), # See If(lambda context: context["data"] != 0, Struct("speed", SBInt16("x"), SBInt16("y"), SBInt16("z"), ) ), ), 0x18: Struct("mob", UBInt32("eid"), Enum(UBInt8("type"), **{ "Creeper": 50, "Skeleton": 51, "Spider": 52, "GiantZombie": 53, "Zombie": 54, "Slime": 55, "Ghast": 56, "ZombiePig": 57, "Enderman": 58, "CaveSpider": 59, "Silverfish": 60, "Blaze": 61, "MagmaCube": 62, "EnderDragon": 63, "Wither": 64, "Bat": 65, "Witch": 66, "Pig": 90, "Sheep": 91, "Cow": 92, "Chicken": 93, "Squid": 94, "Wolf": 95, "Mooshroom": 96, "Snowman": 97, "Ocelot": 98, "IronGolem": 99, "Villager": 120 }), SBInt32("x"), SBInt32("y"), SBInt32("z"), SBInt8("yaw"), SBInt8("pitch"), SBInt8("head_yaw"), SBInt16("vx"), SBInt16("vy"), SBInt16("vz"), metadata, ), 0x19: Struct("painting", UBInt32("eid"), AlphaString("title"), SBInt32("x"), SBInt32("y"), SBInt32("z"), face, ), 0x1a: Struct("experience", UBInt32("eid"), SBInt32("x"), SBInt32("y"), SBInt32("z"), UBInt16("quantity"), ), 0x1b: Struct("steer", BFloat32("first"), BFloat32("second"), Bool("third"), Bool("fourth"), ), 0x1c: Struct("velocity", UBInt32("eid"), SBInt16("dx"), SBInt16("dy"), SBInt16("dz"), ), 0x1d: Struct("destroy", UBInt8("count"), MetaArray(lambda context: context["count"], UBInt32("eid")), ), 0x1e: Struct("create", UBInt32("eid"), ), 0x1f: Struct("entity-position", UBInt32("eid"), SBInt8("dx"), SBInt8("dy"), SBInt8("dz") ), 0x20: Struct("entity-orientation", UBInt32("eid"), UBInt8("yaw"), UBInt8("pitch") ), 0x21: Struct("entity-location", UBInt32("eid"), SBInt8("dx"), SBInt8("dy"), SBInt8("dz"), UBInt8("yaw"), UBInt8("pitch") ), 0x22: Struct("teleport", UBInt32("eid"), SBInt32("x"), SBInt32("y"), SBInt32("z"), UBInt8("yaw"), UBInt8("pitch"), ), 0x23: Struct("entity-head", UBInt32("eid"), UBInt8("yaw"), ), 0x26: Struct("status", UBInt32("eid"), Enum(UBInt8("status"), damaged=2, killed=3, taming=6, tamed=7, drying=8, eating=9, sheep_eat=10, golem_rose=11, heart_particle=12, angry_particle=13, happy_particle=14, magic_particle=15, shaking=16, firework=17, ), ), 0x27: Struct("attach", UBInt32("eid"), # XXX -1 for detatching UBInt32("vid"), UBInt8("unknown"), ), 0x28: Struct("metadata", UBInt32("eid"), metadata, ), 0x29: Struct("effect", UBInt32("eid"), effect, UBInt8("amount"), UBInt16("duration"), ), 0x2a: Struct("uneffect", UBInt32("eid"), effect, ), 0x2b: Struct("levelup", BFloat32("current"), UBInt16("level"), UBInt16("total"), ), # XXX 0x2c, server to client, needs to be implemented, needs special # UUID-packing techniques 0x33: Struct("chunk", SBInt32("x"), SBInt32("z"), Bool("continuous"), UBInt16("primary"), UBInt16("add"), PascalString("data", length_field=UBInt32("length"), encoding="zlib"), ), 0x34: Struct("batch", SBInt32("x"), SBInt32("z"), UBInt16("count"), PascalString("data", length_field=UBInt32("length")), ), 0x35: Struct("block", SBInt32("x"), UBInt8("y"), SBInt32("z"), UBInt16("type"), UBInt8("meta"), ), # XXX This covers general tile actions, not just note blocks. # TODO: Needs work 0x36: Struct("block-action", SBInt32("x"), SBInt16("y"), SBInt32("z"), UBInt8("byte1"), UBInt8("byte2"), UBInt16("blockid"), ), 0x37: Struct("block-break-anim", UBInt32("eid"), UBInt32("x"), UBInt32("y"), UBInt32("z"), UBInt8("stage"), ), # XXX Server -> Client. Use 0x33 instead. 0x38: Struct("bulk-chunk", UBInt16("count"), UBInt32("length"), UBInt8("sky_light"), MetaField("data", lambda ctx: ctx["length"]), MetaArray(lambda context: context["count"], Struct("metadata", UBInt32("chunk_x"), UBInt32("chunk_z"), UBInt16("bitmap_primary"), UBInt16("bitmap_secondary"), ) ) ), # TODO: Needs work? 0x3c: Struct("explosion", BFloat64("x"), BFloat64("y"), BFloat64("z"), BFloat32("radius"), UBInt32("count"), MetaField("blocks", lambda context: context["count"] * 3), BFloat32("motionx"), BFloat32("motiony"), BFloat32("motionz"), ), 0x3d: Struct("sound", Enum(UBInt32("sid"), click2=1000, click1=1001, bow_fire=1002, door_toggle=1003, extinguish=1004, record_play=1005, charge=1007, fireball=1008, zombie_wood=1010, zombie_metal=1011, zombie_break=1012, wither=1013, smoke=2000, block_break=2001, splash_potion=2002, ender_eye=2003, blaze=2004, ), SBInt32("x"), UBInt8("y"), SBInt32("z"), UBInt32("data"), Bool("volume-mod"), ), 0x3e: Struct("named-sound", AlphaString("name"), UBInt32("x"), UBInt32("y"), UBInt32("z"), BFloat32("volume"), UBInt8("pitch"), ), 0x3f: Struct("particle", AlphaString("name"), BFloat32("x"), BFloat32("y"), BFloat32("z"), BFloat32("x_offset"), BFloat32("y_offset"), BFloat32("z_offset"), BFloat32("speed"), UBInt32("count"), ), 0x46: Struct("state", Enum(UBInt8("state"), bad_bed=0, start_rain=1, stop_rain=2, mode_change=3, run_credits=4, ), mode, ), 0x47: Struct("thunderbolt", UBInt32("eid"), UBInt8("gid"), SBInt32("x"), SBInt32("y"), SBInt32("z"), ), 0x64: Struct("window-open", UBInt8("wid"), Enum(UBInt8("type"), chest=0, workbench=1, furnace=2, dispenser=3, enchatment_table=4, brewing_stand=5, npc_trade=6, beacon=7, anvil=8, hopper=9, ), AlphaString("title"), UBInt8("slots"), UBInt8("use_title"), # XXX iff type == 0xb (currently unknown) write an extra secret int # here. WTF? ), 0x65: Struct("window-close", UBInt8("wid"), ), 0x66: Struct("window-action", UBInt8("wid"), UBInt16("slot"), UBInt8("button"), UBInt16("token"), UBInt8("shift"), # TODO: rename to 'mode' Embed(items), ), 0x67: Struct("window-slot", UBInt8("wid"), UBInt16("slot"), Embed(items), ), 0x68: Struct("inventory", UBInt8("wid"), UBInt16("length"), MetaArray(lambda context: context["length"], items), ), 0x69: Struct("window-progress", UBInt8("wid"), UBInt16("bar"), UBInt16("progress"), ), 0x6a: Struct("window-token", UBInt8("wid"), UBInt16("token"), Bool("acknowledged"), ), 0x6b: Struct("window-creative", UBInt16("slot"), Embed(items), ), 0x6c: Struct("enchant", UBInt8("wid"), UBInt8("enchantment"), ), 0x82: Struct("sign", SBInt32("x"), UBInt16("y"), SBInt32("z"), AlphaString("line1"), AlphaString("line2"), AlphaString("line3"), AlphaString("line4"), ), 0x83: Struct("map", UBInt16("type"), UBInt16("itemid"), PascalString("data", length_field=UBInt16("length")), ), 0x84: Struct("tile-update", SBInt32("x"), UBInt16("y"), SBInt32("z"), UBInt8("action"), PascalString("nbt_data", length_field=UBInt16("length")), # gzipped ), 0x85: Struct("0x85", UBInt8("first"), UBInt32("second"), UBInt32("third"), UBInt32("fourth"), ), 0xc8: Struct("statistics", UBInt32("sid"), # XXX I should be an Enum! UBInt32("count"), ), 0xc9: Struct("players", AlphaString("name"), Bool("online"), UBInt16("ping"), ), 0xca: Struct("abilities", UBInt8("flags"), BFloat32("fly-speed"), BFloat32("walk-speed"), ), 0xcb: Struct("tab", AlphaString("autocomplete"), ), 0xcc: Struct("settings", AlphaString("locale"), UBInt8("distance"), UBInt8("chat"), difficulty, Bool("cape"), ), 0xcd: Struct("statuses", UBInt8("payload") ), 0xce: Struct("score_item", AlphaString("name"), AlphaString("value"), Enum(UBInt8("action"), create=0, remove=1, update=2, ), ), 0xcf: Struct("score_update", AlphaString("item_name"), UBInt8("remove"), If(lambda context: context["remove"] == 0, Embed(Struct("information", AlphaString("score_name"), UBInt32("value"), )) ), ), 0xd0: Struct("score_display", Enum(UBInt8("position"), as_list=0, sidebar=1, below_name=2, ), AlphaString("score_name"), ), 0xd1: Struct("teams", AlphaString("name"), Enum(UBInt8("mode"), team_created=0, team_removed=1, team_updates=2, players_added=3, players_removed=4, ), If(lambda context: context["mode"] in ("team_created", "team_updated"), Embed(Struct("team_info", AlphaString("team_name"), AlphaString("team_prefix"), AlphaString("team_suffix"), Enum(UBInt8("friendly_fire"), off=0, on=1, invisibles=2, ), )) ), If(lambda context: context["mode"] in ("team_created", "players_added", "players_removed"), Embed(Struct("players_info", UBInt16("count"), MetaArray(lambda context: context["count"], AlphaString("player_names")), )) ), ), 0xfa: Struct("plugin-message", AlphaString("channel"), PascalString("data", length_field=UBInt16("length")), ), 0xfc: Struct("key-response", PascalString("key", length_field=UBInt16("key-len")), PascalString("token", length_field=UBInt16("token-len")), ), 0xfd: Struct("key-request", AlphaString("server"), PascalString("key", length_field=UBInt16("key-len")), PascalString("token", length_field=UBInt16("token-len")), ), 0xfe: Struct("poll", Magic("\x01" # Poll packet constant "\xfa" # Followed by a plugin message "\x00\x0b" # Length of plugin channel name + u"MC|PingHost".encode("ucs2") # Plugin channel name ), PascalString("data", length_field=UBInt16("length")), ), # TODO: rename to 'kick' 0xff: Struct("error", AlphaString("message")), } packet_stream = Struct("packet_stream", OptionalGreedyRange( Struct("full_packet", UBInt8("header"), Switch("payload", lambda context: context["header"], packets), ), ), OptionalGreedyRange( UBInt8("leftovers"), ), )
[docs]def parse_packets(bytestream): """ Opportunistically parse out as many packets as possible from a raw bytestream. Returns a tuple containing a list of unpacked packet containers, and any leftover unparseable bytes. """ container = packet_stream.parse(bytestream) l = [(i.header, i.payload) for i in container.full_packet] leftovers = "".join(chr(i) for i in container.leftovers) if DUMP_ALL_PACKETS: for header, payload in l: print "Parsed packet 0x%.2x" % header print payload return l, leftovers
incremental_packet_stream = Struct("incremental_packet_stream", Struct("full_packet", UBInt8("header"), Switch("payload", lambda context: context["header"], packets), ), OptionalGreedyRange( UBInt8("leftovers"), ), )
[docs]def parse_packets_incrementally(bytestream): """ Parse out packets one-by-one, yielding a tuple of packet header and packet payload. This function returns a generator. This function will yield all valid packets in the bytestream up to the first invalid packet. :returns: a generator yielding tuples of headers and payloads """ while bytestream: parsed = incremental_packet_stream.parse(bytestream) header = parsed.full_packet.header payload = parsed.full_packet.payload bytestream = "".join(chr(i) for i in parsed.leftovers) yield header, payload
packets_by_name = dict((, k) for (k, v) in packets.iteritems())
[docs]def make_packet(packet, *args, **kwargs): """ Constructs a packet bytestream from a packet header and payload. The payload should be passed as keyword arguments. Additional containers or dictionaries to be added to the payload may be passed positionally, as well. """ if packet not in packets_by_name: print "Couldn't find packet name %s!" % packet return "" header = packets_by_name[packet] for arg in args: kwargs.update(dict(arg)) container = Container(**kwargs) if DUMP_ALL_PACKETS: print "Making packet <%s> (0x%.2x)" % (packet, header) print container payload = packets[header].build(container) return chr(header) + payload
[docs]def make_error_packet(message): """ Convenience method to generate an error packet bytestream. """ return make_packet("error", message=message)