معالجة الأخطاء

ينفّذ AuroraSOC طبقات متعددة من معالجة الأخطاء لضمان المرونة في بيئة أمنية حرِجة.

فلسفة معالجة الأخطاء

تنسيق استجابة أخطاء API

تعيد جميع أخطاء API بنية JSON متسقة:

{
  "detail": "Human-readable error message"
}

بالنسبة إلى أخطاء التحقق (Pydantic)، يعيد FastAPI ما يلي:

{
  "detail": [
    {
      "loc": ["body", "severity"],
      "msg": "String should match pattern '^(critical|high|medium|low|info)$'",
      "type": "string_pattern_mismatch"
    }
  ]
}

مرجع رموز الأخطاء

الرمز	المعنى	وقت الحدوث
400	طلب غير صالح	إدخال غير صالح، فشل في التحقق
401	غير مصرح	JWT مفقود/منتهي الصلاحية، أو مفتاح API غير صالح
403	ممنوع	مصادقة صحيحة لكن الصلاحيات غير كافية
404	غير موجود	المورد غير موجود
409	تعارض	مورد مكرر (مثل تكرار نوع+قيمة IOC)
429	طلبات كثيرة جدًا	تم تجاوز حد المعدل
503	الخدمة غير متاحة	تعطل قاعدة البيانات، وضع متدهور
500	خطأ خادم داخلي	استثناء غير مُعالَج

التعامل مع عدم توفر قاعدة البيانات

أكثر سيناريوهات الأخطاء حساسية هو توقف PostgreSQL. يتعامل AuroraSOC مع ذلك بسلاسة:

class DatabaseUnavailable(Exception):
    """Raised when PostgreSQL is not reachable."""
    pass

@app.exception_handler(DatabaseUnavailable)
async def _handle_db_unavailable(request, exc):
    return JSONResponse(
        status_code=503,
        content={"detail": "Database unavailable — running in degraded mode"},
    )

في الوضع المتدهور، تخدم API البيانات من مخازن داخل الذاكرة مملوءة ببيانات تجريبية. يضمن ذلك بقاء لوحة التحكم عاملة للمراقبة حتى أثناء صيانة قاعدة البيانات.

نمط قاطع الدارة

يستخدم المنسق قاطع دارة عند الإرسال إلى خدمات الوكلاء:

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_timeout: float = 30.0):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"    # closed | open | half-open
        self.last_failure_time = 0

    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
            else:
                raise CircuitBreakerOpen("Agent service unavailable")

        try:
            result = await func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise

إعادة المحاولة مع تزايد أُسّي لفترة الانتظار

يتضمن إرسال المهام إلى الوكلاء إعادة محاولة تلقائية:

async def dispatch_with_retry(self, agent_type, prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await self.circuit_breaker.call(
                self._dispatch, agent_type, prompt
            )
        except Exception:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1s, 2s, 4s
            await asyncio.sleep(wait)

أخطاء تحديد معدل الطلبات

عند تجاوز حدود المعدل:

{
  "detail": "Rate limit exceeded. Try again in 42 seconds."
}

تستخدم حدود المعدل عدادات النافذة المنزلقة في Redis:

الفئة	الحد	النافذة
API العامة	100 طلب	دقيقة واحدة
التحقيقات	10 طلبات	دقيقة واحدة
تنفيذ الـ Playbook	5 طلبات	دقيقة واحدة

التسجيل المهيكل للأخطاء

تُسجَّل جميع الأخطاء عبر structlog مع سياق التتبّع الخاص بـ OpenTelemetry:

logger.error(
    "agent_dispatch_failed",
    agent_type="threat_hunter",
    error=str(e),
    case_id=case_id,
    attempt=attempt,
    trace_id=get_current_span().get_span_context().trace_id,
)

ينتج عن ذلك سجلات JSON مهيكلة يمكن ربطها بآثار التتبّع في Jaeger:

{
  "event": "agent_dispatch_failed",
  "agent_type": "threat_hunter",
  "error": "Connection refused",
  "case_id": "550e8400-...",
  "attempt": 2,
  "trace_id": "abc123...",
  "timestamp": "2024-01-15T10:30:00Z",
  "level": "error"
}

معالجة أخطاء WebSocket

تتعامل اتصالات WebSocket مع الأخطاء بصمت لتجنّب تعطيل العملاء الآخرين:

async def broadcast(self, data: dict) -> None:
    dead: list[str] = []
    for cid, ws in self.active.items():
        try:
            await ws.send_json(data)
        except Exception:
            dead.append(cid)  # Mark for removal
    for cid in dead:
        self.active.pop(cid, None)  # Clean up dead connections

عمليات الإرسال الفاشلة لا تُطلق استثناءً؛ بل تُزال الوصلة الميتة بصمت ويستمر البث إلى العملاء السليمين.

معالجة أخطاء ناقل الأحداث

تستخدم مستهلكات Redis Streams معالجة أخطاء قائمة على الإقرار:

async for msg_id, data in consumer.consume():
    try:
        await handler(data)
        await consumer.ack(msg_id)      # Success: acknowledge
    except Exception:
        pass  # Message stays in pending — will be re-delivered

تبقى الرسائل غير المُقَرّ بها في قائمة الإدخالات المعلقة (PEL) ضمن Redis Stream، وتُعاد تسليمها بعد انتهاء مهلة الرؤية الخاصة بمجموعة المستهلك.