sandbox: adversarial forged-frame tests per gate (Phase 7)

One forged-frame test per trust-boundary gate:

* fire_event: core (homeassistant_stop/call_service/state_changed) and
  unowned-domain (zha_event/hue_event) events are dropped, never reach the bus.
* register_service: an unowned domain (persistent_notification) is rejected.
* register_entity: a foreign entry_id (entry.sandbox != group) is rejected;
  a device_info colliding with a foreign entry's device is refused (no merge).
* translation: a forged foreign domain returned alongside the owned one is
  dropped; only the requested ∩ returned survives.
* store_save: an overlong key and an oversized value are rejected; nothing
  hits disk.
* context cache: a flood of distinct unknown context_ids stays bounded by
  _CONTEXT_CACHE_MAX.
* channel backpressure: over the max_queued cap, inbound calls are shed with
  a ChannelOverloaded error and the inflight set stops growing.

make_channel_pair gained max_queued_a/b passthrough for the backpressure test.

tests/components/sandbox/_helpers.py

@@ -48,13 +48,17 @@ def make_channel_pair(
     name_b: str = "b",
     max_inflight_a: int | None = None,
     max_inflight_b: int | None = None,
+    max_queued_a: int | None = None,
+    max_queued_b: int | None = None,
     use_json: bool = False,
 ) -> tuple[Channel, Channel]:
     """Return two channels connected to each other in-memory.
 
-    ``max_inflight_a`` / ``max_inflight_b`` override the per-side
-    handler concurrency cap when set; otherwise the channel's default
-    applies. Useful for exercising the bounded-semaphore path.
+    ``max_inflight_a`` / ``max_inflight_b`` override the per-side handler
+    concurrency cap; ``max_queued_a`` / ``max_queued_b`` override the per-side
+    inflight (queued + running) shed cap. Both default to the channel's own
+    defaults. Useful for exercising the bounded-semaphore and read-backpressure
+    paths.
 
     The pair speaks protobuf by default (production parity, so real
     handlers receive typed messages). ``use_json=True`` falls back to the
@@ -65,12 +69,16 @@ def make_channel_pair(
     reader_b = asyncio.StreamReader()
     writer_a = _LoopbackWriter(reader_b)  # a's writes → b's reader
     writer_b = _LoopbackWriter(reader_a)
-    kwargs_a: dict[str, int] = (
-        {"max_inflight": max_inflight_a} if max_inflight_a is not None else {}
-    )
-    kwargs_b: dict[str, int] = (
-        {"max_inflight": max_inflight_b} if max_inflight_b is not None else {}
-    )
+    kwargs_a: dict[str, int] = {}
+    kwargs_b: dict[str, int] = {}
+    if max_inflight_a is not None:
+        kwargs_a["max_inflight"] = max_inflight_a
+    if max_inflight_b is not None:
+        kwargs_b["max_inflight"] = max_inflight_b
+    if max_queued_a is not None:
+        kwargs_a["max_queued"] = max_queued_a
+    if max_queued_b is not None:
+        kwargs_b["max_queued"] = max_queued_b
     codec_a = JsonCodec() if use_json else ProtobufCodec()
     codec_b = JsonCodec() if use_json else ProtobufCodec()
     channel_a = Channel(reader_a, writer_a, name=name_a, codec=codec_a, **kwargs_a)  # type: ignore[arg-type]

tests/components/sandbox/test_bridge.py

@@ -8,6 +8,7 @@ import voluptuous as vol
 
 from homeassistant.components.sandbox._proto import sandbox_pb2 as pb
 from homeassistant.components.sandbox.bridge import (
+    _CONTEXT_CACHE_MAX,
     SandboxBridge,
     SandboxEntityDescription,
     _translate_remote_error,
@@ -753,3 +754,160 @@ async def test_fire_event_lands_on_main_bus(hass: HomeAssistant) -> None:
         await sandbox_channel.close()
 
     assert received == [{"command": "on", "device_ieee": "0a:0b:0c"}]
+
+
+# ---------------------------------------------------------------------------
+# Adversarial / forged-frame gates (trust-boundary hardening)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "event_type",
+    [
+        "homeassistant_stop",  # hard-denied core event
+        "call_service",  # hard-denied core event
+        "state_changed",  # hard-denied core event
+        "zha_event",  # unowned domain — group owns nothing here
+        "hue_event",  # unowned domain
+    ],
+)
+async def test_fire_event_forged_type_dropped(
+    hass: HomeAssistant, event_type: str
+) -> None:
+    """A compromised sandbox cannot fire core/foreign events on main's bus."""
+    # The group owns only ``demo`` — nothing that namespaces the forged events.
+    MockConfigEntry(domain="demo", title="Demo", sandbox="built-in").add_to_hass(hass)
+    _bridge, main_channel, sandbox_channel = await _wire(hass)
+
+    received: list[Any] = []
+
+    @callback
+    def _listener(event: Any) -> None:
+        received.append(event)
+
+    hass.bus.async_listen(event_type, _listener)
+
+    try:
+        forged = pb.FireEvent(event_type=event_type)
+        forged.event_data.update({"injected": True})
+        await sandbox_channel.push("sandbox/fire_event", forged)
+        # Let the push handler run; the event must never reach the bus.
+        for _ in range(20):
+            await asyncio.sleep(0)
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+    assert received == []
+
+
+async def test_register_service_unowned_domain_rejected(
+    hass: HomeAssistant,
+) -> None:
+    """A sandbox cannot register a service in a domain it doesn't own."""
+    # The group owns ``demo``; ``persistent_notification`` is not its to claim.
+    MockConfigEntry(domain="demo", title="Demo", sandbox="built-in").add_to_hass(hass)
+    _bridge, main_channel, sandbox_channel = await _wire(hass)
+
+    try:
+        with pytest.raises(ChannelRemoteError, match="not owned by group"):
+            await sandbox_channel.call(
+                "sandbox/register_service",
+                pb.RegisterService(
+                    domain="persistent_notification",
+                    service="create",
+                    supports_response="none",
+                ),
+            )
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+    assert not hass.services.has_service("persistent_notification", "create")
+
+
+async def test_register_entity_foreign_entry_rejected(
+    hass: HomeAssistant,
+) -> None:
+    """A sandbox cannot attach entities to an entry routed to another group."""
+    # The entry belongs to a *different* sandbox group than this bridge's.
+    foreign = MockConfigEntry(domain="light", title="Victim", sandbox="other-group")
+    foreign.add_to_hass(hass)
+    _bridge, main_channel, sandbox_channel = await _wire(hass)  # group="built-in"
+
+    try:
+        with pytest.raises(ChannelRemoteError, match="not owned by group"):
+            await sandbox_channel.call(
+                "sandbox/register_entity",
+                make_entity_description(
+                    entry_id=foreign.entry_id,
+                    domain="light",
+                    sandbox_entity_id="light.victim",
+                ),
+            )
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+
+async def test_register_entity_foreign_device_merge_rejected(
+    hass: HomeAssistant,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """A sandbox cannot merge into a device owned by a foreign config entry."""
+    # A victim integration owns a device with identifier ("victim", "dev-1").
+    victim = MockConfigEntry(domain="victim", title="Victim")
+    victim.add_to_hass(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=victim.entry_id,
+        identifiers={("victim", "dev-1")},
+        name="Victim Device",
+    )
+
+    # The sandbox owns its own entry but forges device_info colliding with the
+    # victim's identifiers to try to graft onto its device.
+    owned = MockConfigEntry(domain="light", title="Owned", sandbox="built-in")
+    owned.add_to_hass(hass)
+    _bridge, main_channel, sandbox_channel = await _wire(hass)
+
+    try:
+        with pytest.raises(ChannelRemoteError, match="outside group"):
+            await sandbox_channel.call(
+                "sandbox/register_entity",
+                make_entity_description(
+                    entry_id=owned.entry_id,
+                    domain="light",
+                    sandbox_entity_id="light.evil",
+                    unique_id="evil",
+                    supported_features=0,
+                    capabilities={"supported_color_modes": ["onoff"]},
+                    initial_state=STATE_ON,
+                    initial_attributes={"color_mode": "onoff"},
+                    device_info={"identifiers": [["victim", "dev-1"]]},
+                ),
+            )
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+    # The victim's device still belongs only to the victim entry.
+    device = device_registry.async_get_device(identifiers={("victim", "dev-1")})
+    assert device is not None
+    assert device.config_entries == {victim.entry_id}
+
+
+async def test_context_cache_bounded_under_id_flood(
+    hass: HomeAssistant,
+) -> None:
+    """Resolving a flood of distinct unknown context_ids stays cache-bounded."""
+    bridge, main_channel, sandbox_channel = await _wire(hass)
+
+    try:
+        # Each unknown id mints a fresh Context cached under that key; without
+        # eviction on the resolve path this would grow without bound.
+        for i in range(_CONTEXT_CACHE_MAX * 2):
+            bridge._resolve_context(f"forged-{i}")
+        assert len(bridge._contexts) <= _CONTEXT_CACHE_MAX
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()

tests/components/sandbox/test_channel.py

@@ -339,3 +339,54 @@ async def test_close_after_eof_still_closes_transport() -> None:
     # Idempotent: a second close() does no further transport work.
     await channel.close()
     assert transport.close_calls == 1
+
+
+async def test_read_backpressure_sheds_over_queued_cap() -> None:
+    """A frame-flood is bounded: over the cap, calls are shed not queued.
+
+    With a tiny ``max_queued`` and handlers that never return, the reader keeps
+    draining the wire but stops growing handler tasks — once the cap is hit,
+    further calls come back as ``ChannelOverloaded`` instead of piling up
+    unbounded decoded payloads.
+    """
+    channel_a, channel_b = make_channel_pair(
+        max_inflight_b=2, max_queued_b=3, use_json=True
+    )
+    channel_a.start()
+    channel_b.start()
+    release = asyncio.Event()
+
+    async def never(_payload: dict) -> int:
+        await release.wait()
+        return 1
+
+    channel_b.register("test/never", never)
+
+    pending: list[asyncio.Task] = []
+    try:
+        # Fill the queue cap (3) with handlers parked on the semaphore/await.
+        pending = [
+            asyncio.create_task(channel_a.call("test/never", {"idx": i}))
+            for i in range(3)
+        ]
+        for _ in range(50):
+            if len(channel_b._inflight) >= 3:
+                break
+            await asyncio.sleep(0)
+        assert len(channel_b._inflight) == 3
+
+        # The next call is shed with a ChannelOverloaded error frame.
+        with pytest.raises(ChannelRemoteError) as err:
+            await asyncio.wait_for(
+                channel_a.call("test/never", {"idx": 99}), timeout=2.0
+            )
+        assert err.value.error_type == "ChannelOverloaded"
+
+        # The reader threw the excess away rather than growing the inflight set.
+        assert len(channel_b._inflight) == 3
+    finally:
+        release.set()
+        for task in pending:
+            task.cancel()
+        await channel_a.close()
+        await channel_b.close()

tests/components/sandbox/test_store.py

@@ -23,7 +23,11 @@ from typing import Any
 import pytest
 
 from homeassistant.components.sandbox._proto import sandbox_pb2 as pb
-from homeassistant.components.sandbox.bridge import SandboxBridge
+from homeassistant.components.sandbox.bridge import (
+    _STORE_MAX_KEY_LENGTH,
+    _STORE_MAX_VALUE_BYTES,
+    SandboxBridge,
+)
 from homeassistant.components.sandbox.channel import Channel, ChannelRemoteError
 from homeassistant.components.sandbox.messages import struct_to_dict
 from homeassistant.core import HomeAssistant
@@ -183,6 +187,43 @@ async def test_store_rejects_missing_key(hass: HomeAssistant) -> None:
         await sandbox_channel.close()
 
 
+async def test_store_rejects_overlong_key(hass: HomeAssistant) -> None:
+    """A key past the length cap is rejected before any file IO."""
+    _bridge, main_channel, sandbox_channel = await _wire(hass)
+    try:
+        with pytest.raises(ChannelRemoteError, match="too long"):
+            await sandbox_channel.call(
+                "sandbox/store_save",
+                pb.StoreSave(key="k" * (_STORE_MAX_KEY_LENGTH + 1)),
+            )
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+
+async def test_store_rejects_oversized_value(hass: HomeAssistant) -> None:
+    """A value past the per-key byte cap is rejected, nothing hits disk."""
+    _bridge, main_channel, sandbox_channel = await _wire(hass)
+    oversized = "x" * (_STORE_MAX_VALUE_BYTES + 1)
+    save = pb.StoreSave(key="too_big")
+    save.data.update(
+        {
+            "version": 1,
+            "minor_version": 1,
+            "key": "too_big",
+            "data": {"blob": oversized},
+        }
+    )
+    try:
+        with pytest.raises(ChannelRemoteError, match="too large"):
+            await sandbox_channel.call("sandbox/store_save", save)
+    finally:
+        await main_channel.close()
+        await sandbox_channel.close()
+
+    assert not _store_path(hass, "built-in", "too_big").exists()
+
+
 async def test_store_groups_are_isolated(hass: HomeAssistant) -> None:
     """Two bridges with different groups never share a key namespace."""
     _bridge_builtin, main_a, sandbox_a = await _wire(hass, group="built-in")

tests/components/sandbox/test_translation.py

@@ -55,6 +55,30 @@ def _serving_channel(
     return main, sandbox
 
 
+def _forging_channel(
+    hass: HomeAssistant, extra: dict[str, dict[str, Any]]
+) -> tuple[Channel, Channel]:
+    """A channel whose sandbox end injects ``extra`` domains it wasn't asked for.
+
+    Stands in for a compromised sandbox: it returns strings for every
+    requested domain *plus* the foreign ``extra`` domains, attempting to
+    poison a co-resident integration's frontend strings.
+    """
+    main, sandbox = make_channel_pair()
+
+    async def _handler(msg: pb.GetTranslations) -> pb.GetTranslationsResult:
+        result = pb.GetTranslationsResult(language=msg.language)
+        for domain in msg.domains:
+            result.strings.update({domain: _CUSTOM_STRINGS})
+        result.strings.update(extra)
+        return result
+
+    sandbox.register(MSG_GET_TRANSLATIONS, _handler)
+    main.start()
+    sandbox.start()
+    return main, sandbox
+
+
 def _provider_with_bridge(
     hass: HomeAssistant, *, group: str, channel: Channel | None
 ) -> SandboxTranslationProvider:
@@ -201,3 +225,22 @@ async def test_unload_plain_entry_does_not_invalidate(hass: HomeAssistant) -> No
 
     assert result is None
     invalidate.assert_not_called()
+
+
+async def test_foreign_returned_domain_is_dropped(hass: HomeAssistant) -> None:
+    """Strings for a domain the group wasn't asked to resolve are discarded."""
+    mock_integration(hass, MockModule("my_custom"), built_in=False)
+    MockConfigEntry(domain="my_custom", sandbox="custom").add_to_hass(hass)
+    # The sandbox forges a "hue" entry alongside its own "my_custom".
+    main, sandbox = _forging_channel(hass, {"hue": {"title": "PWNED"}})
+    provider = _provider_with_bridge(hass, group="custom", channel=main)
+
+    try:
+        result = await provider.async_get_translations(["en"], {"my_custom"})
+    finally:
+        await main.close()
+        await sandbox.close()
+
+    # Only the requested ∩ returned domain survives; the foreign "hue" is gone.
+    assert result == {"en": {"my_custom": _CUSTOM_STRINGS}}
+    assert "hue" not in result.get("en", {})