yesod-mirror/tools/helm_sync.py

import yaml
import json
import requests
import sys
import os
from urllib.parse import urljoin
from tools import helm_pull


def load_yaml(path):
    with open(path, "r") as f:
        return yaml.safe_load(f)


def fetch_index(repo_url):
    index_url = urljoin(repo_url + "/", "index.yaml")
    print(f"Fetching index from {index_url}...")
    try:
        r = requests.get(index_url)
        r.raise_for_status()
        return yaml.safe_load(r.text)
    except Exception as e:
        print(f"Error fetching {index_url}: {e}")
        return None


def resolve_oci(repo_url, chart_name, version):
    # Construct OCI URL
    # repo_url: oci://ghcr.io/stefanprodan/charts
    # chart_name: podinfo
    # result: oci://ghcr.io/stefanprodan/charts/podinfo

    base_url = repo_url
    if base_url.endswith("/"):
        base_url = base_url[:-1]

    full_url = f"{base_url}/{chart_name}"

    print(f"Resolving OCI chart {full_url}:{version}...")

    # Use helm_pull logic to get manifest and digest
    # Parse URL
    path = full_url[6:]  # strip oci://
    registry, repository = path.split("/", 1)

    token = helm_pull.get_token(registry, repository)
    manifest = helm_pull.get_manifest(registry, repository, version, token)

    valid_media_types = [
        "application/vnd.cncf.helm.chart.content.v1.tar+gzip",
        "application/x-tar",
    ]

    chart_layer = None
    for layer in manifest.get("layers", []):
        if layer.get("mediaType") in valid_media_types:
            chart_layer = layer
            break

    if not chart_layer:
        raise Exception(
            f"No Helm chart layer found in manifest for {full_url}:{version}"
        )

    digest = chart_layer["digest"]

    return {"version": version, "url": full_url, "digest": digest}


def main():
    if len(sys.argv) < 2:
        print("Usage: python helm_sync.py <path_to_chartfile.yaml> [output_lock_file]")
        sys.exit(1)

    chartfile_path = sys.argv[1]
    lockfile_path = (
        sys.argv[2]
        if len(sys.argv) > 2
        else chartfile_path.replace(".yaml", ".lock.json")
    )

    print(f"Reading {chartfile_path}...")
    chartfile = load_yaml(chartfile_path)

    repos = {r["name"]: r["url"] for r in chartfile.get("repositories", [])}
    indices = {}

    lock_data = {"charts": {}}

    for req in chartfile.get("requires", []):
        chart_ref = req["chart"]
        version = req["version"]

        if "/" not in chart_ref:
            print(f"Invalid chart reference: {chart_ref}. Expected repo/name.")
            continue

        repo_name, chart_name = chart_ref.split("/", 1)

        if repo_name not in repos:
            print(f"Repository '{repo_name}' not found for chart {chart_ref}")
            continue

        repo_url = repos[repo_name]

        if repo_url.startswith("oci://"):
            try:
                lock_data["charts"][chart_ref] = resolve_oci(
                    repo_url, chart_name, version
                )
                print(f"Resolved {chart_ref} {version} (OCI)")
            except Exception as e:
                print(f"Error resolving OCI chart {chart_ref}: {e}")
            continue

        if repo_name not in indices:
            indices[repo_name] = fetch_index(repo_url)

        index = indices[repo_name]
        if not index:
            print(f"Skipping {chart_ref} due to missing index.")
            continue

        entries = index.get("entries", {}).get(chart_name, [])

        # Find exact version
        matched_entry = None
        for entry in entries:
            if entry["version"] == version:
                matched_entry = entry
                break

        if not matched_entry:
            print(f"Version {version} not found for chart {chart_ref}")
            continue

        # Resolve URL
        urls = matched_entry.get("urls", [])
        if not urls:
            print(f"No URLs found for {chart_ref} version {version}")
            continue

        # URL can be relative or absolute
        chart_url = urls[0]
        if not chart_url.startswith("http"):
            chart_url = urljoin(repo_url + "/", chart_url)

        digest = matched_entry.get("digest")

        print(f"Resolved {chart_ref} {version} -> {chart_url}")

        lock_data["charts"][chart_ref] = {
            "version": version,
            "url": chart_url,
            "digest": digest,
        }

    print(f"Writing lockfile to {lockfile_path}...")
    with open(lockfile_path, "w") as f:
        json.dump(lock_data, f, indent=2, sort_keys=True)
        f.write("\n")


if __name__ == "__main__":
    main()