#!/usr/bin/env python3 """Fail-closed, read-only placement gate for Darkoza's heterogeneous node fleet. Collection timestamps are assigned by the central collector; source node clocks are recorded separately and never trusted for freshness. This module evaluates proposals only. It never transfers data, starts jobs, or changes a node. """ from __future__ import annotations import argparse import json import math import sys from datetime import datetime, timezone from pathlib import Path from typing import Any ENCRYPTION_REQUIRED = {"internal", "confidential", "restricted"} VALID_CLASSES = {"public", "internal", "confidential", "restricted"} VALID_ENCRYPTION = {"enabled", "disabled", "unknown"} VALID_CLOCK = {"synchronized", "unsynchronized", "unknown"} def parse_time(value: str) -> datetime: if not isinstance(value, str) or not value: raise ValueError("timestamp must be a non-empty ISO-8601 string") parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: raise ValueError("timestamp must include a timezone") return parsed.astimezone(timezone.utc) def validate_snapshot(snapshot: dict[str, Any]) -> list[dict[str, Any]]: if not isinstance(snapshot, dict) or not isinstance(snapshot.get("nodes"), list): raise ValueError("snapshot.nodes must be a list") seen: set[str] = set() nodes = snapshot["nodes"] for node in nodes: if not isinstance(node, dict): raise ValueError("each node must be an object") name = node.get("name") if not isinstance(name, str) or not name.strip() or name in seen: raise ValueError("node names must be non-empty and unique") seen.add(name) if not isinstance(node.get("online"), bool): raise ValueError(f"{name}: online must be boolean") if not isinstance(node.get("roles"), list) or not node["roles"]: raise ValueError(f"{name}: roles must be a non-empty list") free = node.get("disk_free_gib") if isinstance(free, bool) or not isinstance(free, (int, float)) or not math.isfinite(free) or free < 0: raise ValueError(f"{name}: disk_free_gib must be a finite non-negative number") if node.get("encryption_at_rest", "unknown") not in VALID_ENCRYPTION: raise ValueError(f"{name}: invalid encryption_at_rest") if node.get("clock_sync_state", "unknown") not in VALID_CLOCK: raise ValueError(f"{name}: invalid clock_sync_state") parse_time(node.get("collected_at_utc")) if node.get("source_clock_utc") is not None: parse_time(node["source_clock_utc"]) return nodes def evaluate_workload(workload: dict[str, Any], nodes: list[dict[str, Any]], now: datetime, max_age_minutes: int = 15, max_clock_skew_minutes: int = 3) -> dict[str, Any]: name, roles, data_class, min_free = (workload.get(k) for k in ("name", "allowed_roles", "data_class", "min_free_gib")) if not isinstance(name, str) or not name.strip(): raise ValueError("workload.name must be non-empty") if not isinstance(roles, list) or not roles: raise ValueError(f"{name}: allowed_roles must be a non-empty list") if data_class not in VALID_CLASSES: raise ValueError(f"{name}: unsupported data_class") if isinstance(min_free, bool) or not isinstance(min_free, (int, float)) or not math.isfinite(min_free) or min_free < 0: raise ValueError(f"{name}: min_free_gib must be a finite non-negative number") requires_clock = workload.get("requires_synchronized_clock", True) if not isinstance(requires_clock, bool): raise ValueError(f"{name}: requires_synchronized_clock must be boolean") candidates, rejected = [], [] for node in nodes: if not set(node["roles"]).intersection(roles): continue reasons = [] if not node["online"]: reasons.append("node_offline") collected = parse_time(node["collected_at_utc"]) age = (now.astimezone(timezone.utc) - collected).total_seconds() / 60 if age < -max_clock_skew_minutes: reasons.append("collector_timestamp_from_future") elif age > max_age_minutes: reasons.append("collector_observation_stale") if node["disk_free_gib"] < min_free: reasons.append("insufficient_free_disk") encryption = node.get("encryption_at_rest", "unknown") if data_class in ENCRYPTION_REQUIRED and encryption != "enabled": reasons.append("encryption_at_rest_not_verified") clock = node.get("clock_sync_state", "unknown") if requires_clock and clock != "synchronized": reasons.append("node_clock_not_synchronized") if reasons: rejected.append({"node": node["name"], "reasons": reasons}) else: candidates.append({ "node": node["name"], "disk_free_gib": round(float(node["disk_free_gib"]), 2), "observation_age_minutes": round(max(0.0, age), 2), "clock_sync_state": clock, "encryption_at_rest": encryption, }) candidates.sort(key=lambda x: x["disk_free_gib"], reverse=True) return { "workload": name, "status": "eligible" if candidates else "blocked", "data_class": data_class, "selected_candidate": candidates[0] if candidates else None, "eligible_candidates": candidates, "rejected_allowed_role_nodes": rejected, "requires_synchronized_clock": requires_clock, "execution_performed": False, } def evaluate_snapshot(snapshot: dict[str, Any], now: datetime | None = None) -> dict[str, Any]: nodes = validate_snapshot(snapshot) captured = parse_time(snapshot.get("captured_at_utc")) evaluation_time = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) if (evaluation_time - captured).total_seconds() < -180: raise ValueError("snapshot capture time is unexpectedly in the future relative to collector clock") workloads = snapshot.get("workloads") if not isinstance(workloads, list): raise ValueError("snapshot.workloads must be a list") return { "schema": "darkoza.fleet-placement-decision.v1", "captured_at_utc": snapshot["captured_at_utc"], "evaluated_at_utc": evaluation_time.isoformat().replace("+00:00", "Z"), "policy": { "freshness_clock": "central_collector_only", "max_observation_age_minutes": 15, "max_clock_skew_minutes": 3, "encryption_required_for": sorted(ENCRYPTION_REQUIRED), "unknown_encryption_fails_closed": True, "time_sensitive_jobs_require_synchronized_node_clock": True, "performs_execution": False, }, "decisions": [evaluate_workload(w, nodes, evaluation_time) for w in workloads], } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("snapshot", type=Path, help="JSON snapshot with nodes and workload profiles") parser.add_argument("--output", type=Path, help="optional JSON report path") args = parser.parse_args() try: snapshot = json.loads(args.snapshot.read_text(encoding="utf-8")) rendered = json.dumps(evaluate_snapshot(snapshot), indent=2, ensure_ascii=False) + "\n" if args.output: args.output.write_text(rendered, encoding="utf-8") else: sys.stdout.write(rendered) except (OSError, json.JSONDecodeError, ValueError) as exc: print(f"fleet-placement-gate: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())