212 lines
7 KiB
Python
Executable file
212 lines
7 KiB
Python
Executable file
#!/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()
|