If server allows anonymous access, allow anon access to WS too. (#1219)

* If server allows anonymous access, allow anon access to WS too.

* fix: Test logic was not quite right

* Extend timeouts (tests have been flaky)

* tst: Test Subscription without API key

Author: danielballan

tiled/_tests/conftest.py

@@ -350,9 +350,7 @@ def minio_uri():
         raise pytest.skip("No TILED_TEST_BUCKET configured")
 
 
-@pytest.fixture(scope="function")
-def tiled_websocket_context(tmpdir, redis_uri):
-    """Fixture that provides a Tiled context with websocket support."""
+def build_test_app(tmpdir, redis_uri, public=False):
     tree = from_uri(
         "sqlite:///:memory:",
         writable_storage=[
@@ -373,10 +371,25 @@ def tiled_websocket_context(tmpdir, redis_uri):
     )
     app = build_app(
         tree,
-        authentication=Authentication(single_user_api_key="secret"),
+        authentication=Authentication(
+            single_user_api_key="secret",
+            allow_anonymous_access=public,
+        ),
     )
+    return app
+
+
+@pytest.fixture(scope="function")
+def tiled_websocket_context(tmpdir, redis_uri):
+    """Fixture that provides a Tiled context with websocket support."""
+    with Context.from_app(build_test_app(tmpdir, redis_uri, public=False)) as context:
+        yield context
 
-    with Context.from_app(app) as context:
+
+@pytest.fixture(scope="function")
+def tiled_websocket_context_public(tmpdir, redis_uri):
+    """Fixture that provides a Tiled context with websocket support."""
+    with Context.from_app(build_test_app(tmpdir, redis_uri, public=True)) as context:
         yield context
 
 

tiled/_tests/test_subscription.py

@@ -68,7 +68,7 @@ def callback(update):
             streaming_node.write(new_arr)
 
         # Wait for all messages to be received
-        assert received_event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert received_event.wait(timeout=10.0), "Timeout waiting for messages"
 
         # Verify all updates received in order
         assert len(received) == 3
@@ -139,7 +139,7 @@ def callback(update):
             streaming_node.write(new_arr)
 
         # Wait for messages to be received
-        assert received_event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert received_event.wait(timeout=10.0), "Timeout waiting for messages"
 
         # Should only receive the 2 new updates (not the first one)
         assert len(received) == 2
@@ -196,7 +196,7 @@ def callback(update):
             streaming_node.write(new_arr)
 
         # Wait for all messages to be received
-        assert received_event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert received_event.wait(timeout=10.0), "Timeout waiting for messages"
 
         # Should receive: initial array + first update + 2 new updates = 4 total
         assert len(received) == 4
@@ -253,7 +253,7 @@ def child_metadata_updated_cb(update):
             time.sleep(0.1)
             unique_key = f"{uuid.uuid4().hex[:8]}"
             uploaded_nodes.append(client.create_container(unique_key))
-        assert created_3.wait(timeout=5.0), "Timeout waiting for messages"
+        assert created_3.wait(timeout=10.0), "Timeout waiting for messages"
         downloaded_nodes = list(client.values())
         for up, streamed, down in zip(uploaded_nodes, streamed_nodes, downloaded_nodes):
             pass
@@ -262,7 +262,7 @@ def child_metadata_updated_cb(update):
 
         assert len(child_metadata_updated_updates) == 0
         client.values().last().update_metadata({"color": "blue"})
-        assert received_event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert received_event.wait(timeout=10.0), "Timeout waiting for messages"
         assert len(child_metadata_updated_updates) == 1
 
 
@@ -283,7 +283,7 @@ def callback(sub):
         sub.stream_closed.add_callback(callback)
         assert not event.is_set()
         x.close_stream()
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
 
 
 def test_subscribe_to_disconnected(
@@ -305,7 +305,7 @@ def callback(sub):
         sub.disconnected.add_callback(callback)
         assert not event.is_set()
         sub.disconnect()
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
 
     # If the writer closes the stream, the client is disconnected.
     with x.subscribe().start_in_thread() as sub:
@@ -317,7 +317,7 @@ def callback(sub):
         sub.disconnected.add_callback(callback)
         assert not event.is_set()
         x.close_stream()
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
 
 
 def test_subscribe_to_array_registered_with_patch(tiled_websocket_context, tmp_path):
@@ -412,7 +412,7 @@ def on_child_created(update):
             content=safe_json_dump({"data_source": updated_data_source}),
             params=params,
         ).raise_for_status()
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         x.close_stream()
         client.close_stream()
         x.refresh()
@@ -503,7 +503,7 @@ def on_child_created(update):
                 }
             ),
         ).raise_for_status()
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         x.close_stream()
         client.close_stream()
         x.refresh()
