feat: Phase 1 - Data models and migrations (T-004 to T-009)

Implements all database models for the Aegis platform with full
Alembic migration support.

Models created:
- User: Authentication with role-based access control
- Technique: MITRE ATT&CK techniques with coverage status tracking
- Test: Security tests with validation workflow (draft/review/validated)
- Evidence: File metadata for test evidence (stored in MinIO)
- IntelItem: Threat intelligence items linked to techniques
- AuditLog: System-wide audit trail with JSONB details

Enumerations:
- TechniqueStatus: not_evaluated, in_progress, validated, partial, etc.
- TestState: draft, in_review, validated, rejected
- TestResult: detected, not_detected, partially_detected

Services:
- audit_service.py: log_action() helper for audit logging

All models include proper foreign key relationships and PostgreSQL
enum types are managed correctly in migrations (create/drop).
This commit is contained in:
2026-02-06 12:26:26 +01:00
parent b479acdea0
commit ec65991ac1
18 changed files with 561 additions and 1 deletions

View File

@@ -0,0 +1,13 @@
# Import all models here so Alembic can detect them
from app.models.user import User
from app.models.technique import Technique
from app.models.test import Test
from app.models.evidence import Evidence
from app.models.intel import IntelItem
from app.models.audit import AuditLog
from app.models.enums import TechniqueStatus, TestState, TestResult
__all__ = [
"User", "Technique", "Test", "Evidence", "IntelItem", "AuditLog",
"TechniqueStatus", "TestState", "TestResult"
]

View File

@@ -0,0 +1,29 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from app.database import Base
class AuditLog(Base):
"""
Audit log model for tracking all system actions.
Records user actions, entity changes, and system events
for security auditing and compliance purposes.
"""
__tablename__ = "audit_logs"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
action = Column(String, nullable=False)
entity_type = Column(String, nullable=True)
entity_id = Column(String, nullable=True)
timestamp = Column(DateTime, default=datetime.utcnow)
details = Column(JSONB, nullable=True)
# Relationships
user = relationship("User")

View File

@@ -0,0 +1,23 @@
import enum
class TechniqueStatus(str, enum.Enum):
not_evaluated = "not_evaluated"
in_progress = "in_progress"
validated = "validated"
partial = "partial"
not_covered = "not_covered"
review_required = "review_required"
class TestState(str, enum.Enum):
draft = "draft"
in_review = "in_review"
validated = "validated"
rejected = "rejected"
class TestResult(str, enum.Enum):
detected = "detected"
not_detected = "not_detected"
partially_detected = "partially_detected"

View File

@@ -0,0 +1,30 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from app.database import Base
class Evidence(Base):
"""
Evidence model for storing file metadata associated with tests.
Files are stored in MinIO, and this model tracks the file location,
integrity hash, and upload metadata.
"""
__tablename__ = "evidences"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
test_id = Column(UUID(as_uuid=True), ForeignKey("tests.id"), nullable=False)
file_name = Column(String, nullable=False)
file_path = Column(String, nullable=False) # Path in MinIO
sha256_hash = Column(String, nullable=False)
uploaded_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
uploaded_at = Column(DateTime, default=datetime.utcnow)
# Relationships
test = relationship("Test", back_populates="evidences")
uploader = relationship("User", foreign_keys=[uploaded_by])

View File

@@ -0,0 +1,29 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from app.database import Base
class IntelItem(Base):
"""
Intelligence item model for tracking threat intelligence related to techniques.
Stores URLs and metadata from automated intel scans that may indicate
new attack variations or detection bypasses for specific techniques.
"""
__tablename__ = "intel_items"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
technique_id = Column(UUID(as_uuid=True), ForeignKey("techniques.id"), nullable=True)
url = Column(String, nullable=False)
title = Column(String, nullable=True)
source = Column(String, nullable=True)
detected_at = Column(DateTime, default=datetime.utcnow)
reviewed = Column(Boolean, default=False)
# Relationships
technique = relationship("Technique")

