96 lines
3.9 KiB
Python
96 lines
3.9 KiB
Python
from django.core.management.base import BaseCommand, CommandError
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from coldfront.plugins.cmmc_audit.backfill import RUN_NAME, run_backfill
|
|
from coldfront.plugins.cmmc_audit.models import BackfillRun
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Reconstruct best-effort historical CMMC audit events from existing records."
|
|
|
|
def add_arguments(self, parser):
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Report events that would be reconstructed without writing rows.",
|
|
)
|
|
mode.add_argument(
|
|
"--commit",
|
|
action="store_true",
|
|
help="Create reconstructed audit rows.",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Rerun after a completed backfill. Duplicate source/action checks still apply.",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
dry_run = options["dry_run"]
|
|
commit = options["commit"]
|
|
force = options["force"]
|
|
|
|
if force and dry_run:
|
|
raise CommandError("--force is only valid with --commit")
|
|
|
|
existing_run = BackfillRun.objects.filter(name=RUN_NAME).first()
|
|
if commit and existing_run and existing_run.completed and not force:
|
|
self.stdout.write(
|
|
self.style.WARNING(
|
|
f"Backfill '{RUN_NAME}' completed at {existing_run.completed}; use --force to rerun."
|
|
)
|
|
)
|
|
return
|
|
|
|
if dry_run:
|
|
report = run_backfill(dry_run=True)
|
|
self._print_report(report, dry_run=True)
|
|
return
|
|
|
|
with transaction.atomic():
|
|
if existing_run is None:
|
|
backfill_run = BackfillRun.objects.create(name=RUN_NAME, dry_run=False)
|
|
else:
|
|
backfill_run = existing_run
|
|
backfill_run.completed = None
|
|
backfill_run.created_events = 0
|
|
backfill_run.dry_run = False
|
|
backfill_run.notes = "Forced rerun." if force else ""
|
|
backfill_run.save(update_fields=["completed", "created_events", "dry_run", "notes"])
|
|
|
|
report = run_backfill(dry_run=False)
|
|
backfill_run.completed = timezone.now()
|
|
backfill_run.created_events = report.created_events
|
|
backfill_run.notes = self._summary_notes(report, forced=force)
|
|
backfill_run.save(update_fields=["completed", "created_events", "notes"])
|
|
|
|
self._print_report(report, dry_run=False)
|
|
|
|
def _summary_notes(self, report, *, forced):
|
|
prefix = "Forced rerun. " if forced else ""
|
|
return (
|
|
f"{prefix}Created {report.created_events} events; "
|
|
f"duplicates skipped {report.duplicates}; ambiguous skipped {report.ambiguous}."
|
|
)
|
|
|
|
def _print_report(self, report, *, dry_run):
|
|
self.stdout.write("CMMC audit historical reconstruction report")
|
|
self.stdout.write(f"Mode: {'dry-run' if dry_run else 'commit'}")
|
|
self.stdout.write("Records examined by source:")
|
|
for source, count in sorted(report.examined.items()):
|
|
self.stdout.write(f" {source}: {count}")
|
|
|
|
self.stdout.write(f"Events that would be created: {report.would_create_events}")
|
|
self.stdout.write(f"Events created: {report.created_events}")
|
|
self.stdout.write(f"Duplicates skipped: {report.duplicates}")
|
|
self.stdout.write(f"Ambiguous records skipped: {report.ambiguous}")
|
|
|
|
self.stdout.write("Counts by action:")
|
|
for action, count in sorted(report.by_action.items(), key=lambda item: str(item[0])):
|
|
self.stdout.write(f" {action}: {count}")
|
|
|
|
self.stdout.write("Counts by evidence category:")
|
|
for category, count in sorted(report.by_evidence.items()):
|
|
self.stdout.write(f" {category}: {count}")
|