Merge ~mpontillo/maas:beaconing-service into maas:master

[Mike Pontillo]

Add initial beaconing service and protocol.

 * Add ProtocolForObserveBeacons to read input from each beacon
   monitoring process.
 * Add BeaconingSocketProtocol to receive input from either the
   ProtocolForObserveBeacons or the socket layer. (The socket
   layer path is used for unit tests, and in a future test command.)
 * Add a way to age out entries in an OrderedDict based upon UUID keys.
   (The UUIDs must have a valid 'time' field.)
 * Add initial transmit and receive path for beacons.
 * If solicitation packets are received, respond to them with
   advertisements.

[Mike Pontillo (mpontillo)]

I tried to make this branch as small as possible so that it can be reviewed and landed; I've also developed a command to test beaconing, but that is not part of this proposal. That branch can be found here:

https://code.launchpad.net/~mpontillo/maas/+git/maas/+ref/send-beacons-command

For now, MAAS listens for beacons on any interface that is configured for network monitoring (that is, passive ARP observation). This approach may change in the future.

MAAS does not take any action on the received beacons except to (1) reply to any solicitation beacons with advertisement beacons and (2) keep state about beacons received within +/- two minutes of the current system time. This is enough to allow testing, sets the stage for MAAS to make some assumptions about fabric connectivity. But those assumptions are not yet acted upon in this branch.

[Данило Шеган (danilo)]

Looks good, sorry for the delay (it slipped on Friday due to my battles with the lander). A few minor nits inline.

review: Approve

[Mike Pontillo (mpontillo)]

Thanks for the review! Some replies below.

af83e74... by Mike Pontillo on 2017-07-03

Review fixes.

diff --git a/src/maastesting/factory.py b/src/maastesting/factory.py
index b83cffb..d12ac17 100644
--- a/src/maastesting/factory.py
+++ b/src/maastesting/factory.py
@@ -30,7 +30,10 @@ from unittest import mock
 import urllib.error
 import urllib.parse
 import urllib.request
-from uuid import uuid1
+from uuid import (
+    UUID,
+    uuid1,
+)
 
 from distro_info import UbuntuDistroInfo
 from maastesting.fixtures import TempDirectory
@@ -255,6 +258,23 @@ class Factory:
     def make_UUID(self):
         return str(uuid1())
 
+    def make_UUID_with_timestamp(self, timestamp, clock_seq=None, node=None):
+        if node is None:
+            node = random.getrandbits(48) | 0x010000000000
+        if clock_seq is None:
+            clock_seq = random.getrandbits(14)
+        timestamp = int(timestamp * 1e9 / 100) + 0x01b21dd213814000
+        time_low = timestamp & 0xffffffff
+        time_mid = (timestamp >> 32) & 0xffff
+        time_hi_version = (timestamp >> 48) & 0x0fff
+        clock_seq_low = clock_seq & 0xff
+        clock_seq_hi_variant = (clock_seq >> 8) & 0x3f
+        fields = (
+            time_low, time_mid, time_hi_version, clock_seq_hi_variant,
+            clock_seq_low, node
+        )
+        return str(UUID(fields=fields, version=1))
+
     def _make_random_network(
             self, slash=None, but_not=EMPTY_SET, disjoint_from=None,
             random_address_factory=None):
diff --git a/src/provisioningserver/utils/beaconing.py b/src/provisioningserver/utils/beaconing.py
index e8c2384..5b5d286 100644
--- a/src/provisioningserver/utils/beaconing.py
+++ b/src/provisioningserver/utils/beaconing.py
@@ -19,13 +19,16 @@ from gzip import (
     decompress,
 )
 import json
+import math
 import os
 import stat
 import struct
 import subprocess
 import sys
 from textwrap import dedent
+import time
 import uuid
+from uuid import UUID
 
 from bson import BSON
 from bson.errors import BSONError
