16.4-dars: pickle va xavfsizlik

pickle.loads(zararli_baytlar)   →   ishonchsiz kod DARHOL bajariladi

class XavfsizUnpickler(pickle.Unpickler):
    RUXSAT = {("builtins", "list"), ("builtins", "dict")}
    def find_class(self, modul, nom):
        if (modul, nom) in self.RUXSAT:
            return super().find_class(modul, nom)
        raise pickle.UnpicklingError(f"taqiqlangan: {modul}.{nom}")

import pickle

with open("holat.pkl", "wb") as f:
    pickle.dump(obyekt, f, protocol=pickle.HIGHEST_PROTOCOL)
with open("holat.pkl", "rb") as f:            # ⚠️ faqat ishonchli fayl
    obyekt = pickle.load(f)

baytlar = pickle.dumps(obyekt)
obyekt = pickle.loads(baytlar)

ishonchsiz pickle — HECH QACHON load
binar rejim: "wb" / "rb"
uzoq saqlash / almashinuv — JSON, pickle emas
o'zingiznikini imzolang (HMAC)
saqlab bo'lmaydigan qism — __getstate__/__setstate__
faqat Python-ichi, vaqtinchalik holat uchun

"""dumps/loads; protokol va hajm; nima pickle bo'ladi, nima yo'q; o'zaro va rekursiv havolalar; JSON bilan solishtirish; sinf namunasi."""

import json
import pickle
from datetime import datetime


class Nuqta:
    def __init__(self, x: int, y: int) -> None:
        self.x, self.y = x, y

    def __repr__(self) -> str:
        return f"Nuqta({self.x}, {self.y})"


def main() -> None:
    print("=== 1. Aylanib qaytish ===")
    obyekt = {"son": 42, "matn": "salom", "kortej": (1, 2), "toplam": {3, 1, 2},
              "sana": datetime(2026, 9, 17, 9, 30), "ichki": {"a": [1, {"b": 2}]}, "yoq": None}
    qaytgan = pickle.loads(pickle.dumps(obyekt))
    print(f"  teng: {qaytgan == obyekt}")
    print(f"  set turi saqlandi: {type(qaytgan['toplam']).__name__}, kortej: {type(qaytgan['kortej']).__name__}")

    print("\n=== 2. Protokollar va hajm ===")
    for protokol in (0, 2, 5):
        baytlar = pickle.dumps(obyekt, protocol=protokol)
        print(f"  protokol {protokol}: {len(baytlar):3} bayt, boshi: {baytlar[:4]!r}")
    print(f"  DEFAULT={pickle.DEFAULT_PROTOCOL}, HIGHEST={pickle.HIGHEST_PROTOCOL}")

    print("\n=== 3. Nima pickle bo'lmaydi ===")
    def ichki_funksiya() -> None:
        pass
    for nom, qiymat in [("lambda", lambda x: x), ("ichki funksiya", ichki_funksiya)]:
        try:
            pickle.dumps(qiymat)
        except pickle.PicklingError as xato:
            print(f"  {nom}: PicklingError: {str(xato).split(':')[0]}")
    try:
        pickle.dumps(open(__file__, encoding="utf-8"))
    except TypeError as xato:
        print(f"  ochiq fayl: TypeError: {xato}")

    print("\n=== 4. ⭐ O'zaro va rekursiv havolalar ===")
    umumiy = [1, 2, 3]
    juft = [umumiy, umumiy]
    natija = pickle.loads(pickle.dumps(juft))
    print(f"  bir xil obyekt bittaligicha saqlandi: {natija[0] is natija[1]}")
    halqa: list = []
    halqa.append(halqa)
    tiklangan = pickle.loads(pickle.dumps(halqa))
    print(f"  rekursiv tuzilma: {tiklangan[0] is tiklangan}")
    try:
        json.dumps(halqa)
    except ValueError as xato:
        print(f"  JSON rekursivni qila olmaydi: {type(xato).__name__}: {xato}")

    print("\n=== 5. pickle vs JSON hajmi ===")
    malumot = {"foydalanuvchilar": [{"id": i, "faol": i % 2 == 0} for i in range(50)]}
    print(f"  json:   {len(json.dumps(malumot).encode()):4} bayt")
    print(f"  pickle: {len(pickle.dumps(malumot)):4} bayt")


