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

feat: identity-micropython v0.1.0 — Pico W Ed25519 PoC

Browse Source
This commit is contained in:
Niko
2026-05-21 23:07:50 +02:00
commit d23062d54f
7 changed files with 716 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
.gitignore vendored Normal file
View File

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

107
README.md Normal file
View File

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

66
example_main.py Normal file
View File

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

292
identity_iot.py Normal file
View File

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

64
identity_iot_aes.py Normal file
View File

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

163
identity_iot_ed25519.py Normal file
View File

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

8
package.json Normal file
View File

@@ -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"
}