feat: add LXC container support for local chatmail development

Add cmdeploy "lxc-test" command to run cmdeploy against local containers,
with supplementary lxc-start, lxc-stop and lxc-status subcommands.
See doc/source/lxc.rst for full documentation including prerequisites,
DNS setup, TLS handling, DNS-free testing, and known limitations.

Apart from adding lxc-specific docs, tests, and implementation files in the cmdeploy/lxc directory,
this PR adds the --ssh-config option to cmdeploy run/dns/status/test commands and pyinfra invocations,
and also to sshexec (Execnet) handling.  This allows for the host to need no DNS entries for a relay,
and route all resolution through ssh-config.  This is used by the "lxc-test" command, which performs
a completely local setup -- again, see docs for more details.

While working on DNS/SSH things i also unified all zone-file handling
to use actual BIND format as it is easy enough to parse back.

Author: hpk42

.gitignore

@@ -5,6 +5,7 @@ __pycache__/
 *.swp
 *qr-*.png
 chatmail*.ini
+lxconfigs/
 
 
 # C extensions

cmdeploy/src/cmdeploy/chatmail.zone.j2

@@ -18,7 +18,23 @@
 from termcolor import colored
 
 from . import dns, remote
-from .sshexec import LocalExec, SSHExec
+from .lxc.cli import (  # noqa: F401
+    lxc_start_cmd,
+    lxc_start_cmd_options,
+    lxc_status_cmd,
+    lxc_status_cmd_options,
+    lxc_stop_cmd,
+    lxc_stop_cmd_options,
+    lxc_test_cmd,
+    lxc_test_cmd_options,
+)
+from .sshexec import (
+    LocalExec,
+    SSHExec,
+    resolve_host_from_ssh_config,
+    resolve_key_from_ssh_config,
+)
+from .www import main as webdev_main
 
 #
 # cmdeploy sub commands and options
@@ -82,13 +98,14 @@ def run_cmd_options(parser):
         help="disable checks nslookup for dns",
     )
     add_ssh_host_option(parser)
+    add_ssh_config_option(parser)
 
 
 def run_cmd(args, out):
     """Deploy chatmail services on the remote server."""
 
     ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
-    sshexec = get_sshexec(ssh_host)
+    sshexec = get_sshexec(ssh_host, ssh_config=args.ssh_config)
     require_iroh = args.config.enable_iroh_relay
     strict_tls = args.config.tls_cert_mode == "acme"
     if not args.dns_check_disabled:
@@ -108,6 +125,18 @@ def run_cmd(args, out):
     pyinf = "pyinfra --dry" if args.dry_run else "pyinfra"
 
     cmd = f"{pyinf} --ssh-user root {ssh_host} {deploy_path} -y"
+    ssh_config = args.ssh_config
+    if ssh_config:
+        ssh_config = str(Path(ssh_config).resolve())
+
+        # Use pyinfra's native SSH data keys to configure the connection directly
+        # rather than relying on paramiko config parsing (see also sshexec.py)
+        ip = resolve_host_from_ssh_config(ssh_host, ssh_config)
+        key = resolve_key_from_ssh_config(ssh_host, ssh_config)
+        data_args = f"--data ssh_hostname={ip} --data ssh_known_hosts_file=/dev/null"
+        if key:
+            data_args += f" --data ssh_key={key}"
+        cmd = f"{pyinf} --ssh-user root {ssh_host} {deploy_path} -y {data_args}"
     if ssh_host in ["localhost", "@docker"]:
         if ssh_host == "@docker":
             env["CHATMAIL_NOPORTCHECK"] = "True"
@@ -139,15 +168,16 @@ def dns_cmd_options(parser):
         dest="zonefile",
         type=pathlib.Path,
         default=None,
-        help="write out a zonefile",
+        help="write DNS records in standard BIND format to the given file",
     )
     add_ssh_host_option(parser)
+    add_ssh_config_option(parser)
 
 
 def dns_cmd(args, out):
     """Check DNS entries and optionally generate dns zone file."""
     ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
-    sshexec = get_sshexec(ssh_host, verbose=args.verbose)
+    sshexec = get_sshexec(ssh_host, verbose=args.verbose, ssh_config=args.ssh_config)
     tls_cert_mode = args.config.tls_cert_mode
     strict_tls = tls_cert_mode == "acme"
     remote_data = dns.get_initial_remote_data(sshexec, args.config.mail_domain)
@@ -178,13 +208,14 @@ def dns_cmd(args, out):
 
 def status_cmd_options(parser):
     add_ssh_host_option(parser)
+    add_ssh_config_option(parser)
 
 
 def status_cmd(args, out):
     """Display status for online chatmail instance."""
 
     ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
-    sshexec = get_sshexec(ssh_host, verbose=args.verbose)
+    sshexec = get_sshexec(ssh_host, verbose=args.verbose, ssh_config=args.ssh_config)
 
     out.green(f"chatmail domain: {args.config.mail_domain}")
     if args.config.privacy_mail:
@@ -204,14 +235,18 @@ def test_cmd_options(parser):
         help="also run slow tests",
     )
     add_ssh_host_option(parser)