if __name__ == "__main__":
    main()

=== 1. Aylanib qaytish ===
  teng: True
  set turi saqlandi: set, kortej: tuple

=== 2. Protokollar va hajm ===
  protokol 0: 310 bayt, boshi: b'(dp0'
  protokol 2: 261 bayt, boshi: b'\x80\x02}q'
  protokol 5: 158 bayt, boshi: b'\x80\x05\x95\x93'
  DEFAULT=5, HIGHEST=5

=== 3. Nima pickle bo'lmaydi ===
  lambda: PicklingError: Can't pickle local object <function main.<locals>.<lambda> at 0x000001C40CE0BC10>
  ichki funksiya: PicklingError: Can't pickle local object <function main.<locals>.ichki_funksiya at 0x000001C40CE0BB60>
  ochiq fayl: TypeError: cannot pickle 'TextIOWrapper' instances

=== 4. ⭐ O'zaro va rekursiv havolalar ===
  bir xil obyekt bittaligicha saqlandi: True
  rekursiv tuzilma: True
  JSON rekursivni qila olmaydi: ValueError: Circular reference detected

=== 5. pickle vs JSON hajmi ===
  json:   1337 bayt
  pickle:  596 bayt

"""__reduce__ orqali pickle.load kod bajaradi (zararsiz namoyish); find_class bilan cheklangan Unpickler; xavfsiz data baribir yuklanadi; HMAC imzo bilan yaxlitlik."""

import hashlib
import hmac
import io
import pickle

IMZO_KALITI = b"faqat-server-biladigan-maxfiy-kalit"


class Zararsiz_Namoyish:
    """__reduce__ yuklashda print chaqiradi — haqiqiy hujumda bu os.system bo'lardi."""

    def __reduce__(self):
        return (print, ("      [!] pickle.load ushbu kodni BAJARDI — haqiqiy hujumda bu buyruq bo'lardi",))


class XavfsizUnpickler(pickle.Unpickler):
    RUXSAT = {("builtins", "list"), ("builtins", "dict"), ("builtins", "set"),
              ("builtins", "tuple"), ("builtins", "str"), ("builtins", "int")}

    def find_class(self, modul: str, nom: str):
        if (modul, nom) in self.RUXSAT:
            return super().find_class(modul, nom)
        raise pickle.UnpicklingError(f"taqiqlangan turdagi obyekt: {modul}.{nom}")


def imzolab_saqla(obyekt: object) -> bytes:
    data = pickle.dumps(obyekt)
    imzo = hmac.new(IMZO_KALITI, data, hashlib.sha256).digest()
    return imzo + data


def tekshirib_yukla(blob: bytes) -> object:
    imzo, data = blob[:32], blob[32:]
    kutilgan = hmac.new(IMZO_KALITI, data, hashlib.sha256).digest()
    if not hmac.compare_digest(imzo, kutilgan):
        raise ValueError("imzo mos emas — ma'lumot buzilgan yoki soxta")
    return pickle.loads(data)


