Managing your Sitecore secrets within a K8s setup

project-root/
│
├── k8s/                                 # Kubernetes configuration files
│   ├── qa/
│   │   ├── secret-classes/              # Environment-specific secret class YAML files for QA
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml                      # Service-specific YAML files (e.g., cm, cd, id, rh)
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   ├── uat/
│   │   ├── secret-classes/              # Environment-specific secret class YAML files for UAT
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   ├── stg/
│   │   ├── secret-classes/              # Environment-specific secret class YAML files for STG
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   └── prod/
│       ├── secret-classes/              # Environment-specific secret class YAML files for PROD
│       │   ├── sitecore-secrets.yaml
│       │   ├── rh-secrets.yaml
│       │   └── common-secrets.yaml
│       ├── cm.yaml
│       ├── cd.yaml
│       └── rh.yaml

#!/usr/bin/env python3

from ruamel.yaml import YAML
from ruamel.yaml.scalarstring import LiteralScalarString
import os
import argparse
from contextlib import chdir, contextmanager
import collections.abc
from pathlib import Path
from typing import List, Optional, Dict, Any, Iterator, Union
import logging
from enum import Enum, auto

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Environment(Enum):
    QA = auto()
    UAT = auto()
    STG = auto()
    PROD = auto()

class SecretManager:
    def __init__(self):
        self.file_maps = {"sitecore-secrets.yaml": ["cm", "cd", "id"], "rh-secrets.yaml": ["rh"], "common-secrets.yaml": []}
        self.envs: List[str] = [env.name.lower() for env in Environment]

    def get_yaml(self) -> YAML:
        yaml = YAML()
        yaml.preserve_quotes = True
        yaml.width = 4096
        return yaml

    @contextmanager
    def open_yaml(self, file_name: Union[str, Path], mode: str = 'r') -> Iterator[Any]:
        with open(file_name, mode) as file:
            yield file
  
    def get_secret_file_for_services(self, serviceNames: str) -> Optional[str]:
        services: List[str] = [service.strip() for service in serviceNames.split(",")]
        secret_files: List[str] = list(set(key for service in services for key, value in self.file_maps.items() if service in value))
        if len(secret_files) > 1:
            return "common-secrets.yaml"
        elif len(secret_files) == 1:
            return secret_files[0]
        else:
            return None 
  
    def _load_secret_class(self, yaml: YAML, file_path: Path) -> Dict[str, Any]:
        with self.open_yaml(file_path) as file:
            return yaml.load(file)

    def _save_secret_class(self, yaml: YAML, file_path: Path, secret_class: Dict[str, Any]) -> None:
        with self.open_yaml(file_path, 'w') as file:
            yaml.dump(secret_class, file)

    def _find_or_create_secret_object(self, secret_objects: List[Dict[str, Any]], secret_name: str, object_name: str) -> None:
        secret_object = next((obj for obj in secret_objects if obj['secretName'] == secret_name), None)
        if not secret_object:
            secret_object = {'secretName': secret_name, 'type': 'Opaque', 'data': []}
            secret_objects.append(secret_object)
        secret_data = secret_object['data']
        result = filter(lambda x: x['objectName'] == object_name, secret_data)
        secret = next(result, None)
        if not secret:
            logger.info(f"Adding new secret object - {object_name}")
            secret_data.append({'objectName': object_name, 'key': object_name})
            secret = secret_data[-1]
        else:
            logger.warning(f"Secret object found - {object_name}")

    def _update_secret_param_objects(self, secret_class: Dict[str, Any], object_name: str) -> None:
        secret_param_objects = str(secret_class['spec']['parameters']['objects'])
        secret_param_objects = secret_param_objects.rstrip()
        current_array = secret_param_objects.splitlines()
        last_line = current_array[-1]
        leading_whitespace = last_line[:len(last_line) - len(last_line.lstrip())]
        secret_object_entry = f'objectName: {object_name}'
        if secret_object_entry not in secret_param_objects:
            new_object = LiteralScalarString(f"{leading_whitespace}{secret_object_entry}\n{leading_whitespace}objectType: secret\n")
            current_array.append(f"{leading_whitespace[:-2]}- |\n{new_object}")
            secret_class['spec']['parameters']['objects'] = "\n".join(current_array)

    def write_secret_to_secret_class(self, secret_name: str, object_name: str, secret_file: Optional[str]) -> None:
        yaml: YAML = self.get_yaml()
        yaml.indent(mapping=2, sequence=4, offset=2)
        if secret_file is None:
            logger.error("No secret file found for the given services")
            return  
        file_path: Path = Path('secret-classes') / secret_file
        if not file_path.exists():
            logger.error(f"File not found - {file_path}")
            return
        secret_class = self._load_secret_class(yaml, file_path)
        secret_objects = secret_class['spec']['secretObjects']
        self._find_or_create_secret_object(secret_objects, secret_name, object_name)
        self._update_secret_param_objects(secret_class, object_name)
        logger.info(f"Writing changes to file {secret_file} now")
        self._save_secret_class(yaml, file_path, secret_class)


    def _load_spec_file(self, yaml: YAML, file_name: str) -> List[Dict[str, Any]]:
        with self.open_yaml(file_name) as file:
            spec = list(yaml.load_all(file))
        return spec if isinstance(spec, collections.abc.Sequence) else [spec]

    def _save_spec_file(self, yaml: YAML, file_name: str, spec: List[Dict[str, Any]]) -> None:
        with self.open_yaml(file_name, 'w') as file:
            yaml.dump_all(spec, file)

    def _update_env_object(self, env_objects: List[Dict[str, Any]], env_var_name: str, secret_name: str, object_name: str) -> None:
        env_object = next((obj for obj in env_objects if obj['name'] == env_var_name), None)
        if not env_object:
            logger.info(f"Adding new env variable - {env_var_name}")
            env_objects.append({'name': env_var_name, 'valueFrom': {'secretKeyRef': {'name': secret_name, 'key': object_name}}})
        else:
            logger.warning(f"Env variable already present - {env_var_name}, Modifying secret name")
            env_object['valueFrom'] = {'secretKeyRef': {'name': secret_name, 'key': object_name}}

    def write_secret_to_spec_files(self, env: str, env_var_name: str, secret_name: str, object_name: str, serviceNames: str) -> None:
        services = serviceNames.split(",")
        for service in services:
            service = service.strip()
            file_name = rf'{service}.yaml'
            logger.info(f"Processing spec file - {service}")
            self.write_secret_to_spec_file(env, env_var_name, secret_name, object_name, file_name)

    def write_secret_to_spec_file(self, env: str, env_var_name: str, secret_name: str, object_name: str, file_name: str) -> None:
        self.current_env = env
        yaml: YAML = self.get_yaml()
        if "rh" in file_name and env == "prod":
            yaml.indent(mapping=2, sequence=4, offset=2)
        if not os.path.exists(file_name):
            logger.info(f"File not found - {file_name}")
            return  
        spec = self._load_spec_file(yaml, file_name)
        for entry in spec:
            if entry['kind'] == "Deployment" or (entry['kind'] == "Job" and entry['metadata']['name'] == "rh-init"):
                try:
                    env_objects = entry['spec']['template']['spec']['containers'][0]['env']
                    self._update_env_object(env_objects, env_var_name, secret_name, object_name)
                except KeyError:
                    logger.error(f"Warning: Expected structure not found in {file_name}")
                    return
        logger.info(f"Writing changes to file {file_name} now")
        self._save_spec_file(yaml, file_name, spec)

    def main(self, envs: List[str], services: str, env_var_name: str, secret_name: str = '', object_name: str = '') -> None:
        if not services:
            logger.error("Error: No services provided")
            return
        if not env_var_name:
            logger.error("Error: No environment variable name provided")
            return
        if not secret_name:
            secret_name = env_var_name.strip().lower().replace("_", "-")
        if not object_name:
            object_name = env_var_name.strip().lower().replace("_", "-")
        # Changing the script execution context folder to the script root
        # This will help in making sure that the script can be run from any folder in the shell 
        script_root = Path(__file__).resolve().parent
        with chdir(script_root):
            for env in envs:      
                with chdir(Path('../k8s') / env):
                    logger.info(f"Writing secrets for env - {env}")
                    secret_file: Optional[str] = self.get_secret_file_for_services(services)
                    self.write_secret_to_secret_class(secret_name, object_name, secret_file)
                    self.write_secret_to_spec_files(env, env_var_name, secret_name, object_name, services)


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='Add secrets to the k8s spec files. Uses the ruamel.yaml library. If not already installed, install with command: pip install ruamel.yaml')
    parser.add_argument('--services', required=True, help='Comma separated string of services. e.g. cm, cd, rh')
    parser.add_argument('--env_var_name', required=True, help='The environment variable name')
    parser.add_argument('--secret_name', help='The name of the secret group. The env_var_name will be used if not specified')
    parser.add_argument('--object_name', help='The name of the secret in the Key Vault. The env_var_name will be used if not specified')
    return parser.parse_args()

