79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""Entry diagnostics for Nvidia Shield (P4-2b, t21 F1 platform-module move).
|
|
|
|
HA discovers ``async_get_config_entry_diagnostics`` through the integration's
|
|
``diagnostics`` platform module — a placement inside media_player.py never
|
|
serves in a real install (t20 finding F1).
|
|
|
|
The payload: redacted entry config (data + options), the manager's StateStore
|
|
snapshot via the public ``store_snapshot`` accessor, and both transports'
|
|
health. Cert paths are kept readable; key material / pairing secrets never
|
|
exist in stored state, and any key carrying a secret-shaped name is redacted
|
|
defensively.
|
|
"""
|
|
|
|
import time
|
|
from dataclasses import asdict, is_dataclass
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from .connection import ShieldConnectionManager
|
|
|
|
_REDACT_KEY_MARKERS = ("pairing", "pin", "secret", "password", "token", "key_material")
|
|
_REDACTED = "**REDACTED**"
|
|
|
|
|
|
def _redact(obj):
|
|
"""Redact sensitive keys from a JSON-ish diagnostics payload."""
|
|
if isinstance(obj, dict):
|
|
return {
|
|
key: (
|
|
_REDACTED
|
|
if any(m in str(key).lower() for m in _REDACT_KEY_MARKERS)
|
|
else _redact(value)
|
|
)
|
|
for key, value in obj.items()
|
|
}
|
|
if isinstance(obj, list):
|
|
return [_redact(item) for item in obj]
|
|
return obj
|
|
|
|
|
|
def _store_snapshot(manager) -> dict[str, dict]:
|
|
"""TTL-stamped store snapshot; dataclass values expand to dicts."""
|
|
snapshot = {}
|
|
for name, (stamp, value) in manager.store.values_stamped().items():
|
|
if is_dataclass(value) and not isinstance(value, type):
|
|
value = asdict(value)
|
|
snapshot[name] = {"age": round(time.monotonic() - stamp, 3), "value": value}
|
|
return snapshot
|
|
|
|
|
|
async def async_get_config_entry_diagnostics(
|
|
hass: HomeAssistant, entry: ConfigEntry
|
|
) -> dict:
|
|
"""Entry diagnostics: redacted config + store snapshot + transport health."""
|
|
manager = getattr(entry, "runtime_data", None)
|
|
payload: dict = {
|
|
"entry": {
|
|
"data": dict(getattr(entry, "data", {}) or {}),
|
|
"options": dict(getattr(entry, "options", {}) or {}),
|
|
},
|
|
"store": {},
|
|
"transport": {},
|
|
}
|
|
if isinstance(manager, ShieldConnectionManager):
|
|
payload["store"] = manager.store_snapshot()
|
|
payload["transport"] = {
|
|
"adb": {"connected": manager.connected, "awake": manager.awake},
|
|
"protocol": {
|
|
"active": manager.protocol_active,
|
|
"available": manager.is_available,
|
|
"is_on": manager.is_on,
|
|
},
|
|
}
|
|
return _redact(payload)
|
|
|
|
|
|
__all__ = ["async_get_config_entry_diagnostics"]
|