Add Unbound Blocklist Switch

custom_components/opnsense/const.py

@@ -52,6 +52,8 @@
 
 ICON_MEMORY = "mdi:memory"
 
+ATTR_UNBOUND_BLOCKLIST = "unbound_blocklist"
+
 SENSOR_TYPES: Final[dict[str, SensorEntityDescription]] = {
     # pfstate
     "telemetry.pfstate.used": SensorEntityDescription(

custom_components/opnsense/coordinator.py

@@ -7,6 +7,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
+from .const import ATTR_UNBOUND_BLOCKLIST
 from .helpers import dict_get
 from .pyopnsense import OPNsenseClient
 
@@ -144,6 +145,9 @@ async def _async_update_data(self):
             self._state["dhcp_leases"] = []
             self._state["dhcp_stats"] = {}
             self._state["notices"] = await self._get_notices()
+            self._state[ATTR_UNBOUND_BLOCKLIST] = (
+                await self._client.get_unbound_blocklist()
+            )
 
             lease_stats: Mapping[str, int] = {"total": 0, "online": 0, "offline": 0}
             for lease in self._state["dhcp_leases"]:

custom_components/opnsense/pyopnsense/__init__.py

@@ -1393,3 +1393,61 @@ async def close_notice(self, id) -> bool:
                 success = False
         _LOGGER.debug(f"[close_notice] success: {success}")
         return success
+
+    @_log_errors
+    async def get_unbound_blocklist(self) -> Mapping[str, Any]:
+        response: Mapping[str, Any] | list = await self._get(
+            "/api/unbound/settings/get"
+        )
+        if response is None or not isinstance(response, Mapping):
+            _LOGGER.error("Invalid data returned from get_unbound_blocklist")
+            return {}
+        # _LOGGER.debug(f"[get_unbound_blocklist] response: {response}")
+        dnsbl_settings = response.get("unbound", {}).get("dnsbl", {})
+        # _LOGGER.debug(f"[get_unbound_blocklist] dnsbl_settings: {dnsbl_settings}")
+        if not isinstance(dnsbl_settings, Mapping):
+            return {}
+        dnsbl = {}
+        for attr in ["enabled", "safesearch", "nxdomain", "address"]:
+            dnsbl[attr] = dnsbl_settings.get(attr, "")
+        for attr in ["type", "lists", "whitelists", "blocklists", "wildcards"]:
+            if isinstance(dnsbl_settings[attr], Mapping):
+                dnsbl[attr] = ",".join(
+                    [
+                        key
+                        for key, value in dnsbl_settings.get(attr, {}).items()
+                        if isinstance(value, Mapping) and value.get("selected", 0) == 1
+                    ]
+                )
+            else:
+                dnsbl[attr] = ""
+        _LOGGER.debug(f"[get_unbound_blocklist] dnsbl: {dnsbl}")
+        return dnsbl
+
+    async def _set_unbound_blocklist(self, set_state: bool) -> bool:
+        payload = {}
+        payload["unbound"] = {}
+        payload["unbound"]["dnsbl"] = await self.get_unbound_blocklist()
+        if set_state:
+            payload["unbound"]["dnsbl"]["enabled"] = "1"
+        else:
+            payload["unbound"]["dnsbl"]["enabled"] = "0"
+        response: Mapping[str, Any] | list = await self._post(
+            "/api/unbound/settings/set", payload=payload
+        )
+        _LOGGER.debug(
+            f"[set_unbound_blocklist] set_state: {'On' if set_state else 'Off'}, payload: {payload}, response: {response}"
+        )
+        return not (
+            response is None
+            or not isinstance(response, Mapping)
+            or response.get("result", "failed") != "saved"
+        )
+
+    @_log_errors
+    async def enable_unbound_blocklist(self) -> bool:
+        return await self._set_unbound_blocklist(set_state=True)
+
+    @_log_errors
+    async def disable_unbound_blocklist(self) -> bool:
+        return await self._set_unbound_blocklist(set_state=False)

custom_components/opnsense/switch.py

@@ -18,7 +18,7 @@
 from custom_components.opnsense.pyopnsense import OPNsenseClient
 
 from . import CoordinatorEntityManager, OPNsenseEntity
-from .const import COORDINATOR, DOMAIN
+from .const import ATTR_UNBOUND_BLOCKLIST, COORDINATOR, DOMAIN
 from .coordinator import OPNsenseDataUpdateCoordinator
 from .helpers import dict_get
 
@@ -192,6 +192,21 @@ def process_entities_callback(hass, config_entry):
                     ),
                 )
                 entities.append(entity)
