feat(sync): add Katello content-view client and ContentViewSyncer

src/sourceos_syncd/cli.py

@@ -17,6 +17,8 @@
 from .reports import load_report, pretty_json, repair_plan, snapshot, validate_report, verify, with_fresh_diagnosis
 from .scorecard import evaluate_scorecard, validate_scorecard
 from .store_reports import append_store_event, init_store, snapshot_from_store
+from .content_sync import ContentViewSyncer
+from .katello_client import KatelloContentClient
 from .trust import TrustRequest, evaluate_trust, validate_trust_decision
 
 
@@ -146,6 +148,30 @@ def build_parser() -> argparse.ArgumentParser:
     score_validate.add_argument("--file", "-f", required=True, help="scorecard JSON file")
     add_compact(score_validate)
 
+    sync = subcommands.add_parser("sync", help="Katello content view sync planning and apply")
+    sync_sub = sync.add_subparsers(dest="command", required=True)
+
+    def add_katello_args(p: argparse.ArgumentParser) -> None:
+        p.add_argument("--katello-url", default="https://127.0.0.1:8443", help="Foreman+Katello base URL")
+        p.add_argument("--katello-user", default="admin", help="Katello admin username")
+        p.add_argument("--katello-password", default=None, help="Katello admin password (or set KATELLO_PASSWORD env)")
+        p.add_argument("--org", default="SocioProphet", help="Katello organization name")
+        p.add_argument("--content-view", default="sourceos-builder-aarch64", help="content view name")
+        p.add_argument("--lifecycle-env", default="dev", help="lifecycle environment (dev/candidate/stable)")
+        p.add_argument("--locus", default="local", help="execution locus (local/trusted_private)")
+        p.add_argument("--flake-ref", default="github:SociOS-Linux/source-os#builder-aarch64", help="NixOS flake ref")
+        p.add_argument("--current-version", default=None, help="current content view version (skip if up to date)")
+        p.add_argument("--no-verify-ssl", action="store_true", help="skip TLS verification (local dev only)")
+
+    sync_plan = sync_sub.add_parser("plan", help="query Katello and emit a ContentSyncPlan (no changes)")
+    add_katello_args(sync_plan)
+    add_compact(sync_plan)
+
+    sync_apply = sync_sub.add_parser("apply", help="apply a ContentSyncPlan (dry-run unless --execute)")
+    add_katello_args(sync_apply)
+    sync_apply.add_argument("--execute", action="store_true", help="actually run nix copy + nixos-rebuild (default: dry-run)")
+    add_compact(sync_apply)
+
     return parser
 
 
@@ -262,6 +288,33 @@ def main(argv: list[str] | None = None) -> int:
             sys.stdout.write(pretty_json({"valid": not errors, "errors": errors}, pretty=pretty))
             return 0 if not errors else 2
 
+        if args.area == "sync" and args.command in ("plan", "apply"):
+            import os
+            password = args.katello_password or os.environ.get("KATELLO_PASSWORD", "")
+            if not password:
+                sys.stderr.write(pretty_json({"error": "missing password", "message": "pass --katello-password or set KATELLO_PASSWORD"}, pretty=pretty))
+                return 1
+            client = KatelloContentClient(
+                base_url=args.katello_url,
+                username=args.katello_user,
+                password=password,
+                org=args.org,
+                verify_ssl=not args.no_verify_ssl,
+            )
+            manifest = client.get_latest_version(args.content_view, args.lifecycle_env)
+            syncer = ContentViewSyncer(
+                flake_ref=args.flake_ref,
+                locus=args.locus,
+                current_version=args.current_version,
+            )
+            plan = syncer.plan(manifest)
+            if args.command == "plan":
+                sys.stdout.write(pretty_json(plan.to_dict(), pretty=pretty))
+                return 0 if plan.policy_gate in ("allowed", "no-op") else 2
+            result = syncer.execute(plan, dry_run=not args.execute)
+            sys.stdout.write(pretty_json(result, pretty=pretty))
+            return 0 if result["status"] in ("dry_run", "executed") else 2
+
     except Exception as exc:  # noqa: BLE001 - CLI boundary should present clean error JSON.
         sys.stderr.write(pretty_json({"error": type(exc).__name__, "message": str(exc)}, pretty=pretty))
         return 1

src/sourceos_syncd/content_sync.py

