تصميم محفظة ما بعد الكم API: أنماط REST وWebSocket

# تنفيذ نقاط نهاية العنوان (FastAPI) من com.fastapi يستورد FastAPI، يعتمد على HTTPException من متحذلق يستورد BaseModel من الكتابة يستورد القائمة، اختيارية يستورد تطبيق base64 = FastAPI(العنوان="محفظة SynX API"النسخة="1.0.0") فصل عنوانالاستجابة(النموذج الأساسي): """عنوان بمفاتيح عمومية ما بعد الكم""" العنوان: مسار الشارع: شارع kyber_public_key: شارع # Base64 مشفر (1,184 بايت) sphincs_public_key: شارع # Base64 مشفر (32 بايت) الرصيد: int hanging_balance: int create_at: str فصل طلب اشتقاق العنوان(BaseModel): الحساب: int = 0 التغيير: int = 0 الفهرس: اختياري[int] = لا شيء # الزيادة التلقائية إذا لم يكن هناك @app.get("/API/v1/العناوين"، Response_model=قائمة[عنوانالاستجابة]) تعريف غير متزامن list_addresses( Wallet_id: str = Depends(get_authenticated_wallet)، تخطي: int = 0، الحد: int = 50 ): """ قائمة عناوين المحفظة مع الأرصدة ملاحظة: المفاتيح العامة Kyber كبيرة (1.2 كيلو بايت). للإدراج، فكر في استبعاد المفاتيح وجلبها بشكل منفصل. """ العناوين = انتظر Address_service.list_addresses( Wallet_id, Skip=skip, Limit=limit ) يعود [ عنوانالاستجابة( العنوان=addr.address، المسار=addr.path، kyber_public_key=base64.b64encode(addr.kyber_pk).decode()، sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode()، Balance=addr.balance، hanging_balance=addr.pending_balance، create_at=addr.created_at.isoformat() ) ل عنوان in العناوين ] @app.post("/API/v1/العناوين/اشتقاق"، Response_model=عنوانالاستجابة) تعريف غير متزامن derive_address( طلب: طلب اشتقاق العنوان، Wallet_id: str = Depends(get_authenticated_wallet) ): """اشتقاق عنوان جديد في مسار الاشتقاق المحدد""" العنوان = انتظر Address_service.derive_address( معرف المحفظة، الحساب = الطلب. الحساب، التغيير = الطلب. التغيير، الفهرس = الطلب. الفهرس ) يعود عنوانالاستجابة(...)

فصل ملخص المعاملة(النموذج الأساسي): """معاملة بدون بيانات التوقيع الكاملة (للقوائم)""" tx_id: الطابع الزمني للسلسلة: عدد المدخلات str: عدد المخرجات int: المبلغ int: رسوم int: تأكيدات int: حالة int: str # "معلق"، "مؤكد"، "فشل" فصل المعاملة كاملة(النموذج الأساسي): """المعاملة الكاملة بما في ذلك التوقيعات""" tx_id: إصدار str: الطابع الزمني int: مدخلات str: قائمة ["مخطط إدخال المعاملات"] النواتج: قائمة["مخطط إخراج المعاملات"] الرسوم: تأكيدات int: int block_hash: اختياري[str] Raw_hex: str # معاملة متسلسلة كاملة فصل TransactionInputSchema(BaseModel): prev_tx_id: str prev_output_index: مقدار int: عنوان int: توقيع str: str # Base64 (حوالي 10.5 كيلو بايت لـ SPHINCS+-SHAKE-128s، 7856 بايت خام) public_key: شارع # Base64 (44 بايت لـ SPHINCS+) فصل طلب بناء المعاملات(BaseModel): المخرجات: قائمة ["مواصفات الإخراج"]fee_rate: اختياري[int] = لا شيء # الحساب التلقائي إذا لم يكن هناك Change_address: اختياري[str] = لا شيء # التحديد التلقائي إذا لم يكن هناك فصل مواصفات الإخراج(BaseModel): المستلم: مقدار str: مذكرة int: اختياري[str] = لا شيء @app.get("/API/v1/المعاملات"، Response_model=قائمة[ملخص المعاملة]) تعريف غير متزامن list_transactions( Wallet_id: str = Depends(get_authenticated_wallet)، تخطي: int = 0، الحد: int = 20، الحالة: اختياري[str] = لا شيء ): """ معاملات القائمة (الملخصات فقط) يتم استبعاد التوقيعات من استجابات القائمة لتقليل الحمولة. استخدم GET /transactions/{tx_id} للمعاملة الكاملة مع التوقيعات. """ تكس = انتظر Transaction_service.list_transactions( Wallet_id، Skip=skip، Limit=limit، Status=status) يعود [tx.to_summary() ل tx in رسالة نصية] @app.get("/API/v1/transactions/{tx_id}"، Response_model=المعاملة كاملة) تعريف غير متزامن get_transaction( tx_id: str، Wallet_id: str = Depends(get_authenticated_wallet)، include_signatures: bool = True ): """ الحصول على تفاصيل المعاملة الكاملة قم بتعيين include_signatures=false لتقليل حجم الاستجابة إذا كنت تحتاج فقط إلى بيانات تعريف المعاملة. """ تكساس = انتظر معاملة_خدمة.get_transaction(wallet_id, tx_id) إذا لم يكن كذلك تكساس: يرفع HTTPException(status_code=404,detail="لم يتم العثور على المعاملة") يعود tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/المعاملات/الإنشاء") تعريف غير متزامن build_transaction( طلب: طلب بناء المعاملات، Wallet_id: str = Depends(get_authenticated_wallet) ): """ إنشاء معاملة غير موقعة إرجاع بيانات المعاملة جاهزة للتوقيع من جانب العميل. يحدث التوقيع على العميل لإبقاء المفاتيح الخاصة خارج الخادم. """ unsigned_tx = انتظر Transaction_service.build_transaction( Wallet_id، المخرجات=request.outputs،fee_rate=request.fee_rate،change_address=request.change_address) يعود { "unsigned_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "رسالة_التوقيع": base64.b64encode(unsigned_tx.tx_hash()).decode(), "المدخلات_إلى_توقيع": [ { "فِهرِس": i, "عنوان": عنوان inp، "كمية": المبلغ الوارد، "مسار_الاشتقاق": مسار inp } ل أنا، inp in تعداد (unsigned_tx.inputs) ]، "الرسوم_المقدرة": unsigned_tx.fee، "المقدرة_الحجم": unsigned_tx.estimated_size() }

