355 lines
18 KiB
Python
355 lines
18 KiB
Python
"""Legacy V1 discovery, information and MD5+ROM upgrade on TCP 4941.
|
|
|
|
Wire contract: references/v1-protocol.md, verified against supplied Qt source.
|
|
An upgrade acknowledgment is not evidence of a completed flash or boot.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import select
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
INFO_PORT = 4941
|
|
DISCOVERY_PORT = 4942
|
|
INFO_COMMAND = 0x5A0E2104
|
|
UPGRADE_COMMAND = 0x5A0E2170
|
|
DISCOVERY_REQUEST = 0x5A0EFF00
|
|
DISCOVERY_RESPONSE = 0x5A0EFF01
|
|
MAX_JSON_SIZE = 1024 * 1024
|
|
INFO_FIELDS = ("code", "desc", "sn", "dev_type", "firmware", "hardware", "dev_id", "dev_ver")
|
|
DISCOVERY_FIELDS = ("session", "ip", "sn", "dev_type", "mac", "firmware", "hardware")
|
|
|
|
|
|
class ProtocolError(ValueError):
|
|
"""A malformed or mismatched V1 frame; never trigger a retry."""
|
|
|
|
|
|
def encode_frame(command: int, payload: dict | None = None) -> bytes:
|
|
data = b"" if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
return struct.pack("!II", command, len(data)) + data
|
|
|
|
|
|
def decode_frame(packet: bytes, expected_command: int) -> dict:
|
|
if len(packet) < 8:
|
|
raise ProtocolError("V1 header is shorter than 8 bytes")
|
|
command, size = struct.unpack("!II", packet[:8])
|
|
if command != expected_command:
|
|
raise ProtocolError("V1 response command mismatch")
|
|
if not 0 < size <= MAX_JSON_SIZE or len(packet) != 8 + size:
|
|
raise ProtocolError("V1 response length mismatch or out of bounds")
|
|
try:
|
|
result = json.loads(packet[8:].decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ProtocolError("V1 response is not a UTF-8 JSON object") from exc
|
|
if not isinstance(result, dict):
|
|
raise ProtocolError("V1 response JSON must be an object")
|
|
return result
|
|
|
|
|
|
def _string_fields(body: dict, fields):
|
|
for field in fields:
|
|
if field in body and not isinstance(body[field], str):
|
|
raise ProtocolError("V1 field %s must be a string" % field)
|
|
|
|
|
|
def _remaining(deadline: float) -> float:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("V1 query exceeded its total timeout")
|
|
return remaining
|
|
|
|
|
|
def _recv_exact(sock, count: int, deadline: float) -> bytes:
|
|
result = bytearray()
|
|
while len(result) < count:
|
|
sock.settimeout(_remaining(deadline))
|
|
chunk = sock.recv(count - len(result))
|
|
if not chunk:
|
|
raise ProtocolError("V1 connection closed before the complete response")
|
|
result.extend(chunk)
|
|
return bytes(result)
|
|
|
|
|
|
def query_device_info(host: str, timeout: float, logger) -> dict:
|
|
deadline = time.monotonic() + timeout
|
|
logger.emit("v1_info_request", host=host, port=INFO_PORT, command=hex(INFO_COMMAND))
|
|
with socket.create_connection((host, INFO_PORT), timeout=_remaining(deadline)) as sock:
|
|
sock.settimeout(_remaining(deadline))
|
|
sock.sendall(encode_frame(INFO_COMMAND))
|
|
header = _recv_exact(sock, 8, deadline)
|
|
command, size = struct.unpack("!II", header)
|
|
if command != INFO_COMMAND or not 0 < size <= MAX_JSON_SIZE:
|
|
raise ProtocolError("V1 info response command/length is invalid")
|
|
body = decode_frame(header + _recv_exact(sock, size, deadline), INFO_COMMAND)
|
|
if type(body.get("code")) is not int or body["code"] != 0:
|
|
raise ProtocolError("V1 info response did not report integer code=0")
|
|
_string_fields(body, INFO_FIELDS[1:])
|
|
# Keep only the documented read-only fields; do not retain unexpected credentials.
|
|
data = {key: body[key] for key in INFO_FIELDS if key in body}
|
|
# A task may explicitly require a literal nested version if a firmware reports
|
|
# it. Observe that exact path only; never manufacture it from native fields.
|
|
for section in ("inner", "outer"):
|
|
nested = body.get(section)
|
|
if isinstance(nested, dict) and isinstance(nested.get("version"), str):
|
|
data[section] = {"version": nested["version"]}
|
|
response = {"protocol": "v1", "cmd": INFO_COMMAND, "code": 0, "data": data}
|
|
logger.emit("v1_device_info", response=response)
|
|
return response
|
|
|
|
|
|
def validate_identity(response: dict, expected_sn: str, expected_model: str | None):
|
|
if (not isinstance(response, dict) or response.get("protocol") != "v1"
|
|
or type(response.get("cmd")) is not int or response["cmd"] != INFO_COMMAND
|
|
or type(response.get("code")) is not int or response["code"] != 0):
|
|
raise ProtocolError("Expected an explicit V1 0x5a0e2104 response envelope")
|
|
data = response.get("data")
|
|
if not isinstance(data, dict) or type(data.get("code")) is not int or data["code"] != 0:
|
|
raise ProtocolError("V1 identity requires successful query data")
|
|
_string_fields(data, INFO_FIELDS[1:])
|
|
if not isinstance(data.get("sn"), str) or data["sn"] != expected_sn:
|
|
raise ProtocolError("V1 full SN mismatch")
|
|
model = data.get("dev_type")
|
|
models = [model] if isinstance(model, str) and model.strip() else []
|
|
if expected_model is not None and expected_model not in models:
|
|
raise ProtocolError("V1 dev_type does not exactly match the expected model")
|
|
return data, models
|
|
|
|
|
|
def _upgrade_wait(deadline: float, stall_timeout: float) -> float:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("V1 upgrade exceeded its total timeout")
|
|
return min(remaining, stall_timeout)
|
|
|
|
|
|
def transfer_rom(host: str, rom_path: Path, rom_size: int, connect_timeout: float,
|
|
stall_timeout: float, total_timeout: float, chunk_size: int,
|
|
logger, expected_sha256: str) -> dict:
|
|
"""Send one legacy 2170 stream, then require its complete JSON acknowledgment.
|
|
|
|
Data = lowercase ASCII MD5[32] + ROM. Length excludes the 8-byte frame header.
|
|
Hash and stream the same open file, bound each socket operation and the whole
|
|
transaction, and never reconnect or retry a write. Only ROM bytes count toward
|
|
bytes_sent; wire_bytes_sent also includes the 40-byte header/digest prefix.
|
|
"""
|
|
if type(rom_size) is not int or not 0 < rom_size <= 0xFFFFFFFF - 32:
|
|
raise ProtocolError("V1 ROM size is outside its U32 data length")
|
|
if type(chunk_size) is not int or not 0 < chunk_size <= 1024 * 1024:
|
|
raise ProtocolError("V1 chunk size must be in 1..1048576")
|
|
for value in (connect_timeout, stall_timeout, total_timeout):
|
|
if not math.isfinite(value) or value <= 0:
|
|
raise ProtocolError("V1 timeouts must be positive and finite")
|
|
if not isinstance(expected_sha256, str) or re.fullmatch(r"[0-9a-fA-F]{64}", expected_sha256) is None:
|
|
raise ProtocolError("V1 transfer requires the reviewed SHA-256")
|
|
expected_sha256 = expected_sha256.lower()
|
|
deadline = time.monotonic() + total_timeout
|
|
bytes_sent = wire_bytes_sent = 0
|
|
try:
|
|
with Path(rom_path).open("rb") as rom:
|
|
md5, sha256 = hashlib.md5(), hashlib.sha256()
|
|
hashed_size = 0
|
|
while True:
|
|
_upgrade_wait(deadline, stall_timeout)
|
|
chunk = rom.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
hashed_size += len(chunk)
|
|
if hashed_size > rom_size:
|
|
raise ProtocolError("V1 ROM size changed before transfer")
|
|
md5.update(chunk)
|
|
sha256.update(chunk)
|
|
if hashed_size != rom_size or sha256.hexdigest() != expected_sha256:
|
|
raise ProtocolError("V1 ROM size or SHA-256 changed before transfer")
|
|
md5_hex = md5.hexdigest()
|
|
prefix = struct.pack("!II", UPGRADE_COMMAND, 32 + rom_size) + md5_hex.encode("ascii")
|
|
rom.seek(0)
|
|
logger.emit("v1_upgrade_prepared", host=host, port=INFO_PORT,
|
|
command=hex(UPGRADE_COMMAND), data_length=32 + rom_size,
|
|
rom_size=rom_size, md5=md5_hex, sha256=expected_sha256)
|
|
with socket.create_connection((host, INFO_PORT),
|
|
timeout=_upgrade_wait(deadline, connect_timeout)) as sock:
|
|
logger.emit("v1_upgrade_connected", host=host, port=INFO_PORT)
|
|
|
|
def send_part(part, is_rom):
|
|
nonlocal bytes_sent, wire_bytes_sent
|
|
view = memoryview(part)
|
|
while view:
|
|
sock.settimeout(_upgrade_wait(deadline, stall_timeout))
|
|
count = sock.send(view)
|
|
if not 0 < count <= len(view):
|
|
raise ProtocolError("V1 connection stopped accepting upload bytes")
|
|
wire_bytes_sent += count
|
|
if is_rom:
|
|
bytes_sent += count
|
|
view = view[count:]
|
|
|
|
send_part(prefix, False)
|
|
sent_sha256 = hashlib.sha256()
|
|
next_progress = 4 * 1024 * 1024
|
|
while bytes_sent < rom_size:
|
|
chunk = rom.read(min(chunk_size, rom_size - bytes_sent))
|
|
if not chunk:
|
|
raise ProtocolError("V1 ROM ended during upload")
|
|
sent_sha256.update(chunk)
|
|
send_part(chunk, True)
|
|
if bytes_sent >= next_progress or bytes_sent == rom_size:
|
|
logger.emit("v1_upload_progress", bytes_sent=bytes_sent,
|
|
rom_size=rom_size, wire_bytes_sent=wire_bytes_sent)
|
|
next_progress = bytes_sent + 4 * 1024 * 1024
|
|
if rom.read(1) or sent_sha256.hexdigest() != expected_sha256:
|
|
raise ProtocolError("V1 ROM changed during upload; result is uncertain")
|
|
logger.emit("v1_upload_complete", bytes_sent=bytes_sent,
|
|
wire_bytes_sent=wire_bytes_sent, awaiting_acknowledgment=True)
|
|
|
|
def receive(count):
|
|
data = bytearray()
|
|
while len(data) < count:
|
|
sock.settimeout(_upgrade_wait(deadline, stall_timeout))
|
|
chunk = sock.recv(count - len(data))
|
|
if not chunk:
|
|
raise ProtocolError("V1 connection closed before the upgrade acknowledgment")
|
|
data.extend(chunk)
|
|
return bytes(data)
|
|
|
|
header = receive(8)
|
|
command, size = struct.unpack("!II", header)
|
|
if command != UPGRADE_COMMAND or not 0 < size <= MAX_JSON_SIZE:
|
|
raise ProtocolError("V1 upgrade response command/length is invalid")
|
|
body = decode_frame(header + receive(size), UPGRADE_COMMAND)
|
|
if type(body.get("code")) is not int or body["code"] != 0:
|
|
raise ProtocolError("V1 upgrade did not report integer code=0")
|
|
_string_fields(body, ("desc",))
|
|
# Retain only documented acknowledgment fields, not arbitrary data.
|
|
acknowledgment = {k: body[k] for k in ("code", "desc") if k in body}
|
|
logger.emit("v1_upgrade_acknowledged", command=hex(command),
|
|
response=acknowledgment, bytes_sent=bytes_sent)
|
|
return {"acknowledged": True, "response_command": command,
|
|
"response_code": body["code"], "response": acknowledgment,
|
|
"bytes_sent": bytes_sent, "wire_bytes_sent": wire_bytes_sent,
|
|
"md5": md5_hex, "sha256": expected_sha256}
|
|
except (OSError, ValueError) as exc:
|
|
logger.emit("v1_transfer_error", error_type=type(exc).__name__,
|
|
bytes_sent=bytes_sent, wire_bytes_sent=wire_bytes_sent,
|
|
automatic_retry=False)
|
|
raise
|
|
|
|
|
|
def parse_discovery(packet: bytes, session: int) -> dict:
|
|
body = decode_frame(packet, DISCOVERY_RESPONSE)
|
|
if type(body.get("session")) is not int or body["session"] != session:
|
|
raise ProtocolError("V1 discovery session mismatch")
|
|
if not isinstance(body.get("sn"), str) or not body["sn"].strip():
|
|
raise ProtocolError("V1 discovery requires a complete SN")
|
|
if "code" in body and (type(body["code"]) is not int or body["code"] != 0):
|
|
raise ProtocolError("V1 discovery returned a failure code")
|
|
_string_fields(body, DISCOVERY_FIELDS[1:])
|
|
return {key: body[key] for key in DISCOVERY_FIELDS if key in body}
|
|
|
|
|
|
def discover(interfaces, timeout: float, client_port: int, logger,
|
|
expected_sn=None, expected_model=None, session=None) -> dict:
|
|
"""Send ff00 to selected directed broadcasts and receive ff01 on a wildcard socket.
|
|
|
|
The specification says replies are broadcast, so binding the receiver only to a
|
|
unicast interface address can lose them. One exclusive receive socket also keeps
|
|
the same source port for all requests. Routing selects the outgoing interface for
|
|
each subnet broadcast; this does not override Windows routing policy.
|
|
"""
|
|
session = secrets.randbelow(0xFFFFFFFF) + 1 if session is None else session
|
|
frame = encode_frame(DISCOVERY_REQUEST, {"session": session})
|
|
targets = {}
|
|
for interface in interfaces:
|
|
if interface.prefix_length > 30:
|
|
logger.emit("v1_interface_skipped", address=str(interface.address),
|
|
reason="no_subnet_broadcast_for_prefix")
|
|
continue
|
|
targets.setdefault(str(interface.network.broadcast_address), []).append(str(interface.address))
|
|
if not targets:
|
|
raise ProtocolError("No V1 interface has a usable directed broadcast")
|
|
found = []
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
# Do not enable SO_REUSEADDR or stop another process to acquire this port.
|
|
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
|
|
try:
|
|
sock.bind(("0.0.0.0", client_port))
|
|
except OSError as exc:
|
|
logger.emit("v1_bind_error", address="0.0.0.0", port=client_port,
|
|
error_type=type(exc).__name__, automatic_process_stop=False)
|
|
raise ProtocolError("Cannot bind V1 UDP receive port; check port ownership") from exc
|
|
sock.setblocking(False)
|
|
logger.emit("v1_socket_ready", address="0.0.0.0", client_port=client_port)
|
|
sent = 0
|
|
for target, addresses in targets.items():
|
|
try:
|
|
sock.sendto(frame, (target, DISCOVERY_PORT))
|
|
sent += 1
|
|
logger.emit("v1_broadcast_sent", selected_interface_addresses=addresses,
|
|
destination=target, port=DISCOVERY_PORT, session=session)
|
|
except OSError as exc:
|
|
logger.emit("v1_send_error", destination=target, error_type=type(exc).__name__)
|
|
if not sent:
|
|
raise ProtocolError("No V1 broadcast could be sent")
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
readable, _, _ = select.select([sock], [], [], remaining)
|
|
if not readable:
|
|
break
|
|
for ready in readable:
|
|
try:
|
|
packet, peer = ready.recvfrom(65535)
|
|
if packet == frame:
|
|
logger.emit("v1_own_request_ignored")
|
|
continue
|
|
body = parse_discovery(packet, session)
|
|
address = ipaddress.IPv4Address(peer[0])
|
|
if (address.is_unspecified or address.is_multicast or address.is_loopback
|
|
or str(address) == "255.255.255.255"):
|
|
raise ProtocolError("V1 discovery has an invalid source address")
|
|
except (OSError, ValueError) as exc:
|
|
logger.emit("v1_invalid_response", error_type=type(exc).__name__)
|
|
continue
|
|
model = body.get("dev_type")
|
|
found.append({"sn": body["sn"], "source_ip": str(address),
|
|
"source_port": peer[1], "received_on": "0.0.0.0:%d" % client_port,
|
|
"mac": body.get("mac"),
|
|
"models": [model] if isinstance(model, str) and model.strip() else [],
|
|
"response": body})
|
|
finally:
|
|
sock.close()
|
|
unique = {}
|
|
for candidate in found:
|
|
key = (candidate["sn"], candidate["source_ip"], candidate["mac"], tuple(candidate["models"]))
|
|
unique[key] = candidate
|
|
candidates = list(unique.values())
|
|
groups = {}
|
|
for candidate in candidates:
|
|
groups.setdefault(candidate["sn"], []).append(candidate)
|
|
conflicts = [sn for sn, identities in groups.items() if len(identities) > 1]
|
|
matches = [c for c in candidates if (expected_sn is None or c["sn"] == expected_sn)
|
|
and (expected_model is None or expected_model in c["models"])]
|
|
blocked = bool(conflicts or (expected_sn is not None and len(matches) != 1))
|
|
return {"status": "BLOCKED" if blocked else "DISCOVERY_COMPLETE",
|
|
"protocol": "v1", "mode": "read_only", "network_used": True,
|
|
"write_executed": False, "firmware_verified": False,
|
|
"identity_verified": bool(expected_sn is not None and len(matches) == 1 and not conflicts),
|
|
"candidate_count": len(candidates), "candidates": candidates,
|
|
"matches": [] if conflicts else matches, "match_count": 0 if conflicts else len(matches),
|
|
"identity_conflicts": conflicts, "expected_sn": expected_sn,
|
|
"expected_model": expected_model,
|
|
"reason": "identity_conflict" if conflicts else ("target_not_uniquely_found" if blocked else "read_only_discovery"),
|
|
"note": "UDP receive port is configurable; no response does not prove the device is offline."}
|