View File

@@ -0,0 +1,39 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, Text, Boolean, DateTime, Enum
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from app.database import Base
from app.models.enums import TechniqueStatus
class Technique(Base):
"""
MITRE ATT&CK Technique model.
Represents an attack technique from the MITRE ATT&CK framework,
including its coverage status and associated tests.
"""
__tablename__ = "techniques"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
mitre_id = Column(String, unique=True, nullable=False) # e.g., "T1059.001"
name = Column(String, nullable=False)
description = Column(Text, nullable=True)
tactic = Column(String, nullable=True)
platforms = Column(JSONB, nullable=True, default=[])
mitre_version = Column(String, nullable=True)
mitre_last_modified = Column(DateTime, nullable=True)
is_subtechnique = Column(Boolean, default=False)
parent_mitre_id = Column(String, nullable=True)
status_global = Column(
Enum(TechniqueStatus, name="techniquestatus"),
default=TechniqueStatus.not_evaluated
)
review_required = Column(Boolean, default=False)
last_review_date = Column(DateTime, nullable=True)
# Relationships
tests = relationship("Test", back_populates="technique")

View File

@@ -0,0 +1,40 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Enum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from app.database import Base
from app.models.enums import TestState, TestResult
class Test(Base):
"""
Test model representing a security test for a MITRE ATT&CK technique.
Each test documents an attempt to validate coverage of a specific technique,
including the procedure, tools used, and outcome.
"""
__tablename__ = "tests"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
technique_id = Column(UUID(as_uuid=True), ForeignKey("techniques.id"), nullable=False)
name = Column(String, nullable=False)
description = Column(Text, nullable=True)
platform = Column(String, nullable=True)
procedure_text = Column(Text, nullable=True)
tool_used = Column(String, nullable=True)
execution_date = Column(DateTime, nullable=True)
created_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
result = Column(Enum(TestResult, name="testresult"), nullable=True)
state = Column(Enum(TestState, name="teststate"), default=TestState.draft)
validated_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
validated_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
# Relationships
technique = relationship("Technique", back_populates="tests")
evidences = relationship("Evidence", back_populates="test")
creator = relationship("User", foreign_keys=[created_by])
validator = relationship("User", foreign_keys=[validated_by])

View File

@@ -0,0 +1,31 @@
import uuid
from datetime import datetime
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.dialects.postgresql import UUID
from app.database import Base
class User(Base):
"""
User model for authentication and authorization.
Possible roles:
- admin: Full system access
- red_tech: Red team technician - can create and edit tests
- blue_tech: Blue team technician - can create and edit tests
- red_lead: Red team lead - can validate tests
- blue_lead: Blue team lead - can validate tests
- viewer: Read-only access (default)
"""
__tablename__ = "users"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
username = Column(String, unique=True, nullable=False)
email = Column(String, nullable=True)
hashed_password = Column(String, nullable=False)
role = Column(String, nullable=False, default="viewer")
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
last_login = Column(DateTime, nullable=True)

View File

View File

@@ -0,0 +1,33 @@
from sqlalchemy.orm import Session
from app.models.audit import AuditLog
def log_action(
db: Session,
user_id,
action: str,
entity_type: str = None,
entity_id: str = None,
details: dict = None
):
"""
Log an action to the audit log.
Args:
db: Database session
user_id: UUID of the user performing the action (can be None for system actions)
action: Description of the action (e.g., "create_test", "validate_technique")
entity_type: Type of entity affected (e.g., "technique", "test", "user")
entity_id: ID of the entity affected
details: Additional details as a dictionary (stored as JSONB)
"""
log = AuditLog(
user_id=user_id,
action=action,
entity_type=entity_type,
entity_id=str(entity_id) if entity_id else None,
details=details,
)
db.add(log)
db.commit()