منصة صلاحيات مركزية لخدماتك Microservices

في مقال الصلاحيات المركزية Authorization نقلنا قرار «من يستطيع ماذا» من شروط if متفرقة في الكود إلى محرك واحد، وطبقنا ذلك على تطبيق واحد. ولكن الواقع في أغلب الفرق أن لديك خمس خدمات أو عشراً من نوع Microservices، كتبها أكثر من فريق وبأكثر من لغة، ولكل خدمة قاعدة بياناتها، وكل خدمة تحتاج أن تعرف هل يحق لهذا المستخدم أن يفتح هذا المستند أو يعدل هذا الطلب. وهنا لا يكفي أن تختار محركاً، لأن الأسئلة الصعبة تبدأ بعد الاختيار: من يشغل المحرك، وكيف تصل إليه العلاقات من قواعد بيانات الخدمات المختلفة، وماذا يحدث إذا سحبت صلاحية موظف ولم يصل السحب بعد، وماذا تفعل الخدمة إذا توقف المحرك.

المواضيع في السلسلة:

• الصلاحيات المركزية Authorization: النماذج RBAC وABAC وReBAC، والأدوار PEP وPDP وPAP وPIP، ومقارنة المحركات.
• منصة صلاحيات مركزية لخدماتك (الموضوع الحالي).
• OpenFGA عملياً: تصميم نموذج العلاقات واختباره وتغييره وتشغيله.
• OPA عملياً: سياسات الخصائص ABAC لخدماتك وبواباتك.

ولن نعيد هنا شرح النماذج ولا المصطلحات ولا المقارنة بين المحركات، فهي في المقال الأول، وإنما نبني عليها، وسوف نناقش في هذا المقال ما يلي:

• ما الذي يتغير عندما يصبح المحرك منصة تخدم كل الخدمات، ومن يملك كل جزء فيها.
• أين يسأل النظام عن القرار: البوابة أم الخدمة المجاورة Sidecar أم الخدمة نفسها، وكيف تصل الهوية من authentik إلى القرار.
• كيف تبقى العلاقات في المحرك مطابقة لبيانات كل خدمة، ولماذا تفشل الكتابة المزدوجة Dual Write، وكيف يحلها نمط Outbox.
• الاتساق: القرار القديم بعد سحب الصلاحية، ومشكلة New Enemy بين خدمتين، وما يحله مستوى الاتساق في OpenFGA وما لا يحله.
• القوائم والزمن والتخزين المؤقت Caching، وما يحدث عندما يتوقف المحرك، وسجل القرارات، ومن يدير الصلاحيات.
• ما لا تستطيع المنصة فعله مع التطبيقات الجاهزة التي تستضيفها.
• تجربة عملية كاملة: authentik وبوابة APISIX وخدمتان تسألان OpenFGA، مع قياس الزمن، وإيقاف المحرك، وإظهار مشكلة New Enemy بأعيننا.

ما الذي يتغير عندما يصبح المحرك منصة؟

عندما يكون لديك تطبيق واحد، فالمحرك جزء من هذا التطبيق، ويكتب الفريق نفسه السياسة والكود الذي يسأل عنها ويحدث البيانات التي يحتاجها القرار. أما عندما تخدم نقطة القرار PDP عشر خدمات، فإنها تصبح خدمة مشتركة مثل قاعدة البيانات أو مزود الهوية، ولها مالك وعقد واضح، والصورة التالية تبين المنصة التي سوف نبنيها في التجربة:

في الصورة أعلاه لاحظ أن المسؤوليات تنقسم إلى قسمين واضحين، وهذا التقسيم هو أهم قرار في المنصة كلها:

• فريق المنصة يملك المحرك وتشغيله وتوفره العالي، ويملك ملف نموذج الصلاحيات Model في مستودع Git مع اختباراته، ويملك المكتبة الصغيرة التي تستخدمها الخدمات لتسأل المحرك، أي الدالة الواحدة authorize() التي رأيناها في الانتقال من الشروط المتفرقة، ويملك سجل القرارات.
• كل فريق خدمة يملك بياناته، ويملك العلاقات التي تخص أنواعه هو فقط، فخدمة المستندات وحدها تكتب علاقات document وfolder، وخدمة الفرق وحدها تكتب عضوية team، ولا تكتب أي خدمة علاقة تخص خدمة أخرى، والسبب أن من يكتب العلاقة يجب أن يكون هو من يعرف متى تتغير.
• الواجهة بين الطرفين صغيرة وثابتة: من المستخدم، وما الفعل، وما المورد. وهذا الشكل هو ما وحدته مواصفة AuthZEN Authorization API 1.0 التي ذكرناها في المقال الأول، وفي OpenFGA 1.22 دعم تجريبي لها يفعل بالخيار --experimentals authzen كما يظهر في openfga run --help، لذلك اجعل الدالة في مكتبتك بهذا الشكل حتى لو كان المحرك اليوم لا يتحدث به.

وقد يتساءل البعض: هل نحتاج إلى محرك علاقات مثل OpenFGA أم محرك سياسات مثل OPA؟ والإجابة في جدول المقارنة، والقاعدة المختصرة أن المشاركة والمجلدات والفرق علاقات، وحدود المبالغ والمنطقة ووقت العمل خصائص. وفي هذا المقال نستخدم OpenFGA كنقطة قرار مركزية لأن العلاقات بيانات كبيرة تحفظ في مكان واحد، ونشرح في OPA عملياً كيف تضيف محرك سياسات للقواعد التي تعتمد على الخصائص، وكثير من الأنظمة تحتاج الاثنين معاً.

أين يسأل النظام عن القرار؟

شرحنا في المقال الأول الفرق بين المكتبة والخدمة المركزية وبين الفحص في البوابة والفحص في التطبيق، وفي المنصة تجتمع هذه الأماكن الثلاثة في طلب واحد، والصورة التالية تبين ما يعرفه كل مكان:

والقاعدة العملية: اجعل البوابة ترفض مبكراً ما لا يحتاج إلى بيانات، مثل التوكن المزور والمسار الإداري من الإنترنت، فهي توفر على خدماتك طلبات لا فائدة منها، واجعل كل خدمة تسأل عن كل مورد بعينه، ولا تجعل الخدمة تثق بأن البوابة قررت عنها، والسبب أن البوابة لا تعرف أن المستند 42 لمؤسسة أخرى. وقد شرحنا البوابات نفسها في ما هي بوابة API Gateway وفي Apache APISIX عملياً.

الهوية من authentik، والقرار من المنصة

المنصة لا تتحقق من كلمات المرور، فهذا عمل مزود الهوية، وفي تجربتنا هو authentik الذي يصدر توكن JWT موقعاً كما شرحنا في OAuth وOpenID Connect. والمطلوب من التوكن شيئان فقط: معرف ثابت للمستخدم، والمجموعات العامة. وهذا هو محتوى التوكن الذي حصلت عليه alice في التجربة:

{
 "iss": "http://lab-authz-authentik:9000/application/o/platform/",
 "sub": "50907a776f61cb2a6651eda4e4fd991b112333e65976d02ad54db7af43bdc280",
 "aud": "platform",
 "name": "Alice",
 "preferred_username": "alice",
 "groups": [
  "team-admins"
 ],
 "azp": "platform",
 "scope": "openid profile"
}

في المخرج أعلاه لاحظ التالي:

• الحقل sub هاش ثابت لا يتغير إذا غيرت اسم المستخدم، أما preferred_username فيمكن أن يتغير. وفي التجربة استخدمنا الاسم حتى تكون المخرجات مقروءة، وفي الإنتاج اجعل معرف المستخدم في العلاقات هو sub أو معرف ثابت من نظامك، والسبب أن تغيير الاسم سوف يجعل كل علاقات المستخدم القديمة تشير إلى شخص غير موجود، أو أسوأ من ذلك إلى شخص جديد أخذ الاسم نفسه.
• المجموعة team-admins دور عام نستخدمه في خدمة الفرق لفحص عام: من يحق له تعديل عضوية الفرق أصلاً. وقد شرحنا في إدارة الأدوار عبر تطبيقاتك كيف تصل المجموعات إلى التوكن وكيف تسمى.
• لا توجد في التوكن أي صلاحية على مستند، وهذا مقصود كما شرحنا في أين تقع ال Scopes داخل التوكن.

والبوابة APISIX تتحقق من توقيع التوكن بمفاتيح authentik العامة عبر إضافة openid-connect، ثم تمرره إلى الخدمة كما هو، والخدمة في التجربة تقرأ الحقول دون أن تتحقق من التوقيع مرة ثانية لأن منفذها غير منشور خارج شبكة Docker. وهذا مقبول في التجربة فقط، وفي الإنتاج إما أن تتحقق كل خدمة من التوكن بنفسها، وهو رخيص لأن المفاتيح العامة تحفظ في الذاكرة، وإما أن تضمن بشبكة معزولة وشهادات mTLS أن أحداً لا يستطيع الوصول إلى الخدمة إلا عبر البوابة، والسبب أن خدمة تثق بأي ترويسة Header تصلها يكفي لاختراقها أن يصل إليها طلب واحد من داخل الشبكة.

كيف تصل علاقاتك إلى المحرك؟

هذه هي المشكلة التي لا تراها في تجربة على تطبيق واحد. محرك العلاقات يحتاج أن يعرف أن المستند reorg-plan داخل المجلد finance-2026، وأن bob عضو في فريق المالية، ولكن هذه الحقائق تولد في قواعد بيانات الخدمات، فخدمة المستندات تحفظ المستند في جدولها، وخدمة الفرق تحفظ العضوية في جدولها، ويجب أن تصل نسخة منها إلى OpenFGA على شكل علاقات Tuples.

الحل البسيط أن تكتب الخدمة في قاعدة بياناتها ثم تكتب العلاقة في OpenFGA مباشرة، وهذه هي الكتابة المزدوجة Dual Write التي شرحنا مشكلتها مع الرسائل في المعاملات الموزعة ونمط Outbox: لا توجد معاملة واحدة تجمع قاعدة البيانات والمحرك، فإذا نجحت الأولى وفشلت الثانية بقيت البيانات بلا علاقات. والصورة التالية تبين الطريقتين كما ظهرتا في التجربة:

وفي المنصة يأخذ نمط Outbox الشكل التالي:

• كل خدمة تكتب بياناتها وصفاً في جدول outbox لديها في المعاملة نفسها، والصف يحمل العملية write أو delete والعلاقة نفسها.
• برنامج صغير اسمه Relay يقرأ الصفوف غير المرسلة بالترتيب، ويكتبها في OpenFGA، ثم يعلم الصف كمرسل. وكتابة العلاقة في OpenFGA مع الخيار on_duplicate: "ignore" والحذف مع on_missing: "ignore" تجعل الإعادة آمنة، فإذا توقف ال Relay بعد الكتابة وقبل تعليم الصف وأعاد إرسال العلاقة نفسها فلن يحدث خطأ، وهذا تطبيق مباشر لفكرة Idempotency.
• بدل السؤال الدوري تستطيع أن تقرأ سجل معاملات PostgreSQL مباشرة بأداة CDC مثل Debezium، وقد شرحنا الطريقتين وثمن كل منهما في كيف تخرج الأحداث من جدول Outbox، وإذا كانت لديك أحداث في Kafka أصلاً، كما في Event Sourcing وCQRS، فالمحرك يصبح مجرد مستهلك آخر يبني منها علاقاته.
• ومع ذلك سوف يحدث اختلاف بين البيانات والعلاقات يوماً ما، بسبب خطأ في كود قديم أو استعادة نسخة احتياطية، لذلك شغل مهمة مطابقة Reconciliation دورية تقارن ما في جداول الخدمة بما في المحرك وتصلح الفرق، وتستطيع أن تقرأ تغييرات العلاقات بترتيبها من واجهة ReadChanges في OpenFGA.

