Helper functions to manage k8s secrets more easily

k8secret.py

#!/usr/bin/env python3
# Helper functions to manage k8s secrets more easily

import os
import sys
import yaml
import base64
import argparse


template = """apiVersion: v1
kind: Secret
metadata:
  name: {name}
  namespace: {namespace}
type: Opaque
data:
  {secrets}
"""


def create(args):
    if " " in args.name:
        raise ValueError("secret name cannot contain spaces")

    data = {}
    data["apiVersion"] = "v1"
    data["kind"] = "Secret"
    data["metadata"] = {}
    data["type"] = "Opaque"
    data["data"] = {}

    data["metadata"]["name"] = args.name
    if args.namespace:
        data["metadata"]["namespace"] = args.namespace

    for pair in args.pairs:
        key, value = pair.split("=", 1)
        if not key or not value:
            raise ValueError(f"Invalid key/value pair: {pair}")
        value = base64.b64encode(value.encode("utf-8")).decode("utf-8")
        data["data"][key] = value

    out = args.out or f"{args.name}.yaml"
    with open(out, "w") as f:
        yaml.dump(data, f, sort_keys=False)


def upsert(args):
    key = args.key[0]
    value = args.value[0]

    with open(args.secret, "r") as f:
        data = yaml.safe_load(f)

    value = base64.b64encode(value.encode("utf-8")).decode("utf-8")
    data["data"][key] = value

    with open(args.out or args.secret, "w") as f:
        yaml.dump(data, f, sort_keys=False)


def encode(args):
    config = {
        "name": args.name,
        "namespace": args.namespace,
    }

    secrets = {}
    for path in args.files:
        key = os.path.basename(path)
        with open(path, "rb") as f:
            secrets[key] = base64.b64encode(f.read()).decode("ascii")

    config["secrets"] = "\n  ".join([f"{key}: {val}" for key, val in secrets.items()])
    with open(args.out, "w") as f:
        f.write(template.format(**config))


def decode(args):
    for path in args.secrets:
        with open(path, 'r') as f:
            data = yaml.safe_load(f)['data']

            if args.key:
                if args.key in data:
                    if args.print:
                        val = base64.b64decode(data[args.key]).decode("utf-8")
                        print(f"{args.key}: {val}")

                    else:
                        with open(os.path.join(args.out, "wb")) as f:
                            f.write(base64.b64decode(data[args.key]))

            else:
                # Handle data
                for key in data:
                    if args.print:
                        val = base64.b64decode(data[key]).decode("utf-8")
                        print(f"{key}: {val}")

                    else:
                        with open(os.path.join(args.out, key), "wb") as f:
                            f.write(base64.b64decode(data[key]))


def prompt(query, default="yes"):
    suggest = {
        "yes": " [Y/n] ",
        "no": " [y/N] ",
        None: " [y/n] ",
    }[default]

    while True:
        sys.stdout.write(query + suggest)
        choice = input().lower() or default
        if choice.startswith("y"):
            return True
        elif choice.startswith("n"):
            return False
        else:
            sys.stdout.write("please specify yes or no\n")


if __name__ == "__main__":
    # Describe the CLI commands and arguments
    cmds = {
        "create": {
            "help": "create a new k8s secret from key/value pairs",
            "func": create,
            "args": {
                ("-n", "--name"): {
                    "type": str, "required": True,
                    "help": "the name of the k8s secret to create"
                },
                ("-N", "--namespace"): {
                    "type": str, "default": None,
                    "help": "the namespace of the k8s secret to create",
                },
                ("-o", "--out"): {
                    "type": str, "default": None,
                    "help": "the path to write the secret YAML to (default: <name>.yaml)",
                },
                "pairs": {
                    "nargs": "+",
                    "help": "key/value pairs to add to the secret (format: key=value)"
                },
            }
        },
        "upsert": {
            "help": "insert or update a k8s secret key/value pair",
            "func": upsert,
            "args": {
                "key": {
                    "type": str, "nargs": 1,
                    "help": "the key of the secret to insert or update"
                },
                "value": {
                    "type": str, "nargs": 1,
                    "help": "the value of the secret to insert or update"
                },
                ("-s", "--secret"): {
                    "type": str, "required": True, "metavar": "IN",
                    "help": "the path to the k8s secret file to update",
                },
                ("-o", "--out"): {
                    "type": str, "default": None,
                    "help": "the path to write the updated secret file to (default: overwrite the original)",
                },
            }
        },
        "encode": {
            "help": "encode secrets as a k8s secrets object in yaml format",
            "func": encode,
            "args": {
                "files": {
                    "nargs": "+",
                    "help": "file(s) to encode into secret objects (keys will be basename)"
                },
                ("-o", "--out"): {
                    "type": str, "default": "secret.yaml",
                    "help": "path to write the secret YAML to",
                },
                ("-n", "--name"): {
                    "type": str, "default": "mtls-certs",
                    "help": "the name of the k8s secret to add to the metadata",
                },
                ("-N", "--namespace"): {
                    "type": str, "default": "default",
                    "help": "the namespace of the k8s secret to add to the metadata",
                },
            }
        },
        "decode": {
            "help": "decode secrets from a k8s secrets object in yaml format",
            "func": decode,
            "args": {
                "secrets": {
                    "nargs": "+",
                    "help": "path to the secrets file(s) to decode",
                },
                ("-o", "--out"): {
                    "type": str, "default": ".",
                    "help": "directory to write the secret files out to"
                },
                ("-k", "--key"): {
                    "type": str, "default": None,
                    "help": "specify the key of the secret to extract",
                },
                ("-p", "--print"): {
                    "action": "store_true", "default": False,
                    "help": "instead of writing secret files, just print key/value pairs (useful for username/passwords)"
                },
            }
        },
    }

    # Create the CLI parser
    parser = argparse.ArgumentParser(
        description="helper script to manage k8s secrets",
    )
    subparsers = parser.add_subparsers(title="actions", description="utility commands")
    for cmd, cargs in cmds.items():
        cmdparser = subparsers.add_parser(cmd, help=cargs.get('help'))
        cmdparser.set_defaults(func=cargs.get('func'))
        for pargs, kwargs in cargs.get("args", {}).items():
            if isinstance(pargs, str):
                pargs = (pargs,)
            cmdparser.add_argument(*pargs, **kwargs)

    # Execute the command
    args = parser.parse_args()
    if hasattr(args, "func"):
        try:
            args.func(args)
        except Exception as e:
            parser.error(e)
    else:
        parser.print_help()