nginx-configurator/nginx_configurator/main.py

import json
import os
import re
import tempfile
from pathlib import Path
import click
from dotenv import load_dotenv
from . import sysaction, certs
from .sysaction import quit_on_err
from .templating import jinja

load_dotenv(os.getenv("DOTENV_PATH", "/etc/ncc/env"))

NGINX_DIR = Path(os.getenv("NGINX_DIR", "/etc/nginx"))
DOMAINS_TXT = Path(os.getenv("DOMAINS_TXT", "/etc/dehydrated/domains.txt"))
REMOTE = os.getenv("REMOTE")
REMOTE_SSH_KEY = os.getenv("REMOTE_SSH_KEY")
DEHYDRATED_BIN = os.getenv("DEHYDRATED_BIN", "dehydrated")
DEHYDRATED_TRIGGER_FILE = Path(
    os.getenv("DEHYDRATED_TRIGGER_FILE", "/tmp/ncc-ssl-trigger")
)
CLUSTERS_FILE = Path(os.getenv("CLUSTERS_FILE", "/etc/ncc/clusters.json"))


@click.group()
@click.option("--skip-master-check", type=bool, is_flag=True)
def cli(skip_master_check: bool):
    """Update the nginx cluster configuration

    MUST BE RAN ON MASTER (will detect automatically)
    """

    if not skip_master_check and not sysaction.is_keepalived_master():
        click.echo("Refusing to start. Not running on master.", err=True)
        exit(1)


@cli.command()
def reload():
    """Replicate the local config and reload the nginx cluster

    Does the following:
      1. checks the nginx configuration
      2. generates domains.txt for dehydrated
      3. runs dehydrated to obtain certificates
      4. reloads nginx
      5. replicates the validated configuration
    """

    # check nginx config
    quit_on_err(
        sysaction.check_nginx_config(NGINX_DIR),
        print_always=True,
        additional_info="Nginx configuration is incorrect",
    )

    # build domains.txt
    try:
        directives = certs.gather_autossl_directives(NGINX_DIR / "nginx.conf")

        DOMAINS_TXT.write_text(certs.build_domains_txt(directives))
    except ValueError as e:
        click.secho(e, err=True, fg="red")
        exit(1)

    # obtain certs
    quit_on_err(
        sysaction.run_dehydrated(DEHYDRATED_BIN),
        additional_info="Failed to run dehydrated",
    )
    certs.generate_ssl_configs(NGINX_DIR / "ssl", [d[1] for d in directives])

    # reload nginx
    quit_on_err(sysaction.reload_nginx(), additional_info="Failed to reload nginx")

    # replicate to remote
    sysaction.remote_replication(REMOTE, REMOTE_SSH_KEY)
    quit_on_err(
        sysaction.remote_check_nginx_config(REMOTE, REMOTE_SSH_KEY, NGINX_DIR),
        additional_info="Remote nginx configuration is incorrect",
    )
    quit_on_err(
        sysaction.remote_reload_nginx(REMOTE, REMOTE_SSH_KEY),
        additional_info="Remote nginx failed to reload",
    )


def get_upstreams():
    clusters = json.loads(Path(CLUSTERS_FILE).read_text())
    click.echo("Existing clusters:")
    line = ""
    for cluster in clusters:
        if len(line + " " + cluster["name"]) > 78:
            click.echo("  " + line.strip())
            line = cluster["name"]
        else:
            line += " " + cluster["name"]
    click.echo("  " + line.strip())

    upstreams = click.prompt("Comma-separated upstreams, or an existing cluster name")

    cluster = next((c for c in clusters if c["name"] == upstreams), None)
    if cluster:
        upstreams = cluster["upstreams"]
    else:
        upstreams = [u.strip() for u in upstreams.split(",")]

    return upstreams


def get_id():
    return max(int(n.name.split("-")[0]) for n in certs.get_site_files(NGINX_DIR)) + 1


def test_config(config: str, file: Path):
    old_config = None
    if file.exists():
        old_config = file.read_text()

    if config:
        file.write_text(config)
    else:
        file.unlink(missing_ok=True)

    c, out = sysaction.check_nginx_config(NGINX_DIR)
    if c:
        try:
            certs.gather_autossl_directives(NGINX_DIR / "nginx.conf")
        except ValueError as e:
            out = e
            c = False

    # rollback
    if old_config is not None:
        file.write_text(old_config)
    else:
        file.unlink()

    return c, out


def edit_service(config, service_file: Path):
    ok = False

    new_config = None
    while not ok:
        new_config = click.edit(config, extension=".conf")

        if new_config is not None:
            config = new_config

        c, out = test_config(config, service_file)

        if not c:
            click.echo(out, err=True)
            click.secho("Failed to verify configuration", fg="red")

            choice = click.prompt(
                "Edit service configuration?", type=click.Choice(("yes", "abort"))
            )
            if choice == "abort":
                tmp = Path(tempfile.mktemp())
                tmp.write_text(config)
                click.echo(f"Unfinished service written to {tmp}")
                break

        ok = c

    return new_config, ok


