- def build_msg(cmd: str, payload: bytes) -> bytes:
- name = cmd.encode("ascii")
- if len(name) > 12: raise ValueError("command too long")
- header = struct.pack("<I", MAGIC) + name + b"\x00"*(12-len(name))
- header += struct.pack("<I", len(payload)) + sha256d(payload)[:4]
- return header + payload
- def build_version_payload() -> bytes:
- version = VERSION; services = 0; ts = int(time.time())
- nonce = int.from_bytes(sha256d(str(ts).encode())[:8], "little")
- ua = b"/connector/"
- payload = struct.pack("<iQq", version, services, ts)
- payload += struct.pack("<Q", services) + ip6_from_ipv4("0.0.0.0") + struct.pack(">H", DEFAULT_PORT)
- payload += struct.pack("<Q", services) + ip6_from_ipv4("0.0.0.0") + struct.pack(">H", 0)
- payload += struct.pack("<Q", nonce)
- payload += bytes([len(ua)]) + ua
- payload += struct.pack("<i", 0) + b"\x00"
- return payload
- def build_ping_payload(nonce: int) -> bytes:
- return struct.pack("<Q", nonce)