pythonnative/tests/test_devserver_ws.py at main · pythonnative/pythonnative · GitHub

Repository navigation

History

149 lines (110 loc) · 5.92 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

"""RFC 6455 framing and handshake helpers in ``pythonnative.devserver.ws``."""

from __future__ import annotations

from typing import Dict, List, Tuple

import pytest

from pythonnative.devserver import auth, ws

def _decode_all(data: bytes) -> list:

decoder = ws.FrameDecoder()

return list(decoder.feed(data))

def test_accept_key_matches_rfc_example() -> None:

# The worked example from RFC 6455 section 1.3.

assert ws.accept_key("dGhlIHNhbXBsZSBub25jZQ==") == "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="

def test_server_handshake_echoes_accept_header() -> None:

response = ws.server_handshake(

{"upgrade": "websocket", "connection": "Upgrade", "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ=="}

)

text = response.decode("ascii")

assert text.startswith("HTTP/1.1 101")

assert "Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" in text

assert text.endswith("\r\n\r\n")

def test_server_handshake_rejects_missing_key() -> None:

with pytest.raises(ws.HandshakeError):

ws.server_handshake({"upgrade": "websocket"})

def test_parse_http_headers_lowercases_names() -> None:

raw = b"GET /ws?role=client HTTP/1.1\r\nHost: x\r\nSec-WebSocket-Key: abc\r\n\r\n"

request_line, headers = ws.parse_http_headers(raw)

assert request_line.startswith("GET /ws?role=client")

assert headers["sec-websocket-key"] == "abc"

assert headers["host"] == "x"

def test_client_handshake_request_carries_key_and_path() -> None:

request, key = ws.client_handshake_request("localhost:8765", "/ws?role=client", key="dGhlIHNhbXBsZSBub25jZQ==")

text = request.decode("ascii")

assert key == "dGhlIHNhbXBsZSBub25jZQ=="

assert text.startswith("GET /ws?role=client HTTP/1.1\r\n")

assert "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" in text

assert "Sec-WebSocket-Version: 13" in text

@pytest.mark.parametrize("size", [0, 1, 125, 126, 127, 65535, 65536, 70000])

def test_frame_round_trip_across_length_encodings(size: int) -> None:

payload = bytes(i % 251 for i in range(size))

for mask in (False, True):

frame = ws.encode_frame(ws.BINARY, payload, mask=mask)

messages = _decode_all(frame)

assert messages == [(ws.BINARY, payload)]

def test_masked_frames_are_unreadable_on_the_wire() -> None:

payload = b"hello hello hello"

frame = ws.encode_frame(ws.TEXT, payload, mask=True)

assert payload not in frame

assert frame[1] & 0x80 # mask bit set

assert _decode_all(frame) == [(ws.TEXT, payload)]

def test_decoder_reassembles_fragmented_messages_and_interleaved_control_frames() -> None:

first = ws.encode_frame(ws.TEXT, b"hel", fin=False)

ping = ws.encode_frame(ws.PING, b"p")

middle = ws.encode_frame(ws.CONTINUATION, b"lo ", fin=False)

last = ws.encode_frame(ws.CONTINUATION, b"world", fin=True)

decoder = ws.FrameDecoder()

out: List[Tuple[int, bytes]] = []

# Deliver byte by byte to exercise partial buffering.

for byte in first + ping + middle + last:

out.extend(decoder.feed(bytes([byte])))

assert out == [(ws.PING, b"p"), (ws.TEXT, b"hello world")]

def test_decoder_rejects_fragmented_control_frame() -> None:

frame = ws.encode_frame(ws.PING, b"x", fin=False)

with pytest.raises(ws.WebSocketError):

_decode_all(frame)

def test_encode_close_carries_status_code_and_reason() -> None:

frame = ws.encode_close(1001, "going away")

[(opcode, payload)] = _decode_all(frame)

assert opcode == ws.CLOSE

assert payload[:2] == (1001).to_bytes(2, "big")

assert payload[2:] == b"going away"

def test_websocket_client_rejects_non_ws_schemes() -> None:

with pytest.raises(ValueError):

ws.WebSocketClient("wss://example.com/ws")

with pytest.raises(ValueError):

ws.WebSocketClient("ws:///nohost")

client = ws.WebSocketClient("ws://localhost:1234/ws?role=client")

assert (client.host, client.port, client.path) == ("localhost", 1234, "/ws?role=client")

def test_websocket_client_keeps_the_token_in_the_request_path() -> None:

client = ws.WebSocketClient("ws://localhost:1234/ws?role=client&token=abc")

assert client.path == "/ws?role=client&token=abc"

# ----------------------------------------------------------------------

# Upgrade authorization helpers

# ----------------------------------------------------------------------

@pytest.mark.parametrize(

"headers, allowed",

[

({"host": "localhost:8765"}, True), # native dev clients send no Origin

({"host": "localhost:8765", "origin": "http://localhost:8765"}, True),

({"host": "LOCALHOST:8765", "origin": "http://localhost:8765"}, True),

({"host": "192.168.1.5:8765", "origin": "http://192.168.1.5:8765"}, True),

({"host": "localhost:8765", "origin": "http://localhost:3000"}, False),

({"host": "localhost:8765", "origin": "https://evil.example"}, False),

({"host": "localhost:8765", "origin": "null"}, False),

({"origin": "http://localhost:8765"}, False), # no Host to compare against

)

def test_websocket_origin_must_match_host(headers: Dict[str, str], allowed: bool) -> None:

from pythonnative.devserver.server import _same_origin

assert _same_origin(headers) is allowed

def test_request_token_reads_query_header_then_cookie() -> None:

assert auth.request_token({}, {"token": "q"}) == "q"

assert auth.request_token({"x-pn-token": "h"}, {}) == "h"

assert auth.request_token({"cookie": 'a=1; pn_token="c"; b=2'}, {}) == "c"

assert auth.request_token({"cookie": "pn_token_other=x"}, {}) is None

assert auth.request_token({}, {}) is None

assert auth.token_matches("secret", "secret")

assert not auth.token_matches("secreT", "secret")

assert not auth.token_matches(None, "secret")

def test_with_token_replaces_an_existing_token() -> None:

assert auth.with_token("ws://h:1/ws?role=client", "t") == "ws://h:1/ws?role=client&token=t"

assert auth.with_token("ws://h:1/ws?token=old&role=client", "new") == "ws://h:1/ws?role=client&token=new"

pythonnative/tests/test_devserver_ws.py at main · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_devserver_ws.py

Latest commit

History

Breadcrumbs

test_devserver_ws.py

File metadata and controls