~12 min · protocol, pydantic, validation
들어오는 모든 WebSocket 메시지는 검증하기 전까지 적대적인 입력이야. 봉투 구조, type, 모든 필드와 길이를 검사해. Pydantic 을 쓰면 스키마로 구조화된 검증을 하고 오류는 {type: 'error', code: 'validation_error', message: ...} 형태로 돌려줄 수 있어.
메시지 type 마다 Pydantic 모델을 정의하고 Field(discriminator='type') 로 나눠. 프레임워크가 type 필드에 따라 알맞은 모델을 고르고 나머지를 검증한 뒤 자료형이 정해진 객체를 돌려줘. 잘못된 메시지는 처리기가 실행되기 전에 거절돼.
from pydantic import BaseModel, Field, ValidationError, field_validator
from typing import Literal, Annotated, Union, Optional
class ChatPayload(BaseModel):
room: str
text: str
reply_to: Optional[str] = None
@field_validator('text')
@classmethod
def text_ok(cls, v: str):
v = v.strip()
if not v:
raise ValueError('empty')
if len(v) > 5_000:
raise ValueError('too long')
return v
class ChatMessage(BaseModel):
type: Literal['chat.message']
data: ChatPayload
class JoinPayload(BaseModel):
room: str
class RoomJoin(BaseModel):
type: Literal['room.join']
data: JoinPayload
InboundMessage = Annotated[
Union[ChatMessage, RoomJoin],
Field(discriminator='type'),
]
# In your handler
async def handle(ws, raw: dict):
try:
msg = pydantic.TypeAdapter(InboundMessage).validate_python(raw)
except ValidationError as e:
await ws.send_json({
'type': 'error',
'code': 'validation_error',
'message': e.errors()[0]['msg'],
})
return
if isinstance(msg, ChatMessage):
await on_chat(ws, msg.data)
elif isinstance(msg, RoomJoin):
await on_join(ws, msg.data)chat.message, 잘못된 type 인 'chat.unknown', 필드가 빠진 메시지를 각각 보내. 클라이언트가 멈추지 않고 처리할 수 있는 구조화된 오류 응답이 만들어지는지 확인해.아직 댓글이 없어요. 첫 댓글을 남겨보세요.