feat(diagnostics): read-only LAN discovery scan to seed LAN-API work (#57) (#112)

Native LAN control (issue #57) is a large, hardware-dependent feature and the
maintainer has no LAN devices to test against. The agreed first step is the
diagnostics-collection helper: capture which of a user's devices answer Govee's
local UDP discovery, so the community can supply the data the full feature needs.

- api/lan.py: async_scan_lan_devices() — one bounded multicast 'scan' to
  239.255.255.250:4001, collects responses on :4002 for a short timeout, returns
  the deduped per-device discovery fields (ip, device, sku, fw versions). Uses
  asyncio datagram endpoints (event-loop safe). Read-only: no control commands,
  no entities, no persistent socket.
- diagnostics.py: fold a 'lan_discovery' block into the entry diagnostics
  download, defensively (a scan error never breaks the download). Add 'ip' to
  TO_REDACT so a responder's private IP isn't leaked in a publicly-attached dump
  (the device MAC is already redacted; SKU + firmware are kept as the signal).

This touches no control path and adds no entity/config surface. A 'lan'
TransportKind, local control, and per-device merging wait until captures confirm
the protocol per-SKU (the point of this increment).

Adds tests/test_lan.py (scan request shape, parse, dedupe, empty, malformed,
bind failure — all socket-mocked) and LAN diagnostics tests (block present, IP +
MAC redacted, scan-failure isolated). Existing diagnostics tests stub the scan.
Refs #57.

Co-authored-by: Tom Lasswell <tom@calcyon.com>

Author: lasswellt

custom_components/govee/api/lan.py

@@ -0,0 +1,112 @@
+"""Read-only Govee LAN (UDP) discovery helper.
+
+Govee exposes a local UDP control API (must be toggled on per device in the
+Govee app) on a subset of mostly-light SKUs. This module performs ONLY the
+discovery half — a single bounded multicast ``scan`` — so a user can capture
+which of their devices answer on the LAN and what they report, and attach it to
+a diagnostics download. That community data is the prerequisite for the full
+LAN transport requested in issue #57 (the maintainer has no LAN hardware to
+test against).
+
+Deliberately scoped: no control commands, no entities, no persistent socket —
+one scan, collect responses for a short timeout, return them. Protocol per
+``docs/govee-protocol-reference.md`` §6:
+
+- Scan request  -> 239.255.255.250:4001  ``{"msg":{"cmd":"scan",...}}``
+- Scan response -> client:4002            ``{"msg":{"cmd":"scan","data":{...}}}``
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import socket
+from typing import Any
+
+_LOGGER = logging.getLogger(__name__)
+
+# Govee LAN API network parameters (docs/govee-protocol-reference.md §6).
+LAN_MULTICAST_GROUP = "239.255.255.250"
+LAN_DISCOVERY_PORT = 4001  # devices listen here for the scan request
+LAN_RESPONSE_PORT = 4002  # we listen here for scan responses
+
+_SCAN_REQUEST = json.dumps(
+    {"msg": {"cmd": "scan", "data": {"account_topic": "reserve"}}}
+).encode("utf-8")
+
+# Fields a scan response may carry; we surface exactly these (no control data).
+_RESPONSE_FIELDS = (
+    "ip",
+    "device",
+    "sku",
+    "bleVersionHard",
+    "bleVersionSoft",
+    "wifiVersionHard",
+    "wifiVersionSoft",
+)
+
+
+class _ScanProtocol(asyncio.DatagramProtocol):
+    """Collects well-formed Govee ``scan`` responses; ignores everything else."""
+
+    def __init__(self) -> None:
+        self.responses: dict[str, dict[str, Any]] = {}
+
+    def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
+        try:
+            payload = json.loads(data.decode("utf-8", errors="replace"))
+            msg = payload.get("msg", {})
+            if msg.get("cmd") != "scan":
+                return
+            body = msg.get("data", {})
+            if not isinstance(body, dict):
+                return
+        except (ValueError, AttributeError):
+            return
+
+        record = {field: body[field] for field in _RESPONSE_FIELDS if field in body}
+        if not record:
+            return
+        # Dedupe by the device id (MAC) when present, else by source IP.
+        key = str(record.get("device") or addr[0])
+        self.responses[key] = record
+
+    def error_received(self, exc: Exception) -> None:  # pragma: no cover - rare
+        _LOGGER.debug("LAN scan socket error: %s", exc)
+
+
+async def async_scan_lan_devices(timeout: float = 2.0) -> list[dict[str, Any]]:
+    """Send one multicast ``scan`` and collect responses for ``timeout`` seconds.
+
+    Returns a list of per-device dicts (deduped) limited to the discovery fields
+    above — no control surface is touched. Uses asyncio datagram endpoints so it
+    is safe to run on the Home Assistant event loop.
+
+    Raises ``OSError`` if the response socket cannot be bound (e.g. port 4002 in
+    use by another local-control app); callers should treat that as "no data".
+    """
+    loop = asyncio.get_running_loop()
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    # Multicast send TTL of 2 so a scan can cross one router hop if present.
+    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
+    sock.setblocking(False)
+    try:
+        sock.bind(("", LAN_RESPONSE_PORT))
+    except OSError:
+        sock.close()
+        raise
+
+    transport, protocol = await loop.create_datagram_endpoint(
+        _ScanProtocol, sock=sock
+    )
+    assert isinstance(protocol, _ScanProtocol)
+    try:
+        transport.sendto(_SCAN_REQUEST, (LAN_MULTICAST_GROUP, LAN_DISCOVERY_PORT))
+        await asyncio.sleep(timeout)
+    finally:
+        transport.close()
+
+    return list(protocol.responses.values())

custom_components/govee/diagnostics.py

@@ -18,6 +18,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.device_registry import DeviceEntry
 
+from .api.lan import async_scan_lan_devices
 from .const import CONF_API_KEY, CONF_EMAIL, CONF_PASSWORD, DOMAIN
 from .coordinator import GoveeCoordinator
 from .models.transport import TRANSPORT_KINDS
@@ -42,6 +43,9 @@
     "deviceName",
     "hub_device_id",
     "mac",
+    # Local network address from the LAN-discovery scan (#57) — a private IP is
+    # still PII in a publicly-attached diagnostics download.
+    "ip",
 }
 
 # Govee device IDs are MAC-derived: 8 colon-separated hex octets
@@ -200,6 +204,31 @@ def _runtime_diag(coordinator: GoveeCoordinator) -> dict[str, Any]:
     }
 
 
+async def _lan_discovery_diag() -> dict[str, Any]:
+    """Run one read-only LAN scan for the diagnostics download (issue #57).
+
+    Captures which of the user's devices answer Govee's local UDP discovery and
+    what they report, so the community can supply the data the full LAN feature
+    needs. Never raises — diagnostics must always produce output; the IP of each
+    responder is redacted by the shared ``_redact`` pass.
+    """
+    try:
+        devices = await async_scan_lan_devices()
+        return {
+            "scan_attempted": True,
+            "device_count": len(devices),
+            "devices": devices,
+            "error": None,
+        }
+    except Exception as err:  # never break the diagnostics download
+        return {
+            "scan_attempted": True,
+            "device_count": 0,
+            "devices": [],
+            "error": str(err),
+        }
+
+
 def _redact(data: dict[str, Any]) -> dict[str, Any]:
     """Redact sensitive keys, then hash any MAC-format device-map keys."""
     redacted: dict[str, Any] = async_redact_data(data, TO_REDACT)
@@ -235,6 +264,8 @@ async def async_get_config_entry_diagnostics(
         # Verbatim device-list response from the most recent discovery poll.
         "raw_api_devices": coordinator.api_client.last_raw_devices,
         "leak_sensors": _leak_diag(coordinator),
+        # Read-only local-network scan to seed the LAN-API work (issue #57).
+        "lan_discovery": await _lan_discovery_diag(),
         **_runtime_diag(coordinator),
     }
     return _redact(diagnostics_data)

tests/test_diagnostics.py

@@ -4,7 +4,7 @@
 
 import json
 import re
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -24,6 +24,20 @@
 from custom_components.govee.models.transport import TransportHealth
 
 
+@pytest.fixture(autouse=True)
+def _stub_lan_scan(monkeypatch):
+    """Stub the LAN scan so entry-diagnostics tests stay fast + offline (#57).
+
+    async_get_config_entry_diagnostics now runs a real UDP discovery scan;
+    default it to "no devices" so the existing tests don't bind a socket or
+    wait on the scan timeout. Tests that exercise the LAN block override this.
+    """
+    monkeypatch.setattr(
+        "custom_components.govee.diagnostics.async_scan_lan_devices",
+        AsyncMock(return_value=[]),
+    )
+
+
 def _coordinator_stub(**overrides):
     """A coordinator MagicMock with all diagnostics accessors defaulted sanely."""
     coordinator = MagicMock()
@@ -447,3 +461,57 @@ async def test_leak_hub_device_includes_its_sensors(self) -> None:
         assert hub_mac not in rendered
         assert sensor_mac not in rendered
         assert _MAC_RE.search(rendered) is None
+
+
+class TestLanDiscoveryDiag:
+    """Entry diagnostics include a read-only LAN scan, with IP redacted (#57)."""
+
+    @pytest.mark.asyncio
+    async def test_lan_block_present_and_ip_redacted(self, monkeypatch) -> None:
+        monkeypatch.setattr(
+            "custom_components.govee.diagnostics.async_scan_lan_devices",
+            AsyncMock(
+                return_value=[
+                    {
+                        "ip": "192.168.1.23",
+                        "device": "1F:80:C5:32:32:36:72:4E",
+                        "sku": "H6072",
+                        "wifiVersionSoft": "1.02.03",
+                    }
+                ]
+            ),
+        )
+        coordinator = _coordinator_stub()
+        out = await async_get_config_entry_diagnostics(
+            MagicMock(), _entry_stub(coordinator)
+        )
+
+        lan = out["lan_discovery"]
+        assert lan["scan_attempted"] is True
+        assert lan["device_count"] == 1
+        device = lan["devices"][0]
+        # SKU + firmware are kept (the useful signal); IP + MAC are redacted.
+        assert device["sku"] == "H6072"
+        assert device["wifiVersionSoft"] == "1.02.03"
+        assert device["ip"] == "**REDACTED**"
+        assert device["device"] == "**REDACTED**"
+        rendered = json.dumps(out, default=str)
+        assert "192.168.1.23" not in rendered
+        assert "1F:80:C5:32:32:36:72:4E" not in rendered
+
+    @pytest.mark.asyncio
+    async def test_lan_scan_failure_is_isolated(self, monkeypatch) -> None:
+        # A scan error must not break the diagnostics download.
+        monkeypatch.setattr(
+            "custom_components.govee.diagnostics.async_scan_lan_devices",
+            AsyncMock(side_effect=OSError("port 4002 in use")),
+        )
+        coordinator = _coordinator_stub()
+        out = await async_get_config_entry_diagnostics(
+            MagicMock(), _entry_stub(coordinator)
+        )
+
+        lan = out["lan_discovery"]
+        assert lan["scan_attempted"] is True
+        assert lan["device_count"] == 0
+        assert "port 4002 in use" in lan["error"]

tests/test_lan.py

@@ -0,0 +1,122 @@
+"""Tests for the read-only Govee LAN discovery helper (issue #57)."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+from custom_components.govee.api import lan
+
+
+def _install_fake_endpoint(monkeypatch, responses, *, bind_error=None):
+    """Patch socket + event loop so async_scan_lan_devices runs offline.
+
+    Returns the MagicMock transport so tests can assert on sendto. ``responses``
+    are fed to the real _ScanProtocol during the (stubbed) scan wait.
+    """
+    sock = MagicMock()
+    if bind_error is not None:
+        sock.bind.side_effect = bind_error
+    monkeypatch.setattr(lan.socket, "socket", lambda *a, **k: sock)
+
+    proto = lan._ScanProtocol()
+    transport = MagicMock()
+
+    async def _create_endpoint(_factory, sock=None):
+        return transport, proto
+
+    mock_loop = MagicMock()
+    mock_loop.create_datagram_endpoint = _create_endpoint
+    monkeypatch.setattr(lan.asyncio, "get_running_loop", lambda: mock_loop)
+
+    async def _sleep(_timeout):
+        for resp in responses:
+            proto.datagram_received(
+                json.dumps(resp).encode("utf-8"), ("192.168.1.50", 4002)
+            )
+
+    monkeypatch.setattr(lan.asyncio, "sleep", _sleep)
+    return transport, sock
+
+
+def _scan_response(**data):
+    return {"msg": {"cmd": "scan", "data": data}}
+
+
+@pytest.mark.asyncio
+async def test_sends_scan_request_to_multicast(monkeypatch):
+    transport, _ = _install_fake_endpoint(monkeypatch, [])
+
+    await lan.async_scan_lan_devices(timeout=0.01)
+
+    transport.sendto.assert_called_once()
+    payload, addr = transport.sendto.call_args[0]
+    assert addr == (lan.LAN_MULTICAST_GROUP, lan.LAN_DISCOVERY_PORT)
+    sent = json.loads(payload.decode())
+    assert sent == {"msg": {"cmd": "scan", "data": {"account_topic": "reserve"}}}
+
+
+@pytest.mark.asyncio
+async def test_parses_and_returns_devices(monkeypatch):
+    responses = [
+        _scan_response(
+            ip="192.168.1.23",
+            device="1F:80:C5:32:32:36:72:4E",
+            sku="H6072",
+            wifiVersionSoft="1.02.03",
+        )
+    ]
+    _install_fake_endpoint(monkeypatch, responses)
+
+    devices = await lan.async_scan_lan_devices(timeout=0.01)
+
+    assert len(devices) == 1
+    assert devices[0]["sku"] == "H6072"
+    assert devices[0]["ip"] == "192.168.1.23"
+    assert devices[0]["device"] == "1F:80:C5:32:32:36:72:4E"
+    assert devices[0]["wifiVersionSoft"] == "1.02.03"
+
+
+@pytest.mark.asyncio
+async def test_dedupes_by_device_id(monkeypatch):
+    dup = _scan_response(ip="192.168.1.23", device="AA:BB:CC:DD:EE:FF:00:11", sku="H6072")
+    _install_fake_endpoint(monkeypatch, [dup, dup])
+
+    devices = await lan.async_scan_lan_devices(timeout=0.01)
+
+    assert len(devices) == 1
+
+
+@pytest.mark.asyncio
+async def test_no_responses_returns_empty(monkeypatch):
+    _install_fake_endpoint(monkeypatch, [])
+
+    devices = await lan.async_scan_lan_devices(timeout=0.01)
+
+    assert devices == []
+
+
+@pytest.mark.asyncio
+async def test_ignores_malformed_and_non_scan(monkeypatch):
+    proto = lan._ScanProtocol()
+    # Garbage bytes.
+    proto.datagram_received(b"not-json", ("192.168.1.9", 4002))
+    # Valid JSON but a different command.
+    proto.datagram_received(
+        json.dumps({"msg": {"cmd": "devStatus", "data": {"onOff": 1}}}).encode(),
+        ("192.168.1.9", 4002),
+    )
+    # scan but empty data -> no usable record.
+    proto.datagram_received(json.dumps(_scan_response()).encode(), ("192.168.1.9", 4002))
+
+    assert proto.responses == {}
+
+
+@pytest.mark.asyncio
+async def test_bind_failure_raises_oserror(monkeypatch):
+    _install_fake_endpoint(monkeypatch, [], bind_error=OSError("port in use"))
+
+    with pytest.raises(OSError):
+        await lan.async_scan_lan_devices(timeout=0.01)