server: service discovery over UDP.

implement RFC-005
- server broadcasts once on startup and answers to broadcasts
- small tool for listening on the port and sending broadcasts

Change-Id: I02d1184d6be62bef6f964eb9d238220aef062e94
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/34851
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Georg Brandl <g.brandl@fz-juelich.de>

bin/frappy-scan

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+# *****************************************************************************
+#
+# 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 2 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, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+#
+# Module authors:
+#   Alexander Zaft <a.zaft@fz-juelich.de>
+#
+# *****************************************************************************
+
+"""SEC node autodiscovery tool."""
+
+import argparse
+import json
+import os
+import select
+import socket
+import sys
+from collections import namedtuple
+from time import time as currenttime
+
+UDP_PORT = 10767
+
+Answer = namedtuple('Answer',
+                    'address, port, equipment_id, firmware, description')
+
+
+def decode(msg, addr):
+    msg = msg.decode('utf-8')
+    try:
+        data = json.loads(msg)
+    except Exception:
+        return None
+    if not isinstance(data, dict):
+        return None
+    if data.get('SECoP') != 'node':
+        return None
+    try:
+        eq_id = data['equipment_id']
+        fw = data['firmware']
+        desc = data['description']
+        port = data['port']
+    except KeyError:
+        return None
+    addr, _scanport = addr
+    return Answer(addr, port, eq_id, fw, desc)
+
+
+def print_answer(answer, *, short=False):
+    if short:
+        # NOTE: keep this easily parseable!
+        print(f'{answer.equipment_id} {answer.address}:{answer.port}')
+        return
+    print(f'Found {answer.equipment_id} at {answer.address}:')
+    print(f'  Port: {answer.port}')
+    print(f'  Firmware: {answer.firmware}')
+    desc = answer.description.replace('\n', '\n    ')
+    print(f'  Node description: {desc}')
+    print()
+
+
+def scan(max_wait=1.0):
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+    # send a general broadcast
+    try:
+        s.sendto(json.dumps(dict(SECoP='discover')).encode('utf-8'),
+                 ('255.255.255.255', UDP_PORT))
+    except OSError as e:
+        print('could not send the broadcast:', e)
+        # we still keep listening for self-announcements
+    start = currenttime()
+    seen = set()
+    while currenttime() < start + max_wait:
+        res = select.select([s], [], [], 0.1)
+        if res[0]:
+            try:
+                msg, addr = s.recvfrom(1024)
+            except socket.error:  # pragma: no cover
+                continue
+            answer = decode(msg, addr)
+            if answer is None:
+                continue
+            if (answer.address, answer.equipment_id, answer.port) in seen:
+                continue
+            seen.add((answer.address, answer.equipment_id, answer.port))
+            yield answer
+
+
+def listen(*, short=False):
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    if os.name == 'nt':
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    else:
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+    s.bind(('0.0.0.0', UDP_PORT))
+    while True:
+        try:
+            msg, addr = s.recvfrom(1024)
+        except KeyboardInterrupt:
+            break
+        answer = decode(msg, addr)
+        if answer:
+            print_answer(answer, short=short)
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-l', '--listen', action='store_true',
+                        help='Print short info. '
+                        'Keep listening after the broadcast.')
+    args = parser.parse_args(sys.argv[1:])
+    for answer in scan():
+        print_answer(answer, short=args.listen)
+    if args.listen:
+        listen(short=args.listen)

frappy/protocol/discovery.py

@@ -0,0 +1,108 @@
+# *****************************************************************************
+#
+# 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 2 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, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+#
+# Module authors:
+#   Alexander Zaft <a.zaft@fz-juelich.de>
+#
+# *****************************************************************************
+"""Discovery via UDP broadcasts."""
+
+import os
+import json
+import socket
+
+from frappy.lib import closeSocket
+from frappy.protocol.interface.tcp import format_address
+from frappy.version import get_version
+
+UDP_PORT = 10767
+MAX_MESSAGE_LEN = 508
+
+
+class UDPListener:
+    def __init__(self, equipment_id, description, ifaces, logger, *,
+                 startup_broadcast=True):
+        self.equipment_id = equipment_id
+        self.log = logger
+        self.description = description or ''
+        self.firmware = 'FRAPPY ' + get_version()
+        self.ports = [int(iface.split('://')[1])
+                      for iface in ifaces if iface.startswith('tcp')]
+        self.running = False
+        self.is_enabled = True
+        self.startup_broadcast = startup_broadcast
+
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        if os.name == 'nt':
+            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        else:
+            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+        if startup_broadcast:
+            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+        self.sock.bind(('0.0.0.0', UDP_PORT))
+
+        available = MAX_MESSAGE_LEN - len(self._getMessage(2**16-1))
+        if available < 0:
+            desc_length = len(self.description.encode('utf-8'))
+            if available + desc_length < 0:
+                self.log.warn('Equipment id and firmware name exceed 430 byte '
+                              'limit, not answering to udp discovery')
+                self.is_enabled = False
+            else:
+                self.log.debug('truncating description for udp discovery')
+                # with errors='ignore', cutting insite a utf-8 glyph will not
+                # report an error but remove the rest of the glyph from the
+                # output.
+                self.description = self.description \
+                                       .encode('utf-8')[:available] \
+                                       .decode('utf-8', errors='ignore')
+
+    def _getMessage(self, port):
+        return json.dumps({
+            'SECoP': 'node',
+            'port': port,
+            'equipment_id': self.equipment_id,
+            'firmware': self.firmware,
+            'description': self.description,
+        }, ensure_ascii=False, separators=(',', ':')).encode('utf-8')
+
+    def run(self):
+        if self.startup_broadcast:
+            self.log.debug('Sending startup UDP broadcast.')
+            for port in self.ports:
+                self.sock.sendto(self._getMessage(port),
+                                 ('255.255.255.255', UDP_PORT))
+        self.running = True
+        while self.running and self.is_enabled:
+            try:
+                msg, addr = self.sock.recvfrom(1024)
+            except socket.error:
+                return
+            try:
+                request = json.loads(msg.decode('utf-8'))
+            except json.JSONDecodeError:
+                continue
+            if 'SECoP' not in request or request['SECoP'] != 'discover':
+                continue
+            self.log.debug('Answering UDP broadcast from: %s',
+                           format_address(addr))
+            for port in self.ports:
+                self.sock.sendto(self._getMessage(port), addr)
+
+    def shutdown(self):
+        self.log.debug('shut down of discovery listener')
+        self.running = False
+        closeSocket(self.sock)

