#!/usr/bin/env python3 """Common utilities for OTA update modules""" import re import zipfile import xml.etree.ElementTree as ET from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional from dataclasses import dataclass from ...common.utils import log_error, log_info, log_success # Re-exported so callers (and ota/__init__.py) can get sparkle_sign_file # from ota.common alongside the other OTA helpers. from ...common.sparkle import sparkle_sign_file as sparkle_sign_file # Sparkle XML namespace SPARKLE_NS = "http://www.andymatuschak.org/xml-namespaces/sparkle" ET.register_namespace("sparkle", SPARKLE_NS) SERVER_PLATFORMS = [ {"name": "darwin_arm64", "binary": "browseros-server-darwin-arm64", "target": "darwin-arm64", "os": "macos", "arch": "arm64"}, {"name": "darwin_x64", "binary": "browseros-server-darwin-x64", "target": "darwin-x64", "os": "macos", "arch": "x86_64"}, {"name": "linux_arm64", "binary": "browseros-server-linux-arm64", "target": "linux-arm64", "os": "linux", "arch": "arm64"}, {"name": "linux_x64", "binary": "browseros-server-linux-x64", "target": "linux-x64", "os": "linux", "arch": "x86_64"}, {"name": "windows_x64", "binary": "browseros-server-windows-x64.exe", "target": "windows-x64", "os": "windows", "arch": "x86_64"}, ] APPCAST_TEMPLATE = """ {title} {appcast_url} BrowserOS Server binary updates en {version} {pub_date} {enclosures} """ ENCLOSURE_TEMPLATE = """ """ @dataclass class SignedArtifact: """Represents a signed artifact with Sparkle signature""" platform: str zip_path: Path signature: str length: int os: str arch: str @dataclass class ExistingAppcast: """Parsed data from an existing appcast file""" version: str pub_date: str artifacts: Dict[str, SignedArtifact] def find_server_resources_dir(binaries_dir: Path, platform: dict) -> Optional[Path]: """Return the extracted ``resources/`` dir for a platform, or ``None``. ``binaries_dir`` is the temp root created by ``_download_artifacts``; each platform lives at ``//resources/``. """ target = platform.get("target", platform["name"].replace("_", "-")) resources = binaries_dir / target / "resources" return resources if resources.is_dir() else None def parse_existing_appcast(appcast_path: Path) -> Optional[ExistingAppcast]: """Parse existing appcast XML file. Args: appcast_path: Path to existing appcast XML file Returns: ExistingAppcast with version, pubDate, and artifacts, or None if parsing fails """ if not appcast_path.exists(): return None try: tree = ET.parse(appcast_path) root = tree.getroot() # Find the item element (we only support single-item appcasts) channel = root.find("channel") if channel is None: return None item = channel.find("item") if item is None: return None # Extract version version_elem = item.find(f"{{{SPARKLE_NS}}}version") if version_elem is None or version_elem.text is None: return None version = version_elem.text # Extract pubDate pub_date_elem = item.find("pubDate") pub_date = pub_date_elem.text if pub_date_elem is not None and pub_date_elem.text else "" # Extract enclosures artifacts: Dict[str, SignedArtifact] = {} for enclosure in item.findall("enclosure"): url = enclosure.get("url", "") os_type = enclosure.get(f"{{{SPARKLE_NS}}}os", "") arch = enclosure.get(f"{{{SPARKLE_NS}}}arch", "") signature = enclosure.get(f"{{{SPARKLE_NS}}}edSignature", "") length_str = enclosure.get("length", "0") if not all([url, os_type, arch, signature]): continue # Extract platform from URL (e.g., browseros_server_0.0.37_darwin_arm64.zip) filename = url.split("/")[-1] # Match pattern like _darwin_arm64.zip or _windows_x64.zip platform_match = re.search(r"_([a-z]+_[a-z0-9]+)\.zip$", filename) if not platform_match: continue platform = platform_match.group(1) artifacts[platform] = SignedArtifact( platform=platform, zip_path=Path(filename), signature=signature, length=int(length_str), os=os_type, arch=arch, ) return ExistingAppcast(version=version, pub_date=pub_date, artifacts=artifacts) except ET.ParseError as e: log_error(f"Malformed appcast XML: {e}") return None except Exception as e: log_error(f"Failed to parse existing appcast: {e}") return None def generate_server_appcast( version: str, artifacts: List[SignedArtifact], channel: str = "alpha", existing: Optional[ExistingAppcast] = None, ) -> str: """Generate appcast XML for server OTA, merging with existing if same version. Args: version: Version string (e.g., "0.0.36") artifacts: List of new SignedArtifact with signature info channel: "alpha" or "prod" existing: Previously parsed appcast to merge with (if same version) Returns: Complete appcast XML string Merge behavior: - If existing has same version: merge platforms, keep original pubDate - If existing has different version or is None: use only new artifacts """ if channel == "alpha": title = "BrowserOS Server (Alpha)" appcast_url = "https://cdn.browseros.com/appcast-server.alpha.xml" else: title = "BrowserOS Server" appcast_url = "https://cdn.browseros.com/appcast-server.xml" # Determine pubDate and merged artifacts if existing is not None and existing.version == version: # Same version: merge artifacts, keep original pubDate pub_date = existing.pub_date merged_artifacts = dict(existing.artifacts) # Copy existing for artifact in artifacts: merged_artifacts[artifact.platform] = artifact # New overrides existing final_artifacts = list(merged_artifacts.values()) log_info(f"Merging with existing appcast (kept {len(existing.artifacts)} existing, added/updated {len(artifacts)} platforms)") else: # Different version or no existing: start fresh pub_date = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S +0000") final_artifacts = artifacts if existing is not None: log_info(f"Version changed ({existing.version} -> {version}), replacing appcast") # Sort artifacts by platform name for consistent output final_artifacts = sorted(final_artifacts, key=lambda a: a.platform) enclosures = [] for artifact in final_artifacts: comment = f"{artifact.os.capitalize()} {artifact.arch}" if artifact.os == "macos": comment = f"macOS {artifact.arch}" zip_filename = f"browseros_server_{version}_{artifact.platform}.zip" url = f"https://cdn.browseros.com/server/{zip_filename}" enclosure = ENCLOSURE_TEMPLATE.format( comment=comment, url=url, os=artifact.os, arch=artifact.arch, signature=artifact.signature, length=artifact.length, ) enclosures.append(enclosure) return APPCAST_TEMPLATE.format( title=title, appcast_url=appcast_url, version=version, pub_date=pub_date, enclosures="\n\n".join(enclosures), ) def create_server_bundle_zip(resources_dir: Path, output_zip: Path) -> bool: """Zip an extracted ``resources/`` tree into a Sparkle payload. Produces entries like ``resources/bin/browseros_server``, ``resources/bin/third_party/lima/limactl`` — mirroring what the agent build staged and what the Chromium build bakes into the installed app. File modes are preserved by ``ZipFile.write`` so executable bits survive. """ if not resources_dir.is_dir(): log_error(f"Resources dir not found: {resources_dir}") return False bundle_root = resources_dir.parent try: with zipfile.ZipFile(output_zip, "w", zipfile.ZIP_DEFLATED) as zf: for path in sorted(resources_dir.rglob("*")): if not path.is_file(): continue arcname = path.relative_to(bundle_root).as_posix() zf.write(path, arcname) log_success(f"Created {output_zip.name}") return True except Exception as e: log_error(f"Failed to create bundle zip: {e}") return False def get_appcast_path(channel: str = "alpha") -> Path: """Get path to appcast file in config/appcast directory""" appcast_dir = Path(__file__).parent.parent.parent / "config" / "appcast" if channel == "alpha": return appcast_dir / "appcast-server.alpha.xml" return appcast_dir / "appcast-server.xml"