codec_update.py

"""
CODEC — Sparkle-compatible auto-update checker (paid Mac app)
=============================================================

The CODEC app is a Python app (no Cocoa runloop), so instead of embedding
Sparkle.framework we use a Python client that speaks the *same* Sparkle
protocol: it reads the Sparkle appcast feed and verifies each update's
**Ed25519 signature** against the app's `SUPublicEDKey` — the exact key pair
generated by Sparkle's `generate_keys` and used by `generate_appcast` to sign
releases. Same security model, same feed, no fragile framework embed.

Flow:
    info = check_for_update()          # → UpdateInfo | None
    if info:
        dmg = download_and_verify(info)   # Ed25519-verified .dmg on disk
        # surface "Update to {info.version}" in the dashboard; user opens dmg

Feed URL + public key come from ~/.codec/config.json (sparkle_feed_url /
sparkle_public_ed_key), falling back to the shipped defaults below.
"""

from __future__ import annotations

import base64
import json
import os
import re
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature

# Shipped defaults (overridable via ~/.codec/config.json). The public key is the
# operator's Sparkle EdDSA public key — safe to embed.
DEFAULT_FEED_URL   = "https://github.com/AVADSA25/codec-updates/releases/latest/download/appcast.xml"
DEFAULT_PUBLIC_KEY = "Az+iqcmusmWvyLEB7lW4j2J2kxGy6uLobNaH7ZIahys="

_SPARKLE_NS = "http://www.andymatuschak.org/xml-namespaces/sparkle"
_CONFIG_PATH = Path(os.path.expanduser("~/.codec/config.json"))


@dataclass
class UpdateInfo:
    version: str            # short version string (semver, e.g. "3.2.0")
    url: str                # enclosure download URL (.dmg)
    ed_signature: str       # base64 Ed25519 signature of the .dmg
    length: int             # expected byte length (0 if unknown)
    title: str = ""
    notes_url: str = ""

    def to_dict(self) -> dict:
        return {"version": self.version, "url": self.url, "title": self.title,
                "length": self.length, "notes_url": self.notes_url}


# ─── version helpers ─────────────────────────────────────────────────────────────

def _current_version() -> str:
    try:
        import codec_version
        return codec_version.__version__
    except Exception:
        try:
            return (Path(__file__).resolve().parent / "VERSION").read_text().strip()
        except Exception:
            return "0.0.0"


def _semver_tuple(v: str) -> tuple:
    """('3.10.2') → (3,10,2). Tolerant of a 'v' prefix; stops at the first
    non-numeric component so pre-release tags ('3.1.0-beta' → (3,1,0)) don't
    inflate the tuple. Pads to at least 3 components."""
    parts = re.split(r"[.\-+]", (v or "").lstrip("vV").strip())
    out = []
    for p in parts:
        if not re.fullmatch(r"\d+", p):
            break  # stop at first non-numeric (pre-release/build tag)
        out.append(int(p))
    while len(out) < 3:
        out.append(0)
    return tuple(out)


def is_newer(candidate: str, current: str) -> bool:
    return _semver_tuple(candidate) > _semver_tuple(current)


# ─── config ──────────────────────────────────────────────────────────────────────

def _config() -> dict:
    try:
        return json.loads(_CONFIG_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _feed_url(cfg: Optional[dict] = None) -> str:
    cfg = cfg if cfg is not None else _config()
    return cfg.get("sparkle_feed_url") or DEFAULT_FEED_URL


def _public_key(cfg: Optional[dict] = None) -> str:
    cfg = cfg if cfg is not None else _config()
    return cfg.get("sparkle_public_ed_key") or DEFAULT_PUBLIC_KEY


# ─── appcast parsing ─────────────────────────────────────────────────────────────

def parse_appcast(xml_text: str) -> list[UpdateInfo]:
    """Parse a Sparkle appcast into UpdateInfo items (newest first by version)."""
    root = ET.fromstring(xml_text)
    items: list[UpdateInfo] = []
    for item in root.iter("item"):
        enc = item.find("enclosure")
        if enc is None:
            continue
        short = (item.findtext(f"{{{_SPARKLE_NS}}}shortVersionString")
                 or enc.get(f"{{{_SPARKLE_NS}}}shortVersionString")
                 or item.findtext(f"{{{_SPARKLE_NS}}}version")
                 or enc.get(f"{{{_SPARKLE_NS}}}version") or "")
        url = enc.get("url", "")
        sig = enc.get(f"{{{_SPARKLE_NS}}}edSignature", "")
        length = int(enc.get("length") or 0)
        if not (short and url):
            continue
        items.append(UpdateInfo(
            version=short, url=url, ed_signature=sig, length=length,
            title=item.findtext("title") or "",
            notes_url=item.findtext(f"{{{_SPARKLE_NS}}}releaseNotesLink") or "",
        ))
    items.sort(key=lambda i: _semver_tuple(i.version), reverse=True)
    return items


# ─── public API ──────────────────────────────────────────────────────────────────

def check_for_update(feed_url: Optional[str] = None, *, timeout: float = 8.0,
                     cfg: Optional[dict] = None) -> Optional[UpdateInfo]:
    """
    Return the latest UpdateInfo if it's newer than the running version, else None.
    Network/parse errors return None (never raises) — update checks are best-effort.
    """
    url = feed_url or _feed_url(cfg)
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r:
            xml_text = r.read().decode("utf-8", "replace")
        items = parse_appcast(xml_text)
    except Exception:
        return None
    if not items:
        return None
    latest = items[0]
    return latest if is_newer(latest.version, _current_version()) else None


def verify_ed25519(data: bytes, ed_signature_b64: str, public_key_b64: str) -> bool:
    """Verify a Sparkle Ed25519 signature (base64) over `data`. False on any failure."""
    try:
        pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(public_key_b64))
        pub.verify(base64.b64decode(ed_signature_b64), data)
        return True
    except (InvalidSignature, Exception):
        return False


def download_and_verify(info: UpdateInfo, *, dest_dir: Optional[Path] = None,
                        timeout: float = 120.0, cfg: Optional[dict] = None) -> Path:
    """
    Download the update, verify its Ed25519 signature against SUPublicEDKey, and
    return the on-disk path. Raises ValueError if the signature doesn't verify —
    a failed signature means the download is untrusted and MUST NOT be installed.
    """
    dest_dir = dest_dir or Path(os.path.expanduser("~/Downloads"))
    dest_dir.mkdir(parents=True, exist_ok=True)
    fname = info.url.rstrip("/").split("/")[-1] or "CODEC-update.dmg"
    out = dest_dir / fname

    with urllib.request.urlopen(info.url, timeout=timeout) as r:
        data = r.read()

    if info.length and len(data) != info.length:
        raise ValueError(f"length mismatch: got {len(data)}, expected {info.length}")
    if not info.ed_signature:
        raise ValueError("update has no Ed25519 signature — refusing (untrusted)")
    if not verify_ed25519(data, info.ed_signature, _public_key(cfg)):
        raise ValueError("Ed25519 signature verification FAILED — refusing (untrusted)")

    out.write_bytes(data)
    return out