APIKeyHeader for Websockets

[Catastropha]

• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.

Is your feature request related to a problem

Yes. It would be nice to use Security with APIKeyHeader (maybe APIKeyCookie, APIKeyQuery as well) for websocket connections.

The solution you would like

Consistent with the current http way of doing it

api_key_header = APIKeyHeader(name=API_KEY_NAME)

async def get_origin(
        api_key: str = Security(api_key_header),
        db: AsyncIOMotorClient = Depends(get_database), ):

Describe alternatives you've considered

At the moment we plan to protect websocket endpoints with a simple api_key: str = Header(None),
and then we do

if not api_key:
        await websocket.close(code=WS_1008_POLICY_VIOLATION)
        return None

For this I guess we might need the WebSocketException feature in Starlette

If someone can give me some guiding on how to structure this feature I might give it a try and do a PR

[tiangolo]

Yes, you can, it's documented here: https://fastapi.tiangolo.com/advanced/websockets/#using-depends-and-others

Also, have in mind that browsers don't have a way to set headers when creating WebSocket connections: https://stackoverflow.com/a/4361358/219530

[Catastropha]

You are right!

Out of curiosity, (the project is for mobile apps only) I followed the docs and here is what I have:

from fastapi import FastAPI, APIRouter, Security, Depends
from starlette.websockets import WebSocket
from fastapi.security.api_key import APIKeyHeader


router = APIRouter()


async def get_ws_origin(
        websocket: WebSocket,
        api_key: str = Security(APIKeyHeader(name='token')),
):
    print(api_key)
    return {'success': True}


@router.websocket("/ws/")
async def ws_endpoint(
        websocket: WebSocket,
        origin: dict = Depends(get_ws_origin),
):
    await websocket.accept()
    print(origin)

app = FastAPI()
app.include_router(router)

Is this the correct way to do websockets with FastAPI?

I get this error (same error with APIKeyQuery):

Traceback (most recent call last):
  File "/env/lib/python3.6/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 153, in run_asgi
    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
  File "/env/lib/python3.6/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
    return await self.app(scope, receive, send)
  File "/env/lib/python3.6/site-packages/fastapi/applications.py", line 142, in __call__
    await super().__call__(scope, receive, send)  # pragma: no cover
  File "/env/lib/python3.6/site-packages/starlette/applications.py", line 134, in __call__
    await self.error_middleware(scope, receive, send)
  File "/env/lib/python3.6/site-packages/starlette/middleware/errors.py", line 143, in __call__
    await self.app(scope, receive, send)
  File "/env/lib/python3.6/site-packages/sentry_sdk/integrations/asgi.py", line 81, in _run_asgi3
    return await self._run_app(scope, lambda: self.app(scope, receive, send))
  File "/env/lib/python3.6/site-packages/sentry_sdk/integrations/asgi.py", line 86, in _run_app
    return await callback()
  File "/env/lib/python3.6/site-packages/starlette/middleware/cors.py", line 68, in __call__
    await self.app(scope, receive, send)
  File "/env/lib/python3.6/site-packages/starlette/exceptions.py", line 49, in __call__
    await self.app(scope, receive, send)
  File "/env/lib/python3.6/site-packages/starlette/routing.py", line 590, in __call__
    await route(scope, receive, send)
  File "/env/lib/python3.6/site-packages/starlette/routing.py", line 264, in __call__
    await self.app(scope, receive, send)
  File "/env/lib/python3.6/site-packages/starlette/routing.py", line 57, in app
    await func(session)
  File "/env/lib/python3.6/site-packages/fastapi/routing.py", line 169, in app
    dependency_overrides_provider=dependency_overrides_provider,
  File "/env/lib/python3.6/site-packages/fastapi/dependencies/utils.py", line 490, in solve_dependencies
    dependency_cache=dependency_cache,
  File "/env/lib/python3.6/site-packages/fastapi/dependencies/utils.py", line 520, in solve_dependencies
    solved = await call(**sub_values)
TypeError: __call__() missing 1 required positional argument: 'request'

[Catastropha]

@tiangolo can you please help me? I'll close this question and I'm not going to bug you anymore :)

[phy25]

@Catastropha APIKeyHeader requires Request which is not available in WebSocket. I guess you can just implement your own Security object for your use case.

fastapi/fastapi/security/api_key.py

Line 38 in 9c3c9b6

async def __call__(self, request: Request) -> Optional[str]:

[GitToby]

same issue here, but solved with a pretty hacky adapter that pulls the header as a header and then passes to the APIs security:

async def validate_ws_static_token(
    token: str = Header(...),
    db=Depends(database_session),
):
    return await _user_from_token(
        token,
        db
    )

async def _user_from_token(
    token: str = Depends(APIKeyHeader(...)),
    db=Depends(database_session)
):
    ...