C.W.K.
Stream
Lesson 04 of 07 · published

SSL/TLS & Security

~13 min · production, security, wss

Level 0Poller
0 XP0/60 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

wss:// 는 무조건

Production WebSocket 이 TLS 위 돌아. 많은 corporate firewall + proxy 가 ws:// 그냥 막음. Cleartext 가 path 위 모든 사람한테 auth token 노출. setup 이 HTTPS 와 동일 — TLS 가 proxy 또는 application 에서 terminate, WebSocket 이 같은 TLS stream 위 돌아. Let's Encrypt 로 무료 certificate; 자동 renewal.

보안 체크리스트

Handshake 를 auth 경계로 다뤄. Origin validate (브라우저가 WebSocket 한테 CORS enforce 안 함). User 인증 (JWT, cookie, signed token). IP 별 + user 별 rate limit. 들어오는 모든 message validate (Pydantic). Message size cap (Starlette default 16MB; 낮춰). User 별 connection 수 cap (multi-device ok, multi-thousand 의심).

Code

Defense-in-depth WebSocket endpoint·python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import time

ALLOWED_ORIGINS = {'https://app.example.com', 'https://admin.example.com'}
MAX_MSG_PER_SEC = 20

@app.websocket('/ws')
async def secure_ws(websocket: WebSocket):
    # 1. Origin validation (CORS for WebSocket is your job)
    origin = websocket.headers.get('origin', '')
    if origin not in ALLOWED_ORIGINS:
        await websocket.close(code=4003, reason='forbidden origin')
        return

    # 2. Auth — before accept()
    token = websocket.query_params.get('token')
    user = decode_jwt(token) if token else None
    if not user:
        await websocket.close(code=4001, reason='unauthorized')
        return

    await websocket.accept()

    # 3. Per-connection state for rate limiting
    msg_window = []  # timestamps in the last second

    try:
        async for msg in websocket.iter_json():
            now = time.time()
            msg_window = [t for t in msg_window if now - t < 1.0]
            if len(msg_window) >= MAX_MSG_PER_SEC:
                await websocket.send_json({
                    'type': 'error', 'code': 'rate_limited',
                })
                continue
            msg_window.append(now)

            # 4. Validate every message
            if not isinstance(msg, dict) or 'type' not in msg:
                await websocket.send_json({
                    'type': 'error', 'code': 'invalid_message',
                })
                continue

            await dispatch(websocket, user, msg)
    except WebSocketDisconnect:
        pass

External links

Exercise

기존 WebSocket endpoint 를 체크리스트 (origin, auth, rate limit, message validation, message size, connection cap) 에 대비 audit. 각각 0/1/2 (0 missing, 1 partial, 2 done). 2 미만 다 fix. Caddy 또는 Let's Encrypt 로 wss:// deploy + ws:// reject 확인.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.