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

feat: boot manager, provisioning server, node stubs v0.3.0

Browse Source
This commit is contained in:
Niko
2026-06-06 17:25:52 +02:00
parent c22a2ea07e
commit 829a3e5f44
10 changed files with 622 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
.gitignore vendored
View File

@@ -1,17 +1,22 @@
# MicroPython compiled
*.mpy
# Secrets
wifi_config.py
config.py
# Runtime files generated on device — never commit
config.json
tenant_*.py
identity_seed.bin
identity_pk.b64
identity_jwt.txt
identity_node_uid.txt
# OS
.DS_Store
Thumbs.db
# Duplicates / backups
*.copy
*copy*
# IDE
# Dev junk
blink.py
oled_test.py
wifitest.py
test.py
test_*.py
# VSCode
.vscode/
.idea/
__pycache__/
*.code-workspace
.micropico

311
boot_manager.py Normal file
View File

@@ -0,0 +1,311 @@
# ============================================================+
# Identity IoT — Boot Manager
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
# (c) Copyright : WIDE di D. Papa - Naples - Italy
# ============================================================+
# State machine: BOOT → CONFIG_CHECK → PROVISIONING | WIFI_CONNECT
# → CHECKIN → OPERATIONAL
# ============================================================+
import ujson
import utime
import network
import machine
import ubinascii
from machine import Pin, I2C
# ---- Optional OLED ----
try:
from ssd1306 import SSD1306_I2C
_i2c = I2C(0, sda=Pin(4), scl=Pin(5))
_oled = SSD1306_I2C(128, 64, _i2c)
HAS_OLED = True
except Exception:
_oled = None
HAS_OLED = False
# ---- Optional NFC (PN532 I2C) ----
try:
from pn532 import PN532_I2C
HAS_NFC = True
except Exception:
HAS_NFC = False
# ---- Optional sensor (DHT22 on GP15) ----
try:
import dht
_dht = dht.DHT22(Pin(15))
HAS_SENSOR = True
except Exception:
HAS_SENSOR = False
# ---- Optional relay (GP16) ----
try:
_relay = Pin(16, Pin.OUT)
HAS_RELAY = True
except Exception:
HAS_RELAY = False
CONFIG_PATH = 'config.json'
STATES = [
'BOOT',
'CAPABILITY_DETECT',
'CONFIG_CHECK',
'PROVISIONING',
'WIFI_CONNECT',
'CHECKIN',
'OPERATIONAL',
'ERROR',
]
# =============================================================================
# Display helpers
# =============================================================================
def _show(lines):
if not HAS_OLED or _oled is None:
print(' | '.join(str(l) for l in lines))
return
_oled.fill(0)
for i, line in enumerate(lines[:5]):
_oled.text(str(line)[:16], 0, i * 12)
_oled.show()
def _show_state(state, detail=''):
_show(['Identity IoT', 'WIDE / AEL', '', state, detail])
# =============================================================================
# Capability detection
# =============================================================================
def detect_capabilities():
caps = {
'has_oled': HAS_OLED,
'has_nfc': HAS_NFC,
'has_sensor': HAS_SENSOR,
'has_relay': HAS_RELAY,
}
print('[caps]', caps)
return caps
# =============================================================================
# Config
# =============================================================================
def load_config():
try:
with open(CONFIG_PATH) as f:
return ujson.load(f)
except OSError:
return None
def save_config(cfg):
with open(CONFIG_PATH, 'w') as f:
ujson.dump(cfg, f)
def validate_config(cfg):
required = ['wifi_ssid', 'wifi_pass', 'comm_key', 'comm_iv',
'base_url', 'tenant_id', 'node_type', 'node_id']
for k in required:
if k not in cfg:
print('[config] missing key:', k)
return False
return True
# =============================================================================
# WiFi
# =============================================================================
def connect_wifi(cfg, timeout=30):
_show_state('WIFI_CONNECT', cfg['wifi_ssid'])
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
utime.sleep(1)
if not wlan.isconnected():
wlan.connect(cfg['wifi_ssid'], cfg['wifi_pass'])
t = timeout
while t > 0:
if wlan.isconnected() and wlan.ifconfig()[0] != '0.0.0.0':
break
utime.sleep(1)
t -= 1
if not wlan.isconnected() or wlan.ifconfig()[0] == '0.0.0.0':
_show_state('ERROR', 'WiFi failed')
raise Exception('WiFi connect failed')
if cfg.get('wifi_static'):
wlan.ifconfig((
cfg['wifi_ip'],
cfg['wifi_mask'],
cfg['wifi_gw'],
cfg['wifi_dns'],
))
ip = wlan.ifconfig()[0]
print('[wifi] connected:', ip)
_show_state('WIFI_OK', ip)
utime.sleep(1)
return wlan
# =============================================================================
# Checkin
# =============================================================================
def do_checkin(cfg):
_show_state('CHECKIN', '')
import urequests
from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt
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']
sn = ubinascii.hexlify(machine.unique_id()).decode()
payload = ujson.dumps({
'request': 'it_node_checkin',
'node_id': cfg['node_id'],
'tenant_id': cfg['tenant_id'],
'node_type': cfg['node_type'],
'node_sn': sn,
})
cipher = aes_cbc_encrypt(payload, key, iv)
body = ujson.dumps({'data': cipher})
for attempt in range(3):
try:
resp = urequests.post(
cfg['base_url'],
headers={'Content-Type': 'application/json'},
data=body,
timeout=20,
)
outer = ujson.loads(resp.text)
resp.close()
clear = aes_cbc_decrypt(outer['data'], key, iv)
result = ujson.loads(clear)
if result.get('status') == 'ok':
print('[checkin] ok')
_show_state('CHECKIN OK', cfg['node_label'])
utime.sleep(1)
return result
raise Exception('checkin status: ' + str(result.get('status')))
except Exception as e:
print('[checkin] attempt', attempt + 1, ':', e)
utime.sleep(3)
raise Exception('checkin failed after 3 attempts')
# =============================================================================
# Boot Manager — main entry point
# =============================================================================
def run():
state = 'BOOT'
while True:
# ------------------------------------------------------------------
if state == 'BOOT':
_show(['Identity IoT', 'WIDE / AEL', '', 'Booting...', ''])
utime.sleep(1)
state = 'CAPABILITY_DETECT'
# ------------------------------------------------------------------
elif state == 'CAPABILITY_DETECT':
caps = detect_capabilities()
_show([
'Identity IoT',
'NFC:' + ('Y' if caps['has_nfc'] else 'N') +
' SNS:' + ('Y' if caps['has_sensor'] else 'N') +
' RLY:' + ('Y' if caps['has_relay'] else 'N'),
'',
'Checking config...',
'',
])
utime.sleep(1)
state = 'CONFIG_CHECK'
# ------------------------------------------------------------------
elif state == 'CONFIG_CHECK':
cfg = load_config()
if cfg and validate_config(cfg):
print('[config] found and valid')
state = 'WIFI_CONNECT'
else:
print('[config] missing or invalid — provisioning mode')
state = 'PROVISIONING'
# ------------------------------------------------------------------
elif state == 'PROVISIONING':
from provisioning_server import run_provisioning
_show_state('PROVISIONING', 'Waiting app...')
cfg = run_provisioning() # blocks until config received
save_config(cfg)
print('[provisioning] config saved — rebooting')
_show_state('CONFIG SAVED', 'Rebooting...')
utime.sleep(2)
machine.reset()
# ------------------------------------------------------------------
elif state == 'WIFI_CONNECT':
try:
connect_wifi(cfg)
state = 'CHECKIN'
except Exception as e:
print('[wifi] error:', e)
_show_state('ERROR', 'WiFi')
utime.sleep(10)
# retry
state = 'WIFI_CONNECT'
# ------------------------------------------------------------------
elif state == 'CHECKIN':
try:
do_checkin(cfg)
state = 'OPERATIONAL'
except Exception as e:
print('[checkin] error:', e)
_show_state('ERROR', 'Checkin')
utime.sleep(15)
state = 'CHECKIN'
# ------------------------------------------------------------------
elif state == 'OPERATIONAL':
node_type = cfg.get('node_type', 'sensor')
_show_state('OPERATIONAL', node_type)
if node_type == 'access_control':
from node_access_control import run_access_control
run_access_control(cfg, caps)
elif node_type == 'sensor':
from node_sensor import run_sensor
run_sensor(cfg, caps)
elif node_type == 'relay':
from node_relay import run_relay
run_relay(cfg, caps)
else:
print('[operational] unknown node_type:', node_type)
_show_state('ERROR', 'node_type?')
utime.sleep(30)
# ------------------------------------------------------------------
elif state == 'ERROR':
_show_state('ERROR', 'halted')
utime.sleep(60)
machine.reset()

