559 行
22 KiB
Python
可执行文件
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())
|