文件
hyb-oa/scripts/qa/inspect-backup-transfer-incident.py
T

559 行
22 KiB
Python
可执行文件

#!/usr/bin/env python3
"""Read-only, redacted inspection of an interrupted automatic-backup chain."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import stat
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable
BATCH_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))$")
PARTIAL_RE = re.compile(r"^hyb-oa-backup-(auto-(\d{8}T\d{6}Z))\.partial$")
TEMP_RE = re.compile(r"^\.hyb-oa-backup-.+\.tmp\..+$")
EXPECTED_FILES = {"database.dump", "storage.tar.gz", "manifest.json"}
ARTIFACTS = {"database": "database.dump", "storage": "storage.tar.gz"}
RSYNC_CODE_RE = re.compile(r"rsync error:[^\r\n]*\(code (11|12|23|30|255)\)")
RSYNC_CATEGORIES = {
"11": "file_io_error",
"12": "protocol_stream_error",
"23": "partial_transfer_error",
"30": "transport_timeout",
"255": "ssh_session_error",
}
SIMILAR_RELATIVE_PPM = 50_000
SIMILAR_ABSOLUTE_BYTES = 64 * 1024 * 1024
SAFE_KEY_RE = re.compile(r"^[a-z0-9_]+$")
SAFE_VALUE_RE = re.compile(r"^(?:[A-Za-z][A-Za-z0-9_]*|-?[0-9]+)$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
class InspectionError(Exception):
"""Stable, non-sensitive inspection failure."""
def __init__(self, category: str):
super().__init__(category)
self.category = category
@dataclass(frozen=True)
class BatchResult:
valid: bool
total_bytes: int
created_epoch: int | None
expected_artifact_bytes: int | None
received_artifact_bytes: int
def emit(items: Iterable[tuple[str, object]]) -> None:
rendered_items: list[tuple[str, str]] = []
for key, value in items:
if isinstance(value, bool):
rendered = "yes" if value else "no"
else:
rendered = str(value)
if not SAFE_KEY_RE.fullmatch(key) or not SAFE_VALUE_RE.fullmatch(rendered):
print("probe_result=FAIL")
print("probe_category=output_allowlist_violation")
raise SystemExit(1)
rendered_items.append((key, rendered))
for key, rendered in rendered_items:
print(f"{key}={rendered}")
def safe_root(path: Path) -> Path:
try:
if not path.is_absolute() or path.is_symlink() or not path.is_dir():
raise InspectionError("root_invalid")
list(path.iterdir())
except PermissionError as error:
raise InspectionError("access_denied") from error
except OSError as error:
raise InspectionError("root_unreadable") from error
return path
def file_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 tree_bytes(path: Path) -> int:
total = 0
try:
for root, directories, files in os.walk(path, followlinks=False):
directories[:] = [
name for name in directories if not (Path(root) / name).is_symlink()
]
for name in files:
item = Path(root) / name
if item.is_symlink():
continue
total += item.stat().st_size
except PermissionError as error:
raise InspectionError("access_denied") from error
except OSError as error:
raise InspectionError("entry_unreadable") from error
return total
def run_epoch(stamp: str) -> int | None:
try:
parsed = datetime.strptime(stamp, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
except ValueError:
return None
return int(parsed.timestamp())
def validate_batch(
path: Path,
expected_run_id: str,
stamp: str,
*,
verify_hashes: bool = True,
) -> BatchResult:
total = tree_bytes(path)
received_artifact_bytes = 0
expected_artifact_bytes: int | None = None
created_epoch = run_epoch(stamp)
if created_epoch is None:
return BatchResult(False, total, None, None, 0)
try:
entries = list(path.iterdir())
regular_names = {
item.name
for item in entries
if not item.is_symlink() and stat.S_ISREG(item.stat().st_mode)
}
if len(entries) != 3 or regular_names != EXPECTED_FILES:
return BatchResult(False, total, created_epoch, None, 0)
manifest = json.loads((path / "manifest.json").read_text(encoding="utf-8"))
if (
manifest.get("schemaVersion") != 1
or manifest.get("status") != "complete"
or manifest.get("runId") != expected_run_id
or not (manifest.get("buildSha") or manifest.get("sourceImageDigest"))
):
return BatchResult(False, total, created_epoch, None, 0)
artifacts = manifest.get("artifacts") or {}
expected_artifact_bytes = 0
for key, expected_name in ARTIFACTS.items():
item = artifacts.get(key) or {}
target = path / expected_name
if item.get("file") != expected_name or isinstance(item.get("bytes"), bool):
return BatchResult(False, total, created_epoch, None, 0)
expected_size = item.get("bytes")
if not isinstance(expected_size, int) or expected_size < 0:
return BatchResult(False, total, created_epoch, None, 0)
expected_sha = item.get("sha256")
if not isinstance(expected_sha, str) or not SHA256_RE.fullmatch(expected_sha):
return BatchResult(False, total, created_epoch, None, 0)
expected_artifact_bytes += expected_size
actual_size = target.stat().st_size
received_artifact_bytes += actual_size
if actual_size != expected_size or (
verify_hashes and file_sha256(target) != expected_sha
):
return BatchResult(
False,
total,
created_epoch,
expected_artifact_bytes,
received_artifact_bytes,
)
return BatchResult(
True,
total,
created_epoch,
expected_artifact_bytes,
received_artifact_bytes,
)
except PermissionError as error:
raise InspectionError("access_denied") from error
except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
return BatchResult(False, total, created_epoch, expected_artifact_bytes, received_artifact_bytes)
def inspect_partial(path: Path, run_id: str, stamp: str) -> BatchResult:
"""Measure a partial without requiring the complete three-file contract."""
total = tree_bytes(path)
received = 0
expected: int | None = None
try:
manifest_path = path / "manifest.json"
if manifest_path.is_file() and not manifest_path.is_symlink():
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
if manifest.get("runId") == run_id and manifest.get("schemaVersion") == 1:
artifacts = manifest.get("artifacts") or {}
sizes: list[int] = []
for key, expected_name in ARTIFACTS.items():
item = artifacts.get(key) or {}
size = item.get("bytes")
if (
item.get("file") != expected_name
or isinstance(size, bool)
or not isinstance(size, int)
or size < 0
):
sizes = []
break
sizes.append(size)
target = path / expected_name
if target.is_file() and not target.is_symlink():
received += target.stat().st_size
if len(sizes) == 2:
expected = sum(sizes)
validated = validate_batch(path, run_id, stamp)
return BatchResult(validated.valid, total, run_epoch(stamp), expected, received)
except PermissionError as error:
raise InspectionError("access_denied") from error
except (AttributeError, json.JSONDecodeError, OSError, TypeError, ValueError):
return BatchResult(False, total, run_epoch(stamp), expected, received)
def filesystem_fields(prefix: str, root: Path) -> list[tuple[str, object]]:
values = os.statvfs(root)
total = values.f_blocks * values.f_frsize
available = values.f_bavail * values.f_frsize
used_percent = 0 if total == 0 else ((total - available) * 100) // total
return [
(f"{prefix}_disk_used_percent", used_percent),
(f"{prefix}_disk_free_gib", available // (1024**3)),
]
def source_contract(wrapper: Path, backup_script: Path) -> list[tuple[str, object]]:
try:
if (
wrapper.is_symlink()
or not wrapper.is_file()
or backup_script.is_symlink()
or not backup_script.is_file()
):
raise InspectionError("source_contract_invalid")
wrapper_text = wrapper.read_text(encoding="utf-8")
backup_text = backup_script.read_text(encoding="utf-8")
except PermissionError as error:
raise InspectionError("source_contract_access_denied") from error
except OSError as error:
raise InspectionError("source_contract_unreadable") from error
if "run_backup() {" not in wrapper_text or "\ncheck_stale() {" not in wrapper_text:
raise InspectionError("source_contract_invalid")
run_backup_body = wrapper_text.split("run_backup() {", 1)[1].split("\ncheck_stale() {", 1)[0]
markers = {
"local_publish": '[[ -d "$batch_dir" && -f "$batch_dir/manifest.json" ]]',
"transfer": "capture_command offhost_transfer rsync",
"finalize": "capture_command offhost_finalize ssh",
"success": 'write_success_state "$run_id" "$manifest_sha"',
"prune": 'prune_local_backups "$batch_name"',
}
positions = {key: run_backup_body.find(marker) for key, marker in markers.items()}
all_present = all(position >= 0 for position in positions.values())
retained = all_present and positions["local_publish"] < positions["transfer"] < positions["prune"]
pruned = all_present and positions["finalize"] < positions["success"] < positions["prune"]
temp_cleanup = all(
marker in backup_text
for marker in (
"trap cleanup EXIT",
'TEMP_DIR="$(mktemp -d',
'rm -rf -- "$TEMP_DIR"',
'mv "$TEMP_DIR" "$FINAL_DIR"',
'TEMP_DIR=""',
"trap - EXIT",
)
)
return [
("retained_after_transfer_failure", retained),
("pruned_after_offhost_success", pruned),
("generation_temp_cleanup_on_failure", temp_cleanup),
("cleanup_contract_source_verified", all_present and temp_cleanup),
]
def last_backup_phase(state_dir: Path) -> str:
state_file = state_dir / "last-run.json"
if not state_file.exists():
return "NOT_AVAILABLE"
try:
if state_file.is_symlink() or not state_file.is_file():
return "STATE_INVALID"
payload = json.loads(state_file.read_text(encoding="utf-8"))
findings = payload.get("findings") or []
detail = findings[0].get("detail") if len(findings) == 1 and isinstance(findings[0], dict) else None
match = re.fullmatch(r"phase_([a-z_]+)_failed", detail or "")
allowed = {
"runtime_identity",
"paired_backup",
"offhost_transfer",
"offhost_finalize",
"success_state",
"stale_check",
"quarterly_reminder",
}
return match.group(1) if match and match.group(1) in allowed else "NOT_AVAILABLE"
except PermissionError as error:
raise InspectionError("diagnostic_access_denied") from error
except (AttributeError, json.JSONDecodeError, OSError, TypeError):
return "STATE_INVALID"
def parse_transfer_diagnostic(state_dir: Path) -> list[tuple[str, object]]:
phase = last_backup_phase(state_dir)
diagnostic = state_dir / "last-error.log"
if phase != "offhost_transfer" or not diagnostic.exists():
code = "NOT_AVAILABLE"
else:
try:
if diagnostic.is_symlink() or not diagnostic.is_file():
raise InspectionError("diagnostic_invalid")
text = diagnostic.read_text(encoding="utf-8", errors="replace")
except PermissionError as error:
raise InspectionError("diagnostic_access_denied") from error
except OSError as error:
raise InspectionError("diagnostic_unreadable") from error
matches = RSYNC_CODE_RE.findall(text)
code = matches[-1] if matches else "NOT_AVAILABLE"
return [
("last_backup_failure_phase", phase),
("transfer_exit_code", code),
("transfer_exit_code_source", "last_error_log_regex" if code != "NOT_AVAILABLE" else "NOT_AVAILABLE"),
("transfer_failure_category", RSYNC_CATEGORIES.get(code, "not_available")),
("transfer_started_epoch", "NOT_AVAILABLE"),
("transfer_failed_epoch", "NOT_AVAILABLE"),
("transfer_duration_seconds", "NOT_AVAILABLE"),
("transfer_duration_reason", "wrapper_does_not_persist_transfer_clock"),
]
def production(args: argparse.Namespace) -> None:
backup_root = safe_root(args.backup_root)
state_dir = safe_root(args.state_dir)
now = args.now_epoch
metadata_valid: list[tuple[Path, str, str, BatchResult]] = []
invalid_count = invalid_bytes = temp_count = temp_bytes = other_count = 0
for entry in backup_root.iterdir():
match = BATCH_RE.fullmatch(entry.name)
if match and entry.is_dir() and not entry.is_symlink():
result = validate_batch(
entry,
match.group(1),
match.group(2),
verify_hashes=False,
)
if result.valid and result.created_epoch is not None and result.created_epoch <= now:
metadata_valid.append((entry, match.group(1), match.group(2), result))
else:
invalid_count += 1
invalid_bytes += result.total_bytes
elif TEMP_RE.fullmatch(entry.name) and entry.is_dir() and not entry.is_symlink():
temp_count += 1
temp_bytes += tree_bytes(entry)
else:
other_count += 1
metadata_valid.sort(key=lambda item: item[3].created_epoch or -1)
latest_metadata = metadata_valid[-1] if metadata_valid else None
latest_verified: BatchResult | None = None
if latest_metadata is not None:
latest_verified = validate_batch(
latest_metadata[0],
latest_metadata[1],
latest_metadata[2],
verify_hashes=True,
)
complete_local_recovery_point = bool(latest_verified and latest_verified.valid)
latest_result = latest_metadata[3] if latest_metadata else None
output: list[tuple[str, object]] = [
("production_probe_result", "PASS"),
("complete_local_recovery_point", complete_local_recovery_point),
(
"backup_risk_classification",
"production_single_point"
if complete_local_recovery_point
else "no_proven_recovery_point",
),
("fully_verified_local_batch_count", 1 if complete_local_recovery_point else 0),
("metadata_complete_local_batch_count", len(metadata_valid)),
(
"metadata_complete_local_batch_total_bytes",
sum(item[3].total_bytes for item in metadata_valid),
),
(
"older_metadata_batch_hash_status",
"NOT_RUN" if len(metadata_valid) > 1 else "NOT_APPLICABLE",
),
(
"latest_local_candidate_hash_status",
"PASS"
if complete_local_recovery_point
else ("FAIL" if latest_metadata is not None else "NOT_RUN"),
),
(
"latest_local_candidate_age_seconds",
now - latest_result.created_epoch
if latest_result and latest_result.created_epoch is not None
else "NOT_AVAILABLE",
),
(
"latest_local_candidate_bytes",
latest_result.total_bytes if latest_result else "NOT_AVAILABLE",
),
("invalid_local_batch_count", invalid_count),
("invalid_local_batch_total_bytes", invalid_bytes),
("generation_temp_count", temp_count),
("generation_temp_total_bytes", temp_bytes),
("unclassified_local_entry_count", other_count),
(
"transition_recovery_point_action",
"wait_natural" if complete_local_recovery_point else "request_manual_run",
),
]
output.extend(filesystem_fields("production_backup", backup_root))
output.extend(source_contract(args.wrapper_script, args.backup_script))
output.extend(parse_transfer_diagnostic(state_dir))
emit(output)
def stopping_class(samples: list[BatchResult]) -> str:
if len(samples) < 2:
return "insufficient_samples"
sizes = sorted(item.total_bytes for item in samples[-2:])
difference = sizes[1] - sizes[0]
relative_limit = sizes[1] * SIMILAR_RELATIVE_PPM // 1_000_000
similar_limit = min(SIMILAR_ABSOLUTE_BYTES, relative_limit)
return "similar" if difference <= similar_limit else "different"
def arm(args: argparse.Namespace) -> None:
inbox_root = safe_root(args.inbox_root)
archive_root = safe_root(args.archive_root)
now = args.now_epoch
partials: list[BatchResult] = []
other_count = 0
for entry in inbox_root.iterdir():
match = PARTIAL_RE.fullmatch(entry.name)
if match and entry.is_dir() and not entry.is_symlink():
partials.append(inspect_partial(entry, match.group(1), match.group(2)))
else:
other_count += 1
partials.sort(key=lambda item: item.created_epoch or -1)
archives: list[BatchResult] = []
invalid_archive_count = 0
for entry in archive_root.iterdir():
match = BATCH_RE.fullmatch(entry.name)
if match and entry.is_dir() and not entry.is_symlink():
result = validate_batch(entry, match.group(1), match.group(2))
if result.valid and result.created_epoch is not None and result.created_epoch <= now:
archives.append(result)
else:
invalid_archive_count += 1
archives.sort(key=lambda item: item.created_epoch or -1)
output: list[tuple[str, object]] = [
("arm_probe_result", "PASS"),
("arm_partial_count", len(partials)),
("arm_partial_total_bytes", sum(item.total_bytes for item in partials)),
("arm_partial_complete_contract_count", sum(1 for item in partials if item.valid)),
("arm_partial_incomplete_contract_count", sum(1 for item in partials if not item.valid)),
("arm_partial_stop_offset_class", stopping_class(partials)),
("arm_partial_similarity_relative_ppm", SIMILAR_RELATIVE_PPM),
("arm_partial_similarity_absolute_bytes", SIMILAR_ABSOLUTE_BYTES),
("arm_inbox_other_entry_count", other_count),
("arm_valid_atomic_archive_count", len(archives)),
("arm_invalid_atomic_archive_count", invalid_archive_count),
(
"arm_latest_atomic_archive_age_seconds",
now - archives[-1].created_epoch if archives else "NOT_AVAILABLE",
),
]
for index, sample in enumerate(partials, 1):
ratio = (
min(sample.received_artifact_bytes, sample.expected_artifact_bytes)
* 1_000_000
// sample.expected_artifact_bytes
if sample.expected_artifact_bytes and sample.expected_artifact_bytes > 0
else "NOT_AVAILABLE"
)
output.extend(
[
(f"arm_partial_sample_{index}_total_bytes", sample.total_bytes),
(
f"arm_partial_sample_{index}_expected_artifact_bytes",
sample.expected_artifact_bytes
if sample.expected_artifact_bytes is not None
else "NOT_AVAILABLE",
),
(f"arm_partial_sample_{index}_received_ratio_ppm", ratio),
(
f"arm_partial_sample_{index}_age_seconds",
now - sample.created_epoch
if sample.created_epoch is not None
else "NOT_AVAILABLE",
),
(f"arm_partial_sample_{index}_complete_contract", sample.valid),
]
)
output.extend(filesystem_fields("arm_inbox", inbox_root))
output.extend(filesystem_fields("arm_archive", archive_root))
emit(output)
def absolute_path(value: str) -> Path:
path = Path(value)
if not path.is_absolute():
raise argparse.ArgumentTypeError("an explicit absolute path is required")
return path
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Emit redacted backup-chain incident fields")
parser.add_argument("--now-epoch", type=int, default=int(datetime.now(timezone.utc).timestamp()))
subparsers = parser.add_subparsers(dest="mode", required=True)
production_parser = subparsers.add_parser("production")
production_parser.add_argument("--backup-root", type=absolute_path, required=True)
production_parser.add_argument("--state-dir", type=absolute_path, required=True)
production_parser.add_argument("--wrapper-script", type=absolute_path, required=True)
production_parser.add_argument("--backup-script", type=absolute_path, required=True)
production_parser.set_defaults(handler=production)
arm_parser = subparsers.add_parser("arm")
arm_parser.add_argument("--inbox-root", type=absolute_path, required=True)
arm_parser.add_argument("--archive-root", type=absolute_path, required=True)
arm_parser.set_defaults(handler=arm)
return parser
def main() -> int:
args = build_parser().parse_args()
try:
args.handler(args)
except InspectionError as error:
emit([("probe_result", "FAIL"), ("probe_category", error.category)])
return 1
except PermissionError:
emit([("probe_result", "FAIL"), ("probe_category", "access_denied")])
return 1
except (OSError, TypeError, ValueError):
emit([("probe_result", "FAIL"), ("probe_category", "inspection_io_error")])
return 1
except Exception:
emit([("probe_result", "FAIL"), ("probe_category", "inspection_internal_error")])
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())