-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathserver.py
75 lines (57 loc) · 2.36 KB
/
server.py
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import socket
import threading
import application
class Server:
def __init__(self, app: application.Application) -> None:
self.running: bool = False
self.application: application.Application = app
self.ip: str = "0.0.0.0"
self.port: int = 80
self.socket: socket.socket = socket.socket()
self.main_thread: threading.Thread = threading.Thread(target=self.handle_requests, daemon=True)
def start(self) -> None:
self.running = True
self.socket.settimeout(1)
self.socket.bind((self.ip, self.port))
self.socket.listen(20)
self.main_thread.start()
def stop(self) -> None:
self.running = False
def replace_application(self, app: application.Application) -> None:
self.application = app
def handle_requests(self) -> None:
while self.running:
try:
self.handle_request(*self.socket.accept())
except TimeoutError:
pass
# noinspection PyUnusedLocal
def handle_request(self, conn: socket.socket, addr: tuple[str, int]) -> None:
conn.settimeout(1)
try:
request: str = conn.recv(1024).decode("utf-8")
except (ConnectionAbortedError, ConnectionResetError, UnicodeDecodeError, TimeoutError):
conn.close()
return
if request.startswith("GET"):
page: str = request.split()[1].split("?")[0]
parameters: dict[str, str] = {}
if "?" in request.split()[1].strip("/"):
for param in request.split()[1].strip("/").split("?")[1].split("&"):
parameters[param.split("=")[0]] = param.split("=")[1]
cookies: dict[str, str] = {}
if "Cookie: " in request:
cookie_list: list[str] = request.split("Cookie: ")[1].split("\r\n")[0].split("; ")
for cookie in cookie_list:
key, value = cookie.split("=")
cookies[key] = value
if page in self.application.GET:
header, payload = self.application.GET[page][0](parameters, cookies, self.application.GET[page][1])
conn.send(header + payload)
else:
print(page)
elif request.startswith("POST"):
print(request)
else:
print(request)
conn.close()