@@ -533,12 +533,12 @@ def collect(update):
     sub = client[key].subscribe()
     sub.new_data.add_callback(collect)
     with sub.start_in_thread(1):
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         actual = updates[0].data()
         assert_frame_equal(actual, df1)
         event.clear()
         x.write(df2)
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         assert not updates[1].append
         actual_updated = updates[1].data()
         assert_frame_equal(actual_updated, df2)
@@ -565,14 +565,14 @@ def collect(update):
     sub.new_data.add_callback(collect)
     with sub.start_in_thread(1):
         x.append_partition(0, table1)
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         assert updates[0].append
         streamed1 = updates[0].data()
         streamed1_pyarrow = pyarrow.Table.from_pandas(streamed1, preserve_index=False)
         assert streamed1_pyarrow == table1
         event.clear()
         x.append_partition(0, table2)
-        assert event.wait(timeout=5.0), "Timeout waiting for messages"
+        assert event.wait(timeout=10.0), "Timeout waiting for messages"
         assert updates[1].append
         streamed2 = updates[1].data()
         streamed2_pyarrow = pyarrow.Table.from_pandas(streamed2, preserve_index=False)
@@ -641,3 +641,51 @@ def __call__(self, timeout=None):
 
         # Restore original recv before disconnecting to avoid cleanup issues
         subscription._websocket.recv = original_recv
+
+
+def test_subscribe_no_api_key_rejected(tiled_websocket_context):
+    "Private server does not allow anonymous user to subscribe."
+    context = tiled_websocket_context
+    client = from_context(context)
+
+    arr = np.arange(10)
+    streaming_node = client.write_array(arr, key="test_stream_immediate")
+
+    received_event = threading.Event()
+
+    def callback(update):
+        "Set event once any update has been received."
+        received_event.set()
+
+    # Any further requests will be unauthenticated.
+    context.api_key = None
+
+    subscription = streaming_node.subscribe()
+    subscription.new_data.add_callback(callback)
+
+    with pytest.raises(WebSocketDenialResponse):
+        subscription.start(0)
+
+
+def test_subscribe_no_api_key_public(tiled_websocket_context_public):
+    "Public server allows anonymous user to subscribe."
+    context = tiled_websocket_context_public
+    client = from_context(context)
+
+    arr = np.arange(10)
+    streaming_node = client.write_array(arr, key="test_stream_immediate")
+
+    received_event = threading.Event()
+
+    def callback(update):
+        "Set event once any update has been received."
+        received_event.set()
+
+    # Any further requests will be unauthenticated.
+    context.api_key = None
+
+    subscription = streaming_node.subscribe()
+    subscription.new_data.add_callback(callback)
+
+    with subscription.start_in_thread(0):
+        assert received_event.wait(timeout=10.0), "Timeout waiting for messages"

tiled/_tests/test_websockets.py

@@ -4,6 +4,7 @@
 import msgpack
 import numpy as np
 import pytest
+from starlette.testclient import WebSocketDenialResponse
 
 from ..client import from_context
 from ..config import parse_configs
@@ -381,7 +382,6 @@ def test_close_stream_not_found(tiled_websocket_context):
 
 def test_websocket_connection_wrong_api_key(tiled_websocket_context):
     """Test websocket connection with wrong API key fails with 401."""
-    from starlette.testclient import WebSocketDenialResponse
 
     context = tiled_websocket_context
     client = from_context(context)
@@ -402,6 +402,50 @@ def test_websocket_connection_wrong_api_key(tiled_websocket_context):
     assert exc_info.value.status_code == 401
 
 
+def test_websocket_connection_no_api_key(tiled_websocket_context):
+    """Test websocket connection with no API key fails with 401."""
+
+    context = tiled_websocket_context
+    client = from_context(context)
+    test_client = context.http_client
+
+    # Create streaming array node using correct key
+    arr = np.arange(10)
+    client.write_array(arr, key="test_auth_websocket")
+
+    # Strip API key so requests below are unauthenticated.
+    context.api_key = None
+
+    # Try to connect to websocket with no API key
+    with pytest.raises(WebSocketDenialResponse) as exc_info:
+        with test_client.websocket_connect(
+            "/api/v1/stream/single/test_auth_websocket?envelope_format=msgpack",
+        ):
+            pass
+
+    assert exc_info.value.status_code == 401
+
+
+def test_websocket_connection_public_no_api_key(tiled_websocket_context_public):
+    """Test websocket connection to a public server with no API key works."""
+    context = tiled_websocket_context_public
+    client = from_context(context)
+    test_client = context.http_client
+
+    # Create streaming array node using correct key
+    arr = np.arange(10)
+    client.write_array(arr, key="test_auth_websocket")
+
+    # Strip API key so requests below are unauthenticated.
+    context.api_key = None
+
+    # Try to connect to (public) websocket with no API key
+    with test_client.websocket_connect(
+        "/api/v1/stream/single/test_auth_websocket?envelope_format=msgpack",
+    ):
+        pass
+
+
 def test_close_stream_wrong_api_key(tiled_websocket_context):
     """Test close endpoint returns 403 with wrong API key."""
     context = tiled_websocket_context