def main() -> None:
    print("=== 1. ⚠️ pickle.load kod bajaradi ===")
    zararli = pickle.dumps(Zararsiz_Namoyish())
    print("  quyidagi qator zararli pickle ni yuklaydi:")
    pickle.loads(zararli)
    print("  ↑ ma'lumot deb o'ylagan narsangiz kod ekan")

    print("\n=== 2. Cheklangan Unpickler ===")
    try:
        XavfsizUnpickler(io.BytesIO(zararli)).load()
    except pickle.UnpicklingError as xato:
        print(f"  zararli bloklandi: {xato}")
    xavfsiz_data = pickle.dumps({"foydalanuvchi": "aziz", "rollar": ["oquvchi"]})
    print(f"  xavfsiz data yuklanadi: {XavfsizUnpickler(io.BytesIO(xavfsiz_data)).load()}")
    print("  ⚠️ oq ro'yxat ham to'liq yechim emas — eng yaxshisi ishonchsizni umuman yuklamaslik")

    print("\n=== 3. HMAC imzo ===")
    blob = imzolab_saqla({"model": "v3", "koeffitsiyentlar": [0.1, 0.5, 0.9]})
    print(f"  imzolangan: {len(blob)} bayt (32 imzo + {len(blob) - 32} data)")
    print(f"  to'g'ri imzo yuklandi: {tekshirib_yukla(blob)}")
    buzilgan = blob[:-1] + bytes([blob[-1] ^ 0x01])
    try:
        tekshirib_yukla(buzilgan)
    except ValueError as xato:
        print(f"  bitta bayt o'zgardi: ValueError: {xato}")
    soxta = imzolab_saqla({"model": "soxta"})
    soxta = bytes(32) + soxta[32:]        # imzoni nolga almashtirish urinishi
    try:
        tekshirib_yukla(soxta)
    except ValueError as xato:
        print(f"  soxta imzo: ValueError: {xato}")


if __name__ == "__main__":
    main()

=== 1. ⚠️ pickle.load kod bajaradi ===
  quyidagi qator zararli pickle ni yuklaydi:
      [!] pickle.load ushbu kodni BAJARDI — haqiqiy hujumda bu buyruq bo'lardi
  ↑ ma'lumot deb o'ylagan narsangiz kod ekan

=== 2. Cheklangan Unpickler ===
  zararli bloklandi: taqiqlangan turdagi obyekt: builtins.print
  xavfsiz data yuklanadi: {'foydalanuvchi': 'aziz', 'rollar': ['oquvchi']}
  ⚠️ oq ro'yxat ham to'liq yechim emas — eng yaxshisi ishonchsizni umuman yuklamaslik

=== 3. HMAC imzo ===
  imzolangan: 111 bayt (32 imzo + 79 data)
  to'g'ri imzo yuklandi: {'model': 'v3', 'koeffitsiyentlar': [0.1, 0.5, 0.9]}
  bitta bayt o'zgardi: ValueError: imzo mos emas — ma'lumot buzilgan yoki soxta
  soxta imzo: ValueError: imzo mos emas — ma'lumot buzilgan yoki soxta

"""__getstate__/__setstate__: saqlab bo'lmaydigan qismni chetlab o'tish va sukut qiymatlar; persistent_id bilan tashqi havola; sinf o'zgarganda mo'rtlik."""

import io
import pickle


class Ulanish:
    """Soketni saqlab bo'lmaydi — holatdan chiqaramiz, yuklashda qayta ochamiz."""

    def __init__(self, host: str) -> None:
        self.host = host
        self.urinishlar = 0
        self._soket = f"<{host} ga ochiq soket>"      # saqlab bo'lmaydi

    def __getstate__(self) -> dict:
        holat = self.__dict__.copy()
        del holat["_soket"]
        return holat

    def __setstate__(self, holat: dict) -> None:
        self.__dict__.update(holat)
        self.__dict__.setdefault("qayta_ulanish", True)   # yangi maydon uchun sukut
        self._soket = None                                # yuklashda yopiq


class DBYozuv:
    def __init__(self, jadval: str, id: int) -> None:
        self.jadval, self.id = jadval, id


class RefPickler(pickle.Pickler):
    def persistent_id(self, obyekt: object):
        if isinstance(obyekt, DBYozuv):
            return ("db", obyekt.jadval, obyekt.id)       # qiymat emas, havola saqlanadi
        return None


class RefUnpickler(pickle.Unpickler):
    def persistent_load(self, pid: tuple):
        tur, jadval, id = pid
        if tur == "db":
            return f"<{jadval}#{id} bazadan yuklab olindi>"
        raise pickle.UnpicklingError(f"noma'lum havola: {pid}")


