Skip to content

Core

simulator

The thin layer between a transport and a protocol, where chaos gets its turn.

ProtocolSimulator

Bases: Protocol

Anything with a name and a bytes-in, bytes-out handle qualifies. The bar is low.

Simulator dataclass 

Simulator(
    protocol: ProtocolSimulator,
    chaos: ChaosProfile | None = None,
)

Ask the protocol for a reply, then let chaos ruin it if chaos is enabled.

handle 

handle(data: bytes) -> bytes

Return the protocol's reply, possibly dropped, mangled, delayed or sent twice.

Duplicates come back glued together in one buffer, which is exactly how they'd look on the wire anyway.

Source code in src/morpheus_vlt/core/simulator.py 
33
34
35
36
37
38
39
40
41
42
43
def handle(self, data: bytes) -> bytes:
    """Return the protocol's reply, possibly dropped, mangled, delayed or sent twice.

    Duplicates come back glued together in one buffer, which is exactly how they'd look
    on the wire anyway.
    """
    out = self.protocol.handle(data)
    if self.chaos and self.chaos.enabled:
        bufs = self.chaos.apply(out)
        return b"".join(bufs)
    return out

build_handler 

build_handler(
    sim: Simulator, protocol: str
) -> Callable[[bytes], bytes]

Wrap sim in the bytes handler a transport wants. The CLI and the pytest fixture share it.

Unidesa frames arrive however the kernel feels like delivering them: half a frame now, three glued together later. For that family the handler buffers the stream, cuts it into whole frames and answers each one. Everybody else gets their bytes as they come and buffers on their own.

Either way, if the protocol blows up, the error gets logged and the reply is empty. The client sits there wondering, which is honestly what real hardware does, and the server lives to answer the next request instead of taking the whole line down with it.

Source code in src/morpheus_vlt/core/simulator.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def build_handler(sim: Simulator, protocol: str) -> Callable[[bytes], bytes]:
    """Wrap `sim` in the bytes handler a transport wants. The CLI and the pytest fixture share it.

    Unidesa frames arrive however the kernel feels like delivering them: half a frame now, three
    glued together later. For that family the handler buffers the stream, cuts it into whole
    frames and answers each one. Everybody else gets their bytes as they come and buffers on
    their own.

    Either way, if the protocol blows up, the error gets logged and the reply is empty. The
    client sits there wondering, which is honestly what real hardware does, and the server
    lives to answer the next request instead of taking the whole line down with it.
    """
    log = logging.getLogger("morpheus-vlt")

    if protocol in UNIDESA_FAMILY:
        stream_buf = bytearray()

        def unidesa_handler(buf: bytes) -> bytes:
            stream_buf.extend(buf)
            out = bytearray()
            for frame in split_unidesa_frames(stream_buf):
                try:
                    out.extend(sim.handle(frame))
                except Exception:
                    log.exception("Handler error")
            return bytes(out)

        return unidesa_handler

    def handler(buf: bytes) -> bytes:
        try:
            return sim.handle(buf)
        except Exception:
            log.exception("Handler error")
            return b""

    return handler

chaos

Fault injection for outgoing replies.

Real serial lines drop bytes, double up, flip bits and take their sweet time. Your parser has to survive all of that, and the only way to find out is to do it to it on purpose. Chaos is love.

ChaosProfile dataclass 

ChaosProfile(
    enabled: bool = False,
    drop_rate: float = 0.0,
    duplicate_rate: float = 0.0,
    corrupt_rate: float = 0.0,
    flip_bit_rate: float = 0.0,
    delay_ms_min: int = 0,
    delay_ms_max: int = 0,
    seed: int | None = None,
)

Fault injection for outgoing bytes.

Every rate is a probability between 0 and 1, and none of them matter unless enabled is True. Want the same disaster twice? Pass seed, or set MORPHEUS_VLT_CHAOS_SEED. Heads up: the seed goes into the global random module, so anything else in the process using random gets reproducible too, whether it asked or not.

apply 

apply(payload: bytes) -> list[bytes]

Mess with payload and return 0, 1 or 2 copies of whatever survived.

The order is: drop (game over, empty list), corrupt one byte, flip one bit, sleep, then maybe duplicate. A duplicate is a copy of the already mangled reply, because if it's going to be wrong it might as well be consistently wrong.

