websockets: only allow connections to identify once

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2022-04-08 20:22:23 -04:00
parent cbc02d625c
commit da864ab391
No known key found for this signature in database
GPG Key ID: 7027245FBBDDF59A
1 changed files with 10 additions and 0 deletions

View File

@ -380,6 +380,10 @@ class WebsocketManager(APITransport):
ws: WebSocket,
**kwargs
) -> Dict[str, int]:
if ws.identified:
raise self.server.error(
f"Connection already identified: {ws.client_data}"
)
try:
name = str(kwargs["client_name"])
version = str(kwargs["version"])
@ -500,6 +504,7 @@ class WebSocket(WebSocketHandler, Subscribable):
self.message_buf: List[Union[str, Dict[str, Any]]] = []
self.last_pong_time: float = self.event_loop.get_loop_time()
self._connected_time: float = 0.
self._identified: bool = False
self._client_data: Dict[str, str] = {
"name": "unknown",
"version": "",
@ -519,6 +524,10 @@ class WebSocket(WebSocketHandler, Subscribable):
def start_time(self) -> float:
return self._connected_time
@property
def identified(self) -> bool:
return self._identified
@property
def client_data(self) -> Dict[str, str]:
return self._client_data
@ -526,6 +535,7 @@ class WebSocket(WebSocketHandler, Subscribable):
@client_data.setter
def client_data(self, data: Dict[str, str]) -> None:
self._client_data = data
self._identified = True
def open(self, *args, **kwargs) -> None:
self.set_nodelay(True)