+
+        entity = OPNsenseUnboundBlocklistSwitch(
+            config_entry,
+            coordinator,
+            SwitchEntityDescription(
+                key=f"unbound_blocklist.switch",
+                name=f"Unbound Blocklist Switch",
+                # icon=icon,
+                # entity_category=ENTITY_CATEGORY_CONFIG,
+                device_class=SwitchDeviceClass.SWITCH,
+                entity_registry_enabled_default=False,
+            ),
+        )
+        entities.append(entity)
+
         return entities
 
     cem = CoordinatorEntityManager(
@@ -220,13 +235,13 @@ def __init__(
             f"{self.opnsense_device_unique_id}_{entity_description.key}"
         )
 
-    @property
-    def is_on(self):
-        return False
+    # @property
+    # def is_on(self):
+    #     return False
 
-    @property
-    def extra_state_attributes(self):
-        return None
+    # @property
+    # def extra_state_attributes(self):
+    #     return None
 
 
 class OPNsenseFilterSwitch(OPNsenseSwitch):
@@ -421,3 +436,58 @@ def extra_state_attributes(self) -> Mapping[str, Any]:
         for attr in ["id", "name"]:
             attributes[f"service_{attr}"] = service.get(attr, None)
         return attributes
+
+
+class OPNsenseUnboundBlocklistSwitch(OPNsenseSwitch):
+
+    def __init__(
+        self,
+        config_entry,
+        coordinator: OPNsenseDataUpdateCoordinator,
+        entity_description: SwitchEntityDescription,
+    ) -> None:
+        super().__init__(
+            config_entry=config_entry,
+            coordinator=coordinator,
+            entity_description=entity_description,
+        )
+        self._attr_is_on = STATE_UNKNOWN
+        self._attr_extra_state_attributes = {}
+        self._attr_available = False
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        dnsbl = self.coordinator.data.get(ATTR_UNBOUND_BLOCKLIST, {})
+        if not isinstance(dnsbl, Mapping) or len(dnsbl) == 0:
+            self._attr_available = False
+            return
+        self._attr_available = True
+        self._attr_is_on = True if dnsbl.get("enabled", "0") == "1" else False
+        self._attr_extra_state_attributes = {
+            "Force SafeSearch": (
+                True if dnsbl.get("safesearch", "0") == "1" else False
+            ),
+            "Type of DNSBL": dnsbl.get("type", ""),
+            "URLs of Blocklists": dnsbl.get("lists", ""),
+            "Whitelist Domains": dnsbl.get("whitelists", ""),
+            "Blocklist Domains": dnsbl.get("blocklists", ""),
+            "Wildcard Domains": dnsbl.get("wildcards", ""),
+            "Destination Address": dnsbl.get("address", ""),
+            "Return NXDOMAIN": (True if dnsbl.get("nxdomain", "0") == "1" else False),
+        }
+        self.async_write_ha_state()
+
+    async def async_turn_on(self, **kwargs) -> None:
+        """Turn the entity on."""
+        client: OPNsenseClient = self._get_opnsense_client()
+        result: bool = await client.enable_unbound_blocklist()
+        if result:
+            await self.coordinator.async_refresh()
+
+    async def async_turn_off(self, **kwargs) -> None:
+        """Turn the entity off."""
+        client: OPNsenseClient = self._get_opnsense_client()
+        result: bool = await client.disable_unbound_blocklist()
+        if result:
+            await self.coordinator.async_refresh()