tiled/client/stream.py

@@ -67,15 +67,18 @@ def __init__(self, http_client, uri: httpx.URL):
         self._websocket = None
         self._connection_lock = threading.Lock()
 
-    def connect(self, api_key: str, start: Optional[int] = None):
+    def connect(self, api_key: Optional[str], start: Optional[int] = None):
         """Connect to the websocket."""
+        params = self._uri.params
+        headers = {}
+        if api_key:
+            headers["Authorization"] = f"Apikey {api_key}"
+        if start is not None:
+            params = params.set("start", start)
         with self._connection_lock:
-            params = self._uri.params
-            if start is not None:
-                params = params.set("start", start)
             self._websocket = self._http_client.websocket_connect(
                 str(self._uri.copy_with(params=params)),
-                headers={"Authorization": f"Apikey {api_key}"},
+                headers=headers,
             )
             self._websocket.__enter__()
 
@@ -105,14 +108,17 @@ def __init__(self, http_client, uri: httpx.URL):
         self._uri = uri
         self._websocket = None
 
-    def connect(self, api_key: str, start: Optional[int] = None):
+    def connect(self, api_key: Optional[str], start: Optional[int] = None):
         """Connect to the websocket."""
         params = self._uri.params
+        headers = {}
+        if api_key:
+            headers["Authorization"] = f"Apikey {api_key}"
         if start is not None:
             params = params.set("start", start)
         self._websocket = connect(
             str(self._uri.copy_with(params=params)),
-            additional_headers={"Authorization": f"Apikey {api_key}"},
+            additional_headers=headers,
         )
 
     def recv(self, timeout=None):
@@ -341,7 +347,7 @@ def _connect(self, start: Optional[int] = None) -> None:
             )
             api_key = key_info["secret"]
         else:
-            # Use single-user API key.
+            # Use single-user API key or None (if unauthenticated).
             api_key = self.context.api_key
 
         # Connect using the websocket wrapper

tiled/server/authentication.py

@@ -284,12 +284,9 @@ async def get_current_access_tags(
 
 def get_api_key_websocket(
     authorization: Annotated[Optional[str], Header()] = None,
-):
+) -> Optional[str]:
     if authorization is None:
-        raise HTTPException(
-            status_code=HTTP_401_UNAUTHORIZED,
-            detail="An API key must be passed in the Authorization header",
-        )
+        return None
     scheme, api_key = get_authorization_scheme_param(authorization)
     if scheme.lower() != "apikey":
         raise HTTPException(
@@ -489,19 +486,30 @@ async def get_current_principal_from_api_key(
 
 async def get_current_principal_websocket(
     websocket: WebSocket,
-    api_key: str = Depends(get_api_key_websocket),
+    api_key: Optional[str] = Depends(get_api_key_websocket),
     settings: Settings = Depends(get_settings),
     db_factory: Callable[[], Optional[AsyncSession]] = Depends(
         get_database_session_factory
     ),
 ):
-    async with db_factory() as db:
-        principal = await get_current_principal_from_api_key(
-            api_key, websocket.app.state.authenticated, db, settings
-        )
-    if principal is None and websocket.app.state.authenticated:
-        raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid API key")
-    return principal
+    if api_key is not None:
+        async with db_factory() as db:
+            principal = await get_current_principal_from_api_key(
+                api_key, websocket.app.state.authenticated, db, settings
+            )
+        if (principal is None) and websocket.app.state.authenticated:
+            raise HTTPException(
+                status_code=HTTP_401_UNAUTHORIZED, detail="Invalid API key"
+            )
+        return principal
+    else:
+        if settings.allow_anonymous_access:
+            return None
+        else:
+            raise HTTPException(
+                status_code=HTTP_401_UNAUTHORIZED,
+                detail="No API key was provided with this request.",
+            )
 
 
 async def get_current_principal(