# Tlamatini Author Banner — Angela López Mendoza · @angelahack1 """Bounded, installation-scoped recovery for the standalone setup clients. The journal lives under the installation's Temp directory. It contains paths and hashes, never configuration values. Backups remain private local files. An unfinished transaction is restored before another operation starts. """ from contextlib import contextmanager import ctypes import hashlib import json import os from pathlib import Path, PurePosixPath import shutil import stat import time RECOVERY_REL = "Temp/.setup-recovery" FILE_ACCESS_RETRY_SECONDS = 5 JOURNAL_REPLACE_RETRY_SECONDS = 2 def retry_file_access(action, path, status): """Allow short-lived scanners/closing handles to release a file, then stop.""" deadline = time.monotonic() + FILE_ACCESS_RETRY_SECONDS announced = False while True: try: return action() except PermissionError as exc: if not announced: status(f"Waiting briefly for Windows to release {Path(path).name}…") announced = True if time.monotonic() >= deadline: from setup_windows import file_lock_description owners = file_lock_description(path) raise PermissionError( f"Windows is preventing access to:\n{path}\n\n" + (f"Programs using this file: {owners}.\n" if owners else "") + "Close programs using this installation, check the file's permissions, and retry. " "No application files have been removed during recovery preparation." ) from exc time.sleep(min(.25, max(0, deadline - time.monotonic()))) def check_readable(path): # Test the open before copying gigabytes. A subsequent copy still checks # access and hashes: this is early diagnosis, not a promise against races. with open(path, "rb"): pass def checked_path(root, relative): path = PurePosixPath(relative) if (not relative or path.is_absolute() or path.as_posix() != relative or any(p in (".", "..") or p.rstrip(" .") != p for p in path.parts) or any(c in relative for c in '\\:<>"|?*') or any(ord(c) < 32 for c in relative)): raise ValueError(f"Unsafe recovery path: {relative}") result = Path(root).joinpath(*path.parts) return _check_redirects(Path(root), result) def _check_redirects(root, result): current = result while current != Path(root).parent: if current.is_symlink() or getattr(current, "is_junction", lambda: False)(): raise RuntimeError(f"Redirected setup path is not supported: {current}") current = current.parent return result def _literal_path(path): """Keep filenames from scandir exact (also past MAX_PATH) for Windows I/O.""" path = Path(path).absolute() if ".." in path.parts: raise ValueError(f"Unsafe removal path: {path}") raw = str(path) if os.name != "nt" or raw.startswith("\\\\?\\"): return path if raw.startswith("\\\\"): return Path("\\\\?\\UNC\\" + raw[2:]) return Path("\\\\?\\" + raw) def _removal_entry(root, name): # Older CMD-generated browser installations can coexist with the legitimate # cache under an NTFS name ending in spaces/dots. Only this known root # entry gets literal-name access; package and general journal paths stay strict. if os.name == "nt" and name.casefold() != "ms-playwright" and name.rstrip(" .").casefold() == "ms-playwright": literal_root = _literal_path(root) return _check_redirects(literal_root, literal_root / name) return checked_path(root, name) def clean_update_browser_caches(installation): """Clean only malformed cache siblings beside the installation being updated.""" root = Path(installation) if (os.name != "nt" or not root.is_absolute() or root == root.parent or root.resolve(strict=True) != root or Path(__file__).resolve().parent != root / "Temp" / "_update" / "staging" or not (root / "Tlamatini.exe").is_file()): raise ValueError("Invalid browser-cache update boundary") for entry in root.iterdir(): name = entry.name if name.casefold() != "ms-playwright" and name.rstrip(" .").casefold() == "ms-playwright": remove_tree(_removal_entry(root, name)) def digest(path): with open(path, "rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def atomic_json(path, document): pending = path.with_suffix(".new") with open(pending, "w", encoding="utf-8") as stream: json.dump(document, stream, ensure_ascii=True, indent=2) stream.flush() os.fsync(stream.fileno()) deadline = time.monotonic() + JOURNAL_REPLACE_RETRY_SECONDS while True: try: os.replace(pending, path) return except PermissionError: # Windows can briefly hold a just-written journal. Keep the old # durable document and the new one intact; never delete to replace. if time.monotonic() >= deadline: raise time.sleep(min(.05, max(0, deadline-time.monotonic()))) @contextmanager def setup_lock(): """One operation per Windows session: the shell registrations are shared.""" kernel = ctypes.WinDLL("kernel32", use_last_error=True) kernel.CreateMutexW.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_wchar_p] kernel.CreateMutexW.restype = ctypes.c_void_p kernel.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_ulong] kernel.ReleaseMutex.argtypes = [ctypes.c_void_p] kernel.CloseHandle.argtypes = [ctypes.c_void_p] handle = kernel.CreateMutexW(None, False, "Local\\XAIHT.Tlamatini.Setup.v1") if not handle: raise ctypes.WinError(ctypes.get_last_error()) acquired = False try: result = kernel.WaitForSingleObject(handle, 0) # WAIT_ABANDONED also transfers ownership; the journal handles recovery. if result not in (0, 0x80): raise RuntimeError("Another Tlamatini installation or removal is running. Close it or wait for it to finish, then retry.") acquired = True yield finally: if acquired: kernel.ReleaseMutex(handle) kernel.CloseHandle(handle) class Recovery: def __init__(self, target, status=lambda _message: None, progress=lambda _fraction: None, rollback_progress=lambda _fraction: None): self.target = Path(os.path.abspath(target)) self.folder = checked_path(self.target, RECOVERY_REL) self.journal = self.folder / "journal.json" self.status = status self.progress = progress self.rollback_progress = rollback_progress self.cleanup_deadline = None self.data = None @classmethod def pending(cls, target): try: recovery = cls(target) recovery.load() return recovery.data["state"] in ("active", "restoring") except (OSError, ValueError, KeyError, RuntimeError): return False @classmethod def removal_cleanup_pending(cls, target): try: recovery = cls(target) recovery.load() return (recovery.data.get("mode") == "removal" and recovery.data["state"] == "committed" and recovery.data.get("outcome") == "removed") except (OSError, ValueError, KeyError, RuntimeError): return False def load(self): with open(self.journal, encoding="utf-8") as stream: data = json.load(stream) if (data.get("schema") not in (1, 2) or os.path.normcase(data.get("target", "")) != os.path.normcase(str(self.target)) or data.get("state") not in ("preparing", "active", "restoring", "committed")): raise RuntimeError(f"Unrecognized recovery journal. Keep it for inspection: {self.folder}") for name in data["files"]: if name.casefold().startswith(RECOVERY_REL.casefold()): raise RuntimeError("Recovery journal cannot target itself") checked_path(self.target, name) checked_path(self.folder / "files", name) for name in data["directories"]: checked_path(self.target, name) for name in data.get("restore_directories", []): checked_path(self.target, name) if data.get("schema") == 2: if data.get("mode") != "removal" or not isinstance(data.get("moves"), dict): raise RuntimeError("Unrecognized removal recovery journal") for name, record in data["moves"].items(): if len(PurePosixPath(name).parts) != 1 or name.casefold() in {"temp", "agents"}: raise RuntimeError(f"Unsafe removal recovery entry: {name}") _removal_entry(self.target, name) _removal_entry(self.folder / "quarantine", name) if record.get("state") not in {"waiting", "moving", "staged", "restored"}: raise RuntimeError("Unrecognized removal move state") identity = record.get("identity") if not isinstance(identity, list) or len(identity) != 2 or not all(isinstance(x, int) for x in identity): raise RuntimeError("Invalid removal file identity") self.data = data def recover_previous(self): if not self.folder.exists(): return False if not self.journal.is_file(): raise RuntimeError(f"An incomplete recovery folder needs inspection: {self.folder}") self.load() if self.data["state"] in ("active", "restoring"): self.status("Restoring the interrupted operation before continuing…") self.rollback() else: # 'preparing' has never mutated the installation; 'committed' is done. self.discard() self.data = None return True def prepare(self, names, *, required_space=0, windows=True, restore_directories=(), created_directories=()): """Finish and verify ALL backup copies before allowing any mutation.""" if self.folder.exists(): raise RuntimeError("Recover the previous operation before starting another") names = sorted(set(names), key=str.casefold) records, directories, inspected_directories = {}, set(), set() total = 0 # ZIPs carry intentionally empty folders too. They have no file receipt # entry, but must still be removed if this installation rolls back. for name in created_directories: path = checked_path(self.target, name) if path.exists() and not path.is_dir(): raise RuntimeError(f"A directory destination is occupied by a file: {path}") relative = PurePosixPath(name) for item in (relative, *relative.parents): if item.as_posix() != "." and not checked_path(self.target, item.as_posix()).exists(): directories.add(item.as_posix()) for index, name in enumerate(names, 1): path = checked_path(self.target, name) if name.casefold().startswith(RECOVERY_REL.casefold()): raise RuntimeError("A package cannot overwrite setup recovery data") if path.is_dir(): raise RuntimeError(f"A file destination is occupied by a directory: {path}") exists = path.is_file() records[name] = {"existed": exists} if exists: retry_file_access(lambda: check_readable(path), path, self.status) total += path.stat().st_size for parent in PurePosixPath(name).parents: relative = parent.as_posix() # File paths are still independently checked for redirections. # Only deduplicate the unchanged parent-existence inventory. if relative != "." and relative not in inspected_directories: inspected_directories.add(relative) if not checked_path(self.target, relative).exists(): directories.add(relative) if index == 1 or index % 250 == 0 or index == len(names): self.status(f"Checking destination… {index:,} / {len(names):,} files") self.progress(.3 * index / max(len(names), 1)) anchor = self.target while not anchor.exists(): anchor = anchor.parent if shutil.disk_usage(anchor).free < total + required_space + 32*1024*1024: raise RuntimeError("Not enough free disk space for setup and its recovery copy. Free space on the destination drive and retry.") target_existed = self.target.exists() temp_existed = (self.target / "Temp").exists() self.folder.mkdir(parents=True) self.data = {"schema": 1, "target": str(self.target), "state": "preparing", "files": records, "directories": sorted(directories), "restore_directories": list(restore_directories), "target_existed": target_existed, "temp_existed": temp_existed} for name in self.data["restore_directories"]: checked_path(self.target, name) atomic_json(self.journal, self.data) try: copied_bytes = 0 self.status(f"Preparing recovery copy… 0 / {total / (1024*1024):,.1f} MB") for i, (name, record) in enumerate(records.items(), 1): if record["existed"]: source = checked_path(self.target, name) backup = checked_path(self.folder / "files", name) backup.parent.mkdir(parents=True, exist_ok=True) retry_file_access(lambda: shutil.copy2(source, backup), source, self.status) mode = backup.stat().st_mode backup.chmod(mode | stat.S_IWRITE) with open(backup, "r+b") as durable: os.fsync(durable.fileno()) backup.chmod(mode) record["sha256"] = digest(backup) if digest(source) != record["sha256"]: raise RuntimeError(f"File changed while preparing recovery: {name}") copied_bytes += source.stat().st_size if i == 1 or i % 100 == 0 or i == len(records): self.status(f"Preparing recovery… {i:,} / {len(records):,} files · " f"{copied_bytes / (1024*1024):,.1f} / {total / (1024*1024):,.1f} MB") fraction = copied_bytes / total if total else i / max(len(records), 1) self.progress(.3 + .7 * fraction) if windows: from setup_windows import WindowsSnapshot WindowsSnapshot(self.folder).capture() self.data["windows"] = True self.data["state"] = "active" atomic_json(self.journal, self.data) self.progress(1) except Exception: self.discard() raise def prepare_removal(self, kept_names, *, windows=True): """Journal same-volume renames; never copy/hash the installed payload. Only metadata edited by setup is copied. Temp is disposable cleanup, handled AFTER commit, so a cache-directory lock cannot undo removal. Schema 2 makes older setup clients refuse this journal, not discard it. """ kept = {name.casefold() for name in kept_names} | {"temp", "agents"} moves = {} for item in self.target.iterdir(): if item.name.casefold() in kept: continue path = _removal_entry(self.target, item.name) info = path.stat(follow_symlinks=False) moves[item.name] = {"state": "waiting", "identity": [info.st_dev, info.st_ino]} metadata = ["Tlamatini.lnk"] if (self.target / "agents").is_dir(): metadata += ["agents/.tlamatini-preserved-agents.json", "agents/_tlamatini_agents_manifest.json"] self.prepare(metadata, windows=windows) self.data.update(schema=2, mode="removal", moves=moves) atomic_json(self.journal, self.data) self.status(f"Recovery ready: {len(moves):,} application entries; no payload copy needed") @staticmethod def _identity(path): info = path.stat(follow_symlinks=False) return [info.st_dev, info.st_ino] def stage_removal(self, name): """Atomically set aside one validated root entry, retaining its identity.""" if self.data.get("mode") != "removal" or name not in self.data["moves"]: raise RuntimeError(f"Removal entry was not prepared: {name}") source = _removal_entry(self.target, name) destination = _removal_entry(self.folder / "quarantine", name) record = self.data["moves"][name] if self._identity(source) != record["identity"]: raise RuntimeError(f"Installation entry changed during removal: {source}") if destination.exists(): raise RuntimeError(f"Recovery destination is already occupied: {destination}") destination.parent.mkdir(exist_ok=True) if source.stat().st_dev != destination.parent.stat().st_dev: raise RuntimeError("Removal recovery must remain on the installation drive") record["state"] = "moving" atomic_json(self.journal, self.data) # No copy fallback, no replacement of an existing destination. source.rename(destination) record["state"] = "staged" atomic_json(self.journal, self.data) def _restore_moves(self): moves = list(self.data.get("moves", {}).items()) failures = [] for index, (name, record) in enumerate(reversed(moves), 1): if record["state"] in {"moving", "staged"}: try: source = _removal_entry(self.folder / "quarantine", name) destination = _removal_entry(self.target, name) if source.exists(): if destination.exists(): raise RuntimeError("Restoration destination is occupied; existing content was kept") if self._identity(source) != record["identity"]: raise RuntimeError("Recovery entry identity changed") source.rename(destination) elif not destination.exists() or self._identity(destination) != record["identity"]: raise RuntimeError("Neither the original entry nor its recovery copy was found") record["state"] = "restored" atomic_json(self.journal, self.data) except OSError as exc: failures.append(f"{name}: {exc}") except RuntimeError as exc: failures.append(f"{name}: {exc}") self.status(f"Restoring application entries… {index:,} / {len(moves):,}") self.rollback_progress(.8 * index / max(len(moves), 1)) if failures: raise RuntimeError("Recovery needs attention. Retained data: " + str(self.folder) + "\n\n" + "\n".join(failures[:10])) def rollback(self): if self.data is None: self.load() if self.data["state"] not in ("active", "restoring"): return self.data["state"] = "restoring" atomic_json(self.journal, self.data) self.rollback_progress(0) self._restore_moves() failures = [] for name in self.data.get("restore_directories", []): try: checked_path(self.target, name).mkdir(parents=True, exist_ok=True) except OSError as exc: failures.append(f"{name}: {exc}") for i, (name, record) in enumerate(self.data["files"].items(), 1): try: path = checked_path(self.target, name) if record["existed"]: backup = checked_path(self.folder / "files", name) if not backup.is_file() or digest(backup) != record["sha256"]: raise RuntimeError("Recovery copy is missing or damaged") if not path.is_file() or digest(path) != record["sha256"]: path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): path.chmod(stat.S_IWRITE | stat.S_IREAD) shutil.copy2(backup, path) if digest(path) != record["sha256"]: raise RuntimeError("Restored file did not pass verification") elif path.is_file(): path.chmod(stat.S_IWRITE | stat.S_IREAD) path.unlink() except Exception as exc: failures.append(f"{name}: {exc}") if i == 1 or i % 100 == 0 or i == len(self.data["files"]): self.status(f"Restoring previous files… {i:,} / {len(self.data['files']):,}") base = .8 if self.data.get("moves") else 0 self.rollback_progress(base + (.9-base) * i / max(len(self.data["files"]), 1)) if self.data.get("windows"): self.status("Restoring Windows registrations…") self.rollback_progress(.9) try: from setup_windows import WindowsSnapshot WindowsSnapshot(self.folder).restore() except Exception as exc: failures.append(f"Windows integration: {exc}") for name in sorted(self.data["directories"], key=len, reverse=True): try: checked_path(self.target, name).rmdir() except OSError: pass # Never recursively erase a directory created by another actor. if failures: raise RuntimeError("Recovery needs attention. Its copies are retained at " + str(self.folder) + "\n\n" + "\n".join(failures[:10])) self.status("Finishing recovery cleanup…") self.rollback_progress(.95) # Restoration is complete. An interruption during housekeeping must # retry housekeeping, not require backup files already deleted here. self.data["state"] = "committed" self.data["outcome"] = "restored" atomic_json(self.journal, self.data) self.discard() self.rollback_progress(1) def commit(self, *, defer_cleanup=False): self.data["state"] = "committed" self.data["outcome"] = "removed" if self.data.get("mode") == "removal" else "installed" atomic_json(self.journal, self.data) if defer_cleanup: return "" try: self.discard() return "" except OSError as exc: # Never roll a completed installation back just because housekeeping failed. return (f"Temporary recovery copy could not be cleared: {self.folder}. " f"{exc}. Cleanup will be retried on the next setup run.") def discard(self): folder = checked_path(self.target, RECOVERY_REL) if folder.exists(): # Keep the durable journal until every payload has been cleared. # A locked file must not leave an unrecognizable partial journal. for child in folder.iterdir(): if child.name == "journal.json": continue remove_tree(child, self.status, deadline=self.cleanup_deadline) self.journal.unlink(missing_ok=True) folder.rmdir() if self.data and not self.data.get("temp_existed", True): try: (self.target / "Temp").rmdir() except OSError: pass if self.data and not self.data.get("target_existed", True): try: self.target.rmdir() except OSError: pass def remove_tree(path, status=lambda _message: None, *, deadline=None): """Delete an explicit tree once, without following links or junctions. Count actual removals while traversing, without a preliminary content scan, copy or hash pass. Callers retain their journal on failure/interruption. """ path = _literal_path(path) stack = [(path, False)] removed = 0 last = time.monotonic() while stack: if deadline is not None and time.monotonic() >= deadline: raise TimeoutError(f"Cleanup time limit reached; remaining files are at {path}") current, visited = stack.pop() try: info = current.lstat() except FileNotFoundError: continue redirected = stat.S_ISLNK(info.st_mode) or bool( getattr(info, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)) directory = stat.S_ISDIR(info.st_mode) if directory and not redirected and not visited: stack.append((current, True)) with os.scandir(current) as entries: stack.extend((Path(item.path), False) for item in entries) continue action = current.rmdir if directory else current.unlink try: action() except PermissionError: if redirected: raise current.chmod(stat.S_IWRITE | stat.S_IREAD) action() removed += 1 now = time.monotonic() if removed == 1 or removed % 250 == 0 or now-last >= 1: status(f"Deleting files and folders… {removed:,} removed ({path.name})") last = now status(f"Deleted {removed:,} files and folders ({path.name})") def removal_files(target, kept_names): """Enumerate rollback files without traversing redirected directories.""" target = Path(target) kept = {name.casefold() for name in kept_names} names = [] for parent, dirs, files in os.walk(target, followlinks=False): relative = Path(parent).relative_to(target) dirs[:] = [name for name in dirs if not ( (relative == Path('.') and name.casefold() in kept) or (relative.as_posix().casefold() == "temp" and name.casefold() == ".setup-recovery"))] for name in dirs + files: checked_path(target, (relative / name).as_posix()) names.extend((relative / name).as_posix() for name in files) for name in ("agents/.tlamatini-preserved-agents.json", "agents/_tlamatini_agents_manifest.json"): if (target / "agents").is_dir(): names.append(name) return names def removal_directories(target, kept_names): """Remember original empty directories as well as directories holding files.""" target = Path(target) kept = {name.casefold() for name in kept_names} names = [] for parent, dirs, _files in os.walk(target, followlinks=False): relative = Path(parent).relative_to(target) dirs[:] = [name for name in dirs if not ( (relative == Path('.') and name.casefold() in kept) or (relative.as_posix().casefold() == "temp" and name.casefold() == ".setup-recovery"))] for name in dirs: item = (relative / name).as_posix() checked_path(target, item) names.append(item) return names if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--clean-update-browser-caches", required=True) clean_update_browser_caches(parser.parse_args().clean_update_browser_caches)