(feature) Require delete count confirmation for retention apply

Make manual retention application more explicit by requiring operators to
confirm both the host name and the current number of planned deletions.

This reduces the risk of applying a stale or misunderstood retention plan
when the delete set changes between review and confirmation.
This commit is contained in:
2026-05-21 01:19:08 +02:00
parent 90e293facd
commit f76b6cad14
4 changed files with 74 additions and 2 deletions

View File

@@ -249,12 +249,18 @@ class RetentionApplyForm(forms.Form):
kind = forms.ChoiceField(choices=(("scheduled", "Scheduled"), ("manual", "Manual"), ("all", "All"))) kind = forms.ChoiceField(choices=(("scheduled", "Scheduled"), ("manual", "Manual"), ("all", "All")))
protect_bases = forms.BooleanField(required=False) protect_bases = forms.BooleanField(required=False)
max_delete = forms.IntegerField(min_value=0, initial=10) max_delete = forms.IntegerField(min_value=0, initial=10)
confirm_delete_count = forms.IntegerField(min_value=0)
confirm_host = forms.CharField() confirm_host = forms.CharField()
def __init__(self, *args, host_name: str, **kwargs) -> None: def __init__(self, *args, host_name: str, expected_delete_count: int | None = None, **kwargs) -> None:
self.host_name = host_name self.host_name = host_name
self.expected_delete_count = expected_delete_count
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.fields["confirm_host"].help_text = f"Type {host_name} to confirm deletion." self.fields["confirm_host"].help_text = f"Type {host_name} to confirm deletion."
if expected_delete_count is not None:
self.fields["confirm_delete_count"].help_text = (
f"Type {expected_delete_count} to confirm the current number of planned deletions."
)
def clean_confirm_host(self) -> str: def clean_confirm_host(self) -> str:
value = self.cleaned_data["confirm_host"].strip() value = self.cleaned_data["confirm_host"].strip()
@@ -262,6 +268,12 @@ class RetentionApplyForm(forms.Form):
raise forms.ValidationError(f"Type {self.host_name} to confirm.") raise forms.ValidationError(f"Type {self.host_name} to confirm.")
return value return value
def clean_confirm_delete_count(self) -> int:
value = self.cleaned_data["confirm_delete_count"]
if self.expected_delete_count is not None and value != self.expected_delete_count:
raise forms.ValidationError(f"Type {self.expected_delete_count} to confirm the delete count.")
return value
class ScheduleConfigForm(forms.ModelForm): class ScheduleConfigForm(forms.ModelForm):
cron_expr = forms.CharField( cron_expr = forms.CharField(

View File

@@ -123,6 +123,13 @@
<div class="helptext">{{ apply_form.confirm_host.help_text }}</div> <div class="helptext">{{ apply_form.confirm_host.help_text }}</div>
</div> </div>
<div class="field">
{{ apply_form.confirm_delete_count.errors }}
<label for="{{ apply_form.confirm_delete_count.id_for_label }}">Confirm delete count</label>
{{ apply_form.confirm_delete_count }}
<div class="helptext">{{ apply_form.confirm_delete_count.help_text }}</div>
</div>
<div class="actions"> <div class="actions">
<button type="submit">Apply retention</button> <button type="submit">Apply retention</button>
</div> </div>

View File

@@ -1412,6 +1412,8 @@ class ViewTests(TestCase):
self.assertContains(response, "newest") self.assertContains(response, "newest")
self.assertContains(response, "Would Delete") self.assertContains(response, "Would Delete")
self.assertContains(response, "outside retention policy") self.assertContains(response, "outside retention policy")
self.assertContains(response, "Confirm delete count")
self.assertContains(response, "Type 1 to confirm the current number of planned deletions.")
def test_retention_plan_warns_when_scheduled_prune_limit_is_exceeded(self) -> None: def test_retention_plan_warns_when_scheduled_prune_limit_is_exceeded(self) -> None:
self.client.force_login(self.staff_user) self.client.force_login(self.staff_user)
@@ -1546,6 +1548,7 @@ class ViewTests(TestCase):
"kind": "scheduled", "kind": "scheduled",
"max_delete": "1", "max_delete": "1",
"confirm_host": host.host, "confirm_host": host.host,
"confirm_delete_count": "1",
}, },
follow=True, follow=True,
) )
@@ -1569,6 +1572,7 @@ class ViewTests(TestCase):
"kind": "scheduled", "kind": "scheduled",
"max_delete": "1", "max_delete": "1",
"confirm_host": "wrong", "confirm_host": "wrong",
"confirm_delete_count": "1",
}, },
follow=True, follow=True,
) )
@@ -1577,6 +1581,34 @@ class ViewTests(TestCase):
self.assertContains(response, "Retention apply confirmation is invalid.") self.assertContains(response, "Retention apply confirmation is invalid.")
self.assertEqual(SnapshotRecord.objects.count(), 1) self.assertEqual(SnapshotRecord.objects.count(), 1)
def test_retention_apply_rejects_mismatched_delete_count_confirmation(self) -> None:
self.client.force_login(self.staff_user)
host = HostConfig.objects.create(
host="web-01",
address="web-01.example.test",
retention_daily=0,
retention_weekly=0,
retention_monthly=0,
retention_yearly=0,
)
self._snapshot(host, "20260518-021500Z__OLDSNAP")
self._snapshot(host, "20260519-021500Z__NEWSNAP")
response = self.client.post(
reverse("apply_host_retention", args=[host.host]),
{
"kind": "scheduled",
"max_delete": "1",
"confirm_host": host.host,
"confirm_delete_count": "0",
},
follow=True,
)
self.assertRedirects(response, reverse("host_retention_plan", args=[host.host]))
self.assertContains(response, "Retention apply confirmation is invalid.")
self.assertEqual(SnapshotRecord.objects.count(), 2)
def test_retention_apply_requires_post(self) -> None: def test_retention_apply_requires_post(self) -> None:
self.client.force_login(self.staff_user) self.client.force_login(self.staff_user)
host = HostConfig.objects.create(host="web-01", address="web-01.example.test") host = HostConfig.objects.create(host="web-01", address="web-01.example.test")

