در احراز هویت مبتنی بر توکن (Token-Based)، کلاینت ابتدا یک توکن معتبر مانند JWT از سرور دریافت می‌کند و هنگام اتصال به WebSocket آن را ارسال می‌کند. سرور توکن را بررسی کرده و در صورت معتبر بودن اتصال را می‌پذیرد.
نمونه سرور WebSocket با FastAPI و JWT
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException
from fastapi.security import HTTPBearer
import jwt
SECRET_KEY = "mysecretkey"
app = FastAPI()
def verify_token(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload
except jwt.PyJWTError:
raise HTTPException(status_code=401, detail="Invalid token")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = None):
if token is None:
await websocket.close(code=1008)
return
try:
user = verify_token(token)
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"پیام شما دریافت شد: {data}")
except WebSocketDisconnect:
print("Client disconnected")
except HTTPException:
await websocket.close(code=1008)
نمونه کلاینت پایتون با ارسال توکن
import asyncio
import websockets
TOKEN = "YOUR_JWT_TOKEN_HERE"
uri = f"ws://127.0.0.1:8000/ws?token={TOKEN}"
async def websocket_client():
async with websockets.connect(uri) as websocket:
print("Connected to server with token!")
while True:
message = input("Enter your message: ")
await websocket.send(message)
response = await websocket.recv()
print(f"Server response: {response}")
asyncio.run(websocket_client())
توضیح:
- در سرور، توکن JWT از query parameter هنگام اتصال گرفته شده و با تابع verify_token بررسی می‌شود.
- در صورت نامعتبر بودن توکن، اتصال WebSocket با کد 1008 بسته می‌شود.
- کلاینت، توکن را در URL هنگام اتصال به سرور ارسال می‌کند.
- پس از تایید توکن، کلاینت می‌تواند پیام ارسال و دریافت کند.