fix: correct URLs, add sha512, updated ed25519 and identity_iot
This commit is contained in:
195
identity_iot.py
195
identity_iot.py
@@ -12,9 +12,9 @@
|
||||
#
|
||||
# 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()
|
||||
# identity = IdentityIoT(base_url="https://api.parta.app/API/V1/identity_layer/identity_iot/")
|
||||
# identity.ensure_registered()
|
||||
# token = identity.authenticate()
|
||||
# ============================================================+
|
||||
|
||||
import ujson
|
||||
@@ -23,51 +23,48 @@ import uhashlib
|
||||
import urandom
|
||||
import machine
|
||||
import network
|
||||
import uasyncio as asyncio
|
||||
import utime
|
||||
|
||||
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
|
||||
_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: int = 0,
|
||||
comm_key: bytes = None,
|
||||
comm_iv: bytes = None,
|
||||
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,
|
||||
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
|
||||
# Bootstrap AES-CBC keys — replaced after init_key if used
|
||||
self._comm_key = comm_key
|
||||
self._comm_iv = comm_iv
|
||||
self._wifi_ssid = wifi_ssid
|
||||
self._wifi_pass = wifi_pass
|
||||
|
||||
# =========================================================================
|
||||
# Public API
|
||||
@@ -80,7 +77,7 @@ class IdentityIoT:
|
||||
pk = self._generate_keypair()
|
||||
node_uid = self._get_node_uid()
|
||||
resp = self._post({
|
||||
'request': 'register_node',
|
||||
'request': 'register_node',
|
||||
'public_key_b64': pk,
|
||||
'node_uid': node_uid,
|
||||
'node_type': self.node_type,
|
||||
@@ -92,35 +89,25 @@ class IdentityIoT:
|
||||
return resp
|
||||
|
||||
def authenticate(self):
|
||||
"""
|
||||
Full challenge/response flow. Returns JWT string.
|
||||
Stores JWT on flash for next session.
|
||||
"""
|
||||
# Try existing JWT first
|
||||
"""Full challenge/response flow. Returns JWT string."""
|
||||
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,
|
||||
})
|
||||
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()
|
||||
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,
|
||||
@@ -136,7 +123,6 @@ class IdentityIoT:
|
||||
return token
|
||||
|
||||
def get_node_uid(self):
|
||||
"""Return the hashed node UID (safe to share)."""
|
||||
return self._get_node_uid()
|
||||
|
||||
# =========================================================================
|
||||
@@ -144,11 +130,9 @@ class IdentityIoT:
|
||||
# =========================================================================
|
||||
|
||||
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)
|
||||
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:
|
||||
@@ -170,25 +154,18 @@ class IdentityIoT:
|
||||
return None
|
||||
|
||||
# =========================================================================
|
||||
# Node UID — device fingerprint, hashed before sending
|
||||
# Node UID
|
||||
# =========================================================================
|
||||
|
||||
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)
|
||||
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
|
||||
@@ -213,65 +190,94 @@ class IdentityIoT:
|
||||
parts = jwt.split('.')
|
||||
if len(parts) != 3:
|
||||
return True
|
||||
payload_b64 = _normalize_b64(parts[1])
|
||||
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
|
||||
payload = ujson.loads(payload_json)
|
||||
exp = payload.get('exp', 0)
|
||||
return utime.time() >= exp
|
||||
except Exception:
|
||||
return True # assume expired on any error
|
||||
return True
|
||||
|
||||
# =========================================================================
|
||||
# HTTP transport — ITEMS envelope (AES-CBC) or plain JSON
|
||||
# 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:
|
||||
"""
|
||||
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:
|
||||
for attempt in range(3):
|
||||
try:
|
||||
return ujson.loads(outer['data'])
|
||||
except Exception:
|
||||
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
|
||||
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:
|
||||
"""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)
|
||||
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')
|
||||
|
||||
|
||||
# =============================================================================
|
||||
@@ -279,7 +285,6 @@ class IdentityIoT:
|
||||
# =============================================================================
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user