SEDbgMuxApp: add a command for terminating connections

This is required for a follow-up change moving DebugMux I/O handling
into separate threads.  We need to be able to stop a thread, which
is executing blocking self.io.read() call.

handlers.py

@@ -0,0 +1,191 @@
+# This file is a part of sedbgmux, an open source DebugMux client.
+# Copyright (c) 2022  Vadim Yanitskiy <axilirator@gmail.com>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import logging as log
+import threading
+import socket
+import queue
+import enum
+import abc
+import sys
+
+from typing import Any, Optional
+from construct import Container
+from proto import DbgMuxFrame
+
+
+class DbgMuxFrameHandler(abc.ABC):
+    ''' Abstract DebugMux frame handler '''
+
+    _tx_queue: Optional[queue.Queue] = None
+
+    def send(self, msg_type: DbgMuxFrame.MsgType, msg: Any = b''):
+        ''' Called by child classes to send some message '''
+        assert self._tx_queue is not None
+        self._tx_queue.put((msg_type, msg))
+
+    @abc.abstractmethod
+    def _handle_frame(self, frame: Container) -> None:
+        ''' Handle the given DebugMux frame '''
+
+
+class ConnState(enum.Enum):
+    ''' Connection state for DbgMuxConnHandler '''
+    NotEstablished      = enum.auto()
+    Establishing        = enum.auto()
+    Established         = enum.auto()
+
+
+class DbgMuxConnHandler(DbgMuxFrameHandler):
+    ''' Abstract DebugMux connection handler '''
+
+    def __init__(self):
+        self.conn_state: ConnState = ConnState.NotEstablished
+        self.ConnRef: int = 0xffff
+        self.DPRef: int = 0xffff
+
+    def establish(self, DPRef: int) -> None:
+        assert self.conn_state == ConnState.NotEstablished
+        log.info("Establishing connection with DPRef=0x%04x", DPRef)
+        self.send(DbgMuxFrame.MsgType.ConnEstablish, dict(DPRef=DPRef))
+        self.conn_state = ConnState.Establishing
+        self.DPRef = DPRef
+
+    def send_data(self, data: bytes) -> None:
+        ''' Called by child classes to send connection data '''
+        assert self.conn_state == ConnState.Established
+        msg = dict(ConnRef=self.ConnRef, Data=data)
+        self.send(DbgMuxFrame.MsgType.ConnData, msg)
+
+    def _match(self, frame: Container,
+                     msg_type: DbgMuxFrame.MsgType,
+                     msg_fields: dict = { }) -> bool:
+        if frame['MsgType'] != msg_type:
+            return False
+        for (key, val) in msg_fields.items():
+            if key not in frame['Msg']:
+                return False
+            if frame['Msg'][key] != val:
+                return False
+        return True
+
+    def _handle_frame(self, frame: Container) -> None:
+        ''' Handle the given DebugMux frame '''
+
+        if self.conn_state == ConnState.Established:
+            fields = dict(ConnRef=self.ConnRef)
+            if self._match(frame, DbgMuxFrame.MsgType.ConnData, fields):
+                self._handle_data(frame['Msg']['Data'])
+                self.send(DbgMuxFrame.MsgType.Ack)
+                raise StopIteration
+            elif self._match(frame, DbgMuxFrame.MsgType.ConnTerminated, fields):
+                log.info('Connection terminated (DPRef=0x%04x, ConnRef=0x%04x)',
+                         self.DPRef, self.ConnRef)
+                self.conn_state = ConnState.NotEstablished
+                self._handle_terminate()
+                self.send(DbgMuxFrame.MsgType.Ack)
+                raise StopIteration
+        elif self.conn_state == ConnState.Establishing:
+            # Match ConnEstablished with our DPRef and any ConnRef
+            if self._match(frame, DbgMuxFrame.MsgType.ConnEstablished, dict(DPRef=self.DPRef)):
+                log.info('Connection established (DPRef=0x%04x, ConnRef=0x%04x)',
+                         self.DPRef, frame['Msg']['ConnRef'])
+                self.conn_state = ConnState.Established
+                self.ConnRef = frame['Msg']['ConnRef']
+                self._handle_establish()
+                self.send(DbgMuxFrame.MsgType.Ack)
+                raise StopIteration
+
+    @abc.abstractmethod
+    def _handle_data(self, data: bytes) -> None:
+        ''' Called on reciept of connection data '''
+
+    @abc.abstractmethod
+    def _handle_establish(self) -> None:
+        ''' Called on connection establishment '''
+
+    @abc.abstractmethod
+    def _handle_terminate(self) -> None:
+        ''' Called on connection termination '''
+
+
+class DbgMuxConnInteractiveTerminal(DbgMuxConnHandler):
+    def __init__(self, *args, **kw):
+        self.attached: bool = False
+        super().__init__(*args)
+
+    def attach(self):
+        self.attached = True
+        while True:
+            try:
+                line = input()
+                if self.conn_state == ConnState.Established:
+                        self.send_data(bytes(line, 'ascii'))
+            except (KeyboardInterrupt, EOFError) as e:
+                break
+        self.attached = False
+
+    def _handle_data(self, data: bytes) -> None:
+        if self.attached:
+            sys.stdout.write(data.decode('ascii'))
+
+    def _handle_establish(self) -> None:
+        pass
+
+    def _handle_terminate(self) -> None:
+        pass
+
+
+class DbgMuxConnUdpBridge(DbgMuxConnHandler):
+    DGRAM_MAX_LEN: int = 1024
+
+    def __init__(self, *args, **kw):
+        super().__init__(*args)
+
+        self.raddr: str = kw.get('raddr', '127.0.0.1')
+        self.rport: int = kw.get('rport', 9999)
+        self.laddr: str = kw.get('laddr', '127.0.0.1')
+        self.lport: int = kw.get('lport', 8888)
+
+        self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._sock.bind((self.laddr, self.lport))
+        self._sock.settimeout(0.5)
+
+        self._thread = threading.Thread(target=self._worker,
+                                        daemon=True)
+        self._shutdown = threading.Event()
+
+    def _worker(self) -> None:
+        while not self._shutdown.is_set():
+            try:
+                (data, addr) = self._sock.recvfrom(self.DGRAM_MAX_LEN)
+                if self.conn_state == ConnState.Established:
+                    self.send_data(data)
+            except TimeoutError:
+                pass
+
+    def _handle_data(self, data: bytes) -> None:
+        self._sock.sendto(data, (self.raddr, self.rport))
+
+    def _handle_establish(self) -> None:
+        self._shutdown.clear()
+        self._thread.start()
+
+    def _handle_terminate(self) -> None:
+        self._shutdown.set()

