mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-10-30 12:03:54 +02:00
618 lines
25 KiB
Python
Executable File
618 lines
25 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
mop_console.py -- minimal DEC MOP Remote Console ("console carrier") client.
|
|
|
|
Talks the DEC Maintenance Operations Protocol Remote Console protocol
|
|
(Ethernet protocol type 0x6002) directly to a MAC address. No IP, no
|
|
DECnet stack, no dependencies beyond the Python standard library.
|
|
|
|
It was written against the VT1200 (DWT remote console) but the framing is
|
|
the generic MOP console carrier, so it should talk to a DECserver too.
|
|
|
|
Protocol references (see dwt/notes/mop-protocol.md for the full write-up):
|
|
* DEC MOP Functional Specification v3.0.0 / v4.0.0, section 5.3
|
|
(Reserve Console 13, Release Console 15, Console Command and Poll 17,
|
|
Console Response and Acknowledge 19, System ID 7, Request ID 5).
|
|
* latd's moprc.cc -- a working console requester; establishes the on-wire
|
|
header: 2-byte little-endian length, then the 1-byte function code.
|
|
* VSI DECnet-Plus Network Management Guide section 10.2.2.1 -- the
|
|
8-byte "verification" value in Reserve Console is the MOP service
|
|
password, default 0000000000000000.
|
|
|
|
Wire format of a frame (Ethernet II):
|
|
dst[6] src[6] ethertype=0x6002
|
|
len_lo len_hi little-endian: bytes following this field
|
|
code MOP message code
|
|
body[...] code-specific
|
|
0x00 padding to the 46-byte Ethernet payload minimum
|
|
|
|
Capability: needs CAP_NET_RAW. Either run as root (`sudo python3 ...`) or
|
|
grant the interpreter the capability:
|
|
sudo setcap cap_net_raw+ep "$(readlink -f "$(which python3)")"
|
|
|
|
Examples:
|
|
sudo ./mop_console.py --iface vt1200 --mac 08:00:2b:16:4b:5a HELP
|
|
sudo ./mop_console.py --iface vt1200 --mac 08:00:2b:16:4b:5a --script
|
|
sudo ./mop_console.py --iface vt1200 --mac 08:00:2b:16:4b:5a --raw 48454c50
|
|
./mop_console.py --iface vt1200 --mac 08:00:2b:16:4b:5a --dry-run --script
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import binascii
|
|
import fcntl
|
|
import os
|
|
import select
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
ETHERTYPE_MOPRC = 0x6002
|
|
ETHERTYPE_MOPDL = 0x6001 # documented for contrast; not used
|
|
MOP_MULTICAST = "ab:00:00:02:00:00"
|
|
BROADCAST = "ff:ff:ff:ff:ff:ff"
|
|
|
|
# MOP function codes (MOP spec 5.3; latd moprc.h uses the same numbers)
|
|
CODE_REQUEST_ID = 5 # "Request ID"
|
|
CODE_BOOT = 6
|
|
CODE_SYSTEM_ID = 7 # "System ID"
|
|
CODE_REQUEST_COUNTERS = 9
|
|
CODE_COUNTERS = 11
|
|
CODE_RESERVE_CONSOLE = 13
|
|
CODE_RELEASE_CONSOLE = 15
|
|
CODE_CONSOLE_COMMAND_POLL = 17 # host -> terminal, carries console input
|
|
CODE_CONSOLE_RESPONSE = 19 # terminal -> host, carries console output
|
|
|
|
CODE_NAMES = {
|
|
CODE_REQUEST_ID: "Request ID",
|
|
CODE_BOOT: "Boot",
|
|
CODE_SYSTEM_ID: "System ID",
|
|
CODE_REQUEST_COUNTERS: "Request Counters",
|
|
CODE_COUNTERS: "Counters",
|
|
CODE_RESERVE_CONSOLE: "Reserve Console",
|
|
CODE_RELEASE_CONSOLE: "Release Console",
|
|
CODE_CONSOLE_COMMAND_POLL: "Console Command and Poll",
|
|
CODE_CONSOLE_RESPONSE: "Console Response and Acknowledge",
|
|
}
|
|
|
|
|
|
def mac_bytes(text: str) -> bytes:
|
|
parts = text.replace("-", ":").split(":")
|
|
if len(parts) != 6:
|
|
raise argparse.ArgumentTypeError(f"bad MAC address: {text!r}")
|
|
try:
|
|
return bytes(int(p, 16) for p in parts)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError(f"bad MAC address: {text!r}")
|
|
|
|
|
|
def mac_text(raw: bytes) -> str:
|
|
return ":".join(f"{b:02x}" for b in raw)
|
|
|
|
|
|
def iface_mac(iface: str) -> bytes:
|
|
"""SIOCGIFHWADDR for an interface name, without shelling out."""
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
info = fcntl.ioctl(
|
|
s.fileno(),
|
|
0x8927, # SIOCGIFHWADDR
|
|
struct.pack("256s", iface.encode()[:15]),
|
|
)
|
|
finally:
|
|
s.close()
|
|
return info[18:24]
|
|
|
|
|
|
def printable(data: bytes) -> str:
|
|
return "".join(chr(b) if 32 <= b < 127 else "." for b in data)
|
|
|
|
|
|
def hexdump(data: bytes, prefix: str = " ") -> str:
|
|
"""Classic 16-byte hex dump with printable gutter."""
|
|
lines = []
|
|
for off in range(0, len(data), 16):
|
|
chunk = data[off:off + 16]
|
|
hexpart = " ".join(f"{b:02x}" for b in chunk)
|
|
hexpart = hexpart[:23] + " " + hexpart[23:] if len(chunk) > 8 else hexpart
|
|
lines.append(f"{prefix}{off:04x} {hexpart:<48} {printable(chunk)}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def encode_message(code: int, body: bytes = b"", pad_to: int = 46) -> bytes:
|
|
"""MOP payload: LE length (code+body), code, body, zero pad."""
|
|
length = 1 + len(body)
|
|
payload = struct.pack("<H", length) + bytes([code]) + body
|
|
if len(payload) < pad_to:
|
|
payload += b"\x00" * (pad_to - len(payload))
|
|
return payload
|
|
|
|
|
|
def parse_message(payload: bytes) -> tuple[int, int, bytes]:
|
|
"""Return (declared_length, code, body). body may include padding."""
|
|
if len(payload) < 3:
|
|
raise ValueError(f"short MOP payload ({len(payload)} bytes)")
|
|
length = struct.unpack_from("<H", payload, 0)[0]
|
|
code = payload[2]
|
|
# The sender pads to 46 bytes; the length field is authoritative.
|
|
body = payload[3:min(len(payload), 2 + length)]
|
|
return length, code, body
|
|
|
|
|
|
def ethernet_frame(dst: bytes, src: bytes, payload: bytes,
|
|
ethertype: int = ETHERTYPE_MOPRC) -> bytes:
|
|
return dst + src + struct.pack(">H", ethertype) + payload
|
|
|
|
|
|
class MopConsole:
|
|
def __init__(self, iface: str, dst: bytes, src: bytes | None,
|
|
ethertype: int, verification: bytes, verbose: bool = True):
|
|
self.iface = iface
|
|
self.dst = dst
|
|
self.src = src if src is not None else iface_mac(iface)
|
|
self.ethertype = ethertype
|
|
self.verification = verification
|
|
self.verbose = verbose
|
|
self.sock = None
|
|
self.seq = 0 # host command message-number bit
|
|
self.receipt = 0
|
|
self.attempts = 0
|
|
self.sent_log: list[bytes] = []
|
|
self.recv_log: list[bytes] = []
|
|
|
|
# ---- transport -----------------------------------------------------
|
|
def open(self):
|
|
try:
|
|
self.sock = socket.socket(
|
|
socket.AF_PACKET, socket.SOCK_RAW, socket.htons(self.ethertype)
|
|
)
|
|
except PermissionError:
|
|
sys.exit(
|
|
"mop_console: permission denied opening an AF_PACKET socket.\n"
|
|
" Need root or CAP_NET_RAW, e.g.\n"
|
|
f" sudo python3 {sys.argv[0]} ...\n"
|
|
" or: sudo setcap cap_net_raw+ep \"$(readlink -f \"$(which python3)\")\""
|
|
)
|
|
self.sock.bind((self.iface, 0))
|
|
self.sock.setblocking(False)
|
|
|
|
def close(self):
|
|
if self.sock:
|
|
self.sock.close()
|
|
self.sock = None
|
|
|
|
def send_frame(self, code: int, body: bytes = b"", note: str = ""):
|
|
payload = encode_message(code, body)
|
|
frame = ethernet_frame(self.dst, self.src, payload, self.ethertype)
|
|
self.attempts += 1
|
|
self.sent_log.append(frame)
|
|
if self.verbose:
|
|
label = note or CODE_NAMES.get(code, f"code {code}")
|
|
print(f" TX {len(frame):3d}B {label}")
|
|
print(hexdump(frame))
|
|
self.sock.send(frame)
|
|
|
|
def recv_frames(self, timeout: float):
|
|
"""Yield (frame, code, body) for frames addressed to us."""
|
|
end = time.monotonic() + timeout
|
|
while True:
|
|
remain = end - time.monotonic()
|
|
if remain <= 0:
|
|
return
|
|
r, _, _ = select.select([self.sock], [], [], remain)
|
|
if not r:
|
|
return
|
|
try:
|
|
frame = self.sock.recv(65535)
|
|
except BlockingIOError:
|
|
continue
|
|
if len(frame) < 14 + 3:
|
|
continue
|
|
etype = struct.unpack_from(">H", frame, 12)[0]
|
|
if etype != self.ethertype:
|
|
continue
|
|
src = frame[6:12]
|
|
dst = frame[0:6]
|
|
# Ignore our own transmitted frame echoed back by AF_PACKET
|
|
# (with --dst broadcast the destination check alone would miss it).
|
|
if src == self.src:
|
|
continue
|
|
# Accept unicast to us and broadcast/multicast.
|
|
if dst != self.src and dst[0] & 1 == 0:
|
|
continue
|
|
try:
|
|
_, code, body = parse_message(frame[14:])
|
|
except ValueError:
|
|
continue
|
|
self.recv_log.append(frame)
|
|
yield frame, src, code, body
|
|
|
|
# ---- protocol ------------------------------------------------------
|
|
def reserve(self):
|
|
self.send_frame(CODE_RESERVE_CONSOLE, self.verification,
|
|
"Reserve Console (verification=%s)"
|
|
% self.verification.hex())
|
|
|
|
def request_id(self):
|
|
body = bytes([0]) + struct.pack("<H", self.receipt)
|
|
self.send_frame(CODE_REQUEST_ID, body, "Request ID")
|
|
|
|
@staticmethod
|
|
def parse_system_id(body: bytes):
|
|
if len(body) < 3:
|
|
return None, {}
|
|
info = body[3:]
|
|
entries = {}
|
|
i = 0
|
|
while i + 3 <= len(info):
|
|
itype = info[i] | (info[i + 1] << 8)
|
|
ilen = info[i + 2]
|
|
i += 3
|
|
if i + ilen > len(info):
|
|
break
|
|
entries[itype] = info[i:i + ilen]
|
|
i += ilen
|
|
functions = entries.get(2, b"\x00")
|
|
functions = functions[0] if functions else 0
|
|
return functions, entries
|
|
|
|
def connect(self, timeout: float = 20.0, interval: float = 2.0,
|
|
require_console: bool = True) -> bool:
|
|
"""Reserve + Request ID, wait for System ID. Mirrors moprc's handshake."""
|
|
start = time.monotonic()
|
|
next_send = start
|
|
got_any = False
|
|
print(f"Connecting to {mac_text(self.dst)} on {self.iface} "
|
|
f"(ethertype 0x{self.ethertype:04x}) ...")
|
|
while time.monotonic() - start < timeout:
|
|
if time.monotonic() >= next_send:
|
|
self.reserve()
|
|
self.request_id()
|
|
next_send = time.monotonic() + interval
|
|
for frame, src, code, body in self.recv_frames(0.25):
|
|
got_any = True
|
|
print(f" RX {len(frame):3d}B {CODE_NAMES.get(code, f'code {code}')}"
|
|
f" from {mac_text(src)}")
|
|
print(hexdump(frame))
|
|
if code != CODE_SYSTEM_ID:
|
|
continue
|
|
functions, entries = self.parse_system_id(body)
|
|
self.show_system_id(entries, functions)
|
|
if functions is None:
|
|
continue
|
|
if require_console and not (functions & 0x20):
|
|
print(" System ID says console carrier (function bit 5) "
|
|
"is NOT supported.")
|
|
return False
|
|
print(" Console carrier available -- connected.")
|
|
return True
|
|
print(f" no System ID after {timeout:.0f}s, {self.attempts} frames sent.")
|
|
return False
|
|
|
|
@staticmethod
|
|
def show_system_id(entries: dict, functions: int | None):
|
|
print(" System ID info:")
|
|
if functions is not None:
|
|
names = ["loop", "dump", "primary loader", "multi-block loader",
|
|
"boot", "console carrier", "data link counters",
|
|
"console reservation"]
|
|
bits = [n for b, n in enumerate(names) if functions & (1 << b)]
|
|
print(f" functions = 0x{functions:02x} ({', '.join(bits) or 'none'})")
|
|
labels = {1: "maintenance version", 3: "console user",
|
|
4: "reservation timer", 5: "console command size",
|
|
6: "console response size", 7: "hardware address",
|
|
100: "communication device", 200: "software id",
|
|
300: "system processor", 400: "data link"}
|
|
for itype, value in entries.items():
|
|
if itype in (0, 2):
|
|
continue
|
|
label = labels.get(itype, f"info type {itype}")
|
|
if itype in (4, 5, 6) and len(value) >= 2:
|
|
print(f" {label}: {struct.unpack_from('<H', value)[0]}")
|
|
elif itype in (200,) and value and 0 < value[0] < 32:
|
|
text = value[1:1 + value[0]].decode("latin-1", "replace")
|
|
print(f" {label}: {text!r}")
|
|
else:
|
|
print(f" {label}: {value.hex()}")
|
|
|
|
def send_data(self, data: bytes, note: str = ""):
|
|
"""Console Command and Poll. Control flags: bit0 msg number, bit1 break."""
|
|
flags = self.seq & 1
|
|
self.send_frame(CODE_CONSOLE_COMMAND_POLL, bytes([flags]) + data,
|
|
note or f"Console Command and Poll ({len(data)} data bytes)")
|
|
self.seq ^= 1
|
|
|
|
def poll(self):
|
|
self.send_data(b"", "Console Command and Poll (empty poll)")
|
|
|
|
def collect(self, quiet: float = 0.8, poll_interval: float = 0.3,
|
|
max_time: float = 45.0) -> bytes:
|
|
"""Collect one console response.
|
|
|
|
The protocol is polled. The VT1200 answers a poll by re-sending its
|
|
whole response buffer (and, for an empty response, a CRA whose length
|
|
field is 2 but whose frame still carries stale buffer bytes). So the
|
|
length field is authoritative, and a repeated chunk is either an exact
|
|
duplicate or a *superset* of the previous chunk -- replace, don't
|
|
append, in that case. Polls stop once no new data has appeared for
|
|
`quiet` seconds.
|
|
"""
|
|
out = bytearray()
|
|
last_chunk = b""
|
|
last_new = 0.0
|
|
last_poll = -1.0
|
|
polls = 0
|
|
start = time.monotonic()
|
|
while time.monotonic() - start < max_time:
|
|
now = time.monotonic()
|
|
if now - last_poll >= poll_interval and polls < 40:
|
|
self.poll()
|
|
polls += 1
|
|
last_poll = now
|
|
for frame, src, code, body in self.recv_frames(0.1):
|
|
print(f" RX {len(frame):3d}B {CODE_NAMES.get(code, f'code {code}')}"
|
|
f" from {mac_text(src)}")
|
|
print(hexdump(frame))
|
|
if code == CODE_CONSOLE_RESPONSE:
|
|
if len(body) >= 1:
|
|
data = body[1:]
|
|
flags = body[0]
|
|
print(f" CRA flags=0x{flags:02x} msg={flags & 1} "
|
|
f"cmd_lost={bool(flags & 2)} resp_lost={bool(flags & 4)}"
|
|
f" data={len(data)}B")
|
|
if data:
|
|
if data == last_chunk:
|
|
print(" (duplicate response chunk -- ignored)")
|
|
elif last_chunk and data.startswith(last_chunk):
|
|
out = out[:-len(last_chunk)] + data
|
|
last_chunk = data
|
|
last_new = time.monotonic()
|
|
print(f" text (buffer grew): {printable(data)}")
|
|
elif last_chunk and last_chunk.startswith(data):
|
|
print(" (shorter stale chunk -- ignored)")
|
|
else:
|
|
out += data
|
|
last_chunk = data
|
|
last_new = time.monotonic()
|
|
print(f" text: {printable(data)}")
|
|
last_poll = time.monotonic()
|
|
elif code == CODE_SYSTEM_ID:
|
|
functions, entries = self.parse_system_id(body)
|
|
self.show_system_id(entries, functions)
|
|
else:
|
|
print(f" body: {body.hex()}")
|
|
if last_new and time.monotonic() - last_new >= quiet:
|
|
break
|
|
if not last_new and polls >= 4 and time.monotonic() - last_poll > 1.0:
|
|
break
|
|
return bytes(out)
|
|
|
|
def command(self, line: str, quiet: float = 1.5) -> bytes:
|
|
print(f"\n=== {line!r} ===")
|
|
self.send_data(line.encode("latin-1", "replace") + b"\r",
|
|
f"Console Command and Poll: {line!r}")
|
|
return self.collect(quiet=quiet)
|
|
|
|
def release(self):
|
|
self.send_frame(CODE_RELEASE_CONSOLE, b"", "Release Console")
|
|
|
|
|
|
def diag(args, client: MopConsole, phase: str, rx: bool):
|
|
print("\n--- diagnosis -------------------------------------------------")
|
|
print(f"phase: {phase}")
|
|
print(f"iface={args.iface} dst={args.mac} src={mac_text(client.src)} "
|
|
f"ethertype=0x{client.ethertype:04x}")
|
|
print(f"frames sent: {len(client.sent_log)}, frames received: {len(client.recv_log)}")
|
|
for i, frame in enumerate(client.sent_log[-4:]):
|
|
print(f"last TX[{i}]: {frame.hex()}")
|
|
print(f" payload (no Ethernet header): {frame[14:].hex()}")
|
|
if client.recv_log:
|
|
for i, frame in enumerate(client.recv_log[-4:]):
|
|
print(f"last RX[{i}]: {frame.hex()}")
|
|
else:
|
|
print("last RX: none")
|
|
print("Hints:")
|
|
print(" * DWT is part of the OS session manager; it does not exist in the")
|
|
print(" power-on diagnostics menu. Boot past level 1 first.")
|
|
print(" * The MOP service password is the 8-byte Reserve Console verification")
|
|
print(" value; wrong password = silent ignore (nothing comes back).")
|
|
print(" * A MOP frame payload is padded to 46 bytes; the 2-byte length field")
|
|
print(" is authoritative. Use --dry-run to inspect frames.")
|
|
print("---------------------------------------------------------------")
|
|
|
|
|
|
def parse_verification(text: str) -> bytes:
|
|
"""Accept 0-16 hex digits, '%X...' NCL form, or 0x.... Pads to 8 bytes.
|
|
|
|
NCL/DECnet-Plus semantics: leftmost digit pair is the first byte on the
|
|
wire (lowest address). NCP integer semantics are handled by --password-int.
|
|
"""
|
|
t = text.strip()
|
|
if t.lower().startswith("0x"):
|
|
t = t[2:]
|
|
elif t.lower().startswith("%x"):
|
|
t = t[2:]
|
|
t = t.replace(":", "").replace("-", "")
|
|
if len(t) % 2:
|
|
t = "0" + t
|
|
raw = bytes.fromhex(t)
|
|
if len(raw) > 8:
|
|
raise SystemExit(f"--password: {len(raw)} bytes, maximum is 8")
|
|
return raw + b"\x00" * (8 - len(raw))
|
|
|
|
|
|
def parse_password_int(text: str) -> bytes:
|
|
"""NCP semantics: the password is an integer; byte 0 is the low byte."""
|
|
value = int(text, 16)
|
|
return value.to_bytes(8, "little")
|
|
|
|
|
|
def build_argparser():
|
|
p = argparse.ArgumentParser(
|
|
description="DEC MOP Remote Console (console carrier) client",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=__doc__.split("Examples:")[1] if "Examples:" in __doc__ else None,
|
|
)
|
|
p.add_argument("--iface", default="vt1200", help="Ethernet interface (default vt1200)")
|
|
p.add_argument("--mac", default="08:00:2b:16:4b:5a",
|
|
help="terminal MAC (default 08:00:2b:16:4b:5a)")
|
|
p.add_argument("--src-mac", default=None,
|
|
help="source MAC (default: the interface's own MAC)")
|
|
p.add_argument("--ethertype", default="0x6002",
|
|
help="MOP protocol type (default 0x6002; 0x6001 is Dump/Load)")
|
|
p.add_argument("--dst", default=None,
|
|
help=f"override destination MAC, e.g. {BROADCAST} or {MOP_MULTICAST}")
|
|
p.add_argument("--password", default=None,
|
|
help="8-byte Reserve Console verification value, hex "
|
|
"(NCL/DECnet-Plus byte order; default 0000000000000000)")
|
|
p.add_argument("--password-int", default=None,
|
|
help="service password as an NCP-style integer (low byte first)")
|
|
p.add_argument("--connect-timeout", type=float, default=20.0,
|
|
help="seconds to wait for System ID (default 20)")
|
|
p.add_argument("--retry-interval", type=float, default=2.0,
|
|
help="seconds between Reserve/RequestID retries (default 2)")
|
|
p.add_argument("--quiet", type=float, default=1.5,
|
|
help="stop collecting after this many quiet seconds (default 1.5)")
|
|
p.add_argument("--no-release", action="store_true",
|
|
help="do not send Release Console at the end")
|
|
p.add_argument("--dry-run", action="store_true",
|
|
help="print the frames that would be sent; no socket")
|
|
p.add_argument("--script", nargs="?", const="HELP,SHOW,TEST 0 0",
|
|
default=None, metavar="CMDS",
|
|
help="run comma-separated commands (default HELP,SHOW,TEST 0 0)")
|
|
p.add_argument("--raw", default=None, metavar="HEX",
|
|
help="send arbitrary console data bytes (inside Console "
|
|
"Command and Poll)")
|
|
p.add_argument("--raw-frame", default=None, metavar="HEX",
|
|
help="send a complete MOP payload (length field + code + body)")
|
|
p.add_argument("--no-connect", action="store_true",
|
|
help="skip Reserve/Request ID (send data immediately)")
|
|
p.add_argument("cmd", nargs="*", help="command line to send, e.g. HELP")
|
|
return p
|
|
|
|
|
|
def dry_run(args, dst: bytes, src: bytes, verification: bytes, ethertype: int):
|
|
print(f"dry run: iface={args.iface} src={mac_text(src)} dst={mac_text(dst)} "
|
|
f"ethertype=0x{ethertype:04x}")
|
|
cmds = []
|
|
if args.script is not None:
|
|
cmds = [c.strip() for c in args.script.split(",") if c.strip()]
|
|
elif args.raw_frame:
|
|
pass
|
|
elif args.raw:
|
|
cmds = [None]
|
|
elif args.cmd:
|
|
cmds = [" ".join(args.cmd)]
|
|
else:
|
|
cmds = ["HELP"]
|
|
|
|
def show(label, code, body):
|
|
payload = encode_message(code, body)
|
|
frame = ethernet_frame(dst, src, payload, ethertype)
|
|
print(f"\n{label}")
|
|
print(f" payload ({len(payload)}B, padded to 46): {payload.hex()}")
|
|
print(hexdump(frame, prefix=" "))
|
|
|
|
if not args.no_connect:
|
|
show("Reserve Console (code 13) -- verification = MOP service password",
|
|
CODE_RESERVE_CONSOLE, verification)
|
|
show("Request ID (code 5) -- reserved=00 receipt=0000",
|
|
CODE_REQUEST_ID, b"\x00\x00\x00")
|
|
if args.raw_frame:
|
|
raw = bytes.fromhex(args.raw_frame.replace(" ", ""))
|
|
frame = ethernet_frame(dst, src, raw, ethertype)
|
|
print("\nraw MOP payload as given")
|
|
print(f" payload: {raw.hex()}")
|
|
print(hexdump(frame, prefix=" "))
|
|
else:
|
|
for i, c in enumerate(cmds):
|
|
data = (b"" if c is None else c.encode() + b"\r")
|
|
if c is None:
|
|
data = bytes.fromhex(args.raw.replace(" ", ""))
|
|
show(f"Console Command and Poll (code 17), msg={i & 1}, data={data!r}",
|
|
CODE_CONSOLE_COMMAND_POLL, bytes([i & 1]) + data)
|
|
print("\nDry run: nothing sent. Remove --dry-run (and run with CAP_NET_RAW).")
|
|
|
|
|
|
def main(argv=None):
|
|
args = build_argparser().parse_args(argv)
|
|
dst = mac_bytes(args.dst or args.mac)
|
|
src = mac_bytes(args.src_mac) if args.src_mac else iface_mac(args.iface)
|
|
ethertype = int(args.ethertype, 0)
|
|
|
|
if args.password and args.password_int:
|
|
raise SystemExit("use either --password or --password-int, not both")
|
|
if args.password_int:
|
|
verification = parse_password_int(args.password_int)
|
|
elif args.password:
|
|
verification = parse_verification(args.password)
|
|
else:
|
|
verification = b"\x00" * 8
|
|
|
|
if args.dry_run:
|
|
dry_run(args, dst, src, verification, ethertype)
|
|
return 0
|
|
|
|
client = MopConsole(args.iface, dst, src, ethertype, verification)
|
|
client.open()
|
|
ok = True
|
|
try:
|
|
if not args.no_connect:
|
|
ok = client.connect(args.connect_timeout, args.retry_interval)
|
|
if not ok:
|
|
diag(args, client, "connect (no System ID)", False)
|
|
return 2
|
|
|
|
if args.raw_frame:
|
|
raw = bytes.fromhex(args.raw_frame.replace(" ", ""))
|
|
frame = ethernet_frame(dst, client.src, raw, ethertype)
|
|
client.sent_log.append(frame)
|
|
print(f"\n=== raw MOP payload {raw.hex()} ===")
|
|
print(hexdump(frame))
|
|
client.sock.send(frame)
|
|
out = client.collect(quiet=args.quiet, max_time=10.0)
|
|
if out:
|
|
print(f"response text: {printable(out)}")
|
|
else:
|
|
print("no console response data")
|
|
diag(args, client, "raw-frame", False)
|
|
elif args.script is not None:
|
|
cmds = [c.strip() for c in args.script.split(",") if c.strip()]
|
|
for c in cmds:
|
|
out = client.command(c, quiet=args.quiet)
|
|
print(f"--- {c}: {len(out)} bytes of console output ---")
|
|
if not out:
|
|
print(" (no response data)")
|
|
print("\nScript complete.")
|
|
elif args.raw:
|
|
data = bytes.fromhex(args.raw.replace(" ", ""))
|
|
print(f"\n=== raw console data {len(data)}B: {data.hex()} ===")
|
|
client.send_data(data, "Console Command and Poll (raw)")
|
|
out = client.collect(quiet=args.quiet, max_time=10.0)
|
|
print(f"response: {out.hex()}")
|
|
if out:
|
|
print(f"text: {printable(out)}")
|
|
else:
|
|
print("no console response data")
|
|
diag(args, client, "raw", False)
|
|
else:
|
|
line = " ".join(args.cmd) if args.cmd else "HELP"
|
|
out = client.command(line, quiet=args.quiet)
|
|
if not out:
|
|
diag(args, client, f"command {line!r}", False)
|
|
ok = False
|
|
finally:
|
|
if ok and not args.no_release:
|
|
# No poll after Release: the terminal treats it as a new request
|
|
# and re-sends its (now stale) response buffer.
|
|
client.release()
|
|
time.sleep(0.2)
|
|
client.close()
|
|
return 0 if ok else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|