frappy/secnode.py

@@ -54,8 +54,17 @@ class SecNode:
         self.name = name
 
     def add_secnode_property(self, prop, value):
+        """Add SECNode property. If starting with an underscore, it is exported
+        in the description."""
         self.nodeprops[prop] = value
 
+    def get_secnode_property(self, prop):
+        """Get SECNode property.
+
+        Returns None if not present.
+        """
+        return self.nodeprops.get(prop)
+
     def get_module(self, modulename):
         """ Returns a fully initialized module. Or None, if something went
         wrong during instatiating/initializing the module."""

frappy/server.py

@@ -37,6 +37,7 @@ from frappy.lib.multievent import MultiEvent
 from frappy.logging import init_remote_logging
 from frappy.params import PREDEFINED_ACCESSIBLES
 from frappy.secnode import SecNode
+from frappy.protocol.discovery import UDPListener
 
 try:
     from daemon import DaemonContext
@@ -117,6 +118,8 @@ class Server:
         signal.signal(signal.SIGINT, self.signal_handler)
         signal.signal(signal.SIGTERM, self.signal_handler)
 
+        self.discovery = None
+
     def signal_handler(self, num, frame):
         if hasattr(self, 'interfaces') and self.interfaces:
             self.shutdown()
@@ -165,6 +168,7 @@ class Server:
                 print(formatException(verbose=True))
                 raise
 
+            # client interfaces
             self.interfaces = {}
             iface_threads = []
             # default_timeout 12 sec: TCPServer might need up to 10 sec to wait for Address no longer in use
@@ -197,6 +201,16 @@ class Server:
             self.secnode.add_secnode_property('_interfaces', list(self.interfaces))
             self.log.info('startup done with interface(s) %s',
                           ', '.join(self.interfaces))
+
+            # start discovery interface when we know where we listen
+            self.discovery = UDPListener(
+                self.secnode.equipment_id,
+                self.secnode.get_secnode_property('description'),
+                list(self.interfaces),
+                self.log.getChild('discovery')
+            )
+            mkthread(self.discovery.run)
+
             if systemd:
                 systemd.daemon.notify("READY=1\nSTATUS=accepting requests")
 
@@ -237,6 +251,8 @@ class Server:
 
     def shutdown(self):
         self._restart = False
+        if self.discovery:
+            self.discovery.shutdown()
         for iface in self.interfaces.values():
             iface.shutdown()
 

test/test_discovery.py

@@ -0,0 +1,62 @@
+# *****************************************************************************
+#
+# 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 2 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, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+#
+# Module authors:
+#   Alexander Zaft <a.zaft@fz-juelich.de>
+#
+# *****************************************************************************
+"""Test discovery messages"""
+
+from test.test_modules import LoggerStub
+from frappy.protocol.discovery import UDPListener, MAX_MESSAGE_LEN
+from frappy.version import get_version
+
+
+def test_empty():
+    logger = LoggerStub()
+    udp = UDPListener('', '', ['tcp://0'], logger)
+    udp.firmware = ''
+    # 78 is the maximum overhead
+    assert 78 == len(udp._getMessage(2**16-1))
+
+
+def test_basic():
+    logger = LoggerStub()
+    udp = UDPListener('eq', 'desc', ['tcp://1234'], logger)
+    assert udp.description == 'desc'
+    assert udp.equipment_id == 'eq'
+    assert udp.ports == [1234]
+    assert udp.firmware == 'FRAPPY ' + get_version()
+
+
+def test_ascii_truncation():
+    logger = LoggerStub()
+    desc = 'a' * MAX_MESSAGE_LEN
+    udp = UDPListener('eq', desc, ['tcp://1234'], logger)
+    assert MAX_MESSAGE_LEN == len(udp._getMessage(65535))
+    fw = len(('FRAPPY ' + get_version()).encode('utf-8'))
+    expected_length = 430 - fw - 2
+    assert expected_length == len(udp.description)
+
+
+def test_unicode_truncation():
+    logger = LoggerStub()
+    desc = '\U0001f604' * 400
+    udp = UDPListener('eq', desc, ['tcp://1234'], logger)
+    fw = len(('FRAPPY ' + get_version()).encode('utf-8'))
+    # 4 bytes per symbol, rounded down for the potential cut
+    expected_length = (430 - fw - 2) // 4
+    assert expected_length == len(udp.description)