#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import re import sys from collections import defaultdict from pathlib import Path from typing import Any # Ensure package imports work both for direct CLI execution and imports. _ROOT_DIR = Path(__file__).resolve().parents[1] if str(_ROOT_DIR) not in sys.path: sys.path.insert(0, str(_ROOT_DIR)) from scripts import create_manifest # noqa: E402 from scripts.packlib import ( # noqa: E402 ROOT, PackError, load_json, load_plan, load_state, packet_map, ) REQUIRED_FILES = [ "README_NL.md", "START_HERE.md", "AGENTS.md", "PROJECT_CONTRACT.md", "CURRENT_STATE.md", "BUILD_STATE.json", "VERSION_BASELINE.yaml", "ACCEPTANCE_MATRIX.md", "prompts/AUTONOMOUS_BUILD_LUNA.md", "implementation/WORK_PLAN.json", "docs/02-security/SECURITY_INVARIANTS.md", "contracts/schemas/project.schema.json", "contracts/schemas/connector-manifest.schema.json", "contracts/schemas/tool-manifest.schema.json", "contracts/schemas/policy.schema.json", "infrastructure/compose/docker-compose.blueprint.yml", ] KIND_REQUIRED = { "Workspace": {"apiVersion", "kind", "metadata", "spec"}, "Project": {"apiVersion", "kind", "metadata", "spec"}, "CapabilityPack": {"apiVersion", "kind", "metadata", "spec"}, "ConnectorTemplate": {"apiVersion", "kind", "metadata", "spec"}, "Policy": {"apiVersion", "kind", "metadata", "spec"}, "ToolManifest": {"apiVersion", "kind", "metadata", "spec"}, "ResourceManifest": {"apiVersion", "kind", "metadata", "spec"}, "PromptManifest": {"apiVersion", "kind", "metadata", "spec"}, } FORBIDDEN_TOOL_PREFIXES = ( "shell.", "filesystem.", "docker.raw", "database.execute", "secrets.", "http.request", ) CONTROLLED_GITEA_WRITE_TOOLS = frozenset( { "gitea.repository.create", "gitea.branchprotection.ensure", "gitea.portfolio.governance.apply", "gitea.branch.create", "gitea.issue.create", "gitea.issue.update", "gitea.issue.comment.create", "gitea.file.create", "gitea.file.update", "gitea.file.delete", "gitea.pullrequest.create", "gitea.pullrequest.update", "gitea.pullrequest.close", "gitea.pullrequest.validation.ensure", "gitea.changeset.apply", "gitea.validation.run", "gitea.workspace.run", "gitea.orchestration.start", "gitea.pullrequest.sync", "gitea.pullrequest.merge", "gitea.branch.delete", } ) CONTROLLED_PROJECTBRAIN_WRITE_TOOLS = frozenset( { "projectbrain.brainstorm.save", "projectbrain.knowledge.save", "projectbrain.documentation.save", "projectbrain.memory.save", "projectbrain.contextpack.create", "projectbrain.feedback.record", "projectbrain.delivery.create", "projectbrain.delivery.update", "projectbrain.delivery.start", "projectbrain.delivery.file.write", "projectbrain.delivery.file.delete", "projectbrain.delivery.changeset.apply", "projectbrain.delivery.validation.record", "projectbrain.delivery.validation.sync", "projectbrain.delivery.complete", "projectbrain.delivery.merge", "projectbrain.analysis.start", "projectbrain.analysis.cancel", "projectbrain.finding.issue.sync", "projectbrain.finding.investigation.start", "projectbrain.finding.investigation.save", "projectbrain.finding.suppress", "projectbrain.finding.delivery.create", "projectbrain.finding.verify", "projectbrain.portfolio.repository.reconcile", "projectbrain.portfolio.repository.exclusion.set", "projectbrain.autonomy.executor.heartbeat", "projectbrain.autonomy.configure", "projectbrain.autonomy.roadmap.import", "projectbrain.autonomy.resume", "projectbrain.autonomy.work.claim", "projectbrain.autonomy.work.report", } ) CONTROLLED_APPOPS_WRITE_TOOLS = frozenset( { "server.appops.self_upgrade.apply", "server.operation.request", "server.operation.execute", "server.job.cancel", "server.deployment.apply", "server.deployment.rollback", "server.deployment.target.enroll", "server.project.reconcile", } ) CONTROLLED_STITCH_WRITE_TOOLS = frozenset( { "stitch.project.create", "stitch.screen.edit", "stitch.screen.generate", "stitch.screen.variants", } ) CONTROLLED_OPENCODE_WRITE_TOOLS = frozenset( { "opencode.workspace.ensure", "opencode.job.start", "opencode.job.cancel", "opencode.delivery.start", "opencode.delivery.merge", "opencode.delivery.control.start", "opencode.delivery.control.retry", "opencode.delivery.control.cancel", } ) CONTROLLED_KNOWLEDGE_INGESTION_WRITE_TOOLS = frozenset( { "knowledge.ingestion.analyze", "knowledge.ingestion.text_proposal", "knowledge.ingestion.submit", } ) GITEA_LIFECYCLE_TOOLS = frozenset( {"gitea.pullrequest.sync", "gitea.pullrequest.merge", "gitea.branch.delete"} ) APPOPS_LIFECYCLE_TOOLS = frozenset( { "server.appops.self_upgrade.apply", "server.operation.execute", "server.deployment.apply", "server.deployment.rollback", "server.deployment.target.enroll", "server.project.reconcile", } ) OPENCODE_LIFECYCLE_TOOLS = frozenset( { "opencode.delivery.merge", "opencode.delivery.control.cancel", } ) # Strong patterns only; documentation may legitimately contain terms such as token. SECRET_PATTERNS = { "private key": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), "GitHub token": re.compile(r"\bgh[pousr]_[A-Za-z0-9]{30,}\b"), "OpenAI key": re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"), "JWT-like value": re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), } def add_error(errors: list[str], message: str) -> None: errors.append(message) def validate_required_files(errors: list[str]) -> None: for item in REQUIRED_FILES: if not (ROOT / item).is_file(): add_error(errors, f"missing required file: {item}") def validate_plan_and_state( errors: list[str], warnings: list[str] ) -> tuple[dict[str, Any], dict[str, Any]]: try: plan = load_plan() state = load_state() packets = packet_map(plan) except PackError as exc: add_error(errors, str(exc)) return {}, {} state_packets = state.get("packets", {}) if set(state_packets) != set(packets): add_error( errors, "packet ids differ between WORK_PLAN and BUILD_STATE; " f"plan-only={sorted(set(packets) - set(state_packets))}, " f"state-only={sorted(set(state_packets) - set(packets))}", ) allowed_statuses = {"pending", "in_progress", "completed", "blocked", "failed"} invalid_status = { pid: status for pid, status in state_packets.items() if status not in allowed_statuses } if invalid_status: add_error(errors, f"invalid packet statuses: {invalid_status}") in_progress = [pid for pid, status in state_packets.items() if status == "in_progress"] if len(in_progress) > 1: add_error(errors, f"at most one packet may be in progress: {in_progress}") graph: dict[str, list[str]] = {} for packet_id, packet in packets.items(): deps = packet.get("deps", []) if not isinstance(deps, list) or not all(isinstance(dep, str) for dep in deps): add_error(errors, f"{packet_id} deps must be a string array") deps = [] graph[packet_id] = deps for dep in deps: if dep not in packets: add_error(errors, f"{packet_id} references unknown dependency {dep}") if dep == packet_id: add_error(errors, f"{packet_id} depends on itself") packet_file = ROOT / "implementation" / "work-packets" / f"{packet_id}.md" if not packet_file.is_file(): add_error( errors, f"missing packet document: {packet_file.relative_to(ROOT)}", ) else: first_line = packet_file.read_text(encoding="utf-8").splitlines()[0:1] if not first_line or packet_id not in first_line[0]: add_error( errors, "packet document does not identify " f"{packet_id}: {packet_file.relative_to(ROOT)}", ) for ref in packet.get("read", []): if not isinstance(ref, str): add_error(errors, f"{packet_id} read entry is not a string: {ref!r}") continue if not (ROOT / ref).is_file(): add_error(errors, f"{packet_id} read reference does not exist: {ref}") visiting: set[str] = set() visited: set[str] = set() def visit(node: str, chain: list[str]) -> None: if node in visited: return if node in visiting: add_error( errors, f"work-packet dependency cycle: {' -> '.join(chain + [node])}", ) return visiting.add(node) for dep in graph.get(node, []): if dep in graph: visit(dep, chain + [node]) visiting.remove(node) visited.add(node) for node in graph: visit(node, []) if plan.get("default_model") != "gpt-5.6-luna": warnings.append("WORK_PLAN default_model is not gpt-5.6-luna") return plan, state def read_catalog_documents( errors: list[str], ) -> list[tuple[Path, dict[str, Any]]]: documents: list[tuple[Path, dict[str, Any]]] = [] for path in sorted((ROOT / "catalog").rglob("*.yaml")): try: value = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: add_error( errors, "catalog file is not JSON-compatible YAML: " f"{path.relative_to(ROOT)} line {exc.lineno}: {exc.msg}", ) continue if not isinstance(value, dict): add_error( errors, f"catalog document must be an object: {path.relative_to(ROOT)}", ) continue documents.append((path, value)) return documents def validate_catalog(errors: list[str], warnings: list[str]) -> dict[str, int]: del warnings # reserved for future non-fatal catalog checks docs = read_catalog_documents(errors) by_kind: dict[str, list[dict[str, Any]]] = defaultdict(list) seen: set[tuple[str, str]] = set() for path, value in docs: kind = value.get("kind") metadata = value.get("metadata") if kind not in KIND_REQUIRED: add_error( errors, f"unknown or missing catalog kind in {path.relative_to(ROOT)}: {kind!r}", ) continue missing = KIND_REQUIRED[kind] - set(value) if missing: add_error( errors, f"{path.relative_to(ROOT)} missing top-level keys: {sorted(missing)}", ) if value.get("apiVersion") != "hub.example.com/v1alpha1": add_error(errors, f"{path.relative_to(ROOT)} has unsupported apiVersion") if not isinstance(metadata, dict) or not isinstance(metadata.get("slug"), str): add_error(errors, f"{path.relative_to(ROOT)} requires metadata.slug") continue slug = metadata["slug"] key = (kind, slug) if key in seen: add_error(errors, f"duplicate catalog identity {kind}/{slug}") seen.add(key) by_kind[kind].append(value) workspaces = {d["metadata"]["slug"] for d in by_kind["Workspace"]} packs = {d["metadata"]["slug"] for d in by_kind["CapabilityPack"]} connectors = {d["metadata"]["slug"] for d in by_kind["ConnectorTemplate"]} endpoint_owner: dict[str, str] = {} for project in by_kind["Project"]: slug = project["metadata"]["slug"] spec = project.get("spec", {}) workspace = spec.get("workspace") if workspace not in workspaces: add_error( errors, f"Project/{slug} references unknown workspace {workspace!r}", ) for pack in spec.get("capabilityPacks", []): if pack not in packs: add_error( errors, f"Project/{slug} references unknown capability pack {pack!r}", ) envs = set(spec.get("environments", {})) for binding in spec.get("connectorBindings", []): ref = binding.get("ref") if isinstance(binding, dict) else None if ref not in connectors: add_error( errors, f"Project/{slug} references unknown connector {ref!r}", ) unknown_envs = ( set(binding.get("environments", [])) - envs if isinstance(binding, dict) else set() ) if unknown_envs: add_error( errors, f"Project/{slug} connector {ref!r} references unknown " f"environments {sorted(unknown_envs)}", ) for server in spec.get("virtualServers", []): if not isinstance(server, dict): add_error(errors, f"Project/{slug} virtual server must be an object") continue endpoint = server.get("endpointAlias") env = server.get("environment") if env not in envs: add_error( errors, f"Project/{slug} virtual server references unknown environment {env!r}", ) if endpoint in endpoint_owner: add_error( errors, f"endpoint alias {endpoint!r} reused by {endpoint_owner[endpoint]} and {slug}", ) elif isinstance(endpoint, str): endpoint_owner[endpoint] = slug for pattern in server.get("toolAllowPatterns", []): if isinstance(pattern, str) and pattern.startswith(FORBIDDEN_TOOL_PREFIXES): add_error( errors, f"Project/{slug} allows forbidden generic tool pattern {pattern!r}", ) if server.get("defaultDecision") != "deny": add_error(errors, f"Project/{slug} virtual server must default deny") documentation = spec.get("documentation", {}) if documentation.get("indexSecrets") is not False: add_error( errors, f"Project/{slug} must set documentation.indexSecrets=false", ) for manifest in by_kind["ToolManifest"]: slug = manifest["metadata"]["slug"] spec = manifest.get("spec", {}) canonical_name = spec.get("canonicalName") if not isinstance(canonical_name, str): add_error(errors, f"ToolManifest/{slug} requires spec.canonicalName") elif canonical_name.startswith(FORBIDDEN_TOOL_PREFIXES): add_error( errors, f"ToolManifest/{slug} exposes forbidden capability {canonical_name!r}" ) controlled_write = ( canonical_name in CONTROLLED_GITEA_WRITE_TOOLS or canonical_name in CONTROLLED_PROJECTBRAIN_WRITE_TOOLS or canonical_name in CONTROLLED_APPOPS_WRITE_TOOLS or canonical_name in CONTROLLED_STITCH_WRITE_TOOLS or canonical_name in CONTROLLED_OPENCODE_WRITE_TOOLS or canonical_name in CONTROLLED_KNOWLEDGE_INGESTION_WRITE_TOOLS ) if spec.get("readOnly") is not True and not controlled_write: add_error(errors, f"ToolManifest/{slug} is not an approved controlled write tool") expected_risks = ( {"R3"} if canonical_name in GITEA_LIFECYCLE_TOOLS or canonical_name in APPOPS_LIFECYCLE_TOOLS or canonical_name in OPENCODE_LIFECYCLE_TOOLS else ({"R2"} if controlled_write else {"R0", "R1"}) ) if spec.get("riskClass") not in expected_risks: add_error( errors, f"ToolManifest/{slug} has invalid risk class {spec.get('riskClass')!r}", ) if not isinstance(spec.get("inputSchema"), dict) or not isinstance( spec.get("outputSchema"), dict ): add_error(errors, f"ToolManifest/{slug} requires input and output schemas") if not by_kind["Policy"]: add_error(errors, "catalog contains no policies") if not any(p["metadata"]["slug"] == "global-deny" for p in by_kind["Policy"]): add_error( errors, "catalog/policies/global-deny.yaml is not represented as Policy/global-deny", ) return {kind: len(items) for kind, items in sorted(by_kind.items())} def validate_schemas_and_examples(errors: list[str]) -> tuple[int, int]: schema_count = 0 example_count = 0 for path in sorted((ROOT / "contracts" / "schemas").glob("*.json")): schema_count += 1 try: value = load_json(path) except PackError as exc: add_error(errors, str(exc)) continue if ( not isinstance(value, dict) or value.get("$schema") != "https://json-schema.org/draft/2020-12/schema" ): add_error( errors, f"schema missing draft 2020-12 declaration: {path.relative_to(ROOT)}", ) for path in sorted((ROOT / "contracts" / "examples").glob("*.json")): example_count += 1 try: load_json(path) except PackError as exc: add_error(errors, str(exc)) if schema_count < 6: add_error(errors, f"expected at least 6 contract schemas, found {schema_count}") return schema_count, example_count def validate_catalog_json_schemas(errors: list[str], warnings: list[str]) -> None: try: from jsonschema import Draft202012Validator, FormatChecker # type: ignore[import-untyped] from referencing import Registry, Resource except ImportError: warnings.append( "jsonschema/referencing not installed; full instance-schema validation " "deferred to WP-010" ) return schema_dir = ROOT / "contracts" / "schemas" kind_map = { "Workspace": "workspace.schema.json", "Project": "project.schema.json", "CapabilityPack": "capability-pack.schema.json", "ConnectorTemplate": "connector-manifest.schema.json", "Policy": "policy.schema.json", "ToolManifest": "tool-manifest.schema.json", "ResourceManifest": "resource-manifest.schema.json", "PromptManifest": "prompt-manifest.schema.json", } registry = Registry() schemas: dict[str, dict[str, Any]] = {} for path in sorted(schema_dir.glob("*.json")): schema = load_json(path) if not isinstance(schema, dict): add_error(errors, f"schema is not an object: {path.relative_to(ROOT)}") continue schemas[path.name] = schema resource = Resource.from_contents(schema) registry = registry.with_resource(path.resolve().as_uri(), resource) if isinstance(schema.get("$id"), str): registry = registry.with_resource(schema["$id"], resource) for path, document in read_catalog_documents(errors): kind = document.get("kind") schema_name = kind_map.get(kind) if isinstance(kind, str) else None if not schema_name or schema_name not in schemas: continue validator = Draft202012Validator( schemas[schema_name], registry=registry, format_checker=FormatChecker(), ) for issue in validator.iter_errors(document): location = "/".join(str(item) for item in issue.absolute_path) or "" add_error( errors, f"schema validation failed: {path.relative_to(ROOT)}:{location}: {issue.message}", ) operation_schema = schemas.get("operation.schema.json") for path in sorted((ROOT / "contracts" / "examples").glob("*.json")): document = load_json(path) if not isinstance(document, dict): add_error(errors, f"example is not an object: {path.relative_to(ROOT)}") continue kind = document.get("kind") schema_name = kind_map.get(kind) if isinstance(kind, str) else None schema = schemas.get(schema_name) if schema_name else operation_schema if schema is None: continue validator = Draft202012Validator( schema, registry=registry, format_checker=FormatChecker(), ) for issue in validator.iter_errors(document): location = "/".join(str(item) for item in issue.absolute_path) or "" add_error( errors, f"example schema validation failed: {path.relative_to(ROOT)}:{location}: " f"{issue.message}", ) def validate_compose_blueprint(errors: list[str], warnings: list[str]) -> None: try: import yaml # type: ignore[import-untyped] except ImportError: warnings.append("PyYAML not installed; Compose YAML parse deferred to WP-020") return path = ROOT / "infrastructure" / "compose" / "docker-compose.blueprint.yml" try: document = yaml.safe_load(path.read_text(encoding="utf-8")) except Exception as exc: # PyYAML exposes several parser subclasses. add_error(errors, f"invalid Compose blueprint YAML: {exc}") return if not isinstance(document, dict) or not isinstance(document.get("services"), dict): add_error(errors, "Compose blueprint must contain a services object") return services = document["services"] allowed_publishers = {"web", "dataplane-router", "inspector"} for name, service in services.items(): if not isinstance(service, dict): add_error(errors, f"Compose service {name} is not an object") continue if service.get("ports") and name not in allowed_publishers: add_error(errors, f"internal service unexpectedly publishes ports: {name}") for volume in service.get("volumes", []) or []: serialized = json.dumps(volume) if isinstance(volume, dict) else str(volume) if "/var/run/docker.sock" in serialized and name != "docker-read-broker": add_error(errors, f"Docker socket mounted outside broker: {name}") broker = services.get("docker-read-broker", {}) broker_volumes = "\n".join(str(item) for item in broker.get("volumes", [])) if "/var/run/docker.sock:/var/run/docker.sock:ro" not in broker_volumes: add_error(errors, "Docker read broker must mount the socket read-only") def validate_no_obvious_secrets(errors: list[str]) -> None: roots = [ ROOT / "catalog", ROOT / "contracts", ROOT / "infrastructure", ROOT / "prompts", ROOT / "scripts", ] for base in roots: for path in sorted(p for p in base.rglob("*") if p.is_file()): if path.suffix.lower() in {".png", ".jpg", ".jpeg", ".zip"}: continue text = path.read_text(encoding="utf-8", errors="replace") for label, pattern in SECRET_PATTERNS.items(): if pattern.search(text): add_error(errors, f"possible {label} in {path.relative_to(ROOT)}") def _parse_manifest_entries( manifest: dict[str, Any], errors: list[str] ) -> dict[str, dict[str, Any]]: entries = manifest.get("files", []) if not isinstance(entries, list): add_error(errors, "manifest files must be an array") return {} self_excluded = manifest.get("selfExcluded", "PACK_MANIFEST.json") by_path: dict[str, dict[str, Any]] = {} for entry in entries: if not isinstance(entry, dict) or not isinstance(entry.get("path"), str): add_error(errors, "manifest entry without a string path") continue rel = entry["path"] if rel == self_excluded: add_error(errors, f"manifest illegally includes its own excluded path: {rel}") continue if rel in by_path: add_error(errors, f"duplicate manifest path: {rel}") continue by_path[rel] = entry return by_path def validate_manifest(errors: list[str], warnings: list[str], repo_root: Path = ROOT) -> None: """Validate PACK_MANIFEST.json against the live Git repository at HEAD. Every check here is deliberately Git-native (blob content, digest, porcelain status) rather than raw filesystem bytes, so that a Windows checkout (CRLF) and a Linux checkout (LF) of the identical commit validate identically. See scripts/create_manifest.py and .gitattributes for the full rationale. """ path = repo_root / "PACK_MANIFEST.json" if not path.exists(): warnings.append( "PACK_MANIFEST.json not generated yet; run scripts/create_manifest.py before packaging" ) return try: manifest = load_json(path) except PackError as exc: add_error(errors, str(exc)) return if not isinstance(manifest, dict): add_error(errors, "PACK_MANIFEST.json must be a JSON object") return source = manifest.get("source") if not isinstance(source, dict) or not isinstance(source.get("contentDigest"), str): add_error( errors, "PACK_MANIFEST.json has no source.contentDigest; regenerate " "with scripts/create_manifest.py (schemaVersion 2.0+ required)", ) return manifest_digest = source["contentDigest"] entries = manifest.get("files", []) if not isinstance(entries, list): add_error(errors, "manifest files must be an array") return by_path = _parse_manifest_entries(manifest, errors) # The recorded top-level digest must itself be a true statement about # the "files" array it sits beside — a non-blocking sanity check, # reported alongside (not instead of) the itemized comparison below, so # a hand-edited individual entry is always named specifically rather # than only producing a generic "digest doesn't match" error. entries_digest = create_manifest.compute_content_digest(list(by_path.values())) if entries_digest != manifest_digest: add_error( errors, "PACK_MANIFEST.json is internally inconsistent: source.contentDigest " f"({manifest_digest}) does not match a fresh digest of its own " f"files array ({entries_digest}) — the manifest was hand-edited " "or corrupted after generation.", ) # Any uncommitted change to a tracked file means the manifest — which # only ever describes committed content — cannot be trusted as a # description of "clean Git content". Untracked files are irrelevant # here: they can never enter the manifest in the first place. dirty = create_manifest.list_dirty_tracked_paths(repo_root) if dirty: add_error( errors, "working tree has uncommitted changes to tracked files, so the " f"manifest cannot be trusted as clean committed content: {dirty}", ) return try: head_commit = create_manifest.resolve_commit("HEAD", repo_root) except create_manifest.GitError as exc: add_error(errors, f"cannot verify manifest against Git: {exc}") return # The itemized comparison below is the actual source of truth for # staleness and tampering — it compares every manifest entry directly # against live Git blob content, never against a raw `HEAD^{tree}` hash # or a commit SHA. That is deliberate, for two independent reasons: # (1) PACK_MANIFEST.json is itself a tracked, committed file, so the raw # tree hash always includes its own blob — which changes on every # regeneration via `generatedAt`. Gating on the raw tree hash is # self-referentially impossible to satisfy. # (2) a no-conflict merge commit (e.g. a Gitea PR merge) gets a new # commit SHA even when it changes no tracked content; gating on # commit SHA would spuriously demand a regeneration commit after # every such merge (this happened once already: commit 95329ca). # Comparing actual (path, checksum) pairs avoids both: it is stable # across manifest-only commits and no-op merges, and flags a path if and # only if that path's tracked content actually changed. self_excluded = manifest.get("selfExcluded", "PACK_MANIFEST.json") tracked = set(create_manifest.list_tracked_paths(head_commit, repo_root)) - {self_excluded} manifest_paths = set(by_path) if missing := sorted(tracked - manifest_paths): add_error(errors, f"manifest omits Git-tracked files: {missing}") if extra := sorted(manifest_paths - tracked): add_error(errors, f"manifest contains files no longer tracked by Git: {extra}") verifiable = sorted(manifest_paths & tracked) contents = create_manifest.batch_blob_contents(head_commit, verifiable, repo_root) total_bytes = 0 for rel in verifiable: entry = by_path[rel] data = contents.get(rel) if data is None: add_error(errors, f"could not read Git blob content for {rel}") continue total_bytes += len(data) if entry.get("bytes") != len(data): add_error(errors, f"manifest byte count mismatch: {rel}") digest = hashlib.sha256(data).hexdigest() if digest != entry.get("sha256"): add_error(errors, f"manifest checksum mismatch: {rel}") if not errors: if manifest.get("fileCount") != len(entries): add_error(errors, "manifest fileCount does not match files array") if manifest.get("totalBytes") != total_bytes: add_error(errors, "manifest totalBytes does not match Git blob contents") def validate_manifest_against_directory(directory: Path, manifest_root: Path = ROOT) -> list[str]: """Validate a plain directory (no Git required) against PACK_MANIFEST.json. Used to prove a clean-room artifact — a `git archive` extraction, a fresh clone, or a release-export directory — contains exactly the bytes the manifest claims, on any platform, without needing Git present in that directory. This works because `git archive` and the manifest's own generator both source content from Git blobs, which are never subject to checkout-time line-ending conversion. """ errors: list[str] = [] manifest_path = manifest_root / "PACK_MANIFEST.json" if not manifest_path.is_file(): return [f"no PACK_MANIFEST.json at {manifest_path}"] manifest = load_json(manifest_path) if not isinstance(manifest, dict): return ["PACK_MANIFEST.json must be a JSON object"] by_path = _parse_manifest_entries(manifest, errors) self_excluded = manifest.get("selfExcluded", "PACK_MANIFEST.json") # The directory being verified may legitimately contain the manifest # file itself — e.g. a `git archive` extraction includes it, since # PACK_MANIFEST.json is itself tracked and committed — even though the # manifest's own file list never includes it (see selfExcluded). Its # presence is expected, not a discrepancy. on_disk = { candidate.relative_to(directory).as_posix() for candidate in directory.rglob("*") if candidate.is_file() } - {self_excluded} manifest_paths = set(by_path) if missing := sorted(manifest_paths - on_disk): errors.append(f"directory is missing manifest files: {missing}") if extra := sorted(on_disk - manifest_paths): errors.append(f"directory contains files absent from the manifest: {extra}") for rel in sorted(manifest_paths & on_disk): entry = by_path[rel] file_path = directory / rel data = file_path.read_bytes() if entry.get("bytes") != len(data): errors.append(f"byte count mismatch outside repository: {rel}") digest = hashlib.sha256(data).hexdigest() if digest != entry.get("sha256"): errors.append(f"checksum mismatch outside repository: {rel}") return errors def main() -> int: parser = argparse.ArgumentParser(description="Validate the ITWorx MCP Hub design/build pack.") parser.add_argument("--json", action="store_true", help="emit machine-readable output") parser.add_argument( "--require-manifest", action="store_true", help="fail when PACK_MANIFEST.json is absent", ) parser.add_argument( "--verify-tree", metavar="DIR", help=( "Skip the full design-pack gate and instead verify that DIR " "(e.g. a `git archive` extraction or release-export directory) " "contains exactly the files and checksums PACK_MANIFEST.json " "records. No Git repository is required inside DIR." ), ) args = parser.parse_args() if args.verify_tree: tree_errors = validate_manifest_against_directory(Path(args.verify_tree).resolve()) if args.json: print(json.dumps({"ok": not tree_errors, "errors": tree_errors}, indent=2)) else: status = "PASS" if not tree_errors else "FAIL" print(f"ITWorx MCP Hub manifest tree verification: {status}") for error in tree_errors: print(f"ERROR: {error}") return 0 if not tree_errors else 1 errors: list[str] = [] warnings: list[str] = [] validate_required_files(errors) plan, _state = validate_plan_and_state(errors, warnings) catalog_counts = validate_catalog(errors, warnings) schema_count, example_count = validate_schemas_and_examples(errors) validate_catalog_json_schemas(errors, warnings) validate_compose_blueprint(errors, warnings) validate_no_obvious_secrets(errors) validate_manifest(errors, warnings) if args.require_manifest and not (ROOT / "PACK_MANIFEST.json").is_file(): add_error(errors, "PACK_MANIFEST.json is required") summary = { "ok": not errors, "errors": errors, "warnings": warnings, "packetCount": len(plan.get("packets", [])) if isinstance(plan, dict) else 0, "catalogCounts": catalog_counts, "schemaCount": schema_count, "exampleCount": example_count, } if args.json: print(json.dumps(summary, indent=2)) else: status = "PASS" if not errors else "FAIL" print(f"ITWorx MCP Hub design-pack validation: {status}") print( f"packets={summary['packetCount']} schemas={schema_count} " f"examples={example_count} catalog={catalog_counts}" ) for warning in warnings: print(f"WARNING: {warning}") for error in errors: print(f"ERROR: {error}") return 0 if not errors else 1 if __name__ == "__main__": sys.exit(main())