diff --git a/.gitignore b/.gitignore index 0f18a21..f281145 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ .claude *local* .venv -__pycache__ \ No newline at end of file +__pycache__ +*.DS_store diff --git a/README.md b/README.md index 676f3f0..68a3b90 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,9 @@ The scripts backing the actions live in `src/ci_cd` and are packaged as a uv pro - **`git-clone-ref`**: clones multiple repositories (used by the `git-clone` action) - **`get-kube`**: downloads a kubeconfig from Proton Pass +- **`get-kube-rancher`**: downloads a kubeconfig from a Rancher-managed cluster +- **`put-kube`**: uploads a kubeconfig to a Proton Pass item, replacing the existing attachment with the same filename +- **`refresh-kube`**: generates a kubeconfig from Rancher and replaces the Proton Pass attachment in one command ### get-kube @@ -151,6 +154,85 @@ export KUBECONFIG=$PWD/local.production_eu.kubeconfig.yaml The file is saved readable only by its owner (`0600`), since it contains cluster credentials. Running `uv run get-kube` with no arguments prints the help. +### get-kube-rancher + +Downloads the kubeconfig of a cluster managed by Rancher, using the Rancher v3 API. Each environment (`dev`, `e2e`, `prod`) has defaults hardcoded in the `ENVIRONMENTS` mapping in `src/ci_cd/get_kube_rancher.py`: the Rancher URL, the cluster ID or name, and the environment variable that holds the token. + +**Setup:** +- Fill in `rancher_url` and `cluster` for each environment in `ENVIRONMENTS`. +- Export the token for the environment (`RANCHER_TOKEN_DEV`, `RANCHER_TOKEN_E2E`, `RANCHER_TOKEN_PROD`), or use `RANCHER_TOKEN` as a fallback. Create it in Rancher under **Account & API Keys** and use the **Bearer Token** (`token-xxxxx:secret`). + +**Options:** +- **`-e, --environment`** (required, env `CICD_ENVIRONMENT`): `dev`, `e2e` or `prod` +- **`-c, --cluster`** (optional, env `CICD_RANCHER_CLUSTER`): Cluster ID or name, overriding the environment default +- **`-u, --rancher-url`** (optional, env `CICD_RANCHER_URL`): Rancher URL, overriding the environment default +- **`-o, --output-path`** (optional, env `CICD_OUTPUT_PATH`): Where to save the kubeconfig. Defaults to `local..kubeconfig.yaml`. + +**Example:** +~~~bash +export RANCHER_TOKEN_DEV='token-xxxxx:yyyyyyyy' +uv run get-kube-rancher -e dev +export KUBECONFIG=$PWD/local.desp-dev.kubeconfig.yaml +~~~ + +The file is saved readable only by its owner (`0600`). On Rancher 2.8 and later the token embedded in a generated kubeconfig expires after about 30 days, so regenerate it at the start of each pipeline run rather than storing it as a long-lived secret. + +### put-kube + +Uploads a local kubeconfig file to an existing Proton Pass item. The project is the vault name, and the environment is the item within that vault. The script replaces the remote attachment whose filename matches the uploaded file's basename. + +**Requirements:** +- `proton` CLI installed and on `PATH` +- Logged in with `proton account login` + +**Required options:** +- **`--project `**: Name of the project/vault containing the kubeconfig +- **`--environment `**: Name of the Kubernetes environment item to update +- **`--path `**: Path to the local kubeconfig file to upload + +**Example:** +~~~bash +uv run put-kube --project my-project --environment production --path ./kubeconfig.yaml +~~~ + +The existing attachment in the item must have the same filename as the local file, because the script uses the local file's basename to replace it. Running `uv run put-kube` without arguments displays the help. The command exits with a nonzero status if the file is missing, a Proton CLI command fails, or the project or environment cannot be found. + +### refresh-kube + +Admin pipeline that combines `get-kube-rancher` and `put-kube`: it generates a fresh kubeconfig from Rancher, validates it with `kubectl`, saves a local copy, and replaces the `kubeconfig.yaml` attachment of the matching Proton Pass item. + +One environment maps to one cluster and one Proton Pass item: + +| `--environment` | Rancher | Cluster and Proton item | +| --- | --- | --- | +| `dev` | `https://rancher.dev.desp.space` | `desp-dev` | +| `e2e` | `https://rancher.e2e3.desp.space` | `desp-e2e3` | +| `prod` | `https://rancher.prod.desp.space` | `desp-prod` | + +The Rancher defaults come from `ENVIRONMENTS` in `src/ci_cd/get_kube_rancher.py`. The Proton Pass vault and attachment name are configured in `PROTON_ENVIRONMENTS` in `src/ci_cd/refresh_kube.py`; pass `--project` or set `CICD_PROJECT` until the vault is filled in there. + +**Requirements:** +- `proton` CLI installed and on `PATH`, logged in with `proton account login` +- `kubectl` installed and on `PATH` unless `--no-verify` is used +- The Rancher token exported as `RANCHER_TOKEN_DEV`, `RANCHER_TOKEN_E2E` or `RANCHER_TOKEN_PROD` + +**Options:** +- **`-e, --environment`** (required, env `CICD_ENVIRONMENT`): `dev`, `e2e` or `prod` +- **`-p, --project`** (optional, env `CICD_PROJECT`): Proton Pass vault containing the kubeconfig item; overrides `PROTON_ENVIRONMENTS` +- **`--attachment`** (optional, env `CICD_ATTACHMENT`): Attachment name to replace, defaults to `kubeconfig.yaml` +- **`-o, --output-path`** (optional, env `CICD_OUTPUT_PATH`): Local copy path, defaults to `local..kubeconfig.yaml` +- **`--dry-run`**: generate, validate and write the local copy without uploading +- **`--verify/--no-verify`**: validate the kubeconfig and confirm the uploaded attachment matches; enabled by default + +**Example:** +~~~bash +export RANCHER_TOKEN_E2E='token-xxxxx:yyyyyyyy' +uv run refresh-kube -e e2e -p my-project +export KUBECONFIG=$PWD/local.desp-e2e3.kubeconfig.yaml +~~~ + +The command fails before uploading if the generated kubeconfig cannot reach the cluster, so a broken Rancher token does not replace a working credential. The local copy and the temporary upload file are both written with mode `0600`. + ## License ``` diff --git a/pyproject.toml b/pyproject.toml index d34bba1..0bd2caf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,9 @@ dependencies = [ [project.scripts] git-clone-ref = "ci_cd.git_clone_ref:app" get-kube = "ci_cd.get_kube:app" +get-kube-rancher = "ci_cd.get_kube_rancher:app" +put-kube = "ci_cd.put_kube:app" +refresh-kube = "ci_cd.refresh_kube:app" [build-system] requires = ["uv_build>=0.12.7,<0.13.0"] diff --git a/src/ci_cd/get_kube_rancher.py b/src/ci_cd/get_kube_rancher.py new file mode 100755 index 0000000..a9f5442 --- /dev/null +++ b/src/ci_cd/get_kube_rancher.py @@ -0,0 +1,293 @@ +#!/usr/bin/env python +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "typer", +# ] +# /// + +"""Download the kubeconfig of a cluster managed by Rancher.""" + +import json +import logging +import os +import pathlib +import urllib.error +import urllib.request +from dataclasses import dataclass +from enum import Enum +from typing import Annotated, Any +from urllib.parse import urljoin + +import typer + +logger = logging.getLogger(__name__) + +app = typer.Typer() + +API_VERSION = "v3" +REQUEST_TIMEOUT_SECONDS = 30 + + +class Environment(str, Enum): + """Environments with Rancher defaults configured in this script.""" + + dev = "dev" + e2e = "e2e" + prod = "prod" + + +@dataclass(frozen=True) +class EnvironmentConfig: + """Default Rancher settings for one environment.""" + + rancher_url: str + cluster: str + token_env: str + + +# Defaults per environment. Fill in the Rancher URL and the cluster ID (or +# name) of each instance. Secrets are never hardcoded: the token is read from +# the environment variable named by token_env, falling back to RANCHER_TOKEN. +ENVIRONMENTS: dict[str, EnvironmentConfig] = { + "dev": EnvironmentConfig( + rancher_url="https://rancher.dev.desp.space/", + cluster="desp-dev", + token_env="RANCHER_TOKEN_DEV", + ), + "e2e": EnvironmentConfig( + rancher_url="https://rancher.e2e3.desp.space/", + cluster="desp-e2e3", + token_env="RANCHER_TOKEN_E2E", + ), + "prod": EnvironmentConfig( + rancher_url="https://rancher.prod.desp.space/", + cluster="desp-prod", + token_env="RANCHER_TOKEN_PROD", + ), +} + + +class RancherError(RuntimeError): + """Raised when the Rancher API cannot be used.""" + + +def api_error_message(exc: urllib.error.HTTPError) -> str: + """Extract a readable message from a Rancher error response. + + Args: + exc: The HTTP error raised by urllib. + """ + body = exc.read().decode("utf-8", errors="replace").strip() + try: + payload = json.loads(body) + except json.JSONDecodeError: + return body or str(exc.reason) + if isinstance(payload, dict): + return str(payload.get("message") or payload.get("error") or body) + return body + + +def call_api( + url: str, + token: str, + *, + method: str = "GET", + payload: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Call the Rancher API and return the decoded JSON response. + + Args: + url: Absolute URL to call. + token: Rancher bearer token (``access_key:secret_key``). + method: HTTP method to use. + payload: Optional JSON body. + + Raises: + RancherError: If the request fails or the response is not JSON. + """ + data = json.dumps(payload).encode() if payload is not None else None + request = urllib.request.Request(url, data=data, method=method) + request.add_header("Authorization", f"Bearer {token}") + request.add_header("Accept", "application/json") + request.add_header("User-Agent", "ci-cd-get-kube-rancher") + if data is not None: + request.add_header("Content-Type", "application/json") + + logger.debug(f"{method} {url}") + try: + with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + return json.load(response) + except urllib.error.HTTPError as exc: + message = api_error_message(exc) + if exc.code == 401: + message += " (check that the Rancher token is valid and not expired)" + raise RancherError(f"{method} {url} failed with HTTP {exc.code}: {message}") from exc + except urllib.error.URLError as exc: + raise RancherError(f"cannot reach {url}: {exc.reason}") from exc + except json.JSONDecodeError as exc: + raise RancherError(f"{method} {url} returned an invalid JSON response") from exc + + +def get_token(config: EnvironmentConfig) -> str: + """Return the Rancher token for an environment. + + The environment-specific variable takes precedence over ``RANCHER_TOKEN``. + + Args: + config: Environment configuration holding the token variable name. + + Raises: + RancherError: If no token is set. + """ + for name in (config.token_env, "RANCHER_TOKEN"): + value = os.environ.get(name, "").strip() + if value: + logger.debug(f"Using Rancher token from {name}") + return value + raise RancherError(f"no Rancher token found; set {config.token_env} or RANCHER_TOKEN") + + +def get_cluster(rancher_url: str, token: str, selector: str) -> dict[str, Any]: + """Find a cluster by ID, falling back to an exact name match. + + Args: + rancher_url: Base URL of the Rancher instance. + token: Rancher bearer token. + selector: Cluster ID (``c-...``) or cluster name. + + Raises: + RancherError: If the cluster is missing or the name is ambiguous. + """ + clusters = call_api(f"{rancher_url}/{API_VERSION}/clusters", token).get("data", []) + for cluster in clusters: + if cluster.get("id") == selector: + return cluster + + matches = [cluster for cluster in clusters if cluster.get("name") == selector] + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + ids = ", ".join(str(cluster.get("id")) for cluster in matches) + raise RancherError(f"cluster name {selector!r} is ambiguous; matching IDs: {ids}") + + available = ", ".join( + sorted(f"{cluster.get('name')} ({cluster.get('id')})" for cluster in clusters) + ) + raise RancherError(f"cluster {selector!r} not found; available clusters: {available or 'none'}") + + +def generate_kubeconfig(rancher_url: str, token: str, cluster: dict[str, Any]) -> str: + """Ask Rancher to generate the kubeconfig of a cluster. + + Args: + rancher_url: Base URL of the Rancher instance. + token: Rancher bearer token. + cluster: Cluster resource returned by the Rancher API. + + Raises: + RancherError: If the action is unavailable or returns no kubeconfig. + """ + action = cluster.get("actions", {}).get("generateKubeconfig") + if not action: + raise RancherError( + f"cluster {cluster.get('name')!r} ({cluster.get('id')!r}) does not offer " + f"generateKubeconfig (state: {cluster.get('state')!r})" + ) + action_url = urljoin(f"{rancher_url}/", action) + response = call_api(action_url, token, method="POST") + kubeconfig = response.get("config") + if not kubeconfig: + raise RancherError("Rancher did not return a kubeconfig") + return kubeconfig + + +def write_kubeconfig(kubeconfig: str, output_path: pathlib.Path) -> None: + """Write a kubeconfig to disk with owner-only permissions. + + Args: + kubeconfig: Kubeconfig file content. + output_path: Destination path. + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(kubeconfig) + output_path.chmod(0o600) + logger.info(f"Downloaded kubeconfig to {str(output_path)!r}") + + +@app.command(no_args_is_help=True) +def main( + environment: Annotated[ + Environment, + typer.Option( + "--environment", + "-e", + envvar="CICD_ENVIRONMENT", + help="Target environment: dev, e2e or prod.", + ), + ], + cluster: Annotated[ + str | None, + typer.Option( + "--cluster", + "-c", + envvar="CICD_RANCHER_CLUSTER", + help="Cluster ID or name. Defaults to the value configured for the environment.", + ), + ] = None, + rancher_url: Annotated[ + str | None, + typer.Option( + "--rancher-url", + "-u", + envvar="CICD_RANCHER_URL", + help="Rancher URL. Defaults to the value configured for the environment.", + ), + ] = None, + output_path: Annotated[ + pathlib.Path | None, + typer.Option( + "--output-path", + "-o", + envvar="CICD_OUTPUT_PATH", + help="Path where the kubeconfig file will be saved.", + ), + ] = None, +) -> None: + """Download the kubeconfig of a Rancher-managed cluster.""" + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + try: + config = ENVIRONMENTS[environment.value] + target_url = (rancher_url or config.rancher_url).rstrip("/") + target_cluster = cluster or config.cluster + if not target_url: + raise RancherError( + f"no Rancher URL configured for environment {environment.value!r}; " + "set it in ENVIRONMENTS or pass --rancher-url" + ) + if not target_cluster: + raise RancherError( + f"no cluster configured for environment {environment.value!r}; " + "set it in ENVIRONMENTS or pass --cluster" + ) + if output_path is None: + output_path = pathlib.Path( + f"local.{target_cluster.replace(' ', '_')}.kubeconfig.yaml" + ) + + token = get_token(config) + logger.info( + f"Downloading kubeconfig for environment {environment.value!r} from " + f"{target_url!r} to {str(output_path)!r}" + ) + selected = get_cluster(target_url, token, target_cluster) + logger.debug(f"Using cluster {selected.get('name')!r} ({selected.get('id')!r})") + kubeconfig = generate_kubeconfig(target_url, token, selected) + write_kubeconfig(kubeconfig, output_path) + except RancherError as exc: + logger.error(str(exc)) + raise typer.Exit(1) + + +if __name__ == "__main__": + app() diff --git a/src/ci_cd/put_kube.py b/src/ci_cd/put_kube.py new file mode 100644 index 0000000..443159c --- /dev/null +++ b/src/ci_cd/put_kube.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "typer", +# ] +# /// + +import json +import subprocess +import typer +import os +import pathlib +import logging +from typing import Annotated + +app = typer.Typer() +logger = logging.getLogger(__name__) + +def run_proton_cli(*args: str) -> dict: + """Run Proton CLI with JSON output and return the decoded response.""" + result = subprocess.run( + ["proton", *args, "--output", "json"], + stdout=subprocess.PIPE, + check=True, + text=True + ) + return json.loads(result.stdout) + +def check_connection() -> bool: + """Check that the Proton CLI account is accessible. + + Returns: + True when Proton returns the account information. + + Raises: + subprocess.CalledProcessError: If the Proton command fails. + """ + connected = run_proton_cli("account", "get") + +def get_vault_id(vault: str) -> str: + """Return the share ID for the named Proton Pass vault. + + Args: + vault: Name of the vault to look up. + + Raises: + LookupError: If Proton returns a vault with a different name. + KeyError: If the response does not contain the expected fields. + """ + result = run_proton_cli("pass", "vaults", "get", vault) + if result["name"] != vault: + raise LookupError(f"Expected vault {vault!r}, got {result['name']!r}") + return result["share_id"] + +def get_item_id(project: str, vault_id: str, item: str) -> str: + """Return the ID of an item with the given name in a vault. + + Args: + vault: Share ID of the vault containing the item. + item: Name of the item to find. + + Raises: + LookupError: If no item with that name exists in the vault. + """ + result = run_proton_cli("pass", "items", "list", "--vault", vault_id, "--limit", "0") + for entry in result["items"]: + if entry["name"] == item: + return entry["item_id"] + raise LookupError(f"Item {item!r} not found in vault {project!r}") + +def update_item_attach(item: str, attachment: str) -> dict: + """Replace the matching remote attachment with a local file. + + Args: + vault: Share ID of the vault containing the item. + item: ID of the item to update. + attachment: Path to the local file to upload. Its basename is used + to identify the existing remote attachment to detach. + + Returns: + The decoded JSON response from Proton. + """ + filename = os.path.basename(attachment) + return run_proton_cli( + "pass", "items", "update", item, + "--attach", attachment, + "--detach", filename + ) + +@app.command(help="Update a kubeconfig attachment on an existing Proton Pass item with a local file.", no_args_is_help=True) +def main( + project: Annotated[ + str, + typer.Option(help="Name of the project containing the kubeconfig.") + ], + environment: Annotated[ + str, + typer.Option(help="Name of the kubernetes environment to update/upload.") + ], + path: Annotated[ + pathlib.Path, + typer.Option(help="Path to the local Kubeconfig file to upload") + ], +) -> None: + """Replace an item's kubeconfig attachment in Proton Pass. + + Checks the Proton account connection, resolves the vault and item IDs, + then uploads the local file and detaches the existing file with the same + basename. + """ + logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG) + try: + isconnected = check_connection() + logger.info("You are logged in to Proton Pass.") + vault_id = get_vault_id(project) + logger.info("The Vault defined is corect") + item_id = get_item_id(project, vault_id, environment) + logger.info("The item you want to update exists in the correspondent project.") + update = update_item_attach(item_id, path) + logger.info("The attachment has been updated successfully") + except FileNotFoundError as e: + logger.error(str(e)) + raise typer.Exit(3) + except subprocess.CalledProcessError as e: + logger.error("The proton-cli command exited with error.") + raise typer.Exit(1) + except LookupError as e: + logger.error(str(e)) + raise typer.Exit(5) + + +if __name__ == "__main__": + app() diff --git a/src/ci_cd/refresh_kube.py b/src/ci_cd/refresh_kube.py new file mode 100755 index 0000000..8ba24a8 --- /dev/null +++ b/src/ci_cd/refresh_kube.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "typer", +# ] +# /// + +"""Refresh a Rancher kubeconfig in Proton Pass in one command.""" + +import logging +import pathlib +import shutil +import subprocess +import tempfile +from dataclasses import dataclass +from typing import Annotated + +import typer + +from ci_cd.get_kube_rancher import ( + ENVIRONMENTS as RANCHER_ENVIRONMENTS, + Environment, + RancherError, + generate_kubeconfig, + get_cluster, + get_token, +) +from ci_cd.put_kube import get_item_id, get_vault_id, run_proton_cli, update_item_attach + +logger = logging.getLogger(__name__) + +app = typer.Typer() + + +class PipelineError(RuntimeError): + """Raised when the refresh pipeline cannot proceed.""" + + +@dataclass(frozen=True) +class ProtonConfig: + """Proton Pass settings for one environment.""" + + vault: str + attachment: str + + +# Proton Pass defaults per environment. The item is always named like the +# cluster; fill in the shared vault name (or pass --project / CICD_PROJECT). +PROTON_ENVIRONMENTS: dict[str, ProtonConfig] = { + "dev": ProtonConfig(vault="edh", attachment="kubeconfig.yaml"), + "e2e": ProtonConfig(vault="edh", attachment="kubeconfig.yaml"), + "prod": ProtonConfig(vault="edh", attachment="kubeconfig.yaml"), +} + + +def write_local_copy(source: pathlib.Path, output_path: pathlib.Path) -> None: + """Write the generated kubeconfig to a local file with mode 0600. + + Args: + source: Temporary file containing the generated kubeconfig. + output_path: Destination path. + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(source.read_bytes()) + output_path.chmod(0o600) + logger.info(f"Saved generated kubeconfig to {str(output_path)!r}") + + +def verify_kubeconfig(kubeconfig_path: pathlib.Path) -> None: + """Check that the generated kubeconfig can reach the cluster. + + Args: + kubeconfig_path: Path to the generated kubeconfig. + + Raises: + PipelineError: If kubectl cannot use the kubeconfig. + """ + logger.info("Validating the generated kubeconfig with kubectl") + try: + result = subprocess.run( + ["kubectl", "--kubeconfig", str(kubeconfig_path), "get", "--raw=/version"], + check=True, + capture_output=True, + text=True, + timeout=30, + ) + except subprocess.CalledProcessError as exc: + detail = (exc.stderr or exc.stdout or "").strip() + raise PipelineError( + f"generated kubeconfig cannot reach the cluster: {detail or exc}" + ) from exc + except subprocess.TimeoutExpired as exc: + raise PipelineError("timed out while validating the generated kubeconfig") from exc + logger.debug(f"Cluster reports: {result.stdout.strip()}") + + +def verify_uploaded_attachment( + item_id: str, attachment_name: str, expected_path: pathlib.Path +) -> None: + """Download the uploaded attachment and compare it with the local file. + + Args: + item_id: Proton Pass item ID. + attachment_name: Name of the attachment to download. + expected_path: Local file the remote attachment must match. + + Raises: + PipelineError: If the remote attachment cannot be downloaded or differs. + """ + with tempfile.TemporaryDirectory(prefix="refresh-kube-verify-") as tmp: + downloaded = pathlib.Path(tmp) / attachment_name + try: + subprocess.run( + [ + "proton", + "pass", + "items", + "attachments", + "download", + item_id, + attachment_name, + "--dest", + str(downloaded), + "--force", + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.CalledProcessError as exc: + detail = (exc.stderr or exc.stdout or "").strip() + raise PipelineError( + f"could not download the uploaded attachment back: {detail or exc}" + ) from exc + if downloaded.read_bytes() != expected_path.read_bytes(): + raise PipelineError("uploaded attachment does not match the generated kubeconfig") + logger.info(f"Verified attachment {attachment_name!r} on item {item_id!r}") + + +@app.command(no_args_is_help=True) +def main( + environment: Annotated[ + Environment, + typer.Option( + "--environment", + "-e", + envvar="CICD_ENVIRONMENT", + help="Target environment: dev, e2e or prod.", + ), + ], + project: Annotated[ + str | None, + typer.Option( + "--project", + "-p", + envvar="CICD_PROJECT", + help="Proton Pass vault containing the kubeconfig item.", + ), + ] = None, + attachment: Annotated[ + str | None, + typer.Option( + "--attachment", + envvar="CICD_ATTACHMENT", + help="Name of the Proton Pass attachment to replace.", + ), + ] = None, + output_path: Annotated[ + pathlib.Path | None, + typer.Option( + "--output-path", + "-o", + envvar="CICD_OUTPUT_PATH", + help="Path where the generated kubeconfig will be saved locally.", + ), + ] = None, + dry_run: Annotated[ + bool, + typer.Option( + "--dry-run", + help="Generate and validate the kubeconfig without uploading it.", + ), + ] = False, + verify: Annotated[ + bool, + typer.Option( + "--verify/--no-verify", + help="Validate the kubeconfig and verify the uploaded attachment.", + ), + ] = True, +) -> None: + """Generate a kubeconfig from Rancher and publish it to Proton Pass.""" + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + try: + rancher_config = RANCHER_ENVIRONMENTS[environment.value] + proton_config = PROTON_ENVIRONMENTS[environment.value] + vault = project or proton_config.vault + attachment_name = attachment or proton_config.attachment + if not vault: + raise PipelineError( + f"no Proton Pass vault configured for environment {environment.value!r}; " + "set it in PROTON_ENVIRONMENTS or pass --project" + ) + if not attachment_name: + raise PipelineError( + f"no attachment name configured for environment {environment.value!r}" + ) + if shutil.which("proton") is None: + raise PipelineError("proton CLI is not installed or not on PATH") + if verify and shutil.which("kubectl") is None: + raise PipelineError( + "kubectl is not installed or not on PATH; install it or pass --no-verify" + ) + if output_path is None: + output_path = pathlib.Path(f"local.{rancher_config.cluster}.kubeconfig.yaml") + + logger.info("Checking the Proton Pass session") + run_proton_cli("account", "get") + vault_id = get_vault_id(vault) + item_id = get_item_id(vault, vault_id, rancher_config.cluster) + logger.info(f"Found item {rancher_config.cluster!r} in vault {vault!r}") + + token = get_token(rancher_config) + rancher_url = rancher_config.rancher_url.rstrip("/") + logger.info( + f"Generating kubeconfig for cluster {rancher_config.cluster!r} " + f"from {rancher_url!r}" + ) + cluster = get_cluster(rancher_url, token, rancher_config.cluster) + kubeconfig = generate_kubeconfig(rancher_url, token, cluster) + + with tempfile.TemporaryDirectory(prefix="refresh-kube-") as tmp: + upload_path = pathlib.Path(tmp) / attachment_name + upload_path.write_text(kubeconfig) + upload_path.chmod(0o600) + if verify: + verify_kubeconfig(upload_path) + write_local_copy(upload_path, output_path) + if dry_run: + logger.info( + f"Dry run: would replace attachment {attachment_name!r} " + f"on item {rancher_config.cluster!r}" + ) + return + logger.info( + f"Uploading kubeconfig to Proton Pass item {rancher_config.cluster!r}" + ) + update_item_attach(item_id, upload_path) + if verify: + verify_uploaded_attachment(item_id, attachment_name, upload_path) + logger.info("Kubeconfig refreshed successfully") + except (PipelineError, RancherError) as exc: + logger.error(str(exc)) + raise typer.Exit(1) + except FileNotFoundError as exc: + logger.error(f"required command not found: {exc.filename or exc}") + raise typer.Exit(1) + except subprocess.CalledProcessError as exc: + detail = (exc.stderr or "").strip() + logger.error(f"command failed ({exc.cmd}): {detail or exc}") + raise typer.Exit(1) + except subprocess.TimeoutExpired as exc: + logger.error(f"command timed out: {exc.cmd}") + raise typer.Exit(1) + except (LookupError, KeyError, ValueError) as exc: + logger.error(str(exc)) + raise typer.Exit(1) + + +if __name__ == "__main__": + app()