import sys import time import json import asyncio from http.server import BaseHTTPRequestHandler, HTTPServer import urllib.parse try: from smartcard.System import readers from smartcard.util import toHexString from smartcard.Exceptions import CardConnectionException, NoCardException except ImportError: print("pyscard not installed. Please install it first using: pip install pyscard") sys.exit(1) # Try importing bleak for BLE hardware scanning try: from bleak import BleakScanner BLE_SUPPORT = True except ImportError: BLE_SUPPORT = False PORT = 5000 # ───────────────────────────────────────────────────────────────────────────── # Low-level card helpers — all operate on an ALREADY CONNECTED connection # so there is no RF gap between APDUs. # ───────────────────────────────────────────────────────────────────────────── def apdu_get_uid(connection): """Return UID string or None.""" data, sw1, sw2 = connection.transmit([0xFF, 0xCA, 0x00, 0x00, 0x00]) if sw1 == 0x90 and sw2 == 0x00: return toHexString(data).replace(" ", "").upper() return None def apdu_is_ntag(connection): """Read page 4 without auth — succeeds on NTAG/Ultralight, fails on MIFARE.""" try: data, sw1, sw2 = connection.transmit([0xFF, 0xB0, 0x00, 0x04, 0x04]) return sw1 == 0x90 and sw2 == 0x00 except Exception: return False def apdu_mifare_authenticate(connection, block=4): """ Authenticate a MIFARE Classic block with default Key A (FF FF FF FF FF FF). Tries the modern PC/SC APDU first (0xFF 0x86), then the legacy one (0xFF 0x88). Returns True on success. """ # Load key into slot 0 LOAD_KEY = [0xFF, 0x82, 0x00, 0x00, 0x06, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF] connection.transmit(LOAD_KEY) # Modern auth command (ACR122U, ACS, most CCID readers) try: AUTH_V2 = [0xFF, 0x86, 0x00, 0x00, 0x05, 0x01, 0x00, block, 0x60, 0x00] _, sw1, sw2 = connection.transmit(AUTH_V2) if sw1 == 0x90 and sw2 == 0x00: return True except Exception as e: print(f" MIFARE auth V2 exception: {e}") # Legacy auth command (some older / clone readers) try: AUTH_V1 = [0xFF, 0x88, 0x00, block, 0x60, 0x00] _, sw1, sw2 = connection.transmit(AUTH_V1) if sw1 == 0x90 and sw2 == 0x00: return True except Exception as e: print(f" MIFARE auth V1 exception: {e}") return False def apdu_read_ntag(connection): """Read 4 pages (16 bytes) of NTAG user data (pages 4-7). Returns bytes list.""" data_bytes = [] for page in range(4, 8): d, sw1, sw2 = connection.transmit([0xFF, 0xB0, 0x00, page, 0x04]) if sw1 == 0x90 and sw2 == 0x00: data_bytes.extend(d) else: return None # page read failed return data_bytes def apdu_write_ntag(connection, data_bytes): """Write 16 bytes across NTAG pages 4-7. Returns True on success.""" for i, page in enumerate(range(4, 8)): chunk = data_bytes[i * 4:(i + 1) * 4] WRITE = [0xFF, 0xD6, 0x00, page, 0x04] + chunk d, sw1, sw2 = connection.transmit(WRITE) if sw1 != 0x90 or sw2 != 0x00: print(f" NTAG write page {page} failed: sw1={hex(sw1)} sw2={hex(sw2)}") return False return True def apdu_read_mifare_block4(connection): """Read Block 4 of MIFARE Classic (must be authenticated first). Returns bytes list.""" d, sw1, sw2 = connection.transmit([0xFF, 0xB0, 0x00, 0x04, 0x10]) if sw1 == 0x90 and sw2 == 0x00: return list(d) return None def apdu_write_mifare_block4(connection, data_bytes): """Write 16 bytes to Block 4 of MIFARE Classic (must be authenticated first).""" WRITE = [0xFF, 0xD6, 0x00, 0x04, 0x10] + data_bytes _, sw1, sw2 = connection.transmit(WRITE) return sw1 == 0x90 and sw2 == 0x00 def bytes_to_text(data_bytes): """Convert a list of bytes to a printable ASCII string.""" if not data_bytes: return "" return "".join(chr(b) for b in data_bytes if 32 <= b <= 126).strip() def text_to_padded_bytes(text, length=16): """Convert text to a null-padded byte list of exactly `length` bytes.""" text = text[:length] return [ord(c) for c in text.ljust(length, '\0')] # ───────────────────────────────────────────────────────────────────────────── # HTTP Handler # ───────────────────────────────────────────────────────────────────────────── class NFCBridgeHandler(BaseHTTPRequestHandler): def end_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') super().end_headers() def do_OPTIONS(self): self.send_response(200) self.end_headers() def log_message(self, format, *args): return # keep terminal clean; errors are printed manually def do_GET(self): p = urllib.parse.urlparse(self.path).path if p == '/status': self.handle_status() elif p == '/scan': self.handle_scan() elif p == '/ble/scan': self.handle_ble_scan() else: self.send_response(404) self.end_headers() def do_POST(self): p = urllib.parse.urlparse(self.path).path if p == '/write': self.handle_write() else: self.send_response(404) self.end_headers() # ── /status ─────────────────────────────────────────────────────────────── def handle_status(self): r = readers() response = { "status": "connected" if r else "disconnected", "reader": r[0].name if r else None, "ble_support": BLE_SUPPORT } self._json(response) # ── /scan ───────────────────────────────────────────────────────────────── def handle_scan(self): r = readers() if not r: self.send_error_response("No NFC reader detected.") return reader = r[0] timeout = 15.0 start_time = time.time() while time.time() - start_time < timeout: connection = None try: connection = reader.createConnection() connection.connect() # ── RF stabilisation: give the reader 300 ms to complete # ── the contactless handshake before sending any APDUs. time.sleep(0.3) # ── Everything in ONE connected session ────────────────────── uid = apdu_get_uid(connection) if not uid: continue is_ntag = apdu_is_ntag(connection) existing_text = "" if is_ntag: raw = apdu_read_ntag(connection) existing_text = bytes_to_text(raw) if raw else "[Empty or unreadable]" else: if apdu_mifare_authenticate(connection): raw = apdu_read_mifare_block4(connection) existing_text = bytes_to_text(raw) if raw else "[Empty or unreadable]" else: existing_text = "[Auth failed — card locked?]" self._json({ "status": "success", "uid": uid, "data": existing_text or "[Empty or unreadable]", "card_type": "NTAG/Ultralight" if is_ntag else "MIFARE Classic" }) return except Exception as e: msg = str(e) if "No card" not in msg and "removed" not in msg.lower(): print(f"Scan error: {msg}") finally: if connection: try: connection.disconnect() except Exception: pass time.sleep(0.25) self.send_error_response("No card detected. Please hold the card on the reader and try again.") # ── /write ──────────────────────────────────────────────────────────────── def handle_write(self): # Parse body try: length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) payload = json.loads(body.decode()) text_to_write = payload.get("data", "").strip() except Exception: self.send_error_response("Invalid JSON payload.") return if not text_to_write: self.send_error_response("No data provided to write.") return r = readers() if not r: self.send_error_response("No NFC reader detected.") return reader = r[0] deadline = time.time() + 20.0 # 20 s write window — gives the user time to hold card steady data_bytes = text_to_padded_bytes(text_to_write) while time.time() < deadline: connection = None try: connection = reader.createConnection() connection.connect() # ── RF stabilisation: give the contactless handshake 300 ms to # ── fully settle before sending the first APDU command. time.sleep(0.3) # ── Everything in ONE connected session ────────────────────── # 1. Read UID uid = apdu_get_uid(connection) if not uid: raise RuntimeError("Could not read card UID.") print(f"Card UID: {uid}") # 2. Detect card type is_ntag = apdu_is_ntag(connection) print(f"Card type: {'NTAG/Ultralight' if is_ntag else 'MIFARE Classic (or unknown)'}") # 3. Check for existing data BEFORE writing existing_text = "" if is_ntag: raw = apdu_read_ntag(connection) existing_text = bytes_to_text(raw) if raw else "" else: if apdu_mifare_authenticate(connection): raw = apdu_read_mifare_block4(connection) existing_text = bytes_to_text(raw) if raw else "" else: raise RuntimeError("Card authentication failed — card may be locked or use a non-default key.") if existing_text: print(f"Card already has data: '{existing_text}' — overwriting...") else: print("Card is empty. Writing fresh data...") # 4. Re-authenticate for MIFARE before writing (session may need refresh) if not is_ntag: if not apdu_mifare_authenticate(connection): raise RuntimeError("Re-authentication before write failed.") # 5. Write data if is_ntag: success = apdu_write_ntag(connection, data_bytes) else: success = apdu_write_mifare_block4(connection, data_bytes) if success: print(f"Write successful: '{text_to_write}' → UID {uid}") self._json({ "status": "success", "uid": uid, "written": text_to_write, "previous_data": existing_text or None, "card_type": "NTAG/Ultralight" if is_ntag else "MIFARE Classic" }) return else: raise RuntimeError("Write APDU returned failure status code.") except Exception as e: msg = str(e) # Suppress noise for "card not present yet" messages if "removed" in msg.lower() or "no card" in msg.lower() or "0x80100069" in msg: pass # card not on reader yet — keep polling else: print(f"Write attempt error: {msg}") finally: if connection: try: connection.disconnect() except Exception: pass time.sleep(0.3) self.send_error_response( "Write timed out. Make sure the card is flat on the reader and not moving. " "If it fails repeatedly, the card may be write-protected." ) # ── /ble/scan ───────────────────────────────────────────────────────────── def handle_ble_scan(self): """Scans for nearby Bluetooth Low Energy devices.""" if not BLE_SUPPORT: mock_beacons = [ {"address": "BEACON-01", "name": "Standard Beacon 1", "rssi": -58}, {"address": "BEACON-02", "name": "Standard Beacon 2", "rssi": -74}, {"address": "BEACON-03", "name": "Asset Tracker Tag", "rssi": -85} ] self._json({ "status": "success", "mode": "mocked", "message": "Install bleak (pip install bleak) for live local Bluetooth scans.", "devices": mock_beacons }) return try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) devices_dict = loop.run_until_complete( BleakScanner.discover(timeout=3.0, return_adv=True) ) loop.close() device_list = [] for address, (device, adv_data) in devices_dict.items(): rssi_val = -100 if hasattr(adv_data, 'rssi') and adv_data.rssi is not None: rssi_val = adv_data.rssi elif hasattr(device, 'rssi') and device.rssi is not None: rssi_val = device.rssi name_val = "" if adv_data and hasattr(adv_data, 'local_name') and adv_data.local_name: name_val = adv_data.local_name elif device and hasattr(device, 'name') and device.name: name_val = device.name device_list.append({ "address": device.address, "name": name_val, "rssi": rssi_val }) self._json({ "status": "success", "mode": "hardware", "devices": sorted(device_list, key=lambda x: x['rssi'], reverse=True) }) except Exception as e: self.send_error_response(f"BLE scan failed: {str(e)}") # ── Helpers ─────────────────────────────────────────────────────────────── def _json(self, data, code=200): body = json.dumps(data).encode() self.send_response(code) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(body) def send_error_response(self, message): self._json({"status": "error", "message": message}, code=400) # ───────────────────────────────────────────────────────────────────────────── def run(): print("=== Hardware NFC & BLE HTTP Bridge ===") print(f"Listening on: http://127.0.0.1:{PORT}") print("Keep this window open to allow the CRM to read/write tags & scan BLE beacons.") print("--------------------------------------") server = HTTPServer(('127.0.0.1', PORT), NFCBridgeHandler) try: server.serve_forever() except KeyboardInterrupt: print("\nStopping bridge...") server.server_close() if __name__ == '__main__': run()