cloud-sync/cloud_sync/config.py

"""Per-instance sync.json — the opt-in marker + minimal config.

If ``<pack-folder>/.cloud-sync/sync.json`` doesn't exist, the pyz is a
no-op (PreLaunch hook returns 0, MC launches normally). Its presence
opts the instance into sync; its contents tell the script where to sync
to (`url`) and which restic sub-path to use (`instance_id`).

Intentionally minimal: NO label (UI display name comes from
``$INST_NAME`` / ``$INST_ID`` / ``--instance-label`` at runtime so the
Prism rename flow Just Works), NO token (mode-600 ``token`` lives next
to it), NO state (``state.json`` is separate, written by the sync flow).

instance_id is a 26-char ULID-style string (timestamp + 80 random bits,
base32, no padding) — sortable, URL-safe, no hyphens. Stable forever.
"""

from __future__ import annotations

import base64
import hashlib
import json
import os
import platform
import secrets
import socket
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path


SCHEMA_VERSION = 1


@dataclass(frozen=True)
class SyncConfig:
    url: str
    instance_id: str
    created_at: datetime
    host_fingerprint: str


def config_path(pack_folder: Path) -> Path:
    return pack_folder / ".cloud-sync" / "sync.json"


def exists(pack_folder: Path) -> bool:
    """Cheap opt-in check used by ``pull`` / ``push`` before doing anything else."""
    return config_path(pack_folder).exists()


def read(pack_folder: Path) -> SyncConfig | None:
    """Parse sync.json or return None if missing / corrupt / wrong schema."""
    p = config_path(pack_folder)
    if not p.exists():
        return None
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    if data.get("schema") != SCHEMA_VERSION:
        return None
    try:
        return SyncConfig(
            url=data["url"],
            instance_id=data["instance_id"],
            created_at=_parse_iso(data["created_at"]),
            host_fingerprint=data.get("host_fingerprint", ""),
        )
    except (KeyError, ValueError):
        return None


def write(pack_folder: Path, cfg: SyncConfig) -> None:
    """Persist sync.json (mode 644 — not a secret)."""
    p = config_path(pack_folder)
    p.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "schema": SCHEMA_VERSION,
        "url": cfg.url,
        "instance_id": cfg.instance_id,
        "created_at": cfg.created_at.astimezone(timezone.utc)
        .isoformat()
        .replace("+00:00", "Z"),
        "host_fingerprint": cfg.host_fingerprint,
    }
    p.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    p.chmod(0o644)


def delete(pack_folder: Path) -> bool:
    """Remove sync.json. Used by the ``disable`` subcommand. Returns True if a
    file was removed, False if nothing was there."""
    p = config_path(pack_folder)
    if not p.exists():
        return False
    p.unlink()
    return True


# ---------------------------------------------------------------------------
# minting
# ---------------------------------------------------------------------------


def mint(url: str) -> SyncConfig:
    """Build a fresh SyncConfig with a new ULID-style id and host fingerprint."""
    return SyncConfig(
        url=url,
        instance_id=new_instance_id(),
        created_at=datetime.now(timezone.utc),
        host_fingerprint=_machine_fingerprint(),
    )


def new_instance_id() -> str:
    """26-char base32 string: 6 bytes ms timestamp + 10 bytes random.

    Like a ULID, minus the strict Crockford alphabet. Sortable by mint
    time, URL-safe, no padding, no hyphens.
    """
    ms = int(time.time() * 1000).to_bytes(6, "big")
    rand = secrets.token_bytes(10)
    return base64.b32encode(ms + rand).decode("ascii").rstrip("=")


def _machine_fingerprint() -> str:
    """sha256 of (hostname || node || nanosecond timestamp), first 16 hex chars.

    Not a cryptographic identity — just a hint we can compare across pulls
    to detect "this instance was first set up on a different machine"
    cases later. Don't use it for authz.
    """
    src = f"{platform.node()}:{socket.gethostname()}:{time.time_ns()}"
    return hashlib.sha256(src.encode()).hexdigest()[:16]


def _parse_iso(s: str) -> datetime:
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    dt = datetime.fromisoformat(s)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


# ---------------------------------------------------------------------------
# display label resolution (NOT stored — derived per-run)
# ---------------------------------------------------------------------------


def resolve_label(flag_label: str | None, instance_id: str) -> str:
    """Pick the UI display name for an instance.

    Precedence:
      1. ``--instance-label`` CLI flag (explicit override)
      2. ``$INST_NAME`` (Prism's rename-tracking display name)
      3. ``$INST_ID`` (Prism's stable folder name)
      4. first 8 chars of the instance_id (last-resort)

    Never stored on disk — recomputed every run so Prism renames flow
    through naturally.
    """
    if flag_label:
        return flag_label
    env = os.environ.get("INST_NAME") or os.environ.get("INST_ID")
    if env:
        return env
    return instance_id[:8]