# ============================================================+ # Identity IoT — Node: Access Control # (c) Copyright : ae Aeonian Engineering Limited - Hong Kong # (c) Copyright : WIDE di D. Papa - Naples - Italy # ============================================================+ import utime import ujson import ubinascii MAX_ATTEMPTS = 3 ALARM_TIMEOUT = 30 # seconds DISPLAY_TIMEOUT = 30 # seconds idle before OLED off # ============================================================================= # Display # ============================================================================= def _draw_nfc_idle(oled, frame=0): oled.fill(0) 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) if waves >= 2: oled.hline(cx - 4, cy - 10, 4, 1) oled.vline(cx + 8, cy - 10, 20, 1) oled.hline(cx - 4, cy + 10, 4, 1) if waves >= 3: oled.hline(cx - 8, cy - 14, 4, 1) oled.vline(cx + 12, cy - 14, 28, 1) oled.hline(cx - 8, cy + 14, 4, 1) oled.fill_rect(cx - 2, cy - 2, 5, 5, 1) oled.text('SCAN BADGE', 20, 54) oled.show() def _draw_granted(oled, label=''): oled.fill(0) oled.line(20, 35, 32, 47, 1) oled.line(32, 47, 56, 23, 1) oled.line(21, 35, 33, 47, 1) oled.line(33, 47, 57, 23, 1) oled.text(' GRANTED', 14, 52) if label: oled.text(label[:16], 0, 0) oled.show() def _draw_denied(oled, attempts, max_attempts): oled.fill(0) oled.line(20, 20, 44, 44, 1) oled.line(44, 20, 20, 44, 1) oled.line(21, 20, 45, 44, 1) oled.line(45, 20, 21, 44, 1) oled.text(' DENIED', 14, 52) oled.text(str(attempts) + '/' + str(max_attempts), 104, 0) oled.show() 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.show() # ============================================================================= # Main loop # ============================================================================= def run_access_control(cfg, caps): print('[access_control] starting — NFC:', caps.get('has_nfc')) if not caps.get('has_nfc'): print('[access_control] ERROR: no NFC hardware detected') while True: utime.sleep(60) 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'] import urequests from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt from identity_iot import IdentityIoT tag_url = cfg['identity_tag_url'] attempts = 0 frame = 0 alarm_until = 0 last_activity = utime.time() display_on = True while True: now = utime.time() # ALARM state if attempts >= MAX_ATTEMPTS: if alarm_until == 0: alarm_until = now + ALARM_TIMEOUT if now < alarm_until: _draw_alarm(oled, blink=(now % 2 == 0)) utime.sleep_ms(500) continue else: 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 if display_on: _draw_nfc_idle(oled, frame) frame += 1 utime.sleep_ms(400) # 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 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: identity = IdentityIoT( base_url = cfg['identity_iot_url'], node_type = cfg.get('node_type', 'access_control'), tenant_id = cfg['tenant_id'], node_id = cfg['node_id'], comm_key = key, comm_iv = iv, ) token = identity.authenticate() except Exception as e: print('[access_control] auth error:', e) utime.sleep(2) continue # Step 1 — Challenge try: req = ujson.dumps({ 'request': 'challenge', 'tag_uid': uid_hex, '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) 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', '')) if caps.get('has_relay') and bm._relay: bm._relay.value(1) utime.sleep_ms(500) bm._relay.value(0) utime.sleep(2) else: attempts += 1 print('[access_control] DENIED:', uid_hex, 'attempts:', attempts) _draw_denied(oled, attempts, MAX_ATTEMPTS) utime.sleep(2) except Exception as e: print('[access_control] verify error:', e) utime.sleep(1) nfc.release_target()