peer.py

@@ -20,13 +20,76 @@ import logging as log
 import threading
 import queue
 
-from typing import Any
+from typing import Any, Optional
 from construct import Const, Container, Int16ul
 
+from handlers import DbgMuxFrameHandler
 from transport import Transport
 from proto import DbgMuxFrame
 
 
+class DbgMuxFrameDisp:
+    ''' DebugMux frame dispatcher '''
+
+    def __init__(self, rxq: queue.Queue, txq: queue.Queue):
+        self.__hlist = list()
+        self._rx_queue = rxq
+        self._tx_queue = txq
+
+        self._thread = threading.Thread(target=self._worker,
+                                        name='DbgMuxPeer-Disp',
+                                        daemon=True)
+        self._shutdown = threading.Event()
+
+    def start(self) -> None:
+        self._shutdown.clear()
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._shutdown.set()
+        self._thread.join()
+
+    def find_by_name(self, name: str) -> Optional[DbgMuxFrameHandler]:
+        for (hname, hinst) in self.__hlist:
+            if hname == name:
+                return hinst
+        return None
+
+    def register(self, inst: DbgMuxFrameHandler,
+                       name: Optional[str] = None) -> None:
+        if name is not None and self.find_by_name(name):
+            raise FileExistsError
+        self.__hlist.append((name, inst))
+        # Give a handler access to the Tx queue
+        inst._tx_queue = self._tx_queue
+
+    def unregister(self, name: str):
+        inst = self.find_by_name(name)
+        if inst is None:
+            raise FileNotFoundError
+        self.__hlist.remove((name, inst))
+
+    def _worker(self) -> None:
+        while not self._shutdown.is_set():
+            try:
+                frame = self._rx_queue.get(block=True, timeout=0.5)
+                self._dispatch(frame)
+            except queue.Empty:
+                pass
+        log.debug('Thread \'%s\' is shutting down', threading.current_thread().name)
+
+    def _dispatch(self, frame: DbgMuxFrame) -> None:
+        for (hname, hinst) in self.__hlist:
+            try:
+                hinst._handle_frame(frame)
+            except StopIteration:
+                return
+            except Exception as e:
+                log.error("Handler '%s' raised an exception: %s", hname, e)
+                # TODO: remove this handler?
+        log.error("DebugMux message %s was not handled, dropping...", frame['Msg'])
+
+
 class DbgMuxPeer:
     def __init__(self, io: Transport):
         self.tx_count: int = 0
