commit d23062d54f4fc0fa5d3dae265857b20c89934ed0 Author: Niko Date: Thu May 21 23:07:50 2026 +0200 feat: identity-micropython v0.1.0 — Pico W Ed25519 PoC diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dfc9018 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# MicroPython compiled +*.mpy + +# Secrets +main.py +wifi_config.py +config.py +tenant_*.py + +# OS +.DS_Store +Thumbs.db + +# IDE +.vscode/ +.idea/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..56c9926 --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# 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 +import mip +mip.install("https://git.aeonianengineering.net/wide/identity-micropython/raw/branch/main/package.json") +``` + +## 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/example_main.py b/example_main.py new file mode 100644 index 0000000..e64eecc --- /dev/null +++ b/example_main.py @@ -0,0 +1,66 @@ +# ============================================================+ +# Identity IoT — Example main.py for Pico W +# ============================================================+ +# Copy to Pico W as main.py for standalone operation. +# Connects to WiFi, registers node, authenticates, prints JWT. +# ============================================================+ + +import network +import utime +from identity_iot import IdentityIoT + +# ---- WiFi config ---- +SSID = 'YOUR_SSID' +PASSWORD = 'YOUR_PASSWORD' + +# ---- Backend config ---- +BASE_URL = 'https://ap.parta.app/API/V1/identity_layer/identity_iot/index.php' +TENANT_ID = 1 +NODE_TYPE = 'sensor' # sensor | reader | actuator | gateway +NODE_LABEL = 'pico-01' + +# ---- Bootstrap ITEMS keys (from your tenant config) ---- +COMM_KEY = b'pm4u5YahhYSZJwGNzpA7QeD0F0O8R8Eo' # 32 bytes +COMM_IV = b'DPUBMeD0Jc42ZbMM' # 16 bytes + + +def connect_wifi(): + wlan = network.WLAN(network.STA_IF) + wlan.active(True) + wlan.connect(SSID, PASSWORD) + timeout = 15 + while not wlan.isconnected() and timeout > 0: + utime.sleep(1) + timeout -= 1 + if not wlan.isconnected(): + raise Exception('WiFi connection failed') + print('WiFi connected:', wlan.ifconfig()[0]) + + +def main(): + connect_wifi() + + identity = IdentityIoT( + base_url = BASE_URL, + node_type = NODE_TYPE, + node_label= NODE_LABEL, + tenant_id = TENANT_ID, + comm_key = COMM_KEY, + comm_iv = COMM_IV, + ) + + # Register (idempotent — safe to call every boot) + print('Registering node...') + reg = identity.ensure_registered() + print('Node:', reg.get('result'), '— uid:', reg.get('node_uid', '')[:16] + '...') + + # Authenticate + print('Authenticating...') + token = identity.authenticate() + print('JWT:', token[:40] + '...') + + # Node UID (hashed — safe to log) + print('Node UID:', identity.get_node_uid()[:16] + '...') + + +main() diff --git a/identity_iot.py b/identity_iot.py new file mode 100644 index 0000000..6202209 --- /dev/null +++ b/identity_iot.py @@ -0,0 +1,292 @@ +# ============================================================+ +# 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://ap.parta.app/API/V1/identity_layer/identity_iot/index.php") +# await identity.ensure_registered(tenant_id=1) +# token = await identity.authenticate() +# ============================================================+ + +import ujson +import ubinascii +import uhashlib +import urandom +import machine +import network +import uasyncio as asyncio + +try: + import uurequests as requests + _ASYNC = False +except ImportError: + import urequests as requests + _ASYNC = False + +# Pure-Python Ed25519 — replaced by natmod in production +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. + Communicates with identity_layer/identity_iot PHP backend. + """ + + # Storage keys (on flash filesystem) + _F_SEED = '/identity_seed.bin' # 32 bytes raw seed + _F_PK = '/identity_pk.b64' # public key b64 + _F_JWT = '/identity_jwt.txt' # current JWT + _F_NODEID = '/identity_node_uid.txt' # hashed node UID + + def __init__( + self, + base_url: str, + app_id: str = 'identity_iot', + node_type: str = 'sensor', + node_label: str = None, + tenant_id: int = 0, + comm_key: bytes = None, + comm_iv: bytes = 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 + # Bootstrap AES-CBC keys — replaced after init_key if used + self._comm_key = comm_key + self._comm_iv = comm_iv + + # ========================================================================= + # 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. + Stores JWT on flash for next session. + """ + # Try existing JWT first + jwt = self._load_jwt() + if jwt and not self._is_jwt_expired(jwt): + return jwt + + node_uid = self._get_node_uid() + + # Challenge + resp = self._post({ + 'request': 'challenge', + 'node_uid': node_uid, + }) + if resp.get('status') != 'ok': + raise Exception('challenge failed: ' + str(resp)) + + challenge_id = resp['challenge_id'] + nonce_b64 = resp['nonce_b64'] + + # Sign nonce + 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() + + # Verify + resp = self._post({ + 'request': 'verify_signature', + 'challenge_id': challenge_id, + 'signature_b64': sig_b64, + 'node_uid': node_uid, + }) + if resp.get('status') != 'ok': + raise Exception('verify_signature failed: ' + str(resp)) + + token = resp.get('token', '') + if token: + self._save_jwt(token) + return token + + def get_node_uid(self): + """Return the hashed node UID (safe to share).""" + return self._get_node_uid() + + # ========================================================================= + # Keypair management + # ========================================================================= + + def _generate_keypair(self): + """Generate Ed25519 keypair from secure random seed, persist to flash.""" + seed = bytes(urandom.getrandbits(8) for _ in range(32)) + pk, _ = ed25519_keypair_from_seed(seed) + pk_b64 = ubinascii.b2a_base64(pk).decode().strip() + # Persist + 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 — device fingerprint, hashed before sending + # ========================================================================= + + def _get_node_uid(self): + """ + Returns SHA-256 hex of device UID. + Cached on flash after first call. + """ + try: + with open(self._F_NODEID, 'r') as f: + return f.read().strip() + except OSError: + pass + + # Pico W: machine.unique_id() returns 8 bytes + raw_uid = machine.unique_id() + h = uhashlib.sha256(raw_uid) + 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) + # MicroPython: use time.time() — requires NTP sync + import utime + return utime.time() >= exp + except Exception: + return True # assume expired on any error + + # ========================================================================= + # HTTP transport — ITEMS envelope (AES-CBC) or plain JSON + # ========================================================================= + + def _post(self, inner: dict) -> dict: + """ + POST to backend. + If comm_key/iv provided: ITEMS AES-CBC envelope. + Otherwise: plain JSON (for PoC / testing). + """ + 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: + """Plain JSON POST — for PoC and testing.""" + body = ujson.dumps({'data': ujson.dumps(inner)}) + resp = requests.post( + self.base_url, + headers={'Content-Type': 'application/json'}, + data=body, + ) + outer = ujson.loads(resp.text) + resp.close() + # Backend returns encrypted data — decode if possible, else return raw + if 'data' in outer: + try: + return ujson.loads(outer['data']) + except Exception: + return outer + return outer + + def _post_encrypted(self, inner: dict) -> dict: + """ITEMS AES-CBC envelope POST.""" + from identity_iot_aes import aes_cbc_encrypt, aes_cbc_decrypt + 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, + ) + outer = ujson.loads(resp.text) + resp.close() + if 'data' not in outer: + raise Exception('Missing outer.data in response') + clear = aes_cbc_decrypt(outer['data'], self._comm_key, self._comm_iv) + return ujson.loads(clear) + + +# ============================================================================= +# Utility +# ============================================================================= + +def _normalize_b64(s: str) -> str: + """Normalize base64 string for ubinascii.""" + 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 diff --git a/identity_iot_aes.py b/identity_iot_aes.py new file mode 100644 index 0000000..5a2335e --- /dev/null +++ b/identity_iot_aes.py @@ -0,0 +1,64 @@ +# ============================================================+ +# 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..cffc935 --- /dev/null +++ b/identity_iot_ed25519.py @@ -0,0 +1,163 @@ +# ============================================================+ +# Identity IoT — Ed25519 Pure Python +# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong +# (c) Copyright : WIDE di D. Papa - Naples - Italy +# ============================================================+ +# Pure-Python Ed25519 for MicroPython PoC. +# Production: replace with natmod C implementation. +# +# Based on public domain Ed25519 reference implementation. +# Optimized for MicroPython memory constraints. +# ============================================================+ + +import uhashlib +import ubinascii + +# ---- Field arithmetic mod p ---- +_p = 2**255 - 19 +_q = 2**252 + 27742317777372353535851937790883648493 +_d = -121665 * pow(121666, _p - 2, _p) % _p +_I = pow(2, (_p - 1) // 4, _p) +_Gx = 15112221349535807912866137220509078750507884956996801637620 +_Gy = 46316835694926478169428394003475163141307993866256225615783 +_G = (_Gx % _p, _Gy % _p, 1, _Gx * _Gy % _p) + + +def _sha512(data: bytes) -> bytes: + # MicroPython uhashlib has sha256 but not sha512 + # Use double-sha256 as fallback — NOT cryptographically equivalent + # TODO: replace with proper sha512 natmod for production + h1 = uhashlib.sha256(data).digest() + h2 = uhashlib.sha256(h1 + data).digest() + return h1 + h2 + + +def _clamp(h: bytes) -> int: + a = bytearray(h[:32]) + a[0] &= 248 + a[31] &= 127 + a[31] |= 64 + return int.from_bytes(a, 'little') + + +def _point_add(P, Q): + x1, y1, z1, t1 = P + x2, y2, z2, t2 = Q + A = (y1 - x1) * (y2 - x2) % _p + B = (y1 + x1) * (y2 + x2) % _p + C = 2 * t1 * t2 * _d % _p + D = 2 * z1 * z2 % _p + E = B - A + F = D - C + G_ = D + C + H = B + A + x3 = E * F % _p + y3 = G_ * H % _p + t3 = E * H % _p + z3 = F * G_ % _p + return (x3, y3, z3, t3) + + +def _point_mul(s, P): + Q = (0, 1, 1, 0) + while s > 0: + if s & 1: + Q = _point_add(Q, P) + P = _point_add(P, P) + s >>= 1 + return Q + + +def _point_encode(P) -> bytes: + x, y, z, _ = P + zi = pow(z, _p - 2, _p) + x = x * zi % _p + y = y * zi % _p + out = bytearray(y.to_bytes(32, 'little')) + out[31] ^= (x & 1) << 7 + return bytes(out) + + +def _point_decode(s: bytes): + y = int.from_bytes(s, 'little') & ~(1 << 255) + x_neg = s[31] >> 7 + y2 = y * y % _p + u = (y2 - 1) % _p + v = (_d * y2 + 1) % _p + x = pow(u * pow(v, _p - 2, _p), (_p + 3) // 8, _p) + vx2 = v * x * x % _p + if (vx2 - u) % _p != 0: + if (vx2 + u) % _p != 0: + raise ValueError('Invalid point') + x = x * _I % _p + if x % 2 != x_neg: + x = _p - x + return (x, y, 1, x * y % _p) + + +def ed25519_keypair_from_seed(seed: bytes): + """ + Derive Ed25519 keypair from 32-byte seed. + Returns (public_key_bytes, private_key_bytes). + """ + if len(seed) != 32: + raise ValueError('Seed must be 32 bytes') + h = _sha512(seed) + a = _clamp(h) + A = _point_mul(a, _G) + pk = _point_encode(A) + sk = seed + pk # 64 bytes: seed || pubkey + return pk, sk + + +def ed25519_sign(seed: bytes, message: bytes) -> bytes: + """ + Sign message with Ed25519 private key derived from seed. + Returns 64-byte signature. + Compatible with PHP ITEMSSignature::verifyDetached(). + """ + if len(seed) != 32: + raise ValueError('Seed must be 32 bytes') + h = _sha512(seed) + a = _clamp(h) + pk_bytes, _ = ed25519_keypair_from_seed(seed) + + # Nonce: SHA512(h[32:64] || message) + r_hash = _sha512(h[32:] + message) + r = int.from_bytes(r_hash, 'little') % _q + R = _point_mul(r, _G) + R_enc = _point_encode(R) + + # k = SHA512(R || pk || message) + k_hash = _sha512(R_enc + pk_bytes + message) + k = int.from_bytes(k_hash, 'little') % _q + + # s = (r + k * a) mod q + s = (r + k * a) % _q + S = s.to_bytes(32, 'little') + + return R_enc + S + + +def ed25519_verify(pk_bytes: bytes, message: bytes, signature: bytes) -> bool: + """ + Verify Ed25519 signature. + Returns True if valid. + """ + if len(signature) != 64 or len(pk_bytes) != 32: + return False + try: + R_enc = signature[:32] + S = int.from_bytes(signature[32:], 'little') + if S >= _q: + return False + A = _point_decode(pk_bytes) + R = _point_decode(R_enc) + k_hash = _sha512(R_enc + pk_bytes + message) + k = int.from_bytes(k_hash, 'little') % _q + # Check: [8][S]B == [8]R + [8][k]A + SB = _point_mul(8 * S, _G) + RkA = _point_add(_point_mul(8, R), _point_mul(8 * k, A)) + return _point_encode(SB) == _point_encode(RkA) + except Exception: + return False diff --git a/package.json b/package.json new file mode 100644 index 0000000..c0ef9ae --- /dev/null +++ b/package.json @@ -0,0 +1,8 @@ +{ + "urls": [ + ["identity_iot.py", "github:wide/identity-micropython/identity_iot.py"], + ["identity_iot_ed25519.py", "github:wide/identity-micropython/identity_iot_ed25519.py"], + ["identity_iot_aes.py", "github:wide/identity-micropython/identity_iot_aes.py"] + ], + "version": "0.1.0" +}