sandbox: bound context-cache + channel-flood memory vectors (Phase 6)

Two unbounded-growth vectors closed:

1. Context cache on resolve. _resolve_context minted a fresh Context per
   unknown context_id but never enforced _CONTEXT_CACHE_MAX (only
   _remember_context did), so a sandbox flooding distinct unknown ids grew
   the cache without bound. Factor a single _store_context() helper used by
   both paths so the cap + expiry-ordering apply uniformly.

2. Channel read backpressure (BOTH mirrors). The reader create_task'd a
   handler per inbound frame; the inflight semaphore caps *running*
   handlers but queued tasks — each pinning a decoded payload up to
   MAX_FRAME_SIZE — grew without bound under a flood. _dispatch now sheds
   over a DEFAULT_MAX_QUEUED cap on inflight handler tasks: inbound calls
   are rejected with a ChannelOverloaded error frame, pushes dropped.
   Responses are always handled inline above the gate, so backpressure
   never starves a reply.

The channel.py edit is applied byte-identically to both hand-mirrored
copies (homeassistant/components/sandbox/channel.py and
sandbox/hass_client/hass_client/channel.py), rebased on top of plan
#1's Channel.close() fix, in this separate commit.

Design note: shed (reject/drop over a bounded cap) rather than block the
reader on the semaphore. Blocking the shared reader would deadlock the
documented nested-call pattern — a handler that issues channel.call() and
awaits its reply through the same reader would stall it once all slots are
held by such handlers (real on the client mirror: a call_service handler
doing a store_save round-trip to main). Shedding bounds memory without
that liveness hazard and stays safe in both mirrors.

homeassistant/components/sandbox/bridge.py

@@ -363,6 +363,24 @@ class SandboxBridge:
                 break
             del contexts[key]
 
+    @callback
+    def _store_context(self, key: str, context: Context, now: datetime) -> None:
+        """Insert/refresh a cache entry and enforce the size backstop.
+
+        Shared by :meth:`_remember_context` (real main-issued contexts) and the
+        miss path of :meth:`_resolve_context` (fresh contexts minted for an
+        unknown id). Keeps the cache ordered by expiry (move-to-end) and caps
+        its size so neither path can grow it without bound — a sandbox flooding
+        distinct unknown ``context_id``s is bounded the same as legitimate
+        traffic.
+        """
+        contexts = self._contexts
+        contexts[key] = _CachedContext(context, now + _CONTEXT_TTL)
+        contexts.move_to_end(key)
+        # TTL + low volume keep this tiny; the cap is only a sanity backstop.
+        while len(contexts) > _CONTEXT_CACHE_MAX:
+            contexts.popitem(last=False)
+
     @callback
     def _remember_context(self, context: Context | None) -> None:
         """Record a Context main is handing down to the sandbox.
@@ -378,12 +396,7 @@ class SandboxBridge:
             return
         now = dt_util.utcnow()
         self._prune_contexts(now)
-        contexts = self._contexts
-        contexts[context.id] = _CachedContext(context, now + _CONTEXT_TTL)
-        contexts.move_to_end(context.id)
-        # TTL + low volume keep this tiny; the cap is only a sanity backstop.
-        while len(contexts) > _CONTEXT_CACHE_MAX:
-            contexts.popitem(last=False)
+        self._store_context(context.id, context, now)
 
     @callback
     def _resolve_context(self, context_id: str | None) -> Context:
@@ -413,8 +426,7 @@ class SandboxBridge:
         if cached is not None:
             return cached.context
         context = Context(user_id=None)
-        self._contexts[context_id] = _CachedContext(context, now + _CONTEXT_TTL)
-        self._contexts.move_to_end(context_id)
+        self._store_context(context_id, context, now)
         return context
 
     async def _handle_register_entity(

homeassistant/components/sandbox/channel.py

@@ -59,6 +59,15 @@ Handler = Callable[[Any], Awaitable[Any]]
 
 DEFAULT_MAX_INFLIGHT = 16
 
+# Hard cap on inbound handler tasks (running + queued) a flood may create. The
+# inflight semaphore bounds how many handlers *run*; without this, queued
+# handler tasks — each pinning a decoded payload up to MAX_FRAME_SIZE — could
+# still grow without bound under a frame-flood. Over the cap, inbound calls are
+# rejected with an error frame and pushes are dropped (responses are always
+# handled inline, so backpressure never starves a reply). Generous enough that
+# honest fan-out never trips it.
+DEFAULT_MAX_QUEUED = 1024
+
 # Hard cap on a single frame's body. A length prefix larger than this aborts
 # the channel rather than letting a compromised sandbox allocate the host to
 # death (same hardening spirit as the auth key check).
@@ -324,6 +333,7 @@ class Channel:
         codec: Codec | None = None,
         name: str = "channel",
         max_inflight: int = DEFAULT_MAX_INFLIGHT,
+        max_queued: int = DEFAULT_MAX_QUEUED,
     ) -> None:
         """Wrap a reader/writer pair (or a transport) into a channel.
 
