Restrict methods

CHANGES.rst

@@ -4,6 +4,14 @@ CHANGES
 
 .. towncrier release notes start
 
+3.0 (2025-xx-xx)
+================
+
+- Disallow HTTP methods other than GET by default.
+  Allowing other methods can be done by passing ``allow_all_methods`` to ``EventSourceResponse``.
+  Note that SSE on browsers will only work with GET endpoints, this flag just serves as a
+  reminder.
+
 2.2.0 (2024-02-29)
 ==================
 

aiohttp_sse/__init__.py

@@ -8,10 +8,11 @@
 
 from aiohttp.abc import AbstractStreamWriter
 from aiohttp.web import BaseRequest, ContentCoding, Request, StreamResponse
+from aiohttp.web_exceptions import HTTPMethodNotAllowed
 
 from .helpers import _ContextManager
 
-__version__ = "2.2.0"
+__version__ = "3.0"
 __all__ = ["EventSourceResponse", "sse_response"]
 
 
@@ -40,6 +41,7 @@ def __init__(
         reason: Optional[str] = None,
         headers: Optional[Mapping[str, str]] = None,
         sep: Optional[str] = None,
+        allow_all_methods: bool = False,
     ):
         super().__init__(status=status, reason=reason)
 
@@ -55,6 +57,7 @@ def __init__(
         self._ping_interval: float = self.DEFAULT_PING_INTERVAL
         self._ping_task: Optional[asyncio.Task[None]] = None
         self._sep = sep if sep is not None else self.DEFAULT_SEPARATOR
+        self._allow_all_methods = allow_all_methods
 
     def is_connected(self) -> bool:
         """Check connection is prepared and ping task is not done."""
@@ -73,6 +76,9 @@ async def prepare(self, request: BaseRequest) -> Optional[AbstractStreamWriter]:
 
         :param request: regular aiohttp.web.Request.
         """
+        if not self._allow_all_methods and request.method != "GET":
+            raise HTTPMethodNotAllowed(request.method, ["GET"])
+
         if not self.prepared:
             writer = await super().prepare(request)
             self._ping_task = asyncio.create_task(self._ping())
@@ -234,6 +240,7 @@ def sse_response(
     reason: Optional[str] = None,
     headers: Optional[Mapping[str, str]] = None,
     sep: Optional[str] = None,
+    allow_all_methods: bool = False,
 ) -> _ContextManager[EventSourceResponse]: ...
 
 
@@ -246,6 +253,7 @@ def sse_response(
     headers: Optional[Mapping[str, str]] = None,
     sep: Optional[str] = None,
     response_cls: type[ESR],
+    allow_all_methods: bool = False,
 ) -> _ContextManager[ESR]: ...
 
 
@@ -257,12 +265,19 @@ def sse_response(
     headers: Optional[Mapping[str, str]] = None,
     sep: Optional[str] = None,
     response_cls: type[EventSourceResponse] = EventSourceResponse,
+    allow_all_methods: bool = False,
 ) -> Any:
     if not issubclass(response_cls, EventSourceResponse):
         raise TypeError(
             "response_cls must be subclass of "
             "aiohttp_sse.EventSourceResponse, got {}".format(response_cls)
         )
 
-    sse = response_cls(status=status, reason=reason, headers=headers, sep=sep)
+    sse = response_cls(
+        status=status,
+        reason=reason,
+        headers=headers,
+        sep=sep,
+        allow_all_methods=allow_all_methods,
+    )
     return _ContextManager(sse._prepare(request))

tests/test_sse.py

@@ -510,7 +510,7 @@ async def test_get_before_prepare(self) -> None:
 )
 async def test_http_methods(aiohttp_client: AiohttpClient, http_method: str) -> None:
     async def handler(request: web.Request) -> EventSourceResponse:
-        async with sse_response(request) as sse:
+        async with sse_response(request, allow_all_methods=True) as sse:
             await sse.send("foo")
         return sse
 
@@ -526,6 +526,29 @@ async def handler(request: web.Request) -> EventSourceResponse:
     assert streamed_data == "data: foo\r\n\r\n"
 
 
+@pytest.mark.parametrize(
+    "http_method",
+    ("POST", "PUT", "DELETE", "PATCH"),
+)
+async def test_not_allowed_methods(
+    aiohttp_client: AiohttpClient,
+    http_method: str,
+) -> None:
+    """Check that EventSourceResponse works only with GET method."""
+
+    async def handler(request: web.Request) -> EventSourceResponse:
+        async with sse_response(request) as sse:
+            ...
+        return sse  # pragma: no cover
+
+    app = web.Application()
+    app.router.add_route(http_method, "/", handler)
+
+    client = await aiohttp_client(app)
+    async with client.request(http_method, "/") as resp:
+        assert resp.status == 405
+
+
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
     reason=".cancelling() missing in older versions",