Amaliy loyiha - o'z HTTP serveringiz

import os
import socket
import threading


class Sorov:
    def __init__(self, usul, yol, versiya, sarlavhalar, tana):
        self.usul = usul
        self.yol = yol
        self.versiya = versiya
        self.sarlavhalar = sarlavhalar
        self.tana = tana

    def sarlavha(self, nom, standart=None):
        return self.sarlavhalar.get(nom.lower(), standart)

    def tirik_qolsinmi(self):
        ulanish = self.sarlavha("connection", "").lower()
        if self.versiya == "HTTP/1.1":
            return ulanish != "close"
        return ulanish == "keep-alive"


class SorovOqiuvchi:
    """Bayt oqimidan to'liq so'rovni ajratib oladi (8-bo'lim)."""

    CHEGARA = 64 * 1024

    def __init__(self, ulanish):
        self.ulanish = ulanish
        self.bufer = b""

    def _yana_oqi(self):
        bolak = self.ulanish.recv(4096)
        if not bolak:
            return False
        self.bufer += bolak
        if len(self.bufer) > self.CHEGARA:
            raise ValueError("so'rov juda katta")
        return True

    def oqi(self):
        while b"\r\n\r\n" not in self.bufer:
            if not self._yana_oqi():
                return None
        bosh, qolgan = self.bufer.split(b"\r\n\r\n", 1)
        qatorlar = bosh.decode("latin-1").split("\r\n")

        qismlar = qatorlar[0].split(" ")
        if len(qismlar) != 3:
            raise ValueError("noto'g'ri so'rov qatori")
        usul, yol, versiya = qismlar

        sarlavhalar = {}
        for q in qatorlar[1:]:
            if ":" not in q:
                raise ValueError("noto'g'ri sarlavha")
            nom, qiymat = q.split(":", 1)
            sarlavhalar[nom.strip().lower()] = qiymat.strip()

        uzunlik = int(sarlavhalar.get("content-length", 0))
        while len(qolgan) < uzunlik:
            self.bufer = qolgan
            if not self._yana_oqi():
                return None
            qolgan = self.bufer
        tana, self.bufer = qolgan[:uzunlik], qolgan[uzunlik:]
        return Sorov(usul, yol, versiya, sarlavhalar, tana)

IZOHLAR = {
    200: "OK", 204: "No Content", 400: "Bad Request",
    403: "Forbidden", 404: "Not Found",
    405: "Method Not Allowed", 500: "Internal Server Error",
}


class Javob:
    def __init__(self, kod=200, tana=b"", tur="text/plain; charset=utf-8"):
        if isinstance(tana, str):
            tana = tana.encode()
        self.kod = kod
        self.tana = tana
        self.sarlavhalar = {"Content-Type": tur}

    def baytlar(self, tirik):
        izoh = IZOHLAR.get(self.kod, "Unknown")
        self.sarlavhalar["Content-Length"] = str(len(self.tana))
        self.sarlavhalar["Connection"] = "keep-alive" if tirik else "close"
        qatorlar = [f"HTTP/1.1 {self.kod} {izoh}"]
        for nom, qiymat in self.sarlavhalar.items():
            qatorlar.append(f"{nom}: {qiymat}")
        bosh = "\r\n".join(qatorlar) + "\r\n\r\n"
        return bosh.encode() + self.tana

class Marshrutlovchi:
    def __init__(self):
        self.qoidalar = {}

    def qosh(self, usul, yol, ishlovchi):
        self.qoidalar[(usul, yol)] = ishlovchi

    def top(self, sorov):
        ishlovchi = self.qoidalar.get((sorov.usul, sorov.yol))
        if ishlovchi:
            return ishlovchi
        # Yo'l bor, lekin usul boshqa bo'lsa - 405
        for usul, yol in self.qoidalar:
            if yol == sorov.yol:
                return lambda s: Javob(405, "Usul qo'llab-quvvatlanmaydi")
        return None


def statik_ishlovchi(ildiz):
    def ishlovchi(sorov):
        nisbiy = sorov.yol[len("/statik/"):]
        # XAVFSIZLIK: yo'lni normallashtirib, ildizdan chiqmasligini
        # tekshiramiz - aks holda ../../etc/passwd o'qib ketiladi
        toliq = os.path.realpath(os.path.join(ildiz, nisbiy))
        if not toliq.startswith(os.path.realpath(ildiz) + os.sep):
            return Javob(403, "Ruxsat yo'q")
        if not os.path.isfile(toliq):
            return Javob(404, "Fayl topilmadi")
        with open(toliq, "rb") as f:
            return Javob(200, f.read(), "text/plain; charset=utf-8")
    return ishlovchi

class Server:
    def __init__(self, port, marshrutlovchi, statik=None):
        self.marshrutlovchi = marshrutlovchi
        self.statik = statik
        self.soket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.soket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.soket.bind(("127.0.0.1", port))
        self.soket.listen(8)
        self.jurnal = []

    def ulanishni_ishla(self, ulanish):
        oquvchi = SorovOqiuvchi(ulanish)
        try:
            while True:
                try:
                    sorov = oquvchi.oqi()
                except ValueError:
                    ulanish.sendall(Javob(400, "Buzuq so'rov").baytlar(False))
                    return
                if sorov is None:
                    return

                if self.statik and sorov.yol.startswith("/statik/"):
                    ishlovchi = self.statik
                else:
                    ishlovchi = self.marshrutlovchi.top(sorov)

                if ishlovchi is None:
                    javob = Javob(404, "Sahifa topilmadi")
                else:
                    try:
                        javob = ishlovchi(sorov)
                    except Exception:
                        javob = Javob(500, "Ichki xato")

                tirik = sorov.tirik_qolsinmi()
                self.jurnal.append(f"{sorov.usul} {sorov.yol} -> {javob.kod}")
                ulanish.sendall(javob.baytlar(tirik))
                if not tirik:
                    return
        finally:
            ulanish.close()

    def nechta_ishla(self, nechta):
        for _ in range(nechta):
            ulanish, _ = self.soket.accept()
            self.ulanishni_ishla(ulanish)

    def yop(self):
        self.soket.close()

