1#! /usr/bin/env python
2#
3# (C) 2001-2015 Chris Liechti <[email protected]>
4#
5# SPDX-License-Identifier:    BSD-3-Clause
6"""\
7Multi-port serial<->TCP/IP forwarder.
8- RFC 2217
9- check existence of serial port periodically
10- start/stop forwarders
11- each forwarder creates a server socket and opens the serial port
12- serial ports are opened only once. network connect/disconnect
13  does not influence serial port
14- only one client per connection
15"""
16import os
17import select
18import socket
19import sys
20import time
21import traceback
22
23import serial
24import serial.rfc2217
25import serial.tools.list_ports
26
27import dbus
28
29# Try to import the avahi service definitions properly. If the avahi module is
30# not available, fall back to a hard-coded solution that hopefully still works.
31try:
32    import avahi
33except ImportError:
34    class avahi:
35        DBUS_NAME = "org.freedesktop.Avahi"
36        DBUS_PATH_SERVER = "/"
37        DBUS_INTERFACE_SERVER = "org.freedesktop.Avahi.Server"
38        DBUS_INTERFACE_ENTRY_GROUP = DBUS_NAME + ".EntryGroup"
39        IF_UNSPEC = -1
40        PROTO_UNSPEC, PROTO_INET, PROTO_INET6 = -1, 0, 1
41
42
43class ZeroconfService:
44    """\
45    A simple class to publish a network service with zeroconf using avahi.
46    """
47
48    def __init__(self, name, port, stype="_http._tcp",
49                 domain="", host="", text=""):
50        self.name = name
51        self.stype = stype
52        self.domain = domain
53        self.host = host
54        self.port = port
55        self.text = text
56        self.group = None
57
58    def publish(self):
59        bus = dbus.SystemBus()
60        server = dbus.Interface(
61            bus.get_object(
62                avahi.DBUS_NAME,
63                avahi.DBUS_PATH_SERVER
64            ),
65            avahi.DBUS_INTERFACE_SERVER
66        )
67
68        g = dbus.Interface(
69            bus.get_object(
70                avahi.DBUS_NAME,
71                server.EntryGroupNew()
72            ),
73            avahi.DBUS_INTERFACE_ENTRY_GROUP
74        )
75
76        g.AddService(avahi.IF_UNSPEC, avahi.PROTO_UNSPEC, dbus.UInt32(0),
77                     self.name, self.stype, self.domain, self.host,
78                     dbus.UInt16(self.port), self.text)
79
80        g.Commit()
81        self.group = g
82
83    def unpublish(self):
84        if self.group is not None:
85            self.group.Reset()
86            self.group = None
87
88    def __str__(self):
89        return "{!r} @ {}:{} ({})".format(self.name, self.host, self.port, self.stype)
90
91
92class Forwarder(ZeroconfService):
93    """\
94    Single port serial<->TCP/IP forarder that depends on an external select
95    loop.
96    - Buffers for serial -> network and network -> serial
97    - RFC 2217 state
98    - Zeroconf publish/unpublish on open/close.
99    """
100
101    def __init__(self, device, name, network_port, on_close=None, log=None):
102        ZeroconfService.__init__(self, name, network_port, stype='_serial_port._tcp')
103        self.alive = False
104        self.network_port = network_port
105        self.on_close = on_close
106        self.log = log
107        self.device = device
108        self.serial = serial.Serial()
109        self.serial.port = device
110        self.serial.baudrate = 115200
111        self.serial.timeout = 0
112        self.socket = None
113        self.server_socket = None
114        self.rfc2217 = None  # instantiate later, when connecting
115
116    def __del__(self):
117        try:
118            if self.alive:
119                self.close()
120        except:
121            pass  # XXX errors on shutdown
122
123    def open(self):
124        """open serial port, start network server and publish service"""
125        self.buffer_net2ser = bytearray()
126        self.buffer_ser2net = bytearray()
127
128        # open serial port
129        try:
130            self.serial.rts = False
131            self.serial.open()
132        except Exception as msg:
133            self.handle_serial_error(msg)
134
135        self.serial_settings_backup = self.serial.get_settings()
136
137        # start the socket server
138        # XXX add IPv6 support: use getaddrinfo for socket options, bind to multiple sockets?
139        #       info_list = socket.getaddrinfo(None, port, 0, socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
140        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
141        self.server_socket.setsockopt(
142            socket.SOL_SOCKET,
143            socket.SO_REUSEADDR,
144            self.server_socket.getsockopt(
145                socket.SOL_SOCKET,
146                socket.SO_REUSEADDR
147            ) | 1
148        )
149        self.server_socket.setblocking(0)
150        try:
151            self.server_socket.bind(('', self.network_port))
152            self.server_socket.listen(1)
153        except socket.error as msg:
154            self.handle_server_error()
155            #~ raise
156        if self.log is not None:
157            self.log.info("{}: Waiting for connection on {}...".format(self.device, self.network_port))
158
159        # zeroconfig
160        self.publish()
161
162        # now we are ready
163        self.alive = True
164
165    def close(self):
166        """Close all resources and unpublish service"""
167        if self.log is not None:
168            self.log.info("{}: closing...".format(self.device))
169        self.alive = False
170        self.unpublish()
171        if self.server_socket:
172            self.server_socket.close()
173        if self.socket:
174            self.handle_disconnect()
175        self.serial.close()
176        if self.on_close is not None:
177            # ensure it is only called once
178            callback = self.on_close
179            self.on_close = None
180            callback(self)
181
182    def write(self, data):
183        """the write method is used by serial.rfc2217.PortManager. it has to
184        write to the network."""
185        self.buffer_ser2net += data
186
187    def update_select_maps(self, read_map, write_map, error_map):
188        """Update dictionaries for select call. insert fd->callback mapping"""
189        if self.alive:
190            # always handle serial port reads
191            read_map[self.serial] = self.handle_serial_read
192            error_map[self.serial] = self.handle_serial_error
193            # handle serial port writes if buffer is not empty
194            if self.buffer_net2ser:
195                write_map[self.serial] = self.handle_serial_write
196            # handle network
197            if self.socket is not None:
198                # handle socket if connected
199                # only read from network if the internal buffer is not
200                # already filled. the TCP flow control will hold back data
201                if len(self.buffer_net2ser) < 2048:
202                    read_map[self.socket] = self.handle_socket_read
203                # only check for write readiness when there is data
204                if self.buffer_ser2net:
205                    write_map[self.socket] = self.handle_socket_write
206                error_map[self.socket] = self.handle_socket_error
207            else:
208                # no connection, ensure clear buffer
209                self.buffer_ser2net = bytearray()
210            # check the server socket
211            read_map[self.server_socket] = self.handle_connect
212            error_map[self.server_socket] = self.handle_server_error
213
214    def handle_serial_read(self):
215        """Reading from serial port"""
216        try:
217            data = os.read(self.serial.fileno(), 1024)
218            if data:
219                # store data in buffer if there is a client connected
220                if self.socket is not None:
221                    # escape outgoing data when needed (Telnet IAC (0xff) character)
222                    if self.rfc2217:
223                        data = serial.to_bytes(self.rfc2217.escape(data))
224                    self.buffer_ser2net.extend(data)
225            else:
226                self.handle_serial_error()
227        except Exception as msg:
228            self.handle_serial_error(msg)
229
230    def handle_serial_write(self):
231        """Writing to serial port"""
232        try:
233            # write a chunk
234            n = os.write(self.serial.fileno(), bytes(self.buffer_net2ser))
235            # and see how large that chunk was, remove that from buffer
236            self.buffer_net2ser = self.buffer_net2ser[n:]
237        except Exception as msg:
238            self.handle_serial_error(msg)
239
240    def handle_serial_error(self, error=None):
241        """Serial port error"""
242        # terminate connection
243        self.close()
244
245    def handle_socket_read(self):
246        """Read from socket"""
247        try:
248            # read a chunk from the serial port
249            data = self.socket.recv(1024)
250            if data:
251                # Process RFC 2217 stuff when enabled
252                if self.rfc2217:
253                    data = b''.join(self.rfc2217.filter(data))
254                # add data to buffer
255                self.buffer_net2ser.extend(data)
256            else:
257                # empty read indicates disconnection
258                self.handle_disconnect()
259        except socket.error:
260            if self.log is not None:
261                self.log.exception("{}: error reading...".format(self.device))
262            self.handle_socket_error()
263
264    def handle_socket_write(self):
265        """Write to socket"""
266        try:
267            # write a chunk
268            count = self.socket.send(bytes(self.buffer_ser2net))
269            # and remove the sent data from the buffer
270            self.buffer_ser2net = self.buffer_ser2net[count:]
271        except socket.error:
272            if self.log is not None:
273                self.log.exception("{}: error writing...".format(self.device))
274            self.handle_socket_error()
275
276    def handle_socket_error(self):
277        """Socket connection fails"""
278        self.handle_disconnect()
279
280    def handle_connect(self):
281        """Server socket gets a connection"""
282        # accept a connection in any case, close connection
283        # below if already busy
284        connection, addr = self.server_socket.accept()
285        if self.socket is None:
286            self.socket = connection
287            # More quickly detect bad clients who quit without closing the
288            # connection: After 1 second of idle, start sending TCP keep-alive
289            # packets every 1 second. If 3 consecutive keep-alive packets
290            # fail, assume the client is gone and close the connection.
291            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
292            self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 1)
293            self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 1)
294            self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
295            self.socket.setblocking(0)
296            self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
297            if self.log is not None:
298                self.log.warning('{}: Connected by {}:{}'.format(self.device, addr[0], addr[1]))
299            self.serial.rts = True
300            self.serial.dtr = True
301            if self.log is not None:
302                self.rfc2217 = serial.rfc2217.PortManager(self.serial, self, logger=log.getChild(self.device))
303            else:
304                self.rfc2217 = serial.rfc2217.PortManager(self.serial, self)
305        else:
306            # reject connection if there is already one
307            connection.close()
308            if self.log is not None:
309                self.log.warning('{}: Rejecting connect from {}:{}'.format(self.device, addr[0], addr[1]))
310
311    def handle_server_error(self):
312        """Socket server fails"""
313        self.close()
314
315    def handle_disconnect(self):
316        """Socket gets disconnected"""
317        # signal disconnected terminal with control lines
318        try:
319            self.serial.rts = False
320            self.serial.dtr = False
321        finally:
322            # restore original port configuration in case it was changed
323            self.serial.apply_settings(self.serial_settings_backup)
324            # stop RFC 2217 state machine
325            self.rfc2217 = None
326            # clear send buffer
327            self.buffer_ser2net = bytearray()
328            # close network connection
329            if self.socket is not None:
330                self.socket.close()
331                self.socket = None
332                if self.log is not None:
333                    self.log.warning('{}: Disconnected'.format(self.device))
334
335
336def test():
337    service = ZeroconfService(name="TestService", port=3000)
338    service.publish()
339    input("Press the ENTER key to unpublish the service ")
340    service.unpublish()
341
342
343if __name__ == '__main__':  # noqa
344    import logging
345    import argparse
346
347    VERBOSTIY = [
348        logging.ERROR,      # 0
349        logging.WARNING,    # 1 (default)
350        logging.INFO,       # 2
351        logging.DEBUG,      # 3
352    ]
353
354    parser = argparse.ArgumentParser(
355        usage="""\
356%(prog)s [options]
357
358Announce the existence of devices using zeroconf and provide
359a TCP/IP <-> serial port gateway (implements RFC 2217).
360
361If running as daemon, write to syslog. Otherwise write to stdout.
362""",
363        epilog="""\
364NOTE: no security measures are implemented. Anyone can remotely connect
365to this service over the network.
366
367Only one connection at once, per port, is supported. When the connection is
368terminated, it waits for the next connect.
369""")
370
371    group = parser.add_argument_group("serial port settings")
372
373    group.add_argument(
374        "--ports-regex",
375        help="specify a regex to search against the serial devices and their descriptions (default: %(default)s)",
376        default='/dev/ttyUSB[0-9]+',
377        metavar="REGEX")
378
379    group = parser.add_argument_group("network settings")
380
381    group.add_argument(
382        "--tcp-port",
383        dest="base_port",
384        help="specify lowest TCP port number (default: %(default)s)",
385        default=7000,
386        type=int,
387        metavar="PORT")
388
389    group = parser.add_argument_group("daemon")
390
391    group.add_argument(
392        "-d", "--daemon",
393        dest="daemonize",
394        action="store_true",
395        help="start as daemon",
396        default=False)
397
398    group.add_argument(
399        "--pidfile",
400        help="specify a name for the PID file",
401        default=None,
402        metavar="FILE")
403
404    group = parser.add_argument_group("diagnostics")
405
406    group.add_argument(
407        "-o", "--logfile",
408        help="write messages file instead of stdout",
409        default=None,
410        metavar="FILE")
411
412    group.add_argument(
413        "-q", "--quiet",
414        dest="verbosity",
415        action="store_const",
416        const=0,
417        help="suppress most diagnostic messages",
418        default=1)
419
420    group.add_argument(
421        "-v", "--verbose",
422        dest="verbosity",
423        action="count",
424        help="increase diagnostic messages")
425
426    args = parser.parse_args()
427
428    # set up logging
429    logging.basicConfig(level=VERBOSTIY[min(args.verbosity, len(VERBOSTIY) - 1)])
430    log = logging.getLogger('port_publisher')
431
432    # redirect output if specified
433    if args.logfile is not None:
434        class WriteFlushed:
435            def __init__(self, fileobj):
436                self.fileobj = fileobj
437
438            def write(self, s):
439                self.fileobj.write(s)
440                self.fileobj.flush()
441
442            def close(self):
443                self.fileobj.close()
444        sys.stdout = sys.stderr = WriteFlushed(open(args.logfile, 'a'))
445        # atexit.register(lambda: sys.stdout.close())
446
447    if args.daemonize:
448        # if running as daemon is requested, do the fork magic
449        # args.quiet = True
450        # do the UNIX double-fork magic, see Stevens' "Advanced
451        # Programming in the UNIX Environment" for details (ISBN 0201563177)
452        try:
453            pid = os.fork()
454            if pid > 0:
455                # exit first parent
456                sys.exit(0)
457        except OSError as e:
458            log.critical("fork #1 failed: {} ({})\n".format(e.errno, e.strerror))
459            sys.exit(1)
460
461        # decouple from parent environment
462        os.chdir("/")   # don't prevent unmounting....
463        os.setsid()
464        os.umask(0)
465
466        # do second fork
467        try:
468            pid = os.fork()
469            if pid > 0:
470                # exit from second parent, save eventual PID before
471                if args.pidfile is not None:
472                    open(args.pidfile, 'w').write("{}".format(pid))
473                sys.exit(0)
474        except OSError as e:
475            log.critical("fork #2 failed: {} ({})\n".format(e.errno, e.strerror))
476            sys.exit(1)
477
478        if args.logfile is None:
479            import syslog
480            syslog.openlog("serial port publisher")
481
482            # redirect output to syslog
483            class WriteToSysLog:
484                def __init__(self):
485                    self.buffer = ''
486
487                def write(self, s):
488                    self.buffer += s
489                    if '\n' in self.buffer:
490                        output, self.buffer = self.buffer.split('\n', 1)
491                        syslog.syslog(output)
492
493                def flush(self):
494                    syslog.syslog(self.buffer)
495                    self.buffer = ''
496
497                def close(self):
498                    self.flush()
499            sys.stdout = sys.stderr = WriteToSysLog()
500
501            # ensure the that the daemon runs a normal user, if run as root
502        # if os.getuid() == 0:
503            #    name, passwd, uid, gid, desc, home, shell = pwd.getpwnam('someuser')
504            #    os.setgid(gid)     # set group first
505            #    os.setuid(uid)     # set user
506
507    # keep the published stuff in a dictionary
508    published = {}
509    # get a nice hostname
510    hostname = socket.gethostname()
511
512    def unpublish(forwarder):
513        """when forwarders die, we need to unregister them"""
514        try:
515            del published[forwarder.device]
516        except KeyError:
517            pass
518        else:
519            log.info("unpublish: {}".format(forwarder))
520
521    alive = True
522    next_check = 0
523    # main loop
524    while alive:
525        try:
526            # if it is time, check for serial port devices
527            now = time.time()
528            if now > next_check:
529                next_check = now + 5
530                connected = [d for d, p, i in serial.tools.list_ports.grep(args.ports_regex)]
531                # Handle devices that are published, but no longer connected
532                for device in set(published).difference(connected):
533                    log.info("unpublish: {}".format(published[device]))
534                    unpublish(published[device])
535                # Handle devices that are connected but not yet published
536                for device in sorted(set(connected).difference(published)):
537                    # Find the first available port, starting from specified number
538                    port = args.base_port
539                    ports_in_use = [f.network_port for f in published.values()]
540                    while port in ports_in_use:
541                        port += 1
542                    published[device] = Forwarder(
543                        device,
544                        "{} on {}".format(device, hostname),
545                        port,
546                        on_close=unpublish,
547                        log=log)
548                    log.warning("publish: {}".format(published[device]))
549                    published[device].open()
550
551            # select_start = time.time()
552            read_map = {}
553            write_map = {}
554            error_map = {}
555            for publisher in published.values():
556                publisher.update_select_maps(read_map, write_map, error_map)
557            readers, writers, errors = select.select(
558                read_map.keys(),
559                write_map.keys(),
560                error_map.keys(),
561                5)
562            # select_end = time.time()
563            # print "select used %.3f s" % (select_end - select_start)
564            for reader in readers:
565                read_map[reader]()
566            for writer in writers:
567                write_map[writer]()
568            for error in errors:
569                error_map[error]()
570            # print "operation used %.3f s" % (time.time() - select_end)
571        except KeyboardInterrupt:
572            alive = False
573            sys.stdout.write('\n')
574        except SystemExit:
575            raise
576        except:
577            #~ raise
578            traceback.print_exc()
579