396 lines
15 KiB
Python
Executable File
396 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Snapshots, through snapper.
|
|
|
|
A point in time you can go back to, per btrfs subvolume. This machine already
|
|
had snapper running hourly, but only for / -- /home is a separate subvolume and
|
|
had no configuration at all, so six hundred snapshots existed and not one of
|
|
them contained a document.
|
|
|
|
Deliberately absent: rollback. snapper's rollback works by changing the btrfs
|
|
default subvolume, and this system's fstab pins subvol=root and subvol=home
|
|
explicitly, which overrides it -- so a rollback would appear to succeed and
|
|
change nothing after a reboot. Restoring files and folders out of a snapshot
|
|
needs no reboot, cannot affect booting, and covers the cases people actually
|
|
hit.
|
|
|
|
panama-snapshots snapshot
|
|
panama-snapshots create CONFIG DESCRIPTION
|
|
panama-snapshots delete CONFIG NUMBER
|
|
panama-snapshots set-retention CONFIG HOURLY DAILY WEEKLY
|
|
panama-snapshots set-timeline CONFIG true|false
|
|
panama-snapshots browse CONFIG NUMBER [RELATIVE_PATH]
|
|
panama-snapshots restore CONFIG NUMBER RELATIVE_PATH
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
CONFIG_NAME = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
|
|
|
|
# Directory listings are for choosing something to restore, not for browsing a
|
|
# terabyte. A folder with more entries than this is reported as truncated.
|
|
BROWSE_LIMIT = 400
|
|
|
|
|
|
class BoundaryError(RuntimeError):
|
|
"""A user-visible validation or snapper failure."""
|
|
|
|
|
|
def run(command: list[str], timeout: float = 30.0) -> subprocess.CompletedProcess:
|
|
try:
|
|
return subprocess.run(command, capture_output=True, text=True,
|
|
timeout=timeout, check=False)
|
|
except subprocess.TimeoutExpired as error:
|
|
raise BoundaryError("snapper did not answer in time.") from error
|
|
except OSError as error:
|
|
raise BoundaryError("snapper is not available.") from error
|
|
|
|
|
|
def require_config(name: str) -> str:
|
|
if not CONFIG_NAME.fullmatch(name or ""):
|
|
raise BoundaryError("That is not a snapshot configuration.")
|
|
return name
|
|
|
|
|
|
def require_number(value: str) -> int:
|
|
if not str(value).isdigit():
|
|
raise BoundaryError("That is not a snapshot.")
|
|
number = int(value)
|
|
# 0 is the live filesystem, not a snapshot, and must never be a target.
|
|
if number < 1:
|
|
raise BoundaryError("That is the current state, not a snapshot.")
|
|
return number
|
|
|
|
|
|
# Everything snapper answers is read through --machine-readable csv and a
|
|
# DictReader. The first version split snapper's box-drawing table by column
|
|
# INDEX, which is the presentation layer: it is localized, and its columns
|
|
# have moved between snapper versions -- a German desktop parsed to zero
|
|
# snapshots while `snapper list` showed twelve. Named columns survive both.
|
|
def snapper_csv(args: list[str], timeout: int = 20) -> list[dict]:
|
|
result = run(["snapper", "--machine-readable", "csv", *args], timeout=timeout)
|
|
if result.returncode != 0:
|
|
return []
|
|
return list(csv.DictReader(io.StringIO(result.stdout)))
|
|
|
|
|
|
def config_names() -> list[str]:
|
|
return [row["config"] for row in snapper_csv(["list-configs"])
|
|
if CONFIG_NAME.fullmatch(row.get("config", ""))]
|
|
|
|
|
|
def config_settings(name: str) -> dict:
|
|
return {row["key"]: row.get("value", "")
|
|
for row in snapper_csv(["-c", name, "get-config"]) if row.get("key")}
|
|
|
|
|
|
def snapshots_for(name: str) -> list[dict]:
|
|
entries = []
|
|
for row in snapper_csv(["-c", name, "list"], timeout=45):
|
|
if not str(row.get("number", "")).isdigit():
|
|
continue
|
|
number = int(row["number"])
|
|
if number == 0:
|
|
continue
|
|
cleanup = row.get("cleanup", "") or ""
|
|
entries.append({
|
|
"number": number,
|
|
"kind": row.get("type", ""),
|
|
"date": row.get("date", ""),
|
|
"user": row.get("user", ""),
|
|
"cleanup": cleanup,
|
|
"description": row.get("description", ""),
|
|
# A snapshot with no cleanup algorithm is not on the timeline's
|
|
# list to remove, which is what "kept" means to someone reading it.
|
|
"kept": cleanup == "",
|
|
})
|
|
entries.sort(key=lambda entry: entry["number"], reverse=True)
|
|
return entries
|
|
|
|
|
|
def btrfs_subvolumes() -> list[dict]:
|
|
"""Mounted btrfs subvolumes, so the page can name what is NOT protected."""
|
|
result = run(["findmnt", "-t", "btrfs", "-J", "-o", "TARGET,OPTIONS"])
|
|
if result.returncode != 0:
|
|
return []
|
|
try:
|
|
payload = json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
|
|
found = []
|
|
|
|
def walk(nodes):
|
|
for node in nodes:
|
|
options = str(node.get("options", ""))
|
|
target = str(node.get("target", ""))
|
|
match = re.search(r"subvol=(/[^,]*)", options)
|
|
# .snapshots holds the snapshots themselves and is not a thing to
|
|
# protect; listing it would offer to snapshot the snapshots.
|
|
if match and "/.snapshots" not in target:
|
|
found.append({"path": target, "subvolume": match.group(1)})
|
|
walk(node.get("children", []))
|
|
|
|
walk(payload.get("filesystems", []))
|
|
return found
|
|
|
|
|
|
def free_space() -> dict:
|
|
try:
|
|
usage = shutil.disk_usage("/home")
|
|
except OSError:
|
|
return {"freeBytes": 0, "totalBytes": 0}
|
|
return {"freeBytes": usage.free, "totalBytes": usage.total}
|
|
|
|
|
|
def timeline_running() -> bool:
|
|
return run(["systemctl", "is-active", "snapper-timeline.timer"]).stdout.strip() == "active"
|
|
|
|
|
|
def snapshot() -> dict:
|
|
configs = []
|
|
protected_paths = set()
|
|
for name in config_names():
|
|
settings = config_settings(name)
|
|
subvolume = settings.get("SUBVOLUME", "")
|
|
protected_paths.add(subvolume)
|
|
configs.append({
|
|
"name": name,
|
|
"subvolume": subvolume,
|
|
"timelineEnabled": settings.get("TIMELINE_CREATE", "no") == "yes",
|
|
# Empty when this user cannot read the config at all, which is a
|
|
# different state from "no snapshots".
|
|
"readable": bool(settings),
|
|
"limits": {
|
|
"hourly": int(settings.get("TIMELINE_LIMIT_HOURLY") or 0),
|
|
"daily": int(settings.get("TIMELINE_LIMIT_DAILY") or 0),
|
|
"weekly": int(settings.get("TIMELINE_LIMIT_WEEKLY") or 0),
|
|
"monthly": int(settings.get("TIMELINE_LIMIT_MONTHLY") or 0),
|
|
"yearly": int(settings.get("TIMELINE_LIMIT_YEARLY") or 0),
|
|
},
|
|
"snapshots": snapshots_for(name) if settings else [],
|
|
})
|
|
configs.sort(key=lambda entry: entry["subvolume"])
|
|
|
|
unprotected = [entry for entry in btrfs_subvolumes()
|
|
if entry["path"] not in protected_paths]
|
|
|
|
return {
|
|
"configs": configs,
|
|
"unprotected": unprotected,
|
|
"timelineRunning": timeline_running(),
|
|
"space": free_space(),
|
|
# An ext4 machine and an unconfigured btrfs machine used to look
|
|
# identical: an empty page. "supported" is whether snapshots are even
|
|
# possible here, so absence can say which kind of absence it is.
|
|
"supported": bool(btrfs_subvolumes()),
|
|
"error": "",
|
|
}
|
|
|
|
|
|
def snapshot_root(config: str, number: int) -> Path:
|
|
settings = config_settings(config)
|
|
subvolume = settings.get("SUBVOLUME", "")
|
|
if not subvolume:
|
|
raise BoundaryError("That snapshot configuration cannot be read.")
|
|
path = Path(subvolume) / ".snapshots" / str(number) / "snapshot"
|
|
if not path.is_dir():
|
|
raise BoundaryError("That snapshot is not available.")
|
|
return path
|
|
|
|
|
|
def safe_relative(root: Path, relative: str) -> Path:
|
|
"""Resolve a path inside a snapshot, refusing anything that escapes it.
|
|
|
|
The caller is a settings page passing a path a person clicked, and ".." in
|
|
the wrong place would read or restore from outside the snapshot entirely.
|
|
"""
|
|
candidate = (root / relative.lstrip("/")).resolve()
|
|
if candidate != root.resolve() and root.resolve() not in candidate.parents:
|
|
raise BoundaryError("That path is not inside the snapshot.")
|
|
return candidate
|
|
|
|
|
|
def browse(config: str, number: str, relative: str) -> dict:
|
|
root = snapshot_root(require_config(config), require_number(number))
|
|
target = safe_relative(root, relative)
|
|
if not target.is_dir():
|
|
raise BoundaryError("That is not a folder in this snapshot.")
|
|
|
|
entries = []
|
|
truncated = False
|
|
try:
|
|
with os.scandir(target) as scan:
|
|
for item in scan:
|
|
if len(entries) >= BROWSE_LIMIT:
|
|
truncated = True
|
|
break
|
|
try:
|
|
is_dir = item.is_dir(follow_symlinks=False)
|
|
size = 0 if is_dir else item.stat(follow_symlinks=False).st_size
|
|
except OSError:
|
|
continue
|
|
entries.append({"name": item.name, "directory": is_dir, "bytes": size})
|
|
except PermissionError as error:
|
|
raise BoundaryError("That folder cannot be read from this snapshot.") from error
|
|
except OSError as error:
|
|
raise BoundaryError("That folder could not be listed.") from error
|
|
|
|
entries.sort(key=lambda entry: (not entry["directory"], entry["name"].lower()))
|
|
return {"path": relative, "entries": entries, "truncated": truncated, "error": ""}
|
|
|
|
|
|
def restore(config: str, number: str, relative: str) -> dict:
|
|
"""Copy something out of a snapshot, keeping whatever is there now.
|
|
|
|
The current version is moved aside rather than overwritten. A restore that
|
|
destroys the thing you were about to compare it against is how people lose
|
|
the work they were trying to save.
|
|
"""
|
|
name = require_config(config)
|
|
index = require_number(number)
|
|
root = snapshot_root(name, index)
|
|
source = safe_relative(root, relative)
|
|
if not source.exists():
|
|
raise BoundaryError("That is not in this snapshot.")
|
|
|
|
settings = config_settings(name)
|
|
live_root = Path(settings.get("SUBVOLUME", ""))
|
|
destination = safe_relative(live_root, relative)
|
|
|
|
kept = ""
|
|
if destination.exists():
|
|
kept = str(destination) + f".before-restore-{index}"
|
|
suffix = 1
|
|
while Path(kept).exists():
|
|
suffix += 1
|
|
kept = str(destination) + f".before-restore-{index}-{suffix}"
|
|
try:
|
|
os.rename(destination, kept)
|
|
except OSError as error:
|
|
raise BoundaryError("The current version could not be set aside.") from error
|
|
|
|
try:
|
|
if source.is_dir():
|
|
shutil.copytree(source, destination, symlinks=True)
|
|
else:
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source, destination, follow_symlinks=False)
|
|
except OSError as error:
|
|
raise BoundaryError("That could not be restored.") from error
|
|
|
|
return {"restored": str(destination), "keptAs": kept}
|
|
|
|
|
|
def create(config: str, description: str) -> None:
|
|
if len(description) > 200 or "\n" in description:
|
|
raise BoundaryError("That description cannot be used.")
|
|
result = run(["snapper", "-c", require_config(config), "create",
|
|
"--description", description or "manual snapshot"], timeout=120)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "The snapshot could not be taken."))
|
|
|
|
|
|
def delete(config: str, number: str) -> None:
|
|
result = run(["snapper", "-c", require_config(config), "delete",
|
|
str(require_number(number))], timeout=120)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "That snapshot could not be removed."))
|
|
|
|
|
|
# The three horizons the page can edit, and the largest number it will accept
|
|
# for any of them. 999 hourly snapshots is not a retention policy, it is a typo
|
|
# that fills a drive; the page offers a dropdown and this is its ceiling. The
|
|
# monthly and yearly limits are left exactly as snapper has them -- nothing here
|
|
# writes them, so a config with longer horizons keeps them.
|
|
RETENTION_LIMIT = 50
|
|
|
|
|
|
def set_retention(config: str, hourly: str, daily: str, weekly: str) -> None:
|
|
values = []
|
|
for label, value in (("HOURLY", hourly), ("DAILY", daily), ("WEEKLY", weekly)):
|
|
if not str(value).isdigit():
|
|
raise BoundaryError("Keep counts must be whole numbers.")
|
|
if int(value) > RETENTION_LIMIT:
|
|
raise BoundaryError(f"Keep counts go up to {RETENTION_LIMIT}.")
|
|
values.append(f"TIMELINE_LIMIT_{label}={int(value)}")
|
|
result = run(["snapper", "-c", require_config(config), "set-config", *values])
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "The keep counts could not be changed."))
|
|
|
|
|
|
def set_timeline(config: str, enabled: str) -> None:
|
|
result = run(["snapper", "-c", require_config(config), "set-config",
|
|
f"TIMELINE_CREATE={'yes' if enabled == 'true' else 'no'}"])
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "Automatic snapshots could not be changed."))
|
|
|
|
|
|
def _refusal(result: subprocess.CompletedProcess, fallback: str) -> str:
|
|
text = (result.stderr or result.stdout or "").strip().splitlines()
|
|
if not text:
|
|
return fallback
|
|
last = text[-1]
|
|
if "permission" in last.lower():
|
|
return "This account is not allowed to change that configuration."
|
|
return last[:200]
|
|
|
|
|
|
def main(arguments: list[str]) -> int:
|
|
try:
|
|
if arguments == ["snapshot"]:
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
if len(arguments) in (3, 4) and arguments[0] == "browse":
|
|
print(json.dumps(browse(arguments[1], arguments[2],
|
|
arguments[3] if len(arguments) == 4 else ""),
|
|
separators=(",", ":")))
|
|
return 0
|
|
if len(arguments) == 4 and arguments[0] == "restore":
|
|
outcome = restore(arguments[1], arguments[2], arguments[3])
|
|
state = snapshot()
|
|
state["restored"] = outcome
|
|
print(json.dumps(state, separators=(",", ":")))
|
|
return 0
|
|
|
|
if len(arguments) == 3 and arguments[0] == "create":
|
|
create(arguments[1], arguments[2])
|
|
elif len(arguments) == 3 and arguments[0] == "delete":
|
|
delete(arguments[1], arguments[2])
|
|
elif len(arguments) == 5 and arguments[0] == "set-retention":
|
|
set_retention(arguments[1], arguments[2], arguments[3], arguments[4])
|
|
elif len(arguments) == 3 and arguments[0] == "set-timeline":
|
|
set_timeline(arguments[1], arguments[2])
|
|
else:
|
|
raise BoundaryError(
|
|
"Usage: panama-snapshots snapshot | create CONFIG DESCRIPTION | "
|
|
"delete CONFIG NUMBER | set-retention CONFIG HOURLY DAILY WEEKLY | "
|
|
"set-timeline CONFIG true|false | browse CONFIG NUMBER [PATH] | "
|
|
"restore CONFIG NUMBER PATH")
|
|
except BoundaryError as error:
|
|
try:
|
|
state = snapshot()
|
|
except BoundaryError:
|
|
state = {"configs": [], "unprotected": [], "timelineRunning": False,
|
|
"space": {"freeBytes": 0, "totalBytes": 0}}
|
|
state["error"] = str(error)
|
|
print(json.dumps(state, separators=(",", ":")))
|
|
return 0
|
|
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|