fix not checking ack auth

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
 dependencies = [
     "fastapi>=0.115",
     "aiohttp>=3.11",
-    "lonelypsp>=0.0.38"
+    "lonelypsp>=0.0.40"
 ]
 requires-python = ">=3.9"
 

requirements.txt

@@ -10,7 +10,7 @@ colorama==0.4.6
 fastapi==0.115.5
 frozenlist==1.5.0
 idna==3.10
-lonelypsp==0.0.39
+lonelypsp==0.0.40
 multidict==6.1.0
 mypy==1.13.0
 mypy-extensions==1.0.0

src/lonelypss/routes/notify.py

@@ -444,7 +444,7 @@ async def handle_trusted_notify(
 
         try:
             try:
-                parsed_resp = await _parse_notify_response(resp.content)
+                parsed_resp = await _parse_receive_response(resp.content)
             finally:
                 await raw_resp.__aexit__(None, None, None)
         except Exception:
@@ -559,7 +559,7 @@ class _ReceiveResponseConfirmReceive:
     num_subscribers: int
 
 
-async def _parse_notify_response(
+async def _parse_receive_response(
     rdr: AsyncReadableBytesIO,
 ) -> Union[_ReceiveResponseUnsubscribeImmediate, _ReceiveResponseConfirmReceive]:
     """Raises ValueError typically if the body is bad"""

src/lonelypss/ws/handlers/open/check_read_task.py

@@ -40,35 +40,6 @@ async def check_read_task(state: StateOpen) -> CheckResult:
     message = S2B_AnyMessageParser.parse(prefix.flags, prefix.type, payload_reader)
     state.read_task = make_websocket_read_task(state.websocket)
 
-    # fast track acks as they are common and can be handled synchronously
-    if message.type == SubscriberToBroadcasterStatefulMessageType.CONFIRM_RECEIVE:
-        expected_ack = state.expecting_acks.get_nowait()
-        if expected_ack.type != message.type:
-            raise Exception(
-                f"unexpected confirm receive (expecting a {expected_ack.type})"
-            )
-        if expected_ack.identifier != message.identifier:
-            raise Exception(
-                f"unexpected confirm receive (expecting identifier {expected_ack.identifier!r}, got {message.identifier!r})"
-            )
-        return CheckResult.RESTART
-
-    if message.type == SubscriberToBroadcasterStatefulMessageType.CONTINUE_RECEIVE:
-        expected_ack = state.expecting_acks.get_nowait()
-        if expected_ack.type != message.type:
-            raise Exception(
-                f"unexpected continue receive (expecting a {expected_ack.type})"
-            )
-        if expected_ack.identifier != message.identifier:
-            raise Exception(
-                f"unexpected continue receive (expecting identifier {expected_ack.identifier!r}, got {message.identifier!r})"
-            )
-        if expected_ack.part_id != message.part_id:
-            raise Exception(
-                f"unexpected continue receive (expecting part_id {expected_ack.part_id}, got {message.part_id})"
-            )
-        return CheckResult.RESTART
-
     if state.process_task is None:
         state.process_task = asyncio.create_task(process_any(state, message))
     else:

src/lonelypss/ws/handlers/open/processors/process_confirm_receive.py

@@ -0,0 +1,39 @@
+import time
+from typing import TYPE_CHECKING
+
+from lonelypsp.auth.config import AuthResult
+from lonelypsp.stateful.messages.confirm_receive import S2B_ConfirmReceive
+
+from lonelypss.ws.handlers.open.processors.protocol import S2B_MessageProcessor
+from lonelypss.ws.handlers.open.websocket_url import (
+    make_for_receive_websocket_url_and_change_counter,
+)
+from lonelypss.ws.state import StateOpen
+
+
+async def process_confirm_receive(
+    state: StateOpen, message: S2B_ConfirmReceive
+) -> None:
+    expected_ack = state.expecting_acks.get_nowait()
+    if expected_ack.type != message.type:
+        raise Exception(f"unexpected confirm receive (expecting a {expected_ack.type})")
+    if expected_ack.identifier != message.identifier:
+        raise Exception(
+            f"unexpected confirm receive (expecting identifier {expected_ack.identifier!r}, got {message.identifier!r})"
+        )
+
+    receive_url = make_for_receive_websocket_url_and_change_counter(state)
+    auth_result = await state.broadcaster_config.is_confirm_receive_allowed(
+        tracing=message.tracing,
+        identifier=message.identifier,
+        num_subscribers=message.num_subscribers,
+        url=receive_url,
+        now=time.time(),
+        authorization=message.authorization,
+    )
+    if auth_result != AuthResult.OK:
+        raise Exception(f"confirm receive auth: {auth_result}")
+
+
+if TYPE_CHECKING:
+    _: S2B_MessageProcessor[S2B_ConfirmReceive] = process_confirm_receive

src/lonelypss/ws/handlers/open/processors/process_continue_receive.py

@@ -0,0 +1,42 @@
+import time
+from typing import TYPE_CHECKING
+
+from lonelypsp.auth.config import AuthResult
+from lonelypsp.stateful.messages.continue_receive import S2B_ContinueReceive
+
+from lonelypss.ws.handlers.open.processors.protocol import S2B_MessageProcessor
+from lonelypss.ws.handlers.open.websocket_url import (
+    make_for_receive_websocket_url_and_change_counter,
+)
+from lonelypss.ws.state import StateOpen
+
+
+async def process_continue_receive(
+    state: StateOpen, message: S2B_ContinueReceive
+) -> None:
+    expected_ack = state.expecting_acks.get_nowait()
+    if expected_ack.type != message.type:
+        raise Exception(
+            f"unexpected continue receive (expecting a {expected_ack.type})"
+        )
+    if expected_ack.identifier != message.identifier:
+        raise Exception(
+            f"unexpected continue receive (expecting identifier {expected_ack.identifier!r}, got {message.identifier!r})"
+        )
+    if expected_ack.part_id != message.part_id:
+        raise Exception(
+            f"unexpected continue receive (expecting part_id {expected_ack.part_id}, got {message.part_id})"
+        )
+
+    receive_url = make_for_receive_websocket_url_and_change_counter(state)
+    auth_result = await state.broadcaster_config.is_stateful_continue_receive_allowed(
+        url=receive_url,
+        message=message,
+        now=time.time(),
+    )
+    if auth_result != AuthResult.OK:
+        raise Exception(f"continue receive auth: {auth_result}")
+
+
+if TYPE_CHECKING:
+    _: S2B_MessageProcessor[S2B_ContinueReceive] = process_continue_receive

src/lonelypss/ws/handlers/open/processors/processor.py

@@ -3,6 +3,12 @@
 from lonelypsp.stateful.constants import SubscriberToBroadcasterStatefulMessageType
 from lonelypsp.stateful.message import S2B_Message
 
+from lonelypss.ws.handlers.open.processors.process_confirm_receive import (
+    process_confirm_receive,
+)
+from lonelypss.ws.handlers.open.processors.process_continue_receive import (
+    process_continue_receive,
+)
 from lonelypss.ws.handlers.open.processors.process_notify import process_notify
 from lonelypss.ws.handlers.open.processors.process_notify_stream import (
     process_notify_stream,
@@ -29,6 +35,8 @@
     SubscriberToBroadcasterStatefulMessageType.SUBSCRIBE_GLOB: process_subscribe_glob,
     SubscriberToBroadcasterStatefulMessageType.UNSUBSCRIBE_EXACT: process_unsubscribe_exact,
     SubscriberToBroadcasterStatefulMessageType.UNSUBSCRIBE_GLOB: process_unsubscribe_glob,
+    SubscriberToBroadcasterStatefulMessageType.CONFIRM_RECEIVE: process_confirm_receive,
+    SubscriberToBroadcasterStatefulMessageType.CONTINUE_RECEIVE: process_continue_receive,
 }
 
 

src/lonelypss/ws/handlers/open/send_receive_stream.py

@@ -330,6 +330,7 @@ async def send_receive_stream_given_first_headers(
                     identifier=identifier,
                     authorization=None,
                     tracing=b"",
+                    num_subscribers=1,
                 )
                 if is_done
                 else S2B_ContinueReceive(