View File

@@ -542,10 +542,12 @@ def host_retention_plan(request, host: str):
"scheduled_prune_exceeded": scheduled_prune_limit is not None and delete_count > scheduled_prune_limit, "scheduled_prune_exceeded": scheduled_prune_limit is not None and delete_count > scheduled_prune_limit,
"apply_form": RetentionApplyForm( "apply_form": RetentionApplyForm(
host_name=host_config.host, host_name=host_config.host,
expected_delete_count=delete_count,
initial={ initial={
"kind": kind, "kind": kind,
"protect_bases": protect_bases, "protect_bases": protect_bases,
"max_delete": delete_count, "max_delete": delete_count,
"confirm_delete_count": delete_count,
}, },
), ),
} }
@@ -556,7 +558,26 @@ def host_retention_plan(request, host: str):
@require_POST @require_POST
def apply_host_retention(request, host: str): def apply_host_retention(request, host: str):
host_config = get_object_or_404(HostConfig, host=host) host_config = get_object_or_404(HostConfig, host=host)
form = RetentionApplyForm(request.POST, host_name=host_config.host) raw_kind = request.POST.get("kind", "scheduled")
raw_protect_bases = request.POST.get("protect_bases") in {"1", "true", "on", "yes"}
expected_delete_count = None
if raw_kind in {"scheduled", "manual", "all"}:
try:
plan = run_sql_retention_plan(
host=host_config.host,
kind=raw_kind,
protect_bases=raw_protect_bases,
)
except PobsyncError as exc:
messages.error(request, str(exc))
return redirect("host_retention_plan", host=host_config.host)
expected_delete_count = len(plan.get("delete") or [])
form = RetentionApplyForm(
request.POST,
host_name=host_config.host,
expected_delete_count=expected_delete_count,
)
if not form.is_valid(): if not form.is_valid():
messages.error(request, "Retention apply confirmation is invalid.") messages.error(request, "Retention apply confirmation is invalid.")
return redirect("host_retention_plan", host=host_config.host) return redirect("host_retention_plan", host=host_config.host)