+    add_ssh_config_option(parser)
 
 
 def test_cmd(args, out):
     """Run local and online tests for chatmail deployment."""
 
     env = os.environ.copy()
+    env["CHATMAIL_INI"] = str(args.inipath.resolve())
     if args.ssh_host:
         env["CHATMAIL_SSH"] = args.ssh_host
+    if args.ssh_config:
+        env["CHATMAIL_SSH_CONFIG"] = str(Path(args.ssh_config).resolve())
 
     pytest_path = shutil.which("pytest")
     pytest_args = [
@@ -276,9 +311,7 @@ def bench_cmd(args, out):
 
 def webdev_cmd(args, out):
     """Run local web development loop for static web pages."""
-    from .www import main
-
-    main()
+    webdev_main()
 
 
 #
@@ -321,6 +354,16 @@ def add_ssh_host_option(parser):
     )
 
 
+def add_ssh_config_option(parser):
+    parser.add_argument(
+        "--ssh-config",
+        dest="ssh_config",
+        type=Path,
+        default=None,
+        help="Path to an SSH config file (e.g. lxconfigs/ssh-config).",
+    )
+
+
 def add_config_option(parser):
     parser.add_argument(
         "--config",
@@ -330,6 +373,7 @@ def add_config_option(parser):
         type=Path,
         help="path to the chatmail.ini file",
     )
+
     parser.add_argument(
         "--verbose",
         "-v",
@@ -340,15 +384,16 @@ def add_config_option(parser):
     )
 
 
-def add_subcommand(subparsers, func):
+def add_subcommand(subparsers, func, add_config=True):
     name = func.__name__
     assert name.endswith("_cmd")
-    name = name[:-4]
+    name = name[:-4].replace("_", "-")
     doc = func.__doc__.strip()
     help = doc.split("\n")[0].strip(".")
     p = subparsers.add_parser(name, description=doc, help=help)
     p.set_defaults(func=func)
-    add_config_option(p)
+    if add_config:
+        add_config_option(p)
     return p
 
 
@@ -362,40 +407,43 @@ def get_parser():
     """Return an ArgumentParser for the 'cmdeploy' CLI"""
 
     parser = argparse.ArgumentParser(description=description.strip())
+    parser.set_defaults(func=None, inipath=None)
     subparsers = parser.add_subparsers(title="subcommands")
 
     # find all subcommands in the module namespace
     glob = globals()
     for name, func in glob.items():
         if name.endswith("_cmd"):
-            subparser = add_subcommand(subparsers, func)
+            needs_config = not (name.startswith("lxc_") or name.startswith("fmt_"))
+            subparser = add_subcommand(subparsers, func, add_config=needs_config)
             addopts = glob.get(name + "_options")
             if addopts is not None:
                 addopts(subparser)
 
     return parser
 
 
-def get_sshexec(ssh_host: str, verbose=True):
+def get_sshexec(ssh_host: str, verbose=True, ssh_config=None):
     if ssh_host in ["localhost", "@local"]:
         return LocalExec(verbose, docker=False)
     elif ssh_host == "@docker":
         return LocalExec(verbose, docker=True)
     if verbose:
         print(f"[ssh] login to {ssh_host}")
-    return SSHExec(ssh_host, verbose=verbose)
+    return SSHExec(ssh_host, verbose=verbose, ssh_config=ssh_config)
 
 
 def main(args=None):
     """Provide main entry point for 'cmdeploy' CLI invocation."""
     parser = get_parser()
     args = parser.parse_args(args=args)
-    if not hasattr(args, "func"):
+    if args.func is None:
         return parser.parse_args(["-h"])
 
     out = Out()
     kwargs = {}
-    if args.func.__name__ not in ("init_cmd", "fmt_cmd"):
+
+    if args.inipath is not None and args.func.__name__ != "init_cmd":
         if not args.inipath.exists():
             out.red(f"expecting {args.inipath} to exist, run init first?")
             raise SystemExit(1)

cmdeploy/src/cmdeploy/cmdeploy.py

@@ -18,6 +18,7 @@
 from pyinfra.operations import apt, files, pip, server, systemd
 
 from cmdeploy.cmdeploy import Out
+from cmdeploy.util import get_version_string
 
 from .acmetool import AcmetoolDeployer
 from .basedeploy import (
@@ -271,8 +272,14 @@ def configure(self):
                     logger.warning("Web page build failed, skipping website deployment")
                     return
             # if it is not a hugo page, upload it as is
-            files.rsync(
-                f"{www_path}/", "/var/www/html", flags=["-avz", "--chown=www-data"]
+            # pyinfra files.rsync (experimental) causes problems with ssh-config configuration
+            # the stable files.sync should do
+            files.sync(
+                src=str(www_path),
+                dest="/var/www/html",
+                user="www-data",
+                group="www-data",
+                delete=True,
             )
 
 
@@ -524,17 +531,9 @@ def activate(self):
 
 class GithashDeployer(Deployer):
     def activate(self):
-        try:
-            git_hash = subprocess.check_output(["git", "rev-parse", "HEAD"]).decode()
-        except Exception:
-            git_hash = "unknown\n"
-        try:
-            git_diff = subprocess.check_output(["git", "diff"]).decode()
-        except Exception:
-            git_diff = ""
         files.put(
             name="Upload chatmail relay git commit hash",
-            src=StringIO(git_hash + git_diff),
+            src=StringIO(get_version_string()),
             dest="/etc/chatmail-version",
             mode="700",
         )
@@ -578,11 +577,17 @@ def deploy_chatmail(config_path: Path, disable_mail: bool, website_only: bool) -
         )
 
     # Check if mtail_address interface is available (if configured)
-    if config.mtail_address and config.mtail_address not in ('127.0.0.1', '::1', 'localhost'):
+    if config.mtail_address and config.mtail_address not in (
+        "127.0.0.1",
+        "::1",
+        "localhost",
+    ):
         ipv4_addrs = host.get_fact(hardware.Ipv4Addrs)
         all_addresses = [addr for addrs in ipv4_addrs.values() for addr in addrs]
         if config.mtail_address not in all_addresses:
-            Out().red(f"Deploy failed: mtail_address {config.mtail_address} is not available (VPN up?).\n")
+            Out().red(
+                f"Deploy failed: mtail_address {config.mtail_address} is not available (VPN up?).\n"
+            )
             exit(1)
 
     if not os.environ.get("CHATMAIL_NOPORTCHECK"):

cmdeploy/src/cmdeploy/deployers.py

@@ -1,11 +1,26 @@
 import datetime
-import importlib
-
-from jinja2 import Template
 
 from . import remote
 
 
+def parse_zone_records(text):
+    """Yield ``(name, ttl, rtype, rdata)`` from standard BIND-format text.
+
+    Skips comment lines (starting with ``;``) and blank lines.
+    Each record line must have the format ``name TTL IN type rdata``.
+    """
+    for raw_line in text.strip().splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith(";"):
+            continue
+        parts = line.split(None, 4)
+        if len(parts) < 5:
+            raise ValueError(f"Bad zone record line: {line}")
+        name = parts[0].rstrip(".")
+        # parts[2] is the IN class — ignored
+        yield name, parts[1], parts[3].upper(), parts[4]
+
+
 def get_initial_remote_data(sshexec, mail_domain):
     return sshexec.logged(
         call=remote.rdns.perform_initial_checks, kwargs=dict(mail_domain=mail_domain)
@@ -31,13 +46,36 @@ def get_filled_zone_file(remote_data):
     if not sts_id:
         remote_data["sts_id"] = datetime.datetime.now().strftime("%Y%m%d%H%M")
 
-    template = importlib.resources.files(__package__).joinpath("chatmail.zone.j2")
-    content = template.read_text()
-    zonefile = Template(content).render(**remote_data)
-    lines = [x.strip() for x in zonefile.split("\n") if x.strip()]
+    d = remote_data["mail_domain"]
+    lines = ["; Required DNS entries"]
+    if remote_data.get("A"):
+        lines.append(f"{d}.  3600  IN  A  {remote_data['A']}")
+    if remote_data.get("AAAA"):
+        lines.append(f"{d}.  3600  IN  AAAA  {remote_data['AAAA']}")
+    lines.append(f"{d}.  3600  IN  MX  10 {d}.")
+    if remote_data.get("strict_tls"):
+        lines.append(
+            f'_mta-sts.{d}.  3600  IN  TXT  "v=STSv1; id={remote_data["sts_id"]}"'
+        )
+        lines.append(f"mta-sts.{d}.  3600  IN  CNAME  {d}.")
+    lines.append(f"www.{d}.  3600  IN  CNAME  {d}.")
+    lines.append(remote_data["dkim_entry"])
+    lines.append("")
+    lines.append("; Recommended DNS entries")
+    lines.append(f'{d}.  3600  IN  TXT  "v=spf1 a ~all"')
+    lines.append(f'_dmarc.{d}.  3600  IN  TXT  "v=DMARC1;p=reject;adkim=s;aspf=s"')
+    if remote_data.get("acme_account_url"):
+        lines.append(
+            f'{d}.  3600  IN  CAA  0 issue'
+            f' "letsencrypt.org;accounturi={remote_data["acme_account_url"]}"'
+        )
+    lines.append(f'_adsp._domainkey.{d}.  3600  IN  TXT  "dkim=discardable"')
+    lines.append(f"_submission._tcp.{d}.  3600  IN  SRV  0 1 587 {d}.")
+    lines.append(f"_submissions._tcp.{d}.  3600  IN  SRV  0 1 465 {d}.")
+    lines.append(f"_imap._tcp.{d}.  3600  IN  SRV  0 1 143 {d}.")
+    lines.append(f"_imaps._tcp.{d}.  3600  IN  SRV  0 1 993 {d}.")
     lines.append("")
-    zonefile = "\n".join(lines)
-    return zonefile
+    return "\n".join(lines)
 
 
 def check_full_zone(sshexec, remote_data, out, zonefile) -> int: