برای بهبود مدیریت ارتباط و اجتناب از پیام‌های طولانی، می‌توان از کلمات کلیدی برای هماهنگی Access Token و Refresh Token در احراز هویت In-Band استفاده کرد. به عنوان مثال، وقتی Access Token منقضی شود، سرور یک پیام با کلمه کلیدی مشخص ارسال می‌کند و کلاینت می‌داند که باید Refresh Token را ارسال کند.

نمونه سرور FastAPI با کلمات کلیدی

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from datetime import datetime, timedelta
import jwt

SECRET_KEY = "mysecretkey"
REFRESH_SECRET_KEY = "myrefreshsecret"

app = FastAPI()

def create_access_token(username: str):
    expire = datetime.utcnow() + timedelta(seconds=30)
    payload = {"sub": username, "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def create_refresh_token(username: str):
    expire = datetime.utcnow() + timedelta(minutes=5)
    payload = {"sub": username, "exp": expire}
    return jwt.encode(payload, REFRESH_SECRET_KEY, algorithm="HS256")

def verify_token(token: str, secret: str):
    try:
        payload = jwt.decode(token, secret, algorithms=["HS256"])
        return payload["sub"]
    except jwt.ExpiredSignatureError:
        return None
    except jwt.PyJWTError:
        return None

ACCESS_EXPIRED_KEYWORD = "ACCESS_EXPIRED"
REFRESH_INVALID_KEYWORD = "REFRESH_INVALID"

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        token_message = await websocket.receive_text()
        username = verify_token(token_message, SECRET_KEY)
        if not username:
            # ارسال کلمه کلیدی برای درخواست Refresh Token
            await websocket.send_text(ACCESS_EXPIRED_KEYWORD)
            token_message = await websocket.receive_text()
            username = verify_token(token_message, REFRESH_SECRET_KEY)
            if not username:
                await websocket.send_text(REFRESH_INVALID_KEYWORD)
                await websocket.close(code=1008)
                return
            new_access_token = create_access_token(username)
            await websocket.send_text(new_access_token)
        
        await websocket.send_text(f"Welcome {username}!")
        
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"{username} sent: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

نمونه کلاینت پایتون با کلمات کلیدی

import asyncio
import websockets

ACCESS_TOKEN = "YOUR_ACCESS_TOKEN"
REFRESH_TOKEN = "YOUR_REFRESH_TOKEN"
uri = "ws://127.0.0.1:8000/ws"

ACCESS_EXPIRED_KEYWORD = "ACCESS_EXPIRED"
REFRESH_INVALID_KEYWORD = "REFRESH_INVALID"

async def websocket_client():
    async with websockets.connect(uri) as websocket:
        await websocket.send(ACCESS_TOKEN)
        response = await websocket.recv()
        print(f"Server: {response}")
        
        if response == ACCESS_EXPIRED_KEYWORD:
            await websocket.send(REFRESH_TOKEN)
            new_token_response = await websocket.recv()
            if new_token_response == REFRESH_INVALID_KEYWORD:
                print("Refresh token invalid. Connection closed.")
                return
            print(f"New Access Token received: {new_token_response}")
        
        while True:
            message = input("Enter your message: ")
            await websocket.send(message)
            server_response = await websocket.recv()
            print(f"Server response: {server_response}")

asyncio.run(websocket_client())

توضیح:

• کلمات کلیدی به جای پیام‌های طولانی استفاده می‌شوند تا هماهنگی بین سرور و کلاینت ساده شود.
• سرور هنگام انقضای Access Token، کلمه کلیدی ACCESS_EXPIRED را ارسال می‌کند.
• کلاینت پس از دریافت این کلمه کلیدی، Refresh Token را ارسال می‌کند و سرور Access Token جدید صادر می‌کند.
• در صورت نامعتبر بودن Refresh Token، سرور کلمه کلیدی REFRESH_INVALID را ارسال و اتصال را می‌بندد.