class Foydalanuvchi:
    """Yangi versiyada 'til' maydoni qo'shildi; eski pickle da u yo'q."""

    def __init__(self, ism: str) -> None:
        self.ism = ism
        self.til = "uz"

    def __setstate__(self, holat: dict) -> None:
        self.__dict__.update(holat)
        self.__dict__.setdefault("til", "uz")     # eski pickle uchun sukut


def main() -> None:
    print("=== 1. __getstate__ / __setstate__ ===")
    ulanish = Ulanish("db.example")
    ulanish.urinishlar = 3
    tiklangan = pickle.loads(pickle.dumps(ulanish))
    print(f"  host: {tiklangan.host}, urinishlar: {tiklangan.urinishlar}")
    print(f"  soket saqlanmadi: {tiklangan._soket}")
    print(f"  yangi maydon sukut bilan tiklandi: {tiklangan.qayta_ulanish}")

    print("\n=== 2. persistent_id — tashqi havola ===")
    bufer = io.BytesIO()
    RefPickler(bufer).dump({"buyurtma": 100, "mijoz": DBYozuv("foydalanuvchilar", 7),
                            "mahsulot": DBYozuv("mahsulotlar", 42)})
    natija = RefUnpickler(io.BytesIO(bufer.getvalue())).load()
    print(f"  {natija}")

    print("\n=== 3. ⚠️ Versiyalash mo'rtligi ===")
    # Eski pickle da 'til' maydoni yo'q edi — __setstate__ sukut beradi
    qurilgan = Foydalanuvchi.__new__(Foydalanuvchi)
    qurilgan.__setstate__({"ism": "Aziz"})            # eski holatda 'til' yo'q
    print(f"  eski holatdan tiklandi: ism={qurilgan.ism}, til={qurilgan.til} (sukut qo'yildi)")
    # Sinf nomi topilmasa — yuklash buziladi
    saqlangan = pickle.dumps(Foydalanuvchi("Bek"))
    buzuq = saqlangan.replace(b"Foydalanuvchi", b"Boshqasinf")   # sinf nomini o'zgartirish
    try:
        pickle.loads(buzuq)
    except (AttributeError, pickle.UnpicklingError, KeyError, EOFError, ValueError) as xato:
        print(f"  sinf nomi o'zgarsa: {type(xato).__name__}")
    print("  ⭐ pickle sinf HOLATINI saqlaydi, ta'rifini emas — versiyalararo saqlashga yaramaydi")


if __name__ == "__main__":
    main()

=== 1. __getstate__ / __setstate__ ===
  host: db.example, urinishlar: 3
  soket saqlanmadi: None
  yangi maydon sukut bilan tiklandi: True

=== 2. persistent_id — tashqi havola ===
  {'buyurtma': 100, 'mijoz': '<foydalanuvchilar#7 bazadan yuklab olindi>', 'mahsulot': '<mahsulotlar#42 bazadan yuklab olindi>'}