@@ -0,0 +1,174 @@
+"""Content view sync planner for sourceos-syncd.
+
+Consumes a ContentViewManifest from KatelloContentClient and produces a
+ContentSyncPlan describing the nix copy and nixos-rebuild steps required to
+apply the new content view version.
+
+Boundary invariant: plan() is pure and side-effect-free. execute() is the
+only method that shells out — it requires an explicit caller opt-in and will
+refuse to run if the plan's policy_gate is not 'allowed'.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import shutil
+import subprocess
+from dataclasses import dataclass, field
+from typing import Any
+
+from .katello_client import ContentViewManifest
+
+SYNC_SCHEMA = "sourceos.content-sync-plan/v0.1"
+
+
+@dataclass(frozen=True)
+class ContentSyncPlan:
+    """Non-mutating description of a pending content sync."""
+
+    schema: str
+    org: str
+    content_view: str
+    from_version: str | None
+    to_version: str
+    lifecycle_env: str
+    nix_cache_url: str
+    flake_ref: str
+    policy_gate: str
+    policy_reason: str
+    steps: list[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "schema": self.schema,
+            "org": self.org,
+            "content_view": self.content_view,
+            "from_version": self.from_version,
+            "to_version": self.to_version,
+            "lifecycle_env": self.lifecycle_env,
+            "nix_cache_url": self.nix_cache_url,
+            "flake_ref": self.flake_ref,
+            "policy_gate": self.policy_gate,
+            "policy_reason": self.policy_reason,
+            "steps": self.steps,
+        }
+
+    @property
+    def allowed(self) -> bool:
+        return self.policy_gate == "allowed"
+
+
+class ContentViewSyncer:
+    """Plans and optionally executes a Katello content view sync.
+
+    The syncer enforces the locus gate: only local locus is permitted for
+    Phase 0. burst_cloud and attested_fog require explicit policy elevation
+    (not yet implemented).
+    """
+
+    ALLOWED_LOCI = {"local", "trusted_private"}
+
+    def __init__(
+        self,
+        flake_ref: str = "github:SociOS-Linux/source-os#builder-aarch64",
+        locus: str = "local",
+        current_version: str | None = None,
+    ) -> None:
+        self._flake_ref = flake_ref
+        self._locus = locus
+        self._current_version = current_version
+
+    def plan(self, manifest: ContentViewManifest) -> ContentSyncPlan:
+        """Return a non-mutating ContentSyncPlan. No I/O performed."""
+
+        if self._locus not in self.ALLOWED_LOCI:
+            return ContentSyncPlan(
+                schema=SYNC_SCHEMA,
+                org=manifest.org,
+                content_view=manifest.content_view,
+                from_version=self._current_version,
+                to_version=manifest.version,
+                lifecycle_env=manifest.lifecycle_env,
+                nix_cache_url=manifest.nix_cache_url,
+                flake_ref=self._flake_ref,
+                policy_gate="denied",
+                policy_reason=f"locus '{self._locus}' not in allowed loci {sorted(self.ALLOWED_LOCI)}",
+                steps=[],
+            )
+
+        if self._current_version and self._current_version == manifest.version:
+            return ContentSyncPlan(
+                schema=SYNC_SCHEMA,
+                org=manifest.org,
+                content_view=manifest.content_view,
+                from_version=self._current_version,
+                to_version=manifest.version,
+                lifecycle_env=manifest.lifecycle_env,
+                nix_cache_url=manifest.nix_cache_url,
+                flake_ref=self._flake_ref,
+                policy_gate="no-op",
+                policy_reason="already at latest version",
+                steps=[],
+            )
+
+        steps = [
+            f"nix copy --from '{manifest.nix_cache_url}' --no-check-sigs '{self._flake_ref}'",
+            f"nixos-rebuild switch --flake '{self._flake_ref}'",
+        ]
+
+        return ContentSyncPlan(
+            schema=SYNC_SCHEMA,
+            org=manifest.org,
+            content_view=manifest.content_view,
+            from_version=self._current_version,
+            to_version=manifest.version,
+            lifecycle_env=manifest.lifecycle_env,
+            nix_cache_url=manifest.nix_cache_url,
+            flake_ref=self._flake_ref,
+            policy_gate="allowed",
+            policy_reason=f"locus '{self._locus}' permitted; new version available",
+            steps=steps,
+        )
+
+    def execute(self, plan: ContentSyncPlan, dry_run: bool = True) -> dict[str, Any]:
+        """Execute the sync plan. dry_run=True (default) only prints steps."""
+
+        if not plan.allowed:
+            return {
+                "status": "skipped",
+                "reason": plan.policy_reason,
+                "policy_gate": plan.policy_gate,
+            }
+
+        results = []
+        for step in plan.steps:
+            if dry_run:
+                results.append({"step": step, "status": "dry_run"})
+                continue
+
+            if not shutil.which("nix") and step.startswith("nix "):
+                results.append({"step": step, "status": "skipped", "reason": "nix not found in PATH"})
+                continue
+            if not shutil.which("nixos-rebuild") and step.startswith("nixos-rebuild "):
+                results.append({"step": step, "status": "skipped", "reason": "nixos-rebuild not found in PATH"})
+                continue
+
+            try:
+                proc = subprocess.run(
+                    step, shell=True, capture_output=True, text=True, timeout=600
+                )
+                results.append({
+                    "step": step,
+                    "status": "ok" if proc.returncode == 0 else "failed",
+                    "returncode": proc.returncode,
+                    "stdout": proc.stdout.strip()[:500],
+                    "stderr": proc.stderr.strip()[:500],
+                })
+            except subprocess.TimeoutExpired:
+                results.append({"step": step, "status": "timeout"})
+
+        return {
+            "status": "dry_run" if dry_run else "executed",
+            "plan": plan.to_dict(),
+            "results": results,
+        }