mirror of
https://github.com/gmfricke/cmmc_audit.git
synced 2026-06-28 19:53:04 -06:00
303 lines
11 KiB
Python
303 lines
11 KiB
Python
from django.conf import settings
|
|
from django.db import models
|
|
from django.utils import timezone
|
|
|
|
from .resolvers import actor_label, resolve_target_label, resolve_value_labels
|
|
|
|
|
|
ALLOCATION_TRANSITION_ACTIONS = {
|
|
"allocation_status_changed",
|
|
"allocation_disabled",
|
|
"renewal_requested",
|
|
"renewal_approved",
|
|
"allocation_renewed",
|
|
}
|
|
|
|
|
|
EVIDENCE_AREA_LABELS = {
|
|
"AC": "Access Control",
|
|
"AU": "Audit Logging",
|
|
"CM": "Configuration Management",
|
|
"IA": "Identification and Authentication",
|
|
}
|
|
|
|
SOURCE_LABELS = {
|
|
"runtime": "Runtime",
|
|
"django_admin": "Django admin",
|
|
"coldfront_signal": "ColdFront signal",
|
|
"coldfront_workflow": "ColdFront workflow",
|
|
"reconstructed_django_admin_log": "Reconstructed Django admin log",
|
|
"django_admin_log": "Django admin log",
|
|
"coldfront_history": "ColdFront history table",
|
|
}
|
|
|
|
|
|
class AuditEvent(models.Model):
|
|
class EvidenceCategory(models.TextChoices):
|
|
AC_ACCESS_CONTROL = "AC-Access-Control", "AC - Access Control"
|
|
AU_AUDIT_ACCOUNTABILITY = "AU-Audit-Accountability", "AU - Audit and Accountability"
|
|
AC_AU = "AC-Access-Control; AU-Audit-Accountability", "AC/AU - Access and Audit"
|
|
CM_CONFIGURATION_MANAGEMENT = "CM-Configuration-Management", "CM - Configuration Management"
|
|
IA_IDENTIFICATION_AUTHENTICATION = (
|
|
"IA-Identification-Authentication",
|
|
"IA - Identification and Authentication",
|
|
)
|
|
|
|
class Source(models.TextChoices):
|
|
RUNTIME = "runtime", "Runtime"
|
|
DJANGO_ADMIN = "django_admin", "Django admin"
|
|
COLDFRONT_SIGNAL = "coldfront_signal", "ColdFront signal"
|
|
COLDFRONT_WORKFLOW = "coldfront_workflow", "ColdFront workflow"
|
|
RECONSTRUCTED_DJANGO_ADMIN_LOG = (
|
|
"reconstructed_django_admin_log",
|
|
"Reconstructed Django admin log",
|
|
)
|
|
DJANGO_ADMIN_LOG = "django_admin_log", "Django admin log"
|
|
COLDFRONT_HISTORY = "coldfront_history", "ColdFront history"
|
|
|
|
class Action(models.TextChoices):
|
|
ADMIN_ADDITION = "admin_addition", "Admin addition"
|
|
ADMIN_CHANGE = "admin_change", "Admin change"
|
|
ADMIN_DELETION = "admin_deletion", "Admin deletion"
|
|
USER_PI_UPGRADED = "user_pi_upgraded", "User upgraded to PI"
|
|
PI_STATUS_REVOKED = "pi_status_revoked", "PI status revoked"
|
|
USER_ADMIN_PRIVILEGES_CHANGED = "user_admin_privileges_changed", "User admin privileges changed"
|
|
ALLOCATION_REQUESTED = "allocation_requested", "Allocation created/requested"
|
|
ALLOCATION_CHANGE_REQUESTED = "allocation_change_requested", "Allocation change requested"
|
|
ALLOCATION_STATUS_CHANGED = "allocation_status_changed", "Allocation status changed"
|
|
ALLOCATION_DISABLED = "allocation_disabled", "Allocation disabled"
|
|
RENEWAL_REQUESTED = "renewal_requested", "Renewal requested"
|
|
RENEWAL_APPROVED = "renewal_approved", "Renewal approved"
|
|
ALLOCATION_RENEWED = "allocation_renewed", "Allocation renewed"
|
|
PROJECT_USER_ADDED = "project_user_added", "Project user added"
|
|
PROJECT_USER_REMOVED = "project_user_removed", "Project user removed"
|
|
PROJECT_USER_ROLE_CHANGED = "project_user_role_changed", "Project user role changed"
|
|
PROJECT_CREATED = "project_created", "Project created"
|
|
PROJECT_ARCHIVED = "project_archived", "Project archived"
|
|
PROJECT_DELETED = "project_deleted", "Project deleted"
|
|
PROJECT_STATUS_CHANGED = "project_status_changed", "Project status changed"
|
|
PROJECT_PI_CHANGED = "project_pi_changed", "Project PI changed"
|
|
PROJECT_REVIEW_FORCED = "project_review_forced", "Project review forced"
|
|
PROJECT_REVIEW_SUBMITTED = "project_review_submitted", "Project review submitted"
|
|
PROJECT_REVIEW_COMPLETED = "project_review_completed", "Project review completed"
|
|
PROJECT_REVIEW_STATUS_CHANGED = "project_review_status_changed", "Project review status changed"
|
|
RESOURCE_CREATED = "resource_created", "Resource created"
|
|
RESOURCE_CHANGED = "resource_changed", "Resource changed"
|
|
RESOURCE_DELETED = "resource_deleted", "Resource deleted"
|
|
|
|
timestamp = models.DateTimeField(auto_now_add=True)
|
|
event_time = models.DateTimeField(default=timezone.now)
|
|
|
|
actor = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="cmmc_audit_events",
|
|
)
|
|
|
|
action = models.CharField(max_length=64, choices=Action.choices)
|
|
evidence_category = models.CharField(
|
|
max_length=128,
|
|
choices=EvidenceCategory.choices,
|
|
blank=True,
|
|
)
|
|
control_family = models.CharField(max_length=16, blank=True)
|
|
target_type = models.CharField(max_length=128, blank=True)
|
|
target_id = models.CharField(max_length=128, blank=True)
|
|
target_repr = models.CharField(max_length=255, blank=True)
|
|
|
|
old_values = models.JSONField(default=dict, blank=True)
|
|
new_values = models.JSONField(default=dict, blank=True)
|
|
message = models.TextField(blank=True)
|
|
source = models.CharField(max_length=64, choices=Source.choices, default=Source.RUNTIME)
|
|
source_id = models.CharField(max_length=128, blank=True)
|
|
is_reconstructed = models.BooleanField(default=False)
|
|
request_path = models.CharField(max_length=255, blank=True)
|
|
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-timestamp"]
|
|
indexes = [
|
|
models.Index(fields=["event_time"]),
|
|
models.Index(fields=["action"]),
|
|
models.Index(fields=["evidence_category"]),
|
|
models.Index(fields=["source"]),
|
|
models.Index(fields=["actor"]),
|
|
models.Index(fields=["target_type", "target_id"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.event_time} {self.action_detail_display} {self.target_summary}"
|
|
|
|
@property
|
|
def action_display(self):
|
|
return self.get_action_display()
|
|
|
|
@property
|
|
def action_detail_display(self):
|
|
if self.action not in ALLOCATION_TRANSITION_ACTIONS:
|
|
return self.action_display
|
|
|
|
transition = self.transition_display
|
|
if self.action == self.Action.ALLOCATION_STATUS_CHANGED:
|
|
if self._new_status() == "Active" and self._old_status() == "New":
|
|
label = "Allocation approved"
|
|
else:
|
|
label = "Allocation status changed"
|
|
elif self.action == self.Action.ALLOCATION_DISABLED:
|
|
if self._new_status() == "Revoked":
|
|
label = "Allocation revoked"
|
|
else:
|
|
label = "Allocation disabled"
|
|
elif self.action == self.Action.RENEWAL_APPROVED:
|
|
label = "Allocation renewal approved"
|
|
elif self.action == self.Action.ALLOCATION_RENEWED:
|
|
label = "Allocation renewed"
|
|
else:
|
|
label = self.action_display
|
|
|
|
if transition:
|
|
return f"{label} ({transition})"
|
|
return label
|
|
|
|
@property
|
|
def transition_display(self):
|
|
if self.action not in ALLOCATION_TRANSITION_ACTIONS:
|
|
return ""
|
|
|
|
old_status = self._old_status()
|
|
new_status = self._new_status()
|
|
if old_status and new_status:
|
|
return f"{old_status} \u2192 {new_status}"
|
|
|
|
if self.action == self.Action.ALLOCATION_RENEWED:
|
|
days = self._renewal_days()
|
|
if days:
|
|
return f"+{days} days"
|
|
|
|
return ""
|
|
|
|
def _old_status(self):
|
|
return self._value_from(self.old_values, "status")
|
|
|
|
def _new_status(self):
|
|
return self._value_from(self.new_values, "status")
|
|
|
|
def _renewal_days(self):
|
|
extension = self._value_from(self.new_values, "end_date_extension")
|
|
if isinstance(extension, int) and extension > 0:
|
|
return extension
|
|
if isinstance(extension, str) and extension.isdigit() and int(extension) > 0:
|
|
return int(extension)
|
|
|
|
old_end_date = self._value_from(self.old_values, "end_date")
|
|
new_end_date = self._value_from(self.new_values, "end_date")
|
|
if not old_end_date or not new_end_date:
|
|
return None
|
|
try:
|
|
old_date = timezone.datetime.fromisoformat(old_end_date).date()
|
|
new_date = timezone.datetime.fromisoformat(new_end_date).date()
|
|
except (TypeError, ValueError):
|
|
return None
|
|
days = (new_date - old_date).days
|
|
return days if days > 0 else None
|
|
|
|
def _value_from(self, values, key):
|
|
if not isinstance(values, dict):
|
|
return None
|
|
value = values.get(key)
|
|
if value not in (None, ""):
|
|
return value
|
|
return values.get(f"allocation_{key}")
|
|
|
|
@property
|
|
def actor_display(self):
|
|
return actor_label(self.actor)
|
|
|
|
@property
|
|
def target_summary(self):
|
|
resolved = resolve_target_label(self.target_type, self.target_id)
|
|
if resolved and not (
|
|
self.target_type == "project.project"
|
|
and self.target_repr
|
|
and "(not found)" in resolved
|
|
):
|
|
return resolved
|
|
if self.target_repr:
|
|
return self.target_repr
|
|
if self.target_type and self.target_id:
|
|
return f"{self.target_type} #{self.target_id}"
|
|
return self.target_type or "Unknown target"
|
|
|
|
@property
|
|
def evidence_summary(self):
|
|
if not self.evidence_category:
|
|
return ""
|
|
if self.control_family:
|
|
return f"{self.control_family}: {self.evidence_category}"
|
|
return self.evidence_category
|
|
|
|
@property
|
|
def evidence_area_display(self):
|
|
if self.control_family:
|
|
areas = [
|
|
EVIDENCE_AREA_LABELS.get(family.strip(), family.strip())
|
|
for family in self.control_family.split(",")
|
|
if family.strip()
|
|
]
|
|
if areas:
|
|
return " + ".join(areas)
|
|
|
|
if self.evidence_category:
|
|
category = str(self.evidence_category)
|
|
if category.startswith("AC-"):
|
|
return EVIDENCE_AREA_LABELS["AC"]
|
|
if category.startswith("AU-"):
|
|
return EVIDENCE_AREA_LABELS["AU"]
|
|
if category.startswith("CM-"):
|
|
return EVIDENCE_AREA_LABELS["CM"]
|
|
if category.startswith("IA-"):
|
|
return EVIDENCE_AREA_LABELS["IA"]
|
|
return category
|
|
|
|
return "Unclassified evidence"
|
|
|
|
@property
|
|
def source_display(self):
|
|
return SOURCE_LABELS.get(self.source, f"Unknown source ({self.source or 'blank'})")
|
|
|
|
@property
|
|
def evidence_display(self):
|
|
return f"{self.evidence_area_display}\nSource: {self.source_display}"
|
|
|
|
@property
|
|
def reconstructed_display(self):
|
|
if not self.is_reconstructed:
|
|
return "No"
|
|
if self.source_id:
|
|
return f"Yes ({self.source_id})"
|
|
return "Yes"
|
|
|
|
@property
|
|
def old_values_display(self):
|
|
return resolve_value_labels(self.old_values)
|
|
|
|
@property
|
|
def new_values_display(self):
|
|
return resolve_value_labels(self.new_values)
|
|
|
|
|
|
class BackfillRun(models.Model):
|
|
name = models.CharField(max_length=128, unique=True)
|
|
started = models.DateTimeField(auto_now_add=True)
|
|
completed = models.DateTimeField(null=True, blank=True)
|
|
created_events = models.PositiveIntegerField(default=0)
|
|
dry_run = models.BooleanField(default=False)
|
|
notes = models.TextField(blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-started"]
|
|
|
|
def __str__(self):
|
|
return self.name
|