#!/usr/bin/env python3 """Dependency-free, fail-closed validation for PDF2HTML deployment manifests.""" from __future__ import annotations import json import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent REPO_ROOT = ROOT.parent.parent MANIFEST = ROOT / "production.json" SCHEMA = ROOT / "deployment.schema.json" EXACT_KEYS = { "": {"$schema", "schemaVersion", "environment", "deploymentStatus", "domains", "components", "secretNames", "rollout", "rollback", "unresolvedAuthenticatedValues"}, "/domains": {"frontend", "api"}, "/components": {"cloudflareFrontend", "cloudpanelApi", "cloudpanelWorker", "postgresMigrations", "privateObjectStorage"}, "/components/cloudflareFrontend": {"kind", "artifact", "accountId", "projectName", "deploymentTarget", "observedStatus", "healthCheck"}, "/components/cloudflareFrontend/healthCheck": {"url", "method", "expectedStatuses", "requiresAuthentication", "contentAssertion", "verified"}, "/components/cloudflareFrontend/healthCheck/contentAssertion": {"kind", "value", "verified"}, "/components/cloudpanelApi": {"kind", "artifact", "host", "runtime", "serviceName", "originMapping", "observedStatus", "healthCheck", "readinessCheck"}, "/components/cloudpanelApi/healthCheck": {"url", "method", "expectedStatuses", "requiresAuthentication", "contentAssertion", "verified"}, "/components/cloudpanelApi/healthCheck/contentAssertion": {"kind", "value", "verified"}, "/components/cloudpanelApi/readinessCheck": {"url", "method", "expectedStatuses", "requiresAuthentication", "contentAssertion", "verified"}, "/components/cloudpanelApi/readinessCheck/contentAssertion": {"kind", "value", "verified"}, "/components/cloudpanelWorker": {"kind", "artifact", "host", "serviceName", "imageDigest", "runtime", "separateFromApi", "network", "runAsNonRoot", "readOnlyRootFilesystem", "dropAllCapabilities", "noNewPrivileges", "resourceLimits", "healthCheck"}, "/components/cloudpanelWorker/resourceLimits": {"cpu", "memoryBytes", "pids", "temporaryStorageBytes", "outputBytes", "fileCount", "wallTimeSeconds"}, "/components/cloudpanelWorker/healthCheck": {"mechanism", "expected", "verified"}, "/components/postgresMigrations": {"kind", "artifact", "runner", "databaseTarget", "migrationCommand", "backupVerified", "restoreVerified", "transactional"}, "/components/privateObjectStorage": {"kind", "adapter", "endpoint", "region", "bucket", "publicAccess", "encryption", "signedDownloadTtlSeconds", "lifecycle"}, "/components/privateObjectStorage/lifecycle": {"sourcePrefix", "zipPrefix", "sourceDeadlineHours", "zipDeadlineHours", "providerRuleIds", "deadlineSemanticsVerified", "applicationReconciliation"}, "/secretNames": {"api", "worker", "frontend"}, "/rollout": {"strategy", "automaticPromotion", "steps", "requiredGates"}, "/rollback": {"automatic", "triggers", "steps", "previousFrontendDeployment", "previousApiArtifact", "previousWorkerImageDigest", "databaseRollbackProcedure"}, } def fail(errors: list[str], message: str) -> None: errors.append(message) def pointer_get(doc: Any, pointer: str) -> Any: node = doc for raw in pointer.lstrip("/").split("/") if pointer else []: key = raw.replace("~1", "/").replace("~0", "~") if not isinstance(node, dict) or key not in node: raise KeyError(pointer) node = node[key] return node def walk_exact(node: Any, pointer: str, errors: list[str]) -> None: if pointer in EXACT_KEYS: if not isinstance(node, dict): fail(errors, f"{pointer or '/'} must be an object") return actual = set(node) expected = EXACT_KEYS[pointer] if actual != expected: fail(errors, f"{pointer or '/'} keys differ: missing={sorted(expected-actual)} extra={sorted(actual-expected)}") if isinstance(node, dict): for key, value in node.items(): walk_exact(value, f"{pointer}/{key}", errors) def null_paths(node: Any, pointer: str = "") -> list[str]: if node is None: return [pointer] found: list[str] = [] if isinstance(node, dict): for key, value in node.items(): if pointer == "" and key == "unresolvedAuthenticatedValues": continue found.extend(null_paths(value, f"{pointer}/{key}")) elif isinstance(node, list): for index, value in enumerate(node): found.extend(null_paths(value, f"{pointer}/{index}")) return found def workspace_package_names(errors: list[str]) -> set[str]: names: set[str] = set() for package_file in REPO_ROOT.glob("**/package.json"): if "node_modules" in package_file.parts: continue try: package = json.loads(package_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: fail(errors, f"cannot inspect workspace package {package_file.relative_to(REPO_ROOT)}: {exc}") continue name = package.get("name") if isinstance(name, str): names.add(name) return names def verify_repo_reference(errors: list[str], value: Any, pointer: str, package_names: set[str]) -> None: if value is None: return if not isinstance(value, str) or not value: fail(errors, f"{pointer} must be null or a non-empty repository path/workspace package name") elif value.startswith("@"): if value not in package_names: fail(errors, f"{pointer} references missing workspace package: {value}") else: target = (REPO_ROOT / value).resolve() try: target.relative_to(REPO_ROOT.resolve()) except ValueError: fail(errors, f"{pointer} escapes the repository: {value}") else: if not target.exists(): fail(errors, f"{pointer} references missing repository path: {value}") def main() -> int: errors: list[str] = [] try: manifest = json.loads(MANIFEST.read_text(encoding="utf-8")) schema = json.loads(SCHEMA.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: print(f"FAIL: cannot load manifest inputs: {exc}") return 1 if schema.get("additionalProperties") is not False: fail(errors, "schema root must reject additional properties") walk_exact(manifest, "", errors) expected = { "$schema": "./deployment.schema.json", "schemaVersion": 1, "environment": "production", "deploymentStatus": "blocked", } for key, value in expected.items(): if manifest.get(key) != value: fail(errors, f"/{key} must equal {value!r}") domains = manifest.get("domains", {}) if domains.get("frontend") != "https://tools.press.zone" or domains.get("api") != "https://api.press.zone": fail(errors, "production domains do not match the committed product contract") components = manifest.get("components", {}) worker = components.get("cloudpanelWorker", {}) for key in ("separateFromApi", "runAsNonRoot", "readOnlyRootFilesystem", "dropAllCapabilities", "noNewPrivileges"): if worker.get(key) is not True: fail(errors, f"worker containment /components/cloudpanelWorker/{key} must be true") if worker.get("network") != "disabled-for-conversion-job": fail(errors, "conversion-job network must be disabled") storage = components.get("privateObjectStorage", {}) if storage.get("publicAccess") is not False: fail(errors, "object storage must be private") lifecycle = storage.get("lifecycle", {}) if lifecycle.get("sourceDeadlineHours") != 48 or lifecycle.get("zipDeadlineHours") != 48: fail(errors, "source and ZIP lifecycle deadlines must both be 48 hours") if lifecycle.get("applicationReconciliation") is not True: fail(errors, "application lifecycle reconciliation must be enabled") package_names = workspace_package_names(errors) migrations = components.get("postgresMigrations", {}) verify_repo_reference(errors, migrations.get("artifact"), "/components/postgresMigrations/artifact", package_names) verify_repo_reference(errors, migrations.get("runner"), "/components/postgresMigrations/runner", package_names) verify_repo_reference(errors, storage.get("adapter"), "/components/privateObjectStorage/adapter", package_names) unresolved = manifest.get("unresolvedAuthenticatedValues") if not isinstance(unresolved, list) or not unresolved: fail(errors, "unresolvedAuthenticatedValues must be a non-empty list until authenticated preflight completes") unresolved = [] unresolved_paths: set[str] = set() for index, item in enumerate(unresolved): if not isinstance(item, dict) or set(item) != {"path", "owner", "reason"}: fail(errors, f"unresolvedAuthenticatedValues[{index}] has invalid shape") continue path = item.get("path") if not all(isinstance(item.get(key), str) and item[key].strip() for key in ("path", "owner", "reason")): fail(errors, f"unresolvedAuthenticatedValues[{index}] values must be non-empty strings") continue if path in unresolved_paths: fail(errors, f"duplicate unresolved path: {path}") unresolved_paths.add(path) try: pointer_get(manifest, path) except KeyError: fail(errors, f"unresolved path does not exist: {path}") for path in null_paths(manifest): covered = any(path == unresolved_path or path.startswith(unresolved_path + "/") for unresolved_path in unresolved_paths) if not covered: fail(errors, f"null value is not explicitly unresolved: {path}") for process in ("api", "worker", "frontend"): names = manifest.get("secretNames", {}).get(process) if not isinstance(names, list) or len(names) != len(set(names)): fail(errors, f"secretNames.{process} must be a unique list") elif any(not isinstance(name, str) or not name or name.upper() != name for name in names): fail(errors, f"secretNames.{process} may contain names only") if manifest.get("secretNames", {}).get("frontend") != []: fail(errors, "frontend must not declare server secrets") rollout = manifest.get("rollout", {}) if rollout.get("automaticPromotion") is not False or rollout.get("strategy") != "gated-sequential": fail(errors, "rollout must be manual and gated-sequential") rollback = manifest.get("rollback", {}) if rollback.get("automatic") is not False: fail(errors, "rollback must not claim unverified automatic recovery") verified_flags = [] def collect_verified(node: Any) -> None: if isinstance(node, dict): for key, value in node.items(): if key.endswith("Verified") or key == "verified": verified_flags.append((key, value)) collect_verified(value) elif isinstance(node, list): for value in node: collect_verified(value) collect_verified(manifest.get("components", {})) if any(value is True for _, value in verified_flags): fail(errors, "skeleton must not claim authenticated infrastructure verification") if errors: print(f"FAIL: {len(errors)} manifest validation error(s)") for error in errors: print(f"- {error}") return 1 print(f"PASS: production deployment manifest is structurally valid and remains fail-closed with {len(unresolved_paths)} unresolved authenticated values") return 0 if __name__ == "__main__": sys.exit(main())