65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Low-level wire messages: envelopes to bytes and back.
|
|
|
|
:class:`EnvelopeCodec` owns everything byte-level about one device key:
|
|
serialising a :class:`CommunicationObject` to encrypted TX bytes and
|
|
parsing received RX bytes back into an envelope. Business logic (which
|
|
command/arguments to send, retries, SoC extraction) lives one layer up in
|
|
``device_local`` / ``battery`` / ``power``; standalone scripts must go
|
|
through those layers, never reimplement this.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict
|
|
|
|
from .crypto import (
|
|
DEFAULT_KEY_RADIX,
|
|
decrypt_wire,
|
|
encrypt_object,
|
|
generate_secret_key,
|
|
)
|
|
from .envelope import CommunicationObject
|
|
|
|
|
|
class EnvelopeCodec:
|
|
"""Encrypt/decrypt envelopes under one device key.
|
|
|
|
:param key: ``generate_secret_key`` output for the device serial.
|
|
:param radix: key-parse radix (16 — verified against live captures).
|
|
"""
|
|
|
|
def __init__(self, key: str, radix: int = DEFAULT_KEY_RADIX) -> None:
|
|
self.key = str(key)
|
|
self.radix = int(radix)
|
|
|
|
@classmethod
|
|
def for_serial(cls, equip_no: str,
|
|
radix: int = DEFAULT_KEY_RADIX) -> "EnvelopeCodec":
|
|
"""Codec keyed by a device serial (derives ``generate_secret_key``)."""
|
|
return cls(generate_secret_key(equip_no), radix=radix)
|
|
|
|
# ---------------------------------------------------------------- encode
|
|
def encode(self, envelope: CommunicationObject) -> bytes:
|
|
"""``toJsonString`` → ``S1 || transform(S2)`` TX bytes."""
|
|
return encrypt_object(envelope.to_json_string(), self.key,
|
|
radix=self.radix)
|
|
|
|
# ---------------------------------------------------------------- decode
|
|
def decode(self, wire: bytes) -> CommunicationObject:
|
|
"""RX bytes → envelope (2nd-comma split, key from ``equipNo``).
|
|
|
|
Raises :class:`ValueError` on truncated/non-JSON/undecodable
|
|
input — callers accumulating a stream treat that as "keep
|
|
reading".
|
|
"""
|
|
return CommunicationObject.from_dict(
|
|
json.loads(decrypt_wire(bytes(wire), radix=self.radix)))
|
|
|
|
def try_decode(self, wire: bytes) -> Dict[str, Any] | None:
|
|
"""Like :meth:`decode` but returns the dict, or ``None``."""
|
|
try:
|
|
return self.decode(wire).to_dict()
|
|
except (ValueError, TypeError):
|
|
return None
|