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"
    PIHOLE = "https://dns.liz.coffee/api"
    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/sites-enabled")

    INTERNAL_LOADBALANCER_HOST = "floating.home.arpa"
    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, "")


class IDns(ABC):
    @abstractmethod
    def put_cname(self, of: str, to: str) -> bool:
        pass


@dataclass
class CloudflareDns(IDns):
    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 PiholeDns(IDns):
    web_password: str
    api_url: str

    def __authenticate(self) -> Optional[str]:
        response = requests.post(f"{self.api_url}/auth", json={"password": self.web_password})
        return response.json().get("session", {}).get("sid")

    def put_cname(self, of: str, to: str) -> bool:
        sid = self.__authenticate()
        if not sid:
            logger.error("Pi-hole authentication failed.")
            return False
        url = f"{self.api_url}/config/dns/cnameRecords/{of},{to}?sid={sid}"
        response = requests.put(url)
        if response.ok:
            logger.info("Pi-hole DNS update successful.")
            return True
        logger.error(f"Pi-hole DNS update failed: {response.text}")
        return False


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: Build {self.service} compose dirs
              ansible.builtin.file:
                state: directory
                dest: '{{{{ {self.service}_base }}}}/{{{{ item.path }}}}'
                owner: 1000
                group: 1000
                mode: 0755
              with_filetree: '../templates'
              when: item.state == 'directory'

            - name: Build {self.service} compose files
              ansible.builtin.template:
                src: '{{{{ item.src }}}}'
                dest: '{{{{ {self.service}_base }}}}/{{{{ item.path }}}}'
                owner: 1000
                group: 1000
                mode: 0755
              with_filetree: '../templates'
              when: item.state == 'file'

            - name: Deploy {self.service} stack
              ansible.builtin.command:
                cmd: 'docker stack deploy -c {{{{ {self.service}_base }}}}/stacks/docker-compose.yml {self.service}'
        """))

    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 }}}}
                networks:
                  - proxy
                deploy:
                  mode: replicated
                  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;
            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 X-Forwarded-Proto $scheme;
                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 Pi-hole...")
        pi = PiholeDns(web_password=Config.from_secrets("pihole_webpwd"), api_url=Config.PIHOLE)
        success = pi.put_cname(service_fqdn, Config.INTERNAL_LOADBALANCER_HOST)
        if not success:
            logger.warning("Internal DNS (Pi-hole) CNAME creation failed.")
        else:
            logger.info("Internal DNS (Pi-hole) CNAME 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()