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