diff --git a/boot_manager.py b/boot_manager.py index 053cf15..7c416c7 100644 --- a/boot_manager.py +++ b/boot_manager.py @@ -213,7 +213,7 @@ def do_checkin(cfg): for attempt in range(3): try: _show(['CHECKIN', 'Registering IoT...', 'attempt ' + str(attempt+1) + '/3', '', '']) - identity.ensure_registered() + reg = identity.ensure_registered() break except Exception as e: print('[checkin] IoT register attempt', attempt + 1, ':', e) @@ -222,8 +222,6 @@ def do_checkin(cfg): raise utime.sleep(3) - # Seconda chiamata idempotente - reg = identity.ensure_registered() uid_short = reg.get('node_uid', '')[:12] + '...' _show(['CHECKIN', 'IoT Node ' + reg.get('result', ''), uid_short, '', '']) utime.sleep(1) @@ -257,7 +255,7 @@ def do_checkin(cfg): for attempt in range(3): try: resp = urequests.post( - cfg['identity_node_url'], + cfg.get('identity_node_url', cfg['identity_tag_url']), headers={ 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token.strip(), @@ -291,6 +289,8 @@ def do_checkin(cfg): def run(): state = 'BOOT' + cfg = None + caps = {'has_oled': False, 'has_nfc': False, 'has_sensor': False, 'has_relay': False} while True: diff --git a/config_example.json b/config_example.json index 778fc9a..ef5a16f 100644 --- a/config_example.json +++ b/config_example.json @@ -13,8 +13,7 @@ "node_type": "access_control", "node_label": "terminal-01", "identity_iot_url": "https://api.parta.app/API/V1/identity_layer/identity_iot/", - "identity_tag_url": "https://api.parta.app/API/V1/identity_layer/identity_tag/", - "identity_node_url": "https://api.parta.app/API/V1/identity_tag/", + "identity_tag_url": "https://api.parta.app/API/V1/identity_tag/", "data_destination_url": "", "pinout": { "i2c0_sda": 4, diff --git a/node_access_control.py b/node_access_control.py index 2c15f68..8f93e42 100644 --- a/node_access_control.py +++ b/node_access_control.py @@ -8,8 +8,9 @@ import utime import ujson import ubinascii -MAX_ATTEMPTS = 3 -ALARM_TIMEOUT = 30 # seconds +MAX_ATTEMPTS = 3 +ALARM_TIMEOUT = 30 # seconds +DISPLAY_TIMEOUT = 30 # seconds idle before OLED off # ============================================================================= @@ -17,40 +18,12 @@ ALARM_TIMEOUT = 30 # seconds # ============================================================================= def _draw_nfc_idle(oled, frame=0): - """NFC waves animation — 4 frames.""" oled.fill(0) - cx, cy = 38, 36 - - # N symbol - oled.text('N', cx - 4, cy + 6) - - # Waves — expanding arcs approximated with pixels - radii = [8, 14, 20] - for i, r in enumerate(radii): - if i < (frame % 4): - # Draw arc (top-right quarter approximation) - for angle in range(0, 90, 5): - import math - x = int(cx + r * math.cos(math.radians(angle))) - y = int(cy - r * math.sin(math.radians(angle))) - if 0 <= x < 128 and 0 <= y < 64: - oled.pixel(x, y, 1) - - oled.text('SCAN BADGE', 30, 56) - oled.show() - - -def _draw_nfc_idle_simple(oled, frame=0): - """Simple NFC animation without math — concentric arcs.""" - oled.fill(0) - - # Animate waves based on frame - waves = frame % 4 cx, cy = 40, 28 - + waves = frame % 4 if waves >= 1: - oled.hline(cx, cy - 6, 4, 1) - oled.vline(cx + 4, cy - 6, 12, 1) + oled.hline(cx, cy - 6, 4, 1) + oled.vline(cx + 4, cy - 6, 12, 1) if waves >= 2: oled.hline(cx - 4, cy - 10, 4, 1) oled.vline(cx + 8, cy - 10, 20, 1) @@ -59,18 +32,13 @@ def _draw_nfc_idle_simple(oled, frame=0): oled.hline(cx - 8, cy - 14, 4, 1) oled.vline(cx + 12, cy - 14, 28, 1) oled.hline(cx - 8, cy + 14, 4, 1) - - # NFC dot oled.fill_rect(cx - 2, cy - 2, 5, 5, 1) - - # Bottom text oled.text('SCAN BADGE', 20, 54) oled.show() def _draw_granted(oled, label=''): oled.fill(0) - # Big checkmark oled.line(20, 35, 32, 47, 1) oled.line(32, 47, 56, 23, 1) oled.line(21, 35, 33, 47, 1) @@ -83,7 +51,6 @@ def _draw_granted(oled, label=''): def _draw_denied(oled, attempts, max_attempts): oled.fill(0) - # X mark oled.line(20, 20, 44, 44, 1) oled.line(44, 20, 20, 44, 1) oled.line(21, 20, 45, 44, 1) @@ -97,8 +64,8 @@ def _draw_alarm(oled, blink=True): oled.fill(1 if blink else 0) fg = 0 if blink else 1 oled.text(' !! ALARM !!', 0, 8, fg) - oled.text(' ACCESS', 0, 28, fg) - oled.text(' DENIED', 0, 40, fg) + oled.text(' ACCESS', 0, 28, fg) + oled.text(' DENIED', 0, 40, fg) oled.show() @@ -114,14 +81,12 @@ def run_access_control(cfg, caps): while True: utime.sleep(60) - # Get hardware refs from boot_manager globals import boot_manager as bm oled = bm._oled nfc = bm._nfc key = cfg['comm_key'].encode() if isinstance(cfg['comm_key'], str) else cfg['comm_key'] iv = cfg['comm_iv'].encode() if isinstance(cfg['comm_iv'], str) else cfg['comm_iv'] - relay_pin = cfg.get('pinout', {}).get('relay_pin', 16) import urequests from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt @@ -130,7 +95,9 @@ def run_access_control(cfg, caps): tag_url = cfg['identity_tag_url'] attempts = 0 frame = 0 - alarm_until = 0 + alarm_until = 0 + last_activity = utime.time() + display_on = True while True: now = utime.time() @@ -144,21 +111,59 @@ def run_access_control(cfg, caps): utime.sleep_ms(500) continue else: - attempts = 0 + attempts = 0 alarm_until = 0 + last_activity = now + if oled and not display_on: + oled.poweron() + display_on = True + + # Display timeout — OLED off after inactivity + if display_on and (now - last_activity) > DISPLAY_TIMEOUT: + if oled: + oled.poweroff() + display_on = False # Idle animation - _draw_nfc_idle_simple(oled, frame) + if display_on: + _draw_nfc_idle(oled, frame) frame += 1 utime.sleep_ms(400) - # Poll for badge - uid = nfc.read_uid(timeout=300) - if uid is None: + # Poll for badge — read UID + NDEF in one shot + try: + uid_hex, ndef_text = nfc.read_ndef_iso(timeout=300) + except Exception: + uid_hex, ndef_text = None, None + + if ndef_text is None: + continue + if uid_hex: + uid_hex = uid_hex.upper() + + # Wake display and reset timeout on any badge detection + if not display_on: + if oled: + oled.poweron() + display_on = True + last_activity = utime.time() + + # Parse NDEF payload + try: + payload = ujson.loads(ndef_text) + except Exception: + print('[access_control] invalid NDEF JSON') continue - uid_hex = ubinascii.hexlify(uid).decode() - print('[access_control] badge uid:', uid_hex) + epk = payload.get('epk', '') + df2 = payload.get('df2', '') + mode = payload.get('mode', 'std') + + if not epk or not df2: + print('[access_control] missing epk or df2') + continue + + print('[access_control] badge uid:', uid_hex, 'mode:', mode) # Get fresh JWT try: @@ -176,15 +181,14 @@ def run_access_control(cfg, caps): utime.sleep(2) continue - # Challenge badge against identity_tag + # Step 1 — Challenge try: - payload = ujson.dumps({ - 'request': 'challenge', - 'badge_uid': uid_hex, - 'node_id': str(cfg['node_id']), + req = ujson.dumps({ + 'request': 'challenge', + 'tag_uid': uid_hex, 'tenant_id': str(cfg['tenant_id']), }) - cipher = aes_cbc_encrypt(payload, key, iv) + cipher = aes_cbc_encrypt(req, key, iv) resp = urequests.post( tag_url, headers={ @@ -196,14 +200,94 @@ def run_access_control(cfg, caps): ) raw = resp.text resp.close() - outer = ujson.loads(raw) + outer = ujson.loads(raw) + ch = ujson.loads(aes_cbc_decrypt(outer['data'], key, iv)) + + if ch.get('status') != 'ok': + raise Exception('challenge: ' + str(ch.get('code', 'error'))) + + nonce_b64 = ch['challenge'] + challenge_token = ch['challenge_token'] + df1_enc = ch.get('df1_encrypted', '') + + except Exception as e: + print('[access_control] challenge error:', e) + attempts += 1 + _draw_denied(oled, attempts, MAX_ATTEMPTS) + utime.sleep(2) + nfc.release_target() + continue + + # Step 2 — Reconstruct DEK, decrypt private key, sign nonce + try: + from identity_iot_aes import aes_cbc_decrypt as aes_dec + + # Decrypt df1 from backend + df1 = ubinascii.a2b_base64(aes_dec(df1_enc, key, iv)) + df2_bytes = ubinascii.a2b_base64(df2) + + # Reconstruct DEK = df1 XOR df2 + dek = bytes([df1[i] ^ df2_bytes[i] for i in range(32)]) + + # Decrypt private key (epk) with DEK — AES-CBC, IV = first 16 bytes of DEK + from ucryptolib import aes as _aes + import ubinascii as _ub + epk_bytes = ubinascii.a2b_base64(epk) + cipher_obj = _aes(dek, 2, dek[:16]) + priv_b64_padded = cipher_obj.decrypt(epk_bytes) + # PKCS7 unpad + pad = priv_b64_padded[-1] + priv_b64 = priv_b64_padded[:-pad].decode('utf-8') + seed = ubinascii.a2b_base64(priv_b64)[:32] + + + + from identity_iot_ed25519 import ed25519_keypair_from_seed + pk, _ = ed25519_keypair_from_seed(seed) + + # Sign nonce + from identity_iot_ed25519 import ed25519_sign + nonce_bytes = ubinascii.a2b_base64(nonce_b64) + signature = ed25519_sign(seed, nonce_bytes) + + sig_b64 = ubinascii.b2a_base64(signature).decode().replace('', '').replace('', '').strip() + + except Exception as e: + print('[access_control] sign error:', e) + attempts += 1 + _draw_denied(oled, attempts, MAX_ATTEMPTS) + utime.sleep(2) + nfc.release_target() + continue + + # Step 3 — Verify signature + try: + req = ujson.dumps({ + 'request': 'verify_signature', + 'tag_uid': uid_hex, + 'signature_b64': sig_b64, + 'challenge_token': challenge_token, + 'tenant_id': str(cfg['tenant_id']), + }) + cipher = aes_cbc_encrypt(req, key, iv) + resp = urequests.post( + tag_url, + headers={ + 'Content-Type': 'application/json', + 'Authorization': 'Bearer ' + token.strip(), + }, + data=ujson.dumps({'data': cipher}), + timeout=10, + ) + raw = resp.text + resp.close() + outer = ujson.loads(raw) result = ujson.loads(aes_cbc_decrypt(outer['data'], key, iv)) if result.get('status') == 'ok': print('[access_control] GRANTED:', uid_hex) attempts = 0 _draw_granted(oled, cfg.get('node_label', '')) - # Trigger relay if caps.get('has_relay') and bm._relay: bm._relay.value(1) utime.sleep_ms(500) @@ -216,7 +300,7 @@ def run_access_control(cfg, caps): utime.sleep(2) except Exception as e: - print('[access_control] tag error:', e) + print('[access_control] verify error:', e) utime.sleep(1) nfc.release_target() \ No newline at end of file diff --git a/pn532.py b/pn532.py index 3454c31..4e1b8f5 100644 --- a/pn532.py +++ b/pn532.py @@ -302,7 +302,7 @@ class PN532_I2C: # 0. Poll for tag — activate it for InDataExchange uid = self.read_passive_target(timeout=timeout) if uid is None: - return None + return None, None, None utime.sleep_ms(100) # 1. Select NDEF Application (D2 76 00 00 85 01 01) @@ -310,37 +310,37 @@ class PN532_I2C: 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00] resp = self._apdu(sel_app) if not resp: - return None + return None, None sw1, sw2 = resp[0], resp[1] if sw1 != 0x90: print('[pn532] select app failed: %02x %02x' % (sw1, sw2)) - return None + return None, None # 2. Select NDEF file (E1 04) sel_file = [0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x04] resp = self._apdu(sel_file) if not resp: - return None + return None, None sw1, sw2 = resp[0], resp[1] if sw1 != 0x90: print('[pn532] select file failed: %02x %02x' % (sw1, sw2)) - return None + return None, None # 3. Read first 2 bytes to get NDEF length read_len = [0x00, 0xB0, 0x00, 0x00, 0x02] resp = self._apdu(read_len) if not resp or len(resp) < 4: - return None + return None, None # resp = [len_hi][len_lo][SW1][SW2] if len(resp) < 4: - return None + return None, None sw1, sw2 = resp[2], resp[3] if sw1 != 0x90: print('[pn532] read len failed: %02x %02x' % (sw1, sw2)) - return None + return None, None ndef_len = (resp[0] << 8) | resp[1] if ndef_len == 0: - return None + return None, None # 4. Read NDEF data (max 200 bytes per read) ndef_data = bytearray() @@ -367,12 +367,12 @@ class PN532_I2C: remaining -= chunk if not ndef_data: - return None + return None, None # 5. Parse NDEF message # ndef_data = clean NDEF bytes, no response code prefix if len(ndef_data) < 3: - return None + return None, None flags = ndef_data[0] type_len = ndef_data[1] @@ -395,19 +395,21 @@ class PN532_I2C: idx += type_len # skip type bytes if idx + payload_len > len(ndef_data): - return None + return None, None payload = ndef_data[idx:idx+payload_len] # TextRecord: byte 0 = status (lang_len), then lang, then text if len(payload) < 1: - return None + return None, None lang_len = payload[0] & 0x3F text_start = 1 + lang_len if text_start > len(payload): - return None + return None, None try: - return bytes(payload[text_start:]).decode('utf-8') + import ubinascii as _ub + uid_hex = _ub.hexlify(uid).decode() + return uid_hex, bytes(payload[text_start:]).decode('utf-8') except Exception: - return None \ No newline at end of file + return None, None \ No newline at end of file