Secure FastAPI WebSocket: Membetulkan Ralat Suntikan Ketergantungan

Hei!

Jadi, anda cuba melindungi WebSocket anda, dan ralat suntikan pergantungan ini muncul. Menjengkelkan, bukan? Jangan risau — saya ada penyelesaian yang cepat dan mudah yang akan menyelesaikan masalah anda.

Masalahnya: Ralat Suntikan Ketergantungan

Anda semua teruja untuk mendapatkan WebSocket anda, tetapi boom! Ralat suntikan ketergantungan muncul.

Tetapi inilah penyelesaian yang mudah.

Penyelesaian: JWT dalam Pengepala Permintaan

Inilah helahnya: gunakan Token Web JSON (JWT). Letakkan token itu ke dalam pengepala permintaan, dan anda sudah berjaya. Ia membolehkan anda melakukan beberapa perkara yang menarik — seperti memikirkan siapa pengguna semasa berada di laluan WebSocket anda. Mudah dan berkesan.

Tidak perlu jargon mewah. Lihat coretan kod pantas ini:

`@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")
cuba:

    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"Tidak dapat mengesahkan kelayakan", headers={"WWW-Authenticate": "Pembawa"})

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`

`# pangkalan data.py
daripada import sqlalchemy create_engine
daripada sqlalchemy.ext.declarative import declarative_base
daripada sqlalchemy.orm import sessionmaker

URL_DATABASE_SQLALCHEMY = 'postgresql psycopg://:@/'

enjin = create_enjin(SQLALCHEMY_DATABASE_URL)

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

Asas = declarative_base()

def get_db():
db = SessionLocal()
cuba:
hasil db
akhirnya:
db.close()`

Ia bukan sains roket; ia hanyalah penyelesaian pantas.

Hanya untuk membuktikan ia berfungsi, kami mendapat tangkapan skrin daripada Posmen.

Selam Lebih Dalam: Projek Bilik Sembang

Jika anda mahukan maklumat penuh, pergi ke projek "chatRoom" saya di Github. Anda akan temui segala-galanya di sana — tiada rahsia, hanya panduan ringkas dan keseluruhannya.

Terima Kasih

Terima kasih kerana bertahan di sana! Masa anda penting, dan kami menghargai anda membacanya. Pastikan ia mudah, pastikan ia selamat.

Sola,
Aditya Keshari

Atas ialah kandungan terperinci Secure FastAPI WebSocket: Membetulkan Ralat Suntikan Ketergantungan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!