Prune MAS evidence local runs after 30 days, as the owners accepted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 4541@bnt-lap001
Assistant-Session: 2099b3aa-d7d9-49a9-ab4e-fa8b7d0e39e8
This commit is contained in:
codex 2026-11-03 08:35:45 +01:00
parent 4277088e1c
commit efb17b4984
2 changed files with 57 additions and 1 deletions

View file

@ -13,6 +13,10 @@ freshness
Fails when the latest verified upload is older than STALE_HOURS, its
receipt is not verified, or the local ciphertext no longer matches it.
After each verified upload, local run directories older than
LOCAL_RETENTION_DAYS are pruned. The current run is never pruned, and
nothing at the provider is ever deleted.
Credentials come only from /etc/mas-backup/primary.env through the systemd
EnvironmentFile. No TEM or SMTP credential and no age identity is read here.
"""
@ -22,6 +26,8 @@ from __future__ import annotations
import argparse
import json
import os
import re
import shutil
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
@ -36,6 +42,8 @@ STATE = Path("/var/lib/mas-evidence-backup")
RECIPIENT_FILE = Path("/etc/mas-backup/age-recipient")
LATEST = "latest-offsite.json"
STALE_HOURS = 26
LOCAL_RETENTION_DAYS = 30 # owner-accepted 2026-11-03
_RUN_NAME = re.compile(r"\A\d{8}T\d{6}Z\Z")
APPLICATION = "mas"
@ -120,7 +128,25 @@ def backup(state=STATE, capture=None, client_factory=None, now=None):
archive.checkpoint(state / LATEST, {**receipt, "run": stamp, "capture": public})
receipt = publish_run(run_dir, captured["captured_at"], client, save_latest)
return {**receipt, "run": stamp, "capture": public}
pruned = prune_local(state, stamp, now or datetime.now(UTC))
return {**receipt, "run": stamp, "capture": public,
"local_retention_days": LOCAL_RETENTION_DAYS, "local_runs_pruned": pruned}
def prune_local(state, keep, now, days=LOCAL_RETENTION_DAYS):
"""Remove dated run directories older than `days`, never `keep`. Local only."""
cutoff = now - timedelta(days=days)
removed = []
for entry in sorted(state.iterdir()):
if entry.name == keep or not _RUN_NAME.fullmatch(entry.name):
continue
if entry.is_symlink() or not entry.is_dir() or entry.parent != state:
continue
stamp = datetime.strptime(entry.name, "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC)
if stamp < cutoff:
shutil.rmtree(entry)
removed.append(entry.name)
return removed
def freshness(state=STATE, now=None):