path: root/create.py

#!/usr/bin/env python3

import argparse
import logging
import subprocess
import requests
import getpass
import textwrap
from dataclasses import dataclass
from typing import Dict, Optional
from abc import ABC, abstractmethod
from pathlib import Path
from functools import cache

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


class Config:
    SECRETS = Path("secrets.enc")
    ROOT_DOMAIN = "liz.coffee"
    CLOUDFLARE = "https://api.cloudflare.com/client/v4"

    INVENTORY = Path("inventory")
    ANSIBLE_DEPLOY = Path("deploy.yml")
    ANSIBLE_PLAYBOOKS = Path("playbooks/")
    ANSIBLE_ROLES = ANSIBLE_PLAYBOOKS / Path("roles/")
    GROUP_VARS = Path("group_vars/")
    NGINX_SITES_ENABLED = ANSIBLE_ROLES / Path("outbound/templates/proxy/nginx/conf.d")

    LABDNS_GROUP_VARS = GROUP_VARS / Path("labdns.yml")
    LABDNS_DELIMITER = "internal_services:" + "\n"
    EXTERNAL_LOADBALANCER_HOST = "outbound.liz.coffee"
    LOADBALANCER_INVENTORY_LINE = textwrap.dedent(
        """\
        swarm-one ansible_host=10.128.0.201 ansible_user=serve \
        ansible_connection=ssh ansible_become_password='{{ swarm_become_password }}'
    """
    ).strip()

    @staticmethod
    @cache
    def read_secrets() -> Dict[str, str]:
        logger.info("Reading secrets...")
        result = subprocess.run(
            ["ansible-vault", "view", str(Config.SECRETS)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        secrets = result.stdout.decode().splitlines()
        logger.info("Finished reading secrets.")
        return dict(line.split(": ", 1) for line in secrets if ": " in line)

    @staticmethod
    def from_secrets(name: str) -> str:
        return Config.read_secrets().get(name, "")


@dataclass
class CloudflareDns:
    api_token: str
    zone_id: str
    api_url: str

    def put_cname(self, of: str, to: str) -> bool:
        url = f"{self.api_url}/zones/{self.zone_id}/dns_records"
        headers = {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json",
        }
        data = {"type": "CNAME", "name": of, "content": to, "ttl": 3600}
        logger.info(f"Putting Cloudflare CNAME {of} -> {to}")
        response = requests.post(url, headers=headers, json=data)
        if response.ok:
            logger.info("Cloudflare DNS update successful.")
            return True
        logger.error(f"Cloudflare DNS update failed: {response.text}")
        return False


@dataclass
class HomelabDns:
    group_vars: Path
    delimiter: str

    def put_internal(self, dns_prefix: str) -> bool:
        new_lines = []
        with self.group_vars.open("r") as f:
            lines = f.readlines()
            try:
                internal_services = lines.index(self.delimiter)
            except ValueError:
                logger.error(f"Cannot find delimiter {self.delimiter} in {self.group_vars}")
                return False
            new_lines = (
                lines[0:internal_services + 1]
                + [f"  - {dns_prefix}\n"]
                + lines[internal_services + 1:]
            )
        if not new_lines:
            return False
        with self.group_vars.open("w") as f:
            f.write("".join(new_lines))
        return True


class RoleGenerator:
    def __init__(self, service_name: str, image: str, port: str):
        self.service = service_name
        self.image = image
        self.port = port
        self.role_path = Config.ANSIBLE_ROLES / self.service
        self.templates_path = self.role_path / "templates"
        self.tasks_path = self.role_path / "tasks"

    def create_inventory(self):
        with open(Config.INVENTORY, "a") as f:
            f.write(f"[{self.service}]\n")
            f.write(Config.LOADBALANCER_INVENTORY_LINE + "\n\n")

    def create_tasks(self):
        self.tasks_path.mkdir(parents=True, exist_ok=True)
        task_file = self.tasks_path / "main.yml"
        task_file.write_text(
            textwrap.dedent(
                f"""\
            ---

            - name: Deploy {self.service}
              ansible.builtin.import_tasks: manage-docker-swarm-service.yml
              vars:
                service_name: {self.service}
                template_render_dir: "../templates"
                service_destination_dir: "{{{{ {self.service}_base }}}}"
        """
            )
        )

    def create_compose_template(self):
        (self.templates_path / "stacks").mkdir(parents=True, exist_ok=True)
        compose_file = self.templates_path / "stacks" / "docker-compose.yml"
        compose_file.write_text(
            textwrap.dedent(
                f"""\
            services:
              {self.service}:
                image: {self.image}
                volumes:
                  - {{{{ {self.service}_base }}}}/volumes/data:/data
                environment:
                  - TZ={{{{ timezone }}}}
                  - DEPLOYMENT_TIME={{{{ deployment_time }}}}
                networks:
                  - proxy
                healthcheck:
                  test: ["CMD-SHELL", "curl", "--fail", "http://localhost:8000"]
                  timeout: 15s
                  interval: 30s
                  retries: 3
                  start_period: 5s
                deploy:
                  mode: replicated
                  update_config:
                    parallelism: 1
                    failure_action: rollback
                    order: start-first
                    delay: 5s
                    monitor: 30s
                  replicas: 1
                  labels:
                    - traefik.enable=true
                    - traefik.swarm.network=proxy
                    - traefik.http.routers.{self.service}.tls=true
                    - traefik.http.routers.{self.service}.tls.certResolver=letsencrypt
                    - traefik.http.routers.{self.service}.rule=Host(`{{{{ {self.service}_domain }}}}`)
                    - traefik.http.routers.{self.service}.entrypoints=websecure
                    - traefik.http.services.{self.service}.loadbalancer.server.port={self.port}

            networks:
              proxy:
                external: true
        """
            )
        )

    def create_group_vars(self):
        path = Config.GROUP_VARS / f"{self.service}.yml"
        path.write_text(
            textwrap.dedent(
                f"""\
            ---

            {self.service}_domain: {self.service}.{Config.ROOT_DOMAIN}
            {self.service}_base: "{{{{ swarm_base }}}}/{self.service}"
        """
            )
        )

    def create_volumes(self):
        (self.templates_path / "volumes" / "data").mkdir(parents=True, exist_ok=True)
        (self.templates_path / "volumes" / "data" / ".gitkeep").touch()

    def create_deploy_hook(self):
        path = Config.ANSIBLE_PLAYBOOKS / f"{self.service}.yml"
        path.write_text(
            textwrap.dedent(
                f"""\
          ---

          - name: {self.service} setup
            hosts: {self.service}
            become: true
            roles:
              - {self.service}
        """
            )
        )
        with open(Config.ANSIBLE_DEPLOY, "a") as f:
            f.write("\n")
            f.write(f"- name: {self.service}\n")
            f.write(
                f"  ansible.builtin.import_playbook: playbooks/{self.service}.yml\n"
            )

    def create_all(self):
        self.create_inventory()
        self.create_tasks()
        self.create_volumes()
        self.create_compose_template()
        self.create_group_vars()
        self.create_deploy_hook()


def create_nginx_conf(service_name: str):
    path = Config.NGINX_SITES_ENABLED / f"{service_name}.conf"
    path.write_text(
        textwrap.dedent(
            f"""\
        server {{
            listen 80;
            server_name {service_name}.liz.coffee;

            real_ip_header X-Forwarded-For;
            real_ip_recursive on;
            set_real_ip_from {{{{ docker_network }}}};

            location / {{
                proxy_pass https://{{{{ loadbalancer_ip }}}};
                proxy_ssl_verify off;
                proxy_http_version 1.1;
                proxy_set_header Host $host;
                proxy_set_header X-Real-IP $remote_addr;
                proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
                proxy_set_header Upgrade $http_upgrade;
                proxy_set_header Connection "upgrade";
            }}
        }}
    """
        )
    )


def main():
    parser = argparse.ArgumentParser(description="Service initializer for liz.coffee.")
    parser.add_argument("--service-name", type=str, required=True)
    parser.add_argument("--container-image", type=str, required=True)
    parser.add_argument("--service-port", type=str, default="80")
    parser.add_argument("--external", action="store_true")
    parser.add_argument("--internal", action="store_true")
    args = parser.parse_args()

    logger.info(f"Initializing service setup for '{args.service_name}'")
    logger.debug(f"Arguments received: {args}")

    service_fqdn = f"{args.service_name}.{Config.ROOT_DOMAIN}"
    logger.info(f"Service FQDN resolved as {service_fqdn}")

    if args.external:
        logger.info("Configuring external DNS via Cloudflare...")
        cf = CloudflareDns(
            api_token=Config.from_secrets("cloudflare_dns_api_token"),
            zone_id=Config.from_secrets("cloudflare_zone_id"),
            api_url=Config.CLOUDFLARE,
        )
        success = cf.put_cname(args.service_name, Config.EXTERNAL_LOADBALANCER_HOST)
        if not success:
            logger.warning("External DNS (Cloudflare) CNAME creation failed.")
        else:
            logger.info("External DNS (Cloudflare) CNAME created successfully.")

    if args.internal:
        logger.info("Configuring internal DNS via LabDNS...")
        dns = HomelabDns(
            group_vars=Config.LABDNS_GROUP_VARS, delimiter=Config.LABDNS_DELIMITER
        )
        success = dns.put_internal(args.service_name)
        if not success:
            logger.warning("Internal DNS (LabDNS) Record creation failed.")
        else:
            logger.info("Internal DNS (LabDNS) Record created successfully.")

    logger.info("Generating Ansible role and configuration...")
    generator = RoleGenerator(
        args.service_name, args.container_image, args.service_port
    )
    generator.create_all()
    logger.info("Role generation complete.")

    if args.external:
        logger.info("Creating NGINX configuration for external routing...")
        create_nginx_conf(args.service_name)
        logger.info("NGINX config created.")

    logger.info(f"Service '{args.service_name}' setup complete.")


if __name__ == "__main__":
    main()