diff --git a/ptlinux/import_state.py b/ptlinux/import_state.py new file mode 100644 index 0000000..1c273b6 --- /dev/null +++ b/ptlinux/import_state.py @@ -0,0 +1,85 @@ +"""Persistent trust state for game trees seeded from another installation.""" + +from __future__ import annotations + +from datetime import datetime +import json +import os +from pathlib import Path +from uuid import uuid4 + +from .paths import AppPaths + + +IMPORT_STATE_SCHEMA = 2 + + +def game_import_requires_verification(paths: AppPaths) -> bool: + """Return whether a managed imported tree still needs a full depot repair.""" + + manifest = paths.game_import_manifest + if not manifest.exists(): + return False + try: + value = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return True + if not isinstance(value, dict): + return True + build = value.get("verification_build") + completed_at = value.get("verification_completed_at") + return not ( + value.get("verification_required") is False + and isinstance(build, int) + and not isinstance(build, bool) + and build > 0 + and isinstance(completed_at, str) + and bool(completed_at.strip()) + ) + + +def mark_game_import_verified( + paths: AppPaths, + *, + build: int, + files_quarantined: int, + quarantine_directory: str | None, +) -> None: + """Record a completed full-manifest verification of an imported tree.""" + + if isinstance(build, bool) or build <= 0: + raise ValueError("verified import build must be positive") + if files_quarantined < 0: + raise ValueError("quarantined file count cannot be negative") + try: + value = json.loads(paths.game_import_manifest.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + value = {} + if not isinstance(value, dict): + value = {} + value.update( + { + "schema": IMPORT_STATE_SCHEMA, + "verification_required": False, + "verification_build": build, + "verification_completed_at": datetime.now().astimezone().isoformat(), + "files_quarantined": files_quarantined, + "quarantine_directory": quarantine_directory, + } + ) + _write_json(paths.game_import_manifest, value) + + +def _write_json(path: Path, value: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.{uuid4().hex}.tmp") + try: + with temporary.open("x", encoding="utf-8") as handle: + json.dump(value, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + temporary.replace(path) + except BaseException: + temporary.unlink(missing_ok=True) + raise