549 lines
21 KiB
Python
549 lines
21 KiB
Python
"""Defensive sanitization for the non-WebSocket write paths.
|
|
|
|
The WebSocket schemas enforce length and range caps on every str/int field at
|
|
the boundary, so the WS handlers need no sanitiser (they reject rather than
|
|
truncate). The other two write paths have no such guarantee and DO call the cap
|
|
helpers below right before persisting:
|
|
|
|
* **Config flow** — HA's selectors don't enforce lengths, so a malicious or
|
|
buggy programmatic flow caller could otherwise bloat ConfigEntry.data. Every
|
|
save handler caps.
|
|
* **Services** (``add_task`` / ``update_task``) — their voluptuous schemas
|
|
mirror the WS caps, but ``websocket/tasks_persist.py``'s
|
|
``async_create_task_simple`` / ``async_update_task_simple`` are plain Python
|
|
reachable in-process without going through a schema at all, so they cap too.
|
|
|
|
Keeping all three at parity is the point: a value one surface rejects must not
|
|
be writable through another.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from ..const import (
|
|
MAX_COST,
|
|
MAX_DATE_LENGTH,
|
|
MAX_DURATION_MINUTES,
|
|
MAX_ENTITY_SLUG_LENGTH,
|
|
MAX_ICON_LENGTH,
|
|
MAX_ID_LENGTH,
|
|
MAX_INTERVAL_DAYS,
|
|
MAX_META_LENGTH,
|
|
MAX_NAME_LENGTH,
|
|
MAX_NFC_TAG_LENGTH,
|
|
MAX_NOTIFY_ICON_LENGTH,
|
|
MAX_READING_UNIT_LENGTH,
|
|
MAX_SCHEDULE_TIME_LENGTH,
|
|
MAX_TEXT_LENGTH,
|
|
MAX_TYPE_LENGTH,
|
|
MAX_URL_LENGTH,
|
|
)
|
|
from .task_fields import EARLIEST_COMPLETION_RANGE
|
|
from .url_safety import is_safe_url
|
|
|
|
# Per-field cap for task dicts. Values mirror the voluptuous schemas in
|
|
# websocket/tasks.py so an admin who reaches the same field through the UI
|
|
# can't smuggle past a longer string than they could over the WS API.
|
|
_TASK_STR_LIMITS: dict[str, int] = {
|
|
"name": MAX_NAME_LENGTH,
|
|
"type": MAX_TYPE_LENGTH,
|
|
"schedule_type": MAX_TYPE_LENGTH,
|
|
"interval_anchor": MAX_TYPE_LENGTH,
|
|
"last_performed": MAX_DATE_LENGTH,
|
|
"notes": MAX_TEXT_LENGTH,
|
|
"documentation_url": MAX_URL_LENGTH,
|
|
"custom_icon": MAX_ICON_LENGTH,
|
|
# #185: per-task notification icon override ("mdi:…").
|
|
"notify_icon": MAX_NOTIFY_ICON_LENGTH,
|
|
"nfc_tag_id": MAX_NFC_TAG_LENGTH,
|
|
"responsible_user_id": MAX_META_LENGTH,
|
|
"entity_slug": MAX_ENTITY_SLUG_LENGTH,
|
|
"created_at": MAX_DATE_LENGTH,
|
|
# Lifecycle metadata that now round-trips through JSON import (audit
|
|
# 2026-07-11): an ISO timestamp + a short reason code. Length-capped so a
|
|
# crafted backup can't smuggle oversized strings past the importer.
|
|
"archived_at": MAX_META_LENGTH,
|
|
"archived_reason": MAX_META_LENGTH,
|
|
"schedule_time": MAX_SCHEDULE_TIME_LENGTH,
|
|
"priority": MAX_TYPE_LENGTH,
|
|
"reading_unit": MAX_READING_UNIT_LENGTH,
|
|
}
|
|
|
|
_OBJECT_STR_LIMITS: dict[str, int] = {
|
|
"name": MAX_NAME_LENGTH,
|
|
"manufacturer": MAX_META_LENGTH,
|
|
"model": MAX_META_LENGTH,
|
|
"serial_number": MAX_META_LENGTH,
|
|
"area_id": MAX_META_LENGTH,
|
|
"installation_date": MAX_DATE_LENGTH,
|
|
"warranty_expiry": MAX_DATE_LENGTH, # (#67)
|
|
"documentation_url": MAX_URL_LENGTH, # v1.4.0 #43
|
|
"notes": MAX_TEXT_LENGTH, # v1.4.10 #46
|
|
"ha_device_id": MAX_ID_LENGTH, # 2.19: link to an existing HA device
|
|
"parent_entry_id": MAX_ID_LENGTH, # 2.19: parent object (via_device)
|
|
# 2.20 pause + replace lineage — capped so an imported backup can't smuggle
|
|
# oversized strings into these (import copies them verbatim).
|
|
"paused_at": MAX_META_LENGTH, # ISO timestamp marker (presence = paused)
|
|
"paused_until": MAX_DATE_LENGTH, # auto-resume date
|
|
"predecessor_entry_id": MAX_ID_LENGTH,
|
|
"replaced_by_entry_id": MAX_ID_LENGTH,
|
|
}
|
|
|
|
_GROUP_STR_LIMITS: dict[str, int] = {
|
|
"name": MAX_NAME_LENGTH,
|
|
"description": MAX_TEXT_LENGTH,
|
|
}
|
|
|
|
|
|
def _cap_strings(d: dict[str, Any], limits: dict[str, int]) -> None:
|
|
"""Truncate string fields in-place to their per-field max length."""
|
|
for field, max_len in limits.items():
|
|
v = d.get(field)
|
|
if isinstance(v, str) and len(v) > max_len:
|
|
d[field] = v[:max_len]
|
|
|
|
|
|
# Keys a FRESH task copy must never inherit: per-task-unique identity
|
|
# (entity_slug, nfc_tag_id) and dynamic lifecycle state (history, schedule
|
|
# anchors, a one-shot due_override defer, adaptive tuning, archive markers).
|
|
# One list for task-duplicate, object-duplicate, and object-replace — the
|
|
# three hand-copied lists had drifted (task-duplicate kept archive markers,
|
|
# and none of them dropped due_override).
|
|
_FRESH_COPY_STRIP_KEYS = (
|
|
"entity_slug",
|
|
"nfc_tag_id",
|
|
"history",
|
|
"last_performed",
|
|
"last_planned_due",
|
|
"due_override",
|
|
"adaptive_config",
|
|
"archived_at",
|
|
"archived_reason",
|
|
# #170: a copy is a new task — it gets its own reference number.
|
|
"ref_no",
|
|
# Dynamic state like the history (storage._DYNAMIC_TASK_FIELDS, pinned by
|
|
# tests/test_dry_round4_runtime.py): a copy starts its phase cycle and its
|
|
# checklist from the beginning (DRY audit 2026-09-26 B).
|
|
"phase_cursor",
|
|
"checklist_progress",
|
|
)
|
|
|
|
|
|
def strip_object_reference(obj: dict[str, Any]) -> dict[str, Any]:
|
|
"""A copied / successor object is a new object (#170): drop the reference
|
|
number and the task counter so the setup pass numbers it afresh."""
|
|
obj.pop("ref_no", None)
|
|
obj.pop("next_task_ref", None)
|
|
return obj
|
|
|
|
|
|
def strip_task_runtime_state(task: dict[str, Any]) -> dict[str, Any]:
|
|
"""Remove unique/dynamic state from a task dict copy, in place.
|
|
|
|
Returns the same dict for fluent use.
|
|
"""
|
|
for key in _FRESH_COPY_STRIP_KEYS:
|
|
task.pop(key, None)
|
|
if isinstance(task.get("trigger_config"), dict):
|
|
task["trigger_config"].pop("_trigger_state", None)
|
|
return task
|
|
|
|
|
|
def cap_task_fields(task_data: dict[str, Any], *, keep_action_owner: bool = False) -> dict[str, Any]:
|
|
"""Truncate user-controllable strings + numerics on a task dict in-place.
|
|
|
|
Returns the same dict for fluent use. Mirrors the WS schema caps:
|
|
- String fields → individual length caps from `_TASK_STR_LIMITS`
|
|
- `interval_days` → 1..MAX_INTERVAL_DAYS (negative/zero coerced to 1)
|
|
- `warning_days` → 0..365
|
|
- `checklist` → list of strings, each ≤ 500 chars, list ≤ 100 items
|
|
|
|
``keep_action_owner``: the completion action's ``configured_by`` (the
|
|
user it runs as) survives. Only for RE-WRITES of a stored task whose
|
|
action the caller cannot change (the ``update_task`` service, the
|
|
options-flow task edit) and for the config flow's websocket step, whose
|
|
callers are server-side copies (object duplicate / replace) or an import
|
|
that stamped the owner itself. Stripping it there made an operator's
|
|
action run with system rights again after any unrelated edit (bug audit
|
|
2026-09-27, the SEC-2 regression). A client-supplied owner is never
|
|
trusted: every client write path stamps the connection user instead.
|
|
"""
|
|
_cap_strings(task_data, _TASK_STR_LIMITS)
|
|
_drop_unsafe_url(task_data)
|
|
|
|
iv = task_data.get("interval_days")
|
|
if isinstance(iv, int):
|
|
if iv < 1:
|
|
task_data["interval_days"] = 1
|
|
elif iv > MAX_INTERVAL_DAYS:
|
|
task_data["interval_days"] = MAX_INTERVAL_DAYS
|
|
|
|
wd = task_data.get("warning_days")
|
|
if isinstance(wd, int):
|
|
if wd < 0:
|
|
task_data["warning_days"] = 0
|
|
elif wd > 365:
|
|
task_data["warning_days"] = 365
|
|
|
|
ecd = task_data.get("earliest_completion_days")
|
|
if ecd is not None:
|
|
if not isinstance(ecd, int) or isinstance(ecd, bool):
|
|
task_data.pop("earliest_completion_days", None)
|
|
else:
|
|
lo, hi = EARLIEST_COMPLETION_RANGE
|
|
task_data["earliest_completion_days"] = max(lo, min(ecd, hi))
|
|
|
|
# #170: a reference number is a positive int or nothing — junk from a
|
|
# hand-edited backup must not poison the counters.
|
|
ref = task_data.get("ref_no")
|
|
if ref is not None and (not isinstance(ref, int) or isinstance(ref, bool) or ref < 1):
|
|
task_data.pop("ref_no", None)
|
|
|
|
cl = task_data.get("checklist")
|
|
if cl is not None:
|
|
if not isinstance(cl, list):
|
|
task_data.pop("checklist", None)
|
|
else:
|
|
from ..const import MAX_CHECKLIST_ITEM_LENGTH, MAX_CHECKLIST_ITEMS
|
|
|
|
cleaned = [item.strip()[:MAX_CHECKLIST_ITEM_LENGTH] for item in cl if isinstance(item, str)]
|
|
cleaned = [c for c in cleaned if c]
|
|
task_data["checklist"] = cleaned[:MAX_CHECKLIST_ITEMS]
|
|
|
|
# #161 phase 2: reading slots — validated by the shared helper; an
|
|
# empty/invalid list simply means "single value" and is dropped.
|
|
rs = task_data.get("readings")
|
|
if rs is not None:
|
|
from .reading_slots import sanitize_reading_slots
|
|
|
|
slots = sanitize_reading_slots(rs)
|
|
if slots:
|
|
task_data["readings"] = slots
|
|
else:
|
|
task_data.pop("readings", None)
|
|
lb = task_data.get("labels")
|
|
if lb is not None:
|
|
task_data["labels"] = sanitize_labels(lb)
|
|
|
|
# D#183: mirror targets — todo.* ids only, deduped, capped; an empty
|
|
# result drops the key (absence = mirroring off).
|
|
if task_data.get("mirror_todo_entities") is not None:
|
|
mirrors = sanitize_mirror_todo_entities(task_data["mirror_todo_entities"])
|
|
if mirrors:
|
|
task_data["mirror_todo_entities"] = mirrors
|
|
else:
|
|
task_data.pop("mirror_todo_entities", None)
|
|
|
|
if task_data.get("assignee_pool") is not None:
|
|
task_data["assignee_pool"] = sanitize_assignee_pool(task_data["assignee_pool"])
|
|
|
|
# Absence means "no rotation": a stored None / "" (the panel dialog sends
|
|
# null for "none") is dropped too — as a select default it made the
|
|
# options-flow task form unsaveable (bug audit 2026-09-26).
|
|
if "rotation_strategy" in task_data:
|
|
from ..const import ROTATION_STRATEGIES
|
|
|
|
if task_data["rotation_strategy"] not in ROTATION_STRATEGIES:
|
|
task_data.pop("rotation_strategy", None)
|
|
|
|
if task_data.get("required_completion_fields") is not None:
|
|
from .completion_requirements import sanitize_required_completion_fields
|
|
|
|
task_data["required_completion_fields"] = sanitize_required_completion_fields(
|
|
task_data["required_completion_fields"]
|
|
)
|
|
|
|
seed_rotation_assignee(task_data)
|
|
|
|
# v1.3.0: per-task on_complete_action — embedded HA service-call config.
|
|
# Strict shape: {service: "domain.name", target?: dict, data?: dict}.
|
|
# Drops the field entirely on any structural problem; the action layer
|
|
# treats absence as "no action configured" (not an error).
|
|
cap_action_field(task_data, keep_owner=keep_action_owner)
|
|
|
|
# v1.3.0: per-task quick_complete_defaults — pre-fill values used when
|
|
# the user scans the "quick complete" QR code. Schema mirrors the
|
|
# complete_maintenance kwargs.
|
|
cap_quick_complete_defaults_field(task_data)
|
|
|
|
return task_data
|
|
|
|
|
|
def parse_labels_text(raw: str) -> list[str]:
|
|
"""Split a comma/newline-separated labels string into a trimmed list.
|
|
|
|
The config flow enters labels as free text; the panel sends a real list.
|
|
This only splits + trims — dedup and per-label capping happen in
|
|
:func:`sanitize_labels` (via :func:`cap_task_fields`).
|
|
"""
|
|
parts = str(raw).replace("\n", ",").split(",")
|
|
return [p.strip() for p in parts if p.strip()]
|
|
|
|
|
|
def sanitize_labels(value: object) -> list[str]:
|
|
"""Clean a labels list: str items, trimmed, capped, deduped, ≤ MAX_LABELS."""
|
|
if not isinstance(value, list):
|
|
return []
|
|
from ..const import MAX_LABEL_LENGTH, MAX_LABELS
|
|
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for item in value:
|
|
if not isinstance(item, str):
|
|
continue
|
|
v = item.strip()[:MAX_LABEL_LENGTH]
|
|
if v and v not in seen:
|
|
seen.add(v)
|
|
out.append(v)
|
|
return out[:MAX_LABELS]
|
|
|
|
|
|
def sanitize_mirror_todo_entities(value: object) -> list[str]:
|
|
"""Clean a mirror-target list (D#183): ``todo.*`` entity ids only,
|
|
trimmed, deduped, at most ``MAX_MIRROR_TODO_LISTS``."""
|
|
if not isinstance(value, list):
|
|
return []
|
|
import re
|
|
|
|
from ..const import MAX_MIRROR_TODO_LISTS, MIRROR_TODO_ENTITY_PATTERN
|
|
|
|
pattern = re.compile(MIRROR_TODO_ENTITY_PATTERN)
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for item in value:
|
|
if not isinstance(item, str):
|
|
continue
|
|
v = item.strip()
|
|
if v and v not in seen and pattern.fullmatch(v):
|
|
seen.add(v)
|
|
out.append(v)
|
|
return out[:MAX_MIRROR_TODO_LISTS]
|
|
|
|
|
|
def seed_rotation_assignee(task_data: dict[str, Any]) -> None:
|
|
"""Ensure a rotation task always carries an effective assignee.
|
|
|
|
The rotation resolves "who is on duty" by writing the next pool member
|
|
into ``responsible_user_id`` on completion — the field EVERY user filter
|
|
reads (panel, Lovelace card, calendar card, saved views, per-user
|
|
notifications). But a rotation could be configured without an initial
|
|
assignee, leaving the task invisible to all of those until its first
|
|
completion ran ``advance_rotation`` (discussion #49). Seed the first
|
|
pool member when the assignee is missing — or no longer in the pool
|
|
(the pool was edited out from under the current assignee).
|
|
"""
|
|
pool = [u for u in task_data.get("assignee_pool") or [] if u]
|
|
if not pool or not task_data.get("rotation_strategy"):
|
|
return
|
|
current = task_data.get("responsible_user_id")
|
|
# A pool of one is an inert rotation — never seed a MISSING assignee for
|
|
# it. But a STALE one must still be corrected: the old blanket
|
|
# `len(pool) < 2` early-return meant a pool edited down to one member
|
|
# never ran the not-in-pool check, so a removed assignee kept the task
|
|
# forever (bug audit 2026-08-22).
|
|
if len(pool) < 2 and not current:
|
|
return
|
|
if current not in pool:
|
|
task_data["responsible_user_id"] = pool[0]
|
|
|
|
|
|
def sanitize_assignee_pool(value: object) -> list[str]:
|
|
"""Clean an assignee pool: str user-ids, trimmed, deduped, ≤ MAX_ASSIGNEE_POOL."""
|
|
if not isinstance(value, list):
|
|
return []
|
|
from ..const import MAX_ASSIGNEE_POOL
|
|
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for item in value:
|
|
if not isinstance(item, str):
|
|
continue
|
|
v = item.strip()[:MAX_META_LENGTH]
|
|
if v and v not in seen:
|
|
seen.add(v)
|
|
out.append(v)
|
|
return out[:MAX_ASSIGNEE_POOL]
|
|
|
|
|
|
# ─── v1.3.0: completion-action helpers ──────────────────────────────────
|
|
|
|
# Hard caps. service names rarely exceed 64 chars; data dicts intended
|
|
# for built-in HA services rarely exceed 1 KB serialised.
|
|
_MAX_SERVICE_NAME_LENGTH = 100
|
|
_MAX_ACTION_DATA_BYTES = 1024
|
|
_MAX_TARGET_FIELD_LENGTH = 200
|
|
|
|
# Privileged service domains an on-complete action may NOT call. A completion
|
|
# action is meant to nudge devices/notify — not run shell/scripts, reboot the
|
|
# host, purge the recorder, or stop HA. Blocking these closes an operator->admin
|
|
# escalation (an allowlisted operator setting an action that runs with system
|
|
# rights when the task is completed). Domain-specific services stay available
|
|
# (e.g. light.turn_on instead of the generic homeassistant.turn_on).
|
|
_FORBIDDEN_ACTION_DOMAINS = frozenset({"shell_command", "python_script", "hassio", "homeassistant", "recorder", "backup"})
|
|
|
|
|
|
def cap_action_field(task_data: dict[str, Any], *, keep_owner: bool = False) -> None:
|
|
"""Validate + truncate task_data['on_complete_action'] in-place.
|
|
|
|
Drops the entire field on any structural problem. Passes silently when
|
|
not present (it's optional). ``configured_by`` — the user the action runs
|
|
as (``action_listener``) — is server-stamped: kept only for the WS update
|
|
path's stored action (``keep_owner``), dropped from anything else a
|
|
client or an import could have written.
|
|
"""
|
|
import re
|
|
|
|
action = task_data.get("on_complete_action")
|
|
if action is None:
|
|
return
|
|
if not isinstance(action, dict):
|
|
task_data.pop("on_complete_action", None)
|
|
return
|
|
|
|
service = action.get("service")
|
|
if (
|
|
not isinstance(service, str)
|
|
or len(service) > _MAX_SERVICE_NAME_LENGTH
|
|
or not re.fullmatch(r"[a-z][a-z0-9_]*\.[a-z0-9_]+", service)
|
|
):
|
|
task_data.pop("on_complete_action", None)
|
|
return
|
|
|
|
# Reject privileged service domains (arbitrary code / host control).
|
|
if service.split(".", 1)[0] in _FORBIDDEN_ACTION_DOMAINS:
|
|
task_data.pop("on_complete_action", None)
|
|
return
|
|
|
|
cleaned: dict[str, Any] = {"service": service}
|
|
|
|
target = action.get("target")
|
|
if isinstance(target, dict):
|
|
cleaned_target: dict[str, Any] = {}
|
|
for key in ("entity_id", "device_id", "area_id", "label_id", "floor_id"):
|
|
v = target.get(key)
|
|
if isinstance(v, str) and 0 < len(v) <= _MAX_TARGET_FIELD_LENGTH:
|
|
cleaned_target[key] = v
|
|
elif isinstance(v, list):
|
|
cleaned_list = [s for s in v if isinstance(s, str) and 0 < len(s) <= _MAX_TARGET_FIELD_LENGTH]
|
|
if cleaned_list:
|
|
cleaned_target[key] = cleaned_list[:50] # cap target list length
|
|
if cleaned_target:
|
|
cleaned["target"] = cleaned_target
|
|
|
|
data = action.get("data")
|
|
if isinstance(data, dict):
|
|
# Cheap size guard via JSON serialisation
|
|
import json
|
|
|
|
try:
|
|
serialised = json.dumps(data)
|
|
except (TypeError, ValueError):
|
|
serialised = None
|
|
if serialised is not None and len(serialised) <= _MAX_ACTION_DATA_BYTES:
|
|
cleaned["data"] = data
|
|
|
|
owner = action.get(ACTION_OWNER_KEY)
|
|
if keep_owner and isinstance(owner, str) and 0 < len(owner) <= 64:
|
|
cleaned[ACTION_OWNER_KEY] = owner
|
|
|
|
task_data["on_complete_action"] = cleaned
|
|
|
|
|
|
# The user a completion action runs as (see helpers.action_listener).
|
|
ACTION_OWNER_KEY = "configured_by"
|
|
|
|
|
|
def stamp_action_owner(task_data: dict[str, Any], user_id: str | None) -> None:
|
|
"""Record who configured the task's completion action — it then runs
|
|
with that user's rights, so an operator cannot schedule an admin-only
|
|
service (bug audit 2026-09-26). Call after :func:`cap_action_field`."""
|
|
action = task_data.get("on_complete_action")
|
|
if isinstance(action, dict) and user_id:
|
|
action[ACTION_OWNER_KEY] = user_id
|
|
|
|
|
|
def settle_action_owner(task_data: dict[str, Any], stored_action: Any, user_id: str | None) -> None:
|
|
"""Owner of an EDITED task's completion action, after :func:`cap_action_field`.
|
|
|
|
An unchanged action (the panel dialog sends it back on every save) keeps
|
|
the owner it was stored with — an operator editing the notes must not
|
|
re-author an admin's action, nor may an edit drop the owner (the action
|
|
would then run with system rights). A changed action belongs to
|
|
``user_id``, the saving user (bug audit 2026-09-26 / 2026-09-27).
|
|
"""
|
|
stored_owner = stored_action.get(ACTION_OWNER_KEY) if isinstance(stored_action, dict) else None
|
|
new_action = task_data.get("on_complete_action")
|
|
unchanged = isinstance(stored_action, dict) and new_action == {
|
|
k: v for k, v in stored_action.items() if k != ACTION_OWNER_KEY
|
|
}
|
|
if isinstance(new_action, dict):
|
|
new_action.pop(ACTION_OWNER_KEY, None)
|
|
stamp_action_owner(task_data, stored_owner if unchanged else user_id)
|
|
|
|
|
|
def cap_quick_complete_defaults_field(task_data: dict[str, Any]) -> None:
|
|
"""Validate + truncate task_data['quick_complete_defaults'] in-place.
|
|
|
|
Drops malformed entries silently (per-field), preserves the rest.
|
|
"""
|
|
defaults = task_data.get("quick_complete_defaults")
|
|
if defaults is None:
|
|
return
|
|
if not isinstance(defaults, dict):
|
|
task_data.pop("quick_complete_defaults", None)
|
|
return
|
|
|
|
cleaned: dict[str, Any] = {}
|
|
|
|
notes = defaults.get("notes")
|
|
if isinstance(notes, str) and notes:
|
|
cleaned["notes"] = notes[:MAX_TEXT_LENGTH]
|
|
|
|
cost = defaults.get("cost")
|
|
if isinstance(cost, (int, float)) and 0 <= cost <= MAX_COST:
|
|
cleaned["cost"] = float(cost)
|
|
|
|
duration = defaults.get("duration")
|
|
if isinstance(duration, int) and 0 <= duration <= MAX_DURATION_MINUTES:
|
|
cleaned["duration"] = duration
|
|
|
|
from ..const import MaintenanceFeedback
|
|
|
|
feedback = defaults.get("feedback")
|
|
# Use the enum (needed / not_needed / not_sure) — a bare ("needed",
|
|
# "not_needed") literal silently dropped a valid not_sure feedback.
|
|
if feedback in tuple(MaintenanceFeedback):
|
|
cleaned["feedback"] = feedback
|
|
|
|
if cleaned:
|
|
task_data["quick_complete_defaults"] = cleaned
|
|
else:
|
|
task_data.pop("quick_complete_defaults", None)
|
|
|
|
|
|
def cap_object_fields(obj_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Truncate user-controllable strings on an object dict in-place."""
|
|
_cap_strings(obj_data, _OBJECT_STR_LIMITS)
|
|
_drop_unsafe_url(obj_data)
|
|
return obj_data
|
|
|
|
|
|
def _drop_unsafe_url(data: dict[str, Any]) -> None:
|
|
"""Drop a ``documentation_url`` that is not http(s)/path-relative —
|
|
every create and edit path runs through the cap helpers, so this is the
|
|
one place a ``javascript:`` link from a flow or an import stops (bug
|
|
audit 2026-09-26; the WS layer refuses it with an error before)."""
|
|
url = data.get("documentation_url")
|
|
if isinstance(url, str) and url and not is_safe_url(url):
|
|
import logging
|
|
|
|
logging.getLogger(__name__).warning("Dropped an unsafe documentation link (%.40s)", url)
|
|
data["documentation_url"] = None
|
|
|
|
|
|
def cap_group_fields(group_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Truncate user-controllable strings on a group dict in-place."""
|
|
_cap_strings(group_data, _GROUP_STR_LIMITS)
|
|
return group_data
|