1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
from http.client import HTTPConnection
def websocket_request(
remote_agent_host, remote_agent_port, host=None, origin=None, path="/session"
):
real_host = f"{remote_agent_host}:{remote_agent_port}"
url = f"http://{real_host}{path}"
conn = HTTPConnection(real_host)
skip_host = host is not None
conn.putrequest("GET", url, skip_host)
if host is not None:
conn.putheader("Host", host)
if origin is not None:
conn.putheader("Origin", origin)
conn.putheader("Connection", "upgrade")
conn.putheader("Upgrade", "websocket")
conn.putheader("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
conn.putheader("Sec-WebSocket-Version", "13")
conn.endheaders()
return conn.getresponse()
def http_request(server_host, server_port, path="/status", host=None, origin=None):
url = f"http://{server_host}:{server_port}{path}"
conn = HTTPConnection(server_host, server_port)
custom_host = host is not None
conn.putrequest("GET", url, skip_host=custom_host)
if custom_host:
conn.putheader("Host", host)
if origin is not None:
conn.putheader("Origin", origin)
conn.endheaders()
return conn.getresponse()
def get_host(port_type, hostname, server_port):
if port_type == "default_port":
return hostname
if port_type == "server_port":
return f"{hostname}:{server_port}"
if port_type == "wrong_port":
wrong_port = int(server_port) + 1
return f"{hostname}:{wrong_port}"
raise Exception(f"Unrecognised port_type {port_type}")
|
