Add OPNSense device tracker

.coveragerc

@@ -498,6 +498,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

@@ -238,6 +238,7 @@ homeassistant/components/onboarding/* @home-assistant/core
 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/owlet/* @oblogic7

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/components/opnsense",
+  "requirements": [
+    "pyopnsense==0.2.0"
+  ],
+  "dependencies": [],
+  "codeowners": ["@mtreinish"]
+}

requirements_all.txt

@@ -1409,6 +1409,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

@@ -472,6 +472,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"