scripts/ci-qa-macos-os-repair.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Run the macOS OS-disk repair helper against an isolated staged image."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import platform
import pwd
import shlex
import shutil
import signal
import stat
import subprocess
import sys
import time
from datetime import datetime, timezone
HELPER_TIMEOUT = 120
HELPER_STOP_TIMEOUT = 10
BASE_RESULT_KEYS = {"check_exit", "repair_exit", "verify_exit", "changed"}
RESULT_KEYS = BASE_RESULT_KEYS | {"restored_files"}
REFERENCE_IMAGE_NAME = "debian-system.reference.raw"
DEFAULT_MANAGED_SUFFIX = Path("Library/Application Support/Firecrab/micromanager")
FIXTURE_MARKER = ".firecrab-os-repair-fixture"
FIXTURE_MARKER_CONTENT = "isolated OS repair fixture\n"
PROTECTED_FILES = {
"fixture_marker": (FIXTURE_MARKER, True),
"original_os_disk": ("system/debian-system.raw", True),
"os_kernel": ("system/Image", True),
"os_initrd": ("system/initrd.img", True),
"data_disk": ("data/firecrab-data.raw", True),
"ssh_private_key": ("runtime/manager_ed25519", True),
"ssh_public_key": ("runtime/manager_ed25519.pub", False),
"ssh_known_hosts": ("runtime/known_hosts", False),
}
class QAError(Exception):
"""A safety gate, helper contract, or preservation check failed."""
def require(condition, detail):
if not condition:
raise QAError(detail)
def absolute_path(value, label):
path = Path(value).expanduser()
require(path.is_absolute(), f"{label} must be an absolute path: {path}")
return Path(os.path.abspath(path))
def reject_symlink_components(path, allow_missing=False):
"""Reject every existing symlink in a path, including ancestor links."""
path = Path(os.path.abspath(path))
current = Path(path.anchor)
for part in path.parts[1:]:
current /= part
try:
mode = current.lstat().st_mode
except FileNotFoundError:
if allow_missing:
continue
raise QAError(f"path component does not exist: {current}") from None
except OSError as error:
raise QAError(f"could not inspect path component {current}: {error}") from error
require(not stat.S_ISLNK(mode), f"symlinked path is not allowed: {current}")
def path_overlaps(left, right):
return left == right or left in right.parents or right in left.parents
def actual_user_homes():
homes = []
try:
homes.append(Path(pwd.getpwuid(os.getuid()).pw_dir))
except (KeyError, OSError):
pass
homes.append(Path.home())
unique = []
for home in homes:
normalized = Path(os.path.abspath(home)).resolve(strict=False)
if normalized not in unique:
unique.append(normalized)
return unique
def normalized_existing(path, label):
reject_symlink_components(path)
try:
return path.resolve(strict=True)
except OSError as error:
raise QAError(f"could not resolve {label} {path}: {error}") from error
def protected_homes(user_home=None, environ=None):
environ = os.environ if environ is None else environ
homes = [home / DEFAULT_MANAGED_SUFFIX for home in (
actual_user_homes() if user_home is None else [Path(user_home)])]
configured = environ.get("FIRECRAB_MICROMANAGER_HOME")
if configured:
configured_path = Path(configured).expanduser()
if not configured_path.is_absolute():
configured_path = Path.cwd() / configured_path
homes.append(configured_path)
normalized = []
for home in homes:
candidate = Path(os.path.abspath(home))
try:
candidate = candidate.resolve(strict=False)
except OSError as error:
raise QAError(f"could not resolve protected managed home {home}: {error}") from error
if candidate not in normalized:
normalized.append(candidate)
return normalized
def validate_fixture_home(path, user_home=None, environ=None):
path = absolute_path(path, "--fixture-home")
fixture = normalized_existing(path, "fixture home")
require(fixture.is_dir(), f"fixture home is not a directory: {fixture}")
for protected in protected_homes(user_home, environ):
require(not path_overlaps(fixture, protected),
f"fixture home overlaps a managed installation home: {protected}")
return fixture
def require_fixture_marker(home):
marker = home / FIXTURE_MARKER
required_file(marker)
try:
contents = marker.read_text(encoding="utf-8")
except (OSError, UnicodeError) as error:
raise QAError(f"could not read fixture marker {marker}: {error}") from error
require(contents == FIXTURE_MARKER_CONTENT,
f"fixture marker has unexpected content: {marker}")
def validate_helper(path):
path = absolute_path(path, "--helper")
helper = normalized_existing(path, "helper")
require(helper.is_file(), f"helper is not a regular file: {helper}")
require(os.access(helper, os.X_OK), f"helper is not executable: {helper}")
return helper
def required_file(path):
reject_symlink_components(path, allow_missing=True)
try:
info = path.lstat()
except OSError as error:
raise QAError(f"required fixture file is missing: {path}: {error}") from error
require(stat.S_ISREG(info.st_mode), f"fixture asset is not a regular file: {path}")
return info
def fixture_assets(home):
return {
name: (home / relative, required)
for name, (relative, required) in PROTECTED_FILES.items()
}
def select_staged_image(home):
repairs = home / "runtime/repairs"
reject_symlink_components(repairs)
require(repairs.is_dir(), f"staging directory is missing: {repairs}")
candidates = []
for entry in repairs.iterdir():
require(not entry.is_symlink(), f"symlinked repair staging entry is not allowed: {entry}")
if not entry.is_dir():
continue
candidate = entry / "debian-system.repaired.raw"
if candidate.exists() or candidate.is_symlink():
candidates.append(candidate)
require(len(candidates) == 1,
f"expected exactly one staged OS image under {repairs}; found {len(candidates)}")
staged = candidates[0]
info = required_file(staged)
require(info.st_nlink == 1,
f"staged OS image has {info.st_nlink} hard links; refusing ambiguous ownership: {staged}")
return staged.resolve(strict=True)
def assert_reference_isolated(reference, staged, home, user_home=None, environ=None):
info = required_file(reference)
require(info.st_nlink == 1,
f"reference OS image must not have hard links: {reference}")
protected = [(staged, True)] + list(fixture_assets(home).values())
for protected_home in protected_homes(user_home, environ):
protected.extend((protected_home / relative, required)
for relative, required in PROTECTED_FILES.values())
for path, _required in protected:
try:
if path.exists() or path.is_symlink():
require(not path.is_symlink(),
f"protected disk/SSH asset is a symlink: {path}")
other = path.stat()
require((info.st_dev, info.st_ino) != (other.st_dev, other.st_ino),
f"reference OS image aliases a protected disk/SSH asset: {path}")
except OSError as error:
raise QAError(f"could not inspect protected asset {path}: {error}") from error
def select_reference_image(staged, home, user_home=None, environ=None):
candidate = staged.with_name(REFERENCE_IMAGE_NAME)
if not candidate.exists() and not candidate.is_symlink():
return None
reject_symlink_components(candidate)
info = required_file(candidate)
require(info.st_nlink == 1,
f"reference OS image must not have hard links: {candidate}")
assert_reference_isolated(candidate, staged, home, user_home, environ)
return candidate.resolve(strict=True)
def identity(info):
return {
"device": info.st_dev,
"inode": info.st_ino,
"size": info.st_size,
"links": info.st_nlink,
"mode": stat.S_IMODE(info.st_mode),
}
def hash_file(path, info):
digest = hashlib.sha256()
flags = os.O_RDONLY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
descriptor = os.open(path, flags)
except OSError as error:
raise QAError(f"could not open fixture asset {path}: {error}") from error
try:
opened = os.fstat(descriptor)
require(stat.S_ISREG(opened.st_mode), f"fixture asset changed type while opening: {path}")
require((opened.st_dev, opened.st_ino) == (info.st_dev, info.st_ino),
f"fixture asset changed while opening: {path}")
with os.fdopen(descriptor, "rb", closefd=False) as source:
for block in iter(lambda: source.read(1024 * 1024), b""):
digest.update(block)
finished = os.fstat(descriptor)
require((finished.st_dev, finished.st_ino, finished.st_size,
finished.st_mtime_ns, finished.st_ctime_ns) ==
(opened.st_dev, opened.st_ino, opened.st_size,
opened.st_mtime_ns, opened.st_ctime_ns),
f"fixture asset changed while hashing: {path}")
finally:
os.close(descriptor)
return digest.hexdigest()
def file_record(path, required=True):
if not path.exists() and not path.is_symlink():
require(not required, f"required fixture asset is missing: {path}")
return None
info = required_file(path)
return {"path": str(path), "sha256": hash_file(path, info), "identity": identity(info)}
def capture_preservation(home, staged, reference=None):
protected = {}
for name, (path, required) in fixture_assets(home).items():
try:
protected[name] = file_record(path, required)
except (QAError, OSError) as error:
protected[name] = {"error": str(error)}
if reference is None:
protected["reference_os_disk"] = None
else:
try:
protected["reference_os_disk"] = file_record(reference)
except (QAError, OSError) as error:
protected["reference_os_disk"] = {"error": str(error)}
try:
staged_record = file_record(staged)
except (QAError, OSError) as error:
staged_record = {"error": str(error)}
return {
"protected": protected,
"staged_os_disk": staged_record,
}
def require_snapshot_complete(snapshot):
for name, (_relative, required) in PROTECTED_FILES.items():
record = snapshot["protected"][name]
if required:
require(isinstance(record, dict),
f"could not capture required preservation evidence for {name}: {record}")
else:
require(record is None or isinstance(record, dict),
f"could not capture optional preservation evidence for {name}: {record}")
if isinstance(record, dict) and "error" in record:
raise QAError(f"could not capture preservation evidence for {name}: {record['error']}")
reference = snapshot["protected"].get("reference_os_disk")
require(reference is None or
isinstance(reference, dict) and "error" not in reference,
f"could not capture reference OS image preservation evidence: {reference}")
require(isinstance(snapshot["staged_os_disk"], dict)
and "error" not in snapshot["staged_os_disk"],
f"could not capture staged OS image evidence: {snapshot['staged_os_disk']}")
def require_protected_preserved(before, after):
before_stage = before["staged_os_disk"]
require(isinstance(after["staged_os_disk"], dict)
and "error" not in after["staged_os_disk"],
f"staged OS image evidence is unavailable after helper: {after['staged_os_disk']}")
after_stage = after["staged_os_disk"]
before_identity = before_stage.get("identity", {}) if isinstance(before_stage, dict) else {}
after_identity = after_stage.get("identity", {})
require(before_identity.get("links") == 1 and after_identity.get("links") == 1,
"staged OS image must remain a single-link regular file")
require((before_identity.get("device"), before_identity.get("inode")) ==
(after_identity.get("device"), after_identity.get("inode")),
"helper replaced the staged OS image instead of repairing it in place")
require(before["protected"] == after["protected"],
"helper changed or replaced the original OS disk, data disk, or SSH assets")
def assert_staged_isolated(staged, home, user_home=None, environ=None):
staged_info = required_file(staged)
require(staged_info.st_nlink == 1,
f"staged OS image must not have hard links: {staged}")
protected = list(fixture_assets(home).values())
for protected_home in protected_homes(user_home, environ):
protected.extend((protected_home / relative, required)
for relative, required in PROTECTED_FILES.values())
for path, _required in protected:
try:
if path.exists() or path.is_symlink():
require(not path.is_symlink(), f"protected disk/SSH asset is a symlink: {path}")
info = path.stat()
require((staged_info.st_dev, staged_info.st_ino) != (info.st_dev, info.st_ino),
f"staged OS image aliases a protected disk/SSH asset: {path}")
except OSError as error:
raise QAError(f"could not inspect protected asset {path}: {error}") from error
def lsof_path():
executable = shutil.which("lsof")
require(executable is not None, "lsof is required to prove fixture images are not in use")
return executable
def require_image_idle(path, runner=None, executable=None):
runner = subprocess.run if runner is None else runner
executable = executable or lsof_path()
try:
result = runner([executable, "-t", str(path)], capture_output=True, text=True, timeout=15)
except (OSError, subprocess.TimeoutExpired) as error:
raise QAError(f"could not check whether fixture image is active ({path}): {error}") from error
owners = result.stdout.strip().splitlines()
require(not owners, f"fixture image is open by process(es) {', '.join(owners)}: {path}")
require(not result.stderr.strip(),
f"lsof could not prove fixture image is idle ({path}): {result.stderr.strip()}")
require(result.returncode in (0, 1),
f"lsof could not prove fixture image is idle ({path}); exit {result.returncode}: {result.stderr.strip()}")
return {"path": str(path), "exit_code": result.returncode, "owners": owners}
def validate_helper_result(stdout, helper_exit, expect_restored_files=None):
require(helper_exit == 0, f"repair helper exited with status {helper_exit}")
if isinstance(stdout, bytes):
try:
stdout = stdout.decode("utf-8")
except UnicodeDecodeError as error:
raise QAError(f"helper stdout is not UTF-8 JSON: {error}") from error
try:
result = json.loads(stdout)
except (json.JSONDecodeError, TypeError) as error:
raise QAError(f"helper stdout must contain one JSON object: {error}") from error
require(isinstance(result, dict), "helper stdout JSON is not an object")
fields = set(result)
require(fields in (BASE_RESULT_KEYS, RESULT_KEYS),
f"helper JSON fields differ from the contract: {sorted(result)}")
result.setdefault("restored_files", 0)
for name in ("check_exit", "verify_exit"):
value = result[name]
require(type(value) is int and 0 <= value <= 255,
f"{name} must be an integer exit status")
require(result["check_exit"] in (*range(8), 12),
"check_exit must be 0 through 7 or 12 (aborted with uncorrected errors)")
require(type(result["restored_files"]) is int and result["restored_files"] >= 0,
"restored_files must be a nonnegative integer")
repair_exit = result["repair_exit"]
require(repair_exit is None or
(type(repair_exit) is int and 0 <= repair_exit <= 3),
"repair_exit must be null or an integer from 0 through 3")
require(type(result["changed"]) is bool, "changed must be a JSON boolean")
require(result["verify_exit"] == 0,
f"helper verification failed with exit {result['verify_exit']}")
if result["check_exit"] == 0:
require(result["repair_exit"] is None,
"clean filesystem must skip filesystem repair")
else:
require(result["repair_exit"] is not None,
"filesystem repair requires a repair exit")
expected_changed = result["repair_exit"] is not None or result["restored_files"] > 0
require(result["changed"] is expected_changed,
"changed must report filesystem repair or restored files")
if expect_restored_files is not None:
require(result["restored_files"] == expect_restored_files,
f"expected {expect_restored_files} restored files, got {result['restored_files']}")
return result
def write_json(path, value):
temporary = path.with_name(path.name + ".tmp")
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
temporary.chmod(0o600)
temporary.replace(path)
def as_bytes(output):
if output is None:
return b""
return output if isinstance(output, bytes) else str(output).encode("utf-8", errors="replace")
class QA:
def __init__(self, helper, fixture_home, results_dir, *, user_home=None, environ=None,
expect_restored_files=None):
self.environ = dict(os.environ if environ is None else environ)
self.user_home = None if user_home is None else Path(user_home)
self.fixture_input = absolute_path(fixture_home, "--fixture-home")
self.helper_input = absolute_path(helper, "--helper")
self.fixture_home = None
self.helper = None
self.staged = None
self.reference = None
require(expect_restored_files is None or
(type(expect_restored_files) is int and expect_restored_files >= 0),
"expected restored file count must be a nonnegative integer")
self.expect_restored_files = expect_restored_files
self.results = absolute_path(results_dir, "--results-dir")
reject_symlink_components(self.results, allow_missing=True)
require(not self.results.exists(), f"results directory already exists: {self.results}")
fixture_target = self.fixture_input.resolve(strict=False)
for protected in [self.fixture_input, fixture_target,
*protected_homes(self.user_home, self.environ)]:
protected = Path(os.path.abspath(protected)).resolve(strict=False)
require(not path_overlaps(self.results, protected),
f"results directory overlaps a protected home: {protected}")
self.results.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
reject_symlink_components(self.results.parent)
self.results.mkdir(mode=0o700, exist_ok=False)
self.results.chmod(0o700)
self.log = (self.results / "run.log").open("w", encoding="utf-8", buffering=1)
self.before = None
self.after = None
self.summary = {
"status": "RUNNING",
"started_at": datetime.now(timezone.utc).isoformat(),
"platform": platform.platform(),
"helper": str(self.helper_input),
"fixture_home": str(self.fixture_input),
"results_dir": str(self.results),
"checks": [],
"helper_exit": None,
"preservation": "NOT_RUN",
}
if expect_restored_files is not None:
self.summary["expected_restored_files"] = expect_restored_files
def log_line(self, message):
self.log.write(message.rstrip("\n") + "\n")
self.log.flush()
def check(self, item, name, operation):
entry = {"id": item, "name": name, "status": "RUNNING"}
self.summary["checks"].append(entry)
try:
result = operation()
except BaseException as error:
entry.update(status="FAILED", detail=str(error) or "interrupted")
raise
entry["status"] = "PASS"
return result
def preflight(self):
require(platform.system() == "Darwin", "OS repair E2E requires macOS")
self.fixture_home = validate_fixture_home(
self.fixture_input, self.user_home, self.environ)
require_fixture_marker(self.fixture_home)
self.helper = validate_helper(self.helper_input)
signature = subprocess.run(
["/usr/bin/codesign", "--verify", "--strict", str(self.helper)],
capture_output=True, text=True, timeout=15,
)
require(signature.returncode == 0,
f"helper code signature did not verify: {signature.stderr.strip() or signature.stdout.strip()}")
self.summary["helper_signature"] = "PASS"
self.staged = select_staged_image(self.fixture_home)
self.reference = select_reference_image(
self.staged, self.fixture_home, self.user_home, self.environ)
assets = fixture_assets(self.fixture_home)
for name in ("fixture_marker", "original_os_disk", "os_kernel", "os_initrd",
"data_disk", "ssh_private_key"):
required_file(assets[name][0])
assert_staged_isolated(self.staged, self.fixture_home, self.user_home, self.environ)
image_paths = [assets[name][0] for name in ("original_os_disk", "data_disk")]
image_paths.append(self.staged)
idle_images = [require_image_idle(path) for path in image_paths]
self.summary.update({
"helper_resolved": str(self.helper),
"helper_identity": file_record(self.helper),
"staged_os_disk": str(self.staged),
"reference_os_disk": str(self.reference) if self.reference is not None else None,
"reference_image_check": "PASS" if self.reference is not None else "NOT_PRESENT",
"active_image_check": "PASS",
"active_images_before": idle_images,
})
def capture(self):
return capture_preservation(self.fixture_home, self.staged, self.reference)
def record_snapshot(self, name, snapshot):
write_json(self.results / name, snapshot)
def stop_child(self, process):
try:
os.killpg(process.pid, signal.SIGINT)
except (ProcessLookupError, PermissionError, AttributeError):
try:
process.send_signal(signal.SIGINT)
except (ProcessLookupError, PermissionError, AttributeError):
pass
try:
return process.communicate(timeout=HELPER_STOP_TIMEOUT)
except subprocess.TimeoutExpired:
try:
os.killpg(process.pid, signal.SIGKILL)
except (ProcessLookupError, PermissionError, AttributeError):
try:
process.kill()
except (ProcessLookupError, PermissionError, AttributeError):
pass
return process.communicate()
def invoke_helper(self):
command = [str(self.helper), "repair-os", str(self.staged)]
if self.reference is not None:
command.extend(["--reference", str(self.reference)])
environment = dict(self.environ)
environment["FIRECRAB_MICROMANAGER_HOME"] = str(self.fixture_home)
self.log_line("$ " + shlex.join(command))
self.log_line("FIRECRAB_MICROMANAGER_HOME=" + str(self.fixture_home))
started = time.monotonic()
stdout = stderr = b""
process = None
try:
process = subprocess.Popen(
command,
cwd=self.fixture_home,
env=environment,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True,
)
stdout, stderr = process.communicate(timeout=HELPER_TIMEOUT)
except subprocess.TimeoutExpired:
self.summary["helper_timed_out"] = True
stdout, stderr = self.stop_child(process)
self.summary["helper_exit"] = process.returncode
self.write_helper_output(stdout, stderr)
idle_error = self.record_staged_idle_after()
detail = f"repair helper exceeded {HELPER_TIMEOUT} seconds"
if idle_error:
detail += f"; post-helper image idle check failed: {idle_error}"
raise QAError(detail)
except KeyboardInterrupt:
self.summary["helper_interrupted"] = True
if process is not None:
stdout, stderr = self.stop_child(process)
self.summary["helper_exit"] = process.returncode
self.write_helper_output(stdout, stderr)
self.record_staged_idle_after()
raise
except OSError as error:
self.summary["helper_launch_error"] = str(error)
self.write_helper_output(b"", b"")
self.record_staged_idle_after()
raise QAError(f"could not start repair helper: {error}") from error
finally:
self.summary["helper_elapsed_seconds"] = round(time.monotonic() - started, 3)
self.summary["helper_exit"] = process.returncode
self.write_helper_output(stdout, stderr)
idle_error = self.record_staged_idle_after()
require(idle_error is None,
f"post-helper image idle check failed: {idle_error}")
self.summary["result"] = validate_helper_result(
stdout, process.returncode, self.expect_restored_files)
return self.summary["result"]
def write_helper_output(self, stdout, stderr):
(self.results / "helper.stdout").write_bytes(as_bytes(stdout))
(self.results / "helper.stderr.log").write_bytes(as_bytes(stderr))
(self.results / "helper.stdout").chmod(0o600)
(self.results / "helper.stderr.log").chmod(0o600)
self.log_line(f"helper exit={self.summary.get('helper_exit')}")
def record_staged_idle_after(self):
try:
result = require_image_idle(self.staged)
except QAError as error:
detail = str(error) or type(error).__name__
self.summary["staged_image_idle_after"] = {
"status": "FAILED", "error": detail,
}
return detail
self.summary["staged_image_idle_after"] = {"status": "PASS", **result}
return None
def record_preservation(self):
if self.before is None:
return
try:
self.after = self.capture()
self.record_snapshot("preservation-after.json", self.after)
require_snapshot_complete(self.before)
require_protected_preserved(self.before, self.after)
self.summary["preservation"] = "PASS"
self.summary["staged_os_disk_changed"] = (
self.before["staged_os_disk"]["sha256"] !=
self.after["staged_os_disk"]["sha256"])
entry = {"id": "OSR-P", "name": "original disk/data/SSH preservation", "status": "PASS"}
except BaseException as error:
self.summary["preservation"] = "FAILED"
self.summary["preservation_error"] = str(error) or "interrupted"
entry = {
"id": "OSR-P", "name": "original disk/data/SSH preservation",
"status": "FAILED", "detail": str(error) or "interrupted",
}
if self.after is None:
write_json(self.results / "preservation-after-error.json", {"error": entry["detail"]})
self.summary["checks"].append(entry)
def execute(self):
code = 1
try:
self.check("OSR-G", "fixture isolation and active-image gate", self.preflight)
self.before = self.capture()
self.record_snapshot("preservation-before.json", self.before)
self.check("OSR-B", "complete preservation baseline", lambda: require_snapshot_complete(self.before))
self.check("OSR-H", "repair helper contract", self.invoke_helper)
code = 0
except KeyboardInterrupt:
self.summary["error"] = "interrupted"
code = 130
except BaseException as error:
self.summary["error"] = str(error) or type(error).__name__
code = 1
finally:
self.record_preservation()
if self.summary["preservation"] == "FAILED":
code = 1
self.summary["status"] = "PASS" if code == 0 else "FAILED"
self.summary["exit_code"] = code
self.summary["finished_at"] = datetime.now(timezone.utc).isoformat()
write_json(self.results / "summary.json", self.summary)
self.log_line(f"QA status={self.summary['status']} exit={code}")
self.log.close()
print(f"{self.summary['status']}: OS repair QA; evidence: {self.results}", flush=True)
return code
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--helper", required=True, help="absolute path to signed macOS helper")
parser.add_argument("--fixture-home", required=True,
help="isolated fixture home; installed/default homes are rejected")
parser.add_argument("--results-dir", required=True, help="new private evidence directory")
parser.add_argument("--expect-restored-files", type=int,
help="require this exact number of restored configuration files")
args = parser.parse_args(argv)
try:
qa = QA(args.helper, args.fixture_home, args.results_dir,
expect_restored_files=args.expect_restored_files)
except QAError as error:
print(f"FAILED OS repair QA setup: {error}", file=sys.stderr)
return 2
return qa.execute()
if __name__ == "__main__":
raise SystemExit(main())