250 lines
9.2 KiB
Python
250 lines
9.2 KiB
Python
"""URL-safety, NFC-duplicate, and trigger-config validation helpers.
|
|
|
|
Shared base layer for the task WS handlers (imported by tasks_crud) and by
|
|
several sibling modules (objects, documents, io) + tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from ..const import (
|
|
CONF_OBJECT,
|
|
CONF_OBJECT_NAME,
|
|
CONF_TASKS,
|
|
TriggerType,
|
|
)
|
|
from . import (
|
|
_get_object_entries,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SAFE_URL_SCHEMES = {"http", "https"}
|
|
|
|
|
|
def _is_safe_url(url: str | None) -> bool:
|
|
"""Reject javascript:, data:, protocol-relative and other dangerous URLs.
|
|
|
|
Only http/https and genuine path-relative URLs (no host) pass. ASCII control
|
|
characters and surrounding whitespace are stripped first, since urlparse and
|
|
browsers ignore them and they can otherwise mask a "//host" or scheme-less
|
|
host (e.g. ``" //evil.com"`` or ``"\t//evil.com"``).
|
|
"""
|
|
if not url:
|
|
return True
|
|
from urllib.parse import urlparse
|
|
|
|
cleaned = "".join(ch for ch in url if ch.isprintable()).strip()
|
|
if not cleaned:
|
|
return True
|
|
# Block protocol-relative URLs like //evil.com
|
|
if cleaned.startswith("//"):
|
|
return False
|
|
try:
|
|
parsed = urlparse(cleaned)
|
|
except Exception: # noqa: BLE001 - any malformed URL is rejected as unsafe
|
|
return False
|
|
scheme = parsed.scheme.lower()
|
|
if scheme in _SAFE_URL_SCHEMES:
|
|
return True
|
|
# An empty scheme is only safe for a true path-relative URL with no host.
|
|
return scheme == "" and not parsed.netloc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NFC tag uniqueness check
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _check_nfc_tag_duplicate(hass: HomeAssistant, nfc_tag_id: str, exclude_task_id: str | None = None) -> str | None:
|
|
"""Check if an NFC tag ID is already in use by another task.
|
|
|
|
Returns a warning message if duplicate, or None.
|
|
"""
|
|
for entry in _get_object_entries(hass):
|
|
tasks = entry.data.get(CONF_TASKS, {})
|
|
obj_name = entry.data.get(CONF_OBJECT, {}).get(CONF_OBJECT_NAME, "")
|
|
for tid, tdata in tasks.items():
|
|
if tid == exclude_task_id:
|
|
continue
|
|
if tdata.get("nfc_tag_id") == nfc_tag_id:
|
|
return f"NFC tag '{nfc_tag_id}' is already linked to task '{tdata.get('name', tid)}' on object '{obj_name}'"
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Trigger config validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Derived from the enum so the server validator can't drift from the type set.
|
|
_VALID_TRIGGER_TYPES = frozenset(t.value for t in TriggerType)
|
|
|
|
_TRIGGER_REQUIRED_FIELDS: dict[str, list[str]] = {
|
|
"threshold": [], # at least one of trigger_above/trigger_below checked below
|
|
"counter": ["trigger_target_value"],
|
|
"state_change": [],
|
|
"runtime": ["trigger_runtime_hours"],
|
|
"compound": [], # conditions validated separately
|
|
}
|
|
|
|
_TRIGGER_ALLOWED_KEYS: set[str] = {
|
|
"type",
|
|
"entity_id",
|
|
"entity_ids",
|
|
"entity_logic",
|
|
"attribute",
|
|
# threshold
|
|
"trigger_above",
|
|
"trigger_below",
|
|
"trigger_for_minutes",
|
|
# counter
|
|
"trigger_target_value",
|
|
"trigger_delta_mode",
|
|
# runtime
|
|
"trigger_runtime_hours",
|
|
"trigger_on_states",
|
|
# state_change
|
|
"trigger_from_state",
|
|
"trigger_to_state",
|
|
"trigger_target_changes",
|
|
# compound
|
|
"compound_logic",
|
|
"conditions",
|
|
# record a completion when the trigger clears itself (#53)
|
|
"auto_complete_on_recovery",
|
|
}
|
|
|
|
|
|
def _validate_trigger_config(
|
|
hass: HomeAssistant,
|
|
trigger_config: dict[str, Any],
|
|
) -> tuple[list[str], list[str]]:
|
|
"""Validate trigger_config structure.
|
|
|
|
Returns (errors, warnings).
|
|
Accepts both ``entity_id`` (str) and ``entity_ids`` (list[str]).
|
|
"""
|
|
from ..entity.triggers import normalize_entity_ids
|
|
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
|
|
# Trigger type
|
|
trigger_type = trigger_config.get("type", "threshold")
|
|
if trigger_type not in _VALID_TRIGGER_TYPES:
|
|
errors.append(f"Invalid trigger type '{trigger_type}'. Must be one of: {', '.join(sorted(_VALID_TRIGGER_TYPES))}")
|
|
return errors, warnings
|
|
|
|
# --- Compound triggers ---
|
|
if trigger_type == "compound":
|
|
return _validate_compound_trigger(hass, trigger_config)
|
|
|
|
# --- Non-compound: entity validation ---
|
|
entity_ids = normalize_entity_ids(trigger_config)
|
|
if not entity_ids:
|
|
errors.append("trigger_config requires entity_id or entity_ids")
|
|
else:
|
|
for eid in entity_ids:
|
|
state = hass.states.get(eid)
|
|
if state is None:
|
|
warnings.append(f"Entity {eid} does not exist (yet)")
|
|
elif state.state in ("unavailable", "unknown"):
|
|
warnings.append(f"Entity {eid} is currently '{state.state}'")
|
|
# Ensure entity_id is set for backwards compat
|
|
if not trigger_config.get("entity_id"):
|
|
trigger_config["entity_id"] = entity_ids[0]
|
|
# Always store entity_ids list
|
|
trigger_config["entity_ids"] = entity_ids
|
|
|
|
# Validate entity_logic
|
|
entity_logic = trigger_config.get("entity_logic")
|
|
if entity_logic is not None and entity_logic not in ("any", "all"):
|
|
errors.append(f"trigger_config.entity_logic must be 'any' or 'all', got '{entity_logic}'")
|
|
|
|
# Required fields per type
|
|
for field in _TRIGGER_REQUIRED_FIELDS[trigger_type]:
|
|
if trigger_config.get(field) is None:
|
|
errors.append(f"trigger_config.{field} is required for type '{trigger_type}'")
|
|
|
|
# Threshold: at least one of trigger_above or trigger_below
|
|
if trigger_type == "threshold":
|
|
if trigger_config.get("trigger_above") is None and trigger_config.get("trigger_below") is None:
|
|
errors.append("trigger_config requires at least one of 'trigger_above' or 'trigger_below' for type 'threshold'")
|
|
|
|
# Runtime: validate trigger_on_states if provided
|
|
if trigger_type == "runtime":
|
|
on_states = trigger_config.get("trigger_on_states")
|
|
if on_states is not None:
|
|
if not isinstance(on_states, list) or not all(isinstance(s, str) and s.strip() for s in on_states):
|
|
errors.append("trigger_config.trigger_on_states must be a list of non-empty strings")
|
|
elif len(on_states) == 0:
|
|
errors.append("trigger_config.trigger_on_states must not be empty when provided")
|
|
|
|
# Strip unknown keys to prevent data pollution
|
|
unknown = set(trigger_config) - _TRIGGER_ALLOWED_KEYS
|
|
for key in unknown:
|
|
del trigger_config[key]
|
|
|
|
# Coerce the recovery flag to a real bool (drop it when falsy so stored
|
|
# configs stay minimal — absence means off).
|
|
if "auto_complete_on_recovery" in trigger_config:
|
|
if trigger_config["auto_complete_on_recovery"]:
|
|
trigger_config["auto_complete_on_recovery"] = True
|
|
else:
|
|
del trigger_config["auto_complete_on_recovery"]
|
|
|
|
# Normalise state_change from/to: HA's state machine stores values lowercase
|
|
# ("on"/"off"/"home"/...). Users typing "ON"/"OFF" expect a match — same
|
|
# forgiving treatment as the runtime trigger, which lowercases trigger_on_states.
|
|
if trigger_type == "state_change":
|
|
for key in ("trigger_from_state", "trigger_to_state"):
|
|
val = trigger_config.get(key)
|
|
if isinstance(val, str):
|
|
stripped = val.strip().lower()
|
|
if stripped:
|
|
trigger_config[key] = stripped
|
|
else:
|
|
trigger_config.pop(key, None)
|
|
|
|
return errors, warnings
|
|
|
|
|
|
def _validate_compound_trigger(
|
|
hass: HomeAssistant,
|
|
trigger_config: dict[str, Any],
|
|
) -> tuple[list[str], list[str]]:
|
|
"""Validate a compound trigger config."""
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
|
|
compound_logic = trigger_config.get("compound_logic", "AND").upper()
|
|
if compound_logic not in ("AND", "OR"):
|
|
errors.append(f"compound_logic must be 'AND' or 'OR', got '{compound_logic}'")
|
|
|
|
conditions = trigger_config.get("conditions")
|
|
if not isinstance(conditions, list) or len(conditions) < 2:
|
|
errors.append("Compound trigger requires 'conditions' list with at least 2 entries")
|
|
return errors, warnings
|
|
|
|
for idx, condition in enumerate(conditions):
|
|
if not isinstance(condition, dict):
|
|
errors.append(f"Condition {idx} must be a dict")
|
|
continue
|
|
cond_type = condition.get("type", "threshold")
|
|
if cond_type == "compound":
|
|
errors.append(f"Condition {idx}: nested compound triggers are not allowed")
|
|
continue
|
|
# Validate each condition as a regular trigger
|
|
cond_errors, cond_warnings = _validate_trigger_config(hass, condition)
|
|
for err in cond_errors:
|
|
errors.append(f"Condition {idx}: {err}")
|
|
for warn in cond_warnings:
|
|
warnings.append(f"Condition {idx}: {warn}")
|
|
|
|
return errors, warnings
|