feat: harden CASAN production install lifecycle
This commit is contained in:
thanhnv
@@ -0,0 +1,205 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Copy one CASAN production component through the canonical runtime allowlist."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
|
||||
IGNORED_NAMES = {
|
||||
"__pycache__", ".DS_Store", "node_modules", "dist", "build", "coverage",
|
||||
}
|
||||
IGNORED_SUFFIXES = {".pyc", ".pyo", ".log", ".tmp"}
|
||||
|
||||
|
||||
class LayoutError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def load_layout(source_root: Path) -> dict:
|
||||
path = source_root / "packaging" / "runtime-layout.json"
|
||||
try:
|
||||
layout = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError) as error:
|
||||
raise LayoutError(f"cannot load {path}: {error}") from error
|
||||
if layout.get("schema_version") != 1:
|
||||
raise LayoutError("runtime layout schema_version must be 1")
|
||||
return layout
|
||||
|
||||
|
||||
def is_excluded(relative: Path, exclusions: set[str]) -> bool:
|
||||
normalized = relative.as_posix()
|
||||
if any(part in IGNORED_NAMES for part in relative.parts):
|
||||
return True
|
||||
if relative.suffix in IGNORED_SUFFIXES:
|
||||
return True
|
||||
return any(
|
||||
normalized == excluded or normalized.startswith(excluded.rstrip("/") + "/")
|
||||
for excluded in exclusions
|
||||
)
|
||||
|
||||
|
||||
def transformed_content(path: Path, transform: str) -> bytes:
|
||||
if transform != "empty_project_registry":
|
||||
raise LayoutError(f"unknown runtime transform: {transform}")
|
||||
try:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError) as error:
|
||||
raise LayoutError(f"cannot transform project registry: {error}") from error
|
||||
payload["projects"] = []
|
||||
return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
def copy_component(
|
||||
source_root: Path,
|
||||
destination_root: Path,
|
||||
name: str,
|
||||
preserved: set[str] | None = None,
|
||||
clean: bool = False) -> dict:
|
||||
layout = load_layout(source_root)
|
||||
component = layout.get("components", {}).get(name)
|
||||
if not isinstance(component, dict):
|
||||
raise LayoutError(f"unknown runtime component: {name}")
|
||||
|
||||
relative_source = Path(str(component.get("source", "")))
|
||||
source = (source_root / relative_source).resolve()
|
||||
destination = destination_root / relative_source
|
||||
if not source.is_dir():
|
||||
raise LayoutError(f"component source is missing: {source}")
|
||||
if source_root not in source.parents:
|
||||
raise LayoutError(f"component source escapes source root: {source}")
|
||||
|
||||
exclusions = {
|
||||
str(path).replace("\\", "/").strip("/")
|
||||
for path in component.get("exclude", [])
|
||||
if isinstance(path, str) and path.strip("/")
|
||||
}
|
||||
unsafe = [
|
||||
path for path in exclusions
|
||||
if Path(path).is_absolute() or ".." in Path(path).parts
|
||||
]
|
||||
if unsafe:
|
||||
raise LayoutError(
|
||||
f"runtime component {name} has unsafe exclusions: {', '.join(unsafe)}")
|
||||
transforms = component.get("transforms", {})
|
||||
if not isinstance(transforms, dict):
|
||||
raise LayoutError(f"runtime component {name} transforms must be an object")
|
||||
removed = 0
|
||||
if clean and destination.exists():
|
||||
for excluded in sorted(exclusions, reverse=True):
|
||||
candidate = destination / excluded
|
||||
if candidate.is_dir() and not candidate.is_symlink():
|
||||
shutil.rmtree(candidate)
|
||||
removed += 1
|
||||
elif candidate.exists() or candidate.is_symlink():
|
||||
candidate.unlink()
|
||||
removed += 1
|
||||
for candidate in sorted(destination.rglob("*"), reverse=True):
|
||||
if (candidate.name in IGNORED_NAMES or
|
||||
(candidate.is_file() and candidate.suffix in IGNORED_SUFFIXES)):
|
||||
if candidate.is_dir() and not candidate.is_symlink():
|
||||
shutil.rmtree(candidate)
|
||||
else:
|
||||
candidate.unlink()
|
||||
removed += 1
|
||||
copied = 0
|
||||
skipped = 0
|
||||
preserved = preserved or set()
|
||||
for path in sorted(source.rglob("*")):
|
||||
relative = path.relative_to(source)
|
||||
if is_excluded(relative, exclusions):
|
||||
skipped += path.is_file()
|
||||
continue
|
||||
target = destination / relative
|
||||
if path.is_dir():
|
||||
continue
|
||||
if not path.is_file():
|
||||
continue
|
||||
target_relative = (relative_source / relative).as_posix()
|
||||
if target_relative in preserved and target.exists():
|
||||
skipped += 1
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
transform = transforms.get(relative.as_posix())
|
||||
if transform:
|
||||
target.write_bytes(transformed_content(path, str(transform)))
|
||||
shutil.copymode(path, target)
|
||||
else:
|
||||
shutil.copy2(path, target)
|
||||
copied += 1
|
||||
|
||||
if clean and destination.exists():
|
||||
for candidate in sorted(destination.rglob("*"), reverse=True):
|
||||
if candidate.is_dir() and not candidate.is_symlink():
|
||||
try:
|
||||
candidate.rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
missing = [
|
||||
required for required in component.get("required", [])
|
||||
if not (destination / required).is_file()
|
||||
]
|
||||
if missing:
|
||||
raise LayoutError(
|
||||
f"runtime component {name} is incomplete; missing: {', '.join(missing)}")
|
||||
return {
|
||||
"component": name,
|
||||
"source": str(source),
|
||||
"destination": str(destination),
|
||||
"copied_files": copied,
|
||||
"excluded_files": skipped,
|
||||
"removed_paths": removed,
|
||||
}
|
||||
|
||||
|
||||
def parser() -> argparse.ArgumentParser:
|
||||
value = argparse.ArgumentParser(description=__doc__)
|
||||
value.add_argument("--source-root", required=True)
|
||||
value.add_argument("--destination-root", required=True)
|
||||
value.add_argument("--component", action="append", required=True)
|
||||
value.add_argument(
|
||||
"--preserve", action="append", default=[],
|
||||
help="destination-root-relative file to retain when it already exists")
|
||||
value.add_argument(
|
||||
"--clean", action="store_true",
|
||||
help="remove excluded source-only paths left by an older installation")
|
||||
value.add_argument("--json", action="store_true")
|
||||
return value
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parser().parse_args(argv)
|
||||
source_root = Path(args.source_root).expanduser().resolve()
|
||||
destination_root = Path(args.destination_root).expanduser().resolve()
|
||||
destination_root.mkdir(parents=True, exist_ok=True)
|
||||
preserved = {
|
||||
str(path).replace("\\", "/").strip("/")
|
||||
for path in args.preserve
|
||||
if str(path).strip("/")
|
||||
}
|
||||
try:
|
||||
results = [
|
||||
copy_component(
|
||||
source_root, destination_root, component, preserved, args.clean)
|
||||
for component in args.component
|
||||
]
|
||||
except LayoutError as error:
|
||||
sys.stderr.write(f"copy-runtime: {error}\n")
|
||||
return 1
|
||||
if args.json:
|
||||
print(json.dumps({"status": "ok", "components": results}, indent=2))
|
||||
else:
|
||||
for result in results:
|
||||
print(
|
||||
"copy-runtime: {component} copied={copied_files} "
|
||||
"excluded={excluded_files} removed={removed_paths}".format(**result))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user