# تدفق المصادقة باستخدام Kyber + SPHINCS+ يستورد oqs يستورد هاشليب يستورد hmac من التاريخ والوقت يستورد التاريخ والوقت، دلتا الوقت فصل QuantumSafeAuth: """ تدفق مصادقة API الآمن الكمي: 1. يرسل العميل المفتاح العام Kyber 2. يقوم الخادم بتغليف مفتاح الجلسة 3. يقوم العميل بفك التغليف للحصول على مفتاح الجلسة 4. الطلبات الموقعة مع HMAC باستخدام مفتاح الجلسة """ مواطنه __الحرف الأولي__(الذات): self.session_store = {} # في الإنتاج، استخدم Redis self.session_duration = timedelta(hours=24) تعريف غير متزامن Entire_session(self، Wallet_id: str،client_kyber_pk: بايت) -> dict: """ الخطوة 1: يبدأ العميل الجلسة باستخدام المفتاح العام Kyber يقوم الخادم بتغليف سر الجلسة لمفتاح العميل """ كيم = oqs.KeyEncapsulation("كايبر 768") النص المشفر، Shared_secret = kem.encap_secret(client_kyber_pk) # اشتقاق مفتاح الجلسة من السر المشترك session_key = hashlib.shake_256(Shared_secret + b"مفتاح الجلسة" .ملخص(32) # إنشاء معرف الجلسة session_id = hashlib.Blake2b(Shared_secret + str(datetime.utcnow()).encode(),Digest_size=16).hexdigest() # جلسة المتجر (من جانب الخادم) self.session_store[session_id] = { "معرف_المحفظة": معرف المحفظة، "مفتاح_الجلسة": مفتاح الجلسة، "تنتهي_في": datetime.utcnow() + self.session_duration، "تم الإنشاء": datetime.utcnow() } يعود { "معرف_الجلسة": معرف الجلسة، "النص المشفر": base64.b64encode(ciphertext).decode(), "تنتهي_في": (datetime.utcnow() + self.session_duration).isoformat() } مواطنه Verified_request( self، session_id: str، request_signature: بايت، request_data: بايت، الطابع الزمني: int ) -> اختياري[str]: """ التحقق من توقيع الطلب باستخدام مفتاح الجلسة وإرجاع معرف المحفظة إذا كان صالحًا، ولا شيء بخلاف ذلك """ الجلسة = self.session_store.get(session_id) إذا لم يكن كذلك حصة: يعود لا أحد # التحقق من انتهاء الصلاحية if datetime.utcnow() > الجلسة["تنتهي_في"]: ديل self.session_store[session_id] يعود لا أحد # التحقق من الطابع الزمني (منع إعادة التشغيل) request_time = datetime.fromtimestamp(timestamp) if القيمة المطلقة ((datetime.utcnow() - request_time).total_thans()) > 300: يعود لا أحد # أكثر من 5 دقائق قديمة/مستقبلية # التحقق من توقيع HMAC المتوقع_sig = hmac.new( جلسة["مفتاح_الجلسة"]، request_data + str(timestamp).encode()، hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, المتوقع_sig): يعود حصة["معرف_المحفظة"] يعود لا أحد # تبعية FastAPI للطرق المصادق عليها auth_service = QuantumSafeAuth() تعريف غير متزامن get_authenticated_wallet( x_session_id: str = Header(...)، x_signature: str = Header(...)، x_timestamp: str = Header(...)، الطلب: طلب = لا شيء ) -> str: """التبعية التي تتحقق من صحة المصادقة الكمية الآمنة""" الجسم = انتظر request.body() Wallet_id = auth_service.verify_request( session_id=x_session_id, request_signature=base64.b64decode(x_signature), request_data=body, timestamp=int(x_timestamp)) إذا لم يكن كذلك معرف_المحفظة: يرفع HTTPException(status_code=401,detail=""المصادقة غير صالحة"") يعود Wallet_id

# تنفيذ WebSocket للتحديثات في الوقت الحقيقي من com.fastapi يستورد WebSocket، WebSocketDisconnect يستورد json يستورد غير متزامن فصل مدير الاتصال: """إدارة اتصالات WebSocket لكل محفظة""" مواطنه __الحرف الأولي__(الذات): self.active_connections: dict[str, List[WebSocket]] = {} تعريف غير متزامن يتصل(ذاتي، websocket: WebSocket، Wallet_id: str): انتظر websocket.accept() if Wallet_id ليس في self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) مواطنه قطع الاتصال(ذاتي، websocket: WebSocket، Wallet_id: str): if Wallet_id in self.active_connections: self.active_connections[wallet_id].remove(websocket) تعريف غير متزامن broadcast_to_wallet(ذاتي، Wallet_id: str، message: dict): if Wallet_id in self.active_connections: dead_connections = [] ل اتصال in self.active_connections[wallet_id]: يحاول: انتظر اتصال.send_json (رسالة) يستثني: dead_connections.append (اتصال) # تنظيف الاتصالات الميتة ل كون in dead_connections: self.active_connections[wallet_id].remove(conn) manager = مدير الاتصال() @app.websocket("/ws/{wallet_id}") تعريف غير متزامن websocket_endpoint(websocket: WebSocket، Wallet_id: str): """ WebSocket لتحديثات المحفظة في الوقت الفعلي الأحداث: - تحديث_التوازن: تم تغيير الرصيد - المعاملة_المستلمة: معاملة واردة - تأكيد_المعاملة: تأكيدات وصول الإرسال - المعاملة_المرسلة: بث الإرسال الصادر """ # مصادقة اتصال WebSocket auth_token = websocket.query_params.get("رمز") إذا لم تنتظر validate_ws_token(auth_token، Wallet_id): انتظر websocket.إغلاق (الكود = 4001) يعود انتظر manager.connect(websocket, Wallet_id) يحاول: # إرسال الحالة الأولية انتظر websocket.send_json({ "يكتب": "متصل", "معرف_المحفظة": معرف المحفظة، "الطابع الزمني": datetime.utcnow().isoformat() }) # التعامل مع الرسائل الواردة (الاشتراكات، الأصوات) بينما صحيح: البيانات = انتظر websocket.receive_json() if بيانات.جيت("يكتب") == "بينغ": انتظر websocket.send_json({"يكتب": "بونغ"}) إليف بيانات.جيت("يكتب") == "يشترك": # الاشتراك في عناوين محددة العناوين = data.get("العناوين", []) انتظر اشتراك_خدمة.اشتراك (معرف_المحفظة، العناوين) يستثني WebSocketDisconnect: manager.disconnect(websocket, Wallet_id) # بث الأحداث (يتم استدعاؤه بواسطة مراقب blockchain) تعريف غير متزامن Broadcast_balance_update(wallet_id: str، العنوان: str، new_balance: int): انتظر manager.broadcast_to_wallet(wallet_id, { "يكتب": "تحديث_الرصيد", "عنوان": عنوان، "توازن": توازن جديد، "الطابع الزمني": datetime.utcnow().isoformat() }) تعريف غير متزامن البث_المعاملة_المستلمة(wallet_id: str، tx_summary: dict): انتظر manager.broadcast_to_wallet(wallet_id, { "يكتب": "تم استلام المعاملة", "عملية": tx_summary، # ملخص فقط، وليس التوقيع الكامل "الطابع الزمني": datetime.utcnow().isoformat() })

