Add coverage-gap detection: ignored-hrefs registry, Repairs issue, redaction

Implements the first half of the device-capability-diagnostics spec:

- registry/capabilities/ignored.py: known-noise hrefs (Bixby/voice
  provisioning, WiFi/BLE info, OTA/region housekeeping, and a couple of
  redundant hrefs) declared as no-entity Capability objects, so
  discover()'s existing unknown-resource reporting treats them as covered
  instead of flagging every device as having gaps. Folded into each
  by_type registry and into the global fallback CAPABILITIES set.
- discovery.py: log() callback now passes the raw href instead of a
  formatted message, so callers can collect a clean unbound-hrefs list.
- coordinator.py: wires that callback into self._unbound_hrefs, tracks
  device_type_name/one_ui_version, and raises (or clears) a Repairs issue
  when a device's type is unrecognized or it has genuinely unmodeled
  hrefs left over.
- registry/redact.py: recursive, substring-keyed redaction for anything
  that looks like account/identity data, ahead of the diagnostics.py
  platform that will consume it.

Verified against the real dishwasher/refrigerator fixtures: known-noise
hrefs no longer show up as gaps, while genuine gaps (e.g. /bespoke/vs/0
on the fridge) still do.
This commit is contained in:
Marc Billow
2026-07-07 23:49:58 -05:00
parent b67e0e5cf5
commit 35b47d6b5a
14 changed files with 383 additions and 11 deletions
+4
View File
@@ -12,3 +12,7 @@ CONF_LEAF_KEY_PEM = "leaf_key_pem"
PROBE_PORTS = [49154, 49155]
SUMMARY_INTERVAL_S = 30.0
DEVICE_SUPPORT_ISSUE_URL = (
"https://github.com/mbillow/localthings/issues/new?template=device-support.yml"
)
+38 -3
View File
@@ -10,6 +10,7 @@ import cbor2
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers.device_registry import DeviceInfo
@@ -24,7 +25,7 @@ from .registry.identity import read_identity, DeviceIdentity
from .const import (
DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM,
SUMMARY_INTERVAL_S,
DEVICE_SUPPORT_ISSUE_URL, SUMMARY_INTERVAL_S,
)
_LOGGER = logging.getLogger(__name__)
@@ -62,6 +63,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._subpoll_task: asyncio.Task | None = None
self._hot_hrefs: list[str] = []
self._warm_hrefs: list[str] = []
self.device_type_name: str | None = None
self.one_ui_version: str = ''
self._unbound_hrefs: list[str] = []
# ------------------------------------------------------------------
# Session management (all blocking — must run in executor)
@@ -183,14 +187,20 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
one_ui = (resources.get('/otninformation/vs/0', {})
.get('swVersionInfo', {})
.get('oneUiVersion', ''))
self.one_ui_version = one_ui
reg = for_device(one_ui) if one_ui else None
unbound: list[str] = []
if reg is not None:
_LOGGER.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui)
bound = discover(resources, reg.capabilities, reg.pattern_capabilities)
bound = discover(resources, reg.capabilities, reg.pattern_capabilities,
log=unbound.append)
self.device_type_name = reg.name
else:
_LOGGER.warning("unknown device type oneUiVersion=%r; using common caps", one_ui)
bound = discover(resources, CAPABILITIES)
bound = discover(resources, CAPABILITIES, log=unbound.append)
self.device_type_name = None
self.bound = bound
self._unbound_hrefs = unbound
serial = (resources.get('/information/vs/0', {})
.get('x.com.samsung.da.serialNum', ''))
@@ -209,6 +219,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
manufacturer=mfr,
model=model,
)
self._update_coverage_gap_issue(reg is None, unbound, name)
hot, warm = set(), set()
for be in bound:
tier = be.capability.poll_tier
@@ -225,6 +237,29 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
len(bound), serial, self._hot_hrefs, self._warm_hrefs,
)
def _update_coverage_gap_issue(
self, unknown_type: bool, unbound_hrefs: list[str], device_name: str,
) -> None:
"""Raise or clear a Repairs issue when capability coverage is incomplete.
Fires once, at discovery time, either because the device type itself
wasn't recognized or because some of its resources didn't bind to
any capability. Diagnostics (diagnostics.py) is what a user actually
downloads to help; this just tells them there's something to send.
"""
issue_id = f"device_gap_{self._entry.entry_id}"
if unknown_type or unbound_hrefs:
ir.async_create_issue(
self.hass, DOMAIN, issue_id,
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="device_gap",
translation_placeholders={"device_name": device_name},
learn_more_url=DEVICE_SUPPORT_ISSUE_URL,
)
else:
ir.async_delete_issue(self.hass, DOMAIN, issue_id)
# ------------------------------------------------------------------
# DataUpdateCoordinator hook
# ------------------------------------------------------------------
@@ -1,10 +1,11 @@
"""Dishwasher device registry."""
from ..capabilities import common, dishwasher, fridge, laundry, operational
from ..capabilities import common, dishwasher, fridge, ignored, laundry, operational
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='dishwasher',
capabilities=_build([
*ignored.IGNORED,
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
@@ -3,12 +3,13 @@
Note: dryer devices use port 49155 (not 49154). The config flow handles
this automatically via port probing.
"""
from ..capabilities import common, dryer, fridge, laundry, operational
from ..capabilities import common, dryer, fridge, ignored, laundry, operational
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='dryer',
capabilities=_build([
*ignored.IGNORED,
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
@@ -1,10 +1,11 @@
"""Oven device registry."""
from ..capabilities import common, fridge, oven
from ..capabilities import common, fridge, ignored, oven
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='oven',
capabilities=_build([
*ignored.IGNORED,
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
@@ -1,10 +1,11 @@
"""Refrigerator device registry."""
from ..capabilities import common, fridge
from ..capabilities import common, fridge, ignored
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='refrigerator',
capabilities=_build([
*ignored.IGNORED,
common.POWER,
fridge.STATUS_LOCK,
fridge.DOOR_ALERT,
@@ -1,4 +1,4 @@
from . import common, dishwasher, fridge, laundry, operational, oven
from . import common, dishwasher, fridge, ignored, laundry, operational, oven
from ..capability import Capability
@@ -16,4 +16,4 @@ _OVEN_GLOBAL_CAPS = [
]
ALL = [v for mod in (common, operational, laundry, fridge)
for v in vars(mod).values() if _is_capability(v)] + _OVEN_GLOBAL_CAPS
for v in vars(mod).values() if _is_capability(v)] + _OVEN_GLOBAL_CAPS + ignored.IGNORED
@@ -0,0 +1,67 @@
"""Hrefs deliberately not modeled as capabilities.
Every entry here is either irrelevant plumbing (network/account/OTA
housekeeping that will never be home-automation-relevant) or a duplicate of
state we already expose through a friendlier href. Declaring them here,
rather than leaving them unregistered, keeps discover()'s unknown-resource
reporting limited to genuine coverage gaps.
Capability defaults entities=() and discover() treats any href that
resolves to a capability at all as matched, even one with no entities, so
this list uses the same declaration shape as a real capability with no
extra machinery. Fold into by_type registries (per-family gap detection)
and into the global ALL list (unknown-device-type fallback).
Deliberately exact hrefs only, no href_prefix/pattern capabilities: the
unknown-device-type fallback path (coordinator's `discover(resources,
CAPABILITIES)` call) doesn't pass pattern_caps, so a prefix-based entry
here would silently do nothing on that path. Enumerate each known href
instead; it's a short, stable list.
This list is maintainer-curated only; there is no per-installation
override. Grow it as real /device/0 dumps surface more universal noise —
do not add a href here on a guess. If a href's relevance is unclear, leave
it unbound so it surfaces as a gap for a human to look at.
"""
from ..capability import Capability
IGNORED: list[Capability] = [
# Device serial/model is read directly by the coordinator for HA device
# identity, not modeled as an entity capability.
Capability(href='/information/vs/0'),
# Bixby voice assistant: feature negotiation, account provisioning
# (Samsung account email, access tokens), terms-of-service state, and
# enable/disable status.
Capability(href='/voice/feature/vs/0'),
Capability(href='/voice/provisioning/vs/0'),
Capability(href='/bixby/vs/0'),
Capability(href='/bixby/status/vs/0'),
Capability(href='/bixbyuservalidate/vs/0'),
Capability(href='/bixbyterms/vs/0'),
# Network/WiFi housekeeping — MAC addresses, supported auth/crypto
# types, no controllable or observable appliance state.
Capability(href='/wirelessinfo/vs/0'),
Capability(href='/connectionconfig/vs/0'),
# Static or internal-protocol metadata, not entity-worthy.
Capability(href='/quickcontrol/info/vs/0'),
Capability(href='/realtimenotiforclient/vs/0'),
Capability(href='/file/information/vs/0'),
Capability(href='/configuration/vs/0'), # region/countryCode
Capability(href='/setting/vs/0'), # supported/selected UI language
Capability(href='/timezone/vs/0'), # redundant with HA's own timezone
Capability(href='/wm/setinfo/vs/0'), # model/manufacturing metadata
# Demand Response Load Control — utility-company grid signals; requires
# cloud registration with a utility program we don't support locally.
Capability(href='/drlc/vs/0'),
# Redundant with capabilities already declared elsewhere.
# /speakersound/vs/0 duplicates /settings/sound/volume/vs/0 (laundry.SOUND_VOLUME).
Capability(href='/speakersound/vs/0'),
# /wm/editcourse/vs/0 is the raw encoded course-list byte string already
# decoded and exposed via /course/vs/0 (dishwasher.CYCLE_OPTIONS).
Capability(href='/wm/editcourse/vs/0'),
]
@@ -2,7 +2,8 @@
A device advertises a set of OCF resources keyed by href. For every href
present in the registry, emit the capability's entities bound to that resource.
Unknown hrefs are a coverage gap, logged at debug and skipped.
Unknown hrefs are a coverage gap; each one is passed to the optional `log`
callback (as the raw href, not a formatted message) and otherwise skipped.
"""
from __future__ import annotations
@@ -79,6 +80,6 @@ def discover(
break
if not matched and log is not None:
log(f"unknown resource {href}")
log(href)
return out
@@ -0,0 +1,41 @@
"""Redact account/identity data from a raw resource tree before it leaves
the user's Home Assistant instance (diagnostics downloads, issue reports).
/device/0 dumps mix appliance state with genuinely sensitive data when
Bixby/voice is set up on the device: a Samsung account email, a Bixby
access token, a hashed device ID, WiFi/BLE MAC addresses, the serial
number, and otnDUID. This walks the whole tree and redacts any value whose
key matches a known-sensitive substring, regardless of which href it's
under — new device types will have unknown-shaped data we can't fully
enumerate in advance, so this errs on catching the field by name rather
than only redacting inside hrefs we already recognize.
"""
from __future__ import annotations
REDACTED = "**REDACTED**"
_SENSITIVE_SUBSTRINGS = (
'mac', 'serial', 'token', 'login', 'account', 'email',
'userid', 'deviceid', 'uuid', 'duid', 'password', 'secret',
)
def _is_sensitive_key(key: str) -> bool:
lowered = key.lower()
return any(s in lowered for s in _SENSITIVE_SUBSTRINGS)
def redact_resources(resources):
"""Recursively redact dict values whose key matches a sensitive substring.
Works on the shape produced by parse_device0_batch (dict[href, rep]) or
any nested dict/list structure within a rep.
"""
if isinstance(resources, dict):
return {
key: (REDACTED if _is_sensitive_key(key) else redact_resources(value))
for key, value in resources.items()
}
if isinstance(resources, list):
return [redact_resources(item) for item in resources]
return resources
@@ -73,5 +73,11 @@
"abort": {
"already_configured": "This device is already configured."
}
},
"issues": {
"device_gap": {
"title": "Incomplete capability coverage for {device_name}",
"description": "This device is missing full capability coverage. Either its appliance type wasn't recognized, or some of the resources it exposes aren't modeled yet. It'll keep working with whatever is already supported. You can help expand support by going to Settings > Devices & Services > {device_name} > the menu (top right) > Download diagnostics, then filing it with the linked issue template."
}
}
}
+82
View File
@@ -6,6 +6,7 @@ from unittest.mock import patch
import pytest
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from custom_components.localthings.const import (
DOMAIN, SUMMARY_INTERVAL_S,
@@ -72,3 +73,84 @@ async def test_update_failed_on_persistent_poll_error(
# Setup fails — entry should not be in hass.data
assert not result or mock_entry.entry_id not in hass.data.get(DOMAIN, {})
# ---------------------------------------------------------------------------
# Coverage-gap detection and the Repairs issue it raises
# ---------------------------------------------------------------------------
async def test_discovery_populates_device_type_and_unbound_hrefs(
hass: HomeAssistant, mock_entry, mock_coordinator_session
) -> None:
"""The real refrigerator fixture is a recognized type with some gaps left."""
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
assert coordinator.device_type_name == 'refrigerator'
assert coordinator.one_ui_version == '7.0 Refrigerator'
# Real, currently-unmodeled hrefs on this fixture (not noise — see
# tests/test_registry_ignored.py for the noise-suppression checks).
assert '/bespoke/vs/0' in coordinator._unbound_hrefs
async def test_coverage_gap_issue_created_for_real_fixture_gaps(
hass: HomeAssistant, mock_entry, mock_coordinator_session
) -> None:
"""A recognized device with leftover unbound hrefs still raises the issue."""
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
issue_id = f"device_gap_{mock_entry.entry_id}"
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is not None
def test_update_coverage_gap_issue_creates_issue_for_unknown_type(
hass: HomeAssistant, mock_entry
) -> None:
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._update_coverage_gap_issue(True, [], 'Test Appliance')
issue_id = f"device_gap_{mock_entry.entry_id}"
issue = ir.async_get(hass).async_get_issue(DOMAIN, issue_id)
assert issue is not None
assert issue.translation_key == 'device_gap'
assert issue.translation_placeholders == {'device_name': 'Test Appliance'}
assert issue.is_fixable is False
def test_update_coverage_gap_issue_creates_issue_for_unbound_hrefs(
hass: HomeAssistant, mock_entry
) -> None:
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._update_coverage_gap_issue(False, ['/mystery/vs/0'], 'Test Appliance')
issue_id = f"device_gap_{mock_entry.entry_id}"
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is not None
def test_update_coverage_gap_issue_absent_when_fully_covered(
hass: HomeAssistant, mock_entry
) -> None:
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._update_coverage_gap_issue(False, [], 'Test Appliance')
issue_id = f"device_gap_{mock_entry.entry_id}"
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None
def test_update_coverage_gap_issue_clears_previous_issue(
hass: HomeAssistant, mock_entry
) -> None:
"""A later discovery run with no gap deletes an issue raised earlier."""
coordinator = LocalThingsCoordinator(hass, mock_entry)
issue_id = f"device_gap_{mock_entry.entry_id}"
coordinator._update_coverage_gap_issue(True, [], 'Test Appliance')
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is not None
coordinator._update_coverage_gap_issue(False, [], 'Test Appliance')
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None
+74
View File
@@ -0,0 +1,74 @@
"""Tests for registry.redact — the safety net for diagnostics downloads."""
import json
from pathlib import Path
from custom_components.localthings.registry.batch import parse_device0_batch
from custom_components.localthings.registry.redact import REDACTED, redact_resources
FIXTURES = Path(__file__).resolve().parent / 'fixtures'
def _load(name: str) -> dict:
data = json.loads((FIXTURES / name).read_text())
return parse_device0_batch(data['device0'])
def test_redacts_known_sensitive_fields_in_dishwasher_dump():
resources = _load('dishwasher_device.json')
redacted = redact_resources(resources)
info = redacted['/information/vs/0']
assert info['x.com.samsung.da.serialNum'] == REDACTED
assert info['x.com.samsung.da.otnDUID'] == REDACTED
wireless = redacted['/wirelessinfo/vs/0']
assert wireless['macaddressWiFi'] == REDACTED
assert wireless['macaddressBLE'] == REDACTED
provisioning = redacted['/voice/provisioning/vs/0']
headers = provisioning['voice.provisioning.headers']
assert headers['login_id'] == REDACTED
deviceinfo = provisioning['voice.provisioning.deviceinfo']
assert deviceinfo['voice.provisioning.deviceinfo.accesstoken'] == REDACTED
assert deviceinfo['voice.provisioning.deviceinfo.deviceid'] == REDACTED
assert deviceinfo['voice.provisioning.deviceinfo.userid'] == REDACTED
def test_ordinary_state_fields_survive_untouched():
resources = _load('dishwasher_device.json')
redacted = redact_resources(resources)
op_state = redacted['/operational/state/vs/0']
assert op_state['x.com.samsung.da.state'] == 'Run'
assert op_state['x.com.samsung.da.progress'] == 'Finish'
power = redacted['/power/vs/0']
assert power['x.com.samsung.da.power'] == 'On'
dishwasher = redacted['/dishwasher/vs/0']
assert dishwasher['x.com.samsung.da.sanitize'] == 'On'
assert dishwasher['x.com.samsung.da.rinseLevel'] == '4'
alarms = redacted['/alarms/vs/0']['x.com.samsung.da.items']
assert alarms[0]['x.com.samsung.da.code'] == 'SNSF_Reached'
def test_redacts_known_sensitive_fields_in_refrigerator_dump():
resources = _load('refrigerator_device.json')
redacted = redact_resources(resources)
info = redacted['/information/vs/0']
assert info['x.com.samsung.da.serialNum'] == REDACTED
wireless = redacted['/wirelessinfo/vs/0']
assert wireless['macaddressWiFi'] == REDACTED
assert wireless['macaddressBLE'] == REDACTED
def test_redact_resources_does_not_mutate_input():
resources = _load('dishwasher_device.json')
original_serial = resources['/information/vs/0']['x.com.samsung.da.serialNum']
redact_resources(resources)
assert resources['/information/vs/0']['x.com.samsung.da.serialNum'] == original_serial
+58
View File
@@ -0,0 +1,58 @@
"""Tests that known-noise hrefs don't show up as coverage gaps.
Each device-type registry folds in registry.capabilities.ignored.IGNORED so
that Bixby/voice/network/OTA housekeeping resources aren't reported as
unbound. This guards two things at once: that the ignored hrefs are
actually suppressed, and (via a plain import) that no ignored href
collides with a real capability declared elsewhere for the same family.
"""
import json
from pathlib import Path
from custom_components.localthings.registry.batch import parse_device0_batch
from custom_components.localthings.registry.by_type import dishwasher, refrigerator
from custom_components.localthings.registry.capabilities.ignored import IGNORED
FIXTURES = Path(__file__).resolve().parent / 'fixtures'
_IGNORED_HREFS = {cap.href for cap in IGNORED}
def _load(name: str) -> dict:
data = json.loads((FIXTURES / name).read_text())
return parse_device0_batch(data['device0'])
def _unbound_hrefs(resources, registry) -> list[str]:
unbound = []
from custom_components.localthings.registry.discovery import discover
discover(resources, registry.capabilities, registry.pattern_capabilities,
log=unbound.append)
return unbound
def test_ignored_hrefs_absent_from_dishwasher_gaps():
resources = _load('dishwasher_device.json')
unbound = _unbound_hrefs(resources, dishwasher.REGISTRY)
present_ignored = _IGNORED_HREFS & set(unbound)
assert not present_ignored, f"ignored hrefs leaked into gap report: {present_ignored}"
def test_ignored_hrefs_absent_from_refrigerator_gaps():
resources = _load('refrigerator_device.json')
unbound = _unbound_hrefs(resources, refrigerator.REGISTRY)
present_ignored = _IGNORED_HREFS & set(unbound)
assert not present_ignored, f"ignored hrefs leaked into gap report: {present_ignored}"
def test_dishwasher_fixture_still_has_genuine_gaps():
"""Sanity check that the gap mechanism isn't silencing everything.
If this starts failing because the fixture is now fully covered, that's
good news — just trim the expected set down (or to empty) rather than
deleting the test.
"""
resources = _load('dishwasher_device.json')
unbound = set(_unbound_hrefs(resources, dishwasher.REGISTRY))
assert '/wm/jobbeginingstatus/vs/0' in unbound
assert '/diagnosis/vs/0' in unbound