from bluetooth_packets_python3 import l2cap_packets
from bluetooth_packets_python3.l2cap_packets import CommandCode, LeCommandCode
from cert.capture import Capture
+from cert.matchers import HciMatchers
from cert.matchers import L2capMatchers
from cert.matchers import SecurityMatchers
from security.facade_pb2 import UiMsgType
@staticmethod
def ReadBdAddrCompleteCapture():
return Capture(
- lambda packet: packet.event[0:5] == b'\x0e\x0a\x01\x09\x10', lambda packet: hci_packets.ReadBdAddrCompleteView(
- hci_packets.CommandCompleteView(
- hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet.event))))))
+ HciMatchers.CommandComplete(hci_packets.OpCode.READ_BD_ADDR),
+ lambda packet: hci_packets.ReadBdAddrCompleteView(HciMatchers.ExtractMatchingCommandComplete(packet.event, hci_packets.OpCode.READ_BD_ADDR)))
@staticmethod
def ConnectionRequestCapture():
return Capture(
- lambda packet: packet.event[0:2] == b'\x04\x0a', lambda packet: hci_packets.ConnectionRequestView(
- hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet.event)))))
+ HciMatchers.EventWithCode(hci_packets.EventCode.CONNECTION_REQUEST),
+ lambda packet: hci_packets.ConnectionRequestView(
+ HciMatchers.ExtractEventWithCode(packet.event, hci_packets.EventCode.CONNECTION_REQUEST)))
@staticmethod
def ConnectionCompleteCapture():
return Capture(
- lambda packet: packet.event[0:3] == b'\x03\x0b\x00', lambda packet: hci_packets.ConnectionCompleteView(
- hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet.event)))))
+ HciMatchers.EventWithCode(hci_packets.EventCode.CONNECTION_COMPLETE),
+ lambda packet: hci_packets.ConnectionCompleteView(
+ HciMatchers.ExtractEventWithCode(packet.event, hci_packets.EventCode.CONNECTION_COMPLETE)))
@staticmethod
def DisconnectionCompleteCapture():
return Capture(
- lambda packet: packet.event[0:2] == b'\x05\x04', lambda packet: hci_packets.DisconnectionCompleteView(
- hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet.event)))))
+ HciMatchers.EventWithCode(hci_packets.EventCode.DISCONNECTION_COMPLETE),
+ lambda packet: hci_packets.DisconnectionCompleteView(
+ HciMatchers.ExtractEventWithCode(packet.event, hci_packets.EventCode.DISCONNECTION_COMPLETE)))
@staticmethod
def LeConnectionCompleteCapture():
- return Capture(
- lambda packet: packet.event[0] == 0x3e and (packet.event[2] == 0x01 or packet.event[2] == 0x0a),
- lambda packet: hci_packets.LeConnectionCompleteView(
- hci_packets.LeMetaEventView(
- hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet.event))))))
+ return Capture(HciMatchers.LeConnectionComplete(),
+ lambda packet: HciMatchers.ExtractLeConnectionComplete(packet.event))
class L2capCaptures(object):
class HciMatchers(object):
@staticmethod
- def CommandComplete(opcode=None, num_complete=1):
- return lambda msg: HciMatchers._is_matching_command_complete(msg.event, opcode, num_complete)
+ def CommandComplete(opcode=None):
+ return lambda msg: HciMatchers._is_matching_command_complete(msg.event, opcode)
@staticmethod
- def _is_matching_command_complete(packet_bytes, opcode=None, num_complete=1):
- hci_event = HciMatchers.extract_hci_event_with_code(packet_bytes, EventCode.COMMAND_COMPLETE)
- if hci_event is None:
- return False
- frame = hci_packets.CommandCompleteView(hci_event)
- return (opcode is None or frame.GetCommandOpCode() == opcode) and\
- frame.GetNumHciCommandPackets() == num_complete
+ def ExtractMatchingCommandComplete(packet_bytes, opcode=None):
+ return HciMatchers._extract_matching_command_complete(packet_bytes, opcode)
+
+ @staticmethod
+ def _is_matching_command_complete(packet_bytes, opcode=None):
+ return HciMatchers._extract_matching_command_complete(packet_bytes, opcode) is not None
+
+ @staticmethod
+ def _extract_matching_command_complete(packet_bytes, opcode=None):
+ event = HciMatchers._extract_matching_event(packet_bytes, EventCode.COMMAND_COMPLETE)
+ if event is None:
+ return None
+ complete = hci_packets.CommandCompleteView(event)
+ if opcode is None or complete is None:
+ return complete
+ else:
+ if complete.GetCommandOpCode() != opcode:
+ return None
+ else:
+ return complete
@staticmethod
def EventWithCode(event_code):
- return lambda msg: HciMatchers.extract_hci_event_with_code(msg.event, event_code)
+ return lambda msg: HciMatchers._is_matching_event(msg.event, event_code)
@staticmethod
- def extract_hci_event_with_code(packet_bytes, event_code=None):
- hci_event = hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet_bytes)))
- if hci_event is None:
+ def ExtractEventWithCode(packet_bytes, event_code):
+ return HciMatchers._extract_matching_event(packet_bytes, event_code)
+
+ @staticmethod
+ def _is_matching_event(packet_bytes, event_code):
+ return HciMatchers._extract_matching_event(packet_bytes, event_code) is not None
+
+ @staticmethod
+ def _extract_matching_event(packet_bytes, event_code):
+ event = hci_packets.EventPacketView(bt_packets.PacketViewLittleEndian(list(packet_bytes)))
+ if event is None:
return None
- if event_code is not None and hci_event.GetEventCode() != event_code:
+ if event_code is not None and event.GetEventCode() != event_code:
return None
- return hci_event
+ return event
+
+ @staticmethod
+ def LeEventWithCode(subevent_code):
+ return lambda msg: HciMatchers._extract_matching_le_event(msg.event, subevent_code) is not None
+
+ @staticmethod
+ def ExtractLeEventWithCode(packet_bytes, subevent_code):
+ return HciMatchers._extract_matching_le_event(packet_bytes, subevent_code)
+
+ @staticmethod
+ def _extract_matching_le_event(packet_bytes, subevent_code):
+ event = hci_packets.LeMetaEventView(
+ HciMatchers._extract_matching_event(packet_bytes, hci_packets.EventCode.LE_META_EVENT))
+ if event is None:
+ return None
+ if event.GetSubeventCode() != subevent_code:
+ return None
+ return event
+
+ @staticmethod
+ def LeConnectionComplete():
+ return lambda msg: HciMatchers._extract_le_connection_complete(msg.event) is not None
+
+ @staticmethod
+ def ExtractLeConnectionComplete(packet_bytes):
+ return HciMatchers._extract_le_connection_complete(packet_bytes)
+
+ @staticmethod
+ def _extract_le_connection_complete(packet_bytes):
+ event = hci_packets.LeConnectionCompleteView(
+ HciMatchers._extract_matching_le_event(packet_bytes, hci_packets.SubeventCode.CONNECTION_COMPLETE))
+ if event is not None:
+ return event
+ return hci_packets.LeEnhancedConnectionCompleteView(
+ HciMatchers._extract_matching_le_event(packet_bytes, hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE))
@staticmethod
def LinkKeyRequest():
@staticmethod
def _is_matching_inquiry_result(packet, address):
- hci_event = HciMatchers.extract_hci_event_with_code(packet, EventCode.INQUIRY_RESULT)
+ hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.INQUIRY_RESULT)
if hci_event is None:
return False
inquiry_view = hci_packets.InquiryResultView(hci_event)
@staticmethod
def _is_matching_inquiry_result_with_rssi(packet, address):
- hci_event = HciMatchers.extract_hci_event_with_code(packet, EventCode.INQUIRY_RESULT_WITH_RSSI)
+ hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.INQUIRY_RESULT_WITH_RSSI)
if hci_event is None:
return False
inquiry_view = hci_packets.InquiryResultWithRssiView(hci_event)
@staticmethod
def _is_matching_extended_inquiry_result(packet, address):
- hci_event = HciMatchers.extract_hci_event_with_code(packet, EventCode.EXTENDED_INQUIRY_RESULT)
+ hci_event = HciMatchers.ExtractEventWithCode(packet, EventCode.EXTENDED_INQUIRY_RESULT)
if hci_event is None:
return False
extended_view = hci_packets.ExtendedInquiryResultView(hci_event)