195 lines
8.3 KiB
Python
Executable File
195 lines
8.3 KiB
Python
Executable File
#!/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())
|