| ## This file is part of Scapy |
| ## See http://www.secdev.org/projects/scapy for more informations |
| ## Copyright (C) Philippe Biondi <phil@secdev.org> |
| ## Copyright (C) Mike Ryan <mikeryan@lacklustre.net> |
| ## This program is published under a GPLv2 license |
| |
| """ |
| Bluetooth layers, sockets and send/receive functions. |
| """ |
| |
| import socket,struct,array |
| from ctypes import * |
| from select import select |
| |
| from scapy.config import conf |
| from scapy.data import DLT_BLUETOOTH_HCI_H4 |
| from scapy.packet import * |
| from scapy.fields import * |
| from scapy.supersocket import SuperSocket |
| from scapy.sendrecv import sndrcv |
| from scapy.data import MTU |
| from scapy.consts import WINDOWS |
| from scapy.error import warning, log_loading |
| |
| |
| ########## |
| # Fields # |
| ########## |
| |
| class XLEShortField(LEShortField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| class XLELongField(LEShortField): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<Q") |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| |
| class LEMACField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "6s") |
| def i2m(self, pkt, x): |
| if x is None: |
| return b"\0\0\0\0\0\0" |
| return mac2str(x)[::-1] |
| def m2i(self, pkt, x): |
| return str2mac(x[::-1]) |
| def any2i(self, pkt, x): |
| if isinstance(x, str) and len(x) is 6: |
| x = self.m2i(pkt, x) |
| return x |
| def i2repr(self, pkt, x): |
| x = self.i2h(pkt, x) |
| if self in conf.resolve: |
| x = conf.manufdb._resolve_MAC(x) |
| return x |
| def randval(self): |
| return RandMAC() |
| |
| _bluetooth_packet_types = { |
| 0: "Acknowledgement", |
| 1: "Command", |
| 2: "ACL Data", |
| 3: "Synchronous", |
| 4: "Event", |
| 5: "Reserve", |
| 14: "Vendor", |
| 15: "Link Control" |
| } |
| |
| class HCI_Hdr(Packet): |
| name = "HCI header" |
| fields_desc = [ ByteEnumField("type", 2, _bluetooth_packet_types) ] |
| |
| def mysummary(self): |
| return self.sprintf("HCI %type%") |
| |
| class HCI_ACL_Hdr(Packet): |
| name = "HCI ACL header" |
| fields_desc = [ BitField("handle",0,12), # TODO: Create and use LEBitField |
| BitField("PB",0,2), # They are recieved as a **combined** LE Short |
| BitField("BC",0,2), # Handle is 12 bits, eacg flag is 2 bits. |
| LEShortField("len",None), ] |
| |
| def pre_dissect(self, s): |
| # Recieve data as LE stored as |
| # .... 1111 0100 1100 = handle |
| # 1010 .... .... .... = flags |
| # And turn it into |
| # 1111 0100 1100 .... = handle |
| # .... .... .... 1010 = flags |
| hf = socket.ntohs(struct.unpack("!H", s[:2])[0]) |
| r = ((hf & 0x0fff) << 4) + (hf >> 12) |
| return struct.pack("!H", r) + s[2:] |
| |
| def post_dissect(self, s): |
| self.raw_packet_cache = None # Reset packet to allow post_build |
| return s |
| |
| def post_build(self, p, pay): |
| p += pay |
| if self.len is None: |
| p = p[:2] + struct.pack("<H", len(pay)) + p[4:] |
| # Reverse, opposite of pre_dissect |
| hf = struct.unpack("!H", p[:2])[0] |
| r = socket.ntohs(((hf & 0xf) << 12) + (hf >> 4)) |
| return struct.pack("!H", r) + p[2:] |
| |
| |
| class L2CAP_Hdr(Packet): |
| name = "L2CAP header" |
| fields_desc = [ LEShortField("len",None), |
| LEShortEnumField("cid",0,{1:"control", 4:"attribute"}),] |
| |
| def post_build(self, p, pay): |
| p += pay |
| if self.len is None: |
| p = struct.pack("<H", len(pay)) + p[2:] |
| return p |
| |
| |
| |
| class L2CAP_CmdHdr(Packet): |
| name = "L2CAP command header" |
| fields_desc = [ |
| ByteEnumField("code",8,{1:"rej",2:"conn_req",3:"conn_resp", |
| 4:"conf_req",5:"conf_resp",6:"disconn_req", |
| 7:"disconn_resp",8:"echo_req",9:"echo_resp", |
| 10:"info_req",11:"info_resp", 18:"conn_param_update_req", |
| 19:"conn_param_update_resp"}), |
| ByteField("id",0), |
| LEShortField("len",None) ] |
| def post_build(self, p, pay): |
| p += pay |
| if self.len is None: |
| p = p[:2] + struct.pack("<H", len(pay)) + p[4:] |
| return p |
| def answers(self, other): |
| if other.id == self.id: |
| if self.code == 1: |
| return 1 |
| if other.code in [2,4,6,8,10,18] and self.code == other.code+1: |
| if other.code == 8: |
| return 1 |
| return self.payload.answers(other.payload) |
| return 0 |
| |
| class L2CAP_ConnReq(Packet): |
| name = "L2CAP Conn Req" |
| fields_desc = [ LEShortEnumField("psm",0,{1:"SDP",3:"RFCOMM",5:"telephony control"}), |
| LEShortField("scid",0), |
| ] |
| |
| class L2CAP_ConnResp(Packet): |
| name = "L2CAP Conn Resp" |
| fields_desc = [ LEShortField("dcid",0), |
| LEShortField("scid",0), |
| LEShortEnumField("result",0,["success", "pend", "cr_bad_psm", "cr_sec_block", "cr_no_mem", "reserved","cr_inval_scid", "cr_scid_in_use"]), |
| LEShortEnumField("status",0,["no_info", "authen_pend", "author_pend", "reserved"]), |
| ] |
| def answers(self, other): |
| return isinstance(other, L2CAP_ConnReq) and self.dcid == other.scid |
| |
| class L2CAP_CmdRej(Packet): |
| name = "L2CAP Command Rej" |
| fields_desc = [ LEShortField("reason",0), |
| ] |
| |
| |
| class L2CAP_ConfReq(Packet): |
| name = "L2CAP Conf Req" |
| fields_desc = [ LEShortField("dcid",0), |
| LEShortField("flags",0), |
| ] |
| |
| class L2CAP_ConfResp(Packet): |
| name = "L2CAP Conf Resp" |
| fields_desc = [ LEShortField("scid",0), |
| LEShortField("flags",0), |
| LEShortEnumField("result",0,["success","unaccept","reject","unknown"]), |
| ] |
| def answers(self, other): |
| return isinstance(other, L2CAP_ConfReq) and self.scid == other.dcid |
| |
| |
| class L2CAP_DisconnReq(Packet): |
| name = "L2CAP Disconn Req" |
| fields_desc = [ LEShortField("dcid",0), |
| LEShortField("scid",0), ] |
| |
| class L2CAP_DisconnResp(Packet): |
| name = "L2CAP Disconn Resp" |
| fields_desc = [ LEShortField("dcid",0), |
| LEShortField("scid",0), ] |
| def answers(self, other): |
| return self.scid == other.scid |
| |
| |
| |
| class L2CAP_InfoReq(Packet): |
| name = "L2CAP Info Req" |
| fields_desc = [ LEShortEnumField("type",0,{1:"CL_MTU",2:"FEAT_MASK"}), |
| StrField("data","") |
| ] |
| |
| |
| class L2CAP_InfoResp(Packet): |
| name = "L2CAP Info Resp" |
| fields_desc = [ LEShortField("type",0), |
| LEShortEnumField("result",0,["success","not_supp"]), |
| StrField("data",""), ] |
| def answers(self, other): |
| return self.type == other.type |
| |
| |
| class L2CAP_Connection_Parameter_Update_Request(Packet): |
| name = "L2CAP Connection Parameter Update Request" |
| fields_desc = [ LEShortField("min_interval", 0), |
| LEShortField("max_interval", 0), |
| LEShortField("slave_latency", 0), |
| LEShortField("timeout_mult", 0), ] |
| |
| |
| class L2CAP_Connection_Parameter_Update_Response(Packet): |
| name = "L2CAP Connection Parameter Update Response" |
| fields_desc = [ LEShortField("move_result", 0), ] |
| |
| |
| class ATT_Hdr(Packet): |
| name = "ATT header" |
| fields_desc = [ XByteField("opcode", None), ] |
| |
| |
| class ATT_Error_Response(Packet): |
| name = "Error Response" |
| fields_desc = [ XByteField("request", 0), |
| LEShortField("handle", 0), |
| XByteField("ecode", 0), ] |
| |
| class ATT_Exchange_MTU_Request(Packet): |
| name = "Exchange MTU Request" |
| fields_desc = [ LEShortField("mtu", 0), ] |
| |
| class ATT_Exchange_MTU_Response(Packet): |
| name = "Exchange MTU Response" |
| fields_desc = [ LEShortField("mtu", 0), ] |
| |
| class ATT_Find_Information_Request(Packet): |
| name = "Find Information Request" |
| fields_desc = [ XLEShortField("start", 0x0000), |
| XLEShortField("end", 0xffff), ] |
| |
| class ATT_Find_Information_Response(Packet): |
| name = "Find Information Reponse" |
| fields_desc = [ XByteField("format", 1), |
| StrField("data", "") ] |
| |
| class ATT_Find_By_Type_Value_Request(Packet): |
| name = "Find By Type Value Request" |
| fields_desc = [ XLEShortField("start", 0x0001), |
| XLEShortField("end", 0xffff), |
| XLEShortField("uuid", None), |
| StrField("data", ""), ] |
| |
| class ATT_Find_By_Type_Value_Response(Packet): |
| name = "Find By Type Value Response" |
| fields_desc = [ StrField("handles", ""), ] |
| |
| class ATT_Read_By_Type_Request_128bit(Packet): |
| name = "Read By Type Request" |
| fields_desc = [ XLEShortField("start", 0x0001), |
| XLEShortField("end", 0xffff), |
| XLELongField("uuid1", None), |
| XLELongField("uuid2", None)] |
| @classmethod |
| def dispatch_hook(cls, _pkt=None, *args, **kargs): |
| if _pkt and len(_pkt) == 6: |
| return ATT_Read_By_Type_Request |
| return ATT_Read_By_Type_Request_128bit |
| |
| class ATT_Read_By_Type_Request(Packet): |
| name = "Read By Type Request" |
| fields_desc = [ XLEShortField("start", 0x0001), |
| XLEShortField("end", 0xffff), |
| XLEShortField("uuid", None)] |
| |
| class ATT_Read_By_Type_Response(Packet): |
| name = "Read By Type Response" |
| # fields_desc = [ FieldLenField("len", None, length_of="data", fmt="B"), |
| # StrLenField("data", "", length_from=lambda pkt:pkt.len), ] |
| fields_desc = [ StrField("data", "") ] |
| |
| class ATT_Read_Request(Packet): |
| name = "Read Request" |
| fields_desc = [ XLEShortField("gatt_handle", 0), ] |
| |
| class ATT_Read_Response(Packet): |
| name = "Read Response" |
| fields_desc = [ StrField("value", ""), ] |
| |
| class ATT_Read_By_Group_Type_Request(Packet): |
| name = "Read By Group Type Request" |
| fields_desc = [ XLEShortField("start", 0), |
| XLEShortField("end", 0xffff), |
| XLEShortField("uuid", 0), ] |
| |
| class ATT_Read_By_Group_Type_Response(Packet): |
| name = "Read By Group Type Response" |
| fields_desc = [ XByteField("length", 0), |
| StrField("data", ""), ] |
| |
| class ATT_Write_Request(Packet): |
| name = "Write Request" |
| fields_desc = [ XLEShortField("gatt_handle", 0), |
| StrField("data", ""), ] |
| |
| class ATT_Write_Command(Packet): |
| name = "Write Request" |
| fields_desc = [ XLEShortField("gatt_handle", 0), |
| StrField("data", ""), ] |
| |
| class ATT_Write_Response(Packet): |
| name = "Write Response" |
| fields_desc = [ ] |
| |
| class ATT_Handle_Value_Notification(Packet): |
| name = "Handle Value Notification" |
| fields_desc = [ XLEShortField("handle", 0), |
| StrField("value", ""), ] |
| |
| |
| class SM_Hdr(Packet): |
| name = "SM header" |
| fields_desc = [ ByteField("sm_command", None) ] |
| |
| |
| class SM_Pairing_Request(Packet): |
| name = "Pairing Request" |
| fields_desc = [ ByteEnumField("iocap", 3, {0:"DisplayOnly", 1:"DisplayYesNo", 2:"KeyboardOnly", 3:"NoInputNoOutput", 4:"KeyboardDisplay"}), |
| ByteEnumField("oob", 0, {0:"Not Present", 1:"Present (from remote device)"}), |
| BitField("authentication", 0, 8), |
| ByteField("max_key_size", 16), |
| ByteField("initiator_key_distribution", 0), |
| ByteField("responder_key_distribution", 0), ] |
| |
| class SM_Pairing_Response(Packet): |
| name = "Pairing Response" |
| fields_desc = [ ByteEnumField("iocap", 3, {0:"DisplayOnly", 1:"DisplayYesNo", 2:"KeyboardOnly", 3:"NoInputNoOutput", 4:"KeyboardDisplay"}), |
| ByteEnumField("oob", 0, {0:"Not Present", 1:"Present (from remote device)"}), |
| BitField("authentication", 0, 8), |
| ByteField("max_key_size", 16), |
| ByteField("initiator_key_distribution", 0), |
| ByteField("responder_key_distribution", 0), ] |
| |
| |
| class SM_Confirm(Packet): |
| name = "Pairing Confirm" |
| fields_desc = [ StrFixedLenField("confirm", b'\x00' * 16, 16) ] |
| |
| class SM_Random(Packet): |
| name = "Pairing Random" |
| fields_desc = [ StrFixedLenField("random", b'\x00' * 16, 16) ] |
| |
| class SM_Failed(Packet): |
| name = "Pairing Failed" |
| fields_desc = [ XByteField("reason", 0) ] |
| |
| class SM_Encryption_Information(Packet): |
| name = "Encryption Information" |
| fields_desc = [ StrFixedLenField("ltk", b"\x00" * 16, 16), ] |
| |
| class SM_Master_Identification(Packet): |
| name = "Master Identification" |
| fields_desc = [ XLEShortField("ediv", 0), |
| StrFixedLenField("rand", b'\x00' * 8, 8), ] |
| |
| class SM_Identity_Information(Packet): |
| name = "Identity Information" |
| fields_desc = [ StrFixedLenField("irk", b'\x00' * 16, 16), ] |
| |
| class SM_Identity_Address_Information(Packet): |
| name = "Identity Address Information" |
| fields_desc = [ ByteEnumField("atype", 0, {0:"public"}), |
| LEMACField("address", None), ] |
| |
| class SM_Signing_Information(Packet): |
| name = "Signing Information" |
| fields_desc = [ StrFixedLenField("csrk", b'\x00' * 16, 16), ] |
| |
| |
| class EIR_Hdr(Packet): |
| name = "EIR Header" |
| fields_desc = [ |
| LenField("len", None, fmt="B", adjust=lambda x: x+1), # Add bytes mark |
| ByteEnumField("type", 0, { |
| 0x01: "flags", |
| 0x02: "incomplete_list_16_bit_svc_uuids", |
| 0x03: "complete_list_16_bit_svc_uuids", |
| 0x04: "incomplete_list_32_bit_svc_uuids", |
| 0x05: "complete_list_32_bit_svc_uuids", |
| 0x06: "incomplete_list_128_bit_svc_uuids", |
| 0x07: "complete_list_128_bit_svc_uuids", |
| 0x08: "shortened_local_name", |
| 0x09: "complete_local_name", |
| 0x0a: "tx_power_level", |
| 0x0d: "class_of_device", |
| 0x0e: "simple_pairing_hash", |
| 0x0f: "simple_pairing_rand", |
| 0x10: "sec_mgr_tk", |
| 0x11: "sec_mgr_oob_flags", |
| 0x12: "slave_conn_intvl_range", |
| 0x17: "pub_target_addr", |
| 0x18: "rand_target_addr", |
| 0x19: "appearance", |
| 0x1a: "adv_intvl", |
| 0x1b: "le_addr", |
| 0x1c: "le_role", |
| 0x14: "list_16_bit_svc_sollication_uuids", |
| 0x1f: "list_32_bit_svc_sollication_uuids", |
| 0x15: "list_128_bit_svc_sollication_uuids", |
| 0x16: "svc_data_16_bit_uuid", |
| 0x20: "svc_data_32_bit_uuid", |
| 0x21: "svc_data_128_bit_uuid", |
| 0x22: "sec_conn_confirm", |
| 0x22: "sec_conn_rand", |
| 0x24: "uri", |
| 0xff: "mfg_specific_data", |
| }), |
| ] |
| |
| def mysummary(self): |
| return self.sprintf("EIR %type%") |
| |
| class EIR_Element(Packet): |
| name = "EIR Element" |
| |
| def extract_padding(self, s): |
| # Needed to end each EIR_Element packet and make PacketListField work. |
| return '', s |
| |
| @staticmethod |
| def length_from(pkt): |
| if not pkt.underlayer: |
| warning("Missing an upper-layer") |
| return 0 |
| # 'type' byte is included in the length, so substract 1: |
| return pkt.underlayer.len - 1 |
| |
| class EIR_Raw(EIR_Element): |
| name = "EIR Raw" |
| fields_desc = [ |
| StrLenField("data", "", length_from=EIR_Element.length_from) |
| ] |
| |
| class EIR_Flags(EIR_Element): |
| name = "Flags" |
| fields_desc = [ |
| FlagsField("flags", 0x2, 8, |
| ["limited_disc_mode", "general_disc_mode", |
| "br_edr_not_supported", "simul_le_br_edr_ctrl", |
| "simul_le_br_edr_host"] + 3*["reserved"]) |
| ] |
| |
| class EIR_CompleteList16BitServiceUUIDs(EIR_Element): |
| name = "Complete list of 16-bit service UUIDs" |
| fields_desc = [ |
| FieldListField("svc_uuids", None, XLEShortField("uuid", 0), |
| length_from=EIR_Element.length_from) |
| ] |
| |
| class EIR_IncompleteList16BitServiceUUIDs(EIR_CompleteList16BitServiceUUIDs): |
| name = "Incomplete list of 16-bit service UUIDs" |
| |
| class EIR_CompleteLocalName(EIR_Element): |
| name = "Complete Local Name" |
| fields_desc = [ |
| StrLenField("local_name", "", length_from=EIR_Element.length_from) |
| ] |
| |
| class EIR_ShortenedLocalName(EIR_CompleteLocalName): |
| name = "Shortened Local Name" |
| |
| class EIR_TX_Power_Level(EIR_Element): |
| name = "TX Power Level" |
| fields_desc = [SignedByteField("level", 0)] |
| |
| class EIR_Manufacturer_Specific_Data(EIR_Element): |
| name = "EIR Manufacturer Specific Data" |
| fields_desc = [ |
| XLEShortField("company_id", 0), |
| StrLenField("data", "", |
| length_from=lambda pkt: EIR_Element.length_from(pkt) - 2) |
| ] |
| |
| |
| class HCI_Command_Hdr(Packet): |
| name = "HCI Command header" |
| fields_desc = [ XLEShortField("opcode", 0), |
| ByteField("len", None), ] |
| |
| def post_build(self, p, pay): |
| p += pay |
| if self.len is None: |
| p = p[:2] + struct.pack("B", len(pay)) + p[3:] |
| return p |
| |
| class HCI_Cmd_Reset(Packet): |
| name = "Reset" |
| |
| class HCI_Cmd_Set_Event_Filter(Packet): |
| name = "Set Event Filter" |
| fields_desc = [ ByteEnumField("type", 0, {0:"clear"}), ] |
| |
| class HCI_Cmd_Connect_Accept_Timeout(Packet): |
| name = "Connection Attempt Timeout" |
| fields_desc = [ LEShortField("timeout", 32000) ] # 32000 slots is 20000 msec |
| |
| class HCI_Cmd_LE_Host_Supported(Packet): |
| name = "LE Host Supported" |
| fields_desc = [ ByteField("supported", 1), |
| ByteField("simultaneous", 1), ] |
| |
| class HCI_Cmd_Set_Event_Mask(Packet): |
| name = "Set Event Mask" |
| fields_desc = [ StrFixedLenField("mask", b"\xff\xff\xfb\xff\x07\xf8\xbf\x3d", 8) ] |
| |
| class HCI_Cmd_Read_BD_Addr(Packet): |
| name = "Read BD Addr" |
| |
| |
| class HCI_Cmd_LE_Set_Scan_Parameters(Packet): |
| name = "LE Set Scan Parameters" |
| fields_desc = [ ByteEnumField("type", 1, {1:"active"}), |
| XLEShortField("interval", 16), |
| XLEShortField("window", 16), |
| ByteEnumField("atype", 0, {0:"public"}), |
| ByteEnumField("policy", 0, {0:"all"}), ] |
| |
| class HCI_Cmd_LE_Set_Scan_Enable(Packet): |
| name = "LE Set Scan Enable" |
| fields_desc = [ ByteField("enable", 1), |
| ByteField("filter_dups", 1), ] |
| |
| class HCI_Cmd_Disconnect(Packet): |
| name = "Disconnect" |
| fields_desc = [ XLEShortField("handle", 0), |
| ByteField("reason", 0x13), ] |
| |
| class HCI_Cmd_LE_Create_Connection(Packet): |
| name = "LE Create Connection" |
| fields_desc = [ LEShortField("interval", 96), |
| LEShortField("window", 48), |
| ByteEnumField("filter", 0, {0:"address"}), |
| ByteEnumField("patype", 0, {0:"public", 1:"random"}), |
| LEMACField("paddr", None), |
| ByteEnumField("atype", 0, {0:"public", 1:"random"}), |
| LEShortField("min_interval", 40), |
| LEShortField("max_interval", 56), |
| LEShortField("latency", 0), |
| LEShortField("timeout", 42), |
| LEShortField("min_ce", 0), |
| LEShortField("max_ce", 0), ] |
| |
| class HCI_Cmd_LE_Create_Connection_Cancel(Packet): |
| name = "LE Create Connection Cancel" |
| |
| class HCI_Cmd_LE_Connection_Update(Packet): |
| name = "LE Connection Update" |
| fields_desc = [ XLEShortField("handle", 0), |
| XLEShortField("min_interval", 0), |
| XLEShortField("max_interval", 0), |
| XLEShortField("latency", 0), |
| XLEShortField("timeout", 0), |
| LEShortField("min_ce", 0), |
| LEShortField("max_ce", 0xffff), ] |
| |
| class HCI_Cmd_LE_Read_Buffer_Size(Packet): |
| name = "LE Read Buffer Size" |
| |
| class HCI_Cmd_LE_Set_Random_Address(Packet): |
| name = "LE Set Random Address" |
| fields_desc = [ LEMACField("address", None) ] |
| |
| class HCI_Cmd_LE_Set_Advertising_Parameters(Packet): |
| name = "LE Set Advertising Parameters" |
| fields_desc = [ LEShortField("interval_min", 0x0800), |
| LEShortField("interval_max", 0x0800), |
| ByteEnumField("adv_type", 0, {0:"ADV_IND", 1:"ADV_DIRECT_IND", 2:"ADV_SCAN_IND", 3:"ADV_NONCONN_IND", 4:"ADV_DIRECT_IND_LOW"}), |
| ByteEnumField("oatype", 0, {0:"public", 1:"random"}), |
| ByteEnumField("datype", 0, {0:"public", 1:"random"}), |
| LEMACField("daddr", None), |
| ByteField("channel_map", 7), |
| ByteEnumField("filter_policy", 0, {0:"all:all", 1:"connect:all scan:whitelist", 2:"connect:whitelist scan:all", 3:"all:whitelist"}), ] |
| |
| class HCI_Cmd_LE_Set_Advertising_Data(Packet): |
| name = "LE Set Advertising Data" |
| fields_desc = [ FieldLenField("len", None, length_of="data", fmt="B"), |
| StrLenField("data", "", length_from=lambda pkt:pkt.len), ] |
| |
| class HCI_Cmd_LE_Set_Advertise_Enable(Packet): |
| name = "LE Set Advertise Enable" |
| fields_desc = [ ByteField("enable", 0) ] |
| |
| class HCI_Cmd_LE_Start_Encryption_Request(Packet): |
| name = "LE Start Encryption" |
| fields_desc = [ LEShortField("handle", 0), |
| StrFixedLenField("rand", None, 8), |
| XLEShortField("ediv", 0), |
| StrFixedLenField("ltk", b'\x00' * 16, 16), ] |
| |
| class HCI_Cmd_LE_Long_Term_Key_Request_Negative_Reply(Packet): |
| name = "LE Long Term Key Request Negative Reply" |
| fields_desc = [ LEShortField("handle", 0), ] |
| |
| class HCI_Cmd_LE_Long_Term_Key_Request_Reply(Packet): |
| name = "LE Long Term Key Request Reply" |
| fields_desc = [ LEShortField("handle", 0), |
| StrFixedLenField("ltk", b'\x00' * 16, 16), ] |
| |
| class HCI_Event_Hdr(Packet): |
| name = "HCI Event header" |
| fields_desc = [ XByteField("code", 0), |
| ByteField("length", 0), ] |
| |
| |
| class HCI_Event_Disconnection_Complete(Packet): |
| name = "Disconnection Complete" |
| fields_desc = [ ByteEnumField("status", 0, {0:"success"}), |
| LEShortField("handle", 0), |
| XByteField("reason", 0), ] |
| |
| |
| class HCI_Event_Encryption_Change(Packet): |
| name = "Encryption Change" |
| fields_desc = [ ByteEnumField("status", 0, {0:"change has occurred"}), |
| LEShortField("handle", 0), |
| ByteEnumField("enabled", 0, {0:"OFF", 1:"ON (LE)", 2:"ON (BR/EDR)"}), ] |
| |
| class HCI_Event_Command_Complete(Packet): |
| name = "Command Complete" |
| fields_desc = [ ByteField("number", 0), |
| XLEShortField("opcode", 0), |
| ByteEnumField("status", 0, {0:"success"}), ] |
| |
| |
| class HCI_Cmd_Complete_Read_BD_Addr(Packet): |
| name = "Read BD Addr" |
| fields_desc = [ LEMACField("addr", None), ] |
| |
| |
| |
| class HCI_Event_Command_Status(Packet): |
| name = "Command Status" |
| fields_desc = [ ByteEnumField("status", 0, {0:"pending"}), |
| ByteField("number", 0), |
| XLEShortField("opcode", None), ] |
| |
| class HCI_Event_Number_Of_Completed_Packets(Packet): |
| name = "Number Of Completed Packets" |
| fields_desc = [ ByteField("number", 0) ] |
| |
| class HCI_Event_LE_Meta(Packet): |
| name = "LE Meta" |
| fields_desc = [ ByteEnumField("event", 0, {2:"advertising_report"}) ] |
| |
| class HCI_LE_Meta_Connection_Complete(Packet): |
| name = "Connection Complete" |
| fields_desc = [ ByteEnumField("status", 0, {0:"success"}), |
| LEShortField("handle", 0), |
| ByteEnumField("role", 0, {0:"master"}), |
| ByteEnumField("patype", 0, {0:"public", 1:"random"}), |
| LEMACField("paddr", None), |
| LEShortField("interval", 54), |
| LEShortField("latency", 0), |
| LEShortField("supervision", 42), |
| XByteField("clock_latency", 5), ] |
| |
| class HCI_LE_Meta_Connection_Update_Complete(Packet): |
| name = "Connection Update Complete" |
| fields_desc = [ ByteEnumField("status", 0, {0:"success"}), |
| LEShortField("handle", 0), |
| LEShortField("interval", 54), |
| LEShortField("latency", 0), |
| LEShortField("timeout", 42), ] |
| |
| class HCI_LE_Meta_Advertising_Report(Packet): |
| name = "Advertising Report" |
| fields_desc = [ ByteField("number", 0), |
| ByteEnumField("type", 0, {0:"conn_und", 4:"scan_rsp"}), |
| ByteEnumField("atype", 0, {0:"public", 1:"random"}), |
| LEMACField("addr", None), |
| FieldLenField("len", None, length_of="data", fmt="B"), |
| PacketListField("data", [], EIR_Hdr, |
| length_from=lambda pkt:pkt.len), |
| SignedByteField("rssi", 0)] |
| |
| |
| class HCI_LE_Meta_Long_Term_Key_Request(Packet): |
| name = "Long Term Key Request" |
| fields_desc = [ LEShortField("handle", 0), |
| StrFixedLenField("rand", None, 8), |
| XLEShortField("ediv", 0), ] |
| |
| |
| bind_layers( HCI_Hdr, HCI_Command_Hdr, type=1) |
| bind_layers( HCI_Hdr, HCI_ACL_Hdr, type=2) |
| bind_layers( HCI_Hdr, HCI_Event_Hdr, type=4) |
| bind_layers( HCI_Hdr, conf.raw_layer, ) |
| |
| conf.l2types.register(DLT_BLUETOOTH_HCI_H4, HCI_Hdr) |
| |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Reset, opcode=0x0c03) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Set_Event_Mask, opcode=0x0c01) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Set_Event_Filter, opcode=0x0c05) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Connect_Accept_Timeout, opcode=0x0c16) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Host_Supported, opcode=0x0c6d) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Read_BD_Addr, opcode=0x1009) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Read_Buffer_Size, opcode=0x2002) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Random_Address, opcode=0x2005) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Advertising_Parameters, opcode=0x2006) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Advertising_Data, opcode=0x2008) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Advertise_Enable, opcode=0x200a) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Scan_Parameters, opcode=0x200b) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Set_Scan_Enable, opcode=0x200c) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_Disconnect, opcode=0x406) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Create_Connection, opcode=0x200d) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Create_Connection_Cancel, opcode=0x200e) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Connection_Update, opcode=0x2013) |
| |
| |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Start_Encryption_Request, opcode=0x2019) |
| |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Long_Term_Key_Request_Reply, opcode=0x201a) |
| bind_layers( HCI_Command_Hdr, HCI_Cmd_LE_Long_Term_Key_Request_Negative_Reply, opcode=0x201b) |
| |
| bind_layers( HCI_Event_Hdr, HCI_Event_Disconnection_Complete, code=0x5) |
| bind_layers( HCI_Event_Hdr, HCI_Event_Encryption_Change, code=0x8) |
| bind_layers( HCI_Event_Hdr, HCI_Event_Command_Complete, code=0xe) |
| bind_layers( HCI_Event_Hdr, HCI_Event_Command_Status, code=0xf) |
| bind_layers( HCI_Event_Hdr, HCI_Event_Number_Of_Completed_Packets, code=0x13) |
| bind_layers( HCI_Event_Hdr, HCI_Event_LE_Meta, code=0x3e) |
| |
| bind_layers( HCI_Event_Command_Complete, HCI_Cmd_Complete_Read_BD_Addr, opcode=0x1009) |
| |
| bind_layers( HCI_Event_LE_Meta, HCI_LE_Meta_Connection_Complete, event=1) |
| bind_layers( HCI_Event_LE_Meta, HCI_LE_Meta_Advertising_Report, event=2) |
| bind_layers( HCI_Event_LE_Meta, HCI_LE_Meta_Connection_Update_Complete, event=3) |
| bind_layers( HCI_Event_LE_Meta, HCI_LE_Meta_Long_Term_Key_Request, event=5) |
| |
| bind_layers(EIR_Hdr, EIR_Flags, type=0x01) |
| bind_layers(EIR_Hdr, EIR_IncompleteList16BitServiceUUIDs, type=0x02) |
| bind_layers(EIR_Hdr, EIR_CompleteList16BitServiceUUIDs, type=0x03) |
| bind_layers(EIR_Hdr, EIR_ShortenedLocalName, type=0x08) |
| bind_layers(EIR_Hdr, EIR_CompleteLocalName, type=0x09) |
| bind_layers(EIR_Hdr, EIR_TX_Power_Level, type=0x0a) |
| bind_layers(EIR_Hdr, EIR_Manufacturer_Specific_Data, type=0xff) |
| bind_layers(EIR_Hdr, EIR_Raw) |
| |
| bind_layers( HCI_ACL_Hdr, L2CAP_Hdr, ) |
| bind_layers( L2CAP_Hdr, L2CAP_CmdHdr, cid=1) |
| bind_layers( L2CAP_Hdr, L2CAP_CmdHdr, cid=5) #LE L2CAP Signaling Channel |
| bind_layers( L2CAP_CmdHdr, L2CAP_CmdRej, code=1) |
| bind_layers( L2CAP_CmdHdr, L2CAP_ConnReq, code=2) |
| bind_layers( L2CAP_CmdHdr, L2CAP_ConnResp, code=3) |
| bind_layers( L2CAP_CmdHdr, L2CAP_ConfReq, code=4) |
| bind_layers( L2CAP_CmdHdr, L2CAP_ConfResp, code=5) |
| bind_layers( L2CAP_CmdHdr, L2CAP_DisconnReq, code=6) |
| bind_layers( L2CAP_CmdHdr, L2CAP_DisconnResp, code=7) |
| bind_layers( L2CAP_CmdHdr, L2CAP_InfoReq, code=10) |
| bind_layers( L2CAP_CmdHdr, L2CAP_InfoResp, code=11) |
| bind_layers( L2CAP_CmdHdr, L2CAP_Connection_Parameter_Update_Request, code=18) |
| bind_layers( L2CAP_CmdHdr, L2CAP_Connection_Parameter_Update_Response, code=19) |
| bind_layers( L2CAP_Hdr, ATT_Hdr, cid=4) |
| bind_layers( ATT_Hdr, ATT_Error_Response, opcode=0x1) |
| bind_layers( ATT_Hdr, ATT_Exchange_MTU_Request, opcode=0x2) |
| bind_layers( ATT_Hdr, ATT_Exchange_MTU_Response, opcode=0x3) |
| bind_layers( ATT_Hdr, ATT_Find_Information_Request, opcode=0x4) |
| bind_layers( ATT_Hdr, ATT_Find_Information_Response, opcode=0x5) |
| bind_layers( ATT_Hdr, ATT_Find_By_Type_Value_Request, opcode=0x6) |
| bind_layers( ATT_Hdr, ATT_Find_By_Type_Value_Response, opcode=0x7) |
| bind_layers( ATT_Hdr, ATT_Read_By_Type_Request_128bit, opcode=0x8) |
| bind_layers( ATT_Hdr, ATT_Read_By_Type_Request, opcode=0x8) |
| bind_layers( ATT_Hdr, ATT_Read_By_Type_Response, opcode=0x9) |
| bind_layers( ATT_Hdr, ATT_Read_Request, opcode=0xa) |
| bind_layers( ATT_Hdr, ATT_Read_Response, opcode=0xb) |
| bind_layers( ATT_Hdr, ATT_Read_By_Group_Type_Request, opcode=0x10) |
| bind_layers( ATT_Hdr, ATT_Read_By_Group_Type_Response, opcode=0x11) |
| bind_layers( ATT_Hdr, ATT_Write_Request, opcode=0x12) |
| bind_layers( ATT_Hdr, ATT_Write_Response, opcode=0x13) |
| bind_layers( ATT_Hdr, ATT_Write_Command, opcode=0x52) |
| bind_layers( ATT_Hdr, ATT_Handle_Value_Notification, opcode=0x1b) |
| bind_layers( L2CAP_Hdr, SM_Hdr, cid=6) |
| bind_layers( SM_Hdr, SM_Pairing_Request, sm_command=1) |
| bind_layers( SM_Hdr, SM_Pairing_Response, sm_command=2) |
| bind_layers( SM_Hdr, SM_Confirm, sm_command=3) |
| bind_layers( SM_Hdr, SM_Random, sm_command=4) |
| bind_layers( SM_Hdr, SM_Failed, sm_command=5) |
| bind_layers( SM_Hdr, SM_Encryption_Information, sm_command=6) |
| bind_layers( SM_Hdr, SM_Master_Identification, sm_command=7) |
| bind_layers( SM_Hdr, SM_Identity_Information, sm_command=8) |
| bind_layers( SM_Hdr, SM_Identity_Address_Information, sm_command=9) |
| bind_layers( SM_Hdr, SM_Signing_Information, sm_command=0x0a) |
| |
| class BluetoothSocketError(BaseException): |
| pass |
| |
| class BluetoothCommandError(BaseException): |
| pass |
| |
| class BluetoothL2CAPSocket(SuperSocket): |
| desc = "read/write packets on a connected L2CAP socket" |
| def __init__(self, bt_address): |
| if WINDOWS: |
| warning("Not available on Windows") |
| return |
| s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, |
| socket.BTPROTO_L2CAP) |
| s.connect((bt_address,0)) |
| self.ins = self.outs = s |
| |
| def recv(self, x=MTU): |
| return L2CAP_CmdHdr(self.ins.recv(x)) |
| |
| class BluetoothRFCommSocket(BluetoothL2CAPSocket): |
| """read/write packets on a connected RFCOMM socket""" |
| def __init__(self, bt_address, port=0): |
| s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, |
| socket.BTPROTO_RFCOMM) |
| s.connect((bt_address,port)) |
| self.ins = self.outs = s |
| |
| class BluetoothHCISocket(SuperSocket): |
| desc = "read/write on a BlueTooth HCI socket" |
| def __init__(self, iface=0x10000, type=None): |
| if WINDOWS: |
| warning("Not available on Windows") |
| return |
| s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) |
| s.setsockopt(socket.SOL_HCI, socket.HCI_DATA_DIR,1) |
| s.setsockopt(socket.SOL_HCI, socket.HCI_TIME_STAMP,1) |
| s.setsockopt(socket.SOL_HCI, socket.HCI_FILTER, struct.pack("IIIh2x", 0xffffffff,0xffffffff,0xffffffff,0)) #type mask, event mask, event mask, opcode |
| s.bind((iface,)) |
| self.ins = self.outs = s |
| # s.connect((peer,0)) |
| |
| |
| def recv(self, x): |
| return HCI_Hdr(self.ins.recv(x)) |
| |
| class sockaddr_hci(Structure): |
| _fields_ = [ |
| ("sin_family", c_ushort), |
| ("hci_dev", c_ushort), |
| ("hci_channel", c_ushort), |
| ] |
| |
| class BluetoothUserSocket(SuperSocket): |
| desc = "read/write H4 over a Bluetooth user channel" |
| def __init__(self, adapter_index=0): |
| if WINDOWS: |
| warning("Not available on Windows") |
| return |
| # s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) |
| # s.bind((0,1)) |
| |
| # yeah, if only |
| # thanks to Python's weak ass socket and bind implementations, we have |
| # to call down into libc with ctypes |
| |
| sockaddr_hcip = POINTER(sockaddr_hci) |
| cdll.LoadLibrary("libc.so.6") |
| libc = CDLL("libc.so.6") |
| |
| socket_c = libc.socket |
| socket_c.argtypes = (c_int, c_int, c_int); |
| socket_c.restype = c_int |
| |
| bind = libc.bind |
| bind.argtypes = (c_int, POINTER(sockaddr_hci), c_int) |
| bind.restype = c_int |
| |
| ######## |
| ## actual code |
| |
| s = socket_c(31, 3, 1) # (AF_BLUETOOTH, SOCK_RAW, HCI_CHANNEL_USER) |
| if s < 0: |
| raise BluetoothSocketError("Unable to open PF_BLUETOOTH socket") |
| |
| sa = sockaddr_hci() |
| sa.sin_family = 31 # AF_BLUETOOTH |
| sa.hci_dev = adapter_index # adapter index |
| sa.hci_channel = 1 # HCI_USER_CHANNEL |
| |
| r = bind(s, sockaddr_hcip(sa), sizeof(sa)) |
| if r != 0: |
| raise BluetoothSocketError("Unable to bind") |
| |
| self.ins = self.outs = socket.fromfd(s, 31, 3, 1) |
| |
| def send_command(self, cmd): |
| opcode = cmd.opcode |
| self.send(cmd) |
| while True: |
| r = self.recv() |
| if r.type == 0x04 and r.code == 0xe and r.opcode == opcode: |
| if r.status != 0: |
| raise BluetoothCommandError("Command %x failed with %x" % (opcode, r.status)) |
| return r |
| |
| def recv(self, x=512): |
| return HCI_Hdr(self.ins.recv(x)) |
| |
| def readable(self, timeout=0): |
| (ins, outs, foo) = select([self.ins], [], [], timeout) |
| return len(ins) > 0 |
| |
| def flush(self): |
| while self.readable(): |
| self.recv() |
| |
| conf.BTsocket = BluetoothRFCommSocket |
| |
| ## Bluetooth |
| |
| @conf.commands.register |
| def srbt(bt_address, pkts, inter=0.1, *args, **kargs): |
| """send and receive using a bluetooth socket""" |
| if "port" in kargs.keys(): |
| s = conf.BTsocket(bt_address=bt_address, port=kargs.pop("port")) |
| else: |
| s = conf.BTsocket(bt_address=bt_address) |
| a,b = sndrcv(s,pkts,inter=inter,*args,**kargs) |
| s.close() |
| return a,b |
| |
| @conf.commands.register |
| def srbt1(bt_address, pkts, *args, **kargs): |
| """send and receive 1 packet using a bluetooth socket""" |
| a,b = srbt(bt_address, pkts, *args, **kargs) |
| if len(a) > 0: |
| return a[0][1] |
