Whack multiplexer to handle console too.

Renamed multiplexer to cryptech_muxd, since it now handles both RPC and CTY.

Added new program cryptech_console to act as client for CTY multiplexer.

Might want to add console logging capability eventually, not today.

Probably want to incorporate UART probing (what cryptech_probe does
now) eventually, also not today.

1 files changed, 272 insertions, 0 deletions

diff --git a/cryptech_muxd b/cryptech_muxd
new file mode 100755
index 0000000..80be443
--- /dev/null
+++ b/cryptech_muxd
@@ -0,0 +1,272 @@
+#!/usr/bin/env python
+#
+# Copyright (c) 2016-2017, NORDUnet A/S All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+# - Redistributions of source code must retain the above copyright notice,
+#   this list of conditions and the following disclaimer.
+#
+# - Redistributions in binary form must reproduce the above copyright
+#   notice, this list of conditions and the following disclaimer in the
+#   documentation and/or other materials provided with the distribution.
+#
+# - Neither the name of the NORDUnet nor the names of its contributors may
+#   be used to endorse or promote products derived from this software
+#   without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+# PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+# TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""
+Implementation of Cryptech RPC protocol multiplexer in Python.
+
+Unlike the original C implementation, this uses SLIP encapsulation
+over a SOCK_STREAM channel, because support for SOCK_SEQPACKET is not
+what we might wish.  We outsource all the heavy lifting for serial and
+network I/O to the PySerial and Tornado libraries, respectively.
+"""
+
+import os
+import sys
+import time
+import struct
+import atexit
+import weakref
+import argparse
+
+import serial
+
+import tornado.tcpserver
+import tornado.iostream
+import tornado.netutil
+import tornado.ioloop
+import tornado.queues
+import tornado.locks
+import tornado.gen
+
+
+SLIP_END     = chr(0300)        # Indicates end of SLIP packet
+SLIP_ESC     = chr(0333)        # Indicates byte stuffing
+SLIP_ESC_END = chr(0334)        # ESC ESC_END means END data byte
+SLIP_ESC_ESC = chr(0335)        # ESC ESC_ESC means ESC data byte
+
+
+def slip_encode(buffer):
+    "Encode a buffer using SLIP encapsulation."
+    return SLIP_END + buffer.replace(SLIP_ESC, SLIP_ESC + SLIP_ESC_ESC).replace(SLIP_END, SLIP_ESC + SLIP_ESC_END) + SLIP_END
+
+def slip_decode(buffer):
+    "Decode a SLIP-encapsulated buffer."
+    return buffer.strip(SLIP_END).replace(SLIP_ESC + SLIP_ESC_END, SLIP_END).replace(SLIP_ESC + SLIP_ESC_ESC, SLIP_ESC)
+
+
+def client_handle_get(msg):
+    "Extract client_handle field from a Cryptech RPC message."
+    return struct.unpack(">L", msg[4:8])[0]
+
+def client_handle_set(msg, handle):
+    "Replace client_handle field in a Cryptech RPC message."
+    return msg[:4] + struct.pack(">L", handle) + msg[8:]
+
+
+class SerialIOStream(tornado.iostream.BaseIOStream):
+    """
+    Implementation of a Tornado IOStream over a PySerial device.
+    """
+
+    def __init__(self, device, baudrate = 921600, debug = False, *pargs, **kwargs):
+        self.serial = serial.Serial(device, baudrate, timeout = 0, write_timeout = 0)
+        self.debug  = debug
+        super(SerialIOStream, self).__init__(*pargs, **kwargs)
+
+    def fileno(self):
+        return self.serial.fileno()
+
+    def close_fd(self):
+        self.serial.close()
+
+    def write_to_fd(self, data):
+        return self.serial.write(data)
+
+    def read_from_fd(self):
+        return self.serial.read(self.read_chunk_size) or None
+
+
+class PFUnixServer(tornado.tcpserver.TCPServer):
+    """
+    Variant on tornado.tcpserver.TCPServer, listening on a PF_UNIX
+    (aka PF_LOCAL) socket instead of a TCP socket.
+    """
+
+    def listen(self, filename, mode = 0600):
+        self.socket_filename = filename
+        self.add_socket(tornado.netutil.bind_unix_socket(filename, mode))
+        atexit.register(self.atexit_unlink)
+
+    def atexit_unlink(self):
+        try:
+            os.unlink(self.socket_filename)
+        except:
+            pass
+
+
+class RPCIOStream(SerialIOStream):
+    """
+    Tornado IOStream for a serial RPC channel.
+    """
+
+    def __init__(self, *pargs, **kwargs):
+        super(RPCIOStream, self).__init__(*pargs, **kwargs)
+        self.queues = weakref.WeakValueDictionary()
+        self.rpc_input_lock = tornado.locks.Lock()
+
+    @tornado.gen.coroutine
+    def rpc_input(self, query, handle, queue):
+        "Send a query to the HSM."
+        if self.debug:
+            sys.stdout.write("+send: {}\n".format(":".join("{:02x}".format(ord(c)) for c in query)))
+        self.queues[handle] = queue
+        with (yield self.rpc_input_lock.acquire()):
+            yield self.write(query)
+
+    @tornado.gen.coroutine
+    def rpc_output_loop(self):
+        "Handle reply stream HSM -> network."
+        while True:
+            reply = yield self.read_until(SLIP_END)
+            if self.debug:
+                sys.stdout.write("+recv: {}\n".format(":".join("{:02x}".format(ord(c)) for c in reply)))
+            if len(reply) < 9:
+                continue
+            handle = client_handle_get(slip_decode(reply))
+            self.queues[handle].put_nowait(reply)
+
+
+class RPCServer(PFUnixServer):
+    """
+    Serve multiplexed Cryptech RPC over a PF_UNIX socket.
+    """
+
+    def set_serial(self, serial_stream):
+        self.serial = serial_stream
+
+    @tornado.gen.coroutine
+    def handle_stream(self, stream, address):
+        "Handle one network connection."
+        handle = stream.socket.fileno()
+        queue  = tornado.queues.Queue()
+        closed = False
+        while not closed:
+            try:
+                query = yield stream.read_until(SLIP_END)
+                if len(query) < 9:
+                    continue
+                query = slip_encode(client_handle_set(slip_decode(query), handle))
+                yield self.serial.rpc_input(query, handle, queue)
+                reply = yield queue.get()
+                yield stream.write(SLIP_END + reply)
+            except tornado.iostream.StreamClosedError:
+                closed = True
+
+class CTYIOStream(SerialIOStream):
+    """
+    Tornado IOStream for a serial console channel.
+    """
+
+    def __init__(self, *pargs, **kwargs):
+        super(CTYIOStream, self).__init__(*pargs, **kwargs)
+        self.attached_cty = None
+
+    @tornado.gen.coroutine
+    def cty_output_loop(self):
+        while True:
+            buffer = yield self.read_bytes(1024, partial = True)
+            try:
+                if self.attached_cty is not None:
+                    yield self.attached_cty.write(buffer)
+            except tornado.iostream.StreamClosedError:
+                pass
+
+
+class CTYServer(PFUnixServer):
+    """
+    Serve Cryptech console over a PF_UNIX socket.
+    """
+
+    def set_serial(self, serial_stream):
+        self.serial = serial_stream
+
+    @tornado.gen.coroutine
+    def handle_stream(self, stream, address):
+        "Handle one network connection."
+
+        if self.serial.attached_cty is not None:
+            yield stream.write("[Console already in use, sorry]\n")
+            stream.close()
+            return
+
+        try:
+            self.serial.attached_cty = stream
+            while self.serial.attached_cty is stream:
+                yield self.serial.write((yield stream.read_bytes(1024, partial = True)))
+        except tornado.iostream.StreamClosedError:
+            pass
+        finally:
+            if self.serial.attached_cty is stream:
+                self.serial.attached_cty = None
+
+
+@tornado.gen.coroutine
+def main():
+    parser = argparse.ArgumentParser(formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("-v", "--verbose",  action = "store_true",  help = "produce human-readable output")
+    parser.add_argument("-d", "--debug",    action = "store_true",  help = "blather about what we're doing")
+
+    parser.add_argument("--rpc-device",            help = "RPC serial device name",
+                        default = os.getenv("CRYPTECH_RPC_CLIENT_SERIAL_DEVICE", "/dev/ttyUSB0"))
+    parser.add_argument("--rpc-socket",            help = "RPC PF_UNIX socket name",
+                        default = os.getenv("CRYPTECH_RPC_CLIENT_SOCKET_NAME", "/tmp/.cryptech_muxd.rpc"))
+
+    parser.add_argument("--cty-device",            help = "CTY serial device name",
+                        default = os.getenv("CRYPTECH_CTY_CLIENT_SERIAL_DEVICE", "/dev/ttyUSB0"))
+    parser.add_argument("--cty-socket",            help = "CTY PF_UNIX socket name",
+                        default = os.getenv("CRYPTECH_CTY_CLIENT_SOCKET_NAME", "/tmp/.cryptech_muxd.cty"))
+
+    args = parser.parse_args()
+
+    futures = []
+
+    rpc_stream = RPCIOStream(device = args.rpc_device, debug = args.debug)
+    rpc_server = RPCServer()
+    rpc_server.set_serial(rpc_stream)
+    rpc_server.listen(args.rpc_socket)
+    futures.append(rpc_stream.rpc_output_loop())
+
+    cty_stream = CTYIOStream(device = args.cty_device, debug = args.debug)
+    cty_server = CTYServer()
+    cty_server.set_serial(cty_stream)
+    cty_server.listen(args.cty_socket)
+    futures.append(cty_stream.cty_output_loop())
+
+    # Might want to use WaitIterator(dict(...)) here so we can
+    # diagnose and restart output loops if they fail.  Worry about
+    # that when we get to automatic device probing.
+
+    yield futures
+
+if __name__ == "__main__":
+    try:
+        tornado.ioloop.IOLoop.current().run_sync(main)
+    except KeyboardInterrupt:
+        pass