import tempfile

PORT = 9980
ildiz = tempfile.mkdtemp()
with open(os.path.join(ildiz, "salom.txt"), "w") as f:
    f.write("statik fayl mazmuni")

m = Marshrutlovchi()
m.qosh("GET", "/", lambda s: Javob(200, "Bosh sahifa"))
m.qosh("GET", "/salom", lambda s: Javob(200, "Salom, tarmoq!"))
m.qosh("POST", "/echo", lambda s: Javob(200, s.tana))
m.qosh("GET", "/xato", lambda s: 1 / 0)

server = Server(PORT, m, statik_ishlovchi(ildiz))
oqim = threading.Thread(target=server.nechta_ishla, args=(7,), daemon=True)
oqim.start()


def sorov_yubor(xom):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(5)
    s.connect(("127.0.0.1", PORT))
    s.sendall(xom)
    javob = b""
    while True:
        b = s.recv(4096)
        if not b:
            break
        javob += b
    s.close()
    bosh, tana = javob.split(b"\r\n\r\n", 1)
    return bosh.decode().split("\r\n")[0], tana.decode()


SINOVLAR = [
    ("bosh sahifa", b"GET / HTTP/1.1\r\nHost: x\r\n"
                    b"Connection: close\r\n\r\n"),
    ("mavjud yo'l", b"GET /salom HTTP/1.1\r\nHost: x\r\n"
                    b"Connection: close\r\n\r\n"),
    ("yo'q sahifa", b"GET /yoq HTTP/1.1\r\nHost: x\r\n"
                    b"Connection: close\r\n\r\n"),
    ("noto'g'ri usul", b"DELETE /salom HTTP/1.1\r\nHost: x\r\n"
                       b"Connection: close\r\n\r\n"),
    ("POST tana bilan", b"POST /echo HTTP/1.1\r\nHost: x\r\n"
                        b"Content-Length: 11\r\nConnection: close\r\n\r\n"
                        b"salom tana!"),
    ("statik fayl", b"GET /statik/salom.txt HTTP/1.1\r\nHost: x\r\n"
                    b"Connection: close\r\n\r\n"),
    ("yo'ldan chiqish", b"GET /statik/../../etc/passwd HTTP/1.1\r\n"
                        b"Host: x\r\nConnection: close\r\n\r\n"),
]

for nom, xom in SINOVLAR:
    holat, tana = sorov_yubor(xom)
    print(f"{nom:<18} {holat:<26} {tana[:28]}")

oqim.join()
server.yop()

print()
print("server jurnali:")
for q in server.jurnal:
    print("  ", q)

PORT2 = 9981

m2 = Marshrutlovchi()
m2.qosh("GET", "/", lambda s: Javob(200, "javob"))

server2 = Server(PORT2, m2)
oqim2 = threading.Thread(target=server2.nechta_ishla, args=(1,), daemon=True)
oqim2.start()

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
s.connect(("127.0.0.1", PORT2))

# Bitta ulanishda uchta so'rov
for i in range(1, 4):
    oxirgi = (i == 3)
    ulanish = "close" if oxirgi else "keep-alive"
    s.sendall(f"GET / HTTP/1.1\r\nHost: x\r\n"
              f"Connection: {ulanish}\r\n\r\n".encode())
    javob = s.recv(4096).decode()
    birinchi = javob.split("\r\n")[0]
    ulanish_sarlavhasi = [q for q in javob.split("\r\n")
                          if q.startswith("Connection")][0]
    print(f"  {i}-so'rov: {birinchi}  |  {ulanish_sarlavhasi}")

s.close()
oqim2.join()
server2.yop()

print()
print("qabul qilingan ulanishlar soni: 1")
print("bajarilgan so'rovlar soni     :", len(server2.jurnal))

PORT3 = 9982

m3 = Marshrutlovchi()
m3.qosh("POST", "/echo", lambda s: Javob(200, f"{len(s.tana)} bayt keldi"))

server3 = Server(PORT3, m3)
oqim3 = threading.Thread(target=server3.nechta_ishla, args=(1,), daemon=True)
oqim3.start()

TANA = b"x" * 5000

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
s.connect(("127.0.0.1", PORT3))

# Avval faqat sarlavhalarni yuboramiz - tana hali yo'q
s.sendall(f"POST /echo HTTP/1.1\r\nHost: x\r\n"
          f"Content-Length: {len(TANA)}\r\n"
          f"Connection: close\r\n\r\n".encode())

# Tanani KICHIK bo'laklarda, alohida yuboramiz
for i in range(0, len(TANA), 500):
    s.sendall(TANA[i:i + 500])

javob = b""
while True:
    b = s.recv(4096)
    if not b:
        break
    javob += b
s.close()
oqim3.join()
server3.yop()

bosh, tana = javob.split(b"\r\n\r\n", 1)
print("yuborilgan tana hajmi:", len(TANA))
print("javob                :", tana.decode())
print("to'liq o'qildimi     :", tana.decode().startswith(str(len(TANA))))