@@ -40,36 +103,46 @@ class DbgMuxPeer:
         self._tx_thread = threading.Thread(target=self._tx_worker,
                                            name='DbgMuxPeer-Tx',
                                            daemon=True)
-
         self._shutdown = threading.Event()
 
         # Internal queues for Rx/Tx frames
         self._rx_queue = queue.Queue()
         self._tx_queue = queue.Queue()
 
+        # Init frame dispatcher
+        self.disp = DbgMuxFrameDisp(self._rx_queue,
+                                    self._tx_queue)
+
     def start(self) -> None:
         self._shutdown.clear()
         self._rx_thread.start()
         self._tx_thread.start()
+        self.disp.start()
 
     def stop(self) -> None:
-        # Wait for all Tx messages to be processed
-        self._tx_queue.join()
         # Set the shutdown event
         self._shutdown.set()
+        # Wait for both threads to terminate
+        self._tx_thread.join()
+        self._rx_thread.join()
+        self.disp.stop()
 
     def _rx_worker(self) -> None:
         while not self._shutdown.is_set():
-            frame = self._recv() # blocking
-            self._rx_queue.put(frame)
-        log.debug('Thread \'%s\' is shutting down', threading.currentThread().name)
+            frame = self._recv() # blocking until timeout
+            if frame is not None:
+                self._rx_queue.put(frame)
+        log.debug('Thread \'%s\' is shutting down', threading.current_thread().name)
 
     def _tx_worker(self) -> None:
         while not self._shutdown.is_set():
-            (msg_type, msg) = self._tx_queue.get(block=True)
-            self._send(msg_type, msg)
-            self._tx_queue.task_done()
-        log.debug('Thread \'%s\' is shutting down', threading.currentThread().name)
+            try:
+                (msg_type, msg) = self._tx_queue.get(block=True, timeout=0.5)
+                self._send(msg_type, msg)
+                self._tx_queue.task_done()
+            except queue.Empty:
+                pass
+        log.debug('Thread \'%s\' is shutting down', threading.current_thread().name)
 
     def send(self, msg_type: DbgMuxFrame.MsgType, msg: Any = b'') -> None:
         ''' Send a single message (non-blocking call) '''
@@ -107,13 +180,15 @@ class DbgMuxPeer:
         if msg_type != DbgMuxFrame.MsgType.Ack:
             self.tx_count += 1
 
-    def recv(self) -> Container:
+    def recv(self, timeout: Optional[float] = None) -> Container:
         ''' Receive a single message (blocking call) '''
-        return self._rx_queue.get(block=True)
+        return self._rx_queue.get(block=True, timeout=timeout)
 
-    def _recv(self) -> Container:
+    def _recv(self) -> Optional[Container]:
         frame: bytes = b''
         frame += self.io.read(2)  # Magic
+        if frame == b'':
+            return None
         Const(b'\x42\x42').parse(frame[:2])
         frame += self.io.read(2)  # Length
         length: int = Int16ul.parse(frame[2:])

sedbgmux.py

@@ -23,11 +23,74 @@ import argparse
 import cmd2
 import enum
 import sys
+import time
 
 from transport import TransportModem
 from proto import DbgMuxFrame
 from peer import DbgMuxPeer
 
