
Antigena (Darktrace) → جسر Aruba ClearPass CoA — حجز فوري للمستخدمين/الأجهزة يعتمد على النماذج. بدون نقرات من SOC. بنية سداسية، تغطية اختبار 82%.
جسر Antigena (Darktrace) → Aruba ClearPass CoA — عزل فوري للمستخدمين/الأجهزة مدعوم بالنماذج. بدون نقرات من فريق SOC بين الكشف والاحتواء.
تنفيذ مرجعي مُطَهَّر لنمط تكامل NDR↔NAC إنتاجي تم تشغيله على نطاق قطاع مالي (آلاف نقاط النهاية، مركز عمليات أمنية 24/7). تم استبدال الأجزاء الخاصة بالعميل بإعدادات اصطناعية؛ البنية، تدفق القرار، والأنماط التشغيلية هي الحقيقية.
وعد NDR (Darktrace, ExtraHop, Vectra) هو الكشف في ثوانٍ. لكن الواقع في معظم البنوك: الكشف في ثوانٍ، الاحتواء في ساعات — لأن تسليم المهمة من SOC إلى فرق NAC/جدار الحماية يدوي.
تسد هذه المجموعة الفجوة من خلال ربط Antigena (وحدة الاستجابة المستقلة من Darktrace) بـ Aruba ClearPass عبر REST API الخاص بـ ClearPass. عندما يتم تشغيل نموذج Darktrace فوق عتبة خطورة قابلة للتكوين، تقوم المجموعة بـ:
متوسط زمن الاستجابة من طرف إلى طرف من إطلاق النموذج → تفعيل VLAN العزل: أقل من 5 ثوانٍ.
zero-touch-containment/
├── README.md ← You are here
├── LICENSE
├── .gitignore
├── docs/
│ ├── architecture.md ← Full architecture deep-dive + SOLID trace
│ └── lessons-learned.md ← 10 lessons from running this in prod
│
├── webhook/ ← Inbound HTTP layer (split by SRP)
│ ├── app.py ← FastAPI routes + lifespan only
│ ├── auth.py ← verify_hmac() — HMAC-SHA1 validation
│ ├── replay.py ← ReplayCache — LRU replay protection
│ └── models.py ← AntigenaEvent pydantic schema
│
├── engine/ ← YAML-driven decision engine
│ ├── decision.py ← DecisionEngine (depends on QuarantineReader Protocol)
│ ├── rules.py ← YAML loaders for mapping + allowlist
│ └── models.py ← Action + MappingRule + ActionKind
│
├── clearpass/ ← NAC adapter (implements CoAClient Protocol)
│ ├── client.py ← ClearPassClient — REST CoA-style ops
│ ├── ports.py ← CoAClient Protocol — port for any NAC backend
│ └── auth.py ← OAuth2 TokenCache
│
├── ledger/ ← SQLite ledger (implements 5 ports — ISP applied)
│ ├── store.py ← SqliteLedger — all-in-one implementation
│ ├── ports.py ← EventStore + QuarantineWriter + QuarantineReader
│ │ + ReleaseManager + HealthChecker (segregated)
│ └── schema.py ← SQL DDL constant
│
├── cli/ ← SOC operations CLI
│ └── soc.py ← `ztc release-expired` + planned commands
│
├── config/
│ ├── mapping.example.yaml ← Severity → action mapping
│ └── allowlist.example.yaml ← VIP / never-quarantine list
│
├── deploy/
│ ├── docker-compose.yml
│ ├── Dockerfile
│ └── .env.example
│
├── tests/ ← 60 tests covering every layer
│ ├── test_decision.py
│ ├── test_ledger.py
│ ├── test_webhook_helpers.py
│ ├── test_clearpass_client.py
│ ├── test_protocols.py ← Structural ISP/DIP compliance tests
│ └── fixtures/sample_event.json
│
├── requirements.txt
└── pyproject.toml
git clone https://gitlab.com/zimlama/zero-touch-containment.git
cd zero-touch-containment
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp config/mapping.example.yaml config/mapping.yaml
cp config/allowlist.example.yaml config/allowlist.yaml
cp deploy/.env.example .env # fill in CLEARPASS_HOST, OAUTH creds, HMAC secret
# Run the webhook receiver
uvicorn webhook.app:app --host 0.0.0.0 --port 8080
# In another shell: replay a sample event
curl -X POST http://localhost:8080/antigena \
-H "Content-Type: application/json" \
-H "X-Darktrace-Signature: sha1=$(echo -n @tests/fixtures/sample_event.json | openssl dgst -sha1 -hmac "$HMAC_SECRET" | awk '{print $2}')" \
--data @tests/fixtures/sample_event.json
يقوم الـ webhook بالتحقق من HMAC-SHA1، ويشغل محرك القرار مقابل mapping.yaml، ويقوم إما بـ:
┌──────────────┐ 1. webhook ┌──────────────────┐ 2. validate ┌──────────────────┐
│ Darktrace │ ──────────────▶ │ Webhook │ ───────────────▶ │ Decision │
│ Antigena │ HMAC-SHA1 │ receiver │ parse + auth │ engine │
│ fires model │ │ (FastAPI) │ │ (YAML-driven) │
└──────────────┘ └──────────────────┘ └─────────┬────────┘
│
▼
3. resolve action
(allowlist + rate limit)
│
┌───────────────────────┬───────────────────────┼────────────────────────┐
▼ ▼ ▼ ▼
┌──────────────┐ ┌────────────────┐ ┌──────────────┐ ┌─────────────┐
│ ClearPass │ │ SQLite │ │ Slack/Teams │ │ SIEM │
│ REST API │ │ ledger │ │ notification │ │ (structured │
│ - role swap │ │ - state │ │ │ │ logs) │
│ - disconnect │ │ - auto-release │ │ │ │ │
└──────────────┘ └────────────────┘ └──────────────┘ └─────────────┘
انظر docs/architecture.md للتفصيل الكامل.
| الطبقة | الأدوات |
|---|---|
| اللغة | Python 3.11+ |
| الويب | FastAPI + Uvicorn (مستقبل webhook) |
| عميل HTTP | httpx (غير متزامن) + tenacity (إعادة المحاولة مع التباطؤ) |
| المصادقة | HMAC-SHA1 وارد (Darktrace) · OAuth2 client_credentials صادر (ClearPass) |
| الإعدادات | YAML — تخطيط الخطورة → الإجراء + قائمة السماح |
| الحالة | SQLite + WAL — سجل العزل + الإصدار التلقائي |
| التسجيل | structlog — إخراج JSON لاستيعاب SIEM |
| الاختبار | pytest + respx (محاكاة httpx) + إعدادات مسجلة |
| النشر | Docker Compose، ملائم لجهاز افتراضي واحد |