86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
"""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
|