#!/usr/bin/env python3 """Evidence-first product engineering controls for P2-WP11. The commands in this module validate or render evidence. They do not promote objects, mutate customer source data, switch traffic, or downgrade databases. """ from __future__ import annotations import argparse import ast import hashlib import json import re import subprocess import sys from pathlib import Path from typing import Any READINESS_CATEGORIES = ( "service", "task", "database", "graph", "object_storage", "external_dependency", ) PROMOTABLE_KINDS = {"asset", "standard", "ontology", "rule", "workflow"} ENVIRONMENT_ORDER = {"development": 10, "test": 20, "preproduction": 30} HEALTHY_STATES = {"healthy", "ready", "succeeded", "completed", "available"} SECRET_KEY_PATTERN = re.compile( r"(?:secret|password|token|credential|authorization|api[_-]?key)", re.I ) SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") COMPOSE_SERVICE_CATEGORIES = { "backend": "service", "frontend": "service", "runner": "task", "n8n": "task", "kestra": "task", "kestra-db-init": "task", "postgres": "database", "neo4j": "graph", "minio": "object_storage", "minio-init": "object_storage", "source-postgres": "external_dependency", "source-mysql": "external_dependency", } def _canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _sha256_file(path: Path) -> str: return _sha256_bytes(path.read_bytes()) def _tree_manifest(root: Path) -> dict[str, Any]: files: list[dict[str, str]] = [] if not root.exists(): return {"sha256": None, "file_count": 0} for path in sorted(root.rglob("*")): if not path.is_file(): continue relative = path.relative_to(root) if "__pycache__" in relative.parts or path.suffix in {".pyc", ".pyo"}: continue files.append( {"path": relative.as_posix(), "sha256": _sha256_file(path)} ) return { "sha256": _sha256_bytes(_canonical_bytes(files)), "file_count": len(files), } def _redact(value: Any) -> Any: if isinstance(value, dict): return { str(key): ( "[REDACTED]" if SECRET_KEY_PATTERN.search(str(key)) else _redact(item) ) for key, item in value.items() } if isinstance(value, list): return [_redact(item) for item in value] return value def _read_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"JSON object required: {path}") return value def _write_json(value: Any, output: str | None = None) -> None: rendered = json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n" if not output or output == "-": sys.stdout.write(rendered) return path = Path(output) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(rendered, encoding="utf-8") def validate_environment_profiles(profiles_dir: Path) -> dict[str, Any]: profiles: list[dict[str, Any]] = [] seen: dict[tuple[str, str], str] = {} collisions: list[dict[str, str]] = [] expected = set(ENVIRONMENT_ORDER) for path in sorted(profiles_dir.glob("*.json")): profile = _read_json(path) environment = str(profile.get("environment", "")) if environment not in ENVIRONMENT_ORDER: raise ValueError(f"unknown environment in {path.name}: {environment}") if int(profile.get("promotion_order", -1)) != ENVIRONMENT_ORDER[environment]: raise ValueError(f"invalid promotion order for {environment}") required_text = ("project_name", "network_name", "data_namespace") for field in required_text: if not str(profile.get(field, "")).strip(): raise ValueError(f"{field} is required for {environment}") key = (field, str(profile[field])) if key in seen: collisions.append( { "field": field, "value": str(profile[field]), "environments": f"{seen[key]},{environment}", } ) seen[key] = environment ports = profile.get("ports") if not isinstance(ports, dict) or not ports: raise ValueError(f"ports are required for {environment}") for name, raw_port in sorted(ports.items()): port = int(raw_port) if not 1 <= port <= 65535: raise ValueError(f"invalid port {name}={port} for {environment}") key = ("published_port", str(port)) if key in seen: collisions.append( { "field": "published_port", "value": str(port), "environments": f"{seen[key]},{environment}", } ) seen[key] = environment controls = profile.get("controls") if not isinstance(controls, dict) or "auth_required" not in controls: raise ValueError(f"controls.auth_required is required for {environment}") profiles.append( { "environment": environment, "promotion_order": ENVIRONMENT_ORDER[environment], "project_name": profile["project_name"], "network_name": profile["network_name"], "data_namespace": profile["data_namespace"], "ports": dict(sorted((str(k), int(v)) for k, v in ports.items())), "controls": controls, "profile_sha256": _sha256_file(path), "path": path.name, } ) found = {item["environment"] for item in profiles} if collisions: raise ValueError(f"environment collision detected: {collisions}") if found != expected: raise ValueError( "environment profile set must be development,test,preproduction; " f"found={','.join(sorted(found))}" ) profiles.sort(key=lambda item: int(item["promotion_order"])) differences: list[dict[str, Any]] = [] baseline = profiles[0] for profile in profiles[1:]: changed = [ field for field in ("project_name", "network_name", "data_namespace", "ports", "controls") if profile[field] != baseline[field] ] differences.append( { "from": baseline["environment"], "to": profile["environment"], "changed_fields": changed, } ) return { "schema_version": 1, "status": "ready", "promotion_path": [item["environment"] for item in profiles], "profiles": profiles, "collisions": [], "differences": differences, } def build_readiness_report(request: dict[str, Any]) -> dict[str, Any]: components = request.get("components") if not isinstance(components, list): raise ValueError("components must be a list") category_reports: dict[str, dict[str, Any]] = {} blocking: list[str] = [] redacted_components: list[dict[str, Any]] = [] for component in components: if not isinstance(component, dict): raise ValueError("each component must be an object") category = str(component.get("category", "")) if category not in READINESS_CATEGORIES: raise ValueError(f"unsupported readiness category: {category}") name = str(component.get("name", "")).strip() if not name: raise ValueError("component name is required") status = str(component.get("status", "unknown")).lower() required = bool(component.get("required", True)) ready = status in HEALTHY_STATES if required and not ready: blocking.append(name) redacted_components.append(_redact(component)) for category in READINESS_CATEGORIES: selected = [ item for item in redacted_components if item["category"] == category ] required = [item for item in selected if item.get("required", True)] blocked = [ item["name"] for item in required if str(item.get("status", "unknown")).lower() not in HEALTHY_STATES ] category_reports[category] = { "status": "ready" if required and not blocked else ( "blocked" if blocked else "unconfigured" ), "component_count": len(selected), "required_count": len(required), "blocked_components": blocked, } unconfigured = sorted( category for category, report in category_reports.items() if report["status"] == "unconfigured" ) return { "schema_version": 1, "environment": str(request.get("environment", "unknown")), "status": "blocked" if blocking or unconfigured else "ready", "blocking_components": sorted(blocking), "unconfigured_categories": unconfigured, "categories": category_reports, "components": redacted_components, } def build_compose_readiness_report( rows: list[dict[str, Any]], *, environment: str ) -> dict[str, Any]: components: list[dict[str, Any]] = [] for row in rows: service = str(row.get("Service", "")) category = COMPOSE_SERVICE_CATEGORIES.get(service) if not category: continue state = str(row.get("State", "unknown")).lower() health = str(row.get("Health", "")).lower() exit_code = int(row.get("ExitCode", -1)) completed = state == "exited" and exit_code == 0 and service.endswith("-init") ready = completed or (state == "running" and health in {"", "healthy"}) components.append( { "name": service, "category": category, "status": "completed" if completed else ("healthy" if ready else "unhealthy"), "required": True, "details": { "state": state, "health": health or "not_reported", "exit_code": exit_code, "container_id": str(row.get("ID", ""))[:12], }, } ) report = build_readiness_report( {"environment": environment, "components": components} ) report["collector"] = "docker_compose_ps" return report def collect_compose_readiness( compose_file: Path, *, project_name: str, environment: str, env_file: Path | None = None, ) -> dict[str, Any]: command = ["docker", "compose"] if env_file: command.extend(["--env-file", str(env_file)]) command.extend( ["-p", project_name, "-f", str(compose_file), "ps", "-a", "--format", "json"] ) completed = subprocess.run(command, check=True, capture_output=True, text=True) rendered = completed.stdout.strip() rows: list[dict[str, Any]] = [] if rendered: try: parsed = json.loads(rendered) rows = parsed if isinstance(parsed, list) else [parsed] except json.JSONDecodeError: rows = [json.loads(line) for line in rendered.splitlines() if line.strip()] return build_compose_readiness_report(rows, environment=environment) def _promotion_digest_payload(bundle: dict[str, Any]) -> dict[str, Any]: return { "schema_version": bundle["schema_version"], "source_environment": bundle["source_environment"], "target_environment": bundle["target_environment"], "approval_ref": bundle["approval_ref"], "objects": bundle["objects"], } def create_promotion_bundle(request: dict[str, Any]) -> dict[str, Any]: source = str(request.get("source_environment", "")) target = str(request.get("target_environment", "")) if source not in ENVIRONMENT_ORDER or target not in ENVIRONMENT_ORDER: raise ValueError("known source and target environments are required") if ENVIRONMENT_ORDER[target] - ENVIRONMENT_ORDER[source] != 10: raise ValueError("promotion must target the adjacent environment") approval_ref = str(request.get("approval_ref", "")).strip() if not approval_ref: raise ValueError("approval_ref is required") objects = request.get("objects") if not isinstance(objects, list) or not objects: raise ValueError("at least one promotion object is required") normalized: list[dict[str, Any]] = [] identities: set[tuple[str, str]] = set() for item in objects: if not isinstance(item, dict): raise ValueError("promotion object must be an object") kind = str(item.get("kind", "")) uid = str(item.get("uid", "")).strip() version = str(item.get("version", "")).strip() digest = str(item.get("payload_sha256", "")).lower() rollback_ref = str(item.get("rollback_ref", "")).strip() evidence_refs = item.get("evidence_refs") if kind not in PROMOTABLE_KINDS: raise ValueError(f"unsupported promotion kind: {kind}") if not uid or not version or not rollback_ref: raise ValueError("uid, version and rollback_ref are required") if not SHA256_PATTERN.fullmatch(digest): raise ValueError(f"invalid payload sha256 for {kind}:{uid}") if not isinstance(evidence_refs, list) or not evidence_refs: raise ValueError(f"evidence_refs are required for {kind}:{uid}") identity = (kind, uid) if identity in identities: raise ValueError(f"duplicate promotion object: {kind}:{uid}") identities.add(identity) normalized.append( { "kind": kind, "uid": uid, "version": version, "payload_sha256": digest, "rollback_ref": rollback_ref, "evidence_refs": sorted(str(value) for value in evidence_refs), } ) normalized.sort(key=lambda item: (item["kind"], item["uid"])) bundle: dict[str, Any] = { "schema_version": 1, "status": "approved", "source_environment": source, "target_environment": target, "approval_ref": approval_ref, "objects": normalized, } bundle["bundle_sha256"] = _sha256_bytes( _canonical_bytes(_promotion_digest_payload(bundle)) ) return bundle def verify_promotion_bundle(bundle: dict[str, Any]) -> dict[str, Any]: expected = _sha256_bytes(_canonical_bytes(_promotion_digest_payload(bundle))) actual = str(bundle.get("bundle_sha256", "")) if actual != expected: raise ValueError(f"promotion bundle digest mismatch: expected={expected}, actual={actual}") return { "schema_version": 1, "status": "verified", "bundle_sha256": actual, "object_count": len(bundle.get("objects", [])), } def evaluate_customer_migration(request: dict[str, Any]) -> dict[str, Any]: reasons: list[str] = [] mappings = request.get("mappings") if not isinstance(mappings, list): raise ValueError("mappings must be a list") incomplete = [ str(item.get("source", "unknown")) for item in mappings if isinstance(item, dict) and item.get("required", True) and (not str(item.get("source", "")).strip() or not str(item.get("target", "")).strip()) ] targets = [ str(item.get("target", "")) for item in mappings if isinstance(item, dict) and str(item.get("target", "")).strip() ] duplicates = sorted({item for item in targets if targets.count(item) > 1}) if incomplete or duplicates: reasons.append("mapping_incomplete") incremental = request.get("incremental") or {} latest = int(incremental.get("latest_sequence", 0)) caught_up = int(incremental.get("caught_up_sequence", -1)) lag = max(0, latest - caught_up) if lag: reasons.append("incremental_not_caught_up") reconciliation = request.get("reconciliation") or {} source_count = int(reconciliation.get("source_count", -1)) target_count = int(reconciliation.get("target_count", -2)) source_digest = str(reconciliation.get("source_digest", "")) target_digest = str(reconciliation.get("target_digest", "")) reconciled = ( source_count >= 0 and source_count == target_count and bool(source_digest) and source_digest == target_digest ) if not reconciled: reasons.append("reconciliation_mismatch") rollback = request.get("rollback") or {} rollback_ready = bool(rollback.get("backup_ref") and rollback.get("release_ref")) if not rollback_ready: reasons.append("rollback_not_prepared") snapshot = request.get("source_snapshot") or {} if not snapshot.get("snapshot_ref"): reasons.append("source_snapshot_missing") return { "schema_version": 1, "migration_id": str(request.get("migration_id", "unknown")), "mode": "dry_run", "source_mutated": False, "status": "blocked" if reasons else "ready", "blocking_reasons": sorted(set(reasons)), "source_snapshot": _redact(snapshot), "mapping": { "status": "blocked" if incomplete or duplicates else "ready", "mapping_count": len(mappings), "incomplete_sources": incomplete, "duplicate_targets": duplicates, }, "incremental_catchup": { "status": "caught_up" if lag == 0 else "lagging", "latest_sequence": latest, "caught_up_sequence": caught_up, "lag": lag, }, "reconciliation": { "status": "matched" if reconciled else "mismatch", "source_count": source_count, "target_count": target_count, "source_digest": source_digest, "target_digest": target_digest, }, "checkpoints": sorted(str(value) for value in request.get("checkpoints", [])), "rollback": { "status": "prepared" if rollback_ready else "blocked", "backup_ref": rollback.get("backup_ref"), "release_ref": rollback.get("release_ref"), "strategy": "restore_prior_release_and_pre_migration_backup", "database_downgrade": False, }, } def _git_value(repo_root: Path, *arguments: str, default: str = "unknown") -> str: try: completed = subprocess.run( ["git", "-C", str(repo_root), *arguments], check=True, capture_output=True, text=True, ) return completed.stdout.strip() or default except (OSError, subprocess.CalledProcessError): return default def _migration_head(repo_root: Path) -> str: revisions: dict[str, set[str]] = {} referenced: set[str] = set() for path in sorted((repo_root / "migrations/versions").glob("*.py")): tree = ast.parse(path.read_text(encoding="utf-8")) revision = "" parents: set[str] = set() for node in tree.body: if not isinstance(node, (ast.Assign, ast.AnnAssign)): continue target = node.targets[0] if isinstance(node, ast.Assign) else node.target if not isinstance(target, ast.Name): continue if target.id not in {"revision", "down_revision"}: continue value_node = node.value try: value = ast.literal_eval(value_node) except (ValueError, TypeError): continue if target.id == "revision": revision = str(value) elif value: if isinstance(value, (list, tuple)): parents.update(str(item) for item in value) else: parents.add(str(value)) if revision: revisions[revision] = parents referenced.update(parents) heads = sorted(set(revisions) - referenced) if len(heads) != 1: raise ValueError(f"exactly one migration head required; found={heads}") return heads[0] def build_release_manifest( repo_root: Path, *, version: str, source_date_epoch: int ) -> dict[str, Any]: repo_root = repo_root.resolve() openapi = repo_root / "docs/architecture/OPENAPI.yaml" route_match = re.search( r"^x-route-count:\s*(\d+)\s*$", openapi.read_text(encoding="utf-8"), re.MULTILINE, ) if not route_match: raise ValueError("OpenAPI x-route-count is missing") source_trees = { name: _tree_manifest(repo_root / name) for name in ("app", "database", "migrations") } release_copies: dict[str, dict[str, Any]] = {} for name, source in source_trees.items(): release = _tree_manifest(repo_root / "deployment" / name) release_copies[name] = { **release, "matches_source": release["sha256"] == source["sha256"], } manifest = { "schema_version": 1, "version": version, "source_date_epoch": int(source_date_epoch), "source": { "commit": _git_value(repo_root, "rev-parse", "HEAD"), "tree": _git_value(repo_root, "rev-parse", "HEAD^{tree}"), }, "source_trees": source_trees, "release_copies": release_copies, "migration": { "head": _migration_head(repo_root), "tree_sha256": source_trees["migrations"]["sha256"], }, "openapi": { "route_count": int(route_match.group(1)), "sha256": _sha256_file(openapi), }, "compose": { "sha256": _sha256_file(repo_root / "deploy/docker/docker-compose.yml") }, "environment_profiles": _tree_manifest(repo_root / "deployment/environments"), "dependency_inputs": { "backend_sha256": _sha256_file(repo_root / "requirements.txt"), "frontend_sha256": _sha256_file(repo_root / "frontend/package-lock.json"), }, } if not all(item["matches_source"] for item in release_copies.values()): raise ValueError("release copies do not match source trees") manifest["manifest_sha256"] = _sha256_bytes(_canonical_bytes(manifest)) return manifest def _component(name: str, version: str, ecosystem: str) -> dict[str, str]: return { "type": "library", "name": name, "version": version, "purl": f"pkg:{ecosystem}/{name}@{version}", } def _sbom(name: str, components: list[dict[str, str]], input_digest: str) -> dict[str, Any]: components.sort(key=lambda item: (item["name"].lower(), item["version"])) return { "bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1, "metadata": { "component": {"type": "application", "name": name}, "properties": [ {"name": "dataops:source-input-sha256", "value": input_digest} ], }, "components": components, } def generate_backend_sbom(requirements: Path) -> dict[str, Any]: components: list[dict[str, str]] = [] pattern = re.compile(r"^([A-Za-z0-9_.-]+)\s*(===|==|>=|~=|<=|>|<)\s*([^;\s]+)") for raw_line in requirements.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue match = pattern.match(line) if not match: raise ValueError(f"unsupported requirement line: {line}") name, operator, version = match.groups() rendered_version = version if operator in {"==", "==="} else operator + version components.append(_component(name, rendered_version, "pypi")) return _sbom("dataops-platform-backend", components, _sha256_file(requirements)) def generate_frontend_sbom(package_lock: Path) -> dict[str, Any]: lock = _read_json(package_lock) components: list[dict[str, str]] = [] packages = lock.get("packages") or {} if not isinstance(packages, dict): raise ValueError("package-lock packages object is required") for package_path, value in packages.items(): if not package_path or not isinstance(value, dict): continue name = str(value.get("name", "")) if not name and "node_modules/" in package_path: name = package_path.rsplit("node_modules/", 1)[-1] version = str(value.get("version", "")) if name and version: components.append(_component(name, version, "npm")) return _sbom("dataops-platform-frontend", components, _sha256_file(package_lock)) def generate_sboms(repo_root: Path) -> dict[str, Any]: return { "backend": generate_backend_sbom(repo_root / "requirements.txt"), "frontend": generate_frontend_sbom(repo_root / "frontend/package-lock.json"), } def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) validate = subparsers.add_parser("validate-environments") validate.add_argument("--profiles-dir", type=Path, required=True) validate.add_argument("--output") readiness = subparsers.add_parser("readiness") readiness.add_argument("--input", type=Path, required=True) readiness.add_argument("--output") compose_readiness = subparsers.add_parser("compose-readiness") compose_readiness.add_argument("--compose-file", type=Path, required=True) compose_readiness.add_argument("--project-name", required=True) compose_readiness.add_argument("--environment", required=True) compose_readiness.add_argument("--env-file", type=Path) compose_readiness.add_argument("--output") promotion = subparsers.add_parser("promotion") promotion.add_argument("--input", type=Path, required=True) promotion.add_argument("--output") promotion.add_argument("--verify", action="store_true") migration = subparsers.add_parser("migration-dry-run") migration.add_argument("--input", type=Path, required=True) migration.add_argument("--output") manifest = subparsers.add_parser("manifest") manifest.add_argument("--repo-root", type=Path, required=True) manifest.add_argument("--version", required=True) manifest.add_argument("--source-date-epoch", type=int, required=True) manifest.add_argument("--output") sbom = subparsers.add_parser("sbom") sbom.add_argument("--repo-root", type=Path, required=True) sbom.add_argument("--output-dir", required=True) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) if args.command == "validate-environments": _write_json(validate_environment_profiles(args.profiles_dir), args.output) elif args.command == "readiness": _write_json(build_readiness_report(_read_json(args.input)), args.output) elif args.command == "compose-readiness": _write_json( collect_compose_readiness( args.compose_file, project_name=args.project_name, environment=args.environment, env_file=args.env_file, ), args.output, ) elif args.command == "promotion": value = _read_json(args.input) report = verify_promotion_bundle(value) if args.verify else create_promotion_bundle(value) _write_json(report, args.output) elif args.command == "migration-dry-run": _write_json(evaluate_customer_migration(_read_json(args.input)), args.output) elif args.command == "manifest": _write_json( build_release_manifest( args.repo_root, version=args.version, source_date_epoch=args.source_date_epoch, ), args.output, ) elif args.command == "sbom": sboms = generate_sboms(args.repo_root) if args.output_dir == "-": _write_json(sboms) else: output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) _write_json(sboms["backend"], str(output_dir / "backend.cdx.json")) _write_json(sboms["frontend"], str(output_dir / "frontend.cdx.json")) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (OSError, ValueError, json.JSONDecodeError) as exc: print(f"ERROR: {exc}", file=sys.stderr) raise SystemExit(2)