|
| 1 | +# Test BLE GAP unpair functionality |
| 2 | +# Tests the new gap_unpair method added to MicroPython |
| 3 | +# gap_unpair expects a key from _IRQ_GET_SECRET/_IRQ_SET_SECRET events |
| 4 | + |
| 5 | +from micropython import const |
| 6 | +import time, machine, bluetooth |
| 7 | + |
| 8 | +if not hasattr(bluetooth.BLE, "gap_unpair"): |
| 9 | + print("SKIP") |
| 10 | + raise SystemExit |
| 11 | + |
| 12 | +TIMEOUT_MS = 4000 |
| 13 | + |
| 14 | +_IRQ_CENTRAL_CONNECT = const(1) |
| 15 | +_IRQ_CENTRAL_DISCONNECT = const(2) |
| 16 | +_IRQ_GATTS_READ_REQUEST = const(4) |
| 17 | +_IRQ_PERIPHERAL_CONNECT = const(7) |
| 18 | +_IRQ_PERIPHERAL_DISCONNECT = const(8) |
| 19 | +_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11) |
| 20 | +_IRQ_GATTC_CHARACTERISTIC_DONE = const(12) |
| 21 | +_IRQ_GATTC_READ_RESULT = const(15) |
| 22 | +_IRQ_ENCRYPTION_UPDATE = const(28) |
| 23 | +_IRQ_GET_SECRET = const(29) |
| 24 | +_IRQ_SET_SECRET = const(30) |
| 25 | + |
| 26 | +_FLAG_READ = const(0x0002) |
| 27 | +_FLAG_READ_ENCRYPTED = const(0x0200) |
| 28 | + |
| 29 | +SERVICE_UUID = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A") |
| 30 | +CHAR_UUID = bluetooth.UUID("00000000-1111-2222-3333-444444444444") |
| 31 | +CHAR = (CHAR_UUID, _FLAG_READ | _FLAG_READ_ENCRYPTED) |
| 32 | +SERVICE = (SERVICE_UUID, (CHAR,)) |
| 33 | + |
| 34 | +waiting_events = {} |
| 35 | +bond_keys = [] # Store bond keys for unpair testing |
| 36 | + |
| 37 | + |
| 38 | +def irq(event, data): |
| 39 | + if event == _IRQ_CENTRAL_CONNECT: |
| 40 | + print("_IRQ_CENTRAL_CONNECT") |
| 41 | + waiting_events[event] = data[0] |
| 42 | + elif event == _IRQ_CENTRAL_DISCONNECT: |
| 43 | + print("_IRQ_CENTRAL_DISCONNECT") |
| 44 | + elif event == _IRQ_GATTS_READ_REQUEST: |
| 45 | + print("_IRQ_GATTS_READ_REQUEST") |
| 46 | + elif event == _IRQ_PERIPHERAL_CONNECT: |
| 47 | + print("_IRQ_PERIPHERAL_CONNECT") |
| 48 | + waiting_events[event] = data[0] |
| 49 | + elif event == _IRQ_PERIPHERAL_DISCONNECT: |
| 50 | + print("_IRQ_PERIPHERAL_DISCONNECT") |
| 51 | + elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT: |
| 52 | + if data[-1] == CHAR_UUID: |
| 53 | + print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1]) |
| 54 | + waiting_events[event] = data[2] |
| 55 | + else: |
| 56 | + return |
| 57 | + elif event == _IRQ_GATTC_CHARACTERISTIC_DONE: |
| 58 | + print("_IRQ_GATTC_CHARACTERISTIC_DONE") |
| 59 | + elif event == _IRQ_GATTC_READ_RESULT: |
| 60 | + print("_IRQ_GATTC_READ_RESULT", bytes(data[-1])) |
| 61 | + elif event == _IRQ_ENCRYPTION_UPDATE: |
| 62 | + print("_IRQ_ENCRYPTION_UPDATE", data[1], data[2], data[3]) |
| 63 | + elif event == _IRQ_GET_SECRET: |
| 64 | + print("_IRQ_GET_SECRET", "key:", data[1]) |
| 65 | + bond_keys.append(data[1]) # Store the key for unpair testing |
| 66 | + elif event == _IRQ_SET_SECRET: |
| 67 | + print("_IRQ_SET_SECRET", "key:", data[1]) |
| 68 | + bond_keys.append(data[1]) # Store the key for unpair testing |
| 69 | + |
| 70 | + if event not in waiting_events: |
| 71 | + waiting_events[event] = None |
| 72 | + |
| 73 | + |
| 74 | +def wait_for_event(event, timeout_ms): |
| 75 | + t0 = time.ticks_ms() |
| 76 | + while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms: |
| 77 | + if event in waiting_events: |
| 78 | + return waiting_events.pop(event) |
| 79 | + machine.idle() |
| 80 | + raise ValueError("Timeout waiting for {}".format(event)) |
| 81 | + |
| 82 | + |
| 83 | +# Acting in peripheral role. |
| 84 | +def instance0(): |
| 85 | + multitest.globals(BDADDR=ble.config("mac")) |
| 86 | + ((char_handle,),) = ble.gatts_register_services((SERVICE,)) |
| 87 | + ble.gatts_write(char_handle, "encrypted") |
| 88 | + print("gap_advertise") |
| 89 | + ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY") |
| 90 | + multitest.next() |
| 91 | + try: |
| 92 | + # Wait for central to connect. |
| 93 | + wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS) |
| 94 | + |
| 95 | + # Wait for pairing event. |
| 96 | + wait_for_event(_IRQ_ENCRYPTION_UPDATE, TIMEOUT_MS) |
| 97 | + |
| 98 | + # Wait for GATTS read request. |
| 99 | + wait_for_event(_IRQ_GATTS_READ_REQUEST, TIMEOUT_MS) |
| 100 | + |
| 101 | + multitest.next() |
| 102 | + |
| 103 | + # Wait for central to disconnect after initial pairing. |
| 104 | + wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS) |
| 105 | + |
| 106 | + # Test gap_unpair functionality |
| 107 | + print("gap_unpair_test") |
| 108 | + print("bond_keys_captured:", len(bond_keys)) |
| 109 | + |
| 110 | + # Test gap_unpair with captured bond keys |
| 111 | + if bond_keys: |
| 112 | + for i, key in enumerate(bond_keys): |
| 113 | + try: |
| 114 | + result = ble.gap_unpair(key) |
| 115 | + print(f"gap_unpair_key_{i}_result:", result) |
| 116 | + except Exception as e: |
| 117 | + print(f"gap_unpair_key_{i}_error:", type(e).__name__, str(e)) |
| 118 | + else: |
| 119 | + print("gap_unpair_no_keys_captured") |
| 120 | + |
| 121 | + # Test unpair with non-existent key |
| 122 | + fake_key = b'\x01\x12\x34\x56\x78\x9a\xbc\xde\xf0\x11\x22\x33\x44\x55\x66\x77' |
| 123 | + try: |
| 124 | + result = ble.gap_unpair(fake_key) |
| 125 | + print("gap_unpair_fake_key_result:", result) |
| 126 | + except Exception as e: |
| 127 | + print("gap_unpair_fake_key_error:", type(e).__name__, str(e)) |
| 128 | + |
| 129 | + # Test unpair with wrong key format (should fail) |
| 130 | + try: |
| 131 | + result = ble.gap_unpair(b'\x01\x02\x03\x04\x05\x06') # Too short |
| 132 | + print("gap_unpair_wrong_key_result:", result) |
| 133 | + except Exception as e: |
| 134 | + print("gap_unpair_wrong_key_error:", type(e).__name__) |
| 135 | + |
| 136 | + finally: |
| 137 | + ble.active(0) |
| 138 | + |
| 139 | + |
| 140 | +# Acting in central role. |
| 141 | +def instance1(): |
| 142 | + multitest.next() |
| 143 | + try: |
| 144 | + # Connect to peripheral. |
| 145 | + print("gap_connect") |
| 146 | + ble.gap_connect(*BDADDR) |
| 147 | + conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS) |
| 148 | + |
| 149 | + # Discover characteristics (before pairing, doesn't need to be encrypted). |
| 150 | + ble.gattc_discover_characteristics(conn_handle, 1, 65535) |
| 151 | + value_handle = wait_for_event(_IRQ_GATTC_CHARACTERISTIC_RESULT, TIMEOUT_MS) |
| 152 | + wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS) |
| 153 | + |
| 154 | + # Pair with the peripheral. |
| 155 | + print("gap_pair") |
| 156 | + ble.gap_pair(conn_handle) |
| 157 | + |
| 158 | + # Wait for the pairing event. |
| 159 | + wait_for_event(_IRQ_ENCRYPTION_UPDATE, TIMEOUT_MS) |
| 160 | + |
| 161 | + # Read the peripheral's characteristic, should be encrypted. |
| 162 | + print("gattc_read") |
| 163 | + ble.gattc_read(conn_handle, value_handle) |
| 164 | + wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS) |
| 165 | + |
| 166 | + multitest.next() |
| 167 | + |
| 168 | + # Disconnect from the peripheral. |
| 169 | + print("gap_disconnect:", ble.gap_disconnect(conn_handle)) |
| 170 | + wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS) |
| 171 | + |
| 172 | + # Test gap_unpair on central side too |
| 173 | + print("central_gap_unpair_test") |
| 174 | + print("central_bond_keys_captured:", len(bond_keys)) |
| 175 | + |
| 176 | + # Test gap_unpair with captured bond keys on central side |
| 177 | + if bond_keys: |
| 178 | + for i, key in enumerate(bond_keys): |
| 179 | + try: |
| 180 | + result = ble.gap_unpair(key) |
| 181 | + print(f"central_gap_unpair_key_{i}_result:", result) |
| 182 | + except Exception as e: |
| 183 | + print(f"central_gap_unpair_key_{i}_error:", type(e).__name__, str(e)) |
| 184 | + else: |
| 185 | + print("central_gap_unpair_no_keys") |
| 186 | + |
| 187 | + finally: |
| 188 | + ble.active(0) |
| 189 | + |
| 190 | + |
| 191 | +ble = bluetooth.BLE() |
| 192 | +ble.config(mitm=True, le_secure=True, bond=True) # Enable bonding for unpair test |
| 193 | +ble.active(1) |
| 194 | +ble.irq(irq) |
0 commit comments