diff --git a/CHANGELOG.md b/CHANGELOG.md index 42b7e4c1..ebe978d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ their own. See [RELEASING.md](RELEASING.md). ### Fixed +- **Opening a garage door, gate or door cover now asks for the Critical Action PIN when its entity id doesn't say so** ([#773](https://github.com/goruck/home-generative-agent/issues/773)). The recommended rules looked for `door`, `gate` or `garage` in the cover's entity id only. A cover with a plain id (such as `cover.msg100_abc`) opened with no PIN even when its device class was garage, gate or door, or its name was "Garage Door", and so did a cover group like `cover.all_covers` ("All Covers") holding the garage door. That held whether the agent opened it directly, by name through "turn on", in a scene it activated, or in an automation it wrote. A rule's word now also matches a cover whose device class is that word (the one Home Assistant reports, the one set under "Show as", or the integration's own, so a garage door shown as a blind is still held), or whose name, alias or device name has it (or its plural) with no letter next to it ("Garage Door", "Driveway Gates", "Gate1", not "Indoor Blinds"), unless its integration says it is a blind, shade, curtain, awning or window (setting "Show as", or `customize` in YAML, does not turn this off). Any group holding a matching cover (cover group, MQTT or ZHA group, or old-style group, nested to any depth) is held too. Covers and groups that are not set up yet count, since an automation the agent writes runs later: a cover not set up yet that its integration does not class as a blind, shade, curtain, awning or window, a group member Home Assistant does not know yet, and a group helper whose members can't be read are held; a disabled entity is not. Each extra match names one entity exactly, so `cover.a` being a garage door does not hold `cover.awning`. Rules you added yourself get the same reach: one for `msg100` now also holds a group containing that cover, and one written like an id (`garage_door`) matches the name "Garage Door". The security audit's list of public webhook automations that perform a critical action uses the same rules. Group membership and device class are read when the automation is written, so a garage door added to a group later is not checked again. Still not held, so add a rule naming part of the entity id: a cover with no device class whose id and name both lack the words (including a name in another language or a run-together word like "GarageDoor"), a cover whose only link to the words is its area's name, and the switch a garage cover is built on, switched directly ([#794](https://github.com/goruck/home-generative-agent/issues/794)). - **Opening or closing a cover by voice no longer fails when other device tools rank higher** ([#795](https://github.com/goruck/home-generative-agent/issues/795)). "Open the study shade" could fail with "I couldn't find a suitable tool": the agent offers the model only the few tools that best match the request, and device tools such as a vacuum's start or a media player's unpause could outscore the general "turn on" tool, which is the only one Home Assistant gives an assistant for opening a cover. The model then guessed a tool that does not exist and the call was refused. A request with an English action word ("turn", "open", "close", ...), or a "yes" to the agent offering one ("Want me to open the Study Shade?"), now always offers the turn on and turn off tools, alongside what the ranking chose. Requests in other languages, and replies such as "the left one", still rely on the ranking. A tool you excluded is still never offered. ### Changed diff --git a/custom_components/home_generative_agent/agent/graph.py b/custom_components/home_generative_agent/agent/graph.py index 6f44133c..86c7cc7d 100644 --- a/custom_components/home_generative_agent/agent/graph.py +++ b/custom_components/home_generative_agent/agent/graph.py @@ -137,6 +137,7 @@ service_runs_critical_action, ) from .pin_messages import describe_pending_action, pin_msg +from .rule_entities import expand_critical_rules from .token_counter import count_tokens_cross_provider if TYPE_CHECKING: @@ -459,7 +460,10 @@ def _critical_action_guard( """Enforce critical-action PIN flow when needed.""" if not ctx.pin_enabled: return None - if not _is_critical_action(tool_args, ctx.critical_actions, tool_name, ctx.hass): + # Extend entity_match rules to device classes, names and groups (#773, + # see rule_entities). + rules = expand_critical_rules(ctx.hass, ctx.critical_actions) + if not _is_critical_action(tool_args, rules, tool_name, ctx.hass): return None if not ctx.pin_hash or not ctx.pin_salt: LOGGER.warning( diff --git a/custom_components/home_generative_agent/agent/helpers.py b/custom_components/home_generative_agent/agent/helpers.py index 41e84acb..f2cd1764 100644 --- a/custom_components/home_generative_agent/agent/helpers.py +++ b/custom_components/home_generative_agent/agent/helpers.py @@ -877,9 +877,12 @@ def resolve_critical_action_policy(options: Mapping[str, Any]) -> CriticalAction pin_salt=pin_salt, # Copy: the default is a module-level list handed to every config # entry, so an in-place edit anywhere would corrupt it process-wide. - critical_actions=list( - options.get(CONF_CRITICAL_ACTIONS) or RECOMMENDED_CRITICAL_ACTIONS - ), + # The internal exact-entity marker is never taken from stored rules. + critical_actions=[ + {k: v for k, v in rule.items() if k != CRITICAL_RULE_EXACT_ENTITY} + for rule in options.get(CONF_CRITICAL_ACTIONS) + or RECOMMENDED_CRITICAL_ACTIONS + ], ) @@ -1150,6 +1153,29 @@ def thread_has_pending_action( ) +# Marks a rule copy `rule_entities.expand_critical_rules` makes for one entity: +# it matches that entity id exactly, not as a substring. Internal: removed from +# stored rules, so a user rule can never be narrowed by carrying it. The copy +# keeps the user rule's `entity_match`, so a reader that does not know this +# key still reads it as the rule it was copied from. +CRITICAL_RULE_EXACT_ENTITY = "_exact_entity" + + +def rule_entity_matches(rule: Mapping[str, str], entity_ids: Sequence[str]) -> bool: + """ + Return True when a rule's entity filter admits one of these entity ids. + + The ids are lowercased. A rule with no filter admits any. A comma-joined + id ("cover.a, cover.b") is read as the ids it lists for an exact match. + """ + if exact := (rule.get(CRITICAL_RULE_EXACT_ENTITY) or "").lower(): + return any( + exact == part.strip() for entity in entity_ids for part in entity.split(",") + ) + entity_match = (rule.get("entity_match") or "").lower() + return not entity_match or any(entity_match in e for e in entity_ids) + + def matches_critical_rule( *, domain: str, @@ -1176,17 +1202,12 @@ def matches_critical_rule( for rule in critical_actions: rule_domain = (rule.get("domain") or "").lower() rule_service = (rule.get("service") or "").lower() - entity_match = (rule.get("entity_match") or "").lower() if rule_domain and rule_domain != domain: continue if rule_service and rule_service != service: continue - if ( - entity_match - and not any(entity_match in e for e in entities) - and not unresolved_target - ): + if not unresolved_target and not rule_entity_matches(rule, entities): continue return True return False diff --git a/custom_components/home_generative_agent/agent/indirect_pin.py b/custom_components/home_generative_agent/agent/indirect_pin.py index 88ce0c58..54fdfa37 100644 --- a/custom_components/home_generative_agent/agent/indirect_pin.py +++ b/custom_components/home_generative_agent/agent/indirect_pin.py @@ -45,7 +45,7 @@ find_critical_automation_calls, step_targets, ) -from .helpers import TOOL_NAMESPACE_SEP, matches_critical_rule +from .helpers import TOOL_NAMESPACE_SEP, matches_critical_rule, rule_entity_matches if TYPE_CHECKING: from collections.abc import Mapping, Sequence @@ -506,10 +506,9 @@ def _entity_is_guarded( """Return True if any rule applies to this entity, whatever the service.""" for rule in rules: rule_domain = (rule.get("domain") or "").lower() - entity_match = (rule.get("entity_match") or "").lower() if rule_domain and rule_domain != domain: continue - if entity_match and entity_match not in entity_id: + if not rule_entity_matches(rule, [entity_id]): continue return True return False diff --git a/custom_components/home_generative_agent/agent/rule_entities.py b/custom_components/home_generative_agent/agent/rule_entities.py new file mode 100644 index 00000000..3369e924 --- /dev/null +++ b/custom_components/home_generative_agent/agent/rule_entities.py @@ -0,0 +1,318 @@ +""" +Extend ``entity_match`` rules to the entities a name match misses (#773). + +A rule's ``entity_match`` is matched against the entity ids a call targets. +Three kinds of entity act like a matching one without carrying the word in +their own id: + +* **An entity whose device class is the word.** A cover whose device class is + ``garage`` but whose id is ``cover.msg100_abc`` opens a garage door. The + device class must equal the word, so a rule for ``door`` does not reach + device class ``garage``. The class the user set ("Show as") and the + integration's own both count: a garage opener shown as a blind is held. +* **An entity named with the word.** A cover with no device class, id + ``cover.shelly1_abc`` and name "Garage Door", which "open the garage door" + finds by its name. The word, or its plural, must have no letter next to it + in the friendly name, the registry or integration name, an alias, or the + device's name: "Driveway Gates" and "Gate1" match ``gate``, "Indoor Blinds" + does not match ``door``. A cover its integration classes as a window + covering is not matched by name ("French Door Blinds"); a user's "Show as" + does not turn the name match off. A word written like an id + (``garage_door``) matches "Garage Door". Compound words ("GarageDoor") + and names in other languages are not read; device class or a rule naming + the id covers those. +* **A group holding a matching entity.** Opening ``cover.all_covers``, or a + call on an old-style ``group.*``, acts on every member, the garage door + included. Groups of groups are followed to any depth. A member matches by + the ways above, or by the word anywhere in its id, as a direct call on it + is matched (``cover.backdoor``, and so ``cover.outdoor_awning`` too). A member + Home Assistant has not set up or registered yet could be anything, so its + groups are held. A copy made for an old-style ``group.*`` keeps the rule's + domain, so it holds ``cover.open_cover`` aimed at the group. A call in the + ``group`` or ``homeassistant`` domain on it is not matched here: in an + automation it is held as an unresolved target (``automation_pin``), and + Home Assistant refuses HassTurnOn on it (the group domain has no turn_on). + +``expand_critical_rules`` adds, for each such entity, a copy of the rule that +matches that entity id exactly (``CRITICAL_RULE_EXACT_ENTITY``). Every screen +downstream (the tool-call gate, intents, scripts, scenes, ``add_automation``) +matches on entity ids through ``rule_entity_matches``, so each one holds the +call without knowing why the entity counts, and ``cover.a`` being a garage +door does not hold ``cover.awning``. + +Every check counts entities that are not loaded or not available, because an +automation the agent writes runs later. An entity only in the entity registry +is read from it. One not set up yet (no state, or the placeholder Home +Assistant restores at startup) in the rule's domain is held unless its +integration classes it as a window covering, since it could be a group whose +members cannot be seen yet; a disabled one is not, since it cannot run. An +entity that cannot be read is held the same way and logged, so one bad entity +does not fail every call. + +A group's members are read from its state (``entity_id``, or +``group_entities`` for MQTT and ZHA groups, which Home Assistant writes even +while the group is unavailable), else from the group helper's config entry. A +group helper whose members cannot be read is held. Membership and device +class are read when the check runs: a member added to a group after an +automation is written is not checked again. +""" + +from __future__ import annotations + +import logging +import re +from typing import TYPE_CHECKING, Any, cast + +from homeassistant.const import ( + ATTR_DEVICE_CLASS, + ATTR_ENTITY_ID, + ATTR_FRIENDLY_NAME, + ATTR_RESTORED, + CONF_ENTITIES, +) +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers import entity_registry as er + +from .helpers import CRITICAL_RULE_EXACT_ENTITY + +LOGGER = logging.getLogger(__name__) + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + + from homeassistant.core import HomeAssistant, State + +# MQTT and ZHA group entities list their members here. It is a capability +# attribute, which Home Assistant writes even while the entity is unavailable. +_ATTR_GROUP_ENTITIES = "group_entities" +# The integration whose entities are group helpers (cover, light, ... groups). +_GROUP_PLATFORM = "group" +# Scenes and scripts list the entities they set in `entity_id` too, but they +# are opened and screened by `indirect_pin`, not followed as groups here. +_NOT_GROUP_DOMAINS = ("script.", "scene.") +# Characters a name or a rule's word may use between words. +_SEPARATORS = re.compile(r"[_\-]+") +# Cover device classes that say the cover is a window covering. Its name is +# not read for the word: "French Door Blinds" is not a door. `shutter` is not +# here, since roller garage doors are often classed so. +_WINDOW_COVERINGS = frozenset({"awning", "blind", "curtain", "shade", "window"}) + + +def _text_list(value: object) -> list[str]: + """Return the strings in a list attribute, lowercased, or none.""" + if not isinstance(value, (list, tuple, set)): + return [] + items = cast("Iterable[object]", value) + return [item.lower() for item in items if isinstance(item, str)] + + +def _plain(text: str) -> str: + """Return text lowercased with `_` and `-` read as spaces.""" + return _SEPARATORS.sub(" ", text.lower()) + + +class _EntityIndex: + """Every entity's names, device classes and group membership, read once.""" + + def __init__(self, hass: HomeAssistant) -> None: + self._hass = hass + registry = er.async_get(hass) + self._devices = dr.async_get(hass) + # Every class an entity has been given: the one HA reports, the one + # the user set ("Show as") and the integration's own. A garage opener + # shown as a blind is still a garage opener. Keyed by every entity the + # index knows, so its keys are also the "known" set. + self.classes: dict[str, set[str]] = {} + self.names: dict[str, list[str]] = {} + self.members: dict[str, list[str]] = {} + self.unreadable_groups: set[str] = set() + self.unknown: set[str] = set() + for state in hass.states.async_all(): + entity_id = state.entity_id.lower() + self._read(entity_id, registry.async_get(entity_id), state) + for entry in registry.entities.values(): + if entry.entity_id not in self.classes: + self._read(entry.entity_id, entry, None) + # Member -> the groups holding it, so a match climbs to every group + # holding it in one pass, however deep the nesting. + self.holders: dict[str, set[str]] = {} + for group_id, group_members in self.members.items(): + for member in group_members: + self.holders.setdefault(member, set()).add(group_id) + + def _read( + self, entity_id: str, entry: er.RegistryEntry | None, state: State | None + ) -> None: + """Add one entity; one that cannot be read is held, not the whole call.""" + try: + self._add(entity_id, entry, state) + except Exception: + # Without this, one bad entity fails every guarded call in the + # house. It is held instead: it could be anything. + LOGGER.exception( + "Critical Action PIN: could not read %s; it is held", entity_id + ) + self.classes.setdefault(entity_id, set()) + self.names.setdefault(entity_id, []) + self.unknown.add(entity_id) + + def _add( + self, entity_id: str, entry: er.RegistryEntry | None, state: State | None + ) -> None: + attributes: Mapping[str, Any] = state.attributes if state is not None else {} + reported = attributes.get(ATTR_DEVICE_CLASS) + given: list[object] = [reported] + if entry is not None: + given += [entry.device_class, entry.original_device_class] + self.classes[entity_id] = {c.lower() for c in given if isinstance(c, str)} + # The integration's own class, not the user's "Show as" or YAML + # `customize`: only it can say the entity is a window covering, so a + # garage door relay shown as a blind keeps its name match. With no + # registry entry it is read from the loaded entity; unknown, the name + # is read. + own = ( + entry.original_device_class + if entry is not None + else getattr(self._loaded_entity(entity_id), "device_class", None) + ) + window_covering = isinstance(own, str) and own.lower() in _WINDOW_COVERINGS + names: list[object] = [attributes.get(ATTR_FRIENDLY_NAME)] + if entry is not None: + names += [entry.name, entry.original_name, *entry.aliases] + if entry.device_id and (device := self._devices.async_get(entry.device_id)): + # An entity named by its device ("Garage Door") has no name + # of its own in the registry. + names += [device.name_by_user, device.name] + self.names[entity_id] = ( + [] + if window_covering + else [_plain(name) for name in names if isinstance(name, str)] + ) + not_loaded = state is None or bool(attributes.get(ATTR_RESTORED)) + if ( + not_loaded + and not window_covering + and not (entry is not None and entry.disabled) + ): + # Not loaded (no state, or the placeholder Home Assistant restores + # at startup), so its members cannot be seen: unless its + # integration says it is a window covering, it could be a group + # (an MQTT group classed shutter, say). A disabled entity cannot + # run, so it is not held. + self.unknown.add(entity_id) + if entity_id.startswith(_NOT_GROUP_DOMAINS): + return + members = _text_list(attributes.get(ATTR_ENTITY_ID)) or _text_list( + attributes.get(_ATTR_GROUP_ENTITIES) + ) + if not members and self._platform(entity_id, entry) == _GROUP_PLATFORM: + # An unavailable group helper writes no member list. + members = self._helper_members(entry) + if not members: + self.unreadable_groups.add(entity_id) + if members: + self.members[entity_id] = members + + def _platform(self, entity_id: str, entry: er.RegistryEntry | None) -> str | None: + """Return the integration of an entity, registered or only loaded.""" + if entry is not None: + return entry.platform + entity = self._loaded_entity(entity_id) + platform = getattr(getattr(entity, "platform", None), "platform_name", None) + return platform if isinstance(platform, str) else None + + def _loaded_entity(self, entity_id: str) -> object | None: + """Return the entity object of a loaded entity with no registry entry.""" + # A YAML entity with no unique id has no registry entry. Its domain's + # EntityComponent is stored in `hass.data` under the domain (the + # component's DATA_COMPONENT key equals it) and holds the entity. + component = self._hass.data.get(entity_id.split(".", 1)[0]) + get_entity = getattr(component, "get_entity", None) + return get_entity(entity_id) if callable(get_entity) else None + + def _helper_members(self, entry: er.RegistryEntry | None) -> list[str]: + """Return the members a group helper's config entry lists.""" + if entry is None or entry.config_entry_id is None: + return [] + config_entry = self._hass.config_entries.async_get_entry(entry.config_entry_id) + if config_entry is None: + return [] + return _text_list(config_entry.options.get(CONF_ENTITIES)) + + def reach(self, domain: str, match: str) -> set[str]: + """Return every entity a rule on ``domain`` with this match reaches.""" + # A word with no letter next to it: "Gate1" and "Barn Doors" match, + # "Gatekeeper" and "Indoor" do not. + word = re.compile(rf"(? bool: + return not domain or entity_id.split(".", 1)[0] == domain + + matched = { + entity_id + for entity_id in self.classes + if in_domain(entity_id) + and ( + match in entity_id + or match in self.classes[entity_id] + or any(word.search(name) for name in self.names[entity_id]) + ) + } + # A member no state or registry entry describes (its integration not + # set up yet) could be anything a group in the domain could hold. + matched |= { + member + for member in self.holders + if member not in self.classes and in_domain(member) + } + # A group helper whose members cannot be read could hold anything; an + # entity not loaded (not disabled, and not a window covering by its + # integration's class) could be such a group; and one that could not + # be read could be anything. + matched |= { + entity_id + for entity_id in self.unreadable_groups | self.unknown + if in_domain(entity_id) + } + pending = list(matched) + while pending: + for group_id in self.holders.get(pending.pop(), ()): + if group_id not in matched: + matched.add(group_id) + pending.append(group_id) + return matched + + +def expand_critical_rules( + hass: HomeAssistant | None, rules: Iterable[Mapping[str, str]] +) -> list[dict[str, str]]: + """Return the rules plus one per entity an ``entity_match`` misses.""" + expanded = [dict(rule) for rule in rules] + if hass is None or getattr(hass, "states", None) is None: + if any(rule.get("entity_match") for rule in expanded): + LOGGER.warning( + "Critical Action PIN: no Home Assistant instance; covers are " + "matched by entity id only" + ) + return expanded + if not any(rule.get("entity_match") for rule in expanded): + return expanded + index = _EntityIndex(hass) + reached: dict[tuple[str, str], set[str]] = {} + extra: list[dict[str, str]] = [] + for rule in expanded: + match = (rule.get("entity_match") or "").lower() + if not match: + continue + key = ((rule.get("domain") or "").lower(), match) + if key not in reached: + reached[key] = index.reach(*key) + extra.extend( + {**rule, CRITICAL_RULE_EXACT_ENTITY: entity_id} + for entity_id in sorted(reached[key]) + if match not in entity_id + ) + return expanded + extra diff --git a/custom_components/home_generative_agent/agent/tools.py b/custom_components/home_generative_agent/agent/tools.py index 999d4ddf..2a157580 100644 --- a/custom_components/home_generative_agent/agent/tools.py +++ b/custom_components/home_generative_agent/agent/tools.py @@ -127,6 +127,7 @@ tool_call_data, ) from .pin_messages import describe_pending_action, pin_msg +from .rule_entities import expand_critical_rules if TYPE_CHECKING: from collections.abc import Generator, Mapping @@ -851,7 +852,8 @@ def _maybe_require_pin_for_automation( return None critical_calls = find_critical_automation_calls( - validated_config, policy.critical_actions + validated_config, + expand_critical_rules(cfg.get("hass"), policy.critical_actions), ) if not critical_calls: return None diff --git a/custom_components/home_generative_agent/const.py b/custom_components/home_generative_agent/const.py index 9882c350..436c8a86 100644 --- a/custom_components/home_generative_agent/const.py +++ b/custom_components/home_generative_agent/const.py @@ -61,7 +61,9 @@ # calls and the contents of LLM-authored automations. An `entity_match` rule # also fires when the call's real targets cannot be resolved at check time # (an area, device, label, floor, group, registry ID, or template) — see -# `agent/automation_pin.py`. +# `agent/automation_pin.py`. Where Home Assistant's state is at hand, an +# `entity_match` also matches a device class, a name and any group holding a +# matching entity (`agent/rule_entities.py`, #773). RECOMMENDED_CRITICAL_ACTIONS: list[dict[str, str]] = [ {"domain": "lock", "service": "unlock"}, {"domain": "lock", "service": "open"}, diff --git a/custom_components/home_generative_agent/snapshot/network.py b/custom_components/home_generative_agent/snapshot/network.py index da1738fc..e55dca53 100644 --- a/custom_components/home_generative_agent/snapshot/network.py +++ b/custom_components/home_generative_agent/snapshot/network.py @@ -39,6 +39,9 @@ from custom_components.home_generative_agent.agent.helpers import ( resolve_critical_action_policy, ) +from custom_components.home_generative_agent.agent.rule_entities import ( + expand_critical_rules, +) from custom_components.home_generative_agent.const import DOMAIN from custom_components.home_generative_agent.sentinel.auth_inventory import ( TOKEN_TYPE_LONG_LIVED, @@ -678,6 +681,11 @@ class HaNativeInputs: trusted_networks_bypass_login: bool | None = None addons: list[AddonInput] | None = None automations: list[AutomationInput] | None = None + # The Critical Action rules extended to device classes, names and groups + # (#773), read with `hass`. None when no automation has a public webhook + # (nothing to screen) or extending failed; the configured rules are then + # used. + critical_actions: list[dict[str, str]] | None = None # Conversation engines used by Assist pipelines that this integration's # Critical Action PIN does not cover (built-in agent, other LLM agents). assist_agents_outside_pin: list[str] | None = None @@ -1052,6 +1060,31 @@ def _collect_discovery( return unconfigured, ignored +def _collect_critical_actions( + hass: HomeAssistant, + context: NetworkBuildContext, + automations: Sequence[AutomationInput] | None, +) -> list[dict[str, str]] | None: + """ + Return the rules the add_automation gate applies, as it extends them. + + Read only for the public webhook automations the adapter screens with + them (`ha_native_adapter`, same `automation_has_public_webhook` test; + keep the two in step); with none, extending would read every entity for + nothing. + """ + if not automations or not any( + automation_has_public_webhook(a.config) for a in automations + ): + return None + try: + policy = resolve_critical_action_policy(context.options or {}) + return expand_critical_rules(hass, policy.critical_actions) + except Exception as err: # noqa: BLE001 - one input must not fail the snapshot + _log_input_failure("critical action rules", err) + return None + + async def async_collect_ha_native_inputs( # noqa: PLR0913 hass: HomeAssistant, context: NetworkBuildContext, @@ -1067,6 +1100,7 @@ async def async_collect_ha_native_inputs( # noqa: PLR0913 auth = await async_collect_auth_observation(hass, context.pseudonymizer) unconfigured, ignored = _collect_discovery(hass) sensitive_ids = [e["entity_id"] for e in entities if is_sensitive_entity(e)] + automations = _collect_automations(hass) return HaNativeInputs( now=now, auth=auth, @@ -1076,7 +1110,8 @@ async def async_collect_ha_native_inputs( # noqa: PLR0913 http=_collect_http(hass), trusted_networks_bypass_login=_collect_auth_providers(hass), addons=_collect_addons(hass), - automations=_collect_automations(hass), + automations=automations, + critical_actions=_collect_critical_actions(hass, context, automations), assist_agents_outside_pin=_collect_assist_agents(hass), discovered_unconfigured=unconfigured, discovered_ignored=ignored, @@ -1110,8 +1145,9 @@ def is_sensitive_entity(entity: SnapshotEntity) -> bool: Return True for an entity whose voice exposure is a security concern. Locks and alarm panels always; covers only when they are a door, gate, - or garage (by device class or name), mirroring the critical-action - matcher so "sensitive" means the same thing in both places. + or garage (by device class or name). The Critical Action PIN also holds + a group holding such a cover (`agent/rule_entities.py`); this check does + not follow groups yet (#793). """ domain = entity["domain"] if domain in {"lock", "alarm_control_panel"}: @@ -1336,9 +1372,9 @@ def ha_native_adapter( # noqa: PLR0912 - one branch per input, kept flat on pur if inputs.automations is not None: critical_actions = ( - resolve_critical_action_policy(context.options).critical_actions - if context.options is not None - else resolve_critical_action_policy({}).critical_actions + inputs.critical_actions + if inputs.critical_actions is not None + else resolve_critical_action_policy(context.options or {}).critical_actions ) public = sorted( a.entity_id diff --git a/docs/configuration.md b/docs/configuration.md index b6e12785..81dbd8a4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -430,7 +430,7 @@ Protects sensitive actions (unlocking doors, opening covers) behind a second ver **Protected actions:** - Unlocking or opening locks -- Opening covers whose `entity_id` includes `door`, `gate`, or `garage` — via `open_cover`, `open`, `toggle`, or `set_cover_position`, since all four open a closed door +- Opening covers whose `entity_id` includes `door`, `gate`, or `garage`, whose device class is `door`, `gate` or `garage`, or whose name, an alias or its device's name has one of those words or its plural with no letter next to it ("Garage Door", "Driveway Gates", "Gate1", not "Indoor Blinds"), unless its device class is blind, shade, curtain, awning or window — via `open_cover`, `open`, `toggle`, or `set_cover_position`, since all four open a closed door. A group holding such a cover (a cover group such as `cover.all_covers`, an MQTT or ZHA group, or an old-style group, nested to any depth) is held the same way, since opening it opens the garage door. This applies to every rule with an `entity_match`: a rule you add for `msg100` also holds a group containing `cover.msg100_abc`, and a word written like an id (`garage_door`) matches the name "Garage Door". Covers and groups that are not set up yet count too, because an automation the agent writes runs later: a cover not set up yet with no device class, a group member Home Assistant does not know yet, and a group helper whose members cannot be read are held; a disabled entity is not. Membership is read when the automation is written: a garage door added to a group afterwards is not checked again. Not held: a cover with no device class whose entity id and name both lack the words (a name in another language or a run-together word such as "GarageDoor"); a cover whose only link to the words is the name of its area (a cover named "Main" in the "Garage" area); and the switch a garage cover is built on (`switch_as_x`), switched directly. Add a rule naming part of the entity id for those. - Using HA intent tools on locks - Using HA intent tools on covers and valves the same way: "turn on" opens a cover and "set position" moves it, so the rules above apply to them. The PIN check finds the devices the way Home Assistant will, by name, alias, area or floor, and applies a rule's `entity_match` to each device's `entity_id`, so "open the garage door" is held whether the agent names it, gives its entity id, or names only its area. The PIN prompt names every action such a command takes, so one that would lock a lock and open a gate says both. Rules you add for other devices apply to these commands too: "stop" stops a cover or valve (`stop_cover`, `stop_valve`), "turn on" presses a button (`button.press`) and runs a script (a rule for `script.turn_on` or for the script's own service, such as `script.open_garage`, both match when the agent turns the script on or runs it as its own tool) - **Running a script or activating a scene that does any of the above.** Turning on a script, or running it through the tool Home Assistant gives each exposed script, is held when one of its steps would be held in an automation (see **Automation screening** below): a script that opens the garage door asks for the PIN, one that turns on the porch lights does not. A step that runs another script or scene, fires an event, or builds its action from a template is held too, because what it does is not known until it runs. A scene from Home Assistant's own scene config is checked per entity it sets: one that unlocks the front door or opens the garage door asks for the PIN, a bedtime scene that locks the doors and closes the garage does not. A rule's `entity_match` is applied to each entity the scene names, and the members of an old-style group it sets are checked the same way. A scene stored by another integration cannot be read, so it is held, except for lighting systems whose scenes only set their own lights (Hue, LIFX Cloud, LiteJet, Sunricher DALI). A hub's scene (Fibaro, Vera, Overkiz, MQTT, KNX, deCONZ and others) can open a gate or unlock a door, and a shade system's scene (Hunter Douglas PowerView) moves covers a rule can guard, so these ask for the PIN. A tool that calls services by name, such as an MCP service tool, is checked the way an automation step is (see **Automation screening** below), except that a script or scene it runs is opened as above rather than held. A target of `all`, an area, a label, a group or a registry id is held when a rule applies to the call's domain and service (for a call on `light.turn_on` with an area and no light rule, it runs), and always for `homeassistant.*` calls and calls that run a script or scene. `homeassistant.toggle` on a garage door is checked as `cover.toggle`, a call that gives no domain is also checked as `homeassistant.`, wherever its targets sit (`entity_id`, `entity_ids`, `target`, `data`), when it turns something on or toggles it, reaches several domains, or targets an area, device or label (held, since which devices act is unknown), a field given twice with two values (at the top of the call and inside `data`) is held, and `automation.trigger` is held. A script or scene whose configuration cannot be read is held. The PIN prompt names the script as Home Assistant knows it, not from what the agent passed to it. diff --git a/docs/constants.md b/docs/constants.md index b36eaa93..721ca124 100644 --- a/docs/constants.md +++ b/docs/constants.md @@ -476,7 +476,7 @@ This document covers the named constants that affect integration behaviour, orga The `RECOMMENDED_CRITICAL_ACTIONS` list defines which actions require PIN confirmation. By default this covers `lock.unlock`, `lock.open`, `lock.toggle`, and every cover service that opens a closed door — `open_cover`, `open`, `toggle`, and `set_cover_position` — for entities whose `entity_id` contains `door`, `gate`, or `garage`. -The same rules screen two surfaces: direct tool calls from the conversation agent, and the *contents* of automations the agent writes with `add_automation` (see [Critical Action PIN](configuration.md#critical-action-pin)). In the automation path an `entity_match` rule also fires when the call's real targets cannot be resolved at write time — an area, device, label, floor, group, entity registry ID, or template — because the substring cannot be checked against entities that are only resolved when the automation runs. +The same rules screen two surfaces: direct tool calls from the conversation agent, and the *contents* of automations the agent writes with `add_automation` (see [Critical Action PIN](configuration.md#critical-action-pin)). In the automation path an `entity_match` rule also fires when the call's real targets cannot be resolved at write time — an area, device, label, floor, group, entity registry ID, or template — because the substring cannot be checked against entities that are only resolved when the automation runs. Wherever the gate can see Home Assistant's state and entity registry, an `entity_match` also matches an entity in the rule's domain whose device class equals it (a cover with device class `garage` for `garage`) or whose name, alias or device name has it or its plural with no letter next to it (not for a cover whose device class is a window covering), and any group (cover group, MQTT or ZHA group, or old-style group, at any depth) that holds a matching entity, set up or not. A group helper whose members cannot be read, a group member Home Assistant does not know, and an entity in the rule's domain that is not set up and has no device class (not a disabled one) count as matching. Each such entity gets a rule copy matching its id exactly. --- diff --git a/tests/custom_components/home_generative_agent/test_cover_group_pin_gate.py b/tests/custom_components/home_generative_agent/test_cover_group_pin_gate.py new file mode 100644 index 00000000..ef16a224 --- /dev/null +++ b/tests/custom_components/home_generative_agent/test_cover_group_pin_gate.py @@ -0,0 +1,1015 @@ +# ruff: noqa: S101 +""" +Garage covers are held by device class and through groups (#773). + +The recommended cover rules match `door`, `gate` or `garage` in the entity id +only. A cover whose device class is garage but whose id is plain, and a group +(cover group or old-style group) holding a garage door, opened with no PIN. +""" + +from __future__ import annotations + +import logging +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any, cast + +import pytest +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers import entity_registry as er +from homeassistant.setup import async_setup_component +from homeassistant.util import dt as dt_util +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.home_generative_agent.agent import graph as graph_module +from custom_components.home_generative_agent.agent import tools as tools_module +from custom_components.home_generative_agent.agent.automation_pin import ( + find_critical_automation_calls, +) +from custom_components.home_generative_agent.agent.graph import ( + _is_critical_action, + _prepare_tool_args, +) +from custom_components.home_generative_agent.agent.helpers import ( + CRITICAL_RULE_EXACT_ENTITY, + matches_critical_rule, + resolve_critical_action_policy, +) +from custom_components.home_generative_agent.agent.indirect_pin import ( + runs_critical_action, +) +from custom_components.home_generative_agent.agent.rule_entities import ( + expand_critical_rules, +) +from custom_components.home_generative_agent.agent.tools import ( + _maybe_require_pin_for_automation, +) +from custom_components.home_generative_agent.const import ( + CONF_CRITICAL_ACTION_PIN_ENABLED, + CONF_CRITICAL_ACTION_PIN_HASH, + CONF_CRITICAL_ACTION_PIN_SALT, + CONF_CRITICAL_ACTIONS, + RECOMMENDED_CRITICAL_ACTIONS, +) +from custom_components.home_generative_agent.core.utils import hash_pin +from custom_components.home_generative_agent.snapshot import network as network_module +from custom_components.home_generative_agent.snapshot.network import ( + AutomationInput, + HaNativeInputs, + NetworkBuildContext, + _collect_critical_actions, + ha_native_adapter, +) + +if TYPE_CHECKING: + from homeassistant.core import HomeAssistant + +RULES = RECOMMENDED_CRITICAL_ACTIONS + + +@pytest.fixture +async def home(hass: HomeAssistant) -> HomeAssistant: + """ + Set up covers a name match misses. + + A garage opener with a plain id, a gate whose device class is only in the + registry, blinds, a cover group and old-style groups (one nested) holding + the opener, a group of blinds only, and scenes opening the groups. + """ + hass.states.async_set( + "cover.msg100_abc", "closed", {"device_class": "garage", "friendly_name": "Q"} + ) + hass.states.async_set( + "cover.living_room_blinds", "closed", {"device_class": "shade"} + ) + hass.states.async_set("cover.study_blinds", "closed") + entry = er.async_get(hass).async_get_or_create( + "cover", + "test", + "yard", + suggested_object_id="yard_1", + original_device_class="gate", + ) + hass.states.async_set(entry.entity_id, "closed") + hass.states.async_set( + "cover.all_covers", + "closed", + { + "entity_id": ["cover.living_room_blinds", "cover.msg100_abc"], + "friendly_name": "All Covers", + }, + ) + hass.states.async_set( + "cover.all_blinds", + "closed", + {"entity_id": ["cover.living_room_blinds", "cover.study_blinds"]}, + ) + hass.states.async_set( + "group.openers", "closed", {"entity_id": ["cover.msg100_abc"]} + ) + hass.states.async_set("group.outside", "closed", {"entity_id": ["group.openers"]}) + hass.states.async_set("light.porch", "off") + assert await async_setup_component( + hass, + "scene", + { + "scene": [ + {"name": "Open Everything", "entities": {"cover.all_covers": "open"}}, + {"name": "Open Blinds", "entities": {"cover.all_blinds": "open"}}, + ] + }, + ) + await hass.async_block_till_done() + return hass + + +def _open(entity_id: str) -> dict[str, Any]: + return {"domain": "cover", "service": "open_cover", "entity_id": entity_id} + + +@pytest.mark.parametrize( + ("entity_id", "held_before"), + [ + # Device class garage in the state, id with no rule word. + ("cover.msg100_abc", False), + # Device class gate only in the registry. + ("cover.yard_1", False), + # A cover group. + ("cover.all_covers", False), + # An old-style group, and a group of groups: already held, since the + # screen cannot tell what a `group.*` target reaches. + ("group.openers", True), + ("group.outside", True), + ], +) +async def test_opening_a_garage_cover_however_named_is_held( + home: HomeAssistant, entity_id: str, *, held_before: bool +) -> None: + rules = expand_critical_rules(home, RULES) + assert _is_critical_action(_open(entity_id), rules, "mcp__call_service", home) + assert ( + _is_critical_action(_open(entity_id), RULES, "mcp__call_service", home) + is held_before + ) + + +@pytest.mark.parametrize( + "entity_id", ["cover.living_room_blinds", "cover.study_blinds", "cover.all_blinds"] +) +async def test_blinds_and_a_group_of_blinds_are_not_held( + home: HomeAssistant, entity_id: str +) -> None: + rules = expand_critical_rules(home, RULES) + assert not _is_critical_action(_open(entity_id), rules, "mcp__call_service", home) + + +async def test_turning_on_a_garage_cover_group_by_name_is_held( + home: HomeAssistant, +) -> None: + """The agent's usual intent reaches the group through the same rules.""" + rules = expand_critical_rules(home, RULES) + args = {"name": "All Covers"} + assert _is_critical_action(args, rules, "intent__HassTurnOn", home) + assert not _is_critical_action(args, RULES, "intent__HassTurnOn", home) + + +async def test_closing_a_garage_cover_group_is_not_held(home: HomeAssistant) -> None: + rules = expand_critical_rules(home, RULES) + args = {**_open("cover.all_covers"), "service": "close_cover"} + assert not _is_critical_action(args, rules, "mcp__call_service", home) + + +async def test_a_scene_opening_a_group_with_a_garage_door_is_held( + home: HomeAssistant, +) -> None: + """The scene screen checks the group by the extended rules too.""" + rules = expand_critical_rules(home, RULES) + assert runs_critical_action(home, "scene.open_everything", rules) + assert not runs_critical_action(home, "scene.open_blinds", rules) + + +async def test_an_automation_opening_the_group_is_held(home: HomeAssistant) -> None: + rules = expand_critical_rules(home, RULES) + config = { + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.all_covers"}} + ] + } + assert find_critical_automation_calls(config, rules) + assert not find_critical_automation_calls(config, RULES) + + +async def test_the_gate_holds_a_garage_cover_for_a_pin(home: HomeAssistant) -> None: + """End to end: the tool-call gate extends the rules itself.""" + pending: dict[str, dict[str, Any]] = {} + ctx = cast( + "Any", + SimpleNamespace( + hass=home, + config={"configurable": {"thread_id": "t1"}}, + critical_actions=RULES, + pin_enabled=True, + pin_hash="hash", + pin_salt="salt", + pending_actions=pending, + ), + ) + redirect, _ = _prepare_tool_args( + "mcp__call_service", _open("cover.all_covers"), ctx, {"id": "call1"} + ) + assert redirect is not None + assert "requires_pin" in str(redirect.content) + assert len(pending) == 1 + + +async def test_a_user_rule_reaches_groups_holding_its_entity( + home: HomeAssistant, +) -> None: + """The workaround rule from the issue also covers the groups.""" + rules = expand_critical_rules( + home, [{"domain": "cover", "service": "open_cover", "entity_match": "msg100"}] + ) + assert _is_critical_action( + _open("cover.all_covers"), rules, "mcp__call_service", home + ) + + +async def test_rules_without_entity_match_are_unchanged(home: HomeAssistant) -> None: + rules = [{"domain": "lock", "service": "unlock"}] + assert expand_critical_rules(home, rules) == rules + + +def test_without_hass_the_rules_are_copied_unchanged() -> None: + expanded = expand_critical_rules(None, RULES) + assert expanded == RULES + assert expanded is not RULES + assert all(a is not b for a, b in zip(expanded, RULES, strict=True)) + + +@pytest.mark.parametrize("hass", [None, SimpleNamespace(states=None)]) +def test_without_hass_a_warning_says_covers_match_by_id( + hass: Any, caplog: pytest.LogCaptureFixture +) -> None: + """The fallback is logged, and only when a rule has an entity_match.""" + with caplog.at_level(logging.WARNING): + expand_critical_rules(hass, [{"domain": "lock", "service": "unlock"}]) + assert not caplog.records + with caplog.at_level(logging.WARNING): + expand_critical_rules(hass, RULES) + assert any("no Home Assistant instance" in r.getMessage() for r in caplog.records) + + +async def test_an_entity_already_named_by_the_rule_adds_nothing( + hass: HomeAssistant, +) -> None: + hass.states.async_set("cover.garage_door", "closed", {"device_class": "garage"}) + assert expand_critical_rules(hass, RULES) == RULES + + +def _reaches(hass: HomeAssistant, rules: Any = RULES) -> set[str]: + """Return the entity ids the extended rules name beyond the given ones.""" + extra = expand_critical_rules(hass, rules)[len(rules) :] + return {r[CRITICAL_RULE_EXACT_ENTITY] for r in extra} + + +async def test_a_garage_cover_not_loaded_yet_is_held(hass: HomeAssistant) -> None: + """At startup a registered cover has no state; an automation outlives that.""" + er.async_get(hass).async_get_or_create( + "cover", + "test", + "opener", + suggested_object_id="q1", + original_device_class="garage", + ) + assert "cover.q1" in _reaches(hass) + + +async def test_a_user_set_device_class_and_mixed_case_count( + hass: HomeAssistant, +) -> None: + entities = er.async_get(hass) + entry = entities.async_get_or_create("cover", "test", "a", suggested_object_id="a1") + entities.async_update_entity(entry.entity_id, device_class="gate") + hass.states.async_set(entry.entity_id, "closed") + hass.states.async_set("cover.b1", "closed", {"device_class": "Garage"}) + assert {"cover.a1", "cover.b1"} <= _reaches(hass) + + +async def test_a_garage_shown_as_a_blind_is_still_held(hass: HomeAssistant) -> None: + """ + "Show as" changes the class HA reports, not what the integration says. + + A garage opener the user shows as a blind reports `blind`, which also + turns off the name match; the integration's own class still says garage. + """ + entities = er.async_get(hass) + entry = entities.async_get_or_create( + "cover", + "test", + "opener", + suggested_object_id="msg200_abc", + original_device_class="garage", + ) + entities.async_update_entity(entry.entity_id, device_class="blind") + hass.states.async_set( + entry.entity_id, + "closed", + {"device_class": "blind", "friendly_name": "Garage Door"}, + ) + hass.states.async_set( + "cover.downstairs", "closed", {"entity_id": [entry.entity_id]} + ) + assert {entry.entity_id, "cover.downstairs"} <= _reaches(hass) + + +async def test_a_garage_door_relay_shown_as_a_blind_is_held_by_name( + hass: HomeAssistant, +) -> None: + """A user's "Show as" does not turn off the name match.""" + entities = er.async_get(hass) + entry = entities.async_get_or_create( + "cover", "test", "relay", suggested_object_id="relay_1" + ) + entities.async_update_entity(entry.entity_id, device_class="blind") + hass.states.async_set( + entry.entity_id, + "closed", + {"device_class": "blind", "friendly_name": "Garage Door"}, + ) + assert entry.entity_id in _reaches(hass) + + +async def test_an_unloaded_cover_with_a_non_window_class_is_held( + hass: HomeAssistant, +) -> None: + """ + An MQTT or ZHA group not set up yet shows no members. + + Its class (shutter) does not say it is harmless, so it is held. + """ + entities = er.async_get(hass) + entities.async_get_or_create( + "cover", + "mqtt", + "grp", + suggested_object_id="all_covers", + original_device_class="shutter", + ) + entities.async_get_or_create( + "cover", + "mqtt", + "blind", + suggested_object_id="den", + original_device_class="blind", + ) + reached = _reaches(hass) + assert "cover.all_covers" in reached + assert "cover.den" not in reached + + +async def test_an_entity_that_cannot_be_read_is_held_alone( + hass: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """One bad entity is held in its domain; the others are still read.""" + hass.states.async_set("cover.odd", "closed") + hass.states.async_set("cover.study_shade", "closed", {"device_class": "shade"}) + hass.states.async_set("light.porch", "off") + + class _Broken: + def get_entity(self, entity_id: str) -> None: + if entity_id == "cover.odd": + raise RuntimeError + + monkeypatch.setitem(hass.data, "cover", _Broken()) + reached = _reaches(hass) + assert "cover.odd" in reached + assert "cover.study_shade" not in reached + + +async def test_a_yaml_cover_customized_as_a_blind_keeps_its_name_match( + hass: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + With no registry entry, the integration's class is read from the entity. + + `customize: device_class: blind` changes the state, not the entity, so a + YAML garage relay named "Garage Door" is still held; a YAML cover whose + integration says blind is not matched by name. + """ + hass.states.async_set( + "cover.relay_1", + "closed", + {"device_class": "blind", "friendly_name": "Garage Door"}, + ) + hass.states.async_set( + "cover.patio", + "closed", + {"device_class": "blind", "friendly_name": "French Door Blinds"}, + ) + own = {"cover.relay_1": None, "cover.patio": "blind"} + + class _Component: + def get_entity(self, entity_id: str) -> SimpleNamespace | None: + if entity_id not in own: + return None + return SimpleNamespace(device_class=own[entity_id], platform=None) + + monkeypatch.setitem(hass.data, "cover", _Component()) + reached = _reaches(hass) + assert "cover.relay_1" in reached + assert "cover.patio" not in reached + + +async def test_an_unreadable_entity_is_logged( + hass: HomeAssistant, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + hass.states.async_set("cover.odd", "closed") + + class _Broken: + def get_entity(self, entity_id: str) -> None: + raise RuntimeError + + monkeypatch.setitem(hass.data, "cover", _Broken()) + with caplog.at_level(logging.ERROR): + _reaches(hass) + assert any( + r.levelno == logging.ERROR and "cover.odd" in r.getMessage() + for r in caplog.records + ) + + +async def test_device_class_must_equal_the_word(hass: HomeAssistant) -> None: + """A custom rule is not stretched over every device class containing it.""" + hass.states.async_set("switch.plug", "off", {"device_class": "outlet"}) + rules = [{"domain": "switch", "service": "turn_on", "entity_match": "out"}] + assert _reaches(hass, rules) == set() + + +@pytest.mark.parametrize( + ("name", "held"), + [ + ("Garage Door", True), + ("Back gate", True), + ("Indoor Blinds", False), + ("Garagekeeper", False), + ("Driveway Gates", True), + ("Barn-Doors", True), + ], +) +async def test_a_cover_named_with_the_word_is_held( + hass: HomeAssistant, name: str, *, held: bool +) -> None: + """A relay cover with no device class is found by its name, as HA finds it.""" + hass.states.async_set("cover.shelly1_abc", "closed", {"friendly_name": name}) + assert ("cover.shelly1_abc" in _reaches(hass)) is held + + +async def test_an_alias_counts_as_a_name(hass: HomeAssistant) -> None: + entities = er.async_get(hass) + entry = entities.async_get_or_create("cover", "test", "r", suggested_object_id="r1") + entities.async_update_entity(entry.entity_id, aliases=["garage"]) + hass.states.async_set(entry.entity_id, "closed") + assert "cover.r1" in _reaches(hass) + + +async def test_a_group_listing_group_entities_is_held(hass: HomeAssistant) -> None: + """MQTT and ZHA groups list their members as `group_entities`.""" + hass.states.async_set("cover.garage_door", "closed") + hass.states.async_set( + "cover.mqtt_group", "unavailable", {"group_entities": ["cover.garage_door"]} + ) + assert "cover.mqtt_group" in _reaches(hass) + + +async def test_an_unavailable_group_helper_is_read_from_its_config( + hass: HomeAssistant, +) -> None: + """An unavailable group writes no members; its config entry still lists them.""" + entities = er.async_get(hass) + for options, object_id in ( + ({"entities": ["cover.garage_door"]}, "garage_group"), + ({"entities": ["cover.study_blinds"]}, "blind_group"), + ): + entry = MockConfigEntry(domain="group", options=options) + entry.add_to_hass(hass) + reg = entities.async_get_or_create( + "cover", + "group", + object_id, + config_entry=entry, + suggested_object_id=object_id, + ) + hass.states.async_set(reg.entity_id, "unavailable") + hass.states.async_set("cover.garage_door", "closed") + hass.states.async_set("cover.study_blinds", "closed") + reached = _reaches(hass) + assert "cover.garage_group" in reached + assert "cover.blind_group" not in reached + + +async def test_a_group_helper_with_unreadable_members_is_held( + hass: HomeAssistant, +) -> None: + """With no member list anywhere, the group could hold the garage door.""" + reg = er.async_get(hass).async_get_or_create( + "cover", "group", "x", suggested_object_id="mystery" + ) + hass.states.async_set(reg.entity_id, "unavailable") + assert "cover.mystery" in _reaches(hass) + + +async def test_groups_in_a_loop_and_a_long_chain_are_followed( + hass: HomeAssistant, +) -> None: + hass.states.async_set("cover.msg100_abc", "closed", {"device_class": "garage"}) + hass.states.async_set( + "group.a", "on", {"entity_id": ["group.b", "cover.msg100_abc"]} + ) + hass.states.async_set("group.b", "on", {"entity_id": ["group.a"]}) + # Outermost first, the order a level-by-level scan handles worst. + for i in range(300): + hass.states.async_set( + f"group.chain{i}", "on", {"entity_id": [f"group.chain{i + 1}"]} + ) + hass.states.async_set("group.chain300", "on", {"entity_id": ["cover.msg100_abc"]}) + reached = _reaches(hass) + assert {"group.a", "group.b", "group.chain0", "group.chain300"} <= reached + + +async def test_scenes_and_scripts_are_not_followed_as_groups( + home: HomeAssistant, +) -> None: + """A scene lists its entities too; it is screened by what it sets instead.""" + assert not {e for e in _reaches(home) if e.startswith(("scene.", "script."))} + + +async def test_a_rule_with_no_domain_reaches_every_domain(hass: HomeAssistant) -> None: + hass.states.async_set("switch.relay", "off", {"friendly_name": "Garage"}) + rules = [{"service": "turn_on", "entity_match": "garage"}] + assert "switch.relay" in _reaches(hass, rules) + + +async def test_add_automation_holds_an_automation_opening_the_group( + home: HomeAssistant, +) -> None: + """The add_automation gate extends the rules itself.""" + hashed, salt = hash_pin("1234") + pending: dict[str, Any] = {} + cfg = cast( + "Any", + { + "hass": home, + "options": { + CONF_CRITICAL_ACTION_PIN_ENABLED: True, + CONF_CRITICAL_ACTION_PIN_HASH: hashed, + CONF_CRITICAL_ACTION_PIN_SALT: salt, + }, + "pending_actions": pending, + "thread_id": "t1", + }, + ) + validated = { + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.all_covers"}} + ] + } + out = _maybe_require_pin_for_automation( + cfg=cfg, validated_config=validated, ha_automation_config=dict(validated) + ) + assert out is not None + assert "requires_pin" in out + assert len(pending) == 1 + + +async def test_the_audit_flags_a_webhook_automation_opening_the_group( + home: HomeAssistant, +) -> None: + context = NetworkBuildContext(options={}) + automation = { + "triggers": [{"trigger": "webhook", "webhook_id": "x", "local_only": False}], + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.all_covers"}} + ], + } + automations = [AutomationInput("automation.open_all", automation)] + inputs = HaNativeInputs( + now=dt_util.utcnow(), + automations=automations, + critical_actions=_collect_critical_actions(home, context, automations), + ) + ha = ha_native_adapter(inputs, [], context).ha_security + assert ha["webhook_automations_critical"] == ["automation.open_all"] + + +async def test_the_audit_extends_nothing_without_a_public_webhook( + home: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """Most homes have no public webhook automation: no entity is read.""" + calls: list[object] = [] + + def _record(*args: Any) -> list[dict[str, str]]: + calls.append(args) + return [] + + monkeypatch.setattr(network_module, "expand_critical_rules", _record) + context = NetworkBuildContext(options={}) + + def _webhook(*, local_only: bool) -> dict[str, Any]: + trigger = {"trigger": "webhook", "webhook_id": "x", "local_only": local_only} + return {"triggers": [trigger], "actions": []} + + local = [AutomationInput("automation.l", _webhook(local_only=True))] + assert _collect_critical_actions(home, context, None) is None + assert _collect_critical_actions(home, context, []) is None + assert _collect_critical_actions(home, context, local) is None + assert calls == [] + public = [AutomationInput("automation.p", _webhook(local_only=False))] + assert _collect_critical_actions(home, context, public) == [] + assert len(calls) == 1 + + +async def test_the_collector_extends_the_rules_for_a_public_webhook( + home: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """The audit's collector passes the automations it read to the extension.""" + automation = { + "triggers": [{"trigger": "webhook", "webhook_id": "x", "local_only": False}], + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.all_covers"}} + ], + } + monkeypatch.setattr( + network_module, + "_collect_automations", + lambda _hass: [AutomationInput("automation.open_all", automation)], + ) + context = NetworkBuildContext(options={}) + inputs = await network_module.async_collect_ha_native_inputs( + home, + context, + [], + now=dt_util.utcnow(), + entity_device={}, + device_domains={}, + ) + assert inputs.critical_actions is not None + ha = ha_native_adapter(inputs, [], context).ha_security + assert ha["webhook_automations_critical"] == ["automation.open_all"] + + +async def test_a_rule_word_written_like_an_id_matches_a_name( + hass: HomeAssistant, +) -> None: + """The issue's workaround spells the word as in an id.""" + hass.states.async_set( + "cover.shelly1_abc", "closed", {"friendly_name": "Garage Door"} + ) + rules = [ + {"domain": "cover", "service": "open_cover", "entity_match": "garage_door"} + ] + assert "cover.shelly1_abc" in _reaches(hass, rules) + + +async def test_a_cover_not_loaded_is_known_by_its_device_name( + hass: HomeAssistant, +) -> None: + """An entity named by its device has no name of its own in the registry.""" + config_entry = MockConfigEntry(domain="test") + config_entry.add_to_hass(hass) + device = dr.async_get(hass).async_get_or_create( + config_entry_id=config_entry.entry_id, + identifiers={("test", "relay")}, + name="Garage Door", + ) + er.async_get(hass).async_get_or_create( + "cover", + "test", + "relay", + suggested_object_id="shelly_relay", + device_id=device.id, + original_device_class="shutter", + ) + assert "cover.shelly_relay" in _reaches(hass) + + +@pytest.mark.parametrize(("device_class", "held"), [(None, True), ("shade", False)]) +async def test_a_cover_not_loaded_with_no_device_class_is_held( + hass: HomeAssistant, device_class: str | None, *, held: bool +) -> None: + """Unloaded, it could be an MQTT or ZHA group whose members are unseen.""" + er.async_get(hass).async_get_or_create( + "cover", + "mqtt", + "grp", + suggested_object_id="all_covers", + original_device_class=device_class, + ) + assert ("cover.all_covers" in _reaches(hass)) is held + + +async def test_a_member_not_known_yet_is_matched_by_its_id( + hass: HomeAssistant, +) -> None: + """The garage door's integration is still loading; the group lists it.""" + hass.states.async_set( + "cover.all_covers", "closed", {"entity_id": ["cover.garage_door"]} + ) + assert "cover.all_covers" in _reaches(hass) + + +async def test_a_group_holds_what_a_direct_call_holds(hass: HomeAssistant) -> None: + """ + A member's id is read as the rule reads a direct call's: anywhere in it. + + `cover.backdoor` is held when opened directly, so a group holding it is + held too. So is a group of `cover.outdoor_awning`, which a direct call + already holds for the `door` in it. + """ + hass.states.async_set("cover.backdoor", "closed") + hass.states.async_set("cover.outdoor_awning", "closed", {"device_class": "awning"}) + hass.states.async_set("cover.study_blinds", "closed", {"device_class": "blind"}) + hass.states.async_set( + "cover.all_covers", "closed", {"entity_id": ["cover.backdoor"]} + ) + hass.states.async_set( + "cover.all_shades", "closed", {"entity_id": ["cover.outdoor_awning"]} + ) + hass.states.async_set( + "cover.all_blinds", "closed", {"entity_id": ["cover.study_blinds"]} + ) + reached = _reaches(hass) + assert {"cover.all_covers", "cover.all_shades"} <= reached + assert "cover.all_blinds" not in reached + + +async def test_an_unavailable_yaml_cover_group_is_held(hass: HomeAssistant) -> None: + """A YAML group has no registry entry; its platform says it is a group.""" + hass.states.async_set("cover.garage_door", "closed") + assert await async_setup_component( + hass, + "cover", + {"cover": [{"platform": "group", "name": "Yard", "entities": ["cover.x"]}]}, + ) + await hass.async_block_till_done() + assert er.async_get(hass).async_get("cover.yard") is None + hass.states.async_set("cover.yard", "unavailable") + assert "cover.yard" in _reaches(hass) + + +async def test_an_unreadable_group_is_held_only_for_its_domain( + hass: HomeAssistant, +) -> None: + for domain in ("cover", "light"): + reg = er.async_get(hass).async_get_or_create( + domain, "group", domain, suggested_object_id="mystery" + ) + hass.states.async_set(reg.entity_id, "unavailable") + reached = _reaches(hass) + assert "cover.mystery" in reached + assert "light.mystery" not in reached + + +async def test_the_audit_falls_back_to_the_rules_when_extending_fails( + home: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + def _fail(*_args: Any) -> list[dict[str, str]]: + raise RuntimeError + + monkeypatch.setattr(network_module, "expand_critical_rules", _fail) + context = NetworkBuildContext(options={}) + automation = { + "triggers": [{"trigger": "webhook", "webhook_id": "x", "local_only": False}], + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.garage_x"}} + ], + } + automations = [AutomationInput("automation.open", automation)] + assert _collect_critical_actions(home, context, automations) is None + inputs = HaNativeInputs(now=dt_util.utcnow(), automations=automations) + ha = ha_native_adapter(inputs, [], context).ha_security + assert ha["webhook_automations_critical"] == ["automation.open"] + + +async def test_a_disabled_cover_is_not_held_and_holds_no_neighbor( + hass: HomeAssistant, +) -> None: + """A disabled leftover cannot run; its id must not hold a similar one.""" + er.async_get(hass).async_get_or_create( + "cover", + "test", + "old", + suggested_object_id="living_room", + disabled_by=er.RegistryEntryDisabler.USER, + ) + hass.states.async_set( + "cover.living_room_blinds", "closed", {"device_class": "shade"} + ) + hass.states.async_set( + "cover.all_blinds", + "closed", + {"entity_id": ["cover.living_room", "cover.living_room_blinds"]}, + ) + rules = expand_critical_rules(hass, RULES) + for entity_id in ("cover.living_room_blinds", "cover.all_blinds"): + assert not _is_critical_action( + _open(entity_id), rules, "mcp__call_service", hass + ) + + +async def test_a_restored_placeholder_with_no_device_class_is_held( + hass: HomeAssistant, +) -> None: + """At startup HA writes `restored` states for entities not set up yet.""" + hass.states.async_set("cover.mqtt_grp", "unavailable", {"restored": True}) + # HA restores placeholders from the registry, so a restored entity has an + # entry, which holds its integration's class. + er.async_get(hass).async_get_or_create( + "cover", + "test", + "study", + suggested_object_id="study", + original_device_class="blind", + ) + hass.states.async_set( + "cover.study", "unavailable", {"restored": True, "device_class": "blind"} + ) + reached = _reaches(hass) + assert "cover.mqtt_grp" in reached + assert "cover.study" not in reached + + +async def test_a_member_ha_does_not_know_holds_its_group(hass: HomeAssistant) -> None: + """The member's integration is not set up; its id says nothing.""" + hass.states.async_set( + "cover.all_covers", "closed", {"entity_id": ["cover.msg100_abc"]} + ) + assert "cover.all_covers" in _reaches(hass) + + +@pytest.mark.parametrize( + ("name", "device_class", "held"), + [ + ("Gate1", None, True), + ("Garage2", None, True), + ("Gatekeeper", None, False), + ("French Door Blinds", "blind", False), + ("Patio Door Shade", "shade", False), + ("Garage Door", "shutter", True), + ], +) +async def test_names_digits_and_window_coverings( + hass: HomeAssistant, name: str, device_class: str | None, *, held: bool +) -> None: + attributes = {"friendly_name": name} + if device_class: + attributes["device_class"] = device_class + # The integration's class is in the registry (or the entity object). + er.async_get(hass).async_get_or_create( + "cover", + "test", + "relay", + suggested_object_id="relay_7", + original_device_class=device_class, + ) + hass.states.async_set("cover.relay_7", "closed", attributes) + assert ("cover.relay_7" in _reaches(hass)) is held + + +@pytest.mark.parametrize("source", ["name_by_user", "entity_name"]) +async def test_a_user_given_name_counts(hass: HomeAssistant, source: str) -> None: + config_entry = MockConfigEntry(domain="test") + config_entry.add_to_hass(hass) + device = dr.async_get(hass).async_get_or_create( + config_entry_id=config_entry.entry_id, + identifiers={("test", "relay")}, + name="Relay Board", + ) + entities = er.async_get(hass) + entry = entities.async_get_or_create( + "cover", "test", "relay", suggested_object_id="relay_1", device_id=device.id + ) + if source == "name_by_user": + dr.async_get(hass).async_update_device(device.id, name_by_user="Garage Door") + else: + entities.async_update_entity(entry.entity_id, name="Garage Door") + hass.states.async_set(entry.entity_id, "closed", {"device_class": "damper"}) + assert "cover.relay_1" in _reaches(hass) + + +async def test_a_rule_made_for_one_entity_matches_only_that_entity( + hass: HomeAssistant, +) -> None: + """`cover.a` is held; `cover.awning`, whose id starts the same, is not.""" + hass.states.async_set("cover.a", "closed", {"device_class": "garage"}) + hass.states.async_set("cover.awning", "closed", {"device_class": "awning"}) + rules = expand_critical_rules(hass, RULES) + assert _is_critical_action(_open("cover.a"), rules, "mcp__call_service", hass) + assert not _is_critical_action( + _open("cover.awning"), rules, "mcp__call_service", hass + ) + joined = _open("cover.awning, cover.a") + assert _is_critical_action(joined, rules, "mcp__call_service", hass) + + +def test_a_stored_rule_cannot_carry_the_exact_entity_marker() -> None: + """A user rule is never narrowed to one entity by the internal key.""" + stored = [ + {"domain": "lock", "service": "unlock", CRITICAL_RULE_EXACT_ENTITY: "lock.x"} + ] + rules = resolve_critical_action_policy( + {CONF_CRITICAL_ACTIONS: stored} + ).critical_actions + assert rules == [{"domain": "lock", "service": "unlock"}] + assert matches_critical_rule( + domain="lock", + service="unlock", + entity_ids=["lock.front"], + critical_actions=rules, + ) + + +async def test_a_script_opening_the_garage_group_is_held(home: HomeAssistant) -> None: + """The script screen reads the exact rules the same way scenes do.""" + assert await async_setup_component( + home, + "script", + { + "script": { + "open_all": { + "sequence": [ + { + "action": "cover.open_cover", + "target": {"entity_id": "cover.all_covers"}, + } + ] + } + } + }, + ) + await home.async_block_till_done() + assert runs_critical_action( + home, "script.open_all", expand_critical_rules(home, RULES) + ) + assert not runs_critical_action(home, "script.open_all", RULES) + + +async def test_add_automation_fails_closed_when_extending_fails( + home: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """An error in the extension stops the write; nothing is saved unchecked.""" + + def _fail(*_args: Any) -> list[dict[str, str]]: + raise RuntimeError + + monkeypatch.setattr(tools_module, "expand_critical_rules", _fail) + hashed, salt = hash_pin("1234") + cfg = cast( + "Any", + { + "hass": home, + "options": { + CONF_CRITICAL_ACTION_PIN_ENABLED: True, + CONF_CRITICAL_ACTION_PIN_HASH: hashed, + CONF_CRITICAL_ACTION_PIN_SALT: salt, + }, + "pending_actions": {}, + "thread_id": "t1", + }, + ) + validated = { + "actions": [ + {"action": "cover.open_cover", "target": {"entity_id": "cover.all_covers"}} + ] + } + with pytest.raises(RuntimeError): + _maybe_require_pin_for_automation( + cfg=cfg, validated_config=validated, ha_automation_config=dict(validated) + ) + + +async def test_the_gate_fails_closed_when_extending_fails( + home: HomeAssistant, monkeypatch: pytest.MonkeyPatch +) -> None: + """An error in the extension stops the call; it never runs unchecked.""" + + def _fail(*_args: Any) -> list[dict[str, str]]: + raise RuntimeError + + monkeypatch.setattr(graph_module, "expand_critical_rules", _fail) + ctx = cast( + "Any", + SimpleNamespace( + hass=home, + config={"configurable": {"thread_id": "t1"}}, + critical_actions=RULES, + pin_enabled=True, + pin_hash="hash", + pin_salt="salt", + pending_actions={}, + ), + ) + with pytest.raises(RuntimeError): + _prepare_tool_args( + "mcp__call_service", _open("cover.all_covers"), ctx, {"id": "c"} + )