=== 3. ⚠️ Versiyalash mo'rtligi ===
  eski holatdan tiklandi: ism=Aziz, til=uz (sukut qo'yildi)
  sinf nomi o'zgarsa: UnpicklingError
  ⭐ pickle sinf HOLATINI saqlaydi, ta'rifini emas — versiyalararo saqlashga yaramaydi

"""Diskdagi pickle kesh: HMAC imzo, versiya belgisi, TTL, atomar yozish, buzilgan/soxta/eskirgan keshni rad etib qayta hisoblash."""

import hashlib
import hmac
import os
import pickle
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path

KESH_KALITI = b"server-maxfiy-kalit"
KESH_VERSIYASI = 2


@dataclass
class Hisobot:
    nom: str
    qatorlar: int
    yaratilgan: float


class Kesh:
    def __init__(self, katalog: Path, ttl: float) -> None:
        self.katalog = katalog
        self.ttl = ttl
        self.hisoblashlar = 0

    def _yol(self, kalit: str) -> Path:
        xesh = hashlib.sha256(kalit.encode()).hexdigest()[:16]
        return self.katalog / f"{xesh}.pkl"

    def _yoz(self, yol: Path, qiymat: object) -> None:
        paket = {"versiya": KESH_VERSIYASI, "vaqt": time.time(), "qiymat": qiymat}
        data = pickle.dumps(paket, protocol=pickle.HIGHEST_PROTOCOL)
        imzo = hmac.new(KESH_KALITI, data, hashlib.sha256).digest()
        vaqtinchalik = yol.with_suffix(".tmp")
        vaqtinchalik.write_bytes(imzo + data)
        os.replace(vaqtinchalik, yol)                 # atomar almashtirish

    def _oqi(self, yol: Path) -> object:
        blob = yol.read_bytes()
        imzo, data = blob[:32], blob[32:]
        if not hmac.compare_digest(imzo, hmac.new(KESH_KALITI, data, hashlib.sha256).digest()):
            raise ValueError("imzo mos emas")
        paket = pickle.loads(data)
        if paket["versiya"] != KESH_VERSIYASI:
            raise ValueError(f"versiya {paket['versiya']} ≠ {KESH_VERSIYASI}")
        if time.time() - paket["vaqt"] > self.ttl:
            raise ValueError("muddati o'tgan")
        return paket["qiymat"]

    def ol(self, kalit: str, hisobla):
        yol = self._yol(kalit)
        if yol.exists():
            try:
                return self._oqi(yol), "keshdan"
            except ValueError as xato:
                sabab = str(xato)
        else:
            sabab = "yo'q"
        qiymat = hisobla()
        self.hisoblashlar += 1
        self._yoz(yol, qiymat)
        return qiymat, f"qayta hisoblandi ({sabab})"


def main() -> None:
    with tempfile.TemporaryDirectory() as katalog:
        kesh = Kesh(Path(katalog), ttl=60)

        def qimmat_hisob() -> Hisobot:
            return Hisobot("oylik", qatorlar=10_000, yaratilgan=0.0)

        print("=== 1. Birinchi va ikkinchi murojaat ===")
        q1, holat1 = kesh.ol("oylik-2026-09", qimmat_hisob)
        q2, holat2 = kesh.ol("oylik-2026-09", qimmat_hisob)
        print(f"  1: {holat1}, 2: {holat2}")
        print(f"  qiymat bir xil: {q1 == q2}, jami hisoblashlar: {kesh.hisoblashlar}")

        print("\n=== 2. Buzilgan kesh fayli ===")
        yol = kesh._yol("oylik-2026-09")
        blob = bytearray(yol.read_bytes())
        blob[40] ^= 0x01                              # bitta baytni buzish
        yol.write_bytes(blob)
        q3, holat3 = kesh.ol("oylik-2026-09", qimmat_hisob)
        print(f"  3: {holat3} (imzo mos kelmadi → qayta hisoblandi)")

        print("\n=== 3. Soxta kesh (imzosiz hujum) ===")
        soxta = pickle.dumps({"versiya": KESH_VERSIYASI, "vaqt": time.time(), "qiymat": "soxta"})
        yol.write_bytes(bytes(32) + soxta)            # nol imzo bilan
        q4, holat4 = kesh.ol("oylik-2026-09", qimmat_hisob)
        print(f"  4: {holat4}, qiymat soxta emas: {q4 != 'soxta'}")

        print("\n=== 4. Eskirgan versiya ===")
        eski = pickle.dumps({"versiya": 1, "vaqt": time.time(), "qiymat": "eski format"})
        imzo = hmac.new(KESH_KALITI, eski, hashlib.sha256).digest()
        yol.write_bytes(imzo + eski)
        q5, holat5 = kesh.ol("oylik-2026-09", qimmat_hisob)
        print(f"  5: {holat5}")
        print(f"  jami hisoblashlar: {kesh.hisoblashlar} (har xato → qayta)")


if __name__ == "__main__":
    main()

=== 1. Birinchi va ikkinchi murojaat ===
  1: qayta hisoblandi (yo'q), 2: keshdan
  qiymat bir xil: True, jami hisoblashlar: 1

=== 2. Buzilgan kesh fayli ===
  3: qayta hisoblandi (imzo mos emas) (imzo mos kelmadi → qayta hisoblandi)

=== 3. Soxta kesh (imzosiz hujum) ===
  4: qayta hisoblandi (imzo mos emas), qiymat soxta emas: True

=== 4. Eskirgan versiya ===
  5: qayta hisoblandi (versiya 1 ≠ 2)
  jami hisoblashlar: 4 (har xato → qayta)

model = pickle.loads(sorov.body)                   # ❌ RCE
data = json.loads(sorov.body)                      # ✅ yoki imzolangan/cheklangan

open("a.pkl", "w")                                 # ❌ TypeError yoki buzilgan
open("a.pkl", "wb")                                # ✅

pickle.dump(sozlama, f)                            # ❌ versiya o'zgarsa yo'qoladi
json.dump(sozlama, f, ensure_ascii=False)          # ✅

pickle.dumps(obj_with_socket)                      # ❌ TypeError
# ✅ __getstate__ da soketni chiqaring, __setstate__ da qayta oching

qiymat = pickle.load(kesh_fayli)                   # ❌ kesh buzilsa/almashtirsa
# ✅ HMAC imzo + versiya + TTL tekshiruvi

pool.map(lambda x: x * 2, data)                    # ❌ PicklingError
pool.map(kopaytir, data)                           # ✅ modul darajasidagi funksiya

try: pickle.loads(b)
except Exception: pass                             # ❌ kod allaqachon bajarilgan bo'lishi mumkin
# ✅ yuklashdan OLDIN imzoni tekshiring

pickle.dumps(obj)                                  # protokol 5 — eski Python o'qiy olmaydi
pickle.dumps(obj, protocol=2)                      # ✅ moslik kerak bo'lsa

import pickle, json
1.  print(type(pickle.dumps([1, 2])).__name__)
2.  print(pickle.loads(pickle.dumps((1, 2))) == (1, 2))
3.  print(pickle.loads(pickle.dumps({1, 2, 3})) == {1, 2, 3})
4.  a = [1]; b = [a, a]; r = pickle.loads(pickle.dumps(b)); print(r[0] is r[1])
5.  print(pickle.DEFAULT_PROTOCOL)
6.  print(pickle.dumps({}, protocol=2)[:2])
7.  try:
        pickle.dumps(lambda: 1)
    except Exception as e:
        print(type(e).__name__)
8.  x = []; x.append(x); print(pickle.loads(pickle.dumps(x))[0] is not None)
9.  print(len(pickle.dumps(0)) < len(json.dumps(0)))
10. print(pickle.loads(pickle.dumps("Oʻzbek")) == "Oʻzbek")
11. try:
        json.dumps({1, 2})
    except Exception as e:
        print(type(e).__name__)
12. print(pickle.HIGHEST_PROTOCOL >= pickle.DEFAULT_PROTOCOL)

1.  def saqla(yol, obyekt):
        with open(yol, "w") as f:
            pickle.dump(obyekt, f)

2.  def api_yukla(sorov_tanasi):
        return pickle.loads(sorov_tanasi)          # foydalanuvchidan keladi

3.  def sozlama_saqla(yol, sozlama):
        with open(yol, "wb") as f:
            pickle.dump(sozlama, f)                 # boshqa tildagi xizmat ham o'qiydi

4.  def kesh_oqi(yol):
        with open(yol, "rb") as f:
            return pickle.load(f)                   # kesh papkasiga hamma yoza oladi

5.  def parallel(sonlar):
        with multiprocessing.Pool() as p:
            return p.map(lambda n: n * n, sonlar)