Source code in src/morpheus_vlt/core/chaos.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def apply(self, payload: bytes) -> list[bytes]:
    """Mess with `payload` and return 0, 1 or 2 copies of whatever survived.

    The order is: drop (game over, empty list), corrupt one byte, flip one bit, sleep, then
    maybe duplicate. A duplicate is a copy of the already mangled reply, because if it's going
    to be wrong it might as well be consistently wrong.
    """
    if not self.enabled:
        return [payload]

    if self.drop_rate and random.random() < self.drop_rate:
        return []

    out = payload

    if self.corrupt_rate and len(out) and random.random() < self.corrupt_rate:
        b = bytearray(out)
        idx = random.randrange(0, len(b))
        b[idx] = (b[idx] + random.randrange(1, 255)) & 0xFF
        out = bytes(b)

    if self.flip_bit_rate and len(out) and random.random() < self.flip_bit_rate:
        b = bytearray(out)
        idx = random.randrange(0, len(b))
        bit = 1 << random.randrange(0, 8)
        b[idx] ^= bit
        out = bytes(b)

    if self.delay_ms_max or self.delay_ms_min:
        lo = max(0, self.delay_ms_min)
        hi = max(lo, self.delay_ms_max)
        delay_ms = random.randint(lo, hi)
        if delay_ms:
            time.sleep(delay_ms / 1000.0)

    bufs = [out]
    if self.duplicate_rate and random.random() < self.duplicate_rate:
        bufs.append(out)

    return bufs

transport

Byte pipes: a TCP server and a fake serial port.

Neither one knows or cares what protocol is running. Bytes come in, the handler gets called, bytes go out. The dumbest possible design, which is why it works.

ServerConfig dataclass 

ServerConfig(
    host: str = "127.0.0.1",
    port: int = 9000,
    recv_buffer: int = 4096,
)

Where the TCP server listens and how much it reads per gulp.

PtyConfig dataclass 

PtyConfig(
    link_path: str | None = None,
    recv_buffer: int = 4096,
    raw: bool = True,
)

PTY settings. link_path is the friendly symlink, raw turns off the tty's opinions.

TCPByteServer 

TCPByteServer(cfg: ServerConfig, handler: BytesHandler)

Minimal TCP server: request bytes in, response bytes out.

One thread per client and no framing at all. A single recv can hold half a message or three of them glued together, and sorting that out is the handler's problem. start blocks, so run it in a thread if you want your terminal back.

Source code in src/morpheus_vlt/core/transport.py 
49
50
51
52
53
54
55
def __init__(self, cfg: ServerConfig, handler: BytesHandler) -> None:
    self.cfg = cfg
    self.handler = handler
    self._sock: socket.socket | None = None
    self._stop = threading.Event()
    # Set once the socket is listening. Wait on it before connecting, or enjoy the race.
    self.ready = threading.Event()

start 

start() -> None

Bind, listen, and accept clients until stop closes the socket out from under us.

Source code in src/morpheus_vlt/core/transport.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def start(self) -> None:
    """Bind, listen, and accept clients until `stop` closes the socket out from under us."""
    self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self._sock.bind((self.cfg.host, self.cfg.port))
    self._sock.listen(5)
    self.ready.set()

    while not self._stop.is_set():
        try:
            conn, _addr = self._sock.accept()
        except OSError:
            # stop() closed the socket. That's our cue to leave.
            break
        t = threading.Thread(target=self._serve_client, args=(conn,), daemon=True)
        t.start()

stop 

stop() -> None

Close the listening socket. Clients already connected finish their current request.

Source code in src/morpheus_vlt/core/transport.py 
74
75
76
77
78
79
def stop(self) -> None:
    """Close the listening socket. Clients already connected finish their current request."""
    self._stop.set()
    if self._sock:
        with contextlib.suppress(OSError):
            self._sock.close()

PtyByteServer 

PtyByteServer(cfg: PtyConfig, handler: BytesHandler)

PTY-backed byte server for serial-style clients.

Opens a pseudo-terminal, optionally drops a symlink to it at a name you can remember, and serves whoever opens the other end. Your pyserial code thinks it found a real port. It didn't. Linux/macOS only, because Windows and PTYs have never been on speaking terms.

When nothing arrives for 100 ms the handler gets called with b"". Most protocols shrug and return nothing. SAS uses it to keep announcing its address until the host shows up.

