Secure FastAPI WebSocket : correction des erreurs d'injection de dépendances

Salut !

Donc, vous essayez de sécuriser votre WebSocket, et ces erreurs d’injection de dépendance apparaissent. Ennuyeux, non ? Ne vous inquiétez pas, j'ai une solution simple et rapide qui vous aidera.

Le problème : les erreurs d'injection de dépendances

Vous êtes tous impatients de sécuriser votre WebSocket, mais boum ! Des erreurs d’injection de dépendances apparaissent.

Mais voici une solution simple.

La solution : JWT dans l'en-tête de la requête

Voici l'astuce : utilisez un jeton Web JSON (JWT). Insérez ce jeton dans l’en-tête de la demande et vous êtes en or. Il vous permet de faire des choses intéressantes, comme déterminer qui est l'utilisateur actuel directement dans votre itinéraire WebSocket. Simple et efficace.

Pas besoin de jargon sophistiqué. Découvrez cet extrait de code rapide :

`@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, qualifications_exception) :
ic("verify_access_token")
essayez :

    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):
qualifications_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Impossible de valider les informations d'identification", 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`

`#base de données.py
à partir de sqlalchemy import create_engine
à partir de sqlalchemy.ext.declarative import declarative_base
depuis sqlalchemy.orm importer sessionmaker

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

moteur = create_engine(SQLALCHEMY_DATABASE_URL)

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

Base = déclarative_base()

def get_db() :
db = SessionLocal()
essayez :
rendement db
enfin :
db.close()`

Ce n’est pas sorcier ; c'est juste une solution rapide.

Juste pour prouver que cela fonctionne, nous avons des captures d'écran de Postman.

Plongez plus profondément : projet ChatRoom

Si vous voulez un aperçu complet, rendez-vous sur mon projet « chatRoom » sur Github. Vous y trouverez tout : pas de secrets, juste un guide simple et toute l'affaire.

Un grand merci

Merci d'être resté là ! Votre temps compte et nous vous remercions de le lire. Restez simple, gardez-le en sécurité.

Bravo,
Aditya Keshari

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!