127 行
5.1 KiB
Python
可执行文件
127 行
5.1 KiB
Python
可执行文件
#!/usr/bin/env python3
|
|
"""Validate the newest ARM archive and drive an independent dead-man alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
BATCH_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))$")
|
|
|
|
|
|
def fail(message: str) -> "NoReturn":
|
|
print(f"[offhost-deadman][ERROR] {message}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def validate_latest(archive_root: Path, max_age: int) -> tuple[str, str]:
|
|
if not archive_root.is_dir() or archive_root.is_symlink():
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_root_invalid"
|
|
candidates = sorted(
|
|
(entry for entry in archive_root.iterdir() if entry.is_dir() and not entry.is_symlink() and BATCH_RE.fullmatch(entry.name)),
|
|
key=lambda entry: entry.name,
|
|
)
|
|
if not candidates:
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_missing"
|
|
batch = candidates[-1]
|
|
match = BATCH_RE.fullmatch(batch.name)
|
|
assert match is not None
|
|
run_id, stamp = match.groups()
|
|
expected_names = {"database.dump", "storage.tar.gz", "manifest.json"}
|
|
entries = list(batch.iterdir())
|
|
if {entry.name for entry in entries} != expected_names or any(
|
|
entry.is_symlink() or not stat.S_ISREG(entry.stat().st_mode) for entry in entries
|
|
):
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_files_invalid"
|
|
try:
|
|
manifest = json.loads((batch / "manifest.json").read_text(encoding="utf-8"))
|
|
if manifest.get("schemaVersion") != 1 or manifest.get("status") != "complete" or manifest.get("runId") != run_id:
|
|
raise ValueError
|
|
artifacts = manifest.get("artifacts") or {}
|
|
for key, name in (("database", "database.dump"), ("storage", "storage.tar.gz")):
|
|
item = artifacts.get(key) or {}
|
|
target = batch / name
|
|
if item.get("file") != name or item.get("bytes") != target.stat().st_size or item.get("sha256") != sha256(target):
|
|
raise ValueError
|
|
completed = datetime.strptime(stamp, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_manifest_invalid"
|
|
age = (datetime.now(timezone.utc) - completed).total_seconds()
|
|
if age < 0:
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_time_in_future"
|
|
if age > max_age:
|
|
return "FAIL_OFFHOST_DEADMAN", "archive_stale"
|
|
return "PASS_OFFHOST_DEADMAN_FRESH", "fresh_offhost_archive"
|
|
|
|
|
|
def write_summary(path: Path, status: str, category: str) -> None:
|
|
finding = status.startswith("FAIL_")
|
|
payload = {
|
|
"issueId": "AUTO_BACKUP",
|
|
"status": status,
|
|
"conclusion": status,
|
|
"findingCount": 1 if finding else 0,
|
|
"findings": ([{"source": "offhost_deadman", "status": status, "severity": "P0", "detail": category}] if finding else []),
|
|
"privacy": "status/count/category only; no credentials, host addresses, absolute paths, file names, or payloads",
|
|
}
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
fd, temporary = tempfile.mkstemp(prefix=".deadman-summary.", dir=path.parent)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, ensure_ascii=True, indent=2)
|
|
handle.write("\n")
|
|
os.chmod(temporary, 0o600)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--archive-root", required=True, type=Path)
|
|
parser.add_argument("--state-dir", required=True, type=Path)
|
|
parser.add_argument("--webhook-file", required=True, type=Path)
|
|
parser.add_argument("--alert-helper", required=True, type=Path)
|
|
parser.add_argument("--max-age-seconds", type=int, default=93600)
|
|
args = parser.parse_args()
|
|
if args.max_age_seconds <= 0:
|
|
fail("max age must be positive")
|
|
try:
|
|
args.state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
if args.state_dir.is_symlink() or not args.state_dir.is_dir():
|
|
fail("state directory is invalid")
|
|
os.chmod(args.state_dir, 0o700)
|
|
status, category = validate_latest(args.archive_root, args.max_age_seconds)
|
|
summary = args.state_dir / "last-deadman.json"
|
|
write_summary(summary, status, category)
|
|
result = subprocess.run(
|
|
[sys.executable, str(args.alert_helper), "--webhook-file", str(args.webhook_file), "--summary-file", str(summary), "--state-file", str(args.state_dir / "deadman-alert-state.json")],
|
|
check=False,
|
|
)
|
|
except OSError:
|
|
fail("dead-man local file operation failed")
|
|
if result.returncode != 0:
|
|
fail("alert evaluation failed")
|
|
if status.startswith("FAIL_"):
|
|
fail(category)
|
|
print("[offhost-deadman][PASS] newest archive is fresh and valid")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|