집 > 기사 > 백엔드 개발 > 안전한 FastAPI WebSocket: 종속성 주입 오류 수정

안전한 FastAPI WebSocket: 종속성 주입 오류 수정

DDD원래의: 2024-09-13 16:15:50572검색

안녕하세요!

WebSocket을 보호하려고 하면 이러한 종속성 주입 오류가 나타납니다. 짜증나죠? 너무 걱정하지 마세요. 여러분의 문제를 해결할 빠르고 쉬운 해결책이 있습니다.

문제: 종속성 주입 오류

WebSocket 보안에 대해 다들 기대하고 계시지만 굉장합니다! 종속성 주입 오류가 나타납니다.

Secure FastAPI WebSocket: Fixing Dependency Injection Errors

하지만 간단한 해결 방법이 있습니다.

해결책: 요청 헤더의 JWT

비결은 다음과 같습니다. JWT(JSON 웹 토큰)를 사용하세요. 해당 토큰을 요청 헤더에 넣으면 황금입니다. 이를 통해 WebSocket 경로에서 현재 사용자가 누구인지 알아내는 등 몇 가지 멋진 작업을 수행할 수 있습니다. 간단하고 효과적입니다.

Secure FastAPI WebSocket: Fixing Dependency Injection Errors

고급 전문 용어는 필요하지 않습니다. 이 빠른 코드 스니펫을 확인하세요.

`@router.websocket("/create")
async def create_room(websocket: WebSocket, db: 세션 = 종속(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(토큰: str, 자격 증명_예외):
ic("verify_access_token")
시도해 보세요:

    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(토큰: str):
자격 증명_예외 = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
Detail=f"자격 증명을 확인할 수 없습니다.", 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
sqlalchemy import create_engine
에서 sqlalchemy.ext.declarative import declarative_base
에서 sqlalchemy.orm에서 세션메이커 가져오기

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

엔진 = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, 바인딩=엔진)

기본 = declarative_base()

def get_db():
db = SessionLocal()
시도해 보세요:
DB 수율
드디어:
db.close()`

로켓 과학이 아닙니다. 그것은 단지 빠른 해결책일 뿐입니다.

Secure FastAPI WebSocket: Fixing Dependency Injection Errors