~12 min · library, middleware, rate-limit
connect 처리기는 인증, 버전 검사, 지역 차단, 속도 제한 등록을 하는 미들웨어 자리야. 유용한 페이로드를 담아 ConnectionRefusedError(code, message) 를 일으키면 Socket.IO 가 코드와 문장을 모두 클라이언트의 connect_error 이벤트에 드러내.
속도 제한 데코레이터를 만들어 채팅 전송이나 파일 업로드처럼 필요한 이벤트에만 적용하고 읽기 전용 이벤트는 건너뛰어. sid 를 키로 삼아 연결별 최근 시각을 작은 deque 에 저장하면 충분해.
@sio.event
async def connect(sid, environ, auth):
token = (auth or {}).get('token')
user = decode_jwt(token) if token else None
if not user:
raise socketio.exceptions.ConnectionRefusedError(
'unauthorized',
'token missing or invalid',
)
if user.get('disabled'):
raise socketio.exceptions.ConnectionRefusedError(
'forbidden',
'account disabled',
)
await sio.save_session(sid, {'user': user})from collections import defaultdict, deque
import time, functools
windows = defaultdict(deque) # sid -> deque of timestamps
def rate_limit(per_second=20):
def deco(fn):
@functools.wraps(fn)
async def wrapper(sid, *args, **kwargs):
now = time.time()
w = windows[sid]
while w and now - w[0] > 1.0:
w.popleft()
if len(w) >= per_second:
await sio.emit('error',
{'code': 'rate_limited', 'message': 'slow down'},
to=sid)
return
w.append(now)
return await fn(sid, *args, **kwargs)
return wrapper
return deco
@sio.on('chat:message')
@rate_limit(per_second=20)
async def chat(sid, data):
# ...
pass아직 댓글이 없어요. 첫 댓글을 남겨보세요.