studio/cli
2
0
Fork 0
Code Issues Pull Requests Actions Packages Releases 4 Wiki Activity

refactor: cross-platform tool downloads, configurable infra dir and ACME email

Browse Source 
- Make tool downloads platform-aware (darwin/linux, arm64/amd64)
- Add buildctl to bundled tools
- Add get_infra_dir() with config fallback for REPO_ROOT resolution
- Add ACME email to sunbeam config (set/get)
- Add REGISTRY_HOST_IP substitution in kustomize builds
- Update Kratos admin identity schema to employee
- Fix logs command to use production tunnel and context
This commit is contained in:
Sienna Meridian Satterwhite
2026-03-10 19:37:02 +00:00
parent c82f15b190
commit d5b963253b
6 changed files with 132 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
sunbeam/cluster.py
View File

@@ -12,7 +12,8 @@ from sunbeam.tools import run_tool, CACHE_DIR
from sunbeam.output import step, ok, warn, die
LIMA_VM = "sunbeam"
SECRETS_DIR = Path(__file__).parents[3] / "infrastructure" / "secrets" / "local"
from sunbeam.config import get_infra_dir as _get_infra_dir
SECRETS_DIR = _get_infra_dir() / "secrets" / "local"
GITEA_ADMIN_USER = "gitea_admin"

25
sunbeam/config.py
View File

@@ -11,15 +11,18 @@ CONFIG_PATH = Path.home() / ".sunbeam.json"
class SunbeamConfig:
"""Sunbeam configuration with production host and infrastructure directory."""
def __init__(self, production_host: str = "", infra_directory: str = ""):
def __init__(self, production_host: str = "", infra_directory: str = "",
acme_email: str = ""):
self.production_host = production_host
self.infra_directory = infra_directory
self.acme_email = acme_email
def to_dict(self) -> dict:
"""Convert configuration to dictionary for JSON serialization."""
return {
"production_host": self.production_host,
"infra_directory": self.infra_directory,
"acme_email": self.acme_email,
}
@classmethod
@@ -28,6 +31,7 @@ class SunbeamConfig:
return cls(
production_host=data.get("production_host", ""),
infra_directory=data.get("infra_directory", ""),
acme_email=data.get("acme_email", ""),
)
@@ -71,3 +75,22 @@ def get_infra_directory() -> str:
"""Get infrastructure directory from config."""
config = load_config()
return config.infra_directory
def get_infra_dir() -> "Path":
"""Infrastructure manifests directory as a Path.
Prefers the configured infra_directory; falls back to the package-relative
path (works when running from the development checkout).
"""
from pathlib import Path
configured = load_config().infra_directory
if configured:
return Path(configured)
# Dev fallback: cli/sunbeam/config.py → parents[0]=cli/sunbeam, [1]=cli, [2]=monorepo root
return Path(__file__).resolve().parents[2] / "infrastructure"
def get_repo_root() -> "Path":
"""Monorepo root directory (parent of the infrastructure directory)."""
return get_infra_dir().parent

51
sunbeam/images.py
View File

@@ -645,52 +645,11 @@ def _build_kratos_admin(push: bool = False, deploy: bool = False):
step(f"Building kratos-admin-ui -> {image} ...")
if env.is_prod:
# Cross-compile Deno for x86_64 and package into a minimal image.
if not shutil.which("deno"):
die("deno not found — install Deno: https://deno.land/")
if not shutil.which("npm"):
die("npm not found — install Node.js")
ok("Building UI assets (npm run build)...")
_run(["npm", "run", "build"], cwd=str(kratos_admin_dir / "ui"))
ok("Cross-compiling Deno binary for x86_64-linux-gnu...")
_run([
"deno", "compile",
"--target", "x86_64-unknown-linux-gnu",
"--allow-net", "--allow-read", "--allow-env",
"--include", "ui/dist",
"-o", "kratos-admin-x86_64",
"main.ts",
], cwd=str(kratos_admin_dir))
bin_path = kratos_admin_dir / "kratos-admin-x86_64"
if not bin_path.exists():
die("Deno cross-compilation produced no binary")
pkg_dir = Path(tempfile.mkdtemp(prefix="kratos-admin-pkg-"))
shutil.copy2(str(bin_path), str(pkg_dir / "kratos-admin"))
dockerfile = pkg_dir / "Dockerfile"
dockerfile.write_text(
"FROM gcr.io/distroless/cc-debian12:nonroot\n"
"WORKDIR /app\n"
"COPY kratos-admin ./\n"
"EXPOSE 3000\n"
'ENTRYPOINT ["/app/kratos-admin"]\n'
)
try:
_build_image(env, image, dockerfile, pkg_dir, push=push)
finally:
shutil.rmtree(str(pkg_dir), ignore_errors=True)
else:
# Local: buildkitd handles the full Dockerfile build
_build_image(
env, image,
kratos_admin_dir / "Dockerfile", kratos_admin_dir,
push=push,
)
_build_image(
env, image,
kratos_admin_dir / "Dockerfile", kratos_admin_dir,
push=push,
)
if deploy:
_deploy_rollout(env, ["kratos-admin-ui"], "ory", timeout="120s")

