Determine network status based on ip command (#1827)

Related #1814.

This PR refactors the backend logic of how we determine the connection
status of the network interfaces. It extends the response structure so
that we will be able to show IP address and MAC address in a separate
network dialog. The current WiFi dialog also uses this endpoint, so
there needs to be a small adjustment to accommodate the new structure
there.

Notes:

- I renamed the `network.status()` method to
`network.determine_network_status()`, to reflect that it’s now actively
running a command. In order to avoid code duplication,
`network.determine_network_status()` is actually just a convenience
wrapper around `network.inspect_interface()`, which also encapsulates
the technical interface names `eth0` and `wlan0`. I also renamed and
restructured the `NetworkStatus` class to `InterfaceStatus`.
- I found the JSON output structure of the `ip` command a bit hard to
predict, as I saw numerous invariants of which fields were present
during testing, which I wasn’t able to reproduce. I also couldn’t find
reliable documentation for it. I therefore made the code inside the
`network.inspect_interface()` method very defensive and
failure-tolerant, to avoid producing erratic errors. I thought in the
end it would be more user-friendly to show a network interface as down,
even if that’s a false positive/negative, as opposed to potentially
showing a lot of errors and risking to give the feature an overall
brittle impression.
- For consistency [with the Wake on LAN
feature](https://github.com/tiny-pilot/tinypilot-pro/blob/a2ee01de5d97af4f81b3bb60e2af3f61abb8ed63/app/request_parsers/wake_on_lan.py#L14),
I normalized the MAC address to use dashes as delimiter.

<a data-ca-tag
href="https://codeapprove.com/pr/tiny-pilot/tinypilot/1827"><img
src="https://codeapprove.com/external/github-tag-allbg.png" alt="Review
on CodeApprove" /></a>

---------

Co-authored-by: Jan Heuermann <[email protected]>

app/api.py

@@ -203,23 +203,58 @@ def hostname_set():
 
 @api_blueprint.route('/network/status', methods=['GET'])
 def network_status():
-    """Returns the current network status (i.e., which interfaces are active).
+    """Returns the current status of the available network interfaces.
 
     Returns:
         On success, a JSON data structure with the following properties:
-        ethernet: bool.
-        wifi: bool
+        ethernet: object
+        wifi: object
+        The object contains the following fields:
+        isConnected: bool
+        ipAddress: string or null
+        macAddress: string or null
 
         Example:
         {
-            "ethernet": true,
-            "wifi": false
+            "ethernet": {
+                "isConnected": true,
+                "ipAddress": "192.168.2.41",
+                "macAddress": "e4-5f-01-98-65-03"
+            },
+            "wifi": {
+                "isConnected": false,
+                "ipAddress": null,
+                "macAddress": null
+            }
         }
     """
-    status = network.status()
+    # In dev mode, return dummy data because attempting to read the actual
+    # settings will fail in most non-Raspberry Pi OS environments.
+    if flask.current_app.debug:
+        return json_response.success({
+            'ethernet': {
+                'isConnected': True,
+                'ipAddress': '192.168.2.8',
+                'macAddress': '00-b0-d0-63-c2-26',
+            },
+            'wifi': {
+                'isConnected': False,
+                'ipAddress': None,
+                'macAddress': None,
+            },
+        })
+    ethernet, wifi = network.determine_network_status()
     return json_response.success({
-        'ethernet': status.ethernet,
-        'wifi': status.wifi,
+        'ethernet': {
+            'isConnected': ethernet.is_connected,
+            'ipAddress': ethernet.ip_address,
+            'macAddress': ethernet.mac_address,
+        },
+        'wifi': {
+            'isConnected': wifi.is_connected,
+            'ipAddress': wifi.ip_address,
+            'macAddress': wifi.mac_address,
+        },
     })
 
 

app/network.py

@@ -1,7 +1,11 @@
 import dataclasses
+import json
+import logging
 import re
 import subprocess
 
+logger = logging.getLogger(__name__)
+
 _WIFI_COUNTRY_PATTERN = re.compile(r'^\s*country=(.+)$')
 _WIFI_SSID_PATTERN = re.compile(r'^\s*ssid="(.+)"$')
 
@@ -15,9 +19,10 @@ class NetworkError(Error):
 
 
 @dataclasses.dataclass
-class NetworkStatus:
-    ethernet: bool
-    wifi: bool
+class InterfaceStatus:
+    is_connected: bool
+    ip_address: str  # May be `None` if interface is disabled.
+    mac_address: str  # May be `None` if interface is disabled.
 
 
 @dataclasses.dataclass
@@ -27,26 +32,76 @@ class WifiSettings:
     psk: str  # Optional.
 
 
-def status():
+def determine_network_status():
     """Checks the connectivity of the network interfaces.
 
     Returns:
-        NetworkStatus
+        A tuple of InterfaceStatus objects for the Ethernet and WiFi interface.
+    """
+    return inspect_interface('eth0'), inspect_interface('wlan0')
+
+
+def inspect_interface(interface_name):
+    """Gathers information about a network interface.
+
+    This method relies on the JSON output of the `ip` command. If the interface
+    is available, the JSON structure is an array containing an object, which
+    looks like the following (extra properties omitted for brevity):
+        [{
+            "operstate": "UP",
+            "address": "e4:5f:01:98:65:05",
+            "addr_info": [{"family":"inet", "local":"192.168.12.86"}]
+        }]
+    Note that `addr_info` might be empty, e.g. if `operstate` is `DOWN`;
+    it also might contain additional families, such as `inet6` (IPv6).
+
+    In general, we don’t have too much trust in the consistency of the JSON
+    structure, as there is no reliable documentation for it. We try to handle
+    and parse the output in a defensive and graceful way, to maximize
+    robustness and avoid producing erratic failures.
+
+    Args:
+        interface_name: the technical interface name as string, e.g. `eth0`.
+
+    Returns:
+        InterfaceStatus object
     """
-    network_status = NetworkStatus(False, False)
+    status = InterfaceStatus(False, None, None)
+
     try:
-        with open('/sys/class/net/eth0/operstate', encoding='utf-8') as file:
-            eth0 = file.read().strip()
-            network_status.ethernet = eth0 == 'up'
-    except OSError:
-        pass  # We treat this as if the interface was down altogether.
+        ip_cmd_out_raw = subprocess.check_output([
+            'ip',
+            '-json',
+            'address',
+            'show',
+            interface_name,
+        ],
+                                                 stderr=subprocess.STDOUT,
+                                                 universal_newlines=True)
+    except subprocess.CalledProcessError as e:
+        logger.error('Failed to run `ip` command: %s', str(e))
+        return status
+
     try:
-        with open('/sys/class/net/wlan0/operstate', encoding='utf-8') as file:
-            wlan0 = file.read().strip()
-            network_status.wifi = wlan0 == 'up'
-    except OSError:
-        pass  # We treat this as if the interface was down altogether.
-    return network_status
+        json_output = json.loads(ip_cmd_out_raw)
+    except json.decoder.JSONDecodeError as e:
+        logger.error('Failed to parse JSON output of `ip` command: %s', str(e))
+        return status
+
+    if len(json_output) == 0:
+        return status
+    data = json_output[0]
+
+    if 'operstate' in data:
+        status.is_connected = data['operstate'] == 'UP'
+    if 'address' in data:
+        status.mac_address = data['address'].replace(':', '-')
+    if 'addr_info' in data:
+        status.ip_address = next((addr_info['local']
+                                  for addr_info in data['addr_info']
+                                  if addr_info['family'] == 'inet'), None)
+
+    return status
 
 
 def determine_wifi_settings():

app/network_test.py

@@ -0,0 +1,125 @@
+import subprocess
+import unittest
+from unittest import mock
+
+import network
+
+
+# This test checks the various potential JSON output values that the underlying
+# `ip` command may return.
+class InspectInterfaceTest(unittest.TestCase):
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_treats_empty_response_as_inactive_interface(self, mock_cmd):
+        mock_cmd.return_value = ''
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_treats_empty_array_as_inactive_interface(self, mock_cmd):
+        mock_cmd.return_value = '[]'
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_treats_emtpy_object_as_inactive_interface(self, mock_cmd):
+        mock_cmd.return_value = '[{}]'
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_disregards_command_failure(self, mock_cmd):
+        mock_cmd.side_effect = mock.Mock(
+            side_effect=subprocess.CalledProcessError(returncode=1, cmd='ip'))
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_parses_operstate_down_as_not_connected(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"operstate":"DOWN"}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_parses_operstate_up_as_connected(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"operstate":"UP"}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(True, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_parses_mac_address(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"address":"00-b0-d0-63-c2-26"}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(False, None, '00-b0-d0-63-c2-26'),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_normalizes_mac_address_to_use_dashes(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"address":"00:b0:d0:63:c2:26"}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(False, None, '00-b0-d0-63-c2-26'),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_parses_ip_address(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"addr_info":[{"family":"inet","local":"192.168.2.5"}]}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(False, '192.168.2.5', None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_disregards_other_families_such_as_ipv6(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{"addr_info":[{"family":"inet6","local":"::ffff:c0a8:205"}]}]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_parses_all_data(self, mock_cmd):
+        mock_cmd.return_value = """
+            [{
+                "operstate":"UP",
+                "address":"00-b0-d0-63-c2-26",
+                "addr_info":[{"family":"inet","local":"192.168.2.5"}]
+            }]
+        """
+        self.assertEqual(
+            network.InterfaceStatus(True, '192.168.2.5', '00-b0-d0-63-c2-26'),
+            network.inspect_interface('eth0'),
+        )
+
+    @mock.patch.object(subprocess, 'check_output')
+    def test_disregards_invalid_json(self, mock_cmd):
+        mock_cmd.return_value = '[{"address'
+        self.assertEqual(
+            network.InterfaceStatus(False, None, None),
+            network.inspect_interface('eth0'),
+        )

app/static/js/controllers.js

@@ -208,6 +208,14 @@ export async function getNetworkStatus() {
         if (!response.hasOwnProperty(field)) {
           throw new ControllerError(`Missing expected ${field} field`);
         }
+        ["isConnected", "ipAddress", "macAddress"].forEach((property) => {
+          // eslint-disable-next-line no-prototype-builtins
+          if (!response[field].hasOwnProperty(property)) {
+            throw new ControllerError(
+              `Missing expected ${field}.${property} field`
+            );
+          }
+        });
       });
       return response;
     });

app/templates/custom-elements/wifi-dialog.html

@@ -272,7 +272,7 @@ <h3>Wi-Fi Credentials Removed</h3>
 
           this._elements.noEthernetWarning.hide();
           this._elements.inputError.hide();
-          if (!networkStatus.ethernet) {
+          if (!networkStatus.ethernet.isConnected) {
             this._elements.noEthernetWarning.show();
           }