@cli.command()
@click.pass_context
def new(ctx):
    """Create a new service"""
    config = ""
    should_open_editor = False
    id = get_id()

    if click.confirm("Would you like to use the default template?", default=True):
        template = jinja.get_template("nginx-site.conf")

        domains = [
            d.strip() for d in click.prompt("Comma-separated domains").split(",")
        ]
        upstreams = get_upstreams()
        port = click.prompt("Upstream port", type=int)
        proto = (
            click.prompt(
                "Upstream protocol",
                type=click.Choice(("http", "https")),
                show_choices=True,
            )
            + "://"
        )

        config = template.render(
            id=id, domains=domains, upstreams=upstreams, port=port, proto=proto
        )
        # create ssl file so configuration test passes
        Path(f"/etc/nginx/ssl/{id}.conf").touch()

        filename = f"{id}-{domains[0]}.conf"
    else:
        should_open_editor = True  # force open the config
        template = jinja.get_template("nginx-minimal.conf")
        config = template.render(id=id)
        filename = f'{id}-{click.prompt("Service name (used as filename (eg. 01-service.conf))")}.conf'

    # XXX: beware of weird code below

    should_open_editor = should_open_editor or click.confirm(
        "Would you like to edit the config?", default=True
    )

    # assume the user wants to edit the file if we are opening the editor
    service_file = Path(
        NGINX_DIR / "sites" / ("custom" if should_open_editor else "auto") / filename
    )

    ok = False
    while not ok:
        if should_open_editor:
            new_cfg, ok = edit_service(config, service_file)

            if not ok:
                break
            if new_cfg:
                config = new_cfg
            else:
                # correct previous assumption
                service_file = NGINX_DIR / "sites/auto" / filename
        else:
            ok, out = test_config(config, service_file)
            if not ok:
                click.echo(out, err=True)
                click.secho("Failed to verify nginx configuration", fg="red")
                choice = click.prompt(
                    "Edit service configuration?", type=click.Choice(("yes", "abort"))
                )
                if choice == "abort":
                    break
                should_open_editor = True
                service_file = NGINX_DIR / "sites/custom" / filename

    if not ok:
        Path(f"/etc/nginx/ssl/{id}.conf").unlink(missing_ok=True)
        exit(1)

    service_file.write_text(config)

    ctx.invoke(reload)


def shell_complete_service(ctx, param, incomplete):
    return [file.name for file in certs.get_site_files(NGINX_DIR) if incomplete in file.name]

@cli.command()
@click.argument("service", shell_complete=shell_complete_service)
@click.pass_context
def edit(ctx, service: str):
    """Edit a service"""
    filename = service if service.endswith(".conf") else service + ".conf"
    auto = True

    file = NGINX_DIR / "sites/auto" / filename
    if not file.exists():
        auto = False
        file = NGINX_DIR / "sites/custom" / filename

    if not file.exists():
        click.secho(f"Service {filename} does not exist", fg="red")
        exit(1)

    config = file.read_text()
    new_cfg, ok = edit_service(config, file)

    if not new_cfg:
        exit(0)

    if not ok:
        exit(1)

    if auto:
        file = file.rename(NGINX_DIR / "sites/custom" / filename)

    file.write_text(new_cfg)

    ctx.invoke(reload)


@cli.command()
@click.argument("service", shell_complete=shell_complete_service)
@click.pass_context
def delete(ctx, service: str):
    """Delete a service"""
    filename = service if service.endswith(".conf") else service + ".conf"

    file = NGINX_DIR / "sites/auto" / filename
    if not file.exists():
        file = NGINX_DIR / "sites/custom" / filename

    if not file.exists():
        click.secho(f"Service {filename} does not exist", fg="red")
        exit(1)

    c, out = test_config("", file)
    if not c:
        click.echo(out, err=True)
        click.secho("Failed to verify nginx configuration", fg="red")
        click.echo("Service was not deleted")
        exit(1)

    file.unlink()

    ctx.invoke(reload)


@cli.command("list")
def list_():
    """List exsiting services and domain names associated with them"""
    files = certs.get_site_files(NGINX_DIR)
    for file in files:
        config = file.read_text()
        domain_names = set()
        for directive in re.finditer(r"(?:^|\n\s*|[{;]\s*)server_name (.*);", config):
            for domain in directive.group(1).split():
                domain_names.add(domain)

        click.echo(f"{file.name}: {' '.join(domain_names)}")


@cli.command()
def autossl():
    """Renew SSL certificates and replicate changes"""
    # build domains.txt
    try:
        directives = certs.gather_autossl_directives(NGINX_DIR / "nginx.conf")

        DOMAINS_TXT.write_text(certs.build_domains_txt(directives))
    except ValueError as e:
        click.secho(e, err=True, fg="red")
        exit(1)

    # obtain certs
    quit_on_err(
        sysaction.run_dehydrated(DEHYDRATED_BIN),
        additional_info="Failed to run dehydrated",
    )

    if DEHYDRATED_TRIGGER_FILE.exists():
        click.echo("Certificates changed - reloading cluster")
        certs.generate_ssl_configs(NGINX_DIR / "ssl", [d[1] for d in directives])

        # reload nginx
        quit_on_err(sysaction.reload_nginx(), additional_info="Failed to reload nginx")

        # replicate to remote
        sysaction.remote_replication(REMOTE, REMOTE_SSH_KEY)
        quit_on_err(
            sysaction.remote_check_nginx_config(REMOTE, REMOTE_SSH_KEY, NGINX_DIR),
            additional_info="Remote nginx configuration is incorrect",
        )
        quit_on_err(
            sysaction.remote_reload_nginx(REMOTE, REMOTE_SSH_KEY),
            additional_info="Remote nginx failed to reload",
        )