Co-authored-by: coder3 <[email protected]>
This commit was merged in pull request #1195.
This commit is contained in:
coder3
@@ -0,0 +1,194 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user