4f5370db89
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>
130 lines
5.4 KiB
Python
130 lines
5.4 KiB
Python
"""Phase 11: Knowledge Management models — Playbooks + Lessons Learned."""
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import (
|
|
Boolean, Column, DateTime, ForeignKey,
|
|
Index, Integer, String, Text, UniqueConstraint,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import UUID, JSONB
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.database import Base
|
|
|
|
|
|
# ── Playbooks ──────────────────────────────────────────────────────────────────
|
|
|
|
class Playbook(Base):
|
|
"""
|
|
Structured runbook for a specific technique and playbook type.
|
|
|
|
playbook_type: attack | detect | investigate | respond | hunt
|
|
One playbook per (technique, type). Edits increment ``version``
|
|
and save a snapshot to ``PlaybookVersion``.
|
|
"""
|
|
__tablename__ = "playbooks"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
technique_id = Column(
|
|
UUID(as_uuid=True), ForeignKey("techniques.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
playbook_type = Column(String(32), nullable=False) # attack/detect/investigate/respond/hunt
|
|
title = Column(String(255), nullable=False)
|
|
content = Column(Text, nullable=False, default="")
|
|
version = Column(Integer, default=1, nullable=False)
|
|
tools = Column(JSONB, default=list) # list of tool name strings
|
|
prerequisites = Column(JSONB, default=list) # list of prerequisite strings
|
|
created_by = Column(
|
|
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
|
|
)
|
|
updated_by = Column(
|
|
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
|
|
)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
is_active = Column(Boolean, default=True)
|
|
|
|
# Relationships
|
|
technique = relationship("Technique", foreign_keys=[technique_id])
|
|
creator = relationship("User", foreign_keys=[created_by])
|
|
updater = relationship("User", foreign_keys=[updated_by])
|
|
versions = relationship(
|
|
"PlaybookVersion", back_populates="playbook",
|
|
cascade="all, delete-orphan",
|
|
order_by="PlaybookVersion.version.desc()",
|
|
)
|
|
|
|
__table_args__ = (
|
|
UniqueConstraint("technique_id", "playbook_type", name="uq_playbook_technique_type"),
|
|
Index("ix_playbooks_technique_id", "technique_id"),
|
|
Index("ix_playbooks_type", "playbook_type"),
|
|
)
|
|
|
|
|
|
class PlaybookVersion(Base):
|
|
"""Immutable snapshot of a playbook at a given version number."""
|
|
|
|
__tablename__ = "playbook_versions"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
playbook_id = Column(
|
|
UUID(as_uuid=True), ForeignKey("playbooks.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
version = Column(Integer, nullable=False)
|
|
title = Column(String(255), nullable=False)
|
|
content = Column(Text, nullable=False, default="")
|
|
tools = Column(JSONB, default=list)
|
|
prerequisites = Column(JSONB, default=list)
|
|
changed_by = Column(
|
|
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
|
|
)
|
|
change_note = Column(String(500), nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
playbook = relationship("Playbook", back_populates="versions")
|
|
changer = relationship("User", foreign_keys=[changed_by])
|
|
|
|
__table_args__ = (
|
|
Index("ix_pb_versions_playbook_id", "playbook_id"),
|
|
Index("ix_pb_versions_version", "playbook_id", "version"),
|
|
)
|
|
|
|
|
|
# ── Lessons Learned ────────────────────────────────────────────────────────────
|
|
|
|
class LessonLearned(Base):
|
|
"""
|
|
Immutable post-mortem record linked to a test, campaign, attack-path or
|
|
created manually.
|
|
|
|
severity: critical | high | medium | low | info
|
|
entity_type: test | campaign | attack_path | manual
|
|
"""
|
|
__tablename__ = "lessons_learned"
|
|
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
title = Column(String(255), nullable=False)
|
|
what_happened = Column(Text, nullable=False, default="")
|
|
root_cause = Column(Text, nullable=False, default="")
|
|
fix_applied = Column(Text, nullable=True)
|
|
severity = Column(String(16), nullable=False, default="medium")
|
|
entity_type = Column(String(32), nullable=False, default="manual")
|
|
entity_id = Column(UUID(as_uuid=True), nullable=True)
|
|
technique_ids = Column(JSONB, default=list) # list of UUID strings
|
|
tags = Column(JSONB, default=list)
|
|
created_by = Column(
|
|
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
|
|
)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
is_active = Column(Boolean, default=True) # soft-delete (admin only)
|
|
|
|
creator = relationship("User", foreign_keys=[created_by])
|
|
|
|
__table_args__ = (
|
|
Index("ix_ll_entity", "entity_type", "entity_id"),
|
|
Index("ix_ll_severity", "severity"),
|
|
Index("ix_ll_created_by", "created_by"),
|
|
)
|