+from construct import Container
+from handlers import DbgMuxFrameHandler
+from handlers import DbgMuxConnUdpBridge
+from handlers import DbgMuxConnInteractiveTerminal
+
+
+class CommonFrameHandler(DbgMuxFrameHandler):
+    ''' Handles some common messages '''
+
+    def __init__(self, cmd2):
+        self.cmd2 = cmd2
+
+    def _handle_frame(self, frame: Container) -> None:
+        if frame['MsgType'] == DbgMuxFrame.MsgType.Ident:
+            log.info("Identified target: '%s', IMEI=%s",
+                     frame['Msg']['Ident'][:-15],
+                     frame['Msg']['Ident'][-15:])
+        elif frame['MsgType'] == DbgMuxFrame.MsgType.DPAnnounce:
+            log.info("Data Provider available (DPRef=0x%04x): '%s'",
+                     frame['Msg']['DPRef'], frame['Msg']['Name'])
+        elif frame['MsgType'] == DbgMuxFrame.MsgType.FlowControl:
+            log.warning("Rx FlowControl message, which is not yet supported")
+        elif frame['MsgType'] == DbgMuxFrame.MsgType.ConnEstablished \
+         and frame['Msg']['ConnRef'] == 0xffff:
+            log.warning("Connection establishment failed, "
+                        "no such DPRef=0x%04x?", frame['Msg']['DPRef'])
+        elif frame['MsgType'] == DbgMuxFrame.MsgType.ConnTerminated \
+         and frame['Msg']['ConnRef'] == 0xffff:
+            log.warning("Connection termination failed, "
+                        "no such DPRef=0x%04x?", frame['Msg']['DPRef'])
+        elif frame['MsgType'] == DbgMuxFrame.MsgType.Ack:
+            raise StopIteration
+        else: # To be consumed by other handlers
+            return
+        self.send(DbgMuxFrame.MsgType.Ack)
+        raise StopIteration
+
+
+class PingPongHandler(DbgMuxFrameHandler):
+    ''' Handles DbgMuxFrame.MsgType.Pong '''
+
+    def __init__(self):
+        self.expect_pong: bool = False
+
+    def ping(self, payload: str):
+        log.info('Tx Ping with payload \'%s\'', payload)
+        self.send(DbgMuxFrame.MsgType.Ping, payload)
+        self.expect_pong = True
+        # TODO: start a timer?
+
+    def handle_pong(self, payload: str):
+        if not self.expect_pong:
+            log.warning('Rx unexpected Pong, sending ACK anyway')
+            return
+        log.info('Rx Pong with payload \'%s\'', payload)
+        self.expect_pong = False
+
+    def _handle_frame(self, frame: Container) -> None:
+        if frame['MsgType'] == DbgMuxFrame.MsgType.Pong:
+            self.handle_pong(frame['Msg'])
+            self.send(DbgMuxFrame.MsgType.Ack)
+            raise StopIteration
 
 class SEDbgMuxApp(cmd2.Cmd):
     DESC = 'DebugMux client for [Sony] Ericsson phones and modems'
@@ -48,6 +111,10 @@ class SEDbgMuxApp(cmd2.Cmd):
         self.transport = TransportModem(self.argv)
         self.peer = DbgMuxPeer(self.transport)
 
+        # Register DebugMux frame handlers
+        self.peer.disp.register(CommonFrameHandler(self), 'Common')
+        self.peer.disp.register(PingPongHandler(), 'PingPong')
+
         # Modem connection state
         self.set_connected(False)
 
@@ -88,23 +155,8 @@ class SEDbgMuxApp(cmd2.Cmd):
     def do_enquiry(self, opts) -> None:
         ''' Enquiry target identifier and available Data Providers '''
         self.peer.send(DbgMuxFrame.MsgType.Enquiry)