# تمكين ضغط gzip في FastAPI من fastapi.middleware.gzip يستورد GZipMiddleware app.add_middleware(GZipMiddleware, minor_size=1000) # اختياري: استجابات messagePack لعملاء الهاتف المحمول/المضمنين من fastapi.responses يستورد إجابة يستورد com.msgpack فصل مسجاباكريسبونس(الرد): media_type = "التطبيق/الرسالة" مواطنه يجعل(الذات، المحتوى) -> البايتات: يعود msgpack.packb(المحتوى، use_bin_type=True) @app.get("/API/v1/transactions/{tx_id}/binary") تعريف غير متزامن get_transaction_binary(tx_id: شارع): """الحصول على المعاملة بتنسيق messagePack (أصغر من JSON)""" تكساس = انتظر معاملة_خدمة.get_transaction(tx_id) يعود مسجاباكريسبونس(المحتوى=tx.to_dict())

# تحديد معدل المحفظة API من com.slowapi يستورد المحدد، _rate_limit_exceeded_handler من Slowapi.errors يستورد محدد RateLimitExceeded = المحدد (key_func=get_wallet_id_from_request) app.state.limiter = المحدد app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # حدود مختلفة لعمليات مختلفة RATE_LIMITS = { "يقرأ": "100/دقيقة", # الشيكات التوازن، وقوائم تكساس "يكتب": "20/دقيقة", # اشتقاق العنوان "إذاعة": "5/دقيقة", # بث الصفقة } @app.get("/API/v1/التوازن") @limiter.limit("100/دقيقة") تعريف غير متزامن get_balance(الطلب: الطلب): ... @app.post("/API/v1/المعاملات/البث") @limiter.limit("5/دقيقة") تعريف غير متزامن broadcast_transaction(الطلب: الطلب): """حدود أكثر صرامة للبث لمنع البريد العشوائي""" ...

# استجابات الخطأ الموحدة من التعداد يستورد التعداد فصل رمز الخطأ(شارع، التعداد): INVALID_ADDRESS = "INVALID_ADDRESS" INSUFICIENT_BALANCE = "توازن_غير كاف" INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "تم رفض المعاملة" RATE_LIMITED = "RATE_LIMITED" SESSION_EXPIRED = "SESSION_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" فصل خطأ API(النموذج الأساسي): الكود: رمز الخطأ الرسالة: تفاصيل str: اختياري[dict] = لا شيء @app.exception_handler(خطأ في الرصيد غير كافي) تعريف غير متزامن معالج_التوازن_غير_كافي(طلب، باستثناء): يعود JSONResponse(status_code=400,content=خطأ API(الكود=رمز الخطأ.INSUFFICIENT_BALANCE، الرسالة=""الرصيد غير كافي للمعاملة""التفاصيل={ "متاح": غير متاح، "مطلوب": غير مطلوب، "النقص": exc.required - exc.available } .dict() )