#!/usr/bin/env python3 """Fail-closed comparison of deployed Mosaic tools to manifest-owned shipped tools.""" from __future__ import annotations import argparse import hashlib import os from pathlib import Path import stat import subprocess import sys def digest(path: Path) -> str: value = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): value.update(chunk) return value.hexdigest() def default_source_tools() -> Path: return Path(__file__).resolve().parents[2] def normalize_source(path: Path) -> Path: candidate = path.resolve() return candidate / "tools" if (candidate / "tools").is_dir() else candidate def assert_traversable_directory(path: Path) -> None: mode = stat.S_IMODE(path.stat(follow_symlinks=False).st_mode) # At least one principal class must have both read and search. This catches # mode-000 even for privileged reviewers for whom os.access() would lie. if not any(mode & read and mode & execute for read, execute in ((0o400, 0o100), (0o040, 0o010), (0o004, 0o001))): raise PermissionError(f"directory has no readable/searchable mode: {path}") def census(root: Path, *, reject_symlinks: bool) -> dict[str, Path]: result: dict[str, Path] = {} def onerror(error: OSError) -> None: raise error for current, directories, filenames in os.walk(root, topdown=True, followlinks=False, onerror=onerror): current_path = Path(current) assert_traversable_directory(current_path) for name in directories: entry = current_path / name if entry.is_symlink() and reject_symlinks: # A source symlink makes the shipped census incomplete. Deployed # aliases are assessed later only when they occupy a required # framework path; installed-only aliases remain operator state. raise OSError(f"symlinked directory is not an independent census entry: {entry}") for name in filenames: entry = current_path / name if entry.is_symlink(): if reject_symlinks: raise OSError(f"symlinked file is not an independent census entry: {entry}") result[entry.relative_to(root).as_posix()] = entry continue mode = entry.stat(follow_symlinks=False).st_mode if not stat.S_ISREG(mode): raise OSError(f"non-regular census entry: {entry}") if stat.S_IMODE(mode) & 0o444 == 0: raise PermissionError(f"file has no readable mode: {entry}") result[entry.relative_to(root).as_posix()] = entry return result def classify_with_manifest(source: Path, relatives: list[str]) -> dict[str, str]: framework = source.parent manifest = framework / "framework-manifest.txt" resolver = source / "_lib" / "manifest.sh" if not manifest.is_file() or not os.access(manifest, os.R_OK): raise OSError(f"ownership manifest is missing or unreadable: {manifest}") if not resolver.is_file() or not os.access(resolver, os.R_OK): raise OSError(f"canonical manifest resolver is missing or unreadable: {resolver}") payload = "".join(f"tools/{relative}\n" for relative in relatives) completed = subprocess.run( ["bash", str(resolver), "classify"], input=payload, text=True, capture_output=True, check=False, env={**os.environ, "MANIFEST_FILE": str(manifest)}, ) if completed.returncode != 0: detail = completed.stderr.strip() or f"resolver rc={completed.returncode}" raise OSError(f"ownership manifest failed canonical resolution: {detail}") classified: dict[str, str] = {} for line in completed.stdout.splitlines(): ownership, separator, manifest_path = line.partition("\t") if not separator or not manifest_path.startswith("tools/") or ownership not in {"framework", "operator"}: raise OSError(f"invalid canonical ownership output: {line!r}") relative = manifest_path.removeprefix("tools/") if relative in classified: raise OSError(f"duplicate canonical ownership output: {relative}") classified[relative] = ownership if set(classified) != set(relatives): raise OSError("canonical ownership output did not classify the complete source census") return classified def has_symlinked_component(root: Path, relative: str) -> bool: current = root for component in Path(relative).parts: current = current / component if current.is_symlink(): return True return False def main() -> int: parser = argparse.ArgumentParser(description="Detect deployed Mosaic framework-tool drift") parser.add_argument("--source-root", type=Path, default=Path(os.environ["MOSAIC_FRAMEWORK_SOURCE_ROOT"]) if os.environ.get("MOSAIC_FRAMEWORK_SOURCE_ROOT") else default_source_tools()) parser.add_argument("--installed-root", type=Path, default=Path(os.environ.get("MOSAIC_HOME", Path.home() / ".config/mosaic")) / "tools") parser.add_argument("--verbose", action="store_true") args = parser.parse_args() source = normalize_source(args.source_root) installed = args.installed_root.resolve() try: if not source.is_dir(): raise OSError(f"source tools missing: {source}") if not installed.is_dir(): raise OSError(f"installed tools missing: {installed}") if source.samefile(installed): raise OSError("source and installed roots identify the same filesystem object") source_files = census(source, reject_symlinks=True) if not source_files: raise OSError("source tools census is empty") ownership = classify_with_manifest(source, sorted(source_files)) required = sorted(relative for relative, owner in ownership.items() if owner == "framework") if not required: raise OSError("ownership manifest classifies zero shipped tools as framework-owned") installed_files = census(installed, reject_symlinks=False) except (OSError, PermissionError) as error: print(f"[framework-drift] CANNOT_ASSERT {error}", file=sys.stderr) return 2 in_sync: list[str] = [] stale: list[str] = [] not_installed: list[str] = [] unsafe_alias: list[str] = [] for relative in required: deployed = installed / relative if not deployed.is_file(): not_installed.append(relative) continue if has_symlinked_component(installed, relative): unsafe_alias.append(relative) continue try: if source_files[relative].samefile(deployed): unsafe_alias.append(relative) elif digest(source_files[relative]) == digest(deployed): in_sync.append(relative) else: stale.append(relative) except OSError as error: print(f"[framework-drift] CANNOT_ASSERT cannot compare {relative}: {error}", file=sys.stderr) return 2 source_relative = set(source_files) installed_only = sorted(set(installed_files) - source_relative) if args.verbose: for relative in in_sync: print(f"[framework-drift] IN_SYNC {relative}") for relative in stale: print(f"[framework-drift] STALE {relative}") for relative in not_installed: print(f"[framework-drift] NOT_INSTALLED {relative}") for relative in unsafe_alias: print(f"[framework-drift] UNSAFE_ALIAS {relative}") if args.verbose: for relative in installed_only: print(f"[framework-drift] INSTALLED_ONLY operator-or-unknown {relative}") print( "[framework-drift] summary " f"in-sync={len(in_sync)} stale={len(stale)} not-installed={len(not_installed)} " f"unsafe-alias={len(unsafe_alias)} installed-only={len(installed_only)}" ) print("[framework-drift] classification canonical framework-manifest ownership; installed-only=operator-or-unknown-preserved") if stale or not_installed or unsafe_alias: print("[framework-drift] FAIL deployed framework tools do not match independent shipped source; schedule a reviewed framework reseed", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())