الوحدة 7 — المصادقة بالرمز

الخدمة التي بنيناها حتّى الآن مفتوحة تمامًا: كلّ من يعرف عنوان URL يستطيع استدعاء /predict ألف مرّة في الثانية، أو حتّى تفريغ قاعدة العملاء بالكامل عبر /predict/batch. في مشغّل اتّصالات حقيقيّ، هذا كارثة. هذه الوحدة تُغلق الباب: نُلزم كلّ طلب بحمل رمز يُثبت هويّة العميل، ونُطبّق حدًّا أقصى للمعدّل يمنع الاستنزاف حتّى من عميل مُصرَّح له.

API Key: أبسط ما يعمل

الشكل الأبسط للمصادقة في خدمة داخليّة: مفتاح ثابت طويل يُمرَّر في ترويسة HTTP. لا حاجة لقاعدة بيانات جلسات ولا لتشفير معقّد. المبدأ:

• كلّ عميل يُعرف بمفتاحه (churn-dashboard، crm-nightly-job، mobile-app).
• المفاتيح تُخزَّن مُجزَّأة (hachées) في متغيّر بيئة أو Secret manager.
• كلّ طلب يحمل ترويسة X-API-Key: <valeur>.

الشيفرة الأساسيّة بـFastAPI:

import os
import hashlib
from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader

# اسم الترويسة، ومنع FastAPI من رمي 403 قبل معالجنا:
verificateur = APIKeyHeader(name="X-API-Key", auto_error=False)

# تُقرأ عند البدء (متغيّر بيئة، سطر واحد "cle1:sha256,cle2:sha256")
CLES_AUTORISEES = {
    "churn-dashboard": os.environ["APIKEY_DASHBOARD_SHA256"],
    "crm-nightly-job": os.environ["APIKEY_CRM_SHA256"],
}


def sha256(v: str) -> str:
    return hashlib.sha256(v.encode()).hexdigest()


async def client_authentifie(cle: str = Depends(verificateur)) -> str:
    if cle is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="X-API-Key manquant",
            headers={"WWW-Authenticate": "ApiKey"},
        )
    haché = sha256(cle)
    for nom, ref in CLES_AUTORISEES.items():
        if haché == ref:
            return nom
    raise HTTPException(status_code=401, detail="cle inconnue")

استعمالها في المسار كتبعيّة:

@app.post("/predict", response_model=ChurnResponse)
def predire(req: ChurnRequest, qui: str = Depends(client_authentifie)) -> ChurnResponse:
    # qui يحمل الآن اسم العميل المُصرَّح له، مفيد للتسجيل
    ...

الفارق مع ما سبق: FastAPI يعرف تلقائيًّا أنّ هذه التبعيّة تعبّر عن مصادقة، ويعرضها في التوثيق التفاعليّ بواجهة قفل. أيّ طلب بلا X-API-Key أو بمفتاح خاطئ يُردّ بـ401 قبل أن يبلغ منطق التنبّؤ. وحفظ الجزاءات بدل المفاتيح الخام يمنع تسرّبها لو سُرِّب متغيّر البيئة أو ملفّ .env عن طريق الخطأ.

الفخّ الكلاسيكيّ: المفتاح في URL

بعض الأدلّة تقترح GET /predict?api_key=xxx أو POST /predict/xxx/. خطأ فادح: المفتاح ينتهي في:

• سجلّات خادم HTTP (nginx، ELB) بشكل واضح.
• سجلّات المتصفّح (history) على أجهزة المستخدمين.
• Referer HTTP الذي يُرسَل تلقائيًّا إلى أيّ رابط خارجيّ من نفس الصفحة.
• شرائح الشاشة التي يلتقطها المستخدمون ويشاركونها بلا انتباه.

القاعدة القطعيّة: الرمز في الترويسة أو في جسم POST مُشفَّرًا (TLS)، لا في URL أبدًا. وهذا يعمل بشرط أن تكون الخدمة كلّها خلف HTTPS، فبدونه المفتاح يمرّ نصًّا صريحًا على الشبكة.

JWT: خطوة تالية لعملاء عديدين

مفتاح API ثابت مناسب لعدد صغير من العملاء الآليّين. لكن لتطبيق ويب أو موبايل بمئات آلاف المستخدمين البشر، نحتاج بنية أدقّ: كلّ مستخدم يُصادق مرّة عبر بوّابة تسجيل الدخول، ويحصل على رمز JWT موقّع رقميًّا يحمل هويّته وصلاحيّاته وتاريخ انتهائه. الخدمة تتحقّق من التوقيع دون استعلام قاعدة البيانات في كلّ طلب.

الشيفرة الأساسيّة بمكتبة python-jose:

from jose import jwt, JWTError

JWT_SECRET = os.environ["JWT_SECRET"]  # مفتاح توقيع مشترك مع خدمة الهويّة
JWT_ALGO = "HS256"


async def utilisateur_jwt(cle: str = Depends(APIKeyHeader(name="Authorization", auto_error=False))) -> dict:
    if not cle or not cle.startswith("Bearer "):
        raise HTTPException(401, "en-tete Authorization Bearer requis")
    token = cle.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGO])
    except JWTError as e:
        raise HTTPException(401, f"jeton invalide : {e}")
    return payload  # يحتوي sub، exp، scopes...


@app.post("/predict")
def predire(req: ChurnRequest, user: dict = Depends(utilisateur_jwt)) -> ChurnResponse:
    if "churn:predict" not in user.get("scopes", []):
        raise HTTPException(403, "غير مصرّح لهذا النطاق")
    ...

المزيّة الحاسمة: انتهاء الصلاحيّة داخل الرمز نفسه (exp). العميل يحصل على رمز صالح ساعةً واحدة، ويجدّده تلقائيًّا. لو سُرِّب، مدّة استغلاله محدودة. وبعكس مفتاح API الثابت، لا نحتاج إعادة نشر الخدمة لإلغاء عميل: يكفي إبطال الرمز في خدمة الهويّة.

دوران المفاتيح: مبدأ عمليّ

مفتاح API ثابت خلال سنوات مخاطرة يقينيّة: مطوّر يغادر، شرائح شاشة، مستودع git خاصّ يصير عامًّا. الحلّ: دوران دوريّ، مثلًا كلّ ثلاثة أشهر. البنية العمليّة:

• الخدمة تقبل مفتاحَين متزامنَين خلال فترة انتقاليّة (APIKEY_V1_SHA256، APIKEY_V2_SHA256).
• العميل يبدّل مفتاحه، يختبر، يُخبر النظام.
• بعد أسبوع، APIKEY_V1_SHA256 تُحذف من متغيّرات البيئة.

بدون بنية دوران، الدوران يتحوّل إلى تعطيل خدمة كلّ ثلاثة أشهر، فيُؤجَّل بلا نهاية، وتتراكم المفاتيح القديمة الحيّة.

تحديد المعدّل: الحائط الأخير

حتّى عميل مُصرَّح له قد يُغرق الخدمة عن غير قصد (حلقة for سيّئة الكتابة) أو عن قصد. الحماية: حدّ أقصى لعدد الطلبات لكلّ مفتاح لكلّ دقيقة. مكتبة slowapi تسهّل ذلك:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiteur = Limiter(key_func=lambda req: req.headers.get("X-API-Key", get_remote_address(req)))
app.state.limiter = limiteur


@app.post("/predict")
@limiteur.limit("60/minute")
def predire(req: ChurnRequest, qui: str = Depends(client_authentifie)) -> ChurnResponse:
    ...

الآن كلّ مفتاح يحصل على ستّين طلبًا في الدقيقة، وطلب واحد وستّون يُردّ بـ429 Too Many Requests. الأرقام تعتمد على السعة: تنبّؤ فرديّ صغير قد يسمح بـ600/دقيقة، مسار دفعات قد يُحصر بـ10/دقيقة. الفكرة هي أنّ العميل الشرعيّ لا يتجاوز الحدّ عادةً، والحدّ يُوقف السلوك الشاذّ بلا كشف عن الأسباب الداخليّة.

للحماية على مستوى IP وليس فقط المفتاح، key_func يُغيَّر ليجمع الاثنين. ولإنتاج جادّ (خدمة موزّعة)، slowapi تدعم Redis كمخزن مشترك بين العمّال.

الاختبارات

ثلاثة اختبارات لا غنى عنها:

def test_predict_401_sans_cle():
    r = client.post("/predict", json=payload_valide)
    assert r.status_code == 401


def test_predict_401_cle_invalide():
    r = client.post("/predict", json=payload_valide, headers={"X-API-Key": "faux"})
    assert r.status_code == 401


def test_predict_429_apres_limite(monkeypatch):
    monkeypatch.setattr(limiteur, "_default_limits", ["3/minute"])
    for _ in range(3):
        r = client.post("/predict", json=payload_valide, headers={"X-API-Key": CLE_TEST})
        assert r.status_code == 200
    r = client.post("/predict", json=payload_valide, headers={"X-API-Key": CLE_TEST})
    assert r.status_code == 429

الخلاصة

• مفتاح API في الترويسة يكفي لخدمة داخليّة بعملاء آليّين قليلين؛ يُخزَّن مُجزَّأً لا خامًا.
• JWT ضروريّ لعدد كبير من العملاء البشر: توقيع رقميّ، انتهاء صلاحيّة داخليّ، لا استعلام قاعدة بيانات في كلّ طلب.
• المفتاح في URL خطأ أمنيّ فادح: يظهر في السجلّات، في التاريخ، في Referer.
• دوران المفاتيح كلّ ثلاثة أشهر ببنية «مفتاحَين متزامنَين» يُبقي الأمن حيًّا بلا تعطيل خدمة.
• تحديد المعدّل بslowapi هو الحائط الأخير: يوقف العميل الشرعيّ المُخطئ والمهاجم على السواء بـ429 قبل استنزاف الخدمة.