feat(knowledge): Phase 11 — Knowledge Management (Playbooks + Lessons Learned) [FASE-11]
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

- Playbooks: versioned Markdown runbooks per technique × type (attack/detect/investigate/respond/hunt)
- PlaybookVersion: immutable snapshots on every update; restore to any previous version
- LessonLearned: post-mortem records linked to tests/campaigns/attack-paths or manual
- Alembic migration b037know (raw SQL, idempotent, no PostgreSQL enums)
- Router /api/v1/knowledge: 14 endpoints for playbooks + lessons + stats
- Pydantic validators for playbook_type, severity, entity_type (422 on invalid)
- Knowledge stats endpoint: totals + breakdown by severity and playbook type
- Soft-delete on both resources; include_inactive filter for admin recovery
- QA script: 70+ tests across CRUD, versioning, filtering, auth, soft-delete, regression

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
kitos
2026-05-20 13:39:05 +02:00
parent 080ce56de7
commit 4f5370db89
9 changed files with 1329 additions and 0 deletions

View File

@@ -0,0 +1,133 @@
"""Phase 11: Lesson Learned service."""
from datetime import datetime
from typing import List, Optional
from uuid import UUID
from sqlalchemy.orm import Session
from app.domain.errors import DomainError
from app.models.knowledge import LessonLearned
def _get_or_404(db: Session, ll_id: UUID) -> LessonLearned:
ll = db.query(LessonLearned).filter(
LessonLearned.id == ll_id,
LessonLearned.is_active == True,
).first()
if not ll:
raise DomainError(f"Lesson Learned {ll_id} not found", status_code=404)
return ll
# ── Read ──────────────────────────────────────────────────────────────────────
def get_lesson_learned(db: Session, ll_id: UUID) -> LessonLearned:
return _get_or_404(db, ll_id)
def list_lessons_learned(
db: Session,
entity_type: Optional[str] = None,
entity_id: Optional[UUID] = None,
severity: Optional[str] = None,
tag: Optional[str] = None,
technique_id: Optional[str] = None,
include_inactive: bool = False,
) -> List[LessonLearned]:
q = db.query(LessonLearned)
if not include_inactive:
q = q.filter(LessonLearned.is_active == True)
if entity_type:
q = q.filter(LessonLearned.entity_type == entity_type)
if entity_id:
q = q.filter(LessonLearned.entity_id == entity_id)
if severity:
q = q.filter(LessonLearned.severity == severity)
if tag:
# JSONB contains operator — filter lessons that have this tag
q = q.filter(LessonLearned.tags.contains([tag]))
if technique_id:
q = q.filter(LessonLearned.technique_ids.contains([technique_id]))
return q.order_by(LessonLearned.created_at.desc()).all()
# ── Create ────────────────────────────────────────────────────────────────────
def create_lesson_learned(db: Session, data: dict, user_id: UUID) -> LessonLearned:
ll = LessonLearned(
title = data["title"],
what_happened = data.get("what_happened", ""),
root_cause = data.get("root_cause", ""),
fix_applied = data.get("fix_applied"),
severity = data.get("severity", "medium"),
entity_type = data.get("entity_type", "manual"),
entity_id = data.get("entity_id"),
technique_ids = data.get("technique_ids") or [],
tags = data.get("tags") or [],
created_by = user_id,
)
db.add(ll)
db.commit()
db.refresh(ll)
return ll
# ── Update ────────────────────────────────────────────────────────────────────
def update_lesson_learned(db: Session, ll_id: UUID, data: dict, user_id: UUID) -> LessonLearned:
ll = _get_or_404(db, ll_id)
for field in ("title", "what_happened", "root_cause", "fix_applied",
"severity", "technique_ids", "tags"):
if field in data and data[field] is not None:
setattr(ll, field, data[field])
ll.updated_at = datetime.utcnow()
db.commit()
db.refresh(ll)
return ll
# ── Delete (soft) ─────────────────────────────────────────────────────────────
def delete_lesson_learned(db: Session, ll_id: UUID, user_id: UUID) -> None:
"""Soft-delete — admin-only enforcement is done at the router level."""
ll = _get_or_404(db, ll_id)
ll.is_active = False
ll.updated_at = datetime.utcnow()
db.commit()
# ── Stats ─────────────────────────────────────────────────────────────────────
def get_knowledge_stats(db: Session) -> dict:
"""Summary counts for the knowledge base dashboard."""
from app.models.knowledge import Playbook
total_playbooks = db.query(Playbook).filter(Playbook.is_active == True).count()
total_lessons = db.query(LessonLearned).filter(LessonLearned.is_active == True).count()
severity_counts: dict = {}
for sev in ("critical", "high", "medium", "low", "info"):
severity_counts[sev] = (
db.query(LessonLearned)
.filter(LessonLearned.is_active == True, LessonLearned.severity == sev)
.count()
)
playbook_type_counts: dict = {}
from app.schemas.knowledge_schema import VALID_PLAYBOOK_TYPES
for ptype in VALID_PLAYBOOK_TYPES:
playbook_type_counts[ptype] = (
db.query(Playbook)
.filter(Playbook.is_active == True, Playbook.playbook_type == ptype)
.count()
)
return {
"total_playbooks": total_playbooks,
"total_lessons": total_lessons,
"lessons_by_severity": severity_counts,
"playbooks_by_type": playbook_type_counts,
}