13
config_example.json
View File

@@ -6,10 +6,11 @@
"wifi_mask": "255.255.255.0",
"wifi_gw": "192.168.1.1",
"wifi_dns": "8.8.8.8",
"comm_key": "YOUR_32_CHAR_BOOTSTRAP_KEY......",
"comm_iv": "YOUR_16_CHAR_IV.",
"base_url": "https://api.parta.app/API/V1/identity_layer/identity_iot/",
"comm_key": "YOUR_32_CHAR_KEY________________",
"comm_iv": "YOUR_16_CHAR_IV_",
"base_url": "https://api.parta.app/API/V1/identity_tag/",
"tenant_id": 1,
"node_type": "sensor",
"node_label": "pico-01"
}
"node_id": 0,
"node_type": "access_control",
"node_label": "terminal-01"
}

1
identity_iot_aes.py
View File

@@ -10,7 +10,6 @@
import ucryptolib
import ubinascii
def _pkcs7_pad(data: bytes, block_size: int = 16) -> bytes:
pad_len = block_size - (len(data) % block_size)
return data + bytes([pad_len] * pad_len)

BIN
lib/ssd1306.mpy Normal file
View File

Binary file not shown.

130
main.py
View File

@@ -1,128 +1,8 @@
# ============================================================+
# Identity IoT — main.py for Pico W2
# Config from config.json — OLED display integrated
# Identity IoT — main.py
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
# (c) Copyright : WIDE di D. Papa - Naples - Italy
# ============================================================+
import network
import utime
import ujson
from machine import Pin, I2C
from ssd1306 import SSD1306_I2C
from identity_iot import IdentityIoT
# ---- OLED ----
i2c = I2C(0, sda=Pin(4), scl=Pin(5))
oled = SSD1306_I2C(128, 64, i2c)
def show(lines):
oled.fill(0)
for i, line in enumerate(lines):
oled.text(line[:16], 0, i * 12)
oled.show()
def show_scroll(text, row=0, delay=0.04):
"""Scroll long text on a single row."""
if len(text) <= 16:
return
for i in range(len(text) - 15):
oled.fill_rect(0, row * 12, 128, 12, 0)
oled.text(text[i:i+16], 0, row * 12)
oled.show()
utime.sleep(delay)
# ---- Config ----
def load_config():
with open('config.json') as f:
return ujson.load(f)
def connect_wifi(cfg):
show(['Identity IoT', 'WIDE / AEL', '', 'Connecting...'])
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
utime.sleep(2)
if not wlan.isconnected():
wlan.connect(cfg['wifi_ssid'], cfg['wifi_pass'])
timeout = 30
while timeout > 0:
if wlan.isconnected() and wlan.ifconfig()[0] != '0.0.0.0':
break
utime.sleep(1)
timeout -= 1
if not wlan.isconnected() or wlan.ifconfig()[0] == '0.0.0.0':
show(['WiFi FAILED', '', 'Check config', ''])
raise Exception('WiFi failed')
if cfg.get('wifi_static'):
wlan.ifconfig((
cfg['wifi_ip'],
cfg['wifi_mask'],
cfg['wifi_gw'],
cfg['wifi_dns'],
))
ip = wlan.ifconfig()[0]
show(['WiFi OK', ip, '', ''])
utime.sleep(1)
return wlan
# ---- Main ----
def main():
show(['Identity IoT', 'WIDE / AEL', 'v1.0', 'Boot...'])
utime.sleep(1)
cfg = load_config()
connect_wifi(cfg)
identity = IdentityIoT(
base_url = cfg['base_url'],
node_type = cfg.get('node_type', 'sensor'),
node_label = cfg.get('node_label', ''),
tenant_id = cfg['tenant_id'],
comm_key = cfg['comm_key'].encode(),
comm_iv = cfg['comm_iv'].encode(),
wifi_ssid = cfg['wifi_ssid'],
wifi_pass = cfg['wifi_pass'],
)
# Register con retry visibile
for attempt in range(3):
try:
show(['Registering...', 'attempt ' + str(attempt+1) + '/3', '', ''])
reg = identity.ensure_registered()
break
except Exception as e:
print('register failed:', e)
if attempt == 2:
show(['FAILED', 'Check network', '', ''])
raise
utime.sleep(3)
# Register
reg = identity.ensure_registered()
uid_short = reg.get('node_uid', '')[:12] + '...'
show(['Node ' + reg.get('result',''), uid_short, '', ''])
utime.sleep(1)
# Authenticate
show(['Challenge...', '', '', ''])
utime.sleep(0.5)
show(['Signing...', '', 'Please wait', ''])
token = identity.authenticate()
if token:
show(['VERIFIED', '', 'JWT issued', ':)'])
else:
show(['AUTH FAILED', '', 'Check backend', ''])
utime.sleep(2)
# Idle — show node UID
uid = identity.get_node_uid()
show(['Identity IoT', 'Active', uid[:16], uid[16:32]])
main()
import boot_manager
boot_manager.run()

29
node_access_control.py Normal file
View File

@@ -0,0 +1,29 @@
# ============================================================+
# Identity IoT — Node: Access Control
# (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
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'):
print('[access_control] ERROR: no NFC hardware detected')
return
# TODO: init PN532
# TODO: wait for badge tap
# TODO: read UID from NTAG424
# TODO: challenge/verify against cfg['base_url']
# TODO: trigger relay on success if has_relay
while True:
print('[access_control] waiting for badge...')
utime.sleep(2)

20
node_relay.py Normal file
View File

@@ -0,0 +1,20 @@
# ============================================================+
# Identity IoT — Node: Relay
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
# (c) Copyright : WIDE di D. Papa - Naples - Italy
# ============================================================+
import utime
def run_relay(cfg, caps):
"""Main loop for relay node type. Called from boot_manager."""
print('[relay] starting — relay:', caps.get('has_relay'))
# TODO: listen for trigger from backend (polling or badge)
# TODO: activate/deactivate relay pin
# TODO: schedule support
while True:
print('[relay] waiting for trigger...')
utime.sleep(5)

20
node_sensor.py Normal file
View File

@@ -0,0 +1,20 @@
# ============================================================+
# Identity IoT — Node: Sensor
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
# (c) Copyright : WIDE di D. Papa - Naples - Italy
# ============================================================+
import utime
def run_sensor(cfg, caps):
"""Main loop for sensor node type. Called from boot_manager."""
print('[sensor] starting — sensor:', caps.get('has_sensor'))
# TODO: read DHT22 / other sensor
# TODO: POST readings to backend via AES channel
# TODO: polling interval from cfg
while True:
print('[sensor] polling...')
utime.sleep(10)

212
provisioning_server.py Normal file
View File

@@ -0,0 +1,212 @@
# ============================================================+
# Identity IoT — Provisioning Server
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
# (c) Copyright : WIDE di D. Papa - Naples - Italy
# ============================================================+
# Runs when no config.json found on flash.
# 1. Connects WiFi in AP mode (fallback) or waits on existing LAN
# 2. Broadcasts UDP beacon: "identity-iot|{sn}"
# 3. Exposes HTTP on port 8080:
# GET / → node info (plaintext)
# POST /provision → AES-encrypted config payload → saves config.json
# ============================================================+
import network
import socket
import ujson
import ubinascii
import machine
import utime
from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt
# Hardcoded provisioning keys — same in Flutter app
# Change only with coordinated firmware + app update
PROG_KEY = b'IdentityIoTProv!Key_WIDE_AEL_32b' # 32 bytes
PROG_IV = b'IdentityIoTIV16b' # 16 bytes
UDP_PORT = 5000
HTTP_PORT = 8080
BEACON_INTERVAL_MS = 2000
def _get_sn():
return ubinascii.hexlify(machine.unique_id()).decode()
def _get_ip():
wlan = network.WLAN(network.STA_IF)
if wlan.isconnected():
return wlan.ifconfig()[0]
return '0.0.0.0'
def _start_ap():
"""Fallback: start AP so app can connect even without existing WiFi."""
sn = _get_sn()
ssid = 'IdentityIoT-' + sn[:8]
ap = network.WLAN(network.AP_IF)
ap.active(True)
ap.config(essid=ssid, password='provision', security=4) # WPA2
utime.sleep(2)
print('[provisioning] AP started:', ssid, '— ip:', ap.ifconfig()[0])
return ap.ifconfig()[0]
def _beacon_loop(udp_sock, beacon_bytes, stop_flag):
"""Send UDP broadcast beacon. Called between HTTP polls."""
try:
udp_sock.sendto(beacon_bytes, ('255.255.255.255', UDP_PORT))
except Exception as e:
print('[beacon] error:', e)
def _parse_http_request(conn):
"""Read HTTP request from socket. Returns (method, path, body)."""
try:
data = b''
conn.settimeout(5)
while True:
chunk = conn.recv(1024)
if not chunk:
break
data += chunk
if b'\r\n\r\n' in data:
# Check if we have full body
header_end = data.index(b'\r\n\r\n') + 4
headers_raw = data[:header_end].decode('utf-8', 'ignore')
body = data[header_end:]
content_length = 0
for line in headers_raw.split('\r\n'):
if line.lower().startswith('content-length:'):
content_length = int(line.split(':')[1].strip())
while len(body) < content_length:
chunk = conn.recv(1024)
if not chunk:
break
body += chunk
first_line = headers_raw.split('\r\n')[0]
parts = first_line.split(' ')
method = parts[0] if len(parts) > 0 else 'GET'
path = parts[1] if len(parts) > 1 else '/'
return method, path, body.decode('utf-8', 'ignore')
except Exception as e:
print('[http] parse error:', e)
return 'GET', '/', ''
def _http_response(conn, status, body):
resp = (
'HTTP/1.1 ' + status + '\r\n'
'Content-Type: application/json\r\n'
'Content-Length: ' + str(len(body)) + '\r\n'
'Connection: close\r\n'
'\r\n' + body
)
conn.sendall(resp.encode())
def run_provisioning():
"""
Block until a valid config is received from the Flutter app.
Returns config dict ready to save.
"""
sn = _get_sn()
print('[provisioning] SN:', sn)
# Try STA first, fall back to AP
wlan_sta = network.WLAN(network.STA_IF)
if not wlan_sta.isconnected():
ip = _start_ap()
else:
ip = wlan_sta.ifconfig()[0]
print('[provisioning] using existing WiFi, ip:', ip)
beacon_msg = ujson.dumps({
'type': 'identity-iot',
'sn': sn,
'ip': ip,
'port': HTTP_PORT,
}).encode()
# UDP socket for beacon
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
udp.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
except Exception:
pass
udp.setblocking(False)
# HTTP server socket
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(('0.0.0.0', HTTP_PORT))
srv.listen(1)
srv.setblocking(False)
print('[provisioning] HTTP listening on', ip + ':' + str(HTTP_PORT))
print('[provisioning] UDP beacon on port', UDP_PORT)
last_beacon = 0
received_cfg = None
while received_cfg is None:
# Beacon
now = utime.ticks_ms()
if utime.ticks_diff(now, last_beacon) >= BEACON_INTERVAL_MS:
try:
udp.sendto(beacon_msg, ('255.255.255.255', UDP_PORT))
except Exception:
pass
last_beacon = now
# HTTP accept (non-blocking)
try:
conn, addr = srv.accept()
print('[provisioning] connection from', addr)
method, path, body = _parse_http_request(conn)
if method == 'GET' and path == '/':
info = ujson.dumps({'sn': sn, 'ip': ip, 'status': 'awaiting_provision'})
_http_response(conn, '200 OK', info)
elif method == 'POST' and path == '/provision':
try:
outer = ujson.loads(body)
cipher_b64 = outer.get('data', '')
if not cipher_b64:
raise Exception('missing data')
clear = aes_cbc_decrypt(cipher_b64, PROG_KEY, PROG_IV)
cfg = ujson.loads(clear)
# Basic validation
required = ['wifi_ssid', 'wifi_pass', 'comm_key', 'comm_iv',
'base_url', 'tenant_id', 'node_type', 'node_id']
for k in required:
if k not in cfg:
raise Exception('missing field: ' + k)
_http_response(conn, '200 OK', ujson.dumps({'status': 'ok', 'sn': sn}))
conn.close()
received_cfg = cfg
print('[provisioning] config received OK')
except Exception as e:
print('[provisioning] provision error:', e)
_http_response(conn, '400 Bad Request', ujson.dumps({'status': 'error', 'msg': str(e)}))
conn.close()
else:
_http_response(conn, '404 Not Found', '{}')
conn.close()
except OSError:
# No connection yet — normal in non-blocking mode
utime.sleep_ms(100)
udp.close()
srv.close()
return received_cfg