security(gate3): Security-Review-Haertung ueber 4 Pakete (luna-pro)
Phase 6 / Gate 3 — Security-Review auf den fertigen Code, 4 thematische Pakete: - P1 Secrets+Injection (client/phraser/diagnostics): validate haertet URL/Entity/ Unicode-Umgehungen (// , &entity; , Cc/Cf/Zl/Zp); kein Key-Leak. - P2 XSS+externe (card/ha_entity/frontend): kein XSS; DoS-Caps (Forecast 400, Karte 60). - P3 Mutator+Startup (coordinator/__init__/const): nicht-blockierender Startup via Background-Task, LLM-/Store-Ops defensiv, weather_entity_id-Guard, build_items NaN/inf+Caps, Warn-Log ohne Secret. - P4 Flows (config_flow): test_entity exception-frei, _clean_item defensiv, finite Bandgrenzen + sanitized_options-Normalisierung, echte bool-Coercion, Typ-Log. - 24 Findings, 14 uebernommen (mit Regressionstests), 10 verworfen mit Evidenz. Suite gruen 188/188. Gate 3 BESTANDEN. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Nora
parent
9a74d40194
commit
aae8c38a20
11 changed files with 223 additions and 44 deletions
26
LEDGER.md
26
LEDGER.md
|
|
@ -367,3 +367,29 @@ Pipes-and-Filters-Kern (`logic/`, hass-frei) in Ports-and-Adapters-Schale.
|
|||
echtem HA-Core (phcc) — Install→Beispiel-Set→Sensor-Outfit→Karten-Ressource→Service, NFR-1 (keine
|
||||
Netzlast im Default) + C-2 (keine Blocking-Warnung) asserted. Irreversible Schritte (E-3-Mirror,
|
||||
brands-PR, HACS-Default, Tag/Release) → dem Benutzer vorgelegt, nicht autonom.
|
||||
|
||||
### Gate 3 — Security-Review auf fertigen Code (2026-07-13)
|
||||
- **Modell:** `openai/gpt-5.6-luna-pro` via OpenRouter (qwen offline, dokumentierter Ersatz;
|
||||
Gate 3 ab Feature-Größe ohnehin Cloud-Eskalation). 4 thematische Pakete sequenziell,
|
||||
sicherheitskritischstes zuerst; je geschickte Dateien:
|
||||
- **P1 Secrets+LLM-Injection:** `llm/client.py`, `logic/phraser.py`, `diagnostics.py` — 2 Findings,
|
||||
beide übernommen (validate: `//`+HTML-Entity-Regex; Unicode Cc/Cf/Zl/Zp ablehnen). Kein Key-Leak.
|
||||
- **P2 XSS+externe Schnittstellen:** `www/what-to-wear-card.js`, `weather/ha_entity.py`, `frontend.py`
|
||||
— 2 Findings (DoS), beide übernommen (Forecast-Eintrag-Cap 400; Karte items/gaps-Cap 60). **Kein XSS.**
|
||||
- **P3 Mutator+Startup+Robustheit:** `coordinator.py`, `__init__.py`, `const.py` — 8 Findings,
|
||||
5 übernommen (F1 nicht-blockierender Startup via Background-Task; F2 LLM-/Store-Ops defensiv;
|
||||
F4 fehlende weather_entity_id guarden; F6 build_items NaN/inf+Caps+per-Subentry-try; F7 Warn-Log
|
||||
ohne Secret-Wert), 3 verworfen mit Evidenz (F3 config_entry-Shutdown+harmlos; F5 HA garantiert
|
||||
Mapping; F8 single_config_entry → ein Entry).
|
||||
- **P4 Flows+Validierung:** `config_flow.py` — 6 Findings, 5 übernommen (F2 test_entity exception-frei;
|
||||
F3 _clean_item defensiv; F4 finite Bandgrenzen + sanitized_options-Normalisierung; F5 echte
|
||||
bool-Coercion; F6 nur Exception-Typ loggen), 1 verworfen mit Evidenz (F1 single_config_entry-Manifest).
|
||||
- **Alle Findings selbst geprüft;** übernommene mit Regressionstests abgesichert (188 Tests grün).
|
||||
- **Status: Gate 3 BESTANDEN.**
|
||||
|
||||
## Gates (aktualisiert)
|
||||
| Gate | Gegenstand | Modell | Status |
|
||||
|---|---|---|---|
|
||||
| Gate 1 | PRD | luna-pro | ✅ bestanden |
|
||||
| Gate 2 | Architektur | luna-pro | ✅ bestanden |
|
||||
| Gate 3 | Code (Security, 4 Pakete) | luna-pro | ✅ bestanden |
|
||||
|
|
|
|||
|
|
@ -30,16 +30,22 @@ PLATFORMS: list[Platform] = [Platform.SENSOR]
|
|||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
provider = HAEntityProvider(hass, entry.data[CONF_WEATHER_ENTITY])
|
||||
entity_id = entry.data.get(CONF_WEATHER_ENTITY)
|
||||
provider = HAEntityProvider(hass, entity_id if isinstance(entity_id, str) else "")
|
||||
coordinator = WTWCoordinator(hass, entry, provider)
|
||||
# Do not wait on the weather: the coordinator's error contract yields a
|
||||
# fehler_prognose recommendation instead of blocking setup (AD-7).
|
||||
await coordinator.async_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
|
||||
# Do not block setup on the weather or the LLM: schedule the first refresh in
|
||||
# the background (cancelled on unload). The sensor fills in within seconds;
|
||||
# its error contract yields a fehler_prognose recommendation, not a stall (AD-7).
|
||||
entry.async_create_background_task(
|
||||
hass, coordinator.async_refresh(), f"{DOMAIN}_first_refresh"
|
||||
)
|
||||
|
||||
_register_switchover_listener(hass, entry, coordinator)
|
||||
_register_entity_listener(hass, entry, coordinator)
|
||||
if isinstance(entity_id, str) and entity_id:
|
||||
_register_entity_listener(hass, entry, coordinator, entity_id)
|
||||
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
|
||||
|
||||
await async_setup_card(hass)
|
||||
|
|
@ -106,10 +112,8 @@ def _register_switchover_listener(
|
|||
|
||||
|
||||
def _register_entity_listener(
|
||||
hass: HomeAssistant, entry: ConfigEntry, coordinator: WTWCoordinator
|
||||
hass: HomeAssistant, entry: ConfigEntry, coordinator: WTWCoordinator, entity_id: str
|
||||
) -> None:
|
||||
entity_id = entry.data[CONF_WEATHER_ENTITY]
|
||||
|
||||
@callback
|
||||
def _changed(event) -> None:
|
||||
new_state = event.data.get("new_state")
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from __future__ import annotations
|
|||
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
|
@ -35,6 +36,7 @@ from .const import (
|
|||
DOMAIN,
|
||||
SUBENTRY_TYPE_ITEM,
|
||||
default_options,
|
||||
sanitized_options,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigSubentryData
|
||||
|
||||
|
|
@ -53,12 +55,14 @@ CONF_EXAMPLE = "create_example"
|
|||
_UNAVAILABLE = ("unavailable", "unknown")
|
||||
|
||||
|
||||
async def async_test_entity(hass, entity_id: str, switchover: str | None = None) -> str | None:
|
||||
async def async_test_entity(hass, entity_id: Any, switchover: str | None = None) -> str | None:
|
||||
"""Return an error key, or None if the entity yields a usable forecast (FR-1.2).
|
||||
|
||||
``switchover`` is the effective switchover time; it decides whether the
|
||||
target date is today or tomorrow, which affects the coverage check.
|
||||
"""
|
||||
if not isinstance(entity_id, str) or not entity_id:
|
||||
return "no_weather_entity"
|
||||
state = hass.states.get(entity_id)
|
||||
if state is None:
|
||||
return "no_weather_entity"
|
||||
|
|
@ -72,14 +76,18 @@ async def async_test_entity(hass, entity_id: str, switchover: str | None = None)
|
|||
except ForecastUnsupported:
|
||||
return "no_forecast"
|
||||
except Exception as err: # noqa: BLE001 - any fetch failure is "temporary"
|
||||
# CancelledError is a BaseException and is intentionally not caught.
|
||||
_LOGGER.debug("Test fetch for %s failed: %s", entity_id, err)
|
||||
# CancelledError is a BaseException and is intentionally not caught. Log
|
||||
# only the exception type so a provider error cannot leak secrets.
|
||||
_LOGGER.debug("Test fetch for %s failed: %s", entity_id, type(err).__name__)
|
||||
return "temporary"
|
||||
|
||||
target, _ = target_date_for(dt_util.now(), switchover or default_options()["switchover_time"])
|
||||
if not pipeline.covers_target(raw, target):
|
||||
return "forecast_too_short"
|
||||
return None
|
||||
try:
|
||||
target, _ = target_date_for(dt_util.now(), switchover or default_options()["switchover_time"])
|
||||
covered = pipeline.covers_target(raw, target)
|
||||
except Exception as err: # noqa: BLE001 - malformed forecast structure
|
||||
_LOGGER.debug("Coverage check failed: %s", type(err).__name__)
|
||||
return "temporary"
|
||||
return None if covered else "forecast_too_short"
|
||||
|
||||
|
||||
class WhatToWearConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
|
|
@ -196,21 +204,48 @@ def _validate_item(user_input: dict) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
_TRUE = {"true", "1", "yes", "on"}
|
||||
|
||||
|
||||
def _to_bool(value: Any, default: bool) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return value.strip().lower() in _TRUE
|
||||
if isinstance(value, (int, float)):
|
||||
return bool(value)
|
||||
return default
|
||||
|
||||
|
||||
def _finite(value: Any) -> float | None:
|
||||
try:
|
||||
f = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return f if math.isfinite(f) else None
|
||||
|
||||
|
||||
def _clean_item(user_input: dict) -> dict:
|
||||
try:
|
||||
warmth = max(1, min(5, round(float(user_input.get("warmth", 3)))))
|
||||
except (TypeError, ValueError):
|
||||
warmth = 3
|
||||
data: dict = {
|
||||
"name": str(user_input["name"]).strip(),
|
||||
"category": user_input["category"],
|
||||
"warmth": max(1, min(5, round(float(user_input["warmth"])))),
|
||||
"waterproof": bool(user_input.get("waterproof", False)),
|
||||
"windproof": bool(user_input.get("windproof", False)),
|
||||
"sun_protection": bool(user_input.get("sun_protection", False)),
|
||||
"name": str(user_input.get("name", "")).strip()[:60],
|
||||
"category": user_input.get("category"),
|
||||
"warmth": warmth,
|
||||
"waterproof": _to_bool(user_input.get("waterproof"), False),
|
||||
"windproof": _to_bool(user_input.get("windproof"), False),
|
||||
"sun_protection": _to_bool(user_input.get("sun_protection"), False),
|
||||
"formality": "business" if user_input.get("formality") == "business" else "casual",
|
||||
"active": bool(user_input.get("active", True)),
|
||||
"active": _to_bool(user_input.get("active"), True),
|
||||
}
|
||||
if user_input.get("temp_min") is not None:
|
||||
data["temp_min"] = float(user_input["temp_min"])
|
||||
if user_input.get("temp_max") is not None:
|
||||
data["temp_max"] = float(user_input["temp_max"])
|
||||
tmin = _finite(user_input.get("temp_min")) if user_input.get("temp_min") is not None else None
|
||||
tmax = _finite(user_input.get("temp_max")) if user_input.get("temp_max") is not None else None
|
||||
if tmin is not None:
|
||||
data["temp_min"] = max(-60.0, min(60.0, tmin))
|
||||
if tmax is not None:
|
||||
data["temp_max"] = max(-60.0, min(60.0, tmax))
|
||||
return data
|
||||
|
||||
|
||||
|
|
@ -285,6 +320,8 @@ def _build_options(user_input: dict, current: dict) -> tuple[dict | None, str |
|
|||
edges = [float(user_input[k]) for k in _BAND_KEYS]
|
||||
except (TypeError, ValueError, KeyError):
|
||||
return None, "invalid_input"
|
||||
if not all(math.isfinite(e) for e in edges):
|
||||
return None, "invalid_input"
|
||||
if not all(a < b for a, b in zip(edges, edges[1:])):
|
||||
return None, "band_not_monotone"
|
||||
try:
|
||||
|
|
@ -310,7 +347,7 @@ def _build_options(user_input: dict, current: dict) -> tuple[dict | None, str |
|
|||
)
|
||||
# LLM tone section (Story 4.1). The key is never prefilled; an empty
|
||||
# field keeps the stored key, and disabling clears it (AD-5).
|
||||
enabled = bool(user_input.get("llm_enabled", current["llm_enabled"]))
|
||||
enabled = _to_bool(user_input.get("llm_enabled"), bool(current["llm_enabled"]))
|
||||
options["llm_enabled"] = enabled
|
||||
provider = user_input.get("llm_provider", current["llm_provider"])
|
||||
options["llm_provider"] = provider if provider in ("openai", "anthropic") else "openai"
|
||||
|
|
@ -322,7 +359,12 @@ def _build_options(user_input: dict, current: dict) -> tuple[dict | None, str |
|
|||
options["llm_api_key"] = new_key if new_key else current.get("llm_api_key", "")
|
||||
except (TypeError, ValueError, KeyError):
|
||||
return None, "invalid_input"
|
||||
return options, None
|
||||
# Final normalization: out-of-range / non-finite threshold values fall back
|
||||
# to their defaults so only schema-valid options are ever stored (AD-8/AD-23).
|
||||
key = options.get("llm_api_key", "")
|
||||
normalized = sanitized_options(options)
|
||||
normalized["llm_api_key"] = key # sanitizer only type-checks the key; keep it verbatim
|
||||
return normalized, None
|
||||
|
||||
|
||||
class WhatToWearOptionsFlow(OptionsFlow):
|
||||
|
|
|
|||
|
|
@ -136,7 +136,9 @@ def sanitized_options(stored: dict[str, Any], warn=None) -> dict[str, Any]:
|
|||
if _valid(key, value):
|
||||
opts[key] = value
|
||||
elif warn is not None:
|
||||
warn("Option %s has an unexpected value (%r); using default", key, value)
|
||||
# Never log the raw value of a secret, and never log an unbounded
|
||||
# value (log-DoS): report only the key and the value's type.
|
||||
warn("Option %s has an unexpected value (type %s); using default", key, type(value).__name__)
|
||||
return opts
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ from __future__ import annotations
|
|||
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
from dataclasses import replace
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
|
|
@ -71,18 +72,30 @@ def _float_or_none(value: Any) -> float | None:
|
|||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return float(value)
|
||||
result = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return result if math.isfinite(result) else None # reject NaN/inf
|
||||
|
||||
|
||||
# Upper bound on wardrobe items processed — far above any realistic wardrobe;
|
||||
# bounds CPU/memory if the subentry storage is corrupt or huge.
|
||||
_MAX_ITEMS = 500
|
||||
_MAX_NAME_LEN = 60
|
||||
|
||||
|
||||
def build_items(entry: ConfigEntry) -> list[Item]:
|
||||
"""Map ``item`` subentries to pure Item objects, defensively (AD-8/AD-22)."""
|
||||
items: list[Item] = []
|
||||
for subentry in entry.subentries.values():
|
||||
if len(items) >= _MAX_ITEMS:
|
||||
break
|
||||
if subentry.subentry_type != SUBENTRY_TYPE_ITEM:
|
||||
continue
|
||||
data = dict(subentry.data)
|
||||
try:
|
||||
data = dict(subentry.data)
|
||||
except (TypeError, ValueError):
|
||||
continue # corrupt (non-mapping) subentry data -> skip, never crash
|
||||
try:
|
||||
category = Category(data.get("category"))
|
||||
except ValueError:
|
||||
|
|
@ -93,7 +106,7 @@ def build_items(entry: ConfigEntry) -> list[Item]:
|
|||
items.append(
|
||||
Item(
|
||||
id=subentry.subentry_id,
|
||||
name=name.strip(),
|
||||
name=name.strip()[:_MAX_NAME_LEN],
|
||||
category=category,
|
||||
warmth=_int(data.get("warmth"), 1, 1, 5),
|
||||
waterproof=_bool(data.get("waterproof"), False),
|
||||
|
|
@ -145,7 +158,11 @@ class WTWCoordinator(DataUpdateCoordinator[Recommendation]):
|
|||
async def _async_update_data(self) -> Recommendation:
|
||||
async with self._lock:
|
||||
if not self._sig_loaded:
|
||||
self._stored_sig = await self._store.async_load()
|
||||
try:
|
||||
self._stored_sig = await self._store.async_load()
|
||||
except Exception as err: # noqa: BLE001 - corrupt store must not break setup
|
||||
_LOGGER.debug("Signature store load failed: %s", err)
|
||||
self._stored_sig = None
|
||||
self._sig_loaded = True
|
||||
|
||||
options = self._options()
|
||||
|
|
@ -181,12 +198,20 @@ class WTWCoordinator(DataUpdateCoordinator[Recommendation]):
|
|||
return rec # no-coverage fehler recommendation
|
||||
return self._fehler(language, target.isoformat(), now_iso, source, label)
|
||||
|
||||
# OK path — optionally apply the LLM tone (Story 4.3).
|
||||
rec = await self._apply_llm_tone(rec, options)
|
||||
# OK path — optionally apply the LLM tone (Story 4.3). Any failure
|
||||
# here must never break the update; fall back to the rule text.
|
||||
try:
|
||||
rec = await self._apply_llm_tone(rec, options)
|
||||
except Exception as err: # noqa: BLE001
|
||||
_LOGGER.debug("LLM tone step failed: %s", err)
|
||||
rec = replace(rec, data_notes=rec.data_notes + ("llm_error",))
|
||||
|
||||
changed = self._stored_sig != rec.signature
|
||||
if changed:
|
||||
await self._store.async_save(rec.signature)
|
||||
try:
|
||||
await self._store.async_save(rec.signature)
|
||||
except Exception as err: # noqa: BLE001 - persistence is best-effort
|
||||
_LOGGER.debug("Signature store save failed: %s", err)
|
||||
self._stored_sig = rec.signature
|
||||
self.last_success_utc = dt_util.utcnow()
|
||||
return replace(rec, changed=changed)
|
||||
|
|
|
|||
|
|
@ -10,6 +10,8 @@ text. The LLM only rephrases; it can never change the selection.
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
from . import texts
|
||||
from .model import Recommendation
|
||||
|
|
@ -18,7 +20,12 @@ _MAX_LEN = 700
|
|||
_MIN_LEN = 20
|
||||
# Markup / code / markdown markers — fail closed to the rule text on any hit.
|
||||
_FORBIDDEN_CHARS = ("<", ">", "[", "]", "{", "}", "`", "*", "#", "|", "~")
|
||||
_FORBIDDEN_SUBSTRINGS = ("://", "www.", "javascript:", "data:", "mailto:", "&#", "<", ">")
|
||||
# Links / schemes (incl. protocol-relative "//") and HTML entities.
|
||||
_FORBIDDEN_SUBSTRINGS = ("://", "//", "www.", "javascript:", "data:", "mailto:")
|
||||
_ENTITY_RE = re.compile(r"&#?\w+;")
|
||||
# Character categories that are not plain flowing prose: control, format
|
||||
# (zero-width / bidi), and line/paragraph separators.
|
||||
_FORBIDDEN_CATEGORIES = frozenset({"Cc", "Cf", "Zl", "Zp"})
|
||||
|
||||
_SYSTEM = (
|
||||
"You rephrase a clothing recommendation into ONE short, friendly paragraph in "
|
||||
|
|
@ -68,13 +75,16 @@ def validate(text: object, rec: Recommendation) -> str | None:
|
|||
trimmed = text.strip()
|
||||
if not (_MIN_LEN <= len(trimmed) <= _MAX_LEN): # (b), (e)
|
||||
return None
|
||||
# (a) plain flowing prose: no control characters (incl. newline/tab/ANSI)
|
||||
if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in trimmed):
|
||||
# (a) plain flowing prose: no control/format/separator characters
|
||||
# (covers C0/DEL, U+0085, U+2028/U+2029, zero-width and bidi format chars)
|
||||
if any(unicodedata.category(ch) in _FORBIDDEN_CATEGORIES for ch in trimmed):
|
||||
return None
|
||||
if any(ch in trimmed for ch in _FORBIDDEN_CHARS): # (a) no markup/markdown/code
|
||||
return None
|
||||
if _ENTITY_RE.search(trimmed): # (a) no HTML entities (e.g. : t)
|
||||
return None
|
||||
lowered = trimmed.lower()
|
||||
if any(sub in lowered for sub in _FORBIDDEN_SUBSTRINGS): # (a) no links/schemes/entities
|
||||
if any(sub in lowered for sub in _FORBIDDEN_SUBSTRINGS): # (a) no links/schemes
|
||||
return None
|
||||
for name in _item_names(rec): # (c) every item named
|
||||
if name and name not in trimmed:
|
||||
|
|
|
|||
|
|
@ -27,6 +27,10 @@ _UNIT_ATTRS = {
|
|||
"pressure": "pressure_unit",
|
||||
}
|
||||
|
||||
# Cap on forecast entries processed — far above any real daily/hourly horizon;
|
||||
# bounds memory/CPU if a foreign entity returns a pathologically large list.
|
||||
_MAX_ENTRIES = 400
|
||||
|
||||
|
||||
class HAEntityProvider:
|
||||
"""WeatherProvider backed by a Home Assistant ``weather.*`` entity."""
|
||||
|
|
@ -75,7 +79,7 @@ class HAEntityProvider:
|
|||
if not isinstance(entries, (list, tuple)):
|
||||
return ()
|
||||
parsed: list[dict] = []
|
||||
for entry in entries:
|
||||
for entry in entries[:_MAX_ENTRIES]:
|
||||
if not isinstance(entry, Mapping):
|
||||
continue # skip malformed entries (no AttributeError)
|
||||
dt = _parse_dt(entry.get("datetime"))
|
||||
|
|
|
|||
|
|
@ -82,9 +82,10 @@ class WhatToWearCard extends HTMLElement {
|
|||
|
||||
this._headline.textContent = String(state.state || "");
|
||||
|
||||
// Items (name + reason), textContent only, defensively guarded.
|
||||
// Items (name + reason), textContent only, defensively guarded and capped
|
||||
// so a manipulated entity cannot flood the DOM.
|
||||
this._clear(this._items);
|
||||
const items = Array.isArray(a.items) ? a.items : [];
|
||||
const items = (Array.isArray(a.items) ? a.items : []).slice(0, 60);
|
||||
for (const item of items) {
|
||||
if (!item || typeof item !== "object") {
|
||||
continue;
|
||||
|
|
@ -103,7 +104,7 @@ class WhatToWearCard extends HTMLElement {
|
|||
}
|
||||
|
||||
// Gaps.
|
||||
const gaps = Array.isArray(a.gaps) ? a.gaps : [];
|
||||
const gaps = (Array.isArray(a.gaps) ? a.gaps : []).slice(0, 60);
|
||||
const labels = [];
|
||||
for (const g of gaps) {
|
||||
if (g && typeof g === "object") {
|
||||
|
|
|
|||
|
|
@ -98,6 +98,20 @@ def test_validate_requires_all_items_and_gaps() -> None:
|
|||
assert phraser.validate("Zieh das Thermoshirt und die Jeans an, guten Morgen dir.", rec) is None
|
||||
|
||||
|
||||
def test_validate_rejects_advanced_bypasses() -> None:
|
||||
rec = _rec(item_names=("Thermoshirt", "Jeans"))
|
||||
body = "Thermoshirt und Jeans, Handschuhe fehlen. "
|
||||
# protocol-relative + HTML-entity encoded schemes
|
||||
assert phraser.validate(body + "//attacker.example", rec) is None
|
||||
assert phraser.validate(body + "http://x", rec) is None
|
||||
assert phraser.validate(body + "text", rec) is None
|
||||
# unicode format / bidi / zero-width / line separators
|
||||
assert phraser.validate(body + "ab", rec) is None # bidi override
|
||||
assert phraser.validate(body + "ab", rec) is None # zero-width space
|
||||
assert phraser.validate(body + "a
b", rec) is None # line separator
|
||||
assert phraser.validate(body + "a
b", rec) is None # next line
|
||||
|
||||
|
||||
def test_injection_catalog_outputs_rejected() -> None:
|
||||
# Malicious item names try to make the model emit markup/links/instructions.
|
||||
catalog = [
|
||||
|
|
|
|||
|
|
@ -83,3 +83,24 @@ def test_disclosure_present_in_translations() -> None:
|
|||
low = desc.lower()
|
||||
assert ("privacy" in low) or ("datenschutz" in low)
|
||||
assert ("backup" in low)
|
||||
|
||||
|
||||
def test_string_false_disables_and_clears_key() -> None:
|
||||
# a manipulated submit with llm_enabled="false" must disable + clear (Gate 3)
|
||||
current = {**default_options(), "llm_api_key": "existing-key", "llm_enabled": True}
|
||||
opts, err = _build_options(_base_input(llm_enabled="false"), current)
|
||||
assert err is None
|
||||
assert opts["llm_enabled"] is False
|
||||
assert opts["llm_api_key"] == ""
|
||||
|
||||
|
||||
def test_infinite_band_edge_rejected() -> None:
|
||||
opts, err = _build_options(_base_input(warmth_band_3=float("inf")), default_options())
|
||||
assert opts is None and err == "invalid_input"
|
||||
|
||||
|
||||
def test_out_of_range_threshold_normalized() -> None:
|
||||
# a manipulated over-range rain probability is normalized to the default
|
||||
opts, err = _build_options(_base_input(rain_prob_should=9999), default_options())
|
||||
assert err is None
|
||||
assert opts["rain_prob_should"] == 40 # sanitized back to default
|
||||
|
|
|
|||
|
|
@ -103,3 +103,33 @@ async def test_setup_survives_corrupt_options(hass: HomeAssistant) -> None:
|
|||
assert o["warmth_band_limits"] == [0.0, 8.0, 15.0, 22.0] # defaulted
|
||||
assert o["rain_prob_should"] == 40 # defaulted
|
||||
assert coordinator.data is not None # still produced a recommendation
|
||||
|
||||
|
||||
async def test_build_items_rejects_nan_and_corrupt(hass: HomeAssistant) -> None:
|
||||
from homeassistant.config_entries import ConfigSubentryData
|
||||
|
||||
from custom_components.what_to_wear.coordinator import build_items
|
||||
|
||||
_set_weather(hass)
|
||||
_register_forecast(hass)
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN, data={CONF_WEATHER_ENTITY: ENTITY},
|
||||
subentries_data=[ConfigSubentryData(
|
||||
data={"name": "Weird", "category": "top", "warmth": 2,
|
||||
"temp_min": float("nan"), "temp_max": float("inf")},
|
||||
subentry_type="item", title="Weird", unique_id=None)],
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
item = build_items(entry)[0]
|
||||
assert item.temp_min is None and item.temp_max is None # NaN/inf rejected
|
||||
|
||||
|
||||
async def test_setup_survives_missing_weather_entity(hass: HomeAssistant) -> None:
|
||||
# corrupt entry without the weather entity id must not crash setup (AD-2/AD-7)
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={}, options={})
|
||||
entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
assert coordinator.data is not None
|
||||
assert coordinator.data.status == "fehler_prognose"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue