Secure FastAPI WebSocket: Fixing Dependency Injection Errors

Hey there!

So, you’re trying to secure your WebSocket, and these dependency injection errors pop up. Annoying, right? Don’t sweat it — I’ve got a quick and easy solution that’ll sort you out.

The Problem: Dependency Injection Errors

You’re all excited about securing your WebSocket, but boom! Dependency injection errors show up.

But here’s a straightforward fix.

The Solution: JWT in the Request Header

Here’s the trick: use a JSON Web Token (JWT). Pop that token into the request header, and you’re golden. It lets you do some cool stuff — like figuring out who the current user is right there in your WebSocket route. Simple and effective.

No need for fancy jargon. Check out this quick code snippet:

`@router.websocket("/create")
async def create_room(websocket: WebSocket, db: Session = Depends(get_db)):
request_header_dict = dict(websocket.headers)

# check if access_token is in the header
if('access_token' not in request_header_dict.keys()):
    ic("No access token")
    return HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

# else get access token
access_token = request_header_dict['access_token']

current_user = oauth2.get_current_user(access_token)

# websocket route logic ##

oauth2/py

def verify_access_token(token: str, credentials_exception):
ic("verify_access_token")
try:

    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    id: str = payload.get("user_id")

    if id is None:
        raise credentials_exception
    # token_data = schemas.TokenData(id=id)
except JWTError:
    ic("Error occured")
    raise credentials_exception

# return token_data
return id

def get_current_user(token: str):
credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})

db = SessionLocal()
user_id = verify_access_token(token, credentials_exception)  
user = db.query(models.User).filter(models.User.id == user_id).first()  
db.close()
return user`

`# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = 'postgresql+psycopg://:@/'

engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()`

It’s not rocket science; it’s just a quick solution.

Just to prove it works, we’ve got screenshots from Postman.

Dive Deeper: ChatRoom Project

If you want the full scoop, head over to my “chatRoom” project on Github. You’ll find everything there — no secrets, just a straightforward guide and the whole deal.

Big Thanks

Thanks for hanging in there! Your time matters, and we appreciate you giving this a read. Keep it simple, keep it secure.

Cheers,
Aditya Keshari