وثمن ال Outbox أن العلاقة تتأخر قليلاً عن البيانات، وفي التجربة كان التأخير أقل من ثانية، ولكن هذا التأخير هو بالضبط مصدر المشكلة التالية.

الاتساق: القرار القديم ومشكلة New Enemy

شرحنا في المقال الأول مشكلة New Enemy كما وصفتها ورقة Zanzibar، وحلها برمز Zookie الذي يجعل المحرك يقيم الفحص على بيانات لا تقل حداثة عن تعديل معين. وفي المنصة توجد ثلاثة مصادر للقرار القديم، ولكل منها حل مختلف:

• ال Cache داخل المحرك: يفعل في OpenFGA بالخيار check-query-cache-enabled، ويقول وصفه في openfga run --help إنه يجعل Check وListObjects واجهات متسقة في النهاية Eventually Consistent. وحله أن تطلب HIGHER_CONSISTENCY في الطلبات الحساسة، وهو يتجاوز ال Cache ويقرأ من قاعدة البيانات مباشرة كما في توثيق الاتساق.
• ال Cache داخل الخدمة: إذا حفظت الخدمة القرار خمس ثوان، فلن يفيدها أي مستوى اتساق في المحرك لأنها لا تسأله أصلاً، وحله أن تكون المدة قصيرة وأن تتجاوز الخدمة ال Cache في العمليات الحساسة.
• تأخر ال Relay: إذا لم يصل حذف العضوية إلى المحرك بعد، فالمحرك لا يعرف عنه شيئاً، ولا يغير HIGHER_CONSISTENCY من ذلك شيئاً، لأن الحقيقة الجديدة ما زالت في جدول ال Outbox.

ويجدر الإشارة هنا إلى أن OpenFGA لا يدعم اليوم رموز الاتساق مثل Zookie، ويذكر توثيقه أن الفريق يدرس إضافة ميزة مشابهة، أما SpiceDB فيطبقها باسم ZedToken كما في توثيقه. والأخطر من ال Cache هو الحالة الثالثة عندما تجتمع مع خدمتين لكل منهما Relay خاص بها، والصورة التالية تبين ما حدث في التجربة خطوة خطوة:

وهذا هو شكل مشكلة New Enemy في الخدمات المصغرة: لا يوجد ترتيب بين ما تكتبه خدمتان مختلفتان، فالمحتوى الجديد وصل إلى المحرك قبل سحب الصلاحية الذي حدث قبله في الواقع، فرأى bob مستنداً كتب بعد أن أخرج من الفريق. والحلول الممكنة مرتبة من الأبسط:

• اسحب الصلاحية بشكل متزامن: عندما تحذف الخدمة عضوية أو مشاركة، فإنها تحذف العلاقة من المحرك قبل أن ترد على المستخدم، وتبقي صف ال Outbox كاحتياط، وإذا لم يرد المحرك فإنها ترد بخطأ بدل أن تقول «تم». وهذا ما فعلناه في التجربة بالمتغير SYNC_REVOKE=1، والإضافة يمكن أن تتأخر قليلاً أما السحب فلا، والسبب أن تأخر الإضافة يزعج مستخدماً ينتظر، أما تأخر السحب فيكشف بيانات.
• اجعل التغييرات التي تحتاج ترتيباً في مجرى واحد مرتب، مثل موضوع Topic واحد في Kafka مقسم حسب المورد، فيطبقها مستهلك واحد بترتيبها.
• اختر محركاً يدعم رموز الاتساق، مثل SpiceDB، إذا كانت هذه الحالة متكررة في نظامك، واحفظ الرمز مع المحتوى كما تصف ورقة Zanzibar.

وهناك وجه آخر للتأخير يظهر في الاتجاه المعاكس، وهو أن تنشئ alice مجلداً ثم تشاركه مباشرة فترفض الخدمة لأن علاقة الملكية لم تصل بعد، وقد حدث هذا في أول محاولة لنا في التجربة. وحله أن تقرر الخدمة على ما أنشأته للتو من بياناتها هي، أو أن تكتب علاقة المالك بشكل متزامن عند الإنشاء، وهذا ما يسمى قراءة ما كتبته Read Your Writes.

القوائم: ListObjects أم الترشيح بعد الجلب؟

صفحة «مستنداتي» لا تسأل عن مستند واحد وإنما عن كل ما يستطيع المستخدم رؤيته، وقد ذكرنا في المقال الأول أن محركات العلاقات تجيب عن هذا بأمر مثل ListObjects. وفي التجربة وضعنا 5,056 مستنداً، منها 55 يستطيع bob رؤيتها، وقسنا ثلاث طرق، والصورة التالية تبين النتيجة:

في الصورة أعلاه لاحظ التالي:

• الترشيح بعد الجلب Post-filtering، أي أن تجلب كل الصفوف ثم تسأل عن كل صف، لا يصلح إطلاقاً لقائمة كبيرة، وقد أخذ 35 ثانية، وتجاوزت محاولتان مهلة البوابة، والسبب أن كل صف طلب عبر الشبكة.
• BatchCheck، المتاح منذ OpenFGA 1.8.0 كما يذكر SDK بلغة Python، يجمع خمسين فحصاً في طلب، وهو الحد الافتراضي maxChecksPerBatchCheck، فنزل الزمن إلى أقل من ثانيتين، وهو مناسب عندما تكون القائمة صغيرة أصلاً بعد ترشيح آخر، مثل نتائج بحث من عشرين عنصراً.
• ListObjects يعطي المعرفات مباشرة، ثم تجلبها الخدمة باستعلام واحد، وهو الأسرع هنا بفارق كبير. ولكن له حدان افتراضيان يظهران في openfga run --help: ألف نتيجة في الرد العادي (listObjects-max-results) وثلاث ثوان كحد أقصى (listObjects-deadline). وفي التجربة طلبت alice قائمتها وهي تملك 5,055 مستنداً، فكان الرد 1,000 فقط دون أي خطأ، فإذا لم تنتبه لهذا الحد فسوف يختفي جزء من القائمة بصمت. ولهذا يوجد StreamedListObjects للنتائج الكبيرة، أو تقلب السؤال فتسأل عن المجلدات التي يراها المستخدم ثم تجلب مستنداتها من قاعدة البيانات.

ومع محركات السياسات مثل OPA لا يوجد ListObjects لأن المحرك لا يحفظ البيانات، والحل هناك أن يحول المحرك السياسة إلى شرط SQL تضيفه الخدمة إلى استعلامها، وقد طبقنا ذلك في OPA عملياً.

الزمن والتخزين المؤقت Caching

كل طلب يمر الآن بقرار عبر الشبكة، لذلك قسنا في التجربة زمن قرار واحد في كل طبقة، ألفي طلب أو ثلاثة آلاف على اتصال مفتوح Keep-alive، والصورة التالية تبين الوسيط p50 ومدة بقاء القرار القديم بعد سحب الصلاحية:

وهذه الأرقام تقول ثلاثة أشياء:

• القرار نفسه رخيص: فحص يمشي ثلاث علاقات، من المستند إلى المجلد إلى الفريق، أخذ 2.35 ملي ثانية في الوسيط على PostgreSQL، ومعظم زمن الطلب في الخدمة كان في الخدمة نفسها، أي في Python واستعلام قاعدة البيانات وكتابة سطر السجل.
• ال Cache يقلل الزمن، وكل طبقة منه تضيف مدة يبقى فيها القرار القديم، فمع ال Cache في المحرك (عشر ثوان) وفي الخدمة (خمس ثوان) بقي bob يقرأ المستند 9.1 ثانية بعد حذفه من الفريق، ومع HIGHER_CONSISTENCY دون Cache في الخدمة رفض بعد 0.6 ثانية، وهي مدة ال Relay فقط.
• الطلب الأول في القياس كان سيئاً جداً، فقد كان الوسيط في الخدمة 100 ملي ثانية، والسبب لم يكن المحرك وإنما أن الخدمة كانت تفتح اتصالاً جديداً بقاعدة البيانات في كل طلب، ثم بعد إصلاح ذلك بقي 60 ملي ثانية لأن خادم HTTP في Python كان يرسل الترويسات والمحتوى في دفعتين، فينتظر الطرف الآخر تأكيد TCP المؤجل Delayed ACK، وعندما جمعنا الإرسال في دفعة واحدة نزل الزمن إلى 7 ملي ثوان. لذلك قس قبل أن تلوم المحرك أو تضيف Cache.

وقد شرحنا أنواع ال Cache وطرق إبطاله في التخزين المؤقت Caching، والقاعدة هنا أن مدة ال Cache هي المدة التي تقبل فيها أن يبقى شخص سحبت صلاحيته قادراً على الوصول، فاختر الرقم بهذا المعنى لا بمعنى الأداء فقط.

عندما يتوقف المحرك: الرفض هو الافتراضي

المنصة المركزية نقطة فشل مشتركة، فإذا توقف المحرك توقفت القرارات في كل الخدمات. وفي التجربة أوقفنا OpenFGA وطلبت alice مستندها، فرفضت الخدمة الطلب، وهذا هو السلوك الصحيح كما ذكرنا في إرشادات التصميم، ثم غيرنا متغيراً واحداً في الخدمة يجعلها تسمح عند الخطأ Fail Open، فأصبحت carol، التي لا تملك أي صلاحية على المستند، تقرؤه. وما نضيفه هنا من التجربة ثلاث ملاحظات:

• الرد برمز 403 عند تعطل المحرك يخبر المستخدم أنه لا يملك الصلاحية، وهذا غير صحيح، والأفضل أن ترد الخدمة ب 503 مع رسالة أن الخدمة غير متاحة مؤقتاً، حتى لا يفتح المستخدم تذكرة دعم يطلب فيها صلاحية يملكها أصلاً.
• الفشل كان بطيئاً: كل طلب أخذ نحو خمس ثوان قبل أن يرفض، لأن اسم المحرك لم يعد موجوداً في DNS الشبكة. لذلك اجعل مهلة الاتصال بالمحرك قصيرة، واستخدم قاطع الدائرة Circuit Breaker كما في أنماط المرونة، حتى لا يتحول تعطل المحرك إلى تعطل كل خيوط الخدمة.
• OpenFGA نفسه لا يحفظ حالة دائمة خارج قاعدة البيانات، فتستطيع أن تشغل منه عدة نسخ خلف موزع حمل، والحالة كلها في PostgreSQL، لذلك يصبح السؤال الحقيقي هو التوفر العالي لقاعدة البيانات ونسخها الاحتياطي، وفي OpenFGA 1.22 أضيف دعم نسخة قراءة Read Replica لقاعدة MySQL كما هو موجود في PostgreSQL، والقراءات التي تطلب HIGHER_CONSISTENCY تذهب إلى النسخة الرئيسية كما في سجل التغييرات.