Source code in src/morpheus_vlt/core/transport.py 
104
105
106
107
108
109
110
111
112
113
def __init__(self, cfg: PtyConfig, handler: BytesHandler) -> None:
    self.cfg = cfg
    self.handler = handler
    self._stop = threading.Event()
    # Set once the PTY exists and the symlink points at it. Open the device after that, not before.
    self.ready = threading.Event()
    self._master_fd: int | None = None
    self._slave_fd: int | None = None
    self._link_path: str | None = None
    self.slave_name: str | None = None

start 

start() -> None

Open the PTY, make the symlink, then pump bytes until stop. Blocks, like its cousin.

Source code in src/morpheus_vlt/core/transport.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def start(self) -> None:
    """Open the PTY, make the symlink, then pump bytes until `stop`. Blocks, like its cousin."""
    self._master_fd, self._slave_fd = os.openpty()
    self.slave_name = os.ttyname(self._slave_fd)
    if self.cfg.raw:
        tty.setraw(self._slave_fd)
    if self.cfg.link_path:
        self._link_path = self._create_link(self.cfg.link_path, self.slave_name)
    logging.getLogger("morpheus-vlt").info(
        "PTY ready: %s (link: %s)", self.slave_name, self._link_path or "none"
    )
    self.ready.set()

    while not self._stop.is_set():
        master_fd = self._master_fd
        if master_fd is None:
            break
        try:
            readable, _, _ = select.select([master_fd], [], [], 0.1)
        except OSError:
            break
        if not readable:
            # Quiet line. Give the protocol a chance to speak up unprompted.
            idle_resp = self.handler(b"")
            if idle_resp:
                try:
                    os.write(master_fd, idle_resp)
                except OSError:
                    break
            continue
        try:
            data = os.read(master_fd, self.cfg.recv_buffer)
        except OSError:
            break
        if not data:
            continue
        resp = self.handler(data)
        if resp:
            try:
                os.write(master_fd, resp)
            except OSError:
                break

stop 

stop() -> None

Close both PTY ends, remove the symlink. Leave the campsite cleaner than you found it.

Source code in src/morpheus_vlt/core/transport.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def stop(self) -> None:
    """Close both PTY ends, remove the symlink. Leave the campsite cleaner than you found it."""
    self._stop.set()
    if self._master_fd is not None:
        with contextlib.suppress(OSError):
            os.close(self._master_fd)
        self._master_fd = None
    if self._slave_fd is not None:
        with contextlib.suppress(OSError):
            os.close(self._slave_fd)
        self._slave_fd = None
    if self._link_path:
        with contextlib.suppress(OSError):
            os.unlink(self._link_path)
        self._link_path = None

framing

Unidesa/Cirsa frame helpers.

A frame looks like STX SD SO NM CMD ND DATA CHK EXT. STX is 0x02, EXT is 0x03, ND is the data length and CHK is an "XOR55" checksum, which is a byte sum wearing a 0x55 mask and calling itself security. Eight bytes of overhead, ND bytes of payload, zero bytes of mercy.

UnidesaMessage dataclass 

UnidesaMessage(
    stx: int,
    sd: int,
    so: int,
    nm: int,
    cmd: int,
    nd: int,
    data: bytes,
    chk: int,
    ext: int,
)

One parsed Unidesa frame, field by field, exactly as it came off the wire.

parse classmethod 

parse(raw: bytes) -> UnidesaMessage

Slice raw into fields. Doesn't check anything yet, that's what validate is for.

Source code in src/morpheus_vlt/core/framing.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
@classmethod
def parse(cls, raw: bytes) -> UnidesaMessage:
    """Slice `raw` into fields. Doesn't check anything yet, that's what `validate` is for."""
    if len(raw) < 8:
        raise ValueError("frame too short")
    stx, sd, so, nm, cmd, nd = raw[:6]
    total = 8 + nd
    if len(raw) < total:
        raise ValueError("incomplete frame")
    data = raw[6:6+nd]
    chk = raw[6+nd]
    ext = raw[7+nd]
    return cls(stx, sd, so, nm, cmd, nd, data, chk, ext)

validate 

validate() -> None

Raise ValueError if the delimiters are wrong or the checksum doesn't add up.

Source code in src/morpheus_vlt/core/framing.py 
50
51
52
53
54
55
56
def validate(self) -> None:
    """Raise ValueError if the delimiters are wrong or the checksum doesn't add up."""
    if self.stx != 0x02 or self.ext != 0x03:
        raise ValueError("bad STX/EXT")
    expect = xor55_checksum(bytes([self.stx, self.sd, self.so, self.nm, self.cmd, self.nd]) + self.data)
    if expect != self.chk:
        raise ValueError("checksum mismatch")