@@ -48,6 +51,9 @@ from provisioningserver.utils.tcpip import (
 
 
 BEACON_PORT = 5240
+BEACON_IPV4_MULTICAST = "224.0.0.118"
+BEACON_IPV6_MULTICAST = "ff02::15a"
+
 
 BEACON_TYPES = {
     "solicitation": 1,
@@ -71,6 +77,61 @@ BeaconPayload = namedtuple('BeaconPayload', (
 ))
 
 
+def uuid_to_timestamp(uuid_str):
+    """Given the specified UUID string, returns the timestamp.
+
+    The timestamp returned should be comparable to what would be returned
+    from `import time; time.time()`.
+
+    :param uuid_str: a UUID in string format
+    :return: float
+    """
+    uuid_time = UUID(uuid_str).time
+    # Reverse the algorithm in uuid.py.
+    timestamp = (uuid_time - 0x01b21dd213814000) * 100 / 1e9
+    return timestamp
+
+
+def age_out_uuid_queue(queue, threshold=120.0):
+    """Ages out a ordered dictionary (using UUID-based keys) based on time.
+
+    The given threshold (in seconds) indicates how old an entry can be
+    before it will be removed from the queue.
+
+    :param queue: An `OrderedDict` with UUID strings as keys.
+    :param threshold: The maximum time an entry can remain in the queue.
+    """
+    removals = []
+    current_time = time.time()
+    for key in queue:
+        beacon_timestamp = uuid_to_timestamp(key)
+        # Don't leave beacons from the future in the queue if the clock
+        # suddenly changes. (This shouldn't happen, since the Fernet TTL
+        # should not have allowed them through. But just in case.)
+        difference = math.fabs(current_time - beacon_timestamp)
+        # Age out beacons greater than two minutes old.
+        if difference > threshold:
+            removals.append(key)
+        else:
+            # If we're already encountering packets that haven't met the age
+            # threshold (and were received more recently, and thus are later
+            # in the queue) then it's time to give up. (Yes, it's possible that
+            # clock skew could be an issue here, but after a couple of minutes,
+            # it won't matter.)
+            break
+    for uuid_to_remove in removals:
+        queue.pop(uuid_to_remove, None)
+
+
+def beacon_to_json(beacon_payload):
+    """Converts the specified beacon into a format suitable for JSON."""
+    return {
+        "version": beacon_payload.version,
+        "type": beacon_payload.type,
+        "payload": beacon_payload.payload,
+    }
+
+
 def create_beacon_payload(beacon_type, payload=None, version=PROTOCOL_VERSION):
     """Creates a beacon payload of the specified type, with the given data.
 
@@ -106,7 +167,7 @@ def read_beacon_payload(beacon_bytes):
     Decrypts the inner beacon data if necessary.
 
     :param beacon_bytes: beacon payload (bytes).
-    :return: dict
+    :return: BeaconPayload namedtuple
     """
     if len(beacon_bytes) < BEACON_HEADER_LENGTH_V1:
         raise InvalidBeaconingPacket(
@@ -131,7 +192,7 @@ def read_beacon_payload(beacon_bytes):
         else:
             try:
                 decrypted_data = fernet_decrypt_psk(
-                    payload_bytes, raw=True)
+                    payload_bytes, ttl=60, raw=True)
             except InvalidToken:
                 raise InvalidBeaconingPacket(
                     "Failed to decrypt inner payload: check MAAS secret key.")
@@ -176,18 +237,13 @@ class BeaconingPacket:
         self.valid = None
         self.invalid_reason = None
         self.packet = pkt_bytes
-        self.payload = self.parse()
+        self.data = self.parse()
 
     def parse(self):
-        """Output text-based details about this beaconing packet to the
-        specified file or stream.
-
-        :param out: An object with `write(str)` and `flush()` methods.
-        """
         try:
-            payload = read_beacon_payload(self.packet)
+            beacon = read_beacon_payload(self.packet)
             self.valid = True
-            return payload
+            return beacon
         except InvalidBeaconingPacket as ibp:
             self.valid = False
             self.invalid_reason = ibp.invalid_reason
@@ -213,6 +269,8 @@ def observe_beaconing_packets(input=sys.stdin.buffer, out=sys.stdout):
             try:
                 packet = decode_ethernet_udp_packet(packet_bytes, pcap_header)
                 beacon = BeaconingPacket(packet.payload)
+                if not beacon.valid:
+                    continue
                 output_json = {
                     "source_mac": format_eui(packet.l2.src_eui),
                     "destination_mac": format_eui(packet.l2.dst_eui),
@@ -220,15 +278,15 @@ def observe_beaconing_packets(input=sys.stdin.buffer, out=sys.stdout):
                     "destination_ip": str(packet.l3.dst_ip),
                     "source_port": packet.l4.packet.src_port,
                     "destination_port": packet.l4.packet.dst_port,
+                    "time": pcap_header.timestamp_seconds
                 }
                 if packet.l2.vid is not None:
                     output_json["vid"] = packet.l2.vid
-                if beacon.payload is not None:
-                    output_json['payload'] = beacon.payload
-                    output_json['time'] = pcap_header.timestamp_seconds
-                    out.write(json.dumps(output_json))
-                    out.write('\n')
-                    out.flush()
+                if beacon.data is not None:
+                    output_json.update(beacon_to_json(beacon.data))
+                out.write(json.dumps(output_json))
+                out.write('\n')
+                out.flush()
             except PacketProcessingError as e:
                 err.write(e.error)
                 err.write("\n")
diff --git a/src/provisioningserver/utils/services.py b/src/provisioningserver/utils/services.py
index dd6b399..0ca0394 100644
--- a/src/provisioningserver/utils/services.py
+++ b/src/provisioningserver/utils/services.py
@@ -11,10 +11,12 @@ from abc import (
     ABCMeta,
     abstractmethod,
 )
+from collections import OrderedDict
 from datetime import timedelta
 import json
 from json.decoder import JSONDecodeError
 import os
+from pprint import pformat
 import re
 
 from provisioningserver.config import is_dev_environment
@@ -23,6 +25,14 @@ from provisioningserver.logger import (
     LegacyLogger,
 )
 from provisioningserver.rpc.exceptions import NoConnectionsAvailable
+from provisioningserver.utils.beaconing import (
+    age_out_uuid_queue,
+    BEACON_IPV4_MULTICAST,
+    BEACON_PORT,
+    beacon_to_json,
+    create_beacon_payload,
+    read_beacon_payload,
+)
 from provisioningserver.utils.fs import (
     get_maas_provision_command,
     NamedLock,
@@ -47,8 +57,14 @@ from twisted.internet.error import (
     ProcessDone,
     ProcessTerminated,
 )
-from twisted.internet.protocol import ProcessProtocol
+from twisted.internet.interfaces import IReactorMulticast
+from twisted.internet.protocol import (
+    DatagramProtocol,
+    ProcessProtocol,
+)
 from twisted.internet.threads import deferToThread
+from zope.interface.exceptions import DoesNotImplement
+from zope.interface.verify import verifyObject
 
 
 maaslog = get_maas_logger("networks.monitor")
@@ -129,9 +145,6 @@ class ProtocolForObserveARP(JSONPerLineProtocol):
     The difference between `JSONPerLineProtocol` and `ProtocolForObserveARP`
     is that the neighbour observation protocol needs to insert the interface
     metadata into the resultant object before the callback.
-
-    This also ensures that the spawned process is configured as a process
-    group leader for its own process group.
     """
 
     def __init__(self, interface, *args, **kwargs):
@@ -147,6 +160,28 @@ class ProtocolForObserveARP(JSONPerLineProtocol):
         log.msg("observe-arp[%s]:" % self.interface, line)
 
 
+class ProtocolForObserveBeacons(JSONPerLineProtocol):
+    """Protocol used when spawning `maas-rack observe-beacons`.
+
+    The difference between `JSONPerLineProtocol` and
+    `ProtocolForObserveBeacons` is that the beacon observation protocol needs
+    to insert the interface metadata into the resultant object before the
+    callback.
+    """
+
+    def __init__(self, interface, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.interface = interface
+
+    def objectReceived(self, obj):
+        obj['interface'] = self.interface
+        super().objectReceived(obj)
+
+    def errLineReceived(self, line):
+        line = line.decode("utf-8").rstrip()
+        log.msg("observe-beacons[%s]:" % self.interface, line)
+
+
 class ProtocolForObserveMDNS(JSONPerLineProtocol):
     """Protocol used when spawning `maas-rack observe-mdns`.
 
@@ -257,6 +292,30 @@ class NeighbourDiscoveryService(ProcessProtocolService):
             self.ifname, callback=self.callback)
 
 
+class BeaconingService(ProcessProtocolService):
+    """Service to spawn the per-interface device discovery subprocess."""
+
+    def __init__(self, ifname: str, callback: callable):
+        self.ifname = ifname
+        self.callback = callback
+        super().__init__()
+
+    def getDescription(self) -> str:
+        return "Beaconing process for %s" % self.ifname
+
+    def getProcessParameters(self):
+        maas_rack_cmd = get_maas_provision_command().encode("utf-8")
+        return [
+            maas_rack_cmd,
+            b"observe-beacons",
+            self.ifname.encode("utf-8")
+        ]
+
+    def createProcessProtocol(self):
+        return ProtocolForObserveBeacons(
+            self.ifname, callback=self.callback)
+
+
 class MDNSResolverService(ProcessProtocolService):
     """Service to spawn the per-interface device discovery subprocess."""
 
@@ -278,6 +337,173 @@ class MDNSResolverService(ProcessProtocolService):
         return ProtocolForObserveMDNS(callback=self.callback)
 
 
+class BeaconingSocketProtocol(DatagramProtocol):
+    """Protocol to handle beaconing packets received from the socket layer."""
+
+    def __init__(
+            self, reactor, process_incoming=False, debug=True, interface='::',
+            loopback=False, port=BEACON_PORT):
+        super().__init__()
+        self.reactor = reactor
+        self.process_incoming = process_incoming
+        self.debug = debug
+        # These queues keep track of beacons that have recently been sent
+        # or received by the protocol. Ordering is needed here so that we can
+        # later age out the least-recently-added packets without traversing the
+        # entire dictionary.
+        self.tx_queue = OrderedDict()
+        self.rx_queue = OrderedDict()
+        self.listen_port = None
+        try:
+            # Need to ensure that the passed-in reactor is, in fact, a "real"
+            # reactor, and not None, or a mock reactor used in tests.
+            verifyObject(IReactorMulticast, reactor)
+            self.listen_port = reactor.listenMulticast(
+                port, self, interface=interface, listenMultiple=True)
+            self.transport.joinGroup(BEACON_IPV4_MULTICAST)
+            self.transport.setLoopbackMode(loopback)
+            # XXX mpontillo 2017-06-21: Twisted doesn't support IPv6 here yet.
+            # self.transport.joinGroup(BEACON_IPV6_MULTICAST)
+        except DoesNotImplement:
+            pass
+
+    def stopProtocol(self):
+        super().stopProtocol()
+        if self.listen_port is not None:
+            return self.listen_port.stopListening()
+        return None
+
+    def send_beacon(self, beacon, destination_address):
+        """Send a beacon to the specified destination.
+
+        :param beacon: The `BeaconPayload` namedtuple to send. Must have a
+            `payload` ivar containing a 'uuid' element.
+        :param destination_address: The UDP/IP (destination, port) tuple. IPv4
+            addresses must be in IPv4-mapped IPv6 format.
+        :return: True if the beacon was sent, False otherwise.
+        """
+        try:
+            self.transport.write(beacon.bytes, destination_address)
+            # If the packet cannot be sent for whatever reason, OSError will
+            # be raised, and we won't record sending a beacon we didn't
+            # actually send.
+            self.tx_queue[beacon.payload['uuid']] = beacon
+            age_out_uuid_queue(self.tx_queue)
+            return True
+        except OSError as e:
+            if self.debug is True:
+                log.msg("Error while sending beacon: %s" % e)
+        return False
+
+    def beaconReceived(self, beacon_json):
+        """Called whenever a beacon is received.
+
+        This method is responsible for updating the `tx_queue` and `rx_queue`
+        data structures, and determining if the incoming beacon is meaningful
+        for determining network topology.
+
+        :param beacon_json: The normalized beacon JSON, which can come either
+            from the external tcpdump-based process, or from the sockets layer
+            (with less information about the received packet).
+        """
+        rx_uuid = beacon_json.get('payload', {}).get("uuid")
+        if rx_uuid is None:
+            if self.debug is True:
+                log.msg(
+                    "Rejecting incoming beacon: no UUID found: \n%s" % (
+                        pformat(beacon_json)))
+            return
+        own_beacon = False
+        if self.tx_queue.get(rx_uuid):
+            own_beacon = True
+        is_dup = self.remember_beacon_and_check_duplicate(rx_uuid, beacon_json)
+        if self.debug is True:
+            log.msg("%s %sreceived:\n%s" % (
+                "Own beacon" if own_beacon else "Beacon",
+                "(duplicate) " if is_dup else "",
+                beacon_json))
+        # From what we know so far, we can infer some facts about the network.
+        # (1) If we received our own beacon, that means the interface we sent
+        # the packet out on is on the same fabric as the interface that
+        # received it.
+        # (2) If we receive a duplicate beacon on two different interfaces,
+        # that means those two interfaces are on the same fabric.
+        reply_ip = beacon_json['source_ip']
+        reply_port = beacon_json['source_port']
+        if ':' not in reply_ip:
+            # Since we opened an IPv6-compatible socket, need IPv6 syntax
+            # here to send to IPv4 addresses.
+            reply_ip = '::ffff:' + reply_ip
+        reply_address = (reply_ip, reply_port)
+        beacon_type = beacon_json['type']
+        if beacon_type == "solicitation":
+            receive_interface_info = self.get_receive_interface_info(
+                beacon_json)
+            payload = {
+                "interface": receive_interface_info,
+                "acks": rx_uuid
+            }
+            reply = create_beacon_payload("advertisement", payload)
+            self.send_beacon(reply, reply_address)
+
+    def remember_beacon_and_check_duplicate(self, rx_uuid, beacon_json):
+        """Records an incoming beacon based on its UUID and JSON.
+
+        Organizes incoming beacons in the `rx_queue` by creating a list of
+        beacons received [on different interfaces] per UUID.
+
+        :param rx_uuid: The UUID of the incoming beacon.
+        :param beacon_json: The incoming beacon (in JSON format).
+        :return: True if the beacon was a duplicate, otherwise False.
+        """
+        duplicate_received = False
+        # Need to age out before doing anything else; we don't want to match
+        # a duplicate packet and then delete it immediately after.
+        age_out_uuid_queue(self.rx_queue)
+        rx_packets_for_uuid = self.rx_queue.get(rx_uuid, [])
+        if len(rx_packets_for_uuid) > 0:
+            duplicate_received = True
+        rx_packets_for_uuid.append(beacon_json)
+        self.rx_queue[rx_uuid] = rx_packets_for_uuid
+        return duplicate_received
+
+    def get_receive_interface_info(self, context):
+        """Returns a dictionary representing information about the receive
+        interface, given the context of the beacon. The context can be the
+        limited information received from the socket layer, or the extended
+        information from the monitoring process.
+        """
+        receive_interface_info = {
+            "name": context.get('interface'),
+            "source_ip": context.get('source_ip'),
+            "destination_ip": context.get('destination_ip'),
+            "source_mac": context.get('source_mac'),
+            "destination_mac": context.get('destination_mac'),
+        }
+        if 'vid' in context:
+            receive_interface_info['vid'] = context['vid']
+        return receive_interface_info
+
+    def datagramReceived(self, datagram, addr):
+        """Called by Twisted when a UDP datagram is received.
+
+        Note: In the typical use case, the MAAS server will ignore packets
+        coming into this method. We need to listen to the socket normally,
+        however, so that the underlying network stack will send ICMP
+        destination (port) unreachable replies to anyone trying to send us
+        beacons. However, at other times, (such as while running the test
+        commands), we *will* listen to the socket layer for beacons.
+        """
+        if self.process_incoming is True:
+            context = {
+                "source_ip": addr[0],
+                "source_port": addr[1]
+            }
+            beacon_json = beacon_to_json(read_beacon_payload(datagram))
+            beacon_json.update(context)
+            self.beaconReceived(beacon_json)
+
+
 class NetworksMonitoringLock(NamedLock):
     """Host scoped lock to ensure only one network monitoring service runs."""
 
@@ -324,6 +550,7 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         self.interface_monitor.setName("updateInterfaces")
         self.interface_monitor.clock = self.clock
         self.interface_monitor.setServiceParent(self)
+        self.beaconing_protocol = BeaconingSocketProtocol(clock)
 
     def updateInterfaces(self):
         """Update interfaces, catching and logging errors.
@@ -388,12 +615,19 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         This MUST be overridden in subclasses.
         """
 
+    def reportBeacons(self, beacons):
+        """Receives a report of an observed beacon packet."""
+        for beacon in beacons:
+            log.msg("Received beacon: %r" % beacon)
+            self.beaconing_protocol.beaconReceived(beacon)
+
     def stopService(self):
         """Stop the service.
 
         Ensures that sole responsibility for monitoring networks is released.
         """
         d = super().stopService()
+        self.beaconing_protocol.stopProtocol()
         d.addBoth(callOut, self._releaseSoleResponsibility)
         return d
 
@@ -473,6 +707,13 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
         service.setName("neighbour_discovery:" + ifname)
         service.setServiceParent(self)
 
+    def _startBeaconing(self, ifname):
+        """"Start neighbour discovery service on the specified interface."""
+        service = BeaconingService(ifname, self.reportBeacons)
+        service.clock = self.clock
+        service.setName("beaconing:" + ifname)
+        service.setServiceParent(self)
+
     def _startMDNSDiscoveryService(self):
         """Start resolving mDNS entries on attached networks."""
         try:
@@ -520,6 +761,30 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
                 maaslog.info(
                     "Stopped neighbour observation service for %s." % ifname)
 
+    def _startBeaconingServices(self, new_interfaces):
+        """Start monitoring services for the specified set of interfaces."""
+        for ifname in new_interfaces:
+            # Sanity check to ensure the service isn't already started.
+            try:
+                self.getServiceNamed("beaconing:" + ifname)
+            except KeyError:
+                # This is an expected exception. (The call inside the `try`
+                # is only necessary to ensure the service doesn't exist.)
+                self._startBeaconing(ifname)
+
+    def _stopBeaconingServices(self, deleted_interfaces):
+        """Stop monitoring services for the specified set of interfaces."""
+        for ifname in deleted_interfaces:
+            try:
+                service = self.getServiceNamed("beaconing:" + ifname)
+            except KeyError:
+                # Service doesn't exist, so no need to stop it.
+                pass
+            else:
+                service.disownServiceParent()
+                maaslog.info(
+                    "Stopped beaconing service for %s." % ifname)
+
     def _shouldMonitorMDNS(self, monitoring_state):
         # If any interface is configured for mDNS, we must start the monitoring
         # process. (You cannot select interfaces when using `avahi-browse`.)
@@ -591,11 +856,17 @@ class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
             log.msg("Starting neighbour discovery for interfaces: %r" % (
                 new_interfaces))
             self._startNeighbourDiscoveryServices(new_interfaces)
+            # XXX mpontillo 2017-07-12: for testing, just start beaconing
+            # services on all the interfaces enabled for active discovery.
+            self._startBeaconingServices(new_interfaces)
         if len(deleted_interfaces) > 0:
             log.msg(
                 "Stopping neighbour discovery for interfaces: %r" % (
                     deleted_interfaces))
             self._stopNeighbourDiscoveryServices(deleted_interfaces)
+            # XXX mpontillo 2017-07-12: this should be separately configured.
+            # (see similar comment in the 'start' path above.)
+            self._stopBeaconingServices(deleted_interfaces)
         self._monitored = monitored_interfaces
 
     def _interfacesRecorded(self, interfaces):
diff --git a/src/provisioningserver/utils/tests/test_beaconing.py b/src/provisioningserver/utils/tests/test_beaconing.py
index 2448280..514b209 100644
--- a/src/provisioningserver/utils/tests/test_beaconing.py
+++ b/src/provisioningserver/utils/tests/test_beaconing.py
@@ -6,14 +6,20 @@
 __all__ = []
 
 from argparse import ArgumentParser
+from collections import OrderedDict
 from gzip import compress
 import io
+import math
 import random
 import struct
 import subprocess
 from tempfile import NamedTemporaryFile
+import time
 from unittest.mock import Mock
-from uuid import UUID
+from uuid import (
+    UUID,
+    uuid1,
+)
 
 from maastesting.factory import factory
 from maastesting.matchers import MockCalledOnceWith
@@ -26,23 +32,63 @@ from provisioningserver.tests.test_security import SharedSecretTestCase
 from provisioningserver.utils import beaconing as beaconing_module
 from provisioningserver.utils.beaconing import (
     add_arguments,
+    age_out_uuid_queue,
     BEACON_HEADER_FORMAT_V1,
+    beacon_to_json,
     BEACON_TYPES,
     BeaconingPacket,
+    BeaconPayload,
     create_beacon_payload,
     InvalidBeaconingPacket,
     read_beacon_payload,
     run,
+    uuid_to_timestamp,
 )
 from provisioningserver.utils.script import ActionScriptError
 from testtools.matchers import (
+    Contains,
     Equals,
+    HasLength,
     Is,
     IsInstance,
+    LessThan,
+    Not,
 )
 from testtools.testcase import ExpectedException
 
 
+class TestUUIDToTimestamp(MAASTestCase):
+
+    def test__round_trip_preserves_timestamp(self):
+        expected_timestamp = time.time()
+        uuid = str(uuid1())
+        actual_timestamp = uuid_to_timestamp(uuid)
+        difference = math.fabs(actual_timestamp - expected_timestamp)
+        # Tolerate a difference of ~3 seconds. We'll age out packets on the
+        # order of minutes, so that should be good enough.
+        self.assertThat(difference, LessThan(3.0))
+
+
+class TestBeaconToJSON(MAASTestCase):
+    """Tests for `beacon_to_json()` function."""
+
+    def test__preserves_version_type_and_payload__discards_bytes(self):
+        test_bytes = factory.make_bytes()
+        test_version = factory.make_string()
+        test_type = factory.make_string()
+        test_payload = {
+            factory.make_string(): factory.make_string()
+        }
+        beacon = BeaconPayload(
+            test_bytes, test_version, test_type, test_payload)
+        beacon_json = beacon_to_json(beacon)
+        self.assertThat(beacon_json["version"], Equals(test_version))
+        self.assertThat(beacon_json["type"], Equals(test_type))
+        self.assertThat(beacon_json["payload"], Equals(test_payload))
+        self.assertThat(beacon_json, Not(Contains("bytes")))
+        self.assertThat(beacon_json, HasLength(3))
+
+
 class TestCreateBeaconPayload(SharedSecretTestCase):
 
     def test__requires_maas_shared_secret_for_inner_data_payload(self):
@@ -286,3 +332,67 @@ class TestObserveBeaconsCommand(MAASTestCase):
         os.setpgrp.side_effect = exception_type
         self.assertRaises(exception_type, run, [])
         self.assertThat(os.setpgrp, MockCalledOnceWith())
+
+
+class TestAgeOutUUIDQueue(MAASTestCase):
+    """Tests for `age_out_uuid_queue()` function."""
+
+    def test__does_not_remove_fresh_entries(self):
+        uuid_now = str(uuid1())
+        queue = OrderedDict()
+        queue[uuid_now] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(1))
+
+    def test__keeps_entries_from_the_reasonable_past(self):
+        uuid_from_the_past = factory.make_UUID_with_timestamp(
+            time.time() - 60.0)
+        queue = OrderedDict()
+        queue[uuid_from_the_past] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(1))
+
+    def test__keeps_entries_from_the_reasonable_future(self):
+        uuid_from_the_future = factory.make_UUID_with_timestamp(
+            time.time() + 60.0)
+        queue = OrderedDict()
+        queue[uuid_from_the_future] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(1))
+
+    def test__removes_entries_from_the_past(self):
+        uuid_from_the_past = factory.make_UUID_with_timestamp(
+            time.time() - 123.0)
+        queue = OrderedDict()
+        queue[uuid_from_the_past] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(0))
+
+    def test__removes_entries_from_the_future(self):
+        uuid_from_the_future = factory.make_UUID_with_timestamp(
+            time.time() + 123.0)
+        queue = OrderedDict()
+        queue[uuid_from_the_future] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(0))
+
+    def test__removes_entries_from_the_distant_past(self):
+        uuid_from_the_past = '00000000-0000-1000-aaaa-aaaaaaaaaaaa'
+        queue = OrderedDict()
+        queue[uuid_from_the_past] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(0))
+
+    def test__removes_entries_from_the_far_future(self):
+        uuid_from_the_future = 'ffffffff-ffff-1fff-0000-000000000000'
+        queue = OrderedDict()
+        queue[uuid_from_the_future] = {}
+        self.assertThat(queue, HasLength(1))
+        age_out_uuid_queue(queue)
+        self.assertThat(queue, HasLength(0))
diff --git a/src/provisioningserver/utils/tests/test_services.py b/src/provisioningserver/utils/tests/test_services.py
index e287bec..ce8361f 100644
--- a/src/provisioningserver/utils/tests/test_services.py
+++ b/src/provisioningserver/utils/tests/test_services.py
@@ -6,6 +6,7 @@
 __all__ = []
 
 from functools import partial