@@ -354,6 +364,7 @@ class Channel:
         self._write_lock = asyncio.Lock()
         self._inflight: set[asyncio.Task[None]] = set()
         self._inflight_sem = asyncio.Semaphore(max_inflight)
+        self._max_queued = max_queued
 
     @classmethod
     def from_transport(
@@ -517,6 +528,24 @@ class Channel:
                 )
             return
 
+        # Backpressure: responses are handled inline above and never shed.
+        # Bound the inbound handler tasks (each pins a decoded payload) so a
+        # frame-flood throttles here instead of growing memory without bound —
+        # reject calls with an error frame, silently drop pushes.
+        if len(self._inflight) >= self._max_queued:
+            if frame.kind is FrameKind.CALL:
+                self._spawn_handler(
+                    self._write(
+                        Frame.error_response(
+                            frame.id,
+                            "channel overloaded",
+                            "ChannelOverloaded",
+                            msg_type=frame.type,
+                        )
+                    )
+                )
+            return
+
         handler = self._handlers.get(frame.type)
 
         if frame.kind is FrameKind.PUSH:

sandbox/hass_client/hass_client/channel.py

@@ -33,6 +33,15 @@ Handler = Callable[[Any], Awaitable[Any]]
 
 DEFAULT_MAX_INFLIGHT = 16
 
+# Hard cap on inbound handler tasks (running + queued) a flood may create. The
+# inflight semaphore bounds how many handlers *run*; without this, queued
+# handler tasks — each pinning a decoded payload up to MAX_FRAME_SIZE — could
+# still grow without bound under a frame-flood. Over the cap, inbound calls are
+# rejected with an error frame and pushes are dropped (responses are always
+# handled inline, so backpressure never starves a reply). Generous enough that
+# honest fan-out never trips it.
+DEFAULT_MAX_QUEUED = 1024
+
 # Hard cap on a single frame's body. A length prefix larger than this aborts
 # the channel rather than letting a compromised peer allocate the process to
 # death.
@@ -296,6 +305,7 @@ class Channel:
         codec: Codec | None = None,
         name: str = "channel",
         max_inflight: int = DEFAULT_MAX_INFLIGHT,
+        max_queued: int = DEFAULT_MAX_QUEUED,
     ) -> None:
         """Wrap a reader/writer pair (or a transport) into a channel.
 
@@ -326,6 +336,7 @@ class Channel:
         self._write_lock = asyncio.Lock()
         self._inflight: set[asyncio.Task[None]] = set()
         self._inflight_sem = asyncio.Semaphore(max_inflight)
+        self._max_queued = max_queued
 
     @classmethod
     def from_transport(
@@ -482,6 +493,24 @@ class Channel:
                 )
             return
 
+        # Backpressure: responses are handled inline above and never shed.
+        # Bound the inbound handler tasks (each pins a decoded payload) so a
+        # frame-flood throttles here instead of growing memory without bound —
+        # reject calls with an error frame, silently drop pushes.
+        if len(self._inflight) >= self._max_queued:
+            if frame.kind is FrameKind.CALL:
+                self._spawn_handler(
+                    self._write(
+                        Frame.error_response(
+                            frame.id,
+                            "channel overloaded",
+                            "ChannelOverloaded",
+                            msg_type=frame.type,
+                        )
+                    )
+                )
+            return
+
         handler = self._handlers.get(frame.type)
 
         if frame.kind is FrameKind.PUSH: