id_configurator/id_configurator/cli.py

"""Typer CLI for id-configurator."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Annotated

import typer

from id_configurator import modules  # noqa: F401  (registers the built-in modules)
from id_configurator.config import load_configuration
from id_configurator.context import Context
from id_configurator.modules.base import all_modules

app = typer.Typer(
    name="id-configurator",
    help="Configure a device in an air-gapped environment.",
    no_args_is_help=True,
)

logger = logging.getLogger("id_configurator")

ConfigFileOption = Annotated[
    Path | None,
    typer.Option(
        "--config",
        exists=True,
        dir_okay=False,
        help="Path to a JSON or YAML configuration file.",
    ),
]
ConfigEnvOption = Annotated[
    str | None,
    typer.Option(
        "--config-env",
        envvar="ID_CONFIGURATOR_CONFIG",
        help="Base64-encoded JSON/YAML configuration.",
    ),
]
ConfigUrlOption = Annotated[
    str | None,
    typer.Option(
        "--config-url",
        help="URL to download a JSON or YAML configuration from.",
    ),
]
ConfigUrlCacertContentOption = Annotated[
    str | None,
    typer.Option(
        "--config-url-cacert-content",
        help=(
            "PEM content of the CA certificate to trust when downloading "
            "--config-url over HTTPS (defaults to the system's trust store)."
        ),
    ),
]
ConfigUrlInsecureOption = Annotated[
    bool,
    typer.Option(
        "--config-url-insecure",
        help=(
            "Disable TLS certificate verification when downloading --config-url "
            "(INSECURE: only for bootstrapping from a server whose certificate "
            "isn't trustable yet; overrides --config-url-cacert-content)."
        ),
    ),
]
SrcOption = Annotated[
    Path | None,
    typer.Option(
        "--src",
        exists=True,
        file_okay=False,
        help="Existing filesystem used to detect the current configuration.",
    ),
]
DstOption = Annotated[
    Path,
    typer.Option(
        "--dst",
        file_okay=False,
        help="Destination directory receiving the modified files.",
    ),
]
UserOption = Annotated[
    list[str],
    typer.Option(
        "--user",
        help="A user to prepare per-user configuration for (may be repeated).",
    ),
]
AutoUsersOption = Annotated[
    bool,
    typer.Option(
        "--auto-users",
        help=(
            "Also prepare per-user configuration for every regular "
            "(non-system) user found in <src>/etc/passwd: one with a login "
            "shell (not nologin, sync or false) and a UID at or above "
            "UID_MIN (from <src>/etc/login.defs, 1000 by default) - root is "
            "always included regardless of its UID."
        ),
    ),
]
VerboseOption = Annotated[
    bool, typer.Option("--verbose", "-v", help="Enable verbose (debug) logging.")
]


def _configure_logging(verbose: bool) -> None:
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format="%(levelname)s %(name)s: %(message)s",
    )


def _load_configuration(
    config: Path | None,
    config_env: str | None,
    config_url: str | None,
    config_url_cacert_content: str | None,
    config_url_insecure: bool = False,
):
    if config is None and config_env is None and config_url is None:
        raise typer.BadParameter(
            "either --config, --config-env (or $ID_CONFIGURATOR_CONFIG) or "
            "--config-url must be provided"
        )
    return load_configuration(
        config, config_env, config_url, config_url_cacert_content, config_url_insecure
    )


@app.command()
def apply(
    dst: DstOption,
    config: ConfigFileOption = None,
    config_env: ConfigEnvOption = None,
    config_url: ConfigUrlOption = None,
    config_url_cacert_content: ConfigUrlCacertContentOption = None,
    config_url_insecure: ConfigUrlInsecureOption = False,
    src: SrcOption = None,
    user: UserOption = [],  # noqa: B006  (Typer needs a concrete default to detect --user as repeatable)
    auto_users: AutoUsersOption = False,
    verbose: VerboseOption = False,
) -> None:
    """Apply a configuration to the destination directory."""
    _configure_logging(verbose)
    configuration = _load_configuration(
        config, config_env, config_url, config_url_cacert_content, config_url_insecure
    )
    dst.mkdir(parents=True, exist_ok=True)
    context = Context(
        config=configuration, dst=dst, src=src, users=user, auto_users=auto_users
    )
    for module_cls in all_modules():
        module = module_cls()
        if module.is_relevant(context):
            logger.info("applying module %s", module.name)
            module.apply(context)
        else:
            logger.debug("skipping module %s (not relevant)", module.name)


@app.command()
def validate(
    config: ConfigFileOption = None,
    config_env: ConfigEnvOption = None,
    config_url: ConfigUrlOption = None,
    config_url_cacert_content: ConfigUrlCacertContentOption = None,
    config_url_insecure: ConfigUrlInsecureOption = False,
) -> None:
    """Validate a configuration file without applying it."""
    configuration = _load_configuration(
        config, config_env, config_url, config_url_cacert_content, config_url_insecure
    )
    uuid = configuration.uuid if configuration.uuid is not None else "<no uuid>"
    organisation = configuration.organisation or "<no organisation>"
    typer.echo(f"configuration {uuid} is valid (organisation: {organisation})")


@app.command("list-modules")
def list_modules_command() -> None:
    """List the available configuration modules."""
    for module_cls in all_modules():
        typer.echo(module_cls.name)


def main(name: str = "__main__") -> None:
    if name == "__main__":
        app()