Add OPNSense device tracker (#26834)

* Add OPNSense device_tracker

This commit adds a new component for using an OPNSense router as a
device tracker. It uses pyopnsense to query the api to look at the
arptable for a list of devices on the network.

* Run black formatting locally to appease azure

* Apply suggestions from code review

Co-Authored-By: Fabian Affolter <mail@fabian-affolter.ch>

* Fix issues identified during code review

This commit updates several issues found in the module during code
review.

* Update homeassistant/components/opnsense/__init__.py

Co-Authored-By: Fabian Affolter <mail@fabian-affolter.ch>

* Update CODEOWNERS for recent changes

* Fix lint

* Apply suggestions from code review

Co-Authored-By: Martin Hjelmare <marhje52@kth.se>

* More fixes from review comments

This commit fixes several issues from review comments, including
abandoning all the use of async code. This also completely reworks the
tests to be a bit clearer.

* Revert tests to previous format

* Add device detection to opnsense device_tracker test

This commit adds actual device detection to the unit test for the setup
test. A fake api response is added to mocks for both api clients so that
they will register devices as expected and asserts are added for that.

The pyopnsense import is moved from the module level to be runtime in
the class. This was done because it was the only way to make the
MockDependency() call work as expected.

* Rerun black

* Fix lint

* Move import back to module level

* Return false on configuration errors in setup

This commit updates the connection logic to return false if we're unable
to connect to the configured OPNsense API endpoint for any reason.
Previously we would not catch if an endpoint was incorrectly configured
until we first tried to use it. In this case it would raise an unhandled
exception. To handle this more gracefully this adds an api call early in
the setup and catches any exception raised by that so we can return
False to indicate the setup failed.

* Update tests

* Add pyopnsense to test requirements

* Rerun gen_requirements script

* Fix failing isort lint job step

Since opening the PR originally yet another lint/style checker was added
which failed the PR in CI. This commit makes the adjustments to have
this pass the additional tool's checks.

* Fix comment

* Update manifest.json

Co-authored-by: Fabian Affolter <mail@fabian-affolter.ch>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
Co-authored-by: Pascal Vizeli <pascal.vizeli@syshack.ch>

.coveragerc

@@ -508,6 +508,7 @@ omit =
     homeassistant/components/openuv/sensor.py
     homeassistant/components/openweathermap/sensor.py
     homeassistant/components/openweathermap/weather.py
+    homeassistant/components/opnsense/*
     homeassistant/components/opple/light.py
     homeassistant/components/orangepi_gpio/*
     homeassistant/components/oru/*

CODEOWNERS

@@ -247,6 +247,7 @@ homeassistant/components/onewire/* @garbled1
 homeassistant/components/opentherm_gw/* @mvn23
 homeassistant/components/openuv/* @bachya
 homeassistant/components/openweathermap/* @fabaff
+homeassistant/components/opnsense/* @mtreinish
 homeassistant/components/orangepi_gpio/* @pascallj
 homeassistant/components/oru/* @bvlaicu
 homeassistant/components/panel_custom/* @home-assistant/frontend

homeassistant/components/opnsense/__init__.py

@@ -0,0 +1,77 @@
+"""Support for OPNSense Routers."""
+import logging
+
+from pyopnsense import diagnostics
+from pyopnsense.exceptions import APIException
+import voluptuous as vol
+
+from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.discovery import load_platform
+
+_LOGGER = logging.getLogger(__name__)
+
+CONF_API_SECRET = "api_secret"
+CONF_TRACKER_INTERFACE = "tracker_interfaces"
+
+DOMAIN = "opnsense"
+
+OPNSENSE_DATA = DOMAIN
+
+CONFIG_SCHEMA = vol.Schema(
+    {
+        DOMAIN: vol.Schema(
+            {
+                vol.Required(CONF_URL): cv.url,
+                vol.Required(CONF_API_KEY): cv.string,
+                vol.Required(CONF_API_SECRET): cv.string,
+                vol.Optional(CONF_VERIFY_SSL, default=False): cv.boolean,
+                vol.Optional(CONF_TRACKER_INTERFACE, default=[]): vol.All(
+                    cv.ensure_list, [cv.string]
+                ),
+            }
+        )
+    },
+    extra=vol.ALLOW_EXTRA,
+)
+
+
+def setup(hass, config):
+    """Set up the opnsense component."""
+
+    conf = config[DOMAIN]
+    url = conf[CONF_URL]
+    api_key = conf[CONF_API_KEY]
+    api_secret = conf[CONF_API_SECRET]
+    verify_ssl = conf[CONF_VERIFY_SSL]
+    tracker_interfaces = conf[CONF_TRACKER_INTERFACE]
+
+    interfaces_client = diagnostics.InterfaceClient(
+        api_key, api_secret, url, verify_ssl
+    )
+    try:
+        interfaces_client.get_arp()
+    except APIException:
+        _LOGGER.exception("Failure while connecting to OPNsense API endpoint.")
+        return False
+
+    if tracker_interfaces:
+        # Verify that specified tracker interfaces are valid
+        netinsight_client = diagnostics.NetworkInsightClient(
+            api_key, api_secret, url, verify_ssl
+        )
+        interfaces = list(netinsight_client.get_interfaces().values())
+        for interface in tracker_interfaces:
+            if interface not in interfaces:
+                _LOGGER.error(
+                    "Specified OPNsense tracker interface %s is not found", interface
+                )
+                return False
+
+    hass.data[OPNSENSE_DATA] = {
+        "interfaces": interfaces_client,
+        CONF_TRACKER_INTERFACE: tracker_interfaces,
+    }
+
+    load_platform(hass, "device_tracker", DOMAIN, tracker_interfaces, config)
+    return True

homeassistant/components/opnsense/device_tracker.py

@@ -0,0 +1,66 @@
+"""Device tracker support for OPNSense routers."""
+import logging
+
+from homeassistant.components.device_tracker import DeviceScanner
+from homeassistant.components.opnsense import CONF_TRACKER_INTERFACE, OPNSENSE_DATA
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_get_scanner(hass, config, discovery_info=None):
+    """Configure the OPNSense device_tracker."""
+    interface_client = hass.data[OPNSENSE_DATA]["interfaces"]
+    scanner = OPNSenseDeviceScanner(
+        interface_client, hass.data[OPNSENSE_DATA][CONF_TRACKER_INTERFACE]
+    )
+    return scanner
+
+
+class OPNSenseDeviceScanner(DeviceScanner):
+    """This class queries a router running OPNsense."""
+
+    def __init__(self, client, interfaces):
+        """Initialize the scanner."""
+        self.last_results = {}
+        self.client = client
+        self.interfaces = interfaces
+
+    def _get_mac_addrs(self, devices):
+        """Create dict with mac address keys from list of devices."""
+        out_devices = {}
+        for device in devices:
+            if not self.interfaces:
+                out_devices[device["mac"]] = device
+            elif device["intf_description"] in self.interfaces:
+                out_devices[device["mac"]] = device
+        return out_devices
+
+    def scan_devices(self):
+        """Scan for new devices and return a list with found device IDs."""
+        self.update_info()
+        return list(self.last_results)
+
+    def get_device_name(self, device):
+        """Return the name of the given device or None if we don't know."""
+        if device not in self.last_results:
+            return None
+        hostname = self.last_results[device].get("hostname") or None
+        return hostname
+
+    def update_info(self):
+        """Ensure the information from the OPNSense router is up to date.
+
+        Return boolean if scanning successful.
+        """
+
+        devices = self.client.get_arp()
+        self.last_results = self._get_mac_addrs(devices)
+
+    def get_extra_attributes(self, device):
+        """Return the extra attrs of the given device."""
+        if device not in self.last_results:
+            return None
+        mfg = self.last_results[device].get("manufacturer")
+        if mfg:
+            return {"manufacturer": mfg}
+        return {}

homeassistant/components/opnsense/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "opnsense",
+  "name": "OPNSense",
+  "documentation": "https://www.home-assistant.io/integrations/opnsense",
+  "requirements": [
+    "pyopnsense==0.2.0"
+  ],
+  "dependencies": [],
+  "codeowners": ["@mtreinish"]
+}

requirements_all.txt

@@ -1416,6 +1416,9 @@ pyombi==0.1.10
 # homeassistant.components.openuv
 pyopenuv==1.0.9
 
+# homeassistant.components.opnsense
+pyopnsense==0.2.0
+
 # homeassistant.components.opple
 pyoppleio==1.0.5
 

requirements_test_all.txt

@@ -492,6 +492,9 @@ pynx584==0.4
 # homeassistant.components.openuv
 pyopenuv==1.0.9
 
+# homeassistant.components.opnsense
+pyopnsense==0.2.0
+
 # homeassistant.components.opentherm_gw
 pyotgw==0.5b1
 

tests/components/opnsense/__init__.py

@@ -0,0 +1 @@
+"""Tests for the opnsense component."""

tests/components/opnsense/test_device_tracker.py

@@ -0,0 +1,64 @@
+"""The tests for the opnsense device tracker platform."""
+
+from unittest import mock
+
+import pytest
+
+from homeassistant.components import opnsense
+from homeassistant.components.opnsense import CONF_API_SECRET, DOMAIN
+from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture(name="mocked_opnsense")
+def mocked_opnsense():
+    """Mock for pyopnense.diagnostics."""
+    with mock.patch.object(opnsense, "diagnostics") as mocked_opn:
+        yield mocked_opn
+
+
+async def test_get_scanner(hass, mocked_opnsense):
+    """Test creating an opnsense scanner."""
+    interface_client = mock.MagicMock()
+    mocked_opnsense.InterfaceClient.return_value = interface_client
+    interface_client.get_arp.return_value = [
+        {
+            "hostname": "",
+            "intf": "igb1",
+            "intf_description": "LAN",
+            "ip": "192.168.0.123",
+            "mac": "ff:ff:ff:ff:ff:ff",
+            "manufacturer": "",
+        },
+        {
+            "hostname": "Desktop",
+            "intf": "igb1",
+            "intf_description": "LAN",
+            "ip": "192.168.0.167",
+            "mac": "ff:ff:ff:ff:ff:fe",
+            "manufacturer": "OEM",
+        },
+    ]
+    network_insight_client = mock.MagicMock()
+    mocked_opnsense.NetworkInsightClient.return_value = network_insight_client
+    network_insight_client.get_interfaces.return_value = {"igb0": "WAN", "igb1": "LAN"}
+
+    result = await async_setup_component(
+        hass,
+        DOMAIN,
+        {
+            DOMAIN: {
+                CONF_URL: "https://fake_host_fun/api",
+                CONF_API_KEY: "fake_key",
+                CONF_API_SECRET: "fake_secret",
+                CONF_VERIFY_SSL: False,
+            }
+        },
+    )
+    await hass.async_block_till_done()
+    assert result
+    device_1 = hass.states.get("device_tracker.desktop")
+    assert device_1 is not None
+    assert device_1.state == "home"
+    device_2 = hass.states.get("device_tracker.ff_ff_ff_ff_ff_ff")
+    assert device_2.state == "home"