Personal Infrastructure Part 4: Creating and Storing Initial Secrets

import argparse
import random
import string
import yaml
import sys

def generate_secret(length=64):
    """Generate a random alphanumeric secret of specified length."""
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for _ in range(length))

def create_secrets_yaml(identifiers, existing_secrets=None):
    """Create a dictionary of secrets for given identifiers."""
    secrets = {}
    for identifier in identifiers:
        if existing_secrets and identifier in existing_secrets:
            secrets[identifier] = existing_secrets[identifier]
        else:
            secrets[identifier] = generate_secret()
    return yaml.dump(secrets, default_flow_style=False)

def regenerate_secrets(secrets, identifiers_to_regenerate):
    """Regenerate secrets for specified identifiers."""
    for identifier in identifiers_to_regenerate:
        if identifier in secrets:
            secrets[identifier] = generate_secret()
    return secrets

def get_changed_names(old_secrets, new_secrets):
    """Get the list of added, removed, and changed secret names."""
    added = set(new_secrets.keys()) - set(old_secrets.keys())
    removed = set(old_secrets.keys()) - set(new_secrets.keys())
    changed = {k for k in old_secrets.keys() & new_secrets.keys() if old_secrets[k] != new_secrets[k]}
    return added, removed, changed

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Generate, edit, add, or remove secrets for given identifiers.")
    parser.add_argument('identifiers', nargs='*', help='List of identifiers to generate secrets for')
    parser.add_argument('--regenerate', nargs='*', help='List of identifiers to regenerate secrets for')
    parser.add_argument('--changed-names', action='store_true', help='Output what keys were added, removed or changed')
    args = parser.parse_args()

    # Read existing secrets from stdin if provided
    existing_secrets = {}
    if not sys.stdin.isatty():
        existing_secrets = yaml.safe_load(sys.stdin)

    old_secrets = existing_secrets.copy()

    # Regenerate specified secrets
    if args.regenerate:
        existing_secrets = regenerate_secrets(existing_secrets, args.regenerate)

    # Generate new secrets
    new_secrets = yaml.safe_load(create_secrets_yaml(args.identifiers, existing_secrets))

    # Output changed names if requested
    if args.changed_names:
        added, removed, changed = get_changed_names(old_secrets, new_secrets)
        for name in added:
            print(f"+ {name}")
        for name in removed:
            print(f"- {name}")
        for name in changed:
            print(f"~ {name}")
    else:
        # Output the YAML
        print(yaml.dump(new_secrets, default_flow_style=False))

---
- name: Check if secrets file already exists
  stat:
    path: "{{ playbook_dir }}/secrets/secrets.yml"
  register: secrets_file

- name: Ensure secrets directory exists
  file:
    path: "{{ playbook_dir }}/secrets"
    state: directory
    mode: '0700'

- name: Generate or update secrets
  block:
    - name: View existing secrets
      ansible.builtin.shell: ansible-vault view {{ playbook_dir }}/secrets/secrets.yml
      register: existing_secrets
      when: secrets_file.stat.exists

    - name: Check for changes in secrets
      ansible.builtin.shell: >
        {% if secrets_file.stat.exists %}
        echo "{{ existing_secrets.stdout }}" | 
        {% endif %}
        python3 {{ role_path }}/files/manage_secrets.py {{ initial_secrets | join(' ') }} --changed-names
      register: secrets_changes

    - name: Generate or update secrets using manage_secrets.py
      ansible.builtin.shell: >
        {% if secrets_file.stat.exists %}
        echo "{{ existing_secrets.stdout }}" | 
        {% endif %}
        python3 {{ role_path }}/files/manage_secrets.py {{ initial_secrets | join(' ') }}
      register: secrets_output
      when: secrets_changes.stdout != ""

    - name: Create temporary secrets file
      copy:
        content: "{{ secrets_output.stdout }}"
        dest: "{{ playbook_dir }}/secrets/temp_secrets.yml"
        mode: '0600'
      when: secrets_changes.stdout != ""

    - name: Encrypt secrets file with Ansible Vault
      command: >
        ansible-vault encrypt 
        {{ playbook_dir }}/secrets/temp_secrets.yml
      register: encrypt_result
      when: secrets_changes.stdout != ""

    - name: Rename encrypted secrets file
      command: mv {{ playbook_dir }}/secrets/temp_secrets.yml {{ playbook_dir }}/secrets/secrets.yml
      when: secrets_changes.stdout != "" and encrypt_result.rc == 0

    - name: Clean up temporary files
      file:
        path: "{{ playbook_dir }}/secrets/temp_secrets.yml"
        state: absent
      when: secrets_changes.stdout != "" and encrypt_result.rc != 0

    - name: Display success message
      debug:
        msg: "Secrets file {% if secrets_file.stat.exists %}updated{% else %}created{% endif %} and encrypted successfully at {{ playbook_dir }}/secrets/secrets.yml"
      when: secrets_changes.stdout != ""

    - name: Display no changes message
      debug:
        msg: "No changes detected in secrets file. Skipping re-encryption."
      when: secrets_changes.stdout == ""

- name: Web Servers
  hosts: web:!localhost
  tags: [remote, staging]
  tasks:
    - name: Include the encrypted secrets file as variables
      ansible.builtin.include_vars:
        file: "inventories/{{ env }}/group_vars/secrets.yaml"
      tags: [always]

    - name: Setup Web servers
      ansible.builtin.include_role:
        name: deploy_site