Websocket Redirect Middleware

[x-stride]

Example Code

import uvicorn
from fastapi import FastAPI, WebSocket
from fastapi.responses import RedirectResponse
from fastapi.datastructures import URL
from starlette.types import ASGIApp, Receive, Scope, Send


class WSRedirectMiddleware:
    def __init__(
            self,
            app: ASGIApp,
            redirect_paths: list
    ) -> None:
        self.app = app
        self.redirect_paths = redirect_paths

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        url = URL(scope=scope)
        if scope["type"] == "websocket" and url.path in self.redirect_paths:
            print(f'Got path: {url.path}')

            url = f'{scope["scheme"]}://{url.hostname}:{url.port}/ws2'
            print(f'Redirect to: {url}')

            response = RedirectResponse(url, status_code=307)
            await response(scope, receive, send)
        else:
            await self.app(scope, receive, send)


application = FastAPI()


# Redirect from /ws1 to /ws2
application.add_middleware(WSRedirectMiddleware, redirect_paths=['/ws1'])


@application.websocket("/ws1")
async def websocket(ws: WebSocket):
    await ws.accept()
    await ws.send_json({"msg": "Hello WebSocket1"})
    await ws.close()


@application.websocket("/ws2")
async def websocket(ws: WebSocket):
    await ws.accept()
    await ws.send_json({"msg": "Hello WebSocket2"})
    await ws.close()

if __name__ == '__main__':
    uvicorn.run(application, host="localhost", port=8088)

Description

Hello all,

Created an alternative of HTTPSRedirectMiddleware for redirection, but it's not working as I expected.

Error logs:

INFO:     Started server process [25196]
INFO:     Waiting for application startup.
INFO:     ASGI 'lifespan' protocol appears unsupported.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8088 (Press CTRL+C to quit)
Got path: /ws1
Redirect to: ws://localhost:8088/ws2
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\uvicorn\protocols\websockets\websockets_impl.py", line 225, in run_asgi
    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\uvicorn\middleware\proxy_headers.py", line 78, in __call__
    return await self.app(scope, receive, send)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\fastapi\applications.py", line 270, in __call__
    await super().__call__(scope, receive, send)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\starlette\applications.py", line 124, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\starlette\middleware\errors.py", line 149, in __call__
    await self.app(scope, receive, send)
  File "C:\Users\user\Documents\Signalling\main.py", line 24, in __call__
    await response(scope, receive, send)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\starlette\responses.py", line 160, in __call__
    await send(
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\uvicorn\protocols\websockets\websockets_impl.py", line 288, in asgi_send
    raise RuntimeError(msg % message_type)
RuntimeError: Expected ASGI message 'websocket.accept' or 'websocket.close', but got 'http.response.start'.
INFO:     connection open
INFO:     connection closed

Checking the FastAPI httpsredirect.py and various middleware-guides out there I thought this would work... but looking at the ASGI spec it seems impossible...

Is the HTTPSRedirectMiddleware not working for ws or can this be done?

With best regards,

/x-stride

[MRigal]

@Kludex The discussion is indeed better placed here that at tiangolo/fastapi#9609 and I would be very curious to know how to do/implement that!

Best, Matthieu

[Kludex]

• We need this Support the WebSocket Denial Response ASGI extension #2041 to get in.