18
sunbeam/kube.py
View File

@@ -227,6 +227,7 @@ def cmd_bao(bao_args: list[str]) -> int:
def kustomize_build(overlay: Path, domain: str, email: str = "") -> str:
"""Run kustomize build --enable-helm and apply domain/email substitution."""
import socket as _socket
r = run_tool(
"kustomize", "build", "--enable-helm", str(overlay),
capture_output=True, text=True, check=True,
@@ -235,5 +236,22 @@ def kustomize_build(overlay: Path, domain: str, email: str = "") -> str:
text = domain_replace(text, domain)
if email:
text = text.replace("ACME_EMAIL", email)
if "REGISTRY_HOST_IP" in text:
registry_ip = ""
try:
registry_ip = _socket.gethostbyname(f"src.{domain}")
except _socket.gaierror:
pass
if not registry_ip:
# DNS not resolvable locally (VPN, split-horizon, etc.) — derive IP from SSH host config
from sunbeam.config import get_production_host as _get_host
ssh_host = _get_host()
# ssh_host may be "user@host" or just "host"
raw = ssh_host.split("@")[-1].split(":")[0]
try:
registry_ip = _socket.gethostbyname(raw)
except _socket.gaierror:
registry_ip = raw # raw is already an IP in typical config
text = text.replace("REGISTRY_HOST_IP", registry_ip)
text = text.replace("\n annotations: null", "")
return text

18
sunbeam/tests/test_cli.py
View File

@@ -643,9 +643,13 @@ class TestConfigCli(unittest.TestCase):
def test_config_cli_set_dispatch(self):
"""Test that config set CLI dispatches correctly."""
mock_existing = MagicMock()
mock_existing.production_host = "old@example.com"
mock_existing.infra_directory = "/old/infra"
mock_existing.acme_email = ""
mock_save = MagicMock()
mock_config = MagicMock(
SunbeamConfig=MagicMock(return_value="mock_config"),
load_config=MagicMock(return_value=mock_existing),
save_config=mock_save
)
@@ -660,13 +664,11 @@ class TestConfigCli(unittest.TestCase):
except SystemExit:
pass
# Verify SunbeamConfig was called with correct args
mock_config.SunbeamConfig.assert_called_once_with(
production_host="cli@example.com",
infra_directory="/cli/infra"
)
# Verify save_config was called
mock_save.assert_called_once_with("mock_config")
# Verify existing config was loaded and updated
self.assertEqual(mock_existing.production_host, "cli@example.com")
self.assertEqual(mock_existing.infra_directory, "/cli/infra")
# Verify save_config was called with the updated config
mock_save.assert_called_once_with(mock_existing)
def test_config_cli_get_dispatch(self):
"""Test that config get CLI dispatches correctly."""

91
sunbeam/tools.py
View File

@@ -1,10 +1,13 @@
"""Binary bundler — downloads kubectl, kustomize, helm at pinned versions.
"""Binary bundler — downloads kubectl, kustomize, helm, buildctl at pinned versions.
Binaries are cached in ~/.local/share/sunbeam/bin/ and SHA256-verified.
Platform (OS + arch) is detected at runtime so the same package works on
darwin/arm64 (development Mac), darwin/amd64, linux/arm64, and linux/amd64.
"""
import hashlib
import io
import os
import platform
import stat
import subprocess
import tarfile
@@ -13,26 +16,79 @@ from pathlib import Path
CACHE_DIR = Path.home() / ".local/share/sunbeam/bin"
TOOLS: dict[str, dict] = {
# Tool specs — URL and extract templates use {version}, {os}, {arch}.
# {os} : darwin | linux
# {arch} : arm64 | amd64
_TOOL_SPECS: dict[str, dict] = {
"kubectl": {
"version": "v1.32.2",
"url": "https://dl.k8s.io/release/v1.32.2/bin/darwin/arm64/kubectl",
"sha256": "", # set to actual hash; empty = skip verify
"url": "https://dl.k8s.io/release/{version}/bin/{os}/{arch}/kubectl",
# plain binary, no archive
},
"kustomize": {
"version": "v5.8.1",
"url": "https://github.com/kubernetes-sigs/kustomize/releases/download/kustomize%2Fv5.8.1/kustomize_v5.8.1_darwin_arm64.tar.gz",
"sha256": "",
"url": (
"https://github.com/kubernetes-sigs/kustomize/releases/download/"
"kustomize%2F{version}/kustomize_{version}_{os}_{arch}.tar.gz"
),
"extract": "kustomize",
},
"helm": {
"version": "v4.1.0",
"url": "https://get.helm.sh/helm-v4.1.0-darwin-arm64.tar.gz",
"sha256": "82f7065bf4e08d4c8d7881b85c0a080581ef4968a4ae6df4e7b432f8f7a88d0c",
"extract": "darwin-arm64/helm",
"url": "https://get.helm.sh/helm-{version}-{os}-{arch}.tar.gz",
"extract": "{os}-{arch}/helm",
"sha256": {
"darwin_arm64": "82f7065bf4e08d4c8d7881b85c0a080581ef4968a4ae6df4e7b432f8f7a88d0c",
},
},
"buildctl": {
"version": "v0.28.0",
# BuildKit releases: buildkit-v0.28.0.linux.amd64.tar.gz
"url": (
"https://github.com/moby/buildkit/releases/download/{version}/"
"buildkit-{version}.{os}-{arch}.tar.gz"
),
"extract": "bin/buildctl",
},
}
# Expose as TOOLS for callers that do `if "helm" in TOOLS`.
TOOLS = _TOOL_SPECS
def _detect_platform() -> tuple[str, str]:
"""Return (os_name, arch) for the current host."""
sys_os = platform.system().lower()
machine = platform.machine().lower()
os_name = {"darwin": "darwin", "linux": "linux"}.get(sys_os)
if not os_name:
raise RuntimeError(f"Unsupported OS: {sys_os}")
arch = "arm64" if machine in ("arm64", "aarch64") else "amd64"
return os_name, arch
def _resolve_spec(name: str) -> dict:
"""Return a tool spec with {os} / {arch} / {version} substituted.
Uses the module-level TOOLS dict so that tests can patch it.
"""
if name not in TOOLS:
raise ValueError(f"Unknown tool: {name}")
os_name, arch = _detect_platform()
raw = TOOLS[name]
version = raw.get("version", "")
fmt = {"version": version, "os": os_name, "arch": arch}
spec = dict(raw)
spec["version"] = version
spec["url"] = raw["url"].format(**fmt)
if "extract" in raw:
spec["extract"] = raw["extract"].format(**fmt)
# sha256 may be a per-platform dict {"darwin_arm64": "..."} or a plain string.
sha256_val = raw.get("sha256", {})
if isinstance(sha256_val, dict):
spec["sha256"] = sha256_val.get(f"{os_name}_{arch}", "")
return spec
def _sha256(path: Path) -> str:
h = hashlib.sha256()
@@ -45,12 +101,10 @@ def _sha256(path: Path) -> str:
def ensure_tool(name: str) -> Path:
"""Return path to cached binary, downloading + verifying if needed.
Re-downloads automatically when the pinned version in TOOLS changes.
Re-downloads automatically when the pinned version in _TOOL_SPECS changes.
A <name>.version sidecar file records the version of the cached binary.
"""
if name not in TOOLS:
raise ValueError(f"Unknown tool: {name}")
spec = TOOLS[name]
spec = _resolve_spec(name)
CACHE_DIR.mkdir(parents=True, exist_ok=True)
dest = CACHE_DIR / name
version_file = CACHE_DIR / f"{name}.version"
@@ -58,7 +112,6 @@ def ensure_tool(name: str) -> Path:
expected_sha = spec.get("sha256", "")
expected_version = spec.get("version", "")
# Use cached binary if version matches (or no version pinned) and SHA passes
if dest.exists():
version_ok = (
not expected_version
@@ -67,18 +120,17 @@ def ensure_tool(name: str) -> Path:
sha_ok = not expected_sha or _sha256(dest) == expected_sha
if version_ok and sha_ok:
return dest
# Version mismatch or SHA mismatch — re-download
if dest.exists():
dest.unlink()
if version_file.exists():
version_file.unlink()
# Download
url = spec["url"]
with urllib.request.urlopen(url) as resp: # noqa: S310
data = resp.read()
# Extract from tar.gz if needed
extract_path = spec.get("extract")
if extract_path:
with tarfile.open(fileobj=io.BytesIO(data)) as tf:
@@ -88,10 +140,8 @@ def ensure_tool(name: str) -> Path:
else:
binary_data = data
# Write to cache
dest.write_bytes(binary_data)
# Verify SHA256 (after extraction)
if expected_sha:
actual = _sha256(dest)
if actual != expected_sha:
@@ -100,9 +150,7 @@ def ensure_tool(name: str) -> Path:
f"SHA256 mismatch for {name}: expected {expected_sha}, got {actual}"
)
# Make executable
dest.chmod(dest.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
# Record version so future calls skip re-download when version unchanged
version_file.write_text(expected_version)
return dest
@@ -116,9 +164,8 @@ def run_tool(name: str, *args, **kwargs) -> subprocess.CompletedProcess:
env = kwargs.pop("env", None)
if env is None:
env = os.environ.copy()
# kustomize needs helm on PATH for helm chart rendering
if name == "kustomize":
if "helm" in TOOLS:
ensure_tool("helm") # ensure bundled helm is present before kustomize runs
ensure_tool("helm")
env["PATH"] = str(CACHE_DIR) + os.pathsep + env.get("PATH", "")
return subprocess.run([str(bin_path), *args], env=env, **kwargs)