xor55_checksum 

xor55_checksum(buf: bytes) -> int

Add up every byte, XOR with 0x55, keep the low byte. That's the whole algorithm.

Source code in src/morpheus_vlt/core/framing.py 
14
15
16
17
18
19
def xor55_checksum(buf: bytes) -> int:
    """Add up every byte, XOR with 0x55, keep the low byte. That's the whole algorithm."""
    s = 0
    for b in buf:
        s += b
    return (s ^ 0x55) & 0xFF

build_unidesa_reply 

build_unidesa_reply(
    sd: int, so: int, nm: int, cmd: int, data: bytes
) -> bytes

Wrap data in a complete frame with STX, length, checksum and EXT.

Source code in src/morpheus_vlt/core/framing.py 
59
60
61
62
63
64
65
66
def build_unidesa_reply(sd: int, so: int, nm: int, cmd: int, data: bytes) -> bytes:
    """Wrap `data` in a complete frame with STX, length, checksum and EXT."""
    stx = 0x02
    ext = 0x03
    nd = len(data)
    hdr = bytes([stx, sd, so, nm, cmd, nd])
    chk = xor55_checksum(hdr + data)
    return hdr + data + bytes([chk, ext])

split_unidesa_frames 

split_unidesa_frames(buf: bytearray) -> Iterable[bytes]

Yield every complete frame in buf, eating them as it goes.

Garbage in front of an STX gets thrown away. A partial frame at the end stays in buf and waits for the rest to show up. TCP doesn't care where your frames end, so somebody has to.

Source code in src/morpheus_vlt/core/framing.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def split_unidesa_frames(buf: bytearray) -> Iterable[bytes]:
    """Yield every complete frame in `buf`, eating them as it goes.

    Garbage in front of an STX gets thrown away. A partial frame at the end stays in `buf` and
    waits for the rest to show up. TCP doesn't care where your frames end, so somebody has to.
    """
    while buf:
        # Not an STX? Not our problem anymore.
        if buf[0] != 0x02:
            del buf[0]
            continue
        if len(buf) < 6:
            return
        nd = buf[5]
        total = 8 + nd
        if len(buf) < total:
            return
        frame = bytes(buf[:total])
        del buf[:total]
        yield frame

factory

Turns a protocol name into a ready-to-lie simulator.

build_protocol 

build_protocol(
    name: str,
    state: MachineState,
    protocol_type: str | None,
    addr: int,
    dex_file: str | None = None,
    dex_comm_id: str | None = None,
    dex_revision: str | None = None,
    dex_response_code: str | None = None,
) -> Any

Build the simulator for name and hand it the shared state.

protocol_type only matters to VDAI, addr only to SAS, and the dex_* knobs only to DEX. Everybody else ignores what they don't need, like a sane coworker. An unknown name exits the program instead of guessing, because guessing is how you end up debugging at 2 AM.

Source code in src/morpheus_vlt/core/factory.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def build_protocol(
    name: str,
    state: MachineState,
    protocol_type: str | None,
    addr: int,
    dex_file: str | None = None,
    dex_comm_id: str | None = None,
    dex_revision: str | None = None,
    dex_response_code: str | None = None,
) -> Any:
    """Build the simulator for `name` and hand it the shared `state`.

    `protocol_type` only matters to VDAI, `addr` only to SAS, and the `dex_*` knobs only to DEX.
    Everybody else ignores what they don't need, like a sane coworker. An unknown name exits the
    program instead of guessing, because guessing is how you end up debugging at 2 AM.
    """
    if name in ("unidesa", "cirsa"):
        sim = UnidesaSimulator(name=name)
        sim.state = state
        return sim
    if name == "sas":
        sim = SASSimulator()
        sim.state = state
        sim.addr = addr
        return sim
    if name == "vdai":
        sim = VDAISimulator()
        sim.state = state
        if protocol_type:
            sim.protocol_type = protocol_type
        return sim
    if name == "rfranco":
        sim = RFrancoSimulator()
        sim.state = state
        return sim
    if name == "dex":
        sim = DexSimulator()
        sim.state = state
        sim.configure_identity(
            communication_id=dex_comm_id,
            revision_level=dex_revision,
            response_code=dex_response_code,
        )
        if dex_file:
            sim.load_evadts_file(dex_file)
        return sim
    raise SystemExit(f"Unknown protocol: {name}")