xref: /aosp_15_r20/external/scapy/scapy/layers/bluetooth.py (revision 7dc08ffc4802948ccbc861daaf1e81c405c2c4bd)

## This file is part of Scapy
## See http://www.secdev.org/projects/scapy for more informations
## Copyright (C) Philippe Biondi <[email protected]>
## Copyright (C) Mike Ryan <[email protected]>
## 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]