diff --git a/node_access_control.py b/node_access_control.py index 2be2e18..2c15f68 100644 --- a/node_access_control.py +++ b/node_access_control.py @@ -3,15 +3,110 @@ # (c) Copyright : ae Aeonian Engineering Limited - Hong Kong # (c) Copyright : WIDE di D. Papa - Naples - Italy # ============================================================+ -# Reads NTAG424 badge via PN532, performs challenge/verify -# against identity_tag backend. -# ============================================================+ import utime +import ujson +import ubinascii +MAX_ATTEMPTS = 3 +ALARM_TIMEOUT = 30 # seconds + + +# ============================================================================= +# Display +# ============================================================================= + +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 + + 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) + + # 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) + 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) + # X mark + 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): - """Main loop for access_control node type. Called from boot_manager.""" print('[access_control] starting — NFC:', caps.get('has_nfc')) if not caps.get('has_nfc'): @@ -19,12 +114,109 @@ def run_access_control(cfg, caps): while True: utime.sleep(60) - # TODO: init PN532 - # TODO: wait for badge tap - # TODO: read UID from NTAG424 - # TODO: challenge/verify against cfg['identity_tag_url'] - # TODO: trigger relay on success if has_relay + # 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 + from identity_iot import IdentityIoT + + tag_url = cfg['identity_tag_url'] + attempts = 0 + frame = 0 + alarm_until = 0 while True: - print('[access_control] waiting for badge...') - utime.sleep(2) \ No newline at end of file + 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 + + # Idle animation + _draw_nfc_idle_simple(oled, frame) + frame += 1 + utime.sleep_ms(400) + + # Poll for badge + uid = nfc.read_uid(timeout=300) + if uid is None: + continue + + uid_hex = ubinascii.hexlify(uid).decode() + print('[access_control] badge uid:', uid_hex) + + # 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 + + # Challenge badge against identity_tag + try: + payload = ujson.dumps({ + 'request': 'challenge', + 'badge_uid': uid_hex, + 'node_id': str(cfg['node_id']), + 'tenant_id': str(cfg['tenant_id']), + }) + cipher = aes_cbc_encrypt(payload, 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) + 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] tag error:', e) + utime.sleep(1) + + nfc.release_target() \ No newline at end of file diff --git a/pn532.py b/pn532.py index bfa700c..a43ebf7 100644 --- a/pn532.py +++ b/pn532.py @@ -122,12 +122,16 @@ class PN532_I2C: """Wait for ISO14443A card. Returns UID bytes or None.""" self._send_command([PN532_CMD_INLISTPASSIVETARGET, 0x01, 0x00]) resp = self._read_response(length=32, timeout=timeout) - if not resp or len(resp) < 6: + if not resp: return None - if resp[0] != 0x01: + # resp[0]=cmd_response(0x4B), resp[1]=num_targets, resp[2]=tg + # resp[3][4]=ATQA, resp[5]=SAK, resp[6]=uid_len, resp[7+]=UID + if len(resp) < 8: return None - uid_len = resp[4] - return bytes(resp[5:5 + uid_len]) + if resp[1] != 0x01: + return None + uid_len = resp[6] + return bytes(resp[7:7 + uid_len]) def read_uid(self, timeout=500) -> bytes: return self.read_passive_target(timeout=timeout)