From f76b6cad140e7a227e20ed1195dfe1306ee03fd1 Mon Sep 17 00:00:00 2001 From: Peter van Arkel Date: Thu, 21 May 2026 01:19:08 +0200 Subject: [PATCH] (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. --- src/pobsync_backend/forms.py | 14 +++++++- .../pobsync_backend/retention_plan.html | 7 ++++ src/pobsync_backend/tests/test_views.py | 32 +++++++++++++++++++ src/pobsync_backend/views.py | 23 ++++++++++++- 4 files changed, 74 insertions(+), 2 deletions(-) diff --git a/src/pobsync_backend/forms.py b/src/pobsync_backend/forms.py index 8be9435..ec2f6cc 100644 --- a/src/pobsync_backend/forms.py +++ b/src/pobsync_backend/forms.py @@ -249,12 +249,18 @@ class RetentionApplyForm(forms.Form): kind = forms.ChoiceField(choices=(("scheduled", "Scheduled"), ("manual", "Manual"), ("all", "All"))) protect_bases = forms.BooleanField(required=False) max_delete = forms.IntegerField(min_value=0, initial=10) + confirm_delete_count = forms.IntegerField(min_value=0) 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.expected_delete_count = expected_delete_count super().__init__(*args, **kwargs) 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: 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.") 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): cron_expr = forms.CharField( diff --git a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html index 0d94a11..ccacb38 100644 --- a/src/pobsync_backend/templates/pobsync_backend/retention_plan.html +++ b/src/pobsync_backend/templates/pobsync_backend/retention_plan.html @@ -123,6 +123,13 @@
{{ apply_form.confirm_host.help_text }}
+
+ {{ apply_form.confirm_delete_count.errors }} + + {{ apply_form.confirm_delete_count }} +
{{ apply_form.confirm_delete_count.help_text }}
+
+
diff --git a/src/pobsync_backend/tests/test_views.py b/src/pobsync_backend/tests/test_views.py index 2dbdf40..c6e9cc5 100644 --- a/src/pobsync_backend/tests/test_views.py +++ b/src/pobsync_backend/tests/test_views.py @@ -1412,6 +1412,8 @@ class ViewTests(TestCase): self.assertContains(response, "newest") self.assertContains(response, "Would Delete") 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: self.client.force_login(self.staff_user) @@ -1546,6 +1548,7 @@ class ViewTests(TestCase): "kind": "scheduled", "max_delete": "1", "confirm_host": host.host, + "confirm_delete_count": "1", }, follow=True, ) @@ -1569,6 +1572,7 @@ class ViewTests(TestCase): "kind": "scheduled", "max_delete": "1", "confirm_host": "wrong", + "confirm_delete_count": "1", }, follow=True, ) @@ -1577,6 +1581,34 @@ class ViewTests(TestCase): self.assertContains(response, "Retention apply confirmation is invalid.") 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: self.client.force_login(self.staff_user) host = HostConfig.objects.create(host="web-01", address="web-01.example.test") diff --git a/src/pobsync_backend/views.py b/src/pobsync_backend/views.py index 44d7715..cf469a4 100644 --- a/src/pobsync_backend/views.py +++ b/src/pobsync_backend/views.py @@ -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, "apply_form": RetentionApplyForm( host_name=host_config.host, + expected_delete_count=delete_count, initial={ "kind": kind, "protect_bases": protect_bases, "max_delete": delete_count, + "confirm_delete_count": delete_count, }, ), } @@ -556,7 +558,26 @@ def host_retention_plan(request, host: str): @require_POST def apply_host_retention(request, host: str): 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(): messages.error(request, "Retention apply confirmation is invalid.") return redirect("host_retention_plan", host=host_config.host)