if __name__ == "__main__":
    args = parse_arguments()
    secret_manager = SecretManager()
    secret_manager.main(secret_manager.envs, args.services, args.env_var_name, args.secret_name, args.object_name)

pip install ruamel.yaml

project-root/
│
├── tools/
│   └── secrets/
│       └── add-secret.py                  # The main script for managing secrets
│
├── k8s/                                   # Kubernetes configuration files│
│   ├── qa/
│   │   ├── kustomization.yaml             # Kustomization file for QA environment
│   │   ├── secret-classes/                # Environment-specific secret class YAML files for QA
│   │   │   ├── kustomization.yaml         # Kustomization for QA secret classes
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml                        # Service-specific YAML files (e.g., cm, cd, rh)
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   ├── uat/
│   │   ├── kustomization.yaml             # Kustomization file for UAT environment
│   │   ├── secret-classes/                # Environment-specific secret class YAML files for UAT
│   │   │   ├── kustomization.yaml         # Kustomization for UAT secret classes
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   ├── stg/
│   │   ├── kustomization.yaml             # Kustomization file for STG environment
│   │   ├── secret-classes/                # Environment-specific secret class YAML files for STG
│   │   │   ├── kustomization.yaml         # Kustomization for STG secret classes
│   │   │   ├── sitecore-secrets.yaml
│   │   │   ├── rh-secrets.yaml
│   │   │   └── common-secrets.yaml
│   │   ├── cm.yaml
│   │   ├── cd.yaml
│   │   └── rh.yaml
│   │
│   └── prod/
│       ├── kustomization.yaml             # Kustomization file for PROD environment
│       ├── secret-classes/                # Environment-specific secret class YAML files for PROD
│       │   ├── kustomization.yaml         # Kustomization for PROD secret classes
│       │   ├── sitecore-secrets.yaml
│       │   ├── rh-secrets.yaml
│       │   └── common-secrets.yaml
│       ├── cm.yaml
│       ├── cd.yaml
│       └── rh.yaml