-        while True:
-            f = self.peer.recv()
-            if f['MsgType'] == DbgMuxFrame.MsgType.Ident:
-                log.info("Identified target: '%s', IMEI=%s",
-                         f['Msg']['Ident'][:-15],
-                         f['Msg']['Ident'][-15:])
-            elif f['MsgType'] == DbgMuxFrame.MsgType.DPAnnounce:
-                log.info("Data Provider available (DPRef=0x%04x): '%s'",
-                         f['Msg']['DPRef'], f['Msg']['Name'])
-
-            # No more data in the buffer
-            # FIXME: layer violation!
-            if self.transport._sl.in_waiting == 0:
-                break
-
-        # ACKnowledge reception of the info
-        self.peer.send(DbgMuxFrame.MsgType.Ack)
+        # The responce to be handled by CommonFrameHandler
+        time.sleep(0.5)
 
     ping_parser = cmd2.Cmd2ArgumentParser()
     ping_parser.add_argument('-p', '--payload',
@@ -115,52 +167,42 @@ class SEDbgMuxApp(cmd2.Cmd):
     @cmd2.with_category(CATEGORY_DBGMUX)
     def do_ping(self, opts) -> None:
         ''' Send a Ping to the target, expect Pong '''
-        log.info('Tx Ping with payload \'%s\'', opts.payload)
-        self.peer.send(DbgMuxFrame.MsgType.Ping, opts.payload)
-
-        f = self.peer.recv()
-        assert f['MsgType'] == DbgMuxFrame.MsgType.Pong
-        log.info('Rx Pong with payload \'%s\'', f['Msg'])
-        self.peer.send(DbgMuxFrame.MsgType.Ack)
+        hinst = self.peer.disp.find_by_name('PingPong')
+        hinst.ping(opts.payload)
 
     establish_parser = cmd2.Cmd2ArgumentParser()
     establish_parser.add_argument('DPRef',
                                   type=lambda v: int(v, 16),
                                   help='DPRef of a Data Provider in hex')
+    establish_parser.add_argument('mode',
+                                  choices=['interactive', 'udp-bridge'],
+                                  help='Connection mode')
 
     @cmd2.with_argparser(establish_parser)
     @cmd2.with_category(CATEGORY_DBGMUX)
     def do_establish(self, opts) -> None:
         ''' Establish connection with a Data Provider '''
-        log.info("Establishing connection with DPRef=0x%04x", opts.DPRef)
-        self.peer.send(DbgMuxFrame.MsgType.ConnEstablish,
-                       dict(DPRef=opts.DPRef))
-
-        f = self.peer.recv()
-        assert f['MsgType'] == DbgMuxFrame.MsgType.ConnEstablished
-        if f['Msg']['ConnRef'] == 0xffff:
-            log.warning("Connection failed: unknown DPRef=0x%04x?", opts.DPRef)
-            self.peer.send(DbgMuxFrame.MsgType.Ack)
-            return
-
-        log.info("Connection established (ConnRef=0x%04x)",
-                 f['Msg']['ConnRef'])
-
-        # Read the messages
-        while True:
-            f = self.peer.recv()
-
-            if f['MsgType'] != DbgMuxFrame.MsgType.ConnData:
-                log.warning('Unexpected frame: %s', f)
-                self.peer.send(DbgMuxFrame.MsgType.Ack)
-                continue
-            try:  # FIXME: there can be binary data
-                self.stdout.write(f['Msg']['Data'].decode())
-            except:  # ... ignore it for now
-                continue
+        if opts.mode == 'interactive':
+            ch = DbgMuxConnInteractiveTerminal()
+            self.peer.disp.register(ch)
+            ch.establish(opts.DPRef)
+            ch.attach()
+        elif opts.mode == 'udp-bridge':
+            ch = DbgMuxConnUdpBridge()
+            self.peer.disp.register(ch)
+            ch.establish(opts.DPRef)
+
+    terminate_parser = cmd2.Cmd2ArgumentParser()
+    terminate_parser.add_argument('ConnRef',
+                                  type=lambda v: int(v, 16),
+                                  help='ConnRef in hex')
 
-            # ACKnowledge reception of a frame
-            self.peer.send(DbgMuxFrame.MsgType.Ack)
+    @cmd2.with_argparser(terminate_parser)
+    @cmd2.with_category(CATEGORY_DBGMUX)
+    def do_terminate(self, opts) -> None:
+        ''' Terminate connection with a Data Provider '''
+        msg = dict(ConnRef=opts.ConnRef)
+        self.peer.send(DbgMuxFrame.MsgType.ConnTerminate, msg)
 
 
 ap = argparse.ArgumentParser(prog='sedbgmux', description=SEDbgMuxApp.DESC,
@@ -171,8 +213,8 @@ group.add_argument('-p', '--serial-port', metavar='PORT', type=str, default='/de
                    help='Serial port path (default %(default)s)')
 group.add_argument('--serial-baudrate', metavar='BAUDRATE', type=int, default=115200,
                    help='Serial port speed (default %(default)s)')
-group.add_argument('--serial-timeout', metavar='TIMEOUT', type=int,
-                   help='Serial port timeout')
+group.add_argument('--serial-timeout', metavar='TIMEOUT', type=float, default=0.5,
+                   help='Serial port read timeout (default %(default)s)')
 
 log.basicConfig(
     format='[%(levelname)s] %(filename)s:%(lineno)d %(message)s', level=log.INFO)

transport.py

@@ -30,15 +30,19 @@ class TransportIOError(Exception):
 class Transport(abc.ABC):
     ''' Abstract transport layer for DebugMux '''
 
+    @abc.abstractmethod
     def connect(self, opts: dict) -> None:
         ''' Establish connection to the target and enter DebugMux mode '''
 
+    @abc.abstractmethod
     def disconnect(self) -> None:
         ''' Escape DebugMux mode and terminate connection with the target '''
 
+    @abc.abstractmethod
     def write(self, data: bytes) -> int:
         ''' Write the given data bytes '''
 
+    @abc.abstractmethod
     def read(self, length: int = 0) -> bytes:
         ''' Read the given number of bytes '''