1 #!/usr/bin/python3 2 # 3 # Example nfcpy to wpa_supplicant wrapper for DPP NFC operations 4 # Copyright (c) 2012-2013, Jouni Malinen <j (at] w1.fi> 5 # Copyright (c) 2019-2020, The Linux Foundation 6 # 7 # This software may be distributed under the terms of the BSD license. 8 # See README for more details. 9 10 import binascii 11 import errno 12 import os 13 import struct 14 import sys 15 import time 16 import threading 17 import argparse 18 19 import nfc 20 import ndef 21 22 import logging 23 24 scriptsdir = os.path.dirname(os.path.realpath(sys.modules[__name__].__file__)) 25 sys.path.append(os.path.join(scriptsdir, '..', '..', 'wpaspy')) 26 import wpaspy 27 28 wpas_ctrl = '/var/run/wpa_supplicant' 29 ifname = None 30 init_on_touch = False 31 in_raw_mode = False 32 prev_tcgetattr = 0 33 no_input = False 34 continue_loop = True 35 terminate_now = False 36 summary_file = None 37 success_file = None 38 netrole = None 39 operation_success = False 40 mutex = threading.Lock() 41 42 C_NORMAL = '\033[0m' 43 C_RED = '\033[91m' 44 C_GREEN = '\033[92m' 45 C_YELLOW = '\033[93m' 46 C_BLUE = '\033[94m' 47 C_MAGENTA = '\033[95m' 48 C_CYAN = '\033[96m' 49 50 def summary(txt, color=None): 51 with mutex: 52 if color: 53 print(color + txt + C_NORMAL) 54 else: 55 print(txt) 56 if summary_file: 57 with open(summary_file, 'a') as f: 58 f.write(txt + "\n") 59 60 def success_report(txt): 61 summary(txt) 62 if success_file: 63 with open(success_file, 'a') as f: 64 f.write(txt + "\n") 65 66 def wpas_connect(): 67 ifaces = [] 68 if os.path.isdir(wpas_ctrl): 69 try: 70 ifaces = [os.path.join(wpas_ctrl, i) for i in os.listdir(wpas_ctrl)] 71 except OSError as error: 72 summary("Could not find wpa_supplicant: %s", str(error)) 73 return None 74 75 if len(ifaces) < 1: 76 summary("No wpa_supplicant control interface found") 77 return None 78 79 for ctrl in ifaces: 80 if ifname and ifname not in ctrl: 81 continue 82 if os.path.basename(ctrl).startswith("p2p-dev-"): 83 # skip P2P management interface 84 continue 85 try: 86 summary("Trying to use control interface " + ctrl) 87 wpas = wpaspy.Ctrl(ctrl) 88 return wpas 89 except Exception as e: 90 pass 91 summary("Could not connect to wpa_supplicant") 92 return None 93 94 def dpp_nfc_uri_process(uri): 95 wpas = wpas_connect() 96 if wpas is None: 97 return False 98 peer_id = wpas.request("DPP_NFC_URI " + uri) 99 if "FAIL" in peer_id: 100 summary("Could not parse DPP URI from NFC URI record", color=C_RED) 101 return False 102 peer_id = int(peer_id) 103 summary("peer_id=%d for URI from NFC Tag: %s" % (peer_id, uri)) 104 cmd = "DPP_AUTH_INIT peer=%d" % peer_id 105 global enrollee_only, configurator_only, config_params 106 if enrollee_only: 107 cmd += " role=enrollee" 108 elif configurator_only: 109 cmd += " role=configurator" 110 if config_params: 111 cmd += " " + config_params 112 summary("Initiate DPP authentication: " + cmd) 113 res = wpas.request(cmd) 114 if "OK" not in res: 115 summary("Failed to initiate DPP Authentication", color=C_RED) 116 return False 117 summary("DPP Authentication initiated") 118 return True 119 120 def dpp_hs_tag_read(record): 121 wpas = wpas_connect() 122 if wpas is None: 123 return False 124 summary(record) 125 if len(record.data) < 5: 126 summary("Too short DPP HS", color=C_RED) 127 return False 128 if record.data[0] != 0: 129 summary("Unexpected URI Identifier Code", color=C_RED) 130 return False 131 uribuf = record.data[1:] 132 try: 133 uri = uribuf.decode() 134 except: 135 summary("Invalid URI payload", color=C_RED) 136 return False 137 summary("URI: " + uri) 138 if not uri.startswith("DPP:"): 139 summary("Not a DPP URI", color=C_RED) 140 return False 141 return dpp_nfc_uri_process(uri) 142 143 def get_status(wpas, extra=None): 144 if extra: 145 extra = "-" + extra 146 else: 147 extra = "" 148 res = wpas.request("STATUS" + extra) 149 lines = res.splitlines() 150 vals = dict() 151 for l in lines: 152 try: 153 [name, value] = l.split('=', 1) 154 except ValueError: 155 summary("Ignore unexpected status line: %s" % l) 156 continue 157 vals[name] = value 158 return vals 159 160 def get_status_field(wpas, field, extra=None): 161 vals = get_status(wpas, extra) 162 if field in vals: 163 return vals[field] 164 return None 165 166 def own_addr(wpas): 167 addr = get_status_field(wpas, "address") 168 if addr is None: 169 addr = get_status_field(wpas, "bssid[0]") 170 return addr 171 172 def dpp_bootstrap_gen(wpas, type="qrcode", chan=None, mac=None, info=None, 173 curve=None, key=None): 174 cmd = "DPP_BOOTSTRAP_GEN type=" + type 175 if chan: 176 cmd += " chan=" + chan 177 if mac: 178 if mac is True: 179 mac = own_addr(wpas) 180 if mac is None: 181 summary("Could not determine local MAC address for bootstrap info") 182 else: 183 cmd += " mac=" + mac.replace(':', '') 184 if info: 185 cmd += " info=" + info 186 if curve: 187 cmd += " curve=" + curve 188 if key: 189 cmd += " key=" + key 190 res = wpas.request(cmd) 191 if "FAIL" in res: 192 raise Exception("Failed to generate bootstrapping info") 193 return int(res) 194 195 def dpp_start_listen(wpas, freq): 196 if get_status_field(wpas, "bssid[0]"): 197 summary("Own AP freq: %s MHz" % str(get_status_field(wpas, "freq"))) 198 if get_status_field(wpas, "beacon_set", extra="DRIVER") is None: 199 summary("Enable beaconing to have radio ready for RX") 200 wpas.request("DISABLE") 201 wpas.request("SET start_disabled 0") 202 wpas.request("ENABLE") 203 cmd = "DPP_LISTEN %d" % freq 204 global enrollee_only 205 global configurator_only 206 if enrollee_only: 207 cmd += " role=enrollee" 208 elif configurator_only: 209 cmd += " role=configurator" 210 global netrole 211 if netrole: 212 cmd += " netrole=" + netrole 213 summary(cmd) 214 res = wpas.request(cmd) 215 if "OK" not in res: 216 summary("Failed to start DPP listen", color=C_RED) 217 return False 218 return True 219 220 def wpas_get_nfc_uri(start_listen=True, pick_channel=False, chan_override=None): 221 listen_freq = 2412 222 wpas = wpas_connect() 223 if wpas is None: 224 return None 225 global own_id, chanlist 226 if chan_override: 227 chan = chan_override 228 else: 229 chan = chanlist 230 if chan and chan.startswith("81/"): 231 listen_freq = int(chan[3:].split(',')[0]) * 5 + 2407 232 if chan is None and get_status_field(wpas, "bssid[0]"): 233 freq = get_status_field(wpas, "freq") 234 if freq: 235 freq = int(freq) 236 if freq >= 2412 and freq <= 2462: 237 chan = "81/%d" % ((freq - 2407) / 5) 238 summary("Use current AP operating channel (%d MHz) as the URI channel list (%s)" % (freq, chan)) 239 listen_freq = freq 240 if chan is None and pick_channel: 241 chan = "81/6" 242 summary("Use channel 2437 MHz since no other preference provided") 243 listen_freq = 2437 244 own_id = dpp_bootstrap_gen(wpas, type="nfc-uri", chan=chan, mac=True) 245 res = wpas.request("DPP_BOOTSTRAP_GET_URI %d" % own_id).rstrip() 246 if "FAIL" in res: 247 return None 248 if start_listen: 249 if not dpp_start_listen(wpas, listen_freq): 250 raise Exception("Failed to start listen operation on %d MHz" % listen_freq) 251 return res 252 253 def wpas_report_handover_req(uri): 254 wpas = wpas_connect() 255 if wpas is None: 256 return None 257 global own_id 258 cmd = "DPP_NFC_HANDOVER_REQ own=%d uri=%s" % (own_id, uri) 259 return wpas.request(cmd) 260 261 def wpas_report_handover_sel(uri): 262 wpas = wpas_connect() 263 if wpas is None: 264 return None 265 global own_id 266 cmd = "DPP_NFC_HANDOVER_SEL own=%d uri=%s" % (own_id, uri) 267 return wpas.request(cmd) 268 269 def dpp_handover_client(handover, alt=False): 270 summary("About to start run_dpp_handover_client (alt=%s)" % str(alt)) 271 if alt: 272 handover.i_m_selector = False 273 run_dpp_handover_client(handover, alt) 274 summary("Done run_dpp_handover_client (alt=%s)" % str(alt)) 275 276 def run_client_alt(handover, alt): 277 if handover.start_client_alt and not alt: 278 handover.start_client_alt = False 279 summary("Try to send alternative handover request") 280 dpp_handover_client(handover, alt=True) 281 282 class HandoverClient(nfc.handover.HandoverClient): 283 def __init__(self, handover, llc): 284 super(HandoverClient, self).__init__(llc) 285 self.handover = handover 286 287 def recv_records(self, timeout=None): 288 msg = self.recv_octets(timeout) 289 if msg is None: 290 return None 291 records = list(ndef.message_decoder(msg, 'relax')) 292 if records and records[0].type == 'urn:nfc:wkt:Hs': 293 summary("Handover client received message '{0}'".format(records[0].type)) 294 return list(ndef.message_decoder(msg, 'relax')) 295 summary("Handover client received invalid message: %s" + binascii.hexlify(msg)) 296 return None 297 298 def recv_octets(self, timeout=None): 299 start = time.time() 300 msg = bytearray() 301 while True: 302 poll_timeout = 0.1 if timeout is None or timeout > 0.1 else timeout 303 if not self.socket.poll('recv', poll_timeout): 304 if timeout: 305 timeout -= time.time() - start 306 if timeout <= 0: 307 return None 308 start = time.time() 309 continue 310 try: 311 r = self.socket.recv() 312 if r is None: 313 return None 314 msg += r 315 except TypeError: 316 return b'' 317 try: 318 list(ndef.message_decoder(msg, 'strict', {})) 319 return bytes(msg) 320 except ndef.DecodeError: 321 if timeout: 322 timeout -= time.time() - start 323 if timeout <= 0: 324 return None 325 start = time.time() 326 continue 327 return None 328 329 def run_dpp_handover_client(handover, alt=False): 330 chan_override = None 331 if alt: 332 chan_override = handover.altchanlist 333 handover.alt_proposal_used = True 334 global test_uri, test_alt_uri 335 if test_uri: 336 summary("TEST MODE: Using specified URI (alt=%s)" % str(alt)) 337 uri = test_alt_uri if alt else test_uri 338 else: 339 uri = wpas_get_nfc_uri(start_listen=False, chan_override=chan_override) 340 if uri is None: 341 summary("Cannot start handover client - no bootstrap URI available", 342 color=C_RED) 343 return 344 handover.my_uri = uri 345 uri = ndef.UriRecord(uri) 346 summary("NFC URI record for DPP: " + str(uri)) 347 carrier = ndef.Record('application/vnd.wfa.dpp', 'A', uri.data) 348 global test_crn 349 if test_crn: 350 prev, = struct.unpack('>H', test_crn) 351 summary("TEST MODE: Use specified crn %d" % prev) 352 crn = test_crn 353 test_crn = struct.pack('>H', prev + 0x10) 354 else: 355 crn = os.urandom(2) 356 hr = ndef.HandoverRequestRecord(version="1.4", crn=crn) 357 hr.add_alternative_carrier('active', carrier.name) 358 message = [hr, carrier] 359 summary("NFC Handover Request message for DPP: " + str(message)) 360 361 if handover.peer_crn is not None and not alt: 362 summary("NFC handover request from peer was already received - do not send own[1]") 363 return 364 if handover.client: 365 summary("Use already started handover client") 366 client = handover.client 367 else: 368 summary("Start handover client") 369 client = HandoverClient(handover, handover.llc) 370 try: 371 summary("Trying to initiate NFC connection handover") 372 client.connect() 373 summary("Connected for handover") 374 except nfc.llcp.ConnectRefused: 375 summary("Handover connection refused") 376 client.close() 377 return 378 except Exception as e: 379 summary("Other exception: " + str(e)) 380 client.close() 381 return 382 handover.client = client 383 384 if handover.peer_crn is not None and not alt: 385 summary("NFC handover request from peer was already received - do not send own[2] (except alt)") 386 run_client_alt(handover, alt) 387 return 388 389 summary("Sending handover request") 390 391 handover.my_crn_ready = True 392 393 if not client.send_records(message): 394 handover.my_crn_ready = False 395 summary("Failed to send handover request", color=C_RED) 396 run_client_alt(handover, alt) 397 return 398 399 handover.my_crn, = struct.unpack('>H', crn) 400 401 summary("Receiving handover response") 402 try: 403 start = time.time() 404 message = client.recv_records(timeout=3.0) 405 end = time.time() 406 summary("Received {} record(s) in {} seconds".format(len(message) if message is not None else -1, end - start)) 407 except Exception as e: 408 # This is fine if we are the handover selector 409 if handover.hs_sent: 410 summary("Client receive failed as expected since I'm the handover server: %s" % str(e)) 411 elif handover.alt_proposal_used and not alt: 412 summary("Client received failed for initial proposal as expected since alternative proposal was also used: %s" % str(e)) 413 else: 414 summary("Client receive failed: %s" % str(e), color=C_RED) 415 message = None 416 if message is None: 417 if handover.hs_sent: 418 summary("No response received as expected since I'm the handover server") 419 elif handover.alt_proposal_used and not alt: 420 summary("No response received for initial proposal as expected since alternative proposal was also used") 421 elif handover.try_own and not alt: 422 summary("No response received for initial proposal as expected since alternative proposal will also be sent") 423 else: 424 summary("No response received", color=C_RED) 425 run_client_alt(handover, alt) 426 return 427 summary("Received message: " + str(message)) 428 if len(message) < 1 or \ 429 not isinstance(message[0], ndef.HandoverSelectRecord): 430 summary("Response was not Hs - received: " + message.type) 431 return 432 433 summary("Received handover select message") 434 summary("alternative carriers: " + str(message[0].alternative_carriers)) 435 if handover.i_m_selector: 436 summary("Ignore the received select since I'm the handover selector") 437 run_client_alt(handover, alt) 438 return 439 440 if handover.alt_proposal_used and not alt: 441 summary("Ignore received handover select for the initial proposal since alternative proposal was sent") 442 client.close() 443 return 444 445 dpp_found = False 446 for carrier in message: 447 if isinstance(carrier, ndef.HandoverSelectRecord): 448 continue 449 summary("Remote carrier type: " + carrier.type) 450 if carrier.type == "application/vnd.wfa.dpp": 451 if len(carrier.data) == 0 or carrier.data[0] != 0: 452 summary("URI Identifier Code 'None' not seen", color=C_RED) 453 continue 454 summary("DPP carrier type match - send to wpa_supplicant") 455 dpp_found = True 456 uri = carrier.data[1:].decode("utf-8") 457 summary("DPP URI: " + uri) 458 handover.peer_uri = uri 459 if test_uri: 460 summary("TEST MODE: Fake processing") 461 break 462 res = wpas_report_handover_sel(uri) 463 if res is None or "FAIL" in res: 464 summary("DPP handover report rejected", color=C_RED) 465 break 466 467 success_report("DPP handover reported successfully (initiator)") 468 summary("peer_id=" + res) 469 peer_id = int(res) 470 wpas = wpas_connect() 471 if wpas is None: 472 break 473 474 global enrollee_only 475 global config_params 476 if enrollee_only: 477 extra = " role=enrollee" 478 elif config_params: 479 extra = " role=configurator " + config_params 480 else: 481 # TODO: Single Configurator instance 482 res = wpas.request("DPP_CONFIGURATOR_ADD") 483 if "FAIL" in res: 484 summary("Failed to initiate Configurator", color=C_RED) 485 break 486 conf_id = int(res) 487 extra = " conf=sta-dpp configurator=%d" % conf_id 488 global own_id 489 summary("Initiate DPP authentication") 490 cmd = "DPP_AUTH_INIT peer=%d own=%d" % (peer_id, own_id) 491 cmd += extra 492 res = wpas.request(cmd) 493 if "FAIL" in res: 494 summary("Failed to initiate DPP authentication", color=C_RED) 495 break 496 497 if not dpp_found and handover.no_alt_proposal: 498 summary("DPP carrier not seen in response - do not allow alternative proposal anymore") 499 elif not dpp_found: 500 summary("DPP carrier not seen in response - allow peer to initiate a new handover with different parameters") 501 handover.alt_proposal = True 502 handover.my_crn_ready = False 503 handover.my_crn = None 504 handover.peer_crn = None 505 handover.hs_sent = False 506 summary("Returning from dpp_handover_client") 507 return 508 509 summary("Remove peer") 510 handover.close() 511 summary("Done with handover") 512 global only_one 513 if only_one: 514 print("only_one -> stop loop") 515 global continue_loop 516 continue_loop = False 517 518 global no_wait 519 if no_wait or only_one: 520 summary("Trying to exit..") 521 global terminate_now 522 terminate_now = True 523 524 summary("Returning from dpp_handover_client") 525 526 class HandoverServer(nfc.handover.HandoverServer): 527 def __init__(self, handover, llc): 528 super(HandoverServer, self).__init__(llc) 529 self.sent_carrier = None 530 self.ho_server_processing = False 531 self.success = False 532 self.llc = llc 533 self.handover = handover 534 535 def serve(self, socket): 536 peer_sap = socket.getpeername() 537 summary("Serving handover client on remote sap {0}".format(peer_sap)) 538 send_miu = socket.getsockopt(nfc.llcp.SO_SNDMIU) 539 try: 540 while socket.poll("recv"): 541 req = bytearray() 542 while socket.poll("recv"): 543 r = socket.recv() 544 if r is None: 545 return None 546 summary("Received %d octets" % len(r)) 547 req += r 548 if len(req) == 0: 549 continue 550 try: 551 list(ndef.message_decoder(req, 'strict', {})) 552 except ndef.DecodeError: 553 continue 554 summary("Full message received") 555 resp = self._process_request_data(req) 556 if resp is None or len(resp) == 0: 557 summary("No handover select to send out - wait for a possible alternative handover request") 558 handover.alt_proposal = True 559 req = bytearray() 560 continue 561 562 for offset in range(0, len(resp), send_miu): 563 if not socket.send(resp[offset:offset + send_miu]): 564 summary("Failed to send handover select - connection closed") 565 return 566 summary("Sent out full handover select") 567 if handover.terminate_on_hs_send_completion: 568 handover.delayed_exit() 569 570 except nfc.llcp.Error as e: 571 global terminate_now 572 summary("HandoverServer exception: %s" % e, 573 color=None if e.errno == errno.EPIPE or terminate_now else C_RED) 574 finally: 575 socket.close() 576 summary("Handover serve thread exiting") 577 578 def process_handover_request_message(self, records): 579 handover = self.handover 580 self.ho_server_processing = True 581 global in_raw_mode 582 was_in_raw_mode = in_raw_mode 583 clear_raw_mode() 584 if was_in_raw_mode: 585 print("\n") 586 summary("HandoverServer - request received: " + str(records)) 587 588 for carrier in records: 589 if not isinstance(carrier, ndef.HandoverRequestRecord): 590 continue 591 if carrier.collision_resolution_number: 592 handover.peer_crn = carrier.collision_resolution_number 593 summary("peer_crn: %d" % handover.peer_crn) 594 595 if handover.my_crn is None and handover.my_crn_ready: 596 summary("Still trying to send own handover request - wait a moment to see if that succeeds before checking crn values") 597 for i in range(10): 598 if handover.my_crn is not None: 599 break 600 time.sleep(0.01) 601 if handover.my_crn is not None: 602 summary("my_crn: %d" % handover.my_crn) 603 604 if handover.my_crn is not None and handover.peer_crn is not None: 605 if handover.my_crn == handover.peer_crn: 606 summary("Same crn used - automatic collision resolution failed") 607 # TODO: Should generate a new Handover Request message 608 return '' 609 if ((handover.my_crn & 1) == (handover.peer_crn & 1) and \ 610 handover.my_crn > handover.peer_crn) or \ 611 ((handover.my_crn & 1) != (handover.peer_crn & 1) and \ 612 handover.my_crn < handover.peer_crn): 613 summary("I'm the Handover Selector Device") 614 handover.i_m_selector = True 615 else: 616 summary("Peer is the Handover Selector device") 617 summary("Ignore the received request.") 618 return '' 619 620 hs = ndef.HandoverSelectRecord('1.4') 621 sel = [hs] 622 623 found = False 624 625 for carrier in records: 626 if isinstance(carrier, ndef.HandoverRequestRecord): 627 continue 628 summary("Remote carrier type: " + carrier.type) 629 if carrier.type == "application/vnd.wfa.dpp": 630 summary("DPP carrier type match - add DPP carrier record") 631 if len(carrier.data) == 0 or carrier.data[0] != 0: 632 summary("URI Identifier Code 'None' not seen", color=C_RED) 633 continue 634 uri = carrier.data[1:].decode("utf-8") 635 summary("Received DPP URI: " + uri) 636 637 global test_uri, test_alt_uri 638 if test_uri: 639 summary("TEST MODE: Using specified URI") 640 data = test_sel_uri if test_sel_uri else test_uri 641 elif handover.alt_proposal and handover.altchanlist: 642 summary("Use alternative channel list while processing alternative proposal from peer") 643 data = wpas_get_nfc_uri(start_listen=False, 644 chan_override=handover.altchanlist, 645 pick_channel=True) 646 else: 647 data = wpas_get_nfc_uri(start_listen=False, 648 pick_channel=True) 649 summary("Own URI (pre-processing): %s" % data) 650 651 if test_uri: 652 summary("TEST MODE: Fake processing") 653 res = "OK" 654 data += " [%s]" % uri 655 else: 656 res = wpas_report_handover_req(uri) 657 if res is None or "FAIL" in res: 658 summary("DPP handover request processing failed", 659 color=C_RED) 660 if handover.altchanlist: 661 data = wpas_get_nfc_uri(start_listen=False, 662 chan_override=handover.altchanlist) 663 summary("Own URI (try another channel list): %s" % data) 664 continue 665 666 if test_alt_uri: 667 summary("TEST MODE: Reject initial proposal") 668 continue 669 670 found = True 671 672 if not test_uri: 673 wpas = wpas_connect() 674 if wpas is None: 675 continue 676 global own_id 677 data = wpas.request("DPP_BOOTSTRAP_GET_URI %d" % own_id).rstrip() 678 if "FAIL" in data: 679 continue 680 summary("Own URI (post-processing): %s" % data) 681 handover.my_uri = data 682 handover.peer_uri = uri 683 uri = ndef.UriRecord(data) 684 summary("Own bootstrapping NFC URI record: " + str(uri)) 685 686 if not test_uri: 687 info = wpas.request("DPP_BOOTSTRAP_INFO %d" % own_id) 688 freq = None 689 for line in info.splitlines(): 690 if line.startswith("use_freq="): 691 freq = int(line.split('=')[1]) 692 if freq is None or freq == 0: 693 summary("No channel negotiated over NFC - use channel 6") 694 freq = 2437 695 else: 696 summary("Negotiated channel: %d MHz" % freq) 697 if not dpp_start_listen(wpas, freq): 698 break 699 700 carrier = ndef.Record('application/vnd.wfa.dpp', 'A', uri.data) 701 summary("Own DPP carrier record: " + str(carrier)) 702 hs.add_alternative_carrier('active', carrier.name) 703 sel = [hs, carrier] 704 break 705 706 summary("Sending handover select: " + str(sel)) 707 if found: 708 summary("Handover completed successfully") 709 handover.terminate_on_hs_send_completion = True 710 self.success = True 711 handover.hs_sent = True 712 handover.i_m_selector = True 713 elif handover.no_alt_proposal: 714 summary("Do not try alternative proposal anymore - handover failed", 715 color=C_RED) 716 handover.hs_sent = True 717 else: 718 summary("Try to initiate with alternative parameters") 719 handover.try_own = True 720 handover.hs_sent = False 721 handover.no_alt_proposal = True 722 if handover.client_thread: 723 handover.start_client_alt = True 724 else: 725 handover.client_thread = threading.Thread(target=llcp_worker, 726 args=(self.llc, True)) 727 handover.client_thread.start() 728 return sel 729 730 def clear_raw_mode(): 731 import sys, tty, termios 732 global prev_tcgetattr, in_raw_mode 733 if not in_raw_mode: 734 return 735 fd = sys.stdin.fileno() 736 termios.tcsetattr(fd, termios.TCSADRAIN, prev_tcgetattr) 737 in_raw_mode = False 738 739 def getch(): 740 import sys, tty, termios, select 741 global prev_tcgetattr, in_raw_mode 742 fd = sys.stdin.fileno() 743 prev_tcgetattr = termios.tcgetattr(fd) 744 ch = None 745 try: 746 tty.setraw(fd) 747 in_raw_mode = True 748 [i, o, e] = select.select([fd], [], [], 0.05) 749 if i: 750 ch = sys.stdin.read(1) 751 finally: 752 termios.tcsetattr(fd, termios.TCSADRAIN, prev_tcgetattr) 753 in_raw_mode = False 754 return ch 755 756 def dpp_tag_read(tag): 757 success = False 758 for record in tag.ndef.records: 759 summary(record) 760 summary("record type " + record.type) 761 if record.type == "application/vnd.wfa.dpp": 762 summary("DPP HS tag - send to wpa_supplicant") 763 success = dpp_hs_tag_read(record) 764 break 765 if isinstance(record, ndef.UriRecord): 766 summary("URI record: uri=" + record.uri) 767 summary("URI record: iri=" + record.iri) 768 if record.iri.startswith("DPP:"): 769 summary("DPP URI") 770 if not dpp_nfc_uri_process(record.iri): 771 break 772 success = True 773 else: 774 summary("Ignore unknown URI") 775 break 776 777 if success: 778 success_report("Tag read succeeded") 779 780 return success 781 782 def rdwr_connected_write_tag(tag): 783 summary("Tag found - writing - " + str(tag)) 784 if not tag.ndef: 785 summary("Not a formatted NDEF tag", color=C_RED) 786 return 787 if not tag.ndef.is_writeable: 788 summary("Not a writable tag", color=C_RED) 789 return 790 global dpp_tag_data 791 if tag.ndef.capacity < len(dpp_tag_data): 792 summary("Not enough room for the message") 793 return 794 try: 795 tag.ndef.records = dpp_tag_data 796 except ValueError as e: 797 summary("Writing the tag failed: %s" % str(e), color=C_RED) 798 return 799 success_report("Tag write succeeded") 800 summary("Tag writing completed - remove tag", color=C_GREEN) 801 global only_one, operation_success 802 operation_success = True 803 if only_one: 804 global continue_loop 805 continue_loop = False 806 global dpp_sel_wait_remove 807 return dpp_sel_wait_remove 808 809 def write_nfc_uri(clf, wait_remove=True): 810 summary("Write NFC URI record") 811 data = wpas_get_nfc_uri() 812 if data is None: 813 summary("Could not get NFC URI from wpa_supplicant", color=C_RED) 814 return 815 816 global dpp_sel_wait_remove 817 dpp_sel_wait_remove = wait_remove 818 summary("URI: %s" % data) 819 uri = ndef.UriRecord(data) 820 summary(uri) 821 822 summary("Touch an NFC tag to write URI record", color=C_CYAN) 823 global dpp_tag_data 824 dpp_tag_data = [uri] 825 clf.connect(rdwr={'on-connect': rdwr_connected_write_tag}) 826 827 def write_nfc_hs(clf, wait_remove=True): 828 summary("Write NFC Handover Select record on a tag") 829 data = wpas_get_nfc_uri() 830 if data is None: 831 summary("Could not get NFC URI from wpa_supplicant", color=C_RED) 832 return 833 834 global dpp_sel_wait_remove 835 dpp_sel_wait_remove = wait_remove 836 summary("URI: %s" % data) 837 uri = ndef.UriRecord(data) 838 summary(uri) 839 carrier = ndef.Record('application/vnd.wfa.dpp', 'A', uri.data) 840 hs = ndef.HandoverSelectRecord('1.4') 841 hs.add_alternative_carrier('active', carrier.name) 842 summary(hs) 843 summary(carrier) 844 845 summary("Touch an NFC tag to write HS record", color=C_CYAN) 846 global dpp_tag_data 847 dpp_tag_data = [hs, carrier] 848 summary(dpp_tag_data) 849 clf.connect(rdwr={'on-connect': rdwr_connected_write_tag}) 850 851 def rdwr_connected(tag): 852 global only_one, no_wait 853 summary("Tag connected: " + str(tag)) 854 855 if tag.ndef: 856 summary("NDEF tag: " + tag.type) 857 summary(tag.ndef.records) 858 success = dpp_tag_read(tag) 859 if only_one and success: 860 global continue_loop 861 continue_loop = False 862 else: 863 summary("Not an NDEF tag - remove tag", color=C_RED) 864 return True 865 866 return not no_wait 867 868 def llcp_worker(llc, try_alt): 869 global handover 870 print("Start of llcp_worker()") 871 if try_alt: 872 summary("Starting handover client (try_alt)") 873 dpp_handover_client(handover, alt=True) 874 summary("Exiting llcp_worker thread (try_alt)") 875 return 876 global init_on_touch 877 if init_on_touch: 878 summary("Starting handover client (init_on_touch)") 879 dpp_handover_client(handover) 880 summary("llcp_worker init_on_touch processing completed: try_own={} hs_sent={} no_alt_proposal={} start_client_alt={}".format(handover.try_own, handover.hs_sent, handover.no_alt_proposal, handover.start_client_alt)) 881 if handover.start_client_alt and not handover.hs_sent: 882 summary("Try alternative handover request before exiting llcp_worker") 883 handover.start_client_alt = False 884 dpp_handover_client(handover, alt=True) 885 summary("Exiting llcp_worker thread (init_on_touch)") 886 return 887 888 global no_input 889 if no_input: 890 summary("Wait for handover to complete") 891 else: 892 print("Wait for handover to complete - press 'i' to initiate") 893 while not handover.wait_connection and handover.srv.sent_carrier is None: 894 if handover.try_own: 895 handover.try_own = False 896 summary("Try to initiate another handover with own parameters") 897 handover.my_crn_ready = False 898 handover.my_crn = None 899 handover.peer_crn = None 900 handover.hs_sent = False 901 dpp_handover_client(handover, alt=True) 902 summary("Exiting llcp_worker thread (retry with own parameters)") 903 return 904 if handover.srv.ho_server_processing: 905 time.sleep(0.025) 906 elif no_input: 907 time.sleep(0.5) 908 else: 909 res = getch() 910 if res != 'i': 911 continue 912 clear_raw_mode() 913 summary("Starting handover client") 914 dpp_handover_client(handover) 915 summary("Exiting llcp_worker thread (manual init)") 916 return 917 918 global in_raw_mode 919 was_in_raw_mode = in_raw_mode 920 clear_raw_mode() 921 if was_in_raw_mode: 922 print("\r") 923 summary("Exiting llcp_worker thread") 924 925 class ConnectionHandover(): 926 def __init__(self): 927 self.client = None 928 self.client_thread = None 929 self.reset() 930 self.exit_thread = None 931 932 def reset(self): 933 self.wait_connection = False 934 self.my_crn_ready = False 935 self.my_crn = None 936 self.peer_crn = None 937 self.hs_sent = False 938 self.no_alt_proposal = False 939 self.alt_proposal_used = False 940 self.i_m_selector = False 941 self.start_client_alt = False 942 self.terminate_on_hs_send_completion = False 943 self.try_own = False 944 self.my_uri = None 945 self.peer_uri = None 946 self.connected = False 947 self.alt_proposal = False 948 949 def start_handover_server(self, llc): 950 summary("Start handover server") 951 self.llc = llc 952 self.srv = HandoverServer(self, llc) 953 954 def close(self): 955 if self.client: 956 self.client.close() 957 self.client = None 958 959 def run_delayed_exit(self): 960 summary("Trying to exit (delayed)..") 961 time.sleep(0.25) 962 summary("Trying to exit (after wait)..") 963 global terminate_now 964 terminate_now = True 965 966 def delayed_exit(self): 967 global only_one 968 if only_one: 969 self.exit_thread = threading.Thread(target=self.run_delayed_exit) 970 self.exit_thread.start() 971 972 def llcp_startup(llc): 973 global handover 974 handover.start_handover_server(llc) 975 return llc 976 977 def llcp_connected(llc): 978 summary("P2P LLCP connected") 979 global handover 980 handover.connected = True 981 handover.srv.start() 982 if init_on_touch or not no_input: 983 handover.client_thread = threading.Thread(target=llcp_worker, 984 args=(llc, False)) 985 handover.client_thread.start() 986 return True 987 988 def llcp_release(llc): 989 summary("LLCP release") 990 global handover 991 handover.close() 992 return True 993 994 def terminate_loop(): 995 global terminate_now 996 return terminate_now 997 998 def main(): 999 clf = nfc.ContactlessFrontend() 1000 1001 parser = argparse.ArgumentParser(description='nfcpy to wpa_supplicant integration for DPP NFC operations') 1002 parser.add_argument('-d', const=logging.DEBUG, default=logging.INFO, 1003 action='store_const', dest='loglevel', 1004 help='verbose debug output') 1005 parser.add_argument('-q', const=logging.WARNING, action='store_const', 1006 dest='loglevel', help='be quiet') 1007 parser.add_argument('--only-one', '-1', action='store_true', 1008 help='run only one operation and exit') 1009 parser.add_argument('--init-on-touch', '-I', action='store_true', 1010 help='initiate handover on touch') 1011 parser.add_argument('--no-wait', action='store_true', 1012 help='do not wait for tag to be removed before exiting') 1013 parser.add_argument('--ifname', '-i', 1014 help='network interface name') 1015 parser.add_argument('--no-input', '-a', action='store_true', 1016 help='do not use stdout input to initiate handover') 1017 parser.add_argument('--tag-read-only', '-t', action='store_true', 1018 help='tag read only (do not allow connection handover)') 1019 parser.add_argument('--handover-only', action='store_true', 1020 help='connection handover only (do not allow tag read)') 1021 parser.add_argument('--enrollee', action='store_true', 1022 help='run as Enrollee-only') 1023 parser.add_argument('--configurator', action='store_true', 1024 help='run as Configurator-only') 1025 parser.add_argument('--config-params', default='', 1026 help='configurator parameters') 1027 parser.add_argument('--ctrl', default='/var/run/wpa_supplicant', 1028 help='wpa_supplicant/hostapd control interface') 1029 parser.add_argument('--summary', 1030 help='summary file for writing status updates') 1031 parser.add_argument('--success', 1032 help='success file for writing success update') 1033 parser.add_argument('--device', default='usb', help='NFC device to open') 1034 parser.add_argument('--chan', default=None, help='channel list') 1035 parser.add_argument('--altchan', default=None, help='alternative channel list') 1036 parser.add_argument('--netrole', default=None, help='netrole for Enrollee') 1037 parser.add_argument('--test-uri', default=None, 1038 help='test mode: initial URI') 1039 parser.add_argument('--test-alt-uri', default=None, 1040 help='test mode: alternative URI') 1041 parser.add_argument('--test-sel-uri', default=None, 1042 help='test mode: handover select URI') 1043 parser.add_argument('--test-crn', default=None, 1044 help='test mode: hardcoded crn') 1045 parser.add_argument('command', choices=['write-nfc-uri', 1046 'write-nfc-hs'], 1047 nargs='?') 1048 args = parser.parse_args() 1049 summary(args) 1050 1051 global handover 1052 handover = ConnectionHandover() 1053 1054 global only_one 1055 only_one = args.only_one 1056 1057 global no_wait 1058 no_wait = args.no_wait 1059 1060 global chanlist, netrole, test_uri, test_alt_uri, test_sel_uri 1061 global test_crn 1062 chanlist = args.chan 1063 handover.altchanlist = args.altchan 1064 netrole = args.netrole 1065 test_uri = args.test_uri 1066 test_alt_uri = args.test_alt_uri 1067 test_sel_uri = args.test_sel_uri 1068 if args.test_crn: 1069 test_crn = struct.pack('>H', int(args.test_crn)) 1070 else: 1071 test_crn = None 1072 1073 logging.basicConfig(level=args.loglevel) 1074 for l in ['nfc.clf.rcs380', 1075 'nfc.clf.transport', 1076 'nfc.clf.device', 1077 'nfc.clf.__init__', 1078 'nfc.llcp', 1079 'nfc.handover']: 1080 log = logging.getLogger(l) 1081 log.setLevel(args.loglevel) 1082 1083 global init_on_touch 1084 init_on_touch = args.init_on_touch 1085 1086 global enrollee_only 1087 enrollee_only = args.enrollee 1088 1089 global configurator_only 1090 configurator_only = args.configurator 1091 1092 global config_params 1093 config_params = args.config_params 1094 1095 if args.ifname: 1096 global ifname 1097 ifname = args.ifname 1098 summary("Selected ifname " + ifname) 1099 1100 if args.ctrl: 1101 global wpas_ctrl 1102 wpas_ctrl = args.ctrl 1103 1104 if args.summary: 1105 global summary_file 1106 summary_file = args.summary 1107 1108 if args.success: 1109 global success_file 1110 success_file = args.success 1111 1112 if args.no_input: 1113 global no_input 1114 no_input = True 1115 1116 clf = nfc.ContactlessFrontend() 1117 1118 try: 1119 if not clf.open(args.device): 1120 summary("Could not open connection with an NFC device", color=C_RED) 1121 raise SystemExit(1) 1122 1123 if args.command == "write-nfc-uri": 1124 write_nfc_uri(clf, wait_remove=not args.no_wait) 1125 if not operation_success: 1126 raise SystemExit(1) 1127 raise SystemExit 1128 1129 if args.command == "write-nfc-hs": 1130 write_nfc_hs(clf, wait_remove=not args.no_wait) 1131 if not operation_success: 1132 raise SystemExit(1) 1133 raise SystemExit 1134 1135 global continue_loop 1136 while continue_loop: 1137 global in_raw_mode 1138 was_in_raw_mode = in_raw_mode 1139 clear_raw_mode() 1140 if was_in_raw_mode: 1141 print("\r") 1142 if args.handover_only: 1143 summary("Waiting a peer to be touched", color=C_MAGENTA) 1144 elif args.tag_read_only: 1145 summary("Waiting for a tag to be touched", color=C_BLUE) 1146 else: 1147 summary("Waiting for a tag or peer to be touched", 1148 color=C_GREEN) 1149 handover.wait_connection = True 1150 try: 1151 if args.tag_read_only: 1152 if not clf.connect(rdwr={'on-connect': rdwr_connected}): 1153 break 1154 elif args.handover_only: 1155 if not clf.connect(llcp={'on-startup': llcp_startup, 1156 'on-connect': llcp_connected, 1157 'on-release': llcp_release}, 1158 terminate=terminate_loop): 1159 break 1160 else: 1161 if not clf.connect(rdwr={'on-connect': rdwr_connected}, 1162 llcp={'on-startup': llcp_startup, 1163 'on-connect': llcp_connected, 1164 'on-release': llcp_release}, 1165 terminate=terminate_loop): 1166 break 1167 except Exception as e: 1168 summary("clf.connect failed: " + str(e)) 1169 break 1170 1171 if only_one and handover.connected: 1172 role = "selector" if handover.i_m_selector else "requestor" 1173 summary("Connection handover result: I'm the %s" % role, 1174 color=C_YELLOW) 1175 if handover.peer_uri: 1176 summary("Peer URI: " + handover.peer_uri, color=C_YELLOW) 1177 if handover.my_uri: 1178 summary("My URI: " + handover.my_uri, color=C_YELLOW) 1179 if not (handover.peer_uri and handover.my_uri): 1180 summary("Negotiated connection handover failed", 1181 color=C_YELLOW) 1182 break 1183 1184 except KeyboardInterrupt: 1185 raise SystemExit 1186 finally: 1187 clf.close() 1188 1189 raise SystemExit 1190 1191 if __name__ == '__main__': 1192 main() 1193
Indexes created Sun Mar 01 05:31:48 UTC 2026