+import random
 import threading
 from unittest.mock import (
     call,
@@ -27,8 +28,12 @@ from maastesting.testcase import (
     MAASTwistedRunTest,
 )
 from maastesting.twisted import TwistedLoggerFixture
+from provisioningserver.tests.test_security import SharedSecretTestCase
 from provisioningserver.utils import services
+from provisioningserver.utils.beaconing import create_beacon_payload
 from provisioningserver.utils.services import (
+    BeaconingService,
+    BeaconingSocketProtocol,
     JSONPerLineProtocol,
     MDNSResolverService,
     NeighbourDiscoveryService,
@@ -36,6 +41,7 @@ from provisioningserver.utils.services import (
     NetworksMonitoringService,
     ProcessProtocolService,
     ProtocolForObserveARP,
+    ProtocolForObserveBeacons,
 )
 from testtools import ExpectedException
 from testtools.matchers import (
@@ -47,6 +53,7 @@ from testtools.matchers import (
 from twisted.application.service import MultiService
 from twisted.internet import reactor
 from twisted.internet.defer import (
+    Deferred,
     DeferredQueue,
     inlineCallbacks,
     succeed,
@@ -382,6 +389,21 @@ class TestProtocolForObserveARP(MAASTestCase):
             callback, MockCallsMatch(call([{"interface": ifname}])))
 
 
+class TestProtocolForObserveBeacons(MAASTestCase):
+    """Tests for `ProtocolForObserveBeacons`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def test_adds_interface(self):
+        callback = Mock()
+        ifname = factory.make_name('eth')
+        proto = ProtocolForObserveBeacons(ifname, callback=callback)
+        proto.makeConnection(Mock(pid=None))
+        proto.outReceived(b"{}\n")
+        self.expectThat(
+            callback, MockCallsMatch(call([{"interface": ifname}])))
+
+
 class MockProcessProtocolService(ProcessProtocolService):
 
     def __init__(self):
@@ -566,6 +588,58 @@ class TestNeighbourDiscoveryService(MAASTestCase):
             % (ifname, ifname)))
 
 
+class TestBeaconingService(MAASTestCase):
+    """Tests for `BeaconingService`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def test__returns_expected_arguments(self):
+        ifname = factory.make_name('eth')
+        service = BeaconingService(ifname, Mock())
+        args = service.getProcessParameters()
+        self.assertThat(args, HasLength(3))
+        self.assertTrue(args[0].endswith(b'maas-rack'))
+        self.assertTrue(args[1], Equals(b"observe-beacons"))
+        self.assertTrue(args[2], Equals(ifname.encode('utf-8')))
+
+    @inlineCallbacks
+    def test__restarts_process_after_finishing(self):
+        ifname = factory.make_name('eth')
+        service = BeaconingService(ifname, Mock())
+        mock_process_params = self.patch(service, 'getProcessParameters')
+        mock_process_params.return_value = [b'/bin/echo', b'{}']
+        service.clock = Clock()
+        service.startService()
+        # Wait for the protocol to finish
+        service.clock.advance(0.0)
+        yield service._protocol.done
+        # Advance the clock (should start the service again)
+        interval = service.step
+        service.clock.advance(interval)
+        # The Deferred should have been recreated.
+        self.assertThat(service._protocol.done, Not(IsFiredDeferred()))
+        yield service._protocol.done
+        service.stopService()
+
+    @inlineCallbacks
+    def test__protocol_logs_stderr(self):
+        logger = self.useFixture(TwistedLoggerFixture())
+        ifname = factory.make_name('eth')
+        service = BeaconingService(ifname, lambda _: None)
+        protocol = service.createProcessProtocol()
+        reactor.spawnProcess(protocol, b"sh", (b"sh", b"-c", b"exec cat >&2"))
+        protocol.transport.write(
+            b"Lines written to stderr are logged\n"
+            b"with a prefix, with no exceptions.\n")
+        protocol.transport.closeStdin()
+        yield protocol.done
+        self.assertThat(logger.output, Equals(
+            "observe-beacons[%s]: Lines written to stderr are logged\n"
+            "---\n"
+            "observe-beacons[%s]: with a prefix, with no exceptions."
+            % (ifname, ifname)))
+
+
 class TestMDNSResolverService(MAASTestCase):
     """Tests for `MDNSResolverService`."""
 
@@ -594,3 +668,82 @@ class TestMDNSResolverService(MAASTestCase):
             "observe-mdns: Lines written to stderr are logged\n"
             "---\n"
             "observe-mdns: with a prefix, with one exception:"))
+
+
+def wait_for_rx_packets(beacon_protocol, count, deferred=None):
+    """Waits for a BeaconingSocketProtocol to transmit `count` packets."""
+    if deferred is None:
+        deferred = Deferred()
+    if len(beacon_protocol.rx_queue) >= count:
+        deferred.callback(None)
+    else:
+        reactor.callLater(
+            0.001, wait_for_rx_packets, beacon_protocol, count,
+            deferred=deferred)
+    return deferred
+
+
+class TestBeaconingSocketProtocol(SharedSecretTestCase):
+    """Tests for `BeaconingSocketProtocol`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=15)
+
+    @inlineCallbacks
+    def test__creates_listen_port_when_run_with_IReactorMulticast(self):
+        # Note: Always use a random port for testing. (port=0)
+        protocol = BeaconingSocketProtocol(reactor, port=0)
+        self.assertThat(protocol.listen_port, Not(Is(None)))
+        # This tests that the post gets closed properly; otherwise the test
+        # suite will complain about things left in the reactor.
+        yield protocol.stopProtocol()
+
+    def test__skips_creating_listen_port_when_run_with_fake_reactor(self):
+        # Note: Always use a random port for testing. (port=0)
+        protocol = BeaconingSocketProtocol(Clock(), port=0)
+        self.assertThat(protocol.listen_port, Is(None))
+        # No listen port, so stopProtocol() shouldn't return a Deferred.
+        result = protocol.stopProtocol()
+        self.assertThat(result, Is(None))
+
+    @inlineCallbacks
+    def test__sends_and_receives_beacons(self):
+        # Note: Always use a random port for testing. (port=0)
+        logger = self.useFixture(TwistedLoggerFixture())
+        protocol = BeaconingSocketProtocol(
+            reactor, port=0, process_incoming=True, loopback=True,
+            interface="::", debug=True)
+        self.assertThat(protocol.listen_port, Not(Is(None)))
+        listen_port = protocol.listen_port._realPortNumber
+        self.write_secret()
+        beacon = create_beacon_payload("solicitation", {})
+        rx_uuid = beacon.payload['uuid']
+        # XXX: We can't test IPv6 here until we either join the group
+        # manually, or Twisted gains support for joining IPv6 groups.
+        destination = random.choice(["::ffff:127.0.0.1", "::1"])
+        protocol.send_beacon(beacon, (destination, listen_port))
+        # Since we've instructed the protocol to loop back packets for testing,
+        # it should have sent a multicast solicitation, received it back, sent
+        # an advertisement, then received it back. So we'll wait for two
+        # packets to be sent.
+        yield wait_for_rx_packets(protocol, 2)
+        # Grab the beacon we know we transmitted and then received.
+        transmitted = protocol.tx_queue.pop(rx_uuid, None)
+        received = protocol.rx_queue.pop(rx_uuid, None)
+        self.assertThat(transmitted, Equals(beacon))
+        self.assertThat(received[0]['payload']['uuid'], Equals(rx_uuid))
+        # Grab the subsequent packets from the queues.
+        transmitted = protocol.tx_queue.popitem()[1]
+        received = protocol.rx_queue.popitem()[1]
+        # We should have received a second packet to ack the first beacon.
+        self.assertThat(received[0]['payload']['acks'], Equals(rx_uuid))
+        # We should have transmitted an advertisement in response to the
+        # solicitation.
+        self.assertThat(transmitted.type, Equals('advertisement'))
+        # This tests that the post gets closed properly; otherwise the test
+        # suite will complain about things left in the reactor.
+        yield protocol.stopProtocol()
+        # In debug mode, the logger should have printed each packet.
+        self.assertThat(
+            logger.output,
+            DocTestMatches(
+                '...Own beacon received:...Own beacon received:...'))