fix(streaming): address Greptile review on #4796

- A label longer than the request limit no longer becomes a container's
  name, so every published name is one the desktop endpoint accepts. The
  limit moves to handler.streaming.config as CONTAINER_NAME_MAX_LENGTH.
- A label equal to another container's key in any case no longer names
  its container, matching the case-insensitive lookup.
- An oversized broker manifest now fails the import build instead of
  being dropped, which lost the state typing of the base's entries.
- A broker error body is read against a 5 second deadline, like every
  other broker read.
- Type the _labelled test helper.

backend/endpoints/streaming.py

@@ -74,6 +74,7 @@ from handler.streaming.capabilities import (
     slot_capabilities,
 )
 from handler.streaming.config import (
+    CONTAINER_NAME_MAX_LENGTH,
     ResolvedContainer,
     configured_emulator,
     container_for_session,
@@ -176,12 +177,11 @@ class LoadStateRequest(BaseModel):
     slot: Annotated[int, Field(ge=1, le=MAX_SLOT)] = 1
 
 
-CONTAINER_KEY_MAX_LENGTH = 300
 CLAIMED_AT_MAX_LENGTH = 64
 
 # A claim is named by its container and the stamp it was taken at.
 ContainerQuery = Annotated[
-    str | None, Query(alias="container", max_length=CONTAINER_KEY_MAX_LENGTH)
+    str | None, Query(alias="container", max_length=CONTAINER_NAME_MAX_LENGTH)
 ]
 ClaimedAtQuery = Annotated[str | None, Query(max_length=CLAIMED_AT_MAX_LENGTH)]
 
@@ -190,7 +190,7 @@ class DesktopStreamingSessionRequest(BaseModel):
     # The container to open, by the name or key GET /streaming/containers
     # reports. Named rather than pooled: an admin configuring a container
     # needs that one, not whichever is free.
-    container: Annotated[str, Field(min_length=1, max_length=CONTAINER_KEY_MAX_LENGTH)]
+    container: Annotated[str, Field(min_length=1, max_length=CONTAINER_NAME_MAX_LENGTH)]
 
 
 def platform_capabilities(platform: str) -> PlatformCapabilities:

backend/handler/streaming/broker.py

@@ -100,6 +100,7 @@ _BROKER_JSON_MAX_BYTES = 4 * 1024 * 1024
 # read takes all of it; the log line and 502 detail keep just the head.
 _BROKER_ERROR_MAX_BYTES = 1024 * 1024
 _BROKER_ERROR_SHOWN_CHARS = 8 * 1024
+_BROKER_ERROR_READ_SECONDS = 5
 
 
 def broker_error_body(exc: urllib.error.HTTPError) -> str:
@@ -109,7 +110,9 @@ def broker_error_body(exc: urllib.error.HTTPError) -> str:
     closes it, and its body is as long as the broker cares to make it.
     """
     try:
-        return exc.read(_BROKER_ERROR_MAX_BYTES).decode(errors="replace")
+        deadline = time.monotonic() + _BROKER_ERROR_READ_SECONDS
+        body = _read_bounded(exc, _BROKER_ERROR_MAX_BYTES, deadline)
+        return body[:_BROKER_ERROR_MAX_BYTES].decode(errors="replace")
     except OSError as read_exc:
         log.warning("could not read broker error body, %s", read_exc)
         return ""

backend/handler/streaming/config.py

@@ -619,12 +619,16 @@ def _warn_about_later_pools(
             )
 
 
+# The longest name a container URL or desktop request accepts.
+CONTAINER_NAME_MAX_LENGTH = 300
+
+
 def _container_labels(resolved: Iterable[ResolvedContainer]) -> dict[str, str]:
     """Each claimable container's key to its own label, when it sets one."""
     labels: dict[str, str] = {}
     for container in resolved:
         label = (container.container_label or "").strip()
-        if container.key and label:
+        if container.key and label and len(label) <= CONTAINER_NAME_MAX_LENGTH:
             labels.setdefault(container.key, label)
     return labels
 
@@ -640,11 +644,11 @@ def _keys_by_label(labels: dict[str, str]) -> list[list[str]]:
 def _unique_labels(resolved: Sequence[ResolvedContainer]) -> dict[str, str]:
     """The labels that name exactly one container and are not another's key."""
     labels = _container_labels(resolved)
-    keys = {container.key for container in resolved if container.key}
+    folded_keys = {container.key.casefold() for container in resolved if container.key}
     return {
         group[0]: labels[group[0]]
         for group in _keys_by_label(labels)
-        if len(group) == 1 and labels[group[0]] not in keys
+        if len(group) == 1 and labels[group[0]].casefold() not in folded_keys
     }
 
 

backend/handler/streaming/imports.py

@@ -91,7 +91,7 @@ def _manifest_files(zf: zipfile.ZipFile) -> dict[str, dict[str, Any]]:
     except KeyError:
         return {}
     if info.file_size > _MAX_MANIFEST_BYTES:
-        return {}
+        raise ValueError("broker manifest exceeds its size limit")
     try:
         manifest = json.loads(_read_member(zf, info))
     except (json.JSONDecodeError, UnicodeDecodeError):

backend/tests/endpoints/test_streaming.py

@@ -57,6 +57,7 @@ from handler.streaming.capabilities import (
     state_transfer_limits,
 )
 from handler.streaming.config import (
+    CONTAINER_NAME_MAX_LENGTH,
     ResolvedContainer,
     _derive_broker_host,
     configured_emulator,
@@ -2212,7 +2213,7 @@ def test_the_session_platform_picks_the_config_entry_for_its_container():
 # ── Container names ───────────────────────────────────────────────────────────
 
 
-def _labelled(label, index):
+def _labelled(label: str | None, index: int) -> dict[str, Any]:
     """A webstation container on its own host, with the given label."""
     return {
         "host": f"http://192.168.1.{index}:3000",
@@ -2284,6 +2285,20 @@ def test_a_label_cannot_take_another_containers_key():
     assert key == _key_of(target)
 
 
+def test_a_label_matching_another_containers_key_in_any_case_names_neither():
+    target = _labelled("Target", 10)
+    impostor = _labelled(_key_of(target).upper(), 11)
+    with _streaming(target, impostor):
+        names = container_names()
+    assert names[_key_of(impostor)] == _key_of(impostor)
+
+
+def test_a_label_too_long_for_a_url_names_nothing():
+    entry = _labelled("x" * (CONTAINER_NAME_MAX_LENGTH + 1), 10)
+    with _streaming(entry):
+        assert container_names() == {_key_of(entry): _key_of(entry)}
+
+
 def test_an_unclaimable_container_has_no_name():
     broken = _labelled("Broken", 10) | {"host": "192.168.1.10:3000", "broker_host": ""}
     with _streaming(broken):
@@ -5804,7 +5819,9 @@ def test_build_import_archive_does_not_inflate_past_a_members_declared_size():
     assert peak < 16 * 1024 * 1024
 
 
-def test_manifest_files_ignores_a_manifest_over_the_size_cap():
+def test_manifest_files_rejects_a_manifest_over_the_size_cap():
+    """Dropping it would lose which entries are states, so a replaced state
+    would ride along as a save."""
     from tests._zipfile_shim import reload_zipfile
 
     reload_zipfile()
@@ -5815,8 +5832,9 @@ def test_manifest_files_ignores_a_manifest_over_the_size_cap():
     with (
         patch("handler.streaming.imports._MAX_MANIFEST_BYTES", 32),
         zipfile.ZipFile(buf) as zf,
+        pytest.raises(ValueError),
     ):
-        assert imports._manifest_files(zf) == {}
+        imports._manifest_files(zf)
 
 
 def test_build_import_archive_rebuilds_a_symlink_entry_as_a_regular_file():
@@ -8174,7 +8192,7 @@ def test_raise_http_error_raises_import_refused_as_a_typed_error():
         }
     ).encode()
     exc = _http_error(422)
-    with patch.object(exc, "read", return_value=payload):
+    with patch.object(exc, "read", side_effect=_reads(payload)):
         with pytest.raises(broker.ImportRefusedError) as raised:
             broker.raise_http_error(exc)
     assert raised.value.truncated == 0
@@ -8222,7 +8240,7 @@ def test_raise_http_error_parses_a_refusal_list_at_the_brokers_cap():
 
 def test_raise_http_error_keeps_a_long_plain_error_out_of_the_502_detail():
     exc = _http_error(500)
-    with patch.object(exc, "read", return_value=b"x" * (100 * 1024)):
+    with patch.object(exc, "read", side_effect=_reads(b"x" * (100 * 1024))):
         with pytest.raises(HTTPException) as raised:
             broker.raise_http_error(exc)
     assert len(raised.value.detail) < 9 * 1024
@@ -8231,12 +8249,21 @@ def test_raise_http_error_keeps_a_long_plain_error_out_of_the_502_detail():
 def test_raise_http_error_still_raises_502_for_a_plain_broker_error():
     """An ordinary broker error (not an import refusal) is still a plain 502."""
     exc = _http_error(500)
-    with patch.object(exc, "read", return_value=b"boom"):
+    with patch.object(exc, "read", side_effect=_reads(b"boom")):
         with pytest.raises(HTTPException) as raised:
             broker.raise_http_error(exc)
     assert raised.value.status_code == 502
 
 
+def test_broker_error_body_gives_up_on_a_slow_body():
+    exc = _http_error(500)
+    with (
+        patch.object(exc, "read", side_effect=_reads(b"boom")),
+        patch("handler.streaming.broker._BROKER_ERROR_READ_SECONDS", -1),
+    ):
+        assert broker.broker_error_body(exc) == ""
+
+
 def test_fetch_memory_card_returns_bytes(rom: Rom):
     resp = MagicMock()
     resp.__enter__.return_value.read.side_effect = _reads(b"card-bytes")