commit d1a28d989c103a0ce7f133ce915e8ff5b68d570f Author: Niko Date: Mon Jun 8 18:27:38 2026 +0200 Identity IoT: initial release — boot manager, provisioning, sensor/relay diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5c012cd --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +# Runtime files generated on device — never commit +config.json +identity_seed.bin +identity_pk.b64 +identity_jwt.txt +identity_node_uid.txt + +# Duplicates / backups +*.copy +*copy* + +# Dev junk +blink.py +oled_test.py +wifitest.py +test.py +test_*.py + +# VSCode +.vscode/ +*.code-workspace +.micropico diff --git a/README.md b/README.md new file mode 100644 index 0000000..fd69c08 --- /dev/null +++ b/README.md @@ -0,0 +1,118 @@ +# identity-micropython + +**Identity IoT — MicroPython module for Pico W / ESP32** + +(c) WIDE di D. Papa — Naples, Italy +(c) æ Aeonian Engineering Limited — Hong Kong + +--- + +## Overview + +Ed25519 challenge/response authentication for IoT nodes. +Compatible with `identity_layer/identity_iot` PHP backend. + +Private key never leaves the device — stored on flash, never transmitted. +Node UID is SHA-256 hashed before sending — raw UID never reaches the server. + +--- + +## Files + +| File | Description | +|---|---| +| `identity_iot.py` | Main service — register, challenge, verify | +| `identity_iot_ed25519.py` | Pure-Python Ed25519 (PoC — replace with natmod in production) | +| `identity_iot_aes.py` | AES-CBC ITEMS envelope (uses `ucryptolib`) | +| `example_main.py` | Pico W example | +| `package.json` | `mip` install manifest | + +--- + +## Install via mip (on-board, requires WiFi) + +```python +#Pico W / Pico 2 W +import network, utime +wlan = network.WLAN(network.STA_IF) +wlan.active(True) +wlan.connect('YOUR_SSD', 'YOUR_WIFI_PASSWORD') +while not wlan.isconnected(): utime.sleep(1) + +import mip +mip.install("https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/package.json") + +# Then install main.py to root +mip.install("https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/main.py", target="/") + +``` + +## Install via mpremote (from host) + +```bash +mpremote cp identity_iot.py :identity_iot.py +mpremote cp identity_iot_ed25519.py :identity_iot_ed25519.py +mpremote cp identity_iot_aes.py :identity_iot_aes.py +``` + +--- + +## Usage + +```python +from identity_iot import IdentityIoT + +identity = IdentityIoT( + base_url = 'https://ap.parta.app/API/V1/identity_layer/identity_iot/index.php', + node_type = 'sensor', + node_label = 'pico-01', + tenant_id = 1, + comm_key = b'your-32-char-bootstrap-key......', + comm_iv = b'your-16-char-iv.', +) + +# Register (idempotent — safe every boot) +identity.ensure_registered() + +# Authenticate — returns JWT +token = identity.authenticate() +``` + +--- + +## Architecture notes + +### PoC vs Production + +| Component | PoC (now) | Production | +|---|---|---| +| Ed25519 | Pure Python (~2-3s/sign) | natmod C (~10ms) | +| SHA-512 | Double SHA-256 (approximation) | natmod C (proper) | +| AES-CBC | `ucryptolib` built-in | same | +| Transport | `urequests` plain/encrypted | same | + +**Important:** the pure-Python Ed25519 uses double-SHA256 as SHA-512 approximation. +This is NOT cryptographically correct for production — it works for PoC flow testing only. +Replace `identity_iot_ed25519.py` with the natmod version before production deployment. + +### Key storage + +``` +/identity_seed.bin ← 32-byte Ed25519 seed (never transmitted) +/identity_pk.b64 ← public key base64 +/identity_jwt.txt ← current JWT (refreshed on expiry) +/identity_node_uid.txt ← SHA-256(machine.unique_id()) hex +``` + +--- + +## Backend + +Requires [identity_layer](https://git.aeonianengineering.net/wide/identity_layer) PHP backend with `identity_iot` module. + +--- + +## IP Notice + +- Rights holder: WIDE di D. Papa — P.IVA IT03744531215 +- Exclusive worldwide (RoW) licensee: æ Aeonian Engineering Limited — Hong Kong diff --git a/boot_manager.py b/boot_manager.py new file mode 100644 index 0000000..c0ac318 --- /dev/null +++ b/boot_manager.py @@ -0,0 +1,353 @@ +# ============================================================+ +# 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 + +# Hardware instances — initialised in detect_capabilities() after config load +_oled = None +_nfc = None +_dht = None +_relay = None +HAS_OLED = False +HAS_NFC = False +HAS_SENSOR = False +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(cfg=None): + global _oled, _nfc, _dht, _relay + global HAS_OLED, HAS_NFC, HAS_SENSOR, HAS_RELAY + + pinout = cfg.get('pinout', {}) if cfg else {} + i2c0_sda = pinout.get('i2c0_sda', 4) + i2c0_scl = pinout.get('i2c0_scl', 5) + i2c1_sda = pinout.get('i2c1_sda', 6) + i2c1_scl = pinout.get('i2c1_scl', 7) + dht_pin = pinout.get('dht_pin', 15) + relay_pin = pinout.get('relay_pin', 16) + + # OLED + try: + from ssd1306 import SSD1306_I2C + _i2c0 = I2C(0, sda=Pin(i2c0_sda), scl=Pin(i2c0_scl)) + _oled = SSD1306_I2C(128, 64, _i2c0) + HAS_OLED = True + except Exception: + _oled = None + HAS_OLED = False + + # NFC (PN532 as provisioning dongle only — not present in normal operation) + try: + from pn532 import PN532_I2C + _i2c1 = I2C(1, sda=Pin(i2c1_sda), scl=Pin(i2c1_scl)) + _nfc = PN532_I2C(_i2c1) + HAS_NFC = True + except Exception as e: + print('[nfc] not detected:', e) + _nfc = None + HAS_NFC = False + + # Sensor (DHT22) + try: + import dht + _dht = dht.DHT22(Pin(dht_pin)) + HAS_SENSOR = True + except Exception: + _dht = None + HAS_SENSOR = False + + # Relay + try: + _relay = Pin(relay_pin, Pin.OUT) + HAS_RELAY = True + except Exception: + _relay = None + HAS_RELAY = False + + 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', + 'tenant_id', 'node_type', 'node_id', 'identity_iot_url'] + 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(2) + + 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(3) + return wlan + + +# ============================================================================= +# Checkin +# ============================================================================= + +def do_checkin(cfg): + _show_state('CHECKIN', '') + from identity_iot import IdentityIoT + + 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'] + + identity = IdentityIoT( + base_url = cfg['identity_iot_url'], + node_type = cfg.get('node_type', 'sensor'), + node_label = cfg.get('node_label', ''), + tenant_id = cfg['tenant_id'], + node_id = cfg['node_id'], + comm_key = key, + comm_iv = iv, + wifi_ssid = cfg['wifi_ssid'], + wifi_pass = cfg['wifi_pass'], + ) + + # Register — retry loop + for attempt in range(3): + try: + _show(['CHECKIN', 'Registering...', 'attempt ' + str(attempt+1) + '/3', '', '']) + reg = identity.ensure_registered() + break + except Exception as e: + print('[checkin] register attempt', attempt + 1, ':', e) + if attempt == 2: + _show_state('ERROR', 'Check network') + raise + utime.sleep(3) + + uid_short = reg.get('node_uid', '')[:12] + '...' + _show(['CHECKIN', 'Node ' + reg.get('result', ''), uid_short, '', '']) + utime.sleep(1) + + # Authenticate → JWT + _show(['CHECKIN', 'Challenge...', '', '', '']) + utime.sleep(0.5) + _show(['CHECKIN', 'Signing...', '', 'Please wait', '']) + token = identity.authenticate() + + if not token: + _show_state('ERROR', 'Auth failed') + raise Exception('authenticate returned no token') + + print('[checkin] JWT issued') + _show_state('CHECKIN OK', cfg.get('node_label', '')) + utime.sleep(1) + + return {'status': 'ok', 'token': token, 'identity': identity} + + +# ============================================================================= +# Boot Manager — main entry point +# ============================================================================= + +def run(): + state = 'BOOT' + cfg = None + caps = {'has_oled': False, 'has_nfc': False, 'has_sensor': False, 'has_relay': False} + + while True: + + # ------------------------------------------------------------------ + if state == 'BOOT': + _show(['Identity IoT', 'WIDE / AEL', '', 'Booting...', '']) + 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 = 'CAPABILITY_DETECT' + else: + print('[config] missing or invalid — provisioning mode') + caps = detect_capabilities() + state = 'PROVISIONING' + + # ------------------------------------------------------------------ + elif state == 'CAPABILITY_DETECT': + caps = detect_capabilities(cfg) + _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'), + '', + 'READY', + '', + ]) + utime.sleep(1) + state = 'WIFI_CONNECT' + + # ------------------------------------------------------------------ + elif state == 'PROVISIONING': + if caps.get('has_nfc'): + from provisioning_nfc import run_provisioning_nfc + _show(['PROVISIONING', 'NFC mode', '', 'Tap writer badge', '']) + cfg = run_provisioning_nfc(_nfc) + else: + from provisioning_server import run_provisioning + wlan_sta = network.WLAN(network.STA_IF) + _ip = wlan_sta.ifconfig()[0] if wlan_sta.isconnected() else '0.0.0.0' + _show(['PROVISIONING', 'IP: ' + _ip, 'HTTP:8080', 'UDP :5000', 'Waiting app...']) + cfg = run_provisioning() + 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) + state = 'WIFI_CONNECT' + + # ------------------------------------------------------------------ + elif state == 'CHECKIN': + try: + import gc + gc.collect() + checkin_result = do_checkin(cfg) + identity = checkin_result.get('identity') + 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 == '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() diff --git a/config_example.json b/config_example.json new file mode 100644 index 0000000..7fa4605 --- /dev/null +++ b/config_example.json @@ -0,0 +1,35 @@ +{ + "wifi_ssid": "YOUR_SSID", + "wifi_pass": "YOUR_PASSWORD", + "wifi_static": false, + "wifi_ip": "192.168.1.50", + "wifi_mask": "255.255.255.0", + "wifi_gw": "192.168.1.1", + "wifi_dns": "8.8.8.8", + "comm_key": "YOUR_32_CHAR_KEY________________", + "comm_iv": "YOUR_16_CHAR_IV_", + "tenant_id": "1", + "node_id": "0", + "node_type": "sensor", + "node_label": "node-01", + "identity_iot_url": "https://api.yourhost.com/API/V1/identity_layer/identity_iot/", + "data_destination_url": "", + "pinout": { + "i2c0_sda": 4, + "i2c0_scl": 5, + "i2c1_sda": 6, + "i2c1_scl": 7, + "dht_pin": 15, + "relay_pin": 16, + "rain_pin": 26, + "soil_pin": 27, + "pot_pin": 28, + "switch_pins": [17, 18], + "rotary_a": 19, + "rotary_b": 20, + "rotary_sw": 21, + "gps_uart": 1, + "gps_tx": 8, + "gps_rx": 9 + } +} diff --git a/identity_iot.py b/identity_iot.py new file mode 100644 index 0000000..7481adb --- /dev/null +++ b/identity_iot.py @@ -0,0 +1,305 @@ +# ============================================================+ +# Identity IoT — MicroPython Module +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ +# Ed25519 challenge/response authentication for IoT nodes. +# Compatible with identity_layer/identity_iot PHP backend. +# +# Install via mip: +# import mip +# mip.install("https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/package.json") +# +# Usage: +# from identity_iot import IdentityIoT +# identity = IdentityIoT(base_url="https://api.parta.app/API/V1/identity_layer/identity_iot/") +# identity.ensure_registered() +# token = identity.authenticate() +# ============================================================+ + +import ujson +import ubinascii +import uhashlib +import urandom +import machine +import network +import utime + +try: + import uurequests as requests +except ImportError: + import urequests as requests + +from identity_iot_ed25519 import ed25519_sign, ed25519_keypair_from_seed + + +class IdentityIoT: + """ + Identity IoT authentication service for MicroPython. + Device-bound Ed25519 keypair — private key never leaves the device. + """ + + _F_SEED = '/identity_seed.bin' + _F_PK = '/identity_pk.b64' + _F_JWT = '/identity_jwt.txt' + _F_NODEID = '/identity_node_uid.txt' + + def __init__( + self, + base_url: str, + app_id: str = 'identity_iot', + node_type: str = 'sensor', + node_label: str = None, + tenant_id = 0, + node_id: str = None, + comm_key: bytes = None, + comm_iv: bytes = None, + wifi_ssid: str = None, + wifi_pass: str = None, + ): + self.base_url = base_url + self.app_id = app_id + self.node_type = node_type + self.node_label = node_label + self.tenant_id = tenant_id + self.node_id = node_id + self._comm_key = comm_key + self._comm_iv = comm_iv + self._wifi_ssid = wifi_ssid + self._wifi_pass = wifi_pass + + # ========================================================================= + # Public API + # ========================================================================= + + def ensure_registered(self): + """Register node if not already registered. Idempotent.""" + pk = self._load_pk() + if pk is None: + pk = self._generate_keypair() + node_uid = self._get_node_uid() + resp = self._post({ + 'request': 'register_node', + 'public_key_b64': pk, + 'node_uid': node_uid, + 'node_type': self.node_type, + 'node_label': self.node_label or '', + 'tenant_id': self.tenant_id, + }) + if resp.get('status') != 'ok': + raise Exception('register_node failed: ' + str(resp)) + return resp + + def authenticate(self): + """Full challenge/response flow. Returns JWT string.""" + + #Not using cashing + #jwt = self._load_jwt() + #if jwt and not self._is_jwt_expired(jwt): + # return jwt + + node_uid = self._get_node_uid() + + resp = self._post({'request': 'challenge', 'node_uid': node_uid, 'tenant_id': self.tenant_id}) + if resp.get('status') != 'ok': + raise Exception('challenge failed: ' + str(resp)) + + challenge_id = resp['challenge_id'] + nonce_b64 = resp['nonce_b64'] + + nonce_bytes = ubinascii.a2b_base64(_normalize_b64(nonce_b64)) + seed = self._load_seed() + signature = ed25519_sign(seed, nonce_bytes) + sig_b64 = ubinascii.b2a_base64(signature).decode().strip() + + resp = self._post({ + 'request': 'verify_signature', + 'challenge_id': challenge_id, + 'signature_b64': sig_b64, + 'node_uid': node_uid, + 'tenant_id': self.tenant_id, + }) + if resp.get('status') != 'ok': + raise Exception('verify_signature failed: ' + str(resp)) + + token = resp.get('token', '') + # JWT not cached — fresh challenge/verify every time + # if token: + # self._save_jwt(token) + return token + + def get_node_uid(self): + return self._get_node_uid() + + # ========================================================================= + # Keypair management + # ========================================================================= + + def _generate_keypair(self): + seed = bytes(urandom.getrandbits(8) for _ in range(32)) + pk, _ = ed25519_keypair_from_seed(seed) + pk_b64 = ubinascii.b2a_base64(pk).decode().strip() + with open(self._F_SEED, 'wb') as f: + f.write(seed) + with open(self._F_PK, 'w') as f: + f.write(pk_b64) + return pk_b64 + + def _load_seed(self): + try: + with open(self._F_SEED, 'rb') as f: + return f.read() + except OSError: + raise StateError('No seed found. Call ensure_registered() first.') + + def _load_pk(self): + try: + with open(self._F_PK, 'r') as f: + return f.read().strip() + except OSError: + return None + + # ========================================================================= + # Node UID + # ========================================================================= + + def _get_node_uid(self): + try: + with open(self._F_NODEID, 'r') as f: + return f.read().strip() + except OSError: + pass + # Fresh UID — hardware ID + random entropy so each new registration is unique + raw_uid = machine.unique_id() + entropy = bytes([urandom.getrandbits(8) for _ in range(16)]) + h = uhashlib.sha256(raw_uid + entropy) + uid_hex = ubinascii.hexlify(h.digest()).decode() + with open(self._F_NODEID, 'w') as f: + f.write(uid_hex) + return uid_hex + + # ========================================================================= + # JWT management + # ========================================================================= + + def _load_jwt(self): + try: + with open(self._F_JWT, 'r') as f: + return f.read().strip() + except OSError: + return None + + def _save_jwt(self, token): + with open(self._F_JWT, 'w') as f: + f.write(token) + + def _is_jwt_expired(self, jwt): + try: + parts = jwt.split('.') + if len(parts) != 3: + return True + payload_b64 = _normalize_b64(parts[1]) + payload_json = ubinascii.a2b_base64(payload_b64).decode() + payload = ujson.loads(payload_json) + exp = payload.get('exp', 0) + return utime.time() >= exp + except Exception: + return True + + # ========================================================================= + # WiFi reconnect + # ========================================================================= + + def _ensure_wifi(self): + wlan = network.WLAN(network.STA_IF) + if wlan.isconnected(): + return True + if not self._wifi_ssid: + return False + print('WiFi lost, reconnecting...') + wlan.active(True) + wlan.connect(self._wifi_ssid, self._wifi_pass) + timeout = 15 + while not wlan.isconnected() and timeout > 0: + utime.sleep(1) + timeout -= 1 + return wlan.isconnected() + + # ========================================================================= + # HTTP transport + # ========================================================================= + + def _post(self, inner: dict) -> dict: + if self._comm_key and self._comm_iv: + return self._post_encrypted(inner) + return self._post_plain(inner) + + def _post_plain(self, inner: dict) -> dict: + for attempt in range(3): + try: + if not self._ensure_wifi(): + raise Exception('WiFi not connected') + body = ujson.dumps({'data': ujson.dumps(inner)}) + resp = requests.post( + self.base_url, + headers={'Content-Type': 'application/json'}, + data=body, + timeout=20 + ) + outer = ujson.loads(resp.text) + resp.close() + if 'data' in outer: + try: + return ujson.loads(outer['data']) + except Exception: + return outer + return outer + except Exception as e: + print('plain retry', attempt + 1, ':', e) + utime.sleep(3) + raise Exception('POST failed after 3 attempts') + + def _post_encrypted(self, inner: dict) -> dict: + from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt + for attempt in range(3): + try: + if not self._ensure_wifi(): + raise Exception('WiFi not connected') + inner_json = ujson.dumps(inner) + cipher_b64 = aes_cbc_encrypt(inner_json, self._comm_key, self._comm_iv) + body = ujson.dumps({'data': cipher_b64}) + resp = requests.post( + self.base_url, + headers={'Content-Type': 'application/json'}, + data=body, + timeout=20 + ) + print('status:', resp.status_code) + print('raw:', resp.text[:100]) + outer = ujson.loads(resp.text) + resp.close() + if 'data' not in outer: + raise Exception('Missing outer.data') + clear = aes_cbc_decrypt(outer['data'], self._comm_key, self._comm_iv) + return ujson.loads(clear) + except Exception as e: + print('enc retry', attempt + 1, ':', e) + utime.sleep(3) + raise Exception('POST failed after 3 attempts') + + +# ============================================================================= +# Utility +# ============================================================================= + +def _normalize_b64(s: str) -> str: + s = s.strip().replace('\n', '').replace('\r', '').replace(' ', '') + s = s.replace('-', '+').replace('_', '/') + pad = len(s) % 4 + if pad: + s += '=' * (4 - pad) + return s + + +class StateError(Exception): + pass \ No newline at end of file diff --git a/identity_iot_aes.py b/identity_iot_aes.py new file mode 100644 index 0000000..903139e --- /dev/null +++ b/identity_iot_aes.py @@ -0,0 +1,63 @@ +# ============================================================+ +# Identity IoT — AES-CBC ITEMS Envelope +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ +# AES-CBC encryption/decryption compatible with PHP ITEMSEncrypter. +# Uses ucryptolib (built-in on Pico W / ESP32). +# ============================================================+ + +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) + + +def _pkcs7_unpad(data: bytes) -> bytes: + pad_len = data[-1] + return data[:-pad_len] + + +def aes_cbc_encrypt(plaintext: str, key: bytes, iv: bytes) -> str: + """ + Encrypt plaintext string with AES-CBC. + Returns base64-encoded ciphertext. + Compatible with PHP ITEMSEncrypter::encryptAES(). + """ + if isinstance(key, str): + key = key.encode() + if isinstance(iv, str): + iv = iv.encode() + if isinstance(plaintext, str): + plaintext = plaintext.encode('utf-8') + + padded = _pkcs7_pad(plaintext) + cipher = ucryptolib.aes(key, 2, iv) # mode 2 = CBC + encrypted = cipher.encrypt(padded) + return ubinascii.b2a_base64(encrypted).decode().strip() + + +def aes_cbc_decrypt(ciphertext_b64: str, key: bytes, iv: bytes) -> str: + """ + Decrypt base64-encoded AES-CBC ciphertext. + Returns plaintext string. + Compatible with PHP ITEMSEncrypter::decryptAES(). + """ + if isinstance(key, str): + key = key.encode() + if isinstance(iv, str): + iv = iv.encode() + + # Normalize base64 + s = ciphertext_b64.strip().replace('\n', '').replace('\r', '') + s = s.replace('-', '+').replace('_', '/') + pad = len(s) % 4 + if pad: + s += '=' * (4 - pad) + + encrypted = ubinascii.a2b_base64(s) + cipher = ucryptolib.aes(key, 2, iv) # mode 2 = CBC + decrypted = cipher.decrypt(encrypted) + return _pkcs7_unpad(decrypted).decode('utf-8') diff --git a/identity_iot_ed25519.py b/identity_iot_ed25519.py new file mode 100644 index 0000000..721097f --- /dev/null +++ b/identity_iot_ed25519.py @@ -0,0 +1,126 @@ +# ============================================================+ +# Identity IoT — Ed25519 for MicroPython +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ +# Based on Bernstein et al. reference implementation (public domain) +# Iterative scalarmult — no recursion, safe for MicroPython stack +# ============================================================+ + +from sha512 import sha512 + +_q = 2**255 - 19 +_l = 2**252 + 27742317777372353535851937790883648493 +_d = -121665 * pow(121666, _q - 2, _q) % _q +_I = pow(2, (_q - 1) // 4, _q) + + +def _xrecover(y): + x2 = (y * y - 1) * pow(_d * y * y + 1, _q - 2, _q) + x = pow(x2, (_q + 3) // 8, _q) + if (x * x - x2) % _q != 0: + x = x * _I % _q + if x % 2 != 0: + x = _q - x + return x + + +_By = 4 * pow(5, _q - 2, _q) % _q +_Bx = _xrecover(_By) +_B = [_Bx % _q, _By % _q, 1, _Bx * _By % _q] + + +def _edwards_add(P, Q): + x1, y1, z1, t1 = P + x2, y2, z2, t2 = Q + A = (y1 - x1) * (y2 - x2) % _q + B = (y1 + x1) * (y2 + x2) % _q + C = t1 * 2 * _d * t2 % _q + D = z1 * 2 * z2 % _q + E = B - A; F = D - C; G = D + C; H = B + A + return [E * F % _q, G * H % _q, F * G % _q, E * H % _q] + + +def _scalarmult(P, e): + # Iterative double-and-add — no recursion + Q = [0, 1, 1, 0] # neutral element + while e > 0: + if e & 1: + Q = _edwards_add(Q, P) + P = _edwards_add(P, P) + e >>= 1 + return Q + + +def _encodeint(y): + bits = [(y >> i) & 1 for i in range(256)] + return bytes([sum([bits[i * 8 + j] << j for j in range(8)]) for i in range(32)]) + + +def _encodepoint(P): + x, y, z, _ = P + zi = pow(z, _q - 2, _q) + x = x * zi % _q + y = y * zi % _q + bits = [(y >> i) & 1 for i in range(255)] + [x & 1] + return bytes([sum([bits[i * 8 + j] << j for j in range(8)]) for i in range(32)]) + + +def _bit(h, i): + return (h[i // 8] >> (i % 8)) & 1 + + +def _hint(m): + return int.from_bytes(sha512(m), 'little') + + +def _decodepoint(s): + y = sum(_bit(s, i) * 2 ** i for i in range(255)) + x = _xrecover(y) + if x & 1 != _bit(s, 255): + x = _q - x + return [x, y, 1, x * y % _q] + + +# ============================================================ +# Public API +# ============================================================ + +def ed25519_keypair_from_seed(seed: bytes): + if len(seed) != 32: + raise ValueError('Seed must be 32 bytes') + h = sha512(seed) + a = 2 ** 254 + sum(2 ** i * _bit(h, i) for i in range(3, 254)) + A = _scalarmult(_B, a) + pk = _encodepoint(A) + return pk, seed + pk + + +def ed25519_sign(seed: bytes, message: bytes) -> bytes: + if len(seed) != 32: + raise ValueError('Seed must be 32 bytes') + h = sha512(seed) + a = 2 ** 254 + sum(2 ** i * _bit(h, i) for i in range(3, 254)) + pk = ed25519_keypair_from_seed(seed)[0] + r = _hint(h[32:64] + message) + R = _scalarmult(_B, r) + S = (r + _hint(_encodepoint(R) + pk + message) * a) % _l + return _encodepoint(R) + _encodeint(S) + + +def ed25519_verify(pk: bytes, message: bytes, signature: bytes) -> bool: + if len(signature) != 64 or len(pk) != 32: + return False + try: + R_enc = signature[:32] + S = int.from_bytes(signature[32:], 'little') + if S >= _l: + return False + A = _decodepoint(pk) + R = _decodepoint(R_enc) + h = _hint(R_enc + pk + message) + lhs = _scalarmult(_B, 8 * S) + rhs = _edwards_add(_scalarmult(R, 8), _scalarmult(A, 8 * h)) + return _encodepoint(lhs) == _encodepoint(rhs) + except Exception: + return False diff --git a/install.py b/install.py new file mode 100644 index 0000000..6761347 --- /dev/null +++ b/install.py @@ -0,0 +1,25 @@ +# ============================================================+ +# Identity IoT — install.py +# Run once on a fresh Pico W to install all dependencies. +# Edit SSID and PASS before running. +# ============================================================+ + +SSID = 'YOUR_SSID' +PASS = 'YOUR_PASSWORD' + +# ---- WiFi ---- +import network, utime +wlan = network.WLAN(network.STA_IF) +wlan.active(True) +wlan.connect(SSID, PASS) +print('Connecting to WiFi...') +while not wlan.isconnected(): + utime.sleep(1) +print('Connected:', wlan.ifconfig()[0]) + +# ---- Install ---- +import mip +print('Installing identity-micropython...') +mip.install("https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/package.json") +print('Done. Remove SSID/PASS from install.py before committing.') + \ No newline at end of file diff --git a/lib/ssd1306.mpy b/lib/ssd1306.mpy new file mode 100644 index 0000000..a280785 Binary files /dev/null and b/lib/ssd1306.mpy differ diff --git a/main.py b/main.py new file mode 100644 index 0000000..3debf1d --- /dev/null +++ b/main.py @@ -0,0 +1,8 @@ +# ============================================================+ +# Identity IoT — main.py +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ + +import boot_manager +boot_manager.run() diff --git a/node_relay.py b/node_relay.py new file mode 100644 index 0000000..4f6c79c --- /dev/null +++ b/node_relay.py @@ -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) diff --git a/node_sensor.py b/node_sensor.py new file mode 100644 index 0000000..a19db80 --- /dev/null +++ b/node_sensor.py @@ -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) diff --git a/package.json b/package.json new file mode 100644 index 0000000..7793f4c --- /dev/null +++ b/package.json @@ -0,0 +1,19 @@ +{ + "urls": [ + ["identity_iot.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/identity_iot.py"], + ["identity_iot_ed25519.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/identity_iot_ed25519.py"], + ["identity_iot_aes.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/identity_iot_aes.py"], + ["sha512.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/sha512.py"], + ["boot_manager.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/boot_manager.py"], + ["provisioning_server.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/provisioning_server.py"], + ["provisioning_nfc.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/provisioning_nfc.py"], + ["pn532.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/pn532.py"], + ["node_sensor.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/node_sensor.py"], + ["node_relay.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/node_relay.py"], + ["main.py", "https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/main.py"] + ], + "deps": [ + ["ssd1306", "latest"] + ], + "version": "0.1.0" +} \ No newline at end of file diff --git a/pn532.py b/pn532.py new file mode 100644 index 0000000..4e1b8f5 --- /dev/null +++ b/pn532.py @@ -0,0 +1,415 @@ +# ============================================================+ +# Identity IoT — PN532 NFC Driver for MicroPython +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ + +import utime +from machine import Pin, I2C + +PN532_I2C_ADDRESS = 0x24 +PN532_CMD_GETFIRMWAREVERSION = 0x02 +PN532_CMD_SAMCONFIGURATION = 0x14 +PN532_CMD_INLISTPASSIVETARGET= 0x4A +PN532_CMD_INDATAEXCHANGE = 0x40 +PN532_CMD_INRELEASE = 0x52 +PN532_HOSTTOPN532 = 0xD4 +PN532_PN532TOHOST = 0xD5 + + +class PN532Error(Exception): + pass + + +class PN532_I2C: + + def __init__(self, i2c: I2C, address: int = PN532_I2C_ADDRESS): + self._i2c = i2c + self._addr = address + + # Wake + try: + self._i2c.writeto(self._addr, bytes([0x00])) + except Exception: + pass + utime.sleep_ms(500) + + fw = self.get_firmware_version() + if not fw: + raise PN532Error('PN532 not found or not responding') + print('[pn532] firmware: IC=%02x Ver=%d Rev=%d' % (fw['ic'], fw['ver'], fw['rev'])) + self._sam_config() + + def _build_frame(self, cmd: list) -> bytes: + body = [PN532_HOSTTOPN532] + cmd + length = len(body) + lcs = (~length + 1) & 0xFF + dcs = (~(sum(body) & 0xFF) + 1) & 0xFF + return bytes([0x00, 0x00, 0xFF, length, lcs] + body + [dcs, 0x00]) + + def _send_command(self, cmd: list): + frame = self._build_frame(cmd) + self._i2c.writeto(self._addr, frame) + + def _wait_ready(self, timeout=1000) -> bool: + t = timeout + while t > 0: + try: + b = self._i2c.readfrom(self._addr, 1) + if b[0] == 0x01: + return True + except Exception: + pass + utime.sleep_ms(10) + t -= 10 + return False + + def _read_response(self, length=32, timeout=1000): + """Read after command — PN532 sends ACK then response.""" + # Read ACK (6 bytes + status) + if not self._wait_ready(timeout): + return None + self._i2c.readfrom(self._addr, 7) # status + ACK frame + utime.sleep_ms(10) + + # Read response + if not self._wait_ready(timeout): + return None + buf = self._i2c.readfrom(self._addr, length + 1) + + # Parse: skip status byte, find data after header + # Frame: [status][0x00][0x00][0xFF][len][lcs][TFI][cmd+1][data...][dcs][0x00] + if len(buf) < 8: + return None + data_len = buf[4] # length field + if data_len < 2: + return bytes() + # data starts at index 7 (after status+preamble+start+len+lcs+TFI) + data = buf[7:7 + data_len - 1] + return bytes(data) + + def _sam_config(self): + self._send_command([PN532_CMD_SAMCONFIGURATION, 0x01, 0x14, 0x01]) + self._read_response(timeout=200) + + def get_firmware_version(self) -> dict: + try: + self._send_command([PN532_CMD_GETFIRMWAREVERSION]) + # First read: ACK frame + if not self._wait_ready(500): + return None + self._i2c.readfrom(self._addr, 7) # status + ACK + utime.sleep_ms(10) + # Second read: response frame + if not self._wait_ready(500): + return None + buf = self._i2c.readfrom(self._addr, 14) + # buf[0]=status, buf[1..3]=preamble, buf[4]=len, buf[5]=lcs + # buf[6]=TFI(0xD5), buf[7]=cmd(0x03), buf[8..11]=IC,Ver,Rev,Support + if len(buf) < 12: + return None + return { + 'ic': buf[8], + 'ver': buf[9], + 'rev': buf[10], + 'support': buf[11], + } + except Exception as e: + print('[pn532] fw error:', e) + return None + + def read_passive_target(self, timeout=1000) -> bytes: + """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: + return None + # 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 + 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) + + def read_page(self, page: int) -> bytes: + self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01, 0x30, page & 0xFF]) + resp = self._read_response(length=16, timeout=500) + if not resp or len(resp) < 4: + return None + return bytes(resp[:4]) + + def write_page(self, page: int, data: bytes) -> bool: + if len(data) != 4: + raise PN532Error('write_page: data must be 4 bytes') + self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01, 0xA2, page & 0xFF] + list(data)) + resp = self._read_response(timeout=500) + return resp is not None + + def release_target(self) -> bool: + try: + self._send_command([PN532_CMD_INRELEASE, 0x01]) + self._read_response(timeout=200) + return True + except Exception: + return False + + def read_ndef(self, timeout=2000) -> str: + """ + Read NDEF TextRecord from NTAG. + Returns text content string or None. + NDEF TLV starts at page 4 (byte offset 16). + """ + # Read pages 4..19 (64 bytes) — enough for our payload + data = bytearray() + for page in range(4, 45): + chunk = self.read_page(page) + if chunk is None: + break + data += chunk + + if not data: + return None + + # Parse TLV — find NDEF TLV (type 0x03) + i = 0 + while i < len(data): + t = data[i]; i += 1 + if t == 0xFE: # Terminator TLV + break + if t == 0x00: # Null TLV + continue + # Length + if i >= len(data): + break + l = data[i]; i += 1 + if l == 0xFF: # 3-byte length + if i + 2 >= len(data): + break + l = (data[i] << 8) | data[i+1]; i += 2 + if t != 0x03: # Skip non-NDEF TLVs + i += l + continue + # NDEF message + ndef_msg = data[i:i+l] + break + else: + return None + + if not ndef_msg: + return None + + # Parse NDEF record + # Byte 0: flags (MB|ME|CF|SR|IL|TNF) + # Byte 1: type length + # Byte 2: payload length (SR=1) or 4 bytes (SR=0) + # Byte 3+: type, id?, payload + if len(ndef_msg) < 4: + return None + + flags = ndef_msg[0] + type_len = ndef_msg[1] + sr = (flags >> 4) & 1 # Short Record + il = (flags >> 3) & 1 # ID Length present + + idx = 2 + if sr: + payload_len = ndef_msg[idx]; idx += 1 + else: + if idx + 4 > len(ndef_msg): return None + payload_len = (ndef_msg[idx]<<24 | ndef_msg[idx+1]<<16 | + ndef_msg[idx+2]<<8 | ndef_msg[idx+3]); idx += 4 + + if il: + id_len = ndef_msg[idx]; idx += 1 + idx += id_len + + # Skip type bytes + idx += type_len + + if idx + payload_len > len(ndef_msg): + return None + + payload = ndef_msg[idx:idx+payload_len] + + # TextRecord: first byte = status (lang length), then lang, then text + if len(payload) < 1: + return None + lang_len = payload[0] & 0x3F + text_start = 1 + lang_len + if text_start > len(payload): + return None + + try: + return bytes(payload[text_start:]).decode('utf-8') + except Exception: + return None + + # ========================================================================= + # ISO 7816 / NTAG424 DNA NDEF read + # ========================================================================= + + def _apdu(self, apdu: list, timeout=500, resp_len=64): + """Send ISO 7816 APDU via InDataExchange and return response bytes.""" + self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01] + apdu) + # Step 1: wait ready, read ACK + if not self._wait_ready(timeout): + return None + self._i2c.readfrom(self._addr, 7) # ACK frame + utime.sleep_ms(20) + # Step 2: wait ready, read response + if not self._wait_ready(timeout): + return None + buf = self._i2c.readfrom(self._addr, resp_len + 8) + # buf: [status][preamble x3][len][lcs][TFI=0xD5][cmd=0x41][err][data...][SW1][SW2][dcs][0x00] + # data after err byte: index 9, length = buf[4]-3 (TFI+cmd+err) + if len(buf) < 10: + return None + data_len = buf[4] + if data_len < 3: + return None + err = buf[8] + if err != 0x00: + print('[pn532] apdu err: %02x' % err) + return None + # data = everything from index 9 to 9+(data_len-3), then SW1 SW2 + data = bytes(buf[9:9 + data_len - 2]) + return data + + def _sw(self, resp) -> tuple: + """Extract SW1 SW2 — last 2 bytes of clean APDU response.""" + if not resp or len(resp) < 2: + return (0x00, 0x00) + return (resp[-2], resp[-1]) + + def _sw2(self, resp) -> tuple: + """Extract SW1 SW2 — first 2 bytes when response is SW only.""" + if not resp or len(resp) < 2: + return (0x00, 0x00) + return (resp[0], resp[1]) + + def read_ndef_iso(self, timeout=2000) -> str: + """ + Read NDEF from NTAG424 DNA via ISO 7816 command set. + No authentication required for default NDEF file. + Returns text content or None. + Must be called while tag is in field — polls for tag first. + """ + # 0. Poll for tag — activate it for InDataExchange + uid = self.read_passive_target(timeout=timeout) + if uid is None: + return None, None, None + utime.sleep_ms(100) + + # 1. Select NDEF Application (D2 76 00 00 85 01 01) + sel_app = [0x00, 0xA4, 0x04, 0x00, 0x07, + 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00] + resp = self._apdu(sel_app) + if not resp: + return None, None + sw1, sw2 = resp[0], resp[1] + if sw1 != 0x90: + print('[pn532] select app failed: %02x %02x' % (sw1, sw2)) + 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, None + sw1, sw2 = resp[0], resp[1] + if sw1 != 0x90: + print('[pn532] select file failed: %02x %02x' % (sw1, sw2)) + 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, None + # resp = [len_hi][len_lo][SW1][SW2] + if len(resp) < 4: + return None, None + sw1, sw2 = resp[2], resp[3] + if sw1 != 0x90: + print('[pn532] read len failed: %02x %02x' % (sw1, sw2)) + return None, None + ndef_len = (resp[0] << 8) | resp[1] + if ndef_len == 0: + return None, None + + # 4. Read NDEF data (max 200 bytes per read) + ndef_data = bytearray() + offset = 2 + remaining = ndef_len + + while remaining > 0: + chunk = min(remaining, 50) # PN532 I2C frame limit + read_data = [0x00, 0xB0, + (offset >> 8) & 0xFF, + offset & 0xFF, + chunk & 0xFF] + resp = self._apdu(read_data, timeout=timeout, resp_len=chunk+3) + if not resp: + break + if len(resp) < 2: + break + sw1, sw2 = resp[-2], resp[-1] + if sw1 != 0x90: + print('[pn532] read data failed: %02x %02x' % (sw1, sw2)) + break + ndef_data += resp[:-2] # exclude SW1 SW2 + offset += chunk + remaining -= chunk + + if not ndef_data: + return None, None + + # 5. Parse NDEF message + # ndef_data = clean NDEF bytes, no response code prefix + if len(ndef_data) < 3: + return None, None + + flags = ndef_data[0] + type_len = ndef_data[1] + sr = (flags >> 4) & 1 + il = (flags >> 3) & 1 + + idx = 2 + if sr: + if idx >= len(ndef_data): return None + payload_len = ndef_data[idx]; idx += 1 + else: + if idx + 4 > len(ndef_data): return None + payload_len = (ndef_data[idx]<<24 | ndef_data[idx+1]<<16 | + ndef_data[idx+2]<<8 | ndef_data[idx+3]); idx += 4 + + if il: + id_len = ndef_data[idx]; idx += 1 + idx += id_len + + idx += type_len # skip type bytes + + if idx + payload_len > len(ndef_data): + 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, None + lang_len = payload[0] & 0x3F + text_start = 1 + lang_len + if text_start > len(payload): + return None, None + + try: + import ubinascii as _ub + uid_hex = _ub.hexlify(uid).decode() + return uid_hex, bytes(payload[text_start:]).decode('utf-8') + except Exception: + return None, None \ No newline at end of file diff --git a/provisioning_nfc.py b/provisioning_nfc.py new file mode 100644 index 0000000..2e00e26 --- /dev/null +++ b/provisioning_nfc.py @@ -0,0 +1,32 @@ +# ============================================================+ +# Identity IoT — Provisioning via NFC +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ +# Used when has_nfc = True and no config.json found. +# Waits for writer badge tap, reads AES-encrypted config payload +# from NTAG424, decrypts with PROG_KEY/PROG_IV. +# ============================================================+ + +import utime +from provisioning_server import PROG_KEY, PROG_IV +from identity_iot_aes import aes_cbc_decrypt +import ujson + + +def run_provisioning_nfc(): + """ + Block until writer badge is tapped. + Reads encrypted config from NTAG424, returns config dict. + """ + print('[provisioning_nfc] waiting for writer badge...') + + # TODO: init PN532 + # TODO: wait for NTAG424 tap + # TODO: read encrypted config payload from tag memory + # TODO: aes_cbc_decrypt(payload, PROG_KEY, PROG_IV) + # TODO: validate and return config dict + + while True: + print('[provisioning_nfc] tap writer badge...') + utime.sleep(2) diff --git a/provisioning_server.py b/provisioning_server.py new file mode 100644 index 0000000..44fd719 --- /dev/null +++ b/provisioning_server.py @@ -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 diff --git a/sha512.py b/sha512.py new file mode 100644 index 0000000..4e46143 --- /dev/null +++ b/sha512.py @@ -0,0 +1,87 @@ +# SHA-512 pure Python for MicroPython +# Based on public domain implementation +# Compact version optimized for MicroPython memory constraints + +_K = [ + 0x428a2f98d728ae22, 0x7137449123ef65cd, 0xb5c0fbcfec4d3b2f, 0xe9b5dba58189dbbc, + 0x3956c25bf348b538, 0x59f111f1b605d019, 0x923f82a4af194f9b, 0xab1c5ed5da6d8118, + 0xd807aa98a3030242, 0x12835b0145706fbe, 0x243185be4ee4b28c, 0x550c7dc3d5ffb4e2, + 0x72be5d74f27b896f, 0x80deb1fe3b1696b1, 0x9bdc06a725c71235, 0xc19bf174cf692694, + 0xe49b69c19ef14ad2, 0xefbe4786384f25e3, 0x0fc19dc68b8cd5b5, 0x240ca1cc77ac9c65, + 0x2de92c6f592b0275, 0x4a7484aa6ea6e483, 0x5cb0a9dcbd41fbd4, 0x76f988da831153b5, + 0x983e5152ee66dfab, 0xa831c66d2db43210, 0xb00327c898fb213f, 0xbf597fc7beef0ee4, + 0xc6e00bf33da88fc2, 0xd5a79147930aa725, 0x06ca6351e003826f, 0x142929670a0e6e70, + 0x27b70a8546d22ffc, 0x2e1b21385c26c926, 0x4d2c6dfc5ac42aed, 0x53380d139d95b3df, + 0x650a73548baf63de, 0x766a0abb3c77b2a8, 0x81c2c92e47edaee6, 0x92722c851482353b, + 0xa2bfe8a14cf10364, 0xa81a664bbc423001, 0xc24b8b70d0f89791, 0xc76c51a30654be30, + 0xd192e819d6ef5218, 0xd69906245565a910, 0xf40e35855771202a, 0x106aa07032bbd1b8, + 0x19a4c116b8d2d0c8, 0x1e376c085141ab53, 0x2748774cdf8eeb99, 0x34b0bcb5e19b48a8, + 0x391c0cb3c5c95a63, 0x4ed8aa4ae3418acb, 0x5b9cca4f7763e373, 0x682e6ff3d6b2b8a3, + 0x748f82ee5defb2fc, 0x78a5636f43172f60, 0x84c87814a1f0ab72, 0x8cc702081a6439ec, + 0x90befffa23631e28, 0xa4506cebde82bde9, 0xbef9a3f7b2c67915, 0xc67178f2e372532b, + 0xca273eceea26619c, 0xd186b8c721c0c207, 0xeada7dd6cde0eb1e, 0xf57d4f7fee6ed178, + 0x06f067aa72176fba, 0x0a637dc5a2c898a6, 0x113f9804bef90dae, 0x1b710b35131c471b, + 0x28db77f523047d84, 0x32caab7b40c72493, 0x3c9ebe0a15c9bebc, 0x431d67c49c100d4c, + 0x4cc5d4becb3e42b6, 0x597f299cfc657e2a, 0x5fcb6fab3ad6faec, 0x6c44198c4a475817, +] + +_M64 = 0xffffffffffffffff + +def _rotr64(x, n): + return ((x >> n) | (x << (64 - n))) & _M64 + +def _sha512_compress(state, block): + w = list(block) + for i in range(16, 80): + s0 = _rotr64(w[i-15], 1) ^ _rotr64(w[i-15], 8) ^ (w[i-15] >> 7) + s1 = _rotr64(w[i-2], 19) ^ _rotr64(w[i-2], 61) ^ (w[i-2] >> 6) + w.append((w[i-16] + s0 + w[i-7] + s1) & _M64) + + a, b, c, d, e, f, g, h = state + + for i in range(80): + S1 = _rotr64(e, 14) ^ _rotr64(e, 18) ^ _rotr64(e, 41) + ch = (e & f) ^ ((~e & _M64) & g) + temp1 = (h + S1 + ch + _K[i] + w[i]) & _M64 + S0 = _rotr64(a, 28) ^ _rotr64(a, 34) ^ _rotr64(a, 39) + maj = (a & b) ^ (a & c) ^ (b & c) + temp2 = (S0 + maj) & _M64 + + h = g; g = f; f = e + e = (d + temp1) & _M64 + d = c; c = b; b = a + a = (temp1 + temp2) & _M64 + + return [ + (state[0] + a) & _M64, (state[1] + b) & _M64, + (state[2] + c) & _M64, (state[3] + d) & _M64, + (state[4] + e) & _M64, (state[5] + f) & _M64, + (state[6] + g) & _M64, (state[7] + h) & _M64, + ] + +def sha512(data: bytes) -> bytes: + # Initial hash values + state = [ + 0x6a09e667f3bcc908, 0xbb67ae8584caa73b, + 0x3c6ef372fe94f82b, 0xa54ff53a5f1d36f1, + 0x510e527fade682d1, 0x9b05688c2b3e6c1f, + 0x1f83d9abfb41bd6b, 0x5be0cd19137e2179, + ] + + msg = bytearray(data) + length = len(data) * 8 + msg.append(0x80) + while len(msg) % 128 != 112: + msg.append(0x00) + # Append length as 128-bit big-endian (high 64 bits are 0) + msg += (0).to_bytes(8, 'big') + msg += length.to_bytes(8, 'big') + + for i in range(0, len(msg), 128): + block = [] + for j in range(0, 128, 8): + block.append(int.from_bytes(msg[i+j:i+j+8], 'big')) + state = _sha512_compress(state, block) + + return b''.join(s.to_bytes(8, 'big') for s in state) +