سجل القرارات والتدقيق

عندما يسأل المدقق «من فتح هذا المستند، ومن كان يستطيع فتحه يوم 12 أكتوبر؟» تحتاج إلى سجلين مختلفين. الأول سجل القرارات، وفي التجربة تكتب الدالة authorize() سطراً لكل قرار، وهذا أحد الأسطر الحقيقية:

{"ts": "12:20:40", "svc": "docs", "user": "user:alice", "relation": "viewer", "object": "document:budget-q3", "allowed": false, "source": "error: gaierror", "consistency": "MINIMIZE_LATENCY", "ms": 5235.09}

لاحظ أن السطر يقول لماذا رفض الطلب، فالمصدر error وليس pdp، أي أن الرفض كان بسبب تعطل المحرك وليس بسبب الصلاحية، وبدون هذا الحقل سوف تبحث في السياسة عن خطأ غير موجود. والسجل الثاني هو سجل تغييرات العلاقات، أي من أضاف bob إلى الفريق ومتى، وتجده في جداول ال Outbox لدى كل خدمة وفي واجهة ReadChanges في OpenFGA. وقد شرحنا ما يجب أن يحتويه سجل التدقيق وكيف تحميه من التعديل في سجل التدقيق Audit Trail.

ويجدر الإشارة هنا إلى أن OpenFGA في إعدادنا بسجلات JSON يكتب لكل طلب الطلب كاملاً بما فيه حقل context كما رأينا في OpenFGA عملياً، فإذا وضعت في السياق بيانات شخصية فإنها تصل إلى السجلات، وعندها تنطبق عليها مدة الاحتفاظ وصلاحيات القراءة نفسها.

النموذج ككود، ومن يدير الصلاحيات؟

نموذج الصلاحيات في المنصة يؤثر على كل الخدمات، لذلك يعامل مثل مخطط قاعدة البيانات المشتركة: ملف في Git، وكل تعديل بطلب دمج، واختبارات تعمل في CI قبل النشر، ومراجعة من شخص ثان، ومعرف النموذج مثبت في كل خدمة ولا يتغير إلا بقرار. وقد بينا بالتجربة في OpenFGA عملياً ماذا يحدث عندما ينشر نموذج جديد وتستخدمه الخدمات تلقائياً، وكيف فتح حذف علاقة واحدة مستنداً كان محجوباً.

وأما نقطة الإدارة PAP فهي في هذه المنصة شيئان مختلفان:

• إدارة القواعد، أي النموذج والسياسات، وهي لفريق المنصة عبر Git فقط، ولا يوجد في OpenFGA واجهة إدارة رسومية للإنتاج، وواجهة Playground المدمجة أصبحت معطلة افتراضياً ومعلنة كميزة ستزال منذ الإصدار 1.14.0 الصادر في 3 أبريل 2026 بعد ثغرة أمنية، كما في سجل التغييرات.
• إدارة العلاقات، أي من يشارك المجلد مع من، وهي للمستخدمين أنفسهم عبر واجهات خدماتك، مثل زر «مشاركة» في خدمة المستندات، والخدمة هي التي تكتب العلاقة بعد أن تفحص أن المستخدم يملك حق المشاركة. ولا تعط أي شخص أداة سطر الأوامر على المحرك في الإنتاج، لأن علاقة واحدة مكتوبة يدوياً لا تمر بأي خدمة ولا تترك أثراً في أي Outbox.

التطبيقات الجاهزة لا تسأل منصتك

المنصة تخدم الخدمات التي تكتبها أنت، لأنك تستطيع أن تضع في كودها استدعاء authorize(). أما التطبيقات الجاهزة التي تستضيفها، مثل Gitea وGrafana وNextcloud، فلكل منها نظام أدوار خاص به داخل قاعدة بياناته، ولا يوجد فيها مكان تقول فيه «اسأل OpenFGA قبل أن تفتح هذا المستودع». وما تستطيع فعله معها هو ما شرحناه في إدارة الأدوار عبر تطبيقاتك من authentik: الدخول عبر authentik، والدور العام من مجموعة في التوكن، وإغلاق الحساب عند المغادرة، والصلاحيات الدقيقة داخل كل تطبيق تبقى في التطبيق نفسه. فإذا وعدت الإدارة بأن «كل الصلاحيات في مكان واحد» فكن دقيقاً: صلاحيات خدماتك نعم، وصلاحيات التطبيقات الجاهزة تدار بمجموعات مركزية وتراجع في كل تطبيق.

تجربة عملية: منصة صغيرة كاملة

نبني في هذه التجربة المنصة التي في الصورة الأولى: authentik يصدر التوكن، وAPISIX في وضع Standalone تتحقق منه، وخدمتان بلغة Python، خدمة المستندات وخدمة الفرق، لكل منهما قاعدة بيانات وجدول Outbox وبرنامج Relay، وOpenFGA على PostgreSQL كنقطة قرار مركزية. نفذنا التجربة على جهاز فيه Docker Engine 29.8.0 وDocker Compose 5.5.1، وكل ما ننشئه يبدأ اسمه بالبادئة lab-authz، والمنافذ المنشورة مربوطة على 127.0.0.1 فقط، والإصدارات كما يلي: OpenFGA 1.22.0 وfga CLI 0.8.1 وauthentik 2026.8.3 وAPISIX 3.19.0 وPostgreSQL 18.6 وPython 3.13.16 مع psycopg 3.3.6.

ملفات التجربة

ننشئ مجلداً للتجربة فيه المجلدات svc وapisix وstate، والأخير تكتب فيه السكربتات معرف المخزن وكلمات مرور التطبيقات:

mkdir -p ~/authz-platform/{svc,apisix,state} && cd ~/authz-platform

أولاً ملف .env، وكل القيم فيه للتجربة فقط:

# Lab-only secrets. On a real server generate your own: openssl rand -hex 32
PG_PASS=lab-authz-pg-4c1e9b
AUTHENTIK_SECRET_KEY=lab-authz-authentik-secret-3f8a1c5e7b9d2f4a6c8e0b1d3f5a7c9e
AUTHENTIK_BOOTSTRAP_PASSWORD=Lab-Authz-Admin-2026
AUTHENTIK_BOOTSTRAP_TOKEN=lab-authz-bootstrap-8e2d4f6a1c
# OpenFGA check cache (off by default; the lab turns it on to measure it)
FGA_CACHE=false
# Service behaviour switches used in the break-it steps
FAIL_OPEN=0
LOCAL_CACHE_TTL=0

ثم initdb.sql الذي ينشئ قاعدة بيانات لكل مالك داخل PostgreSQL واحد، وDockerfile الذي تبنى منه الخدمات:

-- One PostgreSQL for the lab, one database per owner: OpenFGA, authentik, and each service.
CREATE DATABASE openfga;
CREATE DATABASE authentik;
CREATE DATABASE docs;
CREATE DATABASE teams;

FROM python:3.13.16-slim
RUN pip install --no-cache-dir "psycopg[binary]==3.3.6"
WORKDIR /app
COPY svc/ /app/

وهذا ملف compose.yaml الكامل:

# Central authorization lab: authentik issues tokens, APISIX checks them, two services ask OpenFGA,
# and each service has its own outbox and relay that keep OpenFGA in sync with its data.
name: lab-authz
services:
  pg:
    image: postgres:18.6
    container_name: lab-authz-pg
    environment: { POSTGRES_USER: lab, POSTGRES_PASSWORD: "${PG_PASS}" }
    volumes:
      - pg-data:/var/lib/postgresql
      - ./initdb.sql:/docker-entrypoint-initdb.d/initdb.sql:ro
    healthcheck: { test: ["CMD", "pg_isready", "-U", "lab"], interval: 3s, retries: 20 }

  openfga-migrate:
    image: openfga/openfga:v1.22.0
    container_name: lab-authz-openfga-migrate
    command: migrate
    environment: &fga
      OPENFGA_DATASTORE_ENGINE: postgres
      OPENFGA_DATASTORE_URI: "postgres://lab:${PG_PASS}@lab-authz-pg:5432/openfga?sslmode=disable"
    depends_on: { pg: { condition: service_healthy } }

  openfga:
    image: openfga/openfga:v1.22.0
    container_name: lab-authz-openfga
    command: run
    environment:
      <<: *fga
      OPENFGA_CHECK_QUERY_CACHE_ENABLED: "${FGA_CACHE}"
      OPENFGA_CHECK_QUERY_CACHE_TTL: 10s
      OPENFGA_LOG_FORMAT: json
    ports:
      - "127.0.0.1:18080:8080"   # HTTP API
      - "127.0.0.1:18112:2112"   # Prometheus metrics
    depends_on: { openfga-migrate: { condition: service_completed_successfully } }

  authentik:
    image: ghcr.io/goauthentik/server:2026.8.3
    container_name: lab-authz-authentik
    command: server
    environment: &ak
      AUTHENTIK_POSTGRESQL__HOST: lab-authz-pg
      AUTHENTIK_POSTGRESQL__NAME: authentik
      AUTHENTIK_POSTGRESQL__USER: lab
      AUTHENTIK_POSTGRESQL__PASSWORD: "${PG_PASS}"
      AUTHENTIK_SECRET_KEY: "${AUTHENTIK_SECRET_KEY}"
      AUTHENTIK_BOOTSTRAP_PASSWORD: "${AUTHENTIK_BOOTSTRAP_PASSWORD}"
      AUTHENTIK_BOOTSTRAP_TOKEN: "${AUTHENTIK_BOOTSTRAP_TOKEN}"
      AUTHENTIK_ERROR_REPORTING__ENABLED: "false"
    ports: ["127.0.0.1:19295:9000"]
    depends_on: { pg: { condition: service_healthy } }
  authentik-worker:
    image: ghcr.io/goauthentik/server:2026.8.3
    container_name: lab-authz-authentik-worker
    command: worker
    environment: *ak
    depends_on: { pg: { condition: service_healthy } }

  apisix:
    image: apache/apisix:3.19.0-debian
    container_name: lab-authz-apisix
    volumes:
      - ./apisix/config.yaml:/usr/local/apisix/conf/config.yaml:ro
      - ./apisix/apisix.yaml:/usr/local/apisix/conf/apisix.yaml:ro
    ports: ["127.0.0.1:19280:9080"]

  docs:
    build: .
    image: lab-authz-py:1
    container_name: lab-authz-docs
    command: python docs.py
    environment: &svc
      PG_DSN: "postgresql://lab:${PG_PASS}@lab-authz-pg:5432/docs"
      FGA_URL: http://lab-authz-openfga:8080
      FAIL_OPEN: "${FAIL_OPEN}"
      LOCAL_CACHE_TTL: "${LOCAL_CACHE_TTL}"
      DUAL_WRITE: "${DUAL_WRITE:-0}"
    volumes: ["./svc:/app:ro", "./state:/state"]
    depends_on: { pg: { condition: service_healthy } }
  teams:
    image: lab-authz-py:1
    container_name: lab-authz-teams
    command: python teams.py
    environment:
      <<: *svc
      PG_DSN: "postgresql://lab:${PG_PASS}@lab-authz-pg:5432/teams"
      SYNC_REVOKE: "${SYNC_REVOKE:-0}"
    volumes: ["./svc:/app:ro", "./state:/state"]
    depends_on: [docs]
  relay-docs:
    image: lab-authz-py:1
    container_name: lab-authz-relay-docs
    command: python relay.py
    environment: { <<: *svc }
    volumes: ["./svc:/app:ro", "./state:/state"]
    depends_on: [docs]
  relay-teams:
    image: lab-authz-py:1
    container_name: lab-authz-relay-teams
    command: python relay.py
    environment:
      <<: *svc
      PG_DSN: "postgresql://lab:${PG_PASS}@lab-authz-pg:5432/teams"
    volumes: ["./svc:/app:ro", "./state:/state"]
    depends_on: [teams]

  client:
    image: lab-authz-py:1
    container_name: lab-authz-client
    command: sleep infinity
    volumes: ["./svc:/app:ro", "./state:/state"]

volumes:
  pg-data:
    name: lab-authz-pg-data
networks:
  default:
    name: lab-authz-net

في الإعداد أعلاه لاحظ التالي:

• الخدمة openfga-migrate تنشئ جداول OpenFGA في PostgreSQL ثم تخرج، والمحرك لا يبدأ إلا بعد نجاحها، وهذا هو الأمر نفسه الذي تشغله عند كل ترقية كما في OpenFGA عملياً.
• المتغير OPENFGA_CHECK_QUERY_CACHE_ENABLED يأتي من FGA_CACHE حتى نقيس الزمن مع ال Cache ودونه، ومدته عشر ثوان.
• الخدمات لا تنشر أي منفذ، فلا يصل إليها أحد إلا عبر APISIX من داخل الشبكة، والعميل client هو Container ننفذ منه الطلبات حتى يكون اسم مزود الهوية في التوكن هو نفسه الذي تعرفه البوابة، وقد شرحنا سبب ذلك في دليل APISIX.
• المتغيرات FAIL_OPEN وLOCAL_CACHE_TTL وDUAL_WRITE وSYNC_REVOKE مفاتيح للتجارب التي نكسر فيها المنصة، وقيمها الافتراضية هي الآمنة.

البوابة في وضع Standalone تقرأ إعدادها من ملفين، الأول apisix/config.yaml:

# APISIX 3.19 in standalone mode: routes come from apisix.yaml, no etcd and no Admin API.
apisix:
  node_listen: 9080
  dns_resolver_valid: 5     # re-resolve service names every 5 s, so a recreated container is found
deployment:
  role: data_plane
  role_data_plane:
    config_provider: yaml
plugins:
  - request-id
  - openid-connect
  - proxy-rewrite

والثاني apisix/apisix.yaml الذي فيه المسارات، وكل مسار يشترط توكناً صالحاً من authentik:

# The gateway only checks that the token is valid and comes from authentik (coarse check).
# The decision on each document stays in the services, which ask OpenFGA.
upstreams:
  - id: docs
    nodes: { "lab-authz-docs:8000": 1 }
  - id: teams
    nodes: { "lab-authz-teams:8000": 1 }
routes:
  - id: docs
    uris: ["/api/docs", "/api/docs/*", "/api/folders", "/api/folders/*"]
    upstream_id: docs
    plugins:
      openid-connect:
        client_id: platform
        client_secret: lab-authz-platform-secret-6d2b
        discovery: http://lab-authz-authentik:9000/application/o/platform/.well-known/openid-configuration
        bearer_only: true
        use_jwks: true
        ssl_verify: false
        claim_validator: { audience: { required: true, match_with_client_id: true } }
      proxy-rewrite: { regex_uri: ["^/api/(.*)", "/$1"] }
  - id: teams
    uris: ["/api/teams/*"]
    upstream_id: teams
    plugins:
      openid-connect:
        client_id: platform
        client_secret: lab-authz-platform-secret-6d2b
        discovery: http://lab-authz-authentik:9000/application/o/platform/.well-known/openid-configuration
        bearer_only: true
        use_jwks: true
        ssl_verify: false
        claim_validator: { audience: { required: true, match_with_client_id: true } }
      proxy-rewrite: { regex_uri: ["^/api/(.*)", "/$1"] }
#END

والخيار dns_resolver_valid مهم في هذه التجربة، فبدونه حصلنا على 502 بعد أن أعدنا إنشاء خدمة المستندات، لأن APISIX بقيت تستخدم العنوان القديم للخدمة حتى أعدنا تشغيلها.

ونموذج الصلاحيات svc/model.fga، وهو أبسط من نموذج المقال الأول لأن التركيز هنا على المنصة لا على النموذج:

model
  schema 1.1

type user

type team
  relations
    define member: [user]

type folder
  relations
    define owner: [user]
    define viewer: [user, team#member] or owner

type document
  relations
    define parent: [folder]
    define owner: [user]
    define viewer: [user] or owner or viewer from parent

ثم الكود المشترك svc/common.py، وأهم ما فيه الدالة authorize()، فهي المكان الوحيد الذي تقرر فيه أي خدمة، وهي التي ترفض افتراضياً وتكتب سطر سجل لكل قرار:

# Shared helpers for the lab services: OpenFGA calls, the authorize() function every service uses,
# the identity from the token the gateway already checked, and a small JSON HTTP server.
import base64, http.client, json, os, threading, time, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

FGA = urllib.parse.urlparse(os.environ.get("FGA_URL", "http://lab-authz-openfga:8080"))
FAIL_OPEN = os.environ.get("FAIL_OPEN", "0") == "1"           # break-it switch: never do this in production
LOCAL_CACHE_TTL = float(os.environ.get("LOCAL_CACHE_TTL", "0"))  # seconds; 0 = no cache in the service
SERVICE = os.environ.get("SERVICE", "svc")
_local = threading.local()


def fga(method, path, body=None):
    """One HTTP call to OpenFGA over a kept-alive connection per thread."""
    for attempt in (1, 2):
        conn = getattr(_local, "conn", None)
        if conn is None:
            conn = _local.conn = http.client.HTTPConnection(FGA.hostname, FGA.port, timeout=2)
        try:
            conn.request(method, path, body=json.dumps(body) if body is not None else None,
                         headers={"Content-Type": "application/json"})
            resp = conn.getresponse()
            data = resp.read()
            return resp.status, json.loads(data or b"{}")
        except (OSError, http.client.HTTPException):
            conn.close()
            _local.conn = None
            if attempt == 2:
                raise


def store():
    with open("/state/fga.json") as f:
        return json.load(f)


_cache, _cache_lock = {}, threading.Lock()


def authorize(user, relation, obj, consistency="MINIMIZE_LATENCY"):
    """The only place a service decides. Deny by default, log every decision."""
    key, t0, source = (user, relation, obj), time.perf_counter(), "pdp"
    allowed = None
    if LOCAL_CACHE_TTL and consistency != "HIGHER_CONSISTENCY":
        with _cache_lock:
            hit = _cache.get(key)
        if hit and hit[1] > time.monotonic():
            allowed, source = hit[0], "local-cache"
    if allowed is None:
        s = store()
        try:
            status, r = fga("POST", f"/stores/{s['store_id']}/check", {
                "tuple_key": {"user": user, "relation": relation, "object": obj},
                "authorization_model_id": s["model_id"], "consistency": consistency})
            if status != 200:
                raise RuntimeError(f"openfga {status}: {r.get('message')}")
            allowed = r.get("allowed") is True
            if LOCAL_CACHE_TTL:
                with _cache_lock:
                    _cache[key] = (allowed, time.monotonic() + LOCAL_CACHE_TTL)
        except Exception as e:
            allowed, source = FAIL_OPEN, f"error: {e.__class__.__name__}"
    print(json.dumps({"ts": time.strftime("%H:%M:%S"), "svc": SERVICE, "user": user, "relation": relation,
                      "object": obj, "allowed": allowed, "source": source, "consistency": consistency,
                      "ms": round((time.perf_counter() - t0) * 1000, 2)}), flush=True)
    return allowed


def identity(headers):
    """The gateway has verified the token; the service only reads who the user is and the groups."""
    auth = headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return None, []
    payload = auth.split(".")[1]
    claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
    return "user:" + claims["preferred_username"], claims.get("groups", [])


class Handler(BaseHTTPRequestHandler):
    routes = []          # (method, path prefix, function(handler, rest, user, groups, body))
    protocol_version = "HTTP/1.1"
    disable_nagle_algorithm = True   # answer in one packet (no 40 ms delayed-ACK wait on keep-alive)
    wbufsize = 65536                 # headers and body leave together; flushed after each request

    def log_message(self, *a):
        pass

    def reply(self, code, obj):
        data = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def handle_any(self, method):
        n = int(self.headers.get("Content-Length") or 0)
        body = json.loads(self.rfile.read(n) or b"{}") if n else {}
        user, groups = identity(self.headers)
        if user is None:
            return self.reply(401, {"error": "no token"})
        path = urllib.parse.urlparse(self.path)
        self.query = dict(urllib.parse.parse_qsl(path.query))
        for m, prefix, fn in self.routes:
            if m == method and (path.path == prefix or path.path.startswith(prefix + "/")):
                return fn(self, path.path[len(prefix) + 1:], user, groups, body)
        self.reply(404, {"error": "not found"})

    def do_GET(self): self.handle_any("GET")
    def do_POST(self): self.handle_any("POST")
    def do_PUT(self): self.handle_any("PUT")
    def do_DELETE(self): self.handle_any("DELETE")


def serve(routes):
    Handler.routes = routes
    ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()

وخدمة المستندات svc/docs.py، وكل تغيير في الملكية أو المكان يكتب في ال Outbox في المعاملة نفسها التي تكتب البيانات:

# Documents service: owns folders and documents. Every change to who-owns-what is written to its
# outbox table in the same transaction as the data; relay.py moves it to OpenFGA.
import json, os, threading, time
import psycopg
import common
from common import authorize, fga, serve, store

common.SERVICE = "docs"
DUAL_WRITE = os.environ.get("DUAL_WRITE", "0") == "1"   # break-it switch: write OpenFGA directly, no outbox
DSN = os.environ["PG_DSN"]

with psycopg.connect(DSN, autocommit=True) as c:
    c.execute("""
      CREATE TABLE IF NOT EXISTS folders (id text PRIMARY KEY, owner text NOT NULL);
      CREATE TABLE IF NOT EXISTS documents (id text PRIMARY KEY, folder text REFERENCES folders,
                                            owner text NOT NULL, title text NOT NULL);
      CREATE TABLE IF NOT EXISTS outbox (id bigserial PRIMARY KEY, op text NOT NULL, tuple jsonb NOT NULL,
                                         created_at timestamptz DEFAULT now(), sent_at timestamptz);""")


_tl = threading.local()


def reader():
    """One kept-alive read connection per thread, so a request does not pay for a new PostgreSQL login."""
    if getattr(_tl, "c", None) is None or _tl.c.closed:
        _tl.c = psycopg.connect(DSN, autocommit=True)
    return _tl.c


def t(user, relation, obj):
    return {"user": user, "relation": relation, "object": obj}


def save(sql_steps, tuples):
    """Data and outbox rows in one transaction: both happen or neither does."""
    with psycopg.connect(DSN) as c:
        for sql, args in sql_steps:
            c.execute(sql, args)
        if not DUAL_WRITE:
            for op, tup in tuples:
                c.execute("INSERT INTO outbox (op, tuple) VALUES (%s, %s)", (op, json.dumps(tup)))
    if DUAL_WRITE:   # the second write happens after the commit, and can fail on its own
        s = store()
        body = {"writes": {"tuple_keys": [tp for op, tp in tuples if op == "write"]}}
        status, r = fga("POST", f"/stores/{s['store_id']}/write", body)
        if status != 200:
            raise RuntimeError(r)


def create_folder(h, rest, user, groups, body):
    save([("INSERT INTO folders VALUES (%s, %s)", (body["id"], user))],
         [("write", t(user, "owner", "folder:" + body["id"]))])
    h.reply(201, {"folder": body["id"]})


def share_folder(h, rest, user, groups, body):
    folder = rest.split("/")[0]
    if not authorize(user, "owner", "folder:" + folder):
        return h.reply(403, {"error": "forbidden"})
    save([], [("write", t(f"team:{body['team']}#member", "viewer", "folder:" + folder))])
    h.reply(200, {"shared": folder, "team": body["team"]})


def create_doc(h, rest, user, groups, body):
    if not authorize(user, "owner", "folder:" + body["folder"]):
        return h.reply(403, {"error": "forbidden"})
    try:
        save([("INSERT INTO documents VALUES (%s, %s, %s, %s)", (body["id"], body["folder"], user, body["title"]))],
             [("write", t("folder:" + body["folder"], "parent", "document:" + body["id"])),
              ("write", t(user, "owner", "document:" + body["id"]))])
    except Exception as e:
        return h.reply(500, {"error": str(e)[:200]})
    h.reply(201, {"document": body["id"], "at": time.strftime("%H:%M:%S")})


def get_doc(h, rest, user, groups, body):
    consistency = "HIGHER_CONSISTENCY" if h.query.get("consistency") == "higher" else "MINIMIZE_LATENCY"
    if not authorize(user, "viewer", "document:" + rest, consistency):
        return h.reply(403, {"error": "forbidden"})
    row = reader().execute("SELECT id, folder, title FROM documents WHERE id = %s", (rest,)).fetchone()
    if not row:
        return h.reply(404, {"error": "not found"})
    h.reply(200, {"id": row[0], "folder": row[1], "title": row[2]})


def list_docs(h, rest, user, groups, body):
    t0 = time.perf_counter()
    mode = h.query.get("mode", "list-objects")
    c = reader()
    if mode == "list-objects":   # ask the PDP which ids, then one SQL query for those ids only
        s = store()
        status, r = fga("POST", f"/stores/{s['store_id']}/list-objects", {
            "authorization_model_id": s["model_id"], "type": "document", "relation": "viewer", "user": user})
        if status != 200:
            return h.reply(503, {"error": "authorization unavailable"})
        ids = [o.split(":", 1)[1] for o in r["objects"]]
        rows = c.execute("SELECT id FROM documents WHERE id = ANY(%s) ORDER BY id", (ids,)).fetchall()
        visible = [x[0] for x in rows]
    else:                        # post-filter: load everything, then one check per row
        rows = c.execute("SELECT id FROM documents ORDER BY id").fetchall()
        visible = [x[0] for x in rows if authorize(user, "viewer", "document:" + x[0])]
    h.reply(200, {"mode": mode, "count": len(visible), "ms": round((time.perf_counter() - t0) * 1000, 1)})


serve([("POST", "/folders", lambda h, rest, *a: share_folder(h, rest, *a) if rest.endswith("/share") else create_folder(h, rest, *a)),
       ("POST", "/docs", create_doc), ("GET", "/docs", lambda h, rest, *a: get_doc(h, rest, *a) if rest else list_docs(h, rest, *a))])

وخدمة الفرق svc/teams.py، وفيها الفحص العام بالمجموعة من التوكن، وخيار السحب المتزامن:

# Teams service: owns team membership. Only members of the authentik group team-admins may change it
# (a coarse role from the token). Changes go to its own outbox; SYNC_REVOKE=1 also removes access in
# OpenFGA before answering, so a revocation never waits for the relay.
import json, os
import psycopg
import common
from common import fga, serve, store

common.SERVICE = "teams"
SYNC_REVOKE = os.environ.get("SYNC_REVOKE", "0") == "1"
DSN = os.environ["PG_DSN"]

with psycopg.connect(DSN, autocommit=True) as c:
    c.execute("""
      CREATE TABLE IF NOT EXISTS memberships (team text, username text, PRIMARY KEY (team, username));
      CREATE TABLE IF NOT EXISTS outbox (id bigserial PRIMARY KEY, op text NOT NULL, tuple jsonb NOT NULL,
                                         created_at timestamptz DEFAULT now(), sent_at timestamptz);""")


def change(h, rest, user, groups, body, op):
    if "team-admins" not in groups:
        return h.reply(403, {"error": "team-admins only"})
    team, _, member = rest.partition("/members/")
    tup = {"user": "user:" + member, "relation": "member", "object": "team:" + team}
    with psycopg.connect(DSN) as c:
        if op == "write":
            c.execute("INSERT INTO memberships VALUES (%s, %s) ON CONFLICT DO NOTHING", (team, member))
        else:
            c.execute("DELETE FROM memberships WHERE team = %s AND username = %s", (team, member))
        c.execute("INSERT INTO outbox (op, tuple) VALUES (%s, %s)", (op, json.dumps(tup)))
    if op == "delete" and SYNC_REVOKE:
        s = store()
        status, r = fga("POST", f"/stores/{s['store_id']}/write",
                        {"deletes": {"tuple_keys": [tup], "on_missing": "ignore"}})
        if status != 200:
            return h.reply(503, {"error": "revocation not applied yet", "detail": r.get("message")})
    h.reply(200, {"team": team, "member": member, "op": op, "sync": op == "delete" and SYNC_REVOKE})


serve([("PUT", "/teams", lambda h, rest, u, g, b: change(h, rest, u, g, b, "write")),
       ("DELETE", "/teams", lambda h, rest, u, g, b: change(h, rest, u, g, b, "delete"))])

وبرنامج ال Relay في svc/relay.py، وتشغل منه نسخة لكل خدمة على قاعدة بياناتها:

# Outbox relay: reads unsent rows of one service's outbox in order and applies them to OpenFGA.
# Writes and deletes are idempotent (on_duplicate / on_missing = ignore), so a row sent twice after
# a crash does no harm. A row is marked sent only after OpenFGA accepted it.
import json, os, time
import psycopg
from common import fga, store

DSN = os.environ["PG_DSN"]
NAME = DSN.rsplit("/", 1)[1]

while True:
    try:
        with psycopg.connect(DSN) as c:
            rows = c.execute("""SELECT id, op, tuple FROM outbox WHERE sent_at IS NULL
                                ORDER BY id LIMIT 50 FOR UPDATE SKIP LOCKED""").fetchall()
            for rid, op, tup in rows:
                s = store()
                body = ({"writes": {"tuple_keys": [tup], "on_duplicate": "ignore"}} if op == "write"
                        else {"deletes": {"tuple_keys": [tup], "on_missing": "ignore"}})
                status, r = fga("POST", f"/stores/{s['store_id']}/write", body)
                if status != 200:
                    raise RuntimeError(f"openfga {status}: {r.get('message')}")
                c.execute("UPDATE outbox SET sent_at = now() WHERE id = %s", (rid,))
                c.commit()
                print(json.dumps({"relay": NAME, "id": rid, "op": op, **tup}), flush=True)
    except Exception as e:
        print(json.dumps({"relay": NAME, "error": str(e)[:160]}), flush=True)
        time.sleep(2)
    time.sleep(0.5)

وبقية الملفات أدوات للتجربة: إعداد authentik، والعميل الذي يحصل على التوكن ويرسل الطلبات، وسكربتات القياس:

# Creates in authentik (through its API, with the bootstrap token): the OAuth2 provider and application
# "platform", the group team-admins, and three users with app passwords so the lab can get their tokens
# without a browser (client_credentials with username + app password).
import json, os, time, urllib.error, urllib.request

B = "http://lab-authz-authentik:9000/api/v3"
H = {"Authorization": "Bearer " + os.environ["AUTHENTIK_BOOTSTRAP_TOKEN"], "Content-Type": "application/json"}


def call(path, data=None, method=None):
    r = urllib.request.Request(B + path, data=json.dumps(data).encode() if data is not None else None,
                               headers=H, method=method or ("POST" if data is not None else "GET"))
    try:
        body = urllib.request.urlopen(r).read()
        return json.loads(body) if body else {}
    except urllib.error.HTTPError as e:
        print(e.code, e.read()[:300]); raise


for _ in range(90):           # the worker creates the bootstrap token shortly after start
    try:
        call("/core/users/me/"); break
    except Exception:
        time.sleep(5)

flow = lambda slug: call(f"/flows/instances/?slug={slug}")["results"][0]["pk"]
key = call("/crypto/certificatekeypairs/?name=authentik%20Self-signed%20Certificate")["results"][0]["pk"]
maps = []
while len(maps) < 2:          # the worker creates the managed scope mappings a little later
    maps = [m["pk"] for m in call("/propertymappings/provider/scope/?page_size=50")["results"] if m["managed"] in (
        "goauthentik.io/providers/oauth2/scope-openid", "goauthentik.io/providers/oauth2/scope-profile")]
    time.sleep(0 if len(maps) == 2 else 5)
p = call("/providers/oauth2/", {
    "name": "platform", "client_type": "confidential", "client_id": "platform",
    "client_secret": "lab-authz-platform-secret-6d2b", "signing_key": key, "property_mappings": maps,
    "authorization_flow": flow("default-provider-authorization-implicit-consent"),
    "invalidation_flow": flow("default-provider-invalidation-flow"),
    "grant_types": ["client_credentials"], "access_token_validity": "minutes=10",
    "redirect_uris": [{"matching_mode": "strict", "url": "http://127.0.0.1/unused"}]})
call("/core/applications/", {"name": "Platform", "slug": "platform", "provider": p["pk"]})
admins = call("/core/groups/", {"name": "team-admins"})

users = {}
for name in ("alice", "bob", "carol"):
    u = call("/core/users/", {"username": name, "name": name.title(), "path": "lab",
                              "groups": [admins["pk"]] if name == "alice" else []})
    ident = f"lab-{name}-app-password"
    call("/core/tokens/", {"identifier": ident, "user": u["pk"], "intent": "app_password", "expiring": False})
    secret = f"{name}-app-password-2026-lab"
    call(f"/core/tokens/{ident}/set_key/", {"key": secret})
    users[name] = secret

with open("/state/users.json", "w") as f:
    json.dump(users, f)
print("provider", p["client_id"], "users", ", ".join(users), "team-admins: alice")

# Client helper for the lab: gets a user's token from authentik and calls the gateway with it.
#   python api.py token alice
#   python api.py alice GET /api/docs/plan-q3
#   python api.py alice POST /api/docs '{"id": "...", "folder": "...", "title": "..."}'
import json, sys, time, urllib.error, urllib.parse, urllib.request

AK = "http://lab-authz-authentik:9000/application/o/token/"
GW = "http://lab-authz-apisix:9080"
_tokens = {}


def token(user):
    if user not in _tokens:
        secrets = json.load(open("/state/users.json"))
        data = urllib.parse.urlencode({"grant_type": "client_credentials", "client_id": "platform",
                                       "client_secret": "lab-authz-platform-secret-6d2b",
                                       "username": user, "password": secrets[user],
                                       "scope": "openid profile"}).encode()
        _tokens[user] = json.load(urllib.request.urlopen(AK, data))["access_token"]
    return _tokens[user]


def call(user, method, path, body=None):
    req = urllib.request.Request(GW + path, method=method, data=json.dumps(body).encode() if body else None,
                                 headers={"Authorization": "Bearer " + token(user), "Content-Type": "application/json"})
    try:
        r = urllib.request.urlopen(req)
        return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw)
        except ValueError:
            return e.code, raw.decode()[:80]


if __name__ == "__main__":
    if sys.argv[1] == "token":
        print(token(sys.argv[2]))
    else:
        user, method, path = sys.argv[1:4]
        body = json.loads(sys.argv[4]) if len(sys.argv) > 4 else None
        status, out = call(user, method, path, body)
        print(time.strftime("%H:%M:%S"), status, json.dumps(out, ensure_ascii=False))

# Latency of one authorization decision, measured from inside the Docker network.
#   python bench.py fga [N] [consistency]   one Check straight to OpenFGA
#   python bench.py svc [N]                 GET /docs/budget-q3 on the service, no gateway
#   python bench.py gw  [N]                 the same request through APISIX (token checked there)
import http.client, json, statistics, sys, time
import api
from common import store

mode = sys.argv[1]
n = int(sys.argv[2]) if len(sys.argv) > 2 else 1000
consistency = sys.argv[3] if len(sys.argv) > 3 else "MINIMIZE_LATENCY"
tok = api.token("bob")

if mode == "fga":
    s = store()
    host, port, method, path = "lab-authz-openfga", 8080, "POST", f"/stores/{s['store_id']}/check"
    body = json.dumps({"tuple_key": {"user": "user:bob", "relation": "viewer", "object": "document:budget-q3"},
                       "authorization_model_id": s["model_id"], "consistency": consistency})
    headers = {"Content-Type": "application/json"}
else:
    host, port = ("lab-authz-docs", 8000) if mode == "svc" else ("lab-authz-apisix", 9080)
    method, path, body = "GET", ("/docs/budget-q3" if mode == "svc" else "/api/docs/budget-q3"), None
    headers = {"Authorization": "Bearer " + tok}

conn = http.client.HTTPConnection(host, port, timeout=5)
times, codes = [], {}
for i in range(n + 50):                      # the first 50 warm up connections and caches
    t0 = time.perf_counter()
    conn.request(method, path, body=body, headers=headers)
    r = conn.getresponse(); r.read()
    if i >= 50:
        times.append((time.perf_counter() - t0) * 1000)
        codes[r.status] = codes.get(r.status, 0) + 1
times.sort()
q = lambda p: times[min(len(times) - 1, int(p * len(times)))]
print(f"{mode:4} {consistency if mode == 'fga' else '':18} n={n} status={codes} "
      f"p50={q(.50):.2f}ms p95={q(.95):.2f}ms p99={q(.99):.2f}ms mean={statistics.mean(times):.2f}ms")

# Asks every second for SECONDS seconds whether a user can still open a document, through the gateway,
# with the default consistency and with HIGHER_CONSISTENCY, starting right after an action.
#   python watch.py default bob budget-q3 15 "alice DELETE /api/teams/finance/members/bob"
#   (first argument: default or higher, the consistency the service asks OpenFGA for)
import sys, time
import api

mode = sys.argv.pop(1)
user, doc, seconds = sys.argv[1], sys.argv[2], int(sys.argv[3])
q = "?consistency=higher" if mode == "higher" else ""
api.token(user)
if len(sys.argv) > 4:
    who, method, path = sys.argv[4].split()
    api.token(who)
    t0 = time.monotonic()
    print(f"t=0.0s  {who} {method} {path} -> {api.call(who, method, path)[0]}")
else:
    t0 = time.monotonic()
last = None
while time.monotonic() - t0 < seconds:
    a = api.call(user, "GET", f"/api/docs/{doc}{q}")[0]
    if a != last:
        print(f"t={time.monotonic() - t0:4.1f}s  {mode}: {user} GET {doc} -> {a}")
        last = a
    time.sleep(0.25)
print(f"t={time.monotonic() - t0:4.1f}s  {mode}: {user} GET {doc} -> {a}  (end)")

-- 5,000 old documents in alice's private archive and 50 more in the finance folder,
-- with their outbox rows, as if they had been created through the API.
BEGIN;
INSERT INTO documents SELECT 'arc-' || g, 'archive', 'user:alice', 'old document ' || g FROM generate_series(1, 5000) g;
INSERT INTO documents SELECT 'fin-' || g, 'finance-2026', 'user:alice', 'finance note ' || g FROM generate_series(1, 50) g;
INSERT INTO outbox (op, tuple)
  SELECT 'write', jsonb_build_object('user', 'folder:' || folder, 'relation', 'parent', 'object', 'document:' || id)
  FROM documents WHERE id LIKE 'arc-%' OR id LIKE 'fin-%';
COMMIT;

# Post-filter with BatchCheck: the same 5,055 rows, 50 checks per request (the server's default limit).
import sys, time, psycopg, os
from common import fga, store

user = "user:" + sys.argv[1]
s = store()
with psycopg.connect(os.environ["PG_DSN"]) as c:
    ids = [r[0] for r in c.execute("SELECT id FROM documents ORDER BY id")]
t0 = time.perf_counter(); visible = 0
for i in range(0, len(ids), 50):
    chunk = ids[i:i + 50]
    status, r = fga("POST", f"/stores/{s['store_id']}/batch-check", {"authorization_model_id": s["model_id"], "checks": [
        {"tuple_key": {"user": user, "relation": "viewer", "object": "document:" + d}, "correlation_id": str(n)}
        for n, d in enumerate(chunk)]})
    assert status == 200, r
    visible += sum(1 for v in r["result"].values() if v.get("allowed"))
print(f"batch-check {user}: rows={len(ids)} visible={visible} requests={-(-len(ids) // 50)} ms={(time.perf_counter() - t0) * 1000:.0f}")

التشغيل وإعداد المخزن والمستخدمين

نبني الخدمات ونشغل البنية الأساسية أولاً، ثم ننشئ المخزن Store في OpenFGA ونحمل فيه النموذج بأداة سطر الأوامر من Docker Image الرسمية:

docker compose build
docker compose up -d pg openfga-migrate openfga authentik authentik-worker apisix client
cd svc
docker run --rm --network lab-authz-net -v "$PWD:/work" -w /work \
  -e FGA_API_URL=http://lab-authz-openfga:8080 openfga/cli:v0.8.1 \
  store create --name platform --model model.fga
cd ..

{"store":{"created_at":"2026-10-12T11:51:17.18648Z","id":"01M4QZ7TY010ZHWPPYAYY08T1T","name":"platform","updated_at":"2026-10-12T11:51:17.18648Z"},"model":{"authorization_model_id":"01M4QZ7TYSZ0319MV7VS5NQB6H"}}

نحفظ المعرفين في ملف تقرؤه كل الخدمات، ثم نجهز authentik، ثم نشغل الخدمات وبرامج ال Relay:

printf '{"store_id": "01M4QZ7TY010ZHWPPYAYY08T1T", "model_id": "01M4QZ7TYSZ0319MV7VS5NQB6H"}\n' > state/fga.json
set -a; . ./.env; set +a
docker exec -e AUTHENTIK_BOOTSTRAP_TOKEN=$AUTHENTIK_BOOTSTRAP_TOKEN lab-authz-client python ak_setup.py
docker compose up -d docs teams relay-docs relay-teams

provider platform users alice, bob, carol team-admins: alice

وقد أخذ authentik نحو سبع دقائق في أول تشغيل قبل أن يقبل التوكن الأولي، فالسكربت ينتظره. وفي المحاولة الأولى خرج التوكن دون أي حقل غير المعرف، وكان حقل scope فارغاً، والسبب أن السكربت سأل عن ربط النطاقات Scope Mappings قبل أن ينشئها ال Worker، فأنشأ المزود دونها، لذلك أصبح السكربت ينتظر حتى يجدها. والطريقة التي نحصل بها على التوكن دون متصفح هي منحة client_credentials مع اسم المستخدم وكلمة مرور تطبيق App Password كما في توثيق authentik، فيصدر التوكن باسم هذا المستخدم.

أول طلبات عبر المنصة

تنشئ alice مجلداً وتشاركه مع فريق المالية وتضع فيه مستنداً، ثم نسأل باسم bob وcarol:

c() { docker exec lab-authz-client python api.py "$@"; }
c alice POST /api/folders '{"id": "finance-2026"}'
c alice POST /api/folders/finance-2026/share '{"team": "finance"}'
c alice POST /api/docs '{"id": "budget-q3", "folder": "finance-2026", "title": "ميزانية الربع الثالث"}'
c alice PUT /api/teams/finance/members/bob
sleep 2
c bob GET /api/docs/budget-q3
c carol GET /api/docs/budget-q3
c bob PUT /api/teams/finance/members/carol

والمخرج سوف يكون كما يلي:

12:01:49 201 {"folder": "finance-2026"}
12:01:50 200 {"shared": "finance-2026", "team": "finance"}
12:01:53 201 {"document": "budget-q3", "at": "12:01:53"}
12:01:21 200 {"team": "finance", "member": "bob", "op": "write", "sync": false}
12:01:58 200 {"id": "budget-q3", "folder": "finance-2026", "title": "ميزانية الربع الثالث"}
12:01:59 403 {"error": "forbidden"}
12:01:28 403 {"error": "team-admins only"}

في المخرج أعلاه لاحظ التالي:

• bob يقرأ المستند دون أن تكتب أي خدمة علاقة بينه وبين المستند، فالمحرك مشى من المستند إلى المجلد إلى فريق المالية إلى bob.
• carol مرفوضة بالمحرك، أما محاولة bob إضافة carol إلى الفريق فمرفوضة بالفحص العام في خدمة الفرق، لأنه ليس في مجموعة team-admins في authentik.
• أوقات الطلبين المأخوذين من تشغيل سابق، 12:01:21 و12:01:28، أقدم من البقية، لأن أول تشغيل فشل في إنشاء المجلد برمز 404، فقد كان مسار /api/folders دون نجمة غير موجود في ملف البوابة، فأضفناه وأعدنا الطلبات الثلاثة الأولى.

ويصل الطلب دون توكن أو بتوكن مزور إلى البوابة فقط، فترد برمز 401 قبل أن يصل إلى أي خدمة:

401 b'<html>\r\n<head><title>401 Authorization Required</title></head>...
401 b'<html>\r\n<head><title>401 Authorization Required</title></head>...

قياس الزمن مع ال Cache ودونه

نقيس زمن قرار bob على المستند budget-q3 في ثلاثة أماكن: المحرك مباشرة، والخدمة دون بوابة، والطلب كاملاً عبر APISIX، ثم نفعل ال Cache في المحرك، ثم ال Cache المحلي في الخدمة خمس ثوان:

b() { docker exec lab-authz-client python bench.py "$@"; }
b fga 3000; b fga 3000 HIGHER_CONSISTENCY; b svc 2000; b gw 2000
FGA_CACHE=true docker compose up -d openfga; sleep 8
b fga 3000; b fga 3000 HIGHER_CONSISTENCY; b svc 2000; b gw 2000
FGA_CACHE=true LOCAL_CACHE_TTL=5 docker compose up -d docs; sleep 8
b svc 2000; b gw 2000

fga  MINIMIZE_LATENCY   n=3000 status={200: 3000} p50=2.35ms p95=3.66ms p99=5.01ms mean=2.51ms
fga  HIGHER_CONSISTENCY n=3000 status={200: 3000} p50=2.33ms p95=3.78ms p99=5.05ms mean=2.50ms
svc                     n=2000 status={200: 2000} p50=6.63ms p95=16.56ms p99=26.76ms mean=7.97ms
gw                      n=2000 status={200: 2000} p50=7.97ms p95=15.16ms p99=28.26ms mean=8.98ms
fga  MINIMIZE_LATENCY   n=3000 status={200: 3000} p50=1.27ms p95=2.80ms p99=5.47ms mean=1.43ms
fga  HIGHER_CONSISTENCY n=3000 status={200: 3000} p50=3.06ms p95=5.88ms p99=8.99ms mean=3.37ms
svc                     n=2000 status={200: 2000} p50=3.98ms p95=5.44ms p99=6.63ms mean=4.14ms
gw                      n=2000 status={200: 2000} p50=5.32ms p95=7.05ms p99=8.84ms mean=5.54ms
svc                     n=2000 status={200: 2000} p50=0.78ms p95=1.32ms p99=1.54ms mean=0.86ms
gw                      n=2000 status={200: 2000} p50=1.88ms p95=2.77ms p99=3.27ms mean=1.98ms

هذه نتيجة التشغيل الثاني، وفي التشغيل الأول على الجهاز نفسه كانت أعلى وأكثر تذبذباً، مثل 4.14 ملي ثانية في الوسيط للمحرك و11.42 عبر البوابة، لأن تجارب أخرى كانت تعمل على الجهاز في الوقت نفسه، لذلك اقرأ الأرقام كمقارنة بين الطبقات لا كأرقام مطلقة. وتستطيع أن تتأكد أن ال Cache يعمل فعلاً من مقاييس المحرك على المنفذ 18112:

curl -s 127.0.0.1:18112/metrics | grep -E '^openfga_check_cache|cache_item_count'

openfga_cache_item_count{entity="authz_model"} 0
openfga_cache_item_count{entity="check_response"} 2
openfga_cache_item_count{entity="typesystem"} 0
openfga_check_cache_hit_count 5096
openfga_check_cache_invalid_hit_count 0
openfga_check_cache_total_count 5104

كم يبقى القرار القديم بعد السحب؟

مع ال Cache في المكانين، نعيد bob إلى الفريق وننتظر حتى يقرأ المستند، ثم تحذفه alice ونسأل كل ربع ثانية، مرة بالاتساق الافتراضي ومرة ب HIGHER_CONSISTENCY، وكل مرة بعد إعادة bob:

w() { docker exec lab-authz-client python watch.py "$@"; }
c alice PUT /api/teams/finance/members/bob; sleep 12; c bob GET /api/docs/budget-q3
w default bob budget-q3 16 "alice DELETE /api/teams/finance/members/bob"
c alice PUT /api/teams/finance/members/bob; sleep 12; c bob GET /api/docs/budget-q3
w higher bob budget-q3 6 "alice DELETE /api/teams/finance/members/bob"

t=0.0s  alice DELETE /api/teams/finance/members/bob -> 200
t= 0.1s  default: bob GET budget-q3 -> 200
t= 9.1s  default: bob GET budget-q3 -> 403
t=16.2s  default: bob GET budget-q3 -> 403  (end)
t=0.0s  alice DELETE /api/teams/finance/members/bob -> 200
t= 0.0s  higher: bob GET budget-q3 -> 200
t= 0.6s  higher: bob GET budget-q3 -> 403
t= 6.1s  higher: bob GET budget-q3 -> 403  (end)

في المخرج أعلاه لاحظ أن الاتساق الأعلى لم يجعل السحب فورياً، فقد بقي 0.6 ثانية هي مدة ال Relay. وفي محاولة سابقة كنا نرسل الطلبين في الحلقة نفسها، فظهر السحب بعد 0.7 ثانية في الحالتين، والسبب أن طلب HIGHER_CONSISTENCY كتب نتيجته الجديدة في ال Cache المحلي للخدمة، فاستفاد منها الطلب العادي، لذلك فصلنا القياسين.

نرى مشكلة New Enemy بأعيننا

نعيد ال Cache إلى وضعه الافتراضي (معطل)، ونوقف برنامج ال Relay الخاص بخدمة الفرق، كأنه تعطل أو تأخر، ثم تحذف alice عضوية bob وتضيف مستنداً جديداً إلى المجلد:

docker compose up -d openfga docs
c alice PUT /api/teams/finance/members/bob; sleep 2
c bob GET /api/docs/budget-q3
docker stop lab-authz-relay-teams
c alice DELETE /api/teams/finance/members/bob
c alice POST /api/docs '{"id": "reorg-plan", "folder": "finance-2026", "title": "خطة إعادة الهيكلة"}'
sleep 2
c bob GET /api/docs/reorg-plan
c bob GET "/api/docs/reorg-plan?consistency=higher"
docker exec lab-authz-pg psql -U lab -d teams -c "SELECT id, op, tuple->>'user' AS user, tuple->>'object' AS object, sent_at FROM outbox ORDER BY id DESC LIMIT 2"
docker start lab-authz-relay-teams; sleep 3
c bob GET /api/docs/reorg-plan

12:17:40 200 {"team": "finance", "member": "bob", "op": "write", "sync": false}
12:17:43 200 {"id": "budget-q3", "folder": "finance-2026", "title": "ميزانية الربع الثالث"}
lab-authz-relay-teams
12:17:50 200 {"team": "finance", "member": "bob", "op": "delete", "sync": false}
12:17:50 201 {"document": "reorg-plan", "at": "12:17:50"}
12:17:53 200 {"id": "reorg-plan", "folder": "finance-2026", "title": "خطة إعادة الهيكلة"}
12:17:54 200 {"id": "reorg-plan", "folder": "finance-2026", "title": "خطة إعادة الهيكلة"}
 id |   op   |   user   |    object    |            sent_at
----+--------+----------+--------------+-------------------------------
 10 | delete | user:bob | team:finance |
  9 | write  | user:bob | team:finance | 2026-10-12 12:17:40.801598+00
(2 rows)

lab-authz-relay-teams
12:18:00 403 {"error": "forbidden"}

خطة إعادة الهيكلة كتبت بعد أن أخرج bob من الفريق، ومع ذلك قرأها مرتين، والثانية ب HIGHER_CONSISTENCY، لأن الحذف ما زال في الصف 10 الذي لم يرسل. والآن نعيد التجربة مع السحب المتزامن:

SYNC_REVOKE=1 docker compose up -d teams; sleep 8
c alice PUT /api/teams/finance/members/bob; sleep 2
c bob GET /api/docs/reorg-plan
docker stop lab-authz-relay-teams
c alice DELETE /api/teams/finance/members/bob
c alice POST /api/docs '{"id": "salaries-2027", "folder": "finance-2026", "title": "رواتب 2027"}'
sleep 2
c bob GET /api/docs/salaries-2027
docker start lab-authz-relay-teams; sleep 3; docker logs lab-authz-relay-teams 2>&1 | tail -2

12:18:24 200 {"team": "finance", "member": "bob", "op": "write", "sync": false}
12:18:27 200 {"id": "reorg-plan", "folder": "finance-2026", "title": "خطة إعادة الهيكلة"}
lab-authz-relay-teams
12:18:34 200 {"team": "finance", "member": "bob", "op": "delete", "sync": true}
12:18:36 201 {"document": "salaries-2027", "at": "12:18:36"}
12:18:39 403 {"error": "forbidden"}
lab-authz-relay-teams
{"relay": "teams", "id": 11, "op": "write", "user": "user:bob", "object": "team:finance", "relation": "member"}
{"relay": "teams", "id": 12, "op": "delete", "user": "user:bob", "object": "team:finance", "relation": "member"}

لاحظ في آخر سطر أن ال Relay أرسل الحذف مرة ثانية بعد عودته، والعلاقة كانت قد حذفت أصلاً بالسحب المتزامن، ولم يحدث أي خطأ بفضل on_missing: "ignore".

نكسرها: الكتابة المزدوجة

نشغل خدمة المستندات بالكتابة المزدوجة، ومع Cache محلي دقيقة حتى يمر فحص ملكية المجلد من ال Cache بعد أن نوقف المحرك، وهذا يشبه حالة حقيقية يتعطل فيها المحرك بين الفحص والكتابة:

DUAL_WRITE=1 LOCAL_CACHE_TTL=60 docker compose up -d docs; sleep 7
c alice POST /api/docs '{"id": "draft-notes", "folder": "finance-2026", "title": "ملاحظات"}'
docker stop lab-authz-openfga
c alice POST /api/docs '{"id": "q4-forecast", "folder": "finance-2026", "title": "توقعات الربع الرابع"}'
docker start lab-authz-openfga; sleep 3
docker exec lab-authz-pg psql -U lab -d docs -c "SELECT id, owner FROM documents WHERE id IN ('draft-notes','q4-forecast')"
c alice GET "/api/docs/q4-forecast?consistency=higher"
docker compose up -d docs

12:23:13 201 {"document": "draft-notes", "at": "12:23:13"}
12:23:21 500 {"error": "[Errno -2] Name or service not known"}
     id      |   owner
-------------+------------
 draft-notes | user:alice
 q4-forecast | user:alice
(2 rows)

12:23:26 403 {"error": "forbidden"}

رأت alice رسالة خطأ، ولكن المستند حفظ فعلاً، وعندما عاد المحرك لم تستطع أن تفتح مستنداً هي صاحبته، لأن علاقاته لم تكتب أبداً، ولن تكتب إلا بإصلاح يدوي. أما مع ال Outbox فأوقفنا برنامج ال Relay بدل المحرك:

docker stop lab-authz-relay-docs
c alice POST /api/docs '{"id": "q4-plan", "folder": "finance-2026", "title": "خطة الربع الرابع"}'
c alice GET /api/docs/q4-plan
docker exec lab-authz-pg psql -U lab -d docs -c "SELECT id, op, tuple->>'relation' AS relation, tuple->>'object' AS object, sent_at FROM outbox WHERE sent_at IS NULL"
docker start lab-authz-relay-docs; sleep 3
c alice GET /api/docs/q4-plan

12:23:53 201 {"document": "q4-plan", "at": "12:23:53"}
12:23:53 403 {"error": "forbidden"}
 id |  op   | relation |      object      | sent_at
----+-------+----------+------------------+---------
  9 | write | parent   | document:q4-plan |
 10 | write | owner    | document:q4-plan |
(2 rows)

12:23:59 200 {"id": "q4-plan", "folder": "finance-2026", "title": "خطة الربع الرابع"}

العلاقات انتظرت في الجدول، ووصلت عندما عاد ال Relay، وهذا هو الفرق كله: التأخير يصلح نفسه، والضياع لا يصلح نفسه.

نقيس القوائم على خمسة آلاف مستند

نعيد bob إلى الفريق، وننشئ مجلد أرشيف خاص ب alice، ثم نضع خمسة آلاف مستند في الأرشيف وخمسين في مجلد المالية عبر الملف seed.sql، مع صفوف ال Outbox الخاصة بها:

c alice PUT /api/teams/finance/members/bob
c alice POST /api/folders '{"id": "archive"}'
docker exec -i lab-authz-pg psql -U lab -d docs < svc/seed.sql
time (while [ "$(docker exec lab-authz-pg psql -U lab -d docs -Atc 'SELECT count(*) FROM outbox WHERE sent_at IS NULL')" != "0" ]; do sleep 5; done)

INSERT 0 5000
INSERT 0 50
INSERT 0 5050

real	2m42.337s

احتاج ال Relay دقيقتين و42 ثانية لينقل 5,050 علاقة، أي نحو 31 علاقة في الثانية، لأنه يرسل كل صف في طلب مستقل ويؤكد كل صف في معاملة. وهذا يكفي للتغييرات اليومية، ولكنه لا يكفي لترحيل بيانات قديمة، فعند الترحيل اجمع العلاقات في طلبات أكبر، مع الانتباه إلى أن OpenFGA يقبل 100 علاقة كحد أقصى في طلب الكتابة الواحد افتراضياً (maxTuplesPerWrite) كما في جدول الإعدادات. ثم نقيس القائمة ثلاث مرات لكل طريقة ولكل مستخدم:

for u in bob alice; do for m in list-objects post-filter; do for i in 1 2 3; do
  c $u GET "/api/docs?mode=$m"; done; done; done
for u in bob alice; do docker exec -e PG_DSN="postgresql://lab:$PG_PASS@lab-authz-pg:5432/docs" lab-authz-client python batch_bench.py $u; done

12:27:06 200 {"mode": "list-objects", "count": 55, "ms": 41.4}
12:27:11 200 {"mode": "list-objects", "count": 55, "ms": 1327.4}
12:27:15 200 {"mode": "list-objects", "count": 55, "ms": 51.9}
12:28:17 504 "<html>\r\n<head><title>504 Gateway Time-out</title></head>\r\n<body>\r\n<center><h1>50"
12:29:21 504 "<html>\r\n<head><title>504 Gateway Time-out</title></head>\r\n<body>\r\n<center><h1>50"
12:29:58 200 {"mode": "post-filter", "count": 55, "ms": 35673.4}
12:29:59 200 {"mode": "list-objects", "count": 1000, "ms": 23.1}
12:30:00 200 {"mode": "list-objects", "count": 1000, "ms": 39.9}
12:30:01 200 {"mode": "list-objects", "count": 1000, "ms": 66.3}
12:30:28 200 {"mode": "post-filter", "count": 5055, "ms": 26273.5}
12:30:56 200 {"mode": "post-filter", "count": 5055, "ms": 27805.0}
12:31:19 200 {"mode": "post-filter", "count": 5055, "ms": 22088.2}
batch-check user:bob: rows=5056 visible=55 requests=102 ms=1653
batch-check user:alice: rows=5056 visible=5055 requests=102 ms=1731

في المخرج أعلاه لاحظ التالي:

• قائمة alice بطريقة list-objects أعطت 1000 بينما الصحيح 5,055، وهو الحد الافتراضي الذي شرحناه في القوائم، ولم يظهر أي خطأ.
• عدد الصفوف 5,056 وما تستطيع alice رؤيته 5,055، والفرق هو المستند q4-forecast الذي تركته الكتابة المزدوجة بلا علاقات، فما زال يظهر في كل قياس بعد ذلك.
• الطلب الثاني ل bob أخذ 1.3 ثانية بدل 40 ملي ثانية، ولم نجد له سبباً في السجلات، وهذا يكفي لتعرف أن الوسيط وحده لا يكفي، وأن تراقب p99 في الإنتاج.

نوقف المحرك

docker stop lab-authz-openfga
c alice GET /api/docs/budget-q3
c carol GET /api/docs/budget-q3
docker logs lab-authz-docs 2>&1 | tail -2
docker start lab-authz-openfga
FAIL_OPEN=1 docker compose up -d docs; sleep 7; docker stop lab-authz-openfga
c carol GET /api/docs/budget-q3
docker start lab-authz-openfga; docker compose up -d docs

12:20:40 403 {"error": "forbidden"}
12:20:46 403 {"error": "forbidden"}
{"ts": "12:20:40", "svc": "docs", "user": "user:alice", "relation": "viewer", "object": "document:budget-q3", "allowed": false, "source": "error: gaierror", "consistency": "MINIMIZE_LATENCY", "ms": 5235.09}
{"ts": "12:20:46", "svc": "docs", "user": "user:carol", "relation": "viewer", "object": "document:budget-q3", "allowed": false, "source": "error: gaierror", "consistency": "MINIMIZE_LATENCY", "ms": 4580.15}
12:21:15 200 {"id": "budget-q3", "folder": "finance-2026", "title": "ميزانية الربع الثالث"}

الرفض عند التعطل منع carol، ومتغير واحد جعل الخدمة تعطيها ميزانية الربع الثالث. وتذكر أن هذا المتغير لا يكتبه أحد بنية سيئة، وإنما يأتي غالباً في صورة except: return True كتبها مطور أراد ألا تتوقف الخدمة أثناء صيانة المحرك.

إزالة التجربة

نحذف ما أنشأناه فقط، باسمه:

docker compose down -v
docker rmi lab-authz-py:1
rm -rf ~/authz-platform

الأخطاء الشائعة

• الكتابة المزدوجة: حفظ البيانات ثم كتابة العلاقة مباشرة دون Outbox، فيبقى مستند بلا علاقات عند أول عطل، كما حدث مع q4-forecast.
• اعتبار HIGHER_CONSISTENCY حلاً لكل شيء: هو يتجاوز ال Cache فقط، ولا يعرف بالسحب الذي لم يصل بعد.
• سحب الصلاحية بشكل غير متزامن: الإضافة يمكن أن تتأخر، أما السحب فيجب أن يطبق قبل أن تقول الخدمة «تم».
• تجاهل حد ListObjects: قائمة مقطوعة عند ألف نتيجة دون أي خطأ.
• Cache في الخدمة مدته طويلة: كل ثانية فيه ثانية يبقى فيها الموظف المغادر قادراً على الوصول.
• معرف المستخدم من الاسم: preferred_username يتغير، والعلاقات تبقى على الاسم القديم.
• الثقة بأي ترويسة تصل إلى الخدمة: إذا لم تتحقق الخدمة من التوكن فيجب أن تكون معزولة بحيث لا يصل إليها إلا البوابة.
• الرد ب 403 عند تعطل المحرك، أو أسوأ من ذلك السماح عند التعطل.
• خدمة تكتب علاقات تخص خدمة أخرى، فلا يعرف أحد من مصدر الحقيقة لهذه العلاقة.
• القياس قبل إصلاح الأساسيات: اتصال جديد بقاعدة البيانات في كل طلب يضيع أكثر من المحرك كله.

الخلاصة

• عندما يخدم المحرك عدة خدمات يصبح منصة: فريق المنصة يملك المحرك والنموذج والمكتبة وسجل القرارات، وكل خدمة تملك بياناتها والعلاقات التي تخص أنواعها فقط.
• البوابة تفحص التوكن والمسار، والخدمة تسأل عن كل مورد بعينه، والهوية من authentik بمعرف ثابت ومجموعات عامة.
• العلاقات تصل إلى المحرك عبر Outbox في المعاملة نفسها وبرنامج Relay يكتب بشكل آمن عند التكرار، مع مهمة مطابقة دورية، والكتابة المزدوجة تركت في التجربة مستنداً لا تستطيع صاحبته فتحه.
• القرار القديم له ثلاثة مصادر: Cache المحرك، وCache الخدمة، وتأخر ال Relay، وHIGHER_CONSISTENCY يعالج الأول فقط، وبين خدمتين ظهرت مشكلة New Enemy فعلاً وحلها السحب المتزامن.
• في التجربة كان القرار نفسه نحو 2.4 ملي ثانية، وال Cache أنزله إلى أقل من ملي ثانية، ولكنه أبقى bob يقرأ 9.1 ثانية بعد حذفه، وListObjects أسرع بكثير من الترشيح بعد الجلب، وتوقف عند ألف نتيجة دون خطأ.
• عند تعطل المحرك ارفض برمز 503 وبمهلة قصيرة، ولا تسمح أبداً، والتطبيقات الجاهزة تبقى بأدوارها الخاصة وتدار بمجموعات مركزية.

وفي المقالين التاليين من السلسلة نتعمق في المحركين: OpenFGA عملياً لتصميم نموذج العلاقات واختباره وتغييره، وOPA عملياً لسياسات الخصائص وتوزيعها على الخدمات والبوابات.

دليلAuthorization: كيف تجمع قرارات الصلاحيات في مكان واحدلماذا تتحول شروط الصلاحيات المتفرقة في الكود إلى ثغرات، وكيف تنقل القرار إلى محرك سياسات واحد.دليلOpenFGA عملياً: نموذج العلاقات من التصميم إلى الإنتاجكيف تصمم نموذج علاقات في OpenFGA للمؤسسة والفرق والمجلدات، وتختبره وتغيره وتشغله على PostgreSQL دون مفاجآت.دليلOPA عملياً: سياسات ABAC لخدماتك وبواباتككيف تكتب سياسات ABAC بلغة Rego وتختبرها، ثم توزعها موقعة على خدماتك وبواباتك وتسجل كل قرار تتخذه.دليلالمعاملات الموزعة: Saga ونمط Outbox بدل 2PCكيف تنفذ عملية تمتد على عدة خدمات دون 2PC، بخطوات محلية وتعويضات وجدول Outbox لا يضيع فيه أي حدث.دليلمستخدم أم مدير؟ إدارة الأدوار عبر تطبيقاتك من authentikكيف تحدد من authentik من يدخل كل تطبيق ومن يكون مديراً فيه، وماذا يحدث للجلسات والتوكنات عندما تسحب الصلاحية.دليلApache APISIX عملياً: من أول Route إلى بوابة إنتاجكيف تشغل Apache APISIX مع etcd، وتحمي واجهاتك بالتوكنات والحدود وauthentik، وتراقبها وتأخذ نسخة احتياطية من إعدادها.

سجل التحديثات

• أكتوبر 2026: كتابة المقال، وتنفيذ التجربة على OpenFGA 1.22.0 وfga CLI 0.8.1 وauthentik 2026.8.3 وAPISIX 3.19.0 وPostgreSQL 18.6 وPython 3.13.16 على Docker Engine 29.8.0.