feat: add scripts/add-secret.py

scripts/add-secret.py

@@ -0,0 +1,212 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# dependencies = ["typer", "ruamel.yaml"]
+# ///
+
+import json
+import shutil
+import subprocess
+import sys
+from difflib import unified_diff
+from pathlib import Path
+from typing import Annotated, Any, Optional
+
+import typer
+from ruamel.yaml import YAML
+
+app = typer.Typer(help="Add or update a secret field in an existing SealedSecret")
+
+
+def resolve_file(file: Path) -> Path:
+    if file.is_dir():
+        candidate = file / "sealedsecrets.yaml"
+        if not candidate.is_file():
+            raise typer.BadParameter(f"{file} is a directory and contains no sealedsecrets.yaml")
+        return candidate
+    if not file.is_file():
+        raise typer.BadParameter(f"{file} does not exist")
+    return file
+
+
+def looks_like_json(text: str) -> bool:
+    for char in text:
+        if char in (" ", "\t", "\n", "\r"):
+            continue
+        return char == "{"
+    return False
+
+
+def extract_header_comments(text: str) -> tuple[str, str]:
+    """Split leading comments/blank lines from the rest of the file."""
+    lines = text.splitlines(keepends=True)
+    idx = 0
+    while idx < len(lines):
+        stripped = lines[idx].strip()
+        if stripped == "" or stripped.startswith("#"):
+            idx += 1
+        else:
+            break
+    return "".join(lines[:idx]), "".join(lines[idx:])
+
+
+def load_documents(text: str) -> list[Any]:
+    if looks_like_json(text):
+        # JSON is valid YAML, but parse it explicitly so we can normalize it to YAML output.
+        return [json.loads(text)]
+    yaml = YAML(typ="rt")
+    yaml.preserve_quotes = True
+    return list(yaml.load_all(text))
+
+
+def _represent_none(representer, data):
+    return representer.represent_scalar("tag:yaml.org,2002:null", "null")
+
+
+def dump_documents(docs: list[Any]) -> str:
+    yaml = YAML(typ="rt")
+    yaml.default_flow_style = False
+    yaml.preserve_quotes = True
+    yaml.width = 4096
+    yaml.explicit_start = True
+    yaml.representer.add_representer(type(None), _represent_none)
+    from ruamel.yaml.compat import StringIO
+
+    stream = StringIO()
+    yaml.dump_all(docs, stream)
+    return stream.getvalue()
+
+
+def detect_scope(doc: Any) -> str:
+    annotations = doc.get("metadata", {}).get("annotations", {}) or {}
+    for key, value in annotations.items():
+        if key == "sealedsecrets.bitnami.com/cluster-wide" and str(value).lower() == "true":
+            return "cluster-wide"
+        if key == "sealedsecrets.bitnami.com/namespace-wide" and str(value).lower() == "true":
+            return "namespace-wide"
+    return "strict"
+
+
+def seal_value(value: str, doc: Any, cert: Optional[Path]) -> str:
+    metadata = doc.get("metadata", {})
+    spec = doc.get("spec", {})
+
+    name = metadata.get("name")
+    if not name:
+        raise typer.BadParameter("SealedSecret has no metadata.name")
+
+    namespace = metadata.get("namespace") or spec.get("template", {}).get("metadata", {}).get("namespace")
+    if not namespace:
+        raise typer.BadParameter("SealedSecret has no namespace in metadata or spec.template.metadata")
+
+    cmd = ["kubeseal", "--raw", "--from-file=/dev/stdin"]
+    if cert:
+        cmd.extend(["--cert", str(cert)])
+
+    scope = detect_scope(doc)
+    if scope == "cluster-wide":
+        cmd.extend(["--scope", "cluster-wide"])
+    elif scope == "namespace-wide":
+        cmd.extend(["--namespace", namespace, "--scope", "namespace-wide"])
+    else:
+        cmd.extend(["--namespace", namespace, "--name", name])
+
+    result = subprocess.run(
+        cmd,
+        input=value,
+        text=True,
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        typer.echo(f"kubeseal failed:\n{result.stderr}", err=True)
+        raise typer.Exit(result.returncode)
+
+    return result.stdout.strip()
+
+
+def find_target_document(docs: list[Any], key: str, secret_name: Optional[str]) -> Any:
+    if secret_name:
+        for doc in docs:
+            if doc.get("metadata", {}).get("name") == secret_name:
+                return doc
+        raise typer.BadParameter(f"No SealedSecret named '{secret_name}' found in file")
+
+    if len(docs) == 1:
+        return docs[0]
+
+    candidates = [
+        doc for doc in docs
+        if key in (doc.get("spec", {}).get("encryptedData", {}) or {})
+    ]
+    if len(candidates) == 1:
+        return candidates[0]
+    if len(candidates) > 1:
+        names = [doc.get("metadata", {}).get("name", "<unnamed>") for doc in candidates]
+        raise typer.BadParameter(
+            f"Key '{key}' exists in multiple SealedSecrets ({', '.join(names)}). "
+            "Use --secret-name to choose one."
+        )
+
+    names = [doc.get("metadata", {}).get("name", "<unnamed>") for doc in docs]
+    raise typer.BadParameter(
+        f"File contains multiple SealedSecrets and key '{key}' does not exist yet. "
+        f"Available: {', '.join(names)}. Use --secret-name to choose one."
+    )
+
+
+@app.command()
+def add(
+    name: Annotated[str, typer.Option("--name", help="Key name in SealedSecret encryptedData")],
+    secret: Annotated[str, typer.Option("--secret", help="Plaintext secret value")],
+    file: Annotated[Path, typer.Option("--file", help="Path to sealedsecrets.yaml or its directory")],
+    secret_name: Annotated[
+        Optional[str],
+        typer.Option("--secret-name", help="Name of the SealedSecret resource to update (required for multi-secret files when the key is new)"),
+    ] = None,
+    cert: Annotated[
+        Optional[Path],
+        typer.Option("--cert", help="Path to a kubeseal certificate for offline sealing"),
+    ] = None,
+    dry_run: Annotated[bool, typer.Option("--dry-run", help="Show diff without writing")] = False,
+) -> None:
+    target_path = resolve_file(file)
+    original_text = target_path.read_text()
+    header, body = extract_header_comments(original_text)
+
+    docs = load_documents(body)
+    if not docs:
+        raise typer.BadParameter(f"{target_path} contains no documents")
+
+    target = find_target_document(docs, name, secret_name)
+    sealed = seal_value(secret, target, cert)
+
+    if "spec" not in target or target["spec"] is None:
+        target["spec"] = {}
+    if "encryptedData" not in target["spec"] or target["spec"]["encryptedData"] is None:
+        target["spec"]["encryptedData"] = {}
+
+    old_value = target["spec"]["encryptedData"].get(name)
+    action = "updated" if old_value is not None else "added"
+    target["spec"]["encryptedData"][name] = sealed
+
+    new_text = header + dump_documents(docs)
+
+    if dry_run:
+        diff = unified_diff(
+            original_text.splitlines(keepends=True),
+            new_text.splitlines(keepends=True),
+            fromfile=str(target_path),
+            tofile=str(target_path),
+        )
+        sys.stdout.writelines(diff)
+        raise typer.Exit()
+
+    backup_path = target_path.with_suffix(target_path.suffix + ".bak")
+    shutil.copy2(target_path, backup_path)
+    target_path.write_text(new_text)
+
+    typer.echo(f"{action} '{name}' in SealedSecret '{target.get('metadata', {}).get('name')}' ({target_path})")
+    typer.echo(f"backup saved to {backup_path}")
+
+
+if __name__ == "__main__":
+    app()