d0a5a27926
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02994-7685-7940-bf34-3555b8256018
221 lines
9.4 KiB
Python
Executable file
221 lines
9.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Validate the declared S1 backup membership and public-recipient policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
SCHEMA_VERSION = "1.0"
|
|
AGE_RECIPIENT_RE = re.compile(r"^age1[023456789acdefghjklmnpqrstuvwxyz]{20,}$")
|
|
REQUIRED_EXCLUSIONS = {
|
|
"/etc/ssl/private",
|
|
"/home",
|
|
"/root",
|
|
"/var/lib/kubelet",
|
|
"/var/lib/rancher",
|
|
}
|
|
FORBIDDEN_PARTS = {"id_rsa", "id_ed25519", "keys.txt", "private", "secrets"}
|
|
|
|
|
|
class BackupContractError(ValueError):
|
|
"""The backup declaration is ambiguous, unsafe, or inconsistent."""
|
|
|
|
|
|
def _absolute_path(value: Any, label: str) -> str:
|
|
if not isinstance(value, str) or not value.startswith("/"):
|
|
raise BackupContractError(f"{label} must be an absolute path")
|
|
path = PurePosixPath(value)
|
|
if ".." in path.parts or str(path) != value.rstrip("/"):
|
|
raise BackupContractError(f"{label} must be normalized without traversal")
|
|
return str(path)
|
|
|
|
|
|
def _sops_recipients(path: Path) -> set[str]:
|
|
try:
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
raise BackupContractError(f"cannot read recipient policy {path}: {exc}") from exc
|
|
recipients: set[str] = set()
|
|
for rule in payload.get("creation_rules", []) if isinstance(payload, dict) else []:
|
|
for group in rule.get("key_groups", []) if isinstance(rule, dict) else []:
|
|
age = group.get("age", []) if isinstance(group, dict) else []
|
|
if isinstance(age, list):
|
|
recipients.update(item.strip() for item in age if isinstance(item, str))
|
|
if not recipients:
|
|
raise BackupContractError("recipient policy contains no age recipients")
|
|
return recipients
|
|
|
|
|
|
def validate_spec(payload: Any, *, policy_path: Path) -> dict[str, Any]:
|
|
if not isinstance(payload, dict):
|
|
raise BackupContractError("backup declaration must be a YAML object")
|
|
errors: list[str] = []
|
|
if str(payload.get("schema_version")) != SCHEMA_VERSION:
|
|
errors.append(f"schema_version must be {SCHEMA_VERSION}")
|
|
try:
|
|
backup_root = _absolute_path(payload.get("backup_root"), "backup_root")
|
|
except BackupContractError as exc:
|
|
errors.append(str(exc))
|
|
backup_root = ""
|
|
if backup_root in {"", "/", "/etc", "/home", "/root", "/var"}:
|
|
errors.append("backup_root must be a dedicated bounded directory")
|
|
|
|
recipients = payload.get("recipients")
|
|
if not isinstance(recipients, list) or not recipients or not all(
|
|
isinstance(item, str) and AGE_RECIPIENT_RE.fullmatch(item) for item in recipients
|
|
):
|
|
errors.append("recipients must be a non-empty list of age public recipients")
|
|
recipients = []
|
|
elif len(set(recipients)) != len(recipients):
|
|
errors.append("recipients contains duplicates")
|
|
try:
|
|
policy_recipients = _sops_recipients(policy_path)
|
|
if set(recipients) != policy_recipients:
|
|
errors.append("backup recipients must exactly match the SOPS public-recipient policy")
|
|
except BackupContractError as exc:
|
|
errors.append(str(exc))
|
|
|
|
retention = payload.get("retention")
|
|
if not isinstance(retention, dict):
|
|
errors.append("retention must be an object")
|
|
else:
|
|
for key, low, high in (
|
|
("keep_complete_bundles", 2, 90),
|
|
("max_total_bytes", 1048576, 10737418240),
|
|
("freshness_hours", 1, 168),
|
|
):
|
|
value = retention.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool) or not low <= value <= high:
|
|
errors.append(f"retention.{key} must be an integer from {low} to {high}")
|
|
|
|
offsite = payload.get("offsite")
|
|
if not isinstance(offsite, dict) or not isinstance(
|
|
offsite.get("required_before_prune"), bool
|
|
):
|
|
errors.append("offsite must declare boolean required_before_prune and receipt_root")
|
|
else:
|
|
try:
|
|
receipt_root = _absolute_path(offsite.get("receipt_root"), "offsite.receipt_root")
|
|
if backup_root and receipt_root != backup_root + "/offsite-receipts":
|
|
errors.append("offsite.receipt_root must be the bounded backup-root receipt directory")
|
|
except BackupContractError as exc:
|
|
errors.append(str(exc))
|
|
|
|
exclusions = payload.get("excluded_prefixes")
|
|
normalized_exclusions: list[str] = []
|
|
if not isinstance(exclusions, list):
|
|
errors.append("excluded_prefixes must be a list")
|
|
else:
|
|
for index, value in enumerate(exclusions):
|
|
try:
|
|
normalized_exclusions.append(
|
|
_absolute_path(value, f"excluded_prefixes[{index}]")
|
|
)
|
|
except BackupContractError as exc:
|
|
errors.append(str(exc))
|
|
missing = sorted(REQUIRED_EXCLUSIONS - set(normalized_exclusions))
|
|
if missing:
|
|
errors.append(f"excluded_prefixes omits governed exclusions {', '.join(missing)}")
|
|
|
|
artifacts = payload.get("artifacts")
|
|
if not isinstance(artifacts, dict) or set(artifacts) != {"os-config", "packages"}:
|
|
errors.append("artifacts must contain exactly os-config and packages")
|
|
artifacts = {}
|
|
os_config = artifacts.get("os-config", {})
|
|
paths = os_config.get("paths") if isinstance(os_config, dict) else None
|
|
declared: list[str] = []
|
|
if not isinstance(paths, list) or not paths:
|
|
errors.append("artifacts.os-config.paths must be a non-empty list")
|
|
else:
|
|
for index, item in enumerate(paths):
|
|
label = f"artifacts.os-config.paths[{index}]"
|
|
if not isinstance(item, dict) or not isinstance(item.get("required"), bool):
|
|
errors.append(f"{label} must contain path and boolean required")
|
|
continue
|
|
expected_mode = item.get("expected_mode")
|
|
if item["required"] and not (
|
|
isinstance(expected_mode, str) and re.fullmatch(r"0[0-7]{3}", expected_mode)
|
|
):
|
|
errors.append(f"{label} requires expected_mode as four octal digits")
|
|
if not item["required"] and expected_mode is not None:
|
|
errors.append(f"{label}.expected_mode is allowed only for required paths")
|
|
try:
|
|
path = _absolute_path(item.get("path"), f"{label}.path")
|
|
except BackupContractError as exc:
|
|
errors.append(str(exc))
|
|
continue
|
|
declared.append(path)
|
|
lowered = {part.lower() for part in PurePosixPath(path).parts}
|
|
if lowered & FORBIDDEN_PARTS:
|
|
errors.append(f"{label}.path selects secret/private-key shaped content")
|
|
if any(path == prefix or path.startswith(prefix + "/") for prefix in normalized_exclusions):
|
|
errors.append(f"{label}.path falls under excluded prefix")
|
|
if len(set(declared)) != len(declared):
|
|
errors.append("artifacts.os-config.paths contains duplicates")
|
|
for left in declared:
|
|
for right in declared:
|
|
if left != right and right.startswith(left + "/"):
|
|
errors.append(f"declared paths overlap: {left} contains {right}")
|
|
if isinstance(os_config, dict) and os_config.get("format") != "tar.gz":
|
|
errors.append("artifacts.os-config.format must be tar.gz")
|
|
packages = artifacts.get("packages", {})
|
|
if not isinstance(packages, dict) or packages.get("format") != "dpkg-selections":
|
|
errors.append("artifacts.packages.format must be dpkg-selections")
|
|
elif packages.get("command") != ["dpkg", "--get-selections"]:
|
|
errors.append("artifacts.packages.command must be the fixed dpkg selection argv")
|
|
|
|
if errors:
|
|
raise BackupContractError("backup contract failed:\n- " + "\n- ".join(errors))
|
|
return payload
|
|
|
|
|
|
def load_spec(path: Path) -> dict[str, Any]:
|
|
try:
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
raise BackupContractError(f"cannot read {path}: {exc}") from exc
|
|
policy_value = payload.get("recipient_policy") if isinstance(payload, dict) else None
|
|
if not isinstance(policy_value, str) or not policy_value:
|
|
raise BackupContractError("recipient_policy must name a repository-relative file")
|
|
policy_relative = PurePosixPath(policy_value)
|
|
if policy_relative.is_absolute() or ".." in policy_relative.parts:
|
|
raise BackupContractError("recipient_policy must be a normalized repository-relative path")
|
|
root = path.resolve().parent.parent
|
|
return validate_spec(payload, policy_path=root / policy_relative)
|
|
|
|
|
|
def declaration_sha256(path: Path) -> str:
|
|
import hashlib
|
|
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("spec", nargs="?", type=Path, default=Path("spec/s1-backup.yaml"))
|
|
args = parser.parse_args()
|
|
payload = load_spec(args.spec)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"artifacts": sorted(payload["artifacts"]),
|
|
"declaration_sha256": declaration_sha256(args.spec),
|
|
"ok": True,
|
|
"path_count": len(payload["artifacts"]["os-config"]["paths"]),
|
|
"recipients": len(payload["recipients"]),
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|