#!/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: 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={{{{ now() }}}} networks: - proxy healthcheck: test: ["CMD-SHELL", "curl", "--fail", "http://localhost:8000"] timeout: 5s interval: 30s retries: 2 start_period: 8s deploy: mode: replicated update_config: parallelism: 1 failure_action: rollback order: start-first delay: 15s monitor: 10s 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()