wide/identity-micropython-tag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: v0.3.8 — full access control stack, GRANTED working, NTAG424 ISO7816 read, PN532 driver

Browse Source
This commit is contained in:
Niko
2026-06-07 21:01:41 +02:00
parent 1467fc4886
commit 29300732c1
4 changed files with 167 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
boot_manager.py
View File

@@ -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:

3
config_example.json
View File

@@ -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,

204
node_access_control.py
View File

@@ -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()

34
pn532.py
View File

@@ -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
return None, None