Protect state.as_dict from mutation

homeassistant/components/diagnostics/util.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterable, Mapping
-from typing import Any, TypeVar, cast
+from typing import Any, TypeVar, cast, overload
 
 from homeassistant.core import callback
 
@@ -11,6 +11,16 @@
 T = TypeVar("T")
 
 
+@overload
+def async_redact_data(data: Mapping, to_redact: Iterable[Any]) -> dict:  # type: ignore
+    ...
+
+
+@overload
+def async_redact_data(data: T, to_redact: Iterable[Any]) -> T:
+    ...
+
+
 @callback
 def async_redact_data(data: T, to_redact: Iterable[Any]) -> T:
     """Redact sensitive data in a dict."""
@@ -25,7 +35,7 @@ def async_redact_data(data: T, to_redact: Iterable[Any]) -> T:
     for key, value in redacted.items():
         if key in to_redact:
             redacted[key] = REDACTED
-        elif isinstance(value, dict):
+        elif isinstance(value, Mapping):
             redacted[key] = async_redact_data(value, to_redact)
         elif isinstance(value, list):
             redacted[key] = [async_redact_data(item, to_redact) for item in value]

homeassistant/core.py

@@ -24,7 +24,6 @@
 import re
 import threading
 from time import monotonic
-from types import MappingProxyType
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -83,6 +82,7 @@
     run_callback_threadsafe,
     shutdown_run_callback_threadsafe,
 )
+from .util.read_only_dict import ReadOnlyDict
 from .util.timeout import TimeoutManager
 from .util.unit_system import IMPERIAL_SYSTEM, METRIC_SYSTEM, UnitSystem
 
@@ -1049,12 +1049,12 @@ def __init__(
 
         self.entity_id = entity_id.lower()
         self.state = state
-        self.attributes = MappingProxyType(attributes or {})
+        self.attributes = ReadOnlyDict(attributes or {})
         self.last_updated = last_updated or dt_util.utcnow()
         self.last_changed = last_changed or self.last_updated
         self.context = context or Context()
         self.domain, self.object_id = split_entity_id(self.entity_id)
-        self._as_dict: dict[str, Collection[Any]] | None = None
+        self._as_dict: ReadOnlyDict[str, Collection[Any]] | None = None
 
     @property
     def name(self) -> str:
@@ -1063,7 +1063,7 @@ def name(self) -> str:
             "_", " "
         )
 
-    def as_dict(self) -> dict[str, Collection[Any]]:
+    def as_dict(self) -> ReadOnlyDict[str, Collection[Any]]:
         """Return a dict representation of the State.
 
         Async friendly.
@@ -1077,14 +1077,16 @@ def as_dict(self) -> dict[str, Collection[Any]]:
                 last_updated_isoformat = last_changed_isoformat
             else:
                 last_updated_isoformat = self.last_updated.isoformat()
-            self._as_dict = {
-                "entity_id": self.entity_id,
-                "state": self.state,
-                "attributes": dict(self.attributes),
-                "last_changed": last_changed_isoformat,
-                "last_updated": last_updated_isoformat,
-                "context": self.context.as_dict(),
-            }
+            self._as_dict = ReadOnlyDict(
+                {
+                    "entity_id": self.entity_id,
+                    "state": self.state,
+                    "attributes": self.attributes,
+                    "last_changed": last_changed_isoformat,
+                    "last_updated": last_updated_isoformat,
+                    "context": ReadOnlyDict(self.context.as_dict()),
+                }
+            )
         return self._as_dict
 
     @classmethod
@@ -1343,7 +1345,7 @@ def async_set(
             last_changed = None
         else:
             same_state = old_state.state == new_state and not force_update
-            same_attr = old_state.attributes == MappingProxyType(attributes)
+            same_attr = old_state.attributes == attributes
             last_changed = old_state.last_changed if same_state else None
 
         if same_state and same_attr:
@@ -1404,7 +1406,7 @@ def __init__(
         """Initialize a service call."""
         self.domain = domain.lower()
         self.service = service.lower()
-        self.data = MappingProxyType(data or {})
+        self.data = ReadOnlyDict(data or {})
         self.context = context or Context()
 
     def __repr__(self) -> str:

homeassistant/util/__init__.py

@@ -2,14 +2,13 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Callable, Coroutine, Iterable, KeysView
+from collections.abc import Callable, Coroutine, Iterable, KeysView, Mapping
 from datetime import datetime, timedelta
 from functools import wraps
 import random
 import re
 import string
 import threading
-from types import MappingProxyType
 from typing import Any, TypeVar
 
 import slugify as unicode_slug
@@ -53,7 +52,7 @@ def slugify(text: str | None, *, separator: str = "_") -> str:
 
 def repr_helper(inp: Any) -> str:
     """Help creating a more readable string representation of objects."""
-    if isinstance(inp, (dict, MappingProxyType)):
+    if isinstance(inp, Mapping):
         return ", ".join(
             f"{repr_helper(key)}={repr_helper(item)}" for key, item in inp.items()
         )

homeassistant/util/read_only_dict.py

@@ -0,0 +1,23 @@
+"""Read only dictionary."""
+from typing import Any, TypeVar
+
+
+def _readonly(*args: Any, **kwargs: Any) -> Any:
+    """Raise an exception when a read only dict is modified."""
+    raise RuntimeError("Cannot modify ReadOnlyDict")
+
+
+Key = TypeVar("Key")
+Value = TypeVar("Value")
+
+
+class ReadOnlyDict(dict[Key, Value]):
+    """Read only version of dict that is compatible with dict types."""
+
+    __setitem__ = _readonly
+    __delitem__ = _readonly
+    pop = _readonly
+    popitem = _readonly
+    clear = _readonly
+    update = _readonly
+    setdefault = _readonly

tests/test_core.py

@@ -39,6 +39,7 @@
     ServiceNotFound,
 )
 import homeassistant.util.dt as dt_util
+from homeassistant.util.read_only_dict import ReadOnlyDict
 from homeassistant.util.unit_system import METRIC_SYSTEM
 
 from tests.common import async_capture_events, async_mock_service
@@ -377,10 +378,14 @@ def test_state_as_dict():
         "last_updated": last_time.isoformat(),
         "state": "on",
     }
-    assert state.as_dict() == expected
+    as_dict_1 = state.as_dict()
+    assert isinstance(as_dict_1, ReadOnlyDict)
+    assert isinstance(as_dict_1["attributes"], ReadOnlyDict)
+    assert isinstance(as_dict_1["context"], ReadOnlyDict)
+    assert as_dict_1 == expected
     # 2nd time to verify cache
     assert state.as_dict() == expected
-    assert state.as_dict() is state.as_dict()
+    assert state.as_dict() is as_dict_1
 
 
 async def test_eventbus_add_remove_listener(hass):

tests/util/test_read_only_dict.py

@@ -0,0 +1,36 @@
+"""Test read only dictionary."""
+import json
+
+import pytest
+
+from homeassistant.util.read_only_dict import ReadOnlyDict
+
+
+def test_read_only_dict():
+    """Test read only dictionary."""
+    data = ReadOnlyDict({"hello": "world"})
+
+    with pytest.raises(RuntimeError):
+        data["hello"] = "universe"
+
+    with pytest.raises(RuntimeError):
+        data["other_key"] = "universe"
+
+    with pytest.raises(RuntimeError):
+        data.pop("hello")
+
+    with pytest.raises(RuntimeError):
+        data.popitem()
+
+    with pytest.raises(RuntimeError):
+        data.clear()
+
+    with pytest.raises(RuntimeError):
+        data.update({"yo": "yo"})
+
+    with pytest.raises(RuntimeError):
+        data.setdefault("yo", "yo")
+
+    assert isinstance(data, dict)
+    assert dict(data) == {"hello": "world"}
+    assert json.dumps(data) == json.dumps({"hello": "world"})