feat(backup): check a backup against what is in place, changing nothing (#5823)
CI / lint (push) Successful in 3s
CI / integration (push) Successful in 1m23s
CI / unit (push) Successful in 1m37s
CI / publish (push) Successful in 1m8s

#5823 "Check a backup: compare what it would restore with what is in
place, without restoring".

- steward/backup/check.py: pg_restore writes the backup's rows as COPY
  text with no database involved. The rows in place go out through COPY in
  the same form. Tables are compared on their primary keys, over the columns
  both sides have, and grouped by area.
- Sealed values are opened with each side's own key and compared as
  plaintext digests, so a secret sealed twice still matches. They are never
  shown. A backup whose key can't open its own secrets fails the check.
- What Steward records as it runs (last-seen times, a schedule's last run,
  backup history, scraped UniFi fields) is left out of the comparison:
  contents.ACTIVITY, STATE_SETTINGS and HAND_SET_ONLY.
- Settings → Backups:
  - a Check button on each sent backup, which fetches it from the bucket
    (new s3.get_file);
  - "Check a file" to upload one, with the size limit raised before the
    CSRF check reads the form;
  - a results page.
- Each backup Steward sends checks itself once written, and its history
  row says how that went.
- New CLI command `steward check-backup FILE`. CI's image smoke step checks a
  real backup against its source and expects a match.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-15 10:48:48 -04:00
co-authored by Claude Opus 5.5
parent e6da0ff299
commit e8b75143f1
16 changed files with 1141 additions and 10 deletions
+481
View File
@@ -0,0 +1,481 @@
"""Checking a backup: what it would restore, compared with what is in place,
without restoring anything (#5823).
The backup's rows come out of pg_restore as COPY text, with no database
involved; the rows in place go out through COPY in the same form. They are
compared table by table on their primary keys, over the columns both sides
have. Nothing is written.
What is compared is what a restore would replace: every table whose rows
the backup carries and this Steward keeps (contents.py). Columns that are
Steward's record of what it last saw (contents.ACTIVITY) and settings it
writes as it runs (contents.STATE_SETTINGS) are left out of the comparison:
they change by the minute and say nothing about whether the backup is good.
A sealed value (crypto's ``enc:v1:`` tokens) is opened with its own side's
key and compared as plaintext, so a secret sealed twice still matches. It
is never shown: the report says only that it differs. A backup whose key
doesn't open its own sealed values is reported, since its secrets would not
come back.
"""
from __future__ import annotations
import asyncio
import hashlib
import io
import json
import re
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from .contents import (ACTIVITY, ACTIVITY_EVERYWHERE, AREAS, EXCLUDED, HAND_SET_ONLY,
STATE_SETTINGS)
from .dump import BackupError, read_backup
from .restore import _async_url
CHECK_DEADLINE_SECONDS = 5 * 60
# Names shown for a row, the first a table has; else its primary key.
LABEL_COLUMNS = ("name", "username", "key", "title", "widget_key", "label", "mac")
# A setting's two values are shown when they fit; longer ones are cut.
SHOWN_CHARS = 120
_TOKEN = re.compile(r"enc:v1:[A-Za-z0-9_\-=]+")
_COPY = re.compile(r'^COPY ([^ ]+) \((.*)\) FROM stdin;$')
@dataclass
class RowDiff:
label: str
columns: list[str]
secret: bool = False
backup_value: str | None = None # shown only for a setting that isn't secret
here_value: str | None = None
@dataclass
class TableCheck:
table: str
same: int = 0
differ: list[RowDiff] = field(default_factory=list)
only_backup: list[str] = field(default_factory=list)
only_here: list[str] = field(default_factory=list)
note: str = "" # a table one side lacks, columns one side lacks
@property
def changes(self) -> int:
return len(self.differ) + len(self.only_backup) + len(self.only_here)
@dataclass
class AreaCheck:
name: str
tables: list[TableCheck]
@property
def same(self) -> int:
return sum(t.same for t in self.tables)
@property
def changes(self) -> int:
return sum(t.changes for t in self.tables)
@dataclass
class CheckReport:
name: str
manifest: dict
schema: str # "same", "older" or "newer"
backup_heads: list[str]
here_heads: list[str]
key_same: bool
sealed: int # sealed values in the backup
unopened: int # of those, how many its own key couldn't open
areas: list[AreaCheck]
checked_at: str = ""
@property
def made_at(self) -> datetime | None:
try:
return datetime.fromisoformat(self.manifest.get("created_at") or "")
except ValueError:
return None
@property
def rows(self) -> int:
return sum(a.same + a.changes for a in self.areas)
@property
def changes(self) -> int:
return sum(a.changes for a in self.areas)
@property
def ok(self) -> bool:
"""The backup reads back whole: its key opens its secrets. Whether it
matches what is in place is `changes`; a difference isn't a fault."""
return self.unopened == 0
def summary(self) -> str:
if self.unopened:
return (f"The backup's key doesn't open {self.unopened} of its {self.sealed} sealed "
f"value{'s' if self.sealed != 1 else ''}: those secrets wouldn't come back.")
if not self.changes:
tables = sum(len(a.tables) for a in self.areas)
return f"Matches what is in place: {self.rows} rows across {tables} table{'s' if tables != 1 else ''}."
parts = [f"{a.changes} in {a.name}" for a in self.areas if a.changes]
return (f"{self.changes} difference{'s' if self.changes != 1 else ''} from what is in place: "
+ ", ".join(parts) + ".")
def as_entry(self) -> dict:
"""What a backup's history row keeps of its own check."""
return {"ok": self.ok, "changes": self.changes, "rows": self.rows, "summary": self.summary()}
# ── COPY text ───────────────────────────────────────────────────────────────
_ESCAPES = {ord("b"): b"\b", ord("f"): b"\f", ord("n"): b"\n", ord("r"): b"\r",
ord("t"): b"\t", ord("v"): b"\v"}
def unescape(field: bytes) -> str | None:
"""One field of COPY's text format: \\N is NULL, backslash escapes are
undone (named, octal \\ooo and hex \\xhh)."""
if field == b"\\N":
return None
if b"\\" not in field:
return field.decode("utf-8", errors="replace")
out = bytearray()
i, n = 0, len(field)
while i < n:
c = field[i]
if c != 0x5C or i + 1 == n:
out.append(c)
i += 1
continue
nxt = field[i + 1]
if nxt in _ESCAPES:
out += _ESCAPES[nxt]
i += 2
elif 0x30 <= nxt <= 0x37:
j = i + 1
while j < n and j < i + 4 and 0x30 <= field[j] <= 0x37:
j += 1
out.append(int(field[i + 1:j], 8) & 0xFF)
i = j
elif nxt == ord("x") and i + 2 < n and chr(field[i + 2]) in "0123456789abcdefABCDEF":
j = i + 2
while j < n and j < i + 4 and chr(field[j]) in "0123456789abcdefABCDEF":
j += 1
out.append(int(field[i + 2:j], 16))
i = j
else:
out.append(nxt)
i += 2
return bytes(out).decode("utf-8", errors="replace")
def copy_rows(data: bytes) -> list[list[str | None]]:
"""COPY text rows (no header, no terminator) as lists of fields."""
return [[unescape(f) for f in line.split(b"\t")]
for line in data.split(b"\n") if line and line != b"\\."]
def _ident(name: str) -> str:
name = name.strip()
return name[1:-1].replace('""', '"') if name.startswith('"') and name.endswith('"') else name
def script_tables(script: bytes) -> dict[str, tuple[list[str], list[list[str | None]]]]:
"""Each table's (columns, rows) from a pg_restore --data-only script."""
out: dict[str, tuple[list[str], list[list[str | None]]]] = {}
lines = iter(script.split(b"\n"))
for line in lines:
m = _COPY.match(line.decode("utf-8", errors="replace"))
if not m:
continue
table = _ident(m.group(1).split(".")[-1])
columns = [_ident(c) for c in m.group(2).split(",")]
body = []
for row in lines:
if row == b"\\.":
break
body.append(row)
out[table] = (columns, copy_rows(b"\n".join(body)))
return out
# ── Comparing ───────────────────────────────────────────────────────────────
def _fernet(key: str):
from steward.core.crypto import _make_fernet
return _make_fernet(key) if key else None
class _Opener:
"""Opens one side's sealed values, counting what it couldn't."""
def __init__(self, key: str):
self.fernet = _fernet(key)
self.seen = 0
self.failed = 0
def open(self, value: str) -> tuple[str, bool]:
"""`value` with each sealed token replaced by a digest of its
plaintext (never the plaintext itself), and whether it held one."""
if "enc:v1:" not in value:
return value, False
def plain(m: re.Match) -> str:
self.seen += 1
try:
opened = self.fernet.decrypt(m.group(0)[len("enc:v1:"):].encode("ascii"))
except Exception:
self.failed += 1
return "\x00unopened:" + m.group(0)
return "\x00sealed:" + hashlib.sha256(opened).hexdigest()
return _TOKEN.sub(plain, value), True
def _offset(v: str) -> str:
return re.sub(r"([+-]\d\d)$", r"\1:00", v)
def normalise(value: str | None, data_type: str) -> object:
"""A value as compared: an instant for a timestamp with zone (the two
sides may print it in different zones), parsed JSON, a float."""
if value is None:
return None
try:
if data_type == "timestamp with time zone":
return datetime.fromisoformat(_offset(value)).astimezone(timezone.utc)
if data_type in ("json", "jsonb"):
return json.dumps(json.loads(value), sort_keys=True)
if data_type in ("double precision", "real", "numeric"):
return float(value)
except ValueError:
pass
return value
def _shown(value: str | None) -> str | None:
if value is None:
return None
return value if len(value) <= SHOWN_CHARS else value[:SHOWN_CHARS - 1] + "…"
@dataclass
class _Side:
columns: list[str]
rows: list[list[str | None]]
def compare_table(table: str, backup: _Side, here: _Side, *, types: dict[str, str],
key: list[str], openers: tuple[_Opener, _Opener],
labels: dict[str, dict[str, str]]) -> TableCheck:
"""One table's rows, backup against here."""
result = TableCheck(table)
skip = ACTIVITY.get(table, frozenset()) | ACTIVITY_EVERYWHERE
common = [c for c in backup.columns if c in here.columns and c not in skip]
lacking = sorted((set(backup.columns) ^ set(here.columns)) - skip)
if lacking:
result.note = "Columns only one side has: " + ", ".join(lacking) + "."
key = [c for c in key if c in common] or common
hand_set = HAND_SET_ONLY.get(table)
def index(side: _Side, opener: _Opener) -> dict[tuple, dict[str, object]]:
at = {c: side.columns.index(c) for c in common}
out = {}
for raw in side.rows:
row: dict[str, object] = {}
sealed: set[str] = set()
for c in common:
v = raw[at[c]]
if isinstance(v, str):
v, was = opener.open(v)
if was:
sealed.add(c)
row[c] = normalise(v, types.get(c, ""))
if table == "app_settings" and row.get("key") in STATE_SETTINGS:
continue
if hand_set and not any(raw[side.columns.index(c)] for c in hand_set if c in side.columns):
continue
row["\x00sealed"] = sealed
row["\x00raw"] = {c: raw[at[c]] for c in common}
out[tuple(str(row[c]) for c in key)] = row
return out
theirs, ours = index(backup, openers[0]), index(here, openers[1])
def label(row: dict) -> str:
raw = row["\x00raw"]
if table == "host_agent_registrations" and raw.get("host_id"):
return labels.get("hosts", {}).get(raw["host_id"], raw["host_id"])
if table == "ansible_target_groups":
return (f"{labels.get('ansible_targets', {}).get(raw.get('target_id'), raw.get('target_id'))} in "
f"{labels.get('ansible_groups', {}).get(raw.get('group_id'), raw.get('group_id'))}")
for c in LABEL_COLUMNS:
if raw.get(c):
return str(raw[c])
return " / ".join(str(raw.get(c)) for c in key)
for k in sorted(theirs.keys() | ours.keys()):
b, h = theirs.get(k), ours.get(k)
if h is None:
result.only_backup.append(label(b))
elif b is None:
result.only_here.append(label(h))
else:
cols = [c for c in common if b[c] != h[c]]
if not cols:
result.same += 1
continue
secret = bool((b["\x00sealed"] | h["\x00sealed"]) & set(cols))
diff = RowDiff(label(b), cols, secret)
if table == "app_settings" and not secret:
diff.backup_value = _shown(b["\x00raw"].get("value_json"))
diff.here_value = _shown(h["\x00raw"].get("value_json"))
result.differ.append(diff)
return result
def _labels(tables: dict[str, _Side]) -> dict[str, dict[str, str]]:
"""id → name for the tables other rows are named through."""
out: dict[str, dict[str, str]] = {}
for t in ("hosts", "ansible_targets", "ansible_groups"):
side = tables.get(t)
if side and "id" in side.columns and "name" in side.columns:
i, n = side.columns.index("id"), side.columns.index("name")
out[t] = {r[i]: r[n] for r in side.rows}
return out
# ── Reading both sides ──────────────────────────────────────────────────────
async def _backup_rows(dump: Path) -> dict[str, tuple[list[str], list[list[str | None]]]]:
"""Every table's rows in the dump, as pg_restore would load them. Run
without a database: pg_restore writes the script and nothing connects."""
try:
proc = await asyncio.create_subprocess_exec(
"pg_restore", "--data-only", "--no-owner", "--no-privileges", "--file=-", str(dump),
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
except FileNotFoundError:
raise BackupError("pg_restore isn't installed here.") from None
try:
out, err = await asyncio.wait_for(proc.communicate(), CHECK_DEADLINE_SECONDS)
except TimeoutError:
proc.kill()
await proc.wait()
raise BackupError(f"Reading the backup didn't finish within "
f"{CHECK_DEADLINE_SECONDS // 60} minutes.") from None
if proc.returncode != 0:
lines = [ln for ln in err.decode(errors="replace").splitlines() if ln.strip()]
raise BackupError("The backup can't be read: " + (lines[-1] if lines else
f"pg_restore exited {proc.returncode}."))
return script_tables(out)
async def _here(database_url: str, tables: list[str]):
"""The rows in place for `tables`, with each column's type, each table's
primary key and the migration heads."""
engine = create_async_engine(_async_url(database_url))
try:
async with engine.connect() as conn:
types: dict[str, dict[str, str]] = {}
order: dict[str, list[str]] = {}
for t, c, dt in (await conn.execute(text(
"SELECT table_name, column_name, data_type FROM information_schema.columns "
"WHERE table_schema = 'public' AND table_name = ANY(:t) "
"ORDER BY table_name, ordinal_position"), {"t": tables})).all():
types.setdefault(t, {})[c] = dt
order.setdefault(t, []).append(c)
keys: dict[str, list[str]] = {}
for t, c in (await conn.execute(text(
"SELECT cl.relname, a.attname FROM pg_index i "
"JOIN pg_class cl ON cl.oid = i.indrelid "
"JOIN pg_namespace ns ON ns.oid = cl.relnamespace "
"JOIN pg_attribute a ON a.attrelid = cl.oid AND a.attnum = ANY(i.indkey) "
"WHERE i.indisprimary AND ns.nspname = 'public' AND cl.relname = ANY(:t) "
"ORDER BY cl.relname, a.attnum"), {"t": tables})).all():
keys.setdefault(t, []).append(c)
heads = sorted((await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars())
raw = (await conn.get_raw_connection()).driver_connection
rows: dict[str, _Side] = {}
for t in tables:
if t not in order:
continue
buf = io.BytesIO()
await raw.copy_from_table(t, columns=order[t], output=buf, format="text")
rows[t] = _Side(order[t], copy_rows(buf.getvalue()))
return rows, types, keys, heads
finally:
await engine.dispose()
def known_revisions(plugin_dirs: list[Path] | None = None) -> set[str]:
"""Every migration revision this Steward has, core and plugins: a backup
head not among them was made by a newer Steward."""
import steward
roots = [Path(steward.__file__).parent / "migrations" / "versions"]
for d in plugin_dirs or [Path(steward.__file__).parent.parent / "plugins"]:
if Path(d).is_dir():
roots += [p / "migrations" / "versions" for p in Path(d).iterdir()]
found = set()
for root in roots:
if root.is_dir():
for f in root.glob("*.py"):
m = re.search(r'^revision\s*(?::\s*str\s*)?=\s*["\']([^"\']+)["\']',
f.read_text(errors="replace"), re.M)
if m:
found.add(m.group(1))
return found
def schema_of(backup_heads: list[str], here_heads: list[str], known: set[str]) -> str:
if sorted(backup_heads) == sorted(here_heads):
return "same"
return "newer" if any(h not in known for h in backup_heads) else "older"
async def check_backup(path: Path, database_url: str, here_key: str, *,
plugin_dirs: list[Path] | None = None, name: str = "") -> CheckReport:
"""Compare the backup at `path` with the database at `database_url`.
Raises BackupError when the file isn't a backup or can't be read."""
with tempfile.TemporaryDirectory(prefix="steward-check-") as tmp:
info, dump, backup_key = await asyncio.to_thread(read_backup, path, Path(tmp))
theirs = await _backup_rows(dump)
tables = sorted(t for t in theirs if t not in EXCLUDED)
here, types, keys, heads = await _here(database_url, tables)
openers = (_Opener(backup_key), _Opener(here_key))
backup_sides = {t: _Side(*theirs[t]) for t in tables}
labels = _labels(backup_sides)
for t, names in _labels(here).items(): # a name in place wins
labels[t] = {**labels.get(t, {}), **names}
checks: dict[str, TableCheck] = {}
for t in tables:
if t not in here:
checks[t] = TableCheck(t, only_backup=[f"{len(theirs[t][1])} rows"],
note="This Steward has no such table.")
continue
checks[t] = compare_table(t, backup_sides[t], here[t], types=types.get(t, {}),
key=keys.get(t, []), openers=openers, labels=labels)
areas = []
placed: set[str] = set()
for area, members in AREAS:
found = [checks[t] for t in members if t in checks]
placed |= set(members)
if found:
areas.append(AreaCheck(area, found))
rest = [checks[t] for t in tables if t not in placed]
if rest:
areas.append(AreaCheck("Other", rest))
backup_heads = sorted(info.get("migration_heads") or [])
return CheckReport(
name=name or path.name, manifest=info,
schema=schema_of(backup_heads, heads, known_revisions(plugin_dirs)),
backup_heads=backup_heads, here_heads=heads, key_same=backup_key == here_key,
sealed=openers[0].seen, unopened=openers[0].failed, areas=areas,
checked_at=datetime.now(timezone.utc).isoformat())
+52
View File
@@ -59,3 +59,55 @@ EXCLUDED: frozenset[str] = frozenset({
"unifi_events", "unifi_fw_rules", "unifi_networks", "unifi_port_forwards",
"unifi_speedtest_results", "unifi_wan_stats", "unifi_wlan_stats",
})
# ── What a check compares (#5823) ────────────────────────────────────────────
# Checking a backup compares its kept rows with the ones in place. Some columns
# of kept tables are Steward's own record of what it last saw, not something
# anyone set up: they change by the minute, so comparing them would call every
# backup out of date. They are left out of the comparison (a restore still
# carries them). `updated_at` is bookkeeping wherever it appears.
ACTIVITY: dict[str, frozenset[str]] = {
"api_tokens": frozenset({"last_used_at"}),
# What the agent reports about its host; the enrolment (host_id,
# token_hash, token_created_at) is what is compared.
"host_agent_registrations": frozenset({
"agent_version", "kernel", "distro", "arch", "host_ip", "last_seen_at",
"clock_offset_s", "clock_checked_at", "updates_manager", "updates_pending",
"updates_security", "updates_checked_at", "updates_error", "reboot_required",
"reboot_required_since", "reboot_packages", "maint_helper", "drive_health",
"drive_health_at"}),
"host_maintenance_plans": frozenset({"last_slot_at"}),
"monitors": frozenset({"last_checked_at"}),
"ansible_schedules": frozenset({"last_run_at", "last_run_id", "last_error"}),
"snmp_devices": frozenset({"last_polled_at", "last_ok_at", "last_error", "missing"}),
"prometheus_targets": frozenset({"last_scraped_at", "last_ok_at", "last_error", "last_kept",
"last_dropped", "last_capped"}),
# Scraped from the controller; the name and note are what someone set.
"unifi_known_clients": frozenset({"hostname", "ip", "mac_vendor", "is_blocked", "first_seen",
"last_seen", "scraped_at"}),
}
ACTIVITY_EVERYWHERE: frozenset[str] = frozenset({"updated_at"})
# Settings Steward writes as it runs: a record of its own work, not setup.
STATE_SETTINGS: frozenset[str] = frozenset({
"backup.history", "backup.last_attempt_at", "backup.last_success_at",
"reports.last_sent_at", "cleanup.last_run",
})
# A network client only counts once someone has named it or noted it; the
# rest are the controller's, and come and go as devices do.
HAND_SET_ONLY: dict[str, tuple[str, ...]] = {"unifi_known_clients": ("name", "note")}
# How a check groups what it found, in the order it shows them. A table
# not named here is shown under "Other".
AREAS: tuple[tuple[str, tuple[str, ...]], ...] = (
("Settings", ("app_settings",)),
("Users and API tokens", ("users", "api_tokens")),
("Hosts and agents", ("hosts", "host_agent_registrations", "host_maintenance_plans")),
("Monitors", ("monitors",)),
("Alert rules and maintenance windows", ("alert_rules", "maintenance_windows")),
("Ansible", ("ansible_targets", "ansible_groups", "ansible_target_groups", "ansible_schedules")),
("Dashboards", ("dashboards", "dashboard_widgets", "dashboard_share_tokens")),
("Plugins", ("snmp_devices", "prometheus_targets", "prometheus_widget_templates",
"unifi_known_clients")),
)
+46 -2
View File
@@ -22,7 +22,7 @@ from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from .dump import BackupError, make_backup
from .s3 import S3Error, S3Target, list_objects, put_file
from .s3 import S3Error, S3Target, get_file, list_objects, put_file
logger = logging.getLogger(__name__)
@@ -133,7 +133,8 @@ async def run_backup(app, by: str = "schedule", user_id: str | None = None) -> d
with tempfile.TemporaryDirectory(prefix="steward-backup-") as tmp:
made = await make_backup(db_url, app.config["SECRET_KEY"], heads, Path(tmp), now=at)
await put_file(target, made.name, made.path)
entry.update(ok=True, name=made.name, size=made.size)
entry.update(ok=True, name=made.name, size=made.size)
entry["check"] = await _self_check(app, made.path, made.name)
except (BackupError, S3Error) as exc:
entry["error"] = str(exc)
except Exception as exc: # recorded and logged: a run never takes the scheduler down
@@ -148,6 +149,49 @@ async def run_backup(app, by: str = "schedule", user_id: str | None = None) -> d
return entry
async def _self_check(app, path: Path, name: str) -> dict:
"""A backup just sent, read back and compared with what it was made from
(#5823): proof the file restores, kept on its history row. A check that
can't run says why; the backup was sent either way."""
try:
report = await check_file(app, path, name)
except Exception as exc:
logger.warning("Backup %s couldn't be checked: %s", name, exc)
return {"ok": False, "changes": 0, "rows": 0, "summary": f"Couldn't check it: {exc}"}
return report.as_entry()
async def check_file(app, path: Path, name: str = ""):
"""Compare a backup file with this Steward's database (check.py)."""
from .check import check_backup
return await check_backup(path, app.config["DATABASE_URL"], app.config["SECRET_KEY"],
plugin_dirs=[Path(d) for d in app.config["PLUGIN_DIRS"]],
name=name or path.name)
def is_backup_name(name: str) -> bool:
"""A file name Steward's backups have, and nothing else: no path, no
other object in the bucket."""
import re
return bool(re.fullmatch(r"steward-\d{8}T\d{6}Z\.tar", name or ""))
async def check_stored(app, name: str):
"""Fetch one of this Steward's backups from the bucket and check it.
Raises BackupError or S3Error with what went wrong."""
from steward.core.settings import get_all_settings
if not is_backup_name(name):
raise BackupError(f"{name!r} isn't the name of a Steward backup.")
async with app.db_sessionmaker() as db:
target = target_from(await get_all_settings(db))
if target is None:
raise BackupError("The bucket isn't set up: fill in the endpoint, bucket and keys.")
with tempfile.TemporaryDirectory(prefix="steward-check-") as tmp:
path = Path(tmp) / name
await get_file(target, name, path)
return await check_file(app, path, name)
def start_backup(app, by: str, user_id: str | None) -> bool:
"""Start a backup in the background for a person; False if one is running."""
if _running.locked():
+33 -3
View File
@@ -3,10 +3,11 @@ path-style addressing (endpoint/bucket/key), the three calls a backup needs.
Fabled Archives speaks S3 with path-style buckets and append-only keys by
default (System 180 ruling: backups go there with an append-only key), so
Steward only ever puts and lists; it never deletes. Retention is the
target's business.
Steward puts, lists and reads back (to check or restore a backup); it never
deletes. Retention is the target's business. A key that may not read says
so as AccessDenied, and the backup can be uploaded by hand instead.
Written by hand rather than with boto3: three calls don't justify a large
Written by hand rather than with boto3: four calls don't justify a large
dependency, and signing is a pure function tests can hold to AWS's own
published examples.
"""
@@ -189,3 +190,32 @@ async def list_objects(target: S3Target, prefix: str = "", limit: int = 1000, *,
out.append(StoredObject(fields.get("Key", ""), int(fields.get("Size") or 0),
fields.get("LastModified", "")))
return out
async def get_file(target: S3Target, key: str, path: Path, *,
transport: httpx.AsyncBaseTransport | None = None,
now: datetime | None = None) -> int:
"""Download one object to `path`, streamed; returns its size. A partial
download is removed."""
url = target.url(key)
headers = sign("GET", url, {}, EMPTY_SHA256, target.access_key, target.secret_key,
target.region, now or datetime.now(timezone.utc))
size = 0
try:
async with httpx.AsyncClient(verify=target.verify_tls, timeout=UPLOAD_TIMEOUT_SECONDS,
transport=transport) as client:
async with client.stream("GET", url, headers=headers) as response:
if response.status_code != 200:
await response.aread()
raise S3Error(f"Reading {key} refused: {_says(response)}")
with open(path, "wb") as f:
async for chunk in response.aiter_bytes(CHUNK):
f.write(chunk)
size += len(chunk)
except httpx.HTTPError as exc:
path.unlink(missing_ok=True)
raise _reach_error(target, exc) from None
except S3Error:
path.unlink(missing_ok=True)
raise
return size