214 lines
9.6 KiB
Python
214 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Strict, dependency-free reader for the CASAN project manifest contract."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
|
|
class ManifestError(ValueError):
|
|
pass
|
|
|
|
|
|
def _fail(message: str) -> None:
|
|
raise ManifestError(f"CASAN_PROJECT_MANIFEST_INVALID: {message}")
|
|
|
|
|
|
def safe_relative(value: object, label: str) -> str:
|
|
if not isinstance(value, str) or not value or "\x00" in value or "\\" in value or os.path.isabs(value):
|
|
_fail(f"{label} must be a non-empty repository-relative POSIX path")
|
|
normalized = os.path.normpath(value).replace(os.sep, "/")
|
|
if normalized == ".." or normalized.startswith("../") or "/../" in normalized:
|
|
_fail(f"{label} escapes the repository root")
|
|
return normalized.removeprefix("./")
|
|
|
|
|
|
def within(root: str, value: object, label: str, must_exist: bool = True) -> str:
|
|
relative = safe_relative(value, label)
|
|
root_real = os.path.realpath(root)
|
|
candidate = os.path.realpath(os.path.join(root_real, relative)) if must_exist else os.path.abspath(os.path.join(root_real, relative))
|
|
if os.path.commonpath([root_real, candidate]) != root_real:
|
|
_fail(f"{label} resolves outside the repository root")
|
|
if must_exist and not os.path.exists(candidate):
|
|
_fail(f"{label} does not exist: {relative}")
|
|
return candidate
|
|
|
|
|
|
def _json(path: str, label: str) -> dict:
|
|
try:
|
|
with open(path, encoding="utf-8") as handle:
|
|
value = json.load(handle)
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
_fail(f"{label} is not valid JSON: {error}")
|
|
if not isinstance(value, dict):
|
|
_fail(f"{label} must be an object")
|
|
return value
|
|
|
|
|
|
def _commands(value: object, label: str, allowed: set[str]) -> list[list[str]]:
|
|
if not isinstance(value, list):
|
|
_fail(f"{label} must be an array")
|
|
result = []
|
|
for index, command in enumerate(value):
|
|
if not isinstance(command, list) or not command or any(not isinstance(part, str) or not part for part in command):
|
|
_fail(f"{label}[{index}] must be a non-empty argv array")
|
|
if command[0] not in allowed:
|
|
_fail(f"{label}[{index}] executable is not allowed: {command[0]}")
|
|
result.append(command.copy())
|
|
return result
|
|
|
|
|
|
def load(root: str, manifest_path: str | None = None, project_id: str | None = None) -> dict:
|
|
root = os.path.realpath(root)
|
|
selected = manifest_path or os.environ.get("CASAN_PROJECT_MANIFEST")
|
|
requested = project_id or os.environ.get("CASAN_PROJECT_ID")
|
|
if selected:
|
|
path = within(root, selected, "manifest")
|
|
elif requested:
|
|
registry = _json(os.path.join(root, "packages/casan-harness/level5/project-registry.json"), "project registry")
|
|
entry = next((item for item in registry.get("projects", []) if item.get("project_id") == requested), None)
|
|
if not entry:
|
|
_fail(f"project is not registered: {requested}")
|
|
selected = entry.get("manifest") or f"{safe_relative(entry.get('domain_root'), 'registry domain_root')}/project.manifest.json"
|
|
path = within(root, selected, "registered manifest")
|
|
else:
|
|
path = within(root, "apps/okr/domain/project.manifest.json", "default manifest")
|
|
|
|
raw = _json(path, "project manifest")
|
|
if raw.get("schema_version") != 1:
|
|
_fail("schema_version must be 1")
|
|
if not re.fullmatch(r"[a-z][a-z0-9-]{1,62}", str(raw.get("project_id", ""))):
|
|
_fail("project_id must be a lowercase slug")
|
|
feature = raw.get("feature")
|
|
if not isinstance(feature, dict) or not re.fullmatch(r"[0-9]{3}-[a-z0-9-]+", str(feature.get("id", ""))):
|
|
_fail("feature.id must match NNN-slug")
|
|
if not re.fullmatch(r"MOD-[0-9]{2,}", str(feature.get("module_id", ""))):
|
|
_fail("feature.module_id must match MOD-NN")
|
|
|
|
for key in ("domain_root", "requirements", "architecture", "quality_profile"):
|
|
raw[key] = safe_relative(raw.get(key), key)
|
|
within(root, raw[key], key)
|
|
raw["artifacts_root"] = safe_relative(raw.get("artifacts_root", "docs/output"), "artifacts_root")
|
|
|
|
source_roots = raw.get("source_roots")
|
|
if not isinstance(source_roots, list) or not source_roots:
|
|
_fail("source_roots must contain at least one path")
|
|
raw["source_roots"] = list(dict.fromkeys(safe_relative(item, "source_roots") for item in source_roots))
|
|
for item in raw["source_roots"]:
|
|
within(root, item, "source_root")
|
|
|
|
profile = _json(within(root, raw["quality_profile"], "quality_profile"), "quality profile")
|
|
allowed = set(profile.get("allowed_command_executables", []))
|
|
if profile.get("schema_version") != 1 or not allowed:
|
|
_fail("quality profile version or command allowlist is invalid")
|
|
raw["quality"] = profile
|
|
raw["commands"] = {
|
|
"build": _commands(raw.get("commands", {}).get("build"), "commands.build", allowed),
|
|
"test": _commands(raw.get("commands", {}).get("test"), "commands.test", allowed),
|
|
}
|
|
if profile.get("require_build_commands") and not raw["commands"]["build"]:
|
|
_fail("build commands are required by the quality profile")
|
|
if profile.get("require_test_commands") and not raw["commands"]["test"]:
|
|
_fail("test commands are required by the quality profile")
|
|
|
|
verification = raw.get("verification")
|
|
if not isinstance(verification, list) or (profile.get("require_verification_mapping") and not verification):
|
|
_fail("verification mapping is required")
|
|
normalized_rules = []
|
|
for index, rule in enumerate(verification):
|
|
if not isinstance(rule, dict):
|
|
_fail(f"verification[{index}] must be an object")
|
|
prefix = safe_relative(rule.get("path_prefix"), f"verification[{index}].path_prefix").rstrip("/") + "/"
|
|
normalized_rules.append({"path_prefix": prefix, "commands": _commands(rule.get("commands"), f"verification[{index}].commands", allowed)})
|
|
raw["verification"] = normalized_rules
|
|
if profile.get("fail_on_unmapped_source_root"):
|
|
for source in raw["source_roots"]:
|
|
prefix = source.rstrip("/") + "/"
|
|
if not any(prefix.startswith(rule["path_prefix"]) or rule["path_prefix"].startswith(prefix) for rule in normalized_rules):
|
|
_fail(f"source root has no verification rule: {source}")
|
|
|
|
raw["manifest_path"] = os.path.relpath(path, root).replace(os.sep, "/")
|
|
raw["root"] = root
|
|
return raw
|
|
|
|
|
|
def verification_commands(manifest: dict, files: list[str]) -> list[list[str]]:
|
|
commands: list[list[str]] = []
|
|
seen: set[tuple[str, ...]] = set()
|
|
for rule in manifest["verification"]:
|
|
if any(safe_relative(path, "changed file").startswith(rule["path_prefix"]) for path in files):
|
|
for command in rule["commands"]:
|
|
key = tuple(command)
|
|
if key not in seen:
|
|
commands.append(command.copy())
|
|
seen.add(key)
|
|
return commands
|
|
|
|
|
|
def run_commands(manifest: dict, kind: str) -> int:
|
|
if kind not in {"build", "test"}:
|
|
_fail(f"unsupported command kind: {kind}")
|
|
for command in manifest["commands"][kind]:
|
|
print(f"CASAN_PROJECT_COMMAND kind={kind} argv={json.dumps(command, ensure_ascii=False)}", flush=True)
|
|
timeout = int(os.environ.get("CASAN_PROJECT_COMMAND_TIMEOUT_SEC", "600"))
|
|
try:
|
|
result = subprocess.run(command, cwd=manifest["root"], check=False, timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
print(f"CASAN_PROJECT_COMMAND_TIMEOUT kind={kind} seconds={timeout}", file=sys.stderr)
|
|
return 124
|
|
if result.returncode != 0:
|
|
print(f"CASAN_PROJECT_COMMAND_FAILED kind={kind} rc={result.returncode}", file=sys.stderr)
|
|
return result.returncode
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("action", choices=["validate", "get", "domain-root", "run", "verification"])
|
|
parser.add_argument("--root", default=os.getcwd())
|
|
parser.add_argument("--manifest")
|
|
parser.add_argument("--project")
|
|
parser.add_argument("--kind", choices=["build", "test"])
|
|
parser.add_argument("--file", action="append", default=[])
|
|
parser.add_argument("--field", choices=["project_id", "requirements", "architecture", "domain_root", "quality_profile"])
|
|
args = parser.parse_args()
|
|
try:
|
|
manifest = load(args.root, args.manifest, args.project)
|
|
if args.action == "domain-root":
|
|
print(os.path.join(manifest["root"], manifest["domain_root"]))
|
|
return 0
|
|
if args.action == "get":
|
|
if not args.field:
|
|
_fail("--field is required for get")
|
|
print(manifest[args.field])
|
|
return 0
|
|
if args.action == "validate":
|
|
print(json.dumps({
|
|
"status": "valid",
|
|
"project_id": manifest["project_id"],
|
|
"manifest": manifest["manifest_path"],
|
|
"quality_profile": manifest["quality"]["profile_id"],
|
|
"build_commands": len(manifest["commands"]["build"]),
|
|
"test_commands": len(manifest["commands"]["test"]),
|
|
"verification_rules": len(manifest["verification"]),
|
|
}, ensure_ascii=False))
|
|
return 0
|
|
if args.action == "run":
|
|
if not args.kind:
|
|
_fail("--kind is required for run")
|
|
return run_commands(manifest, args.kind)
|
|
print(json.dumps(verification_commands(manifest, args.file), ensure_ascii=False))
|
|
return 0
|
|
except (ManifestError, OSError, ValueError) as error:
|
|
print(str(error), file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|