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.
This commit is contained in:
holger krekel
2026-03-05 19:31:41 +01:00
parent 4fabfb31f8
commit 861fdf7a50
23 changed files with 2028 additions and 140 deletions

1
.gitignore vendored
View File

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

View File

@@ -1,32 +0,0 @@
;
; Required DNS entries for chatmail servers
;
{% if A %}
{{ mail_domain }}. A {{ A }}
{% endif %}
{% if AAAA %}
{{ mail_domain }}. AAAA {{ AAAA }}
{% endif %}
{{ mail_domain }}. MX 10 {{ mail_domain }}.
{% if strict_tls %}
_mta-sts.{{ mail_domain }}. TXT "v=STSv1; id={{ sts_id }}"
mta-sts.{{ mail_domain }}. CNAME {{ mail_domain }}.
{% endif %}
www.{{ mail_domain }}. CNAME {{ mail_domain }}.
{{ dkim_entry }}
;
; Recommended DNS entries for interoperability and security-hardening
;
{{ mail_domain }}. TXT "v=spf1 a ~all"
_dmarc.{{ mail_domain }}. TXT "v=DMARC1;p=reject;adkim=s;aspf=s"
{% if acme_account_url %}
{{ mail_domain }}. CAA 0 issue "letsencrypt.org;accounturi={{ acme_account_url }}"
{% endif %}
_adsp._domainkey.{{ mail_domain }}. TXT "dkim=discardable"
_submission._tcp.{{ mail_domain }}. SRV 0 1 587 {{ mail_domain }}.
_submissions._tcp.{{ mail_domain }}. SRV 0 1 465 {{ mail_domain }}.
_imap._tcp.{{ mail_domain }}. SRV 0 1 143 {{ mail_domain }}.
_imaps._tcp.{{ mail_domain }}. SRV 0 1 993 {{ mail_domain }}.

View File

@@ -18,7 +18,23 @@ from packaging import version
from termcolor import colored from termcolor import colored
from . import dns, remote 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 # cmdeploy sub commands and options
@@ -82,18 +98,21 @@ def run_cmd_options(parser):
help="disable checks nslookup for dns", help="disable checks nslookup for dns",
) )
add_ssh_host_option(parser) add_ssh_host_option(parser)
add_ssh_config_option(parser)
def run_cmd(args, out): def run_cmd(args, out):
"""Deploy chatmail services on the remote server.""" """Deploy chatmail services on the remote server."""
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain 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 require_iroh = args.config.enable_iroh_relay
strict_tls = args.config.tls_cert_mode == "acme" strict_tls = args.config.tls_cert_mode == "acme"
if not args.dns_check_disabled: if not args.dns_check_disabled:
remote_data = dns.get_initial_remote_data(sshexec, args.config.mail_domain) remote_data = dns.get_initial_remote_data(sshexec, args.config.mail_domain)
if not dns.check_initial_remote_data(remote_data, strict_tls=strict_tls, print=out.red): if not dns.check_initial_remote_data(
remote_data, strict_tls=strict_tls, print=out.red
):
return 1 return 1
env = os.environ.copy() env = os.environ.copy()
@@ -108,6 +127,18 @@ def run_cmd(args, out):
pyinf = "pyinfra --dry" if args.dry_run else "pyinfra" pyinf = "pyinfra --dry" if args.dry_run else "pyinfra"
cmd = f"{pyinf} --ssh-user root {ssh_host} {deploy_path} -y" 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 in ["localhost", "@docker"]:
if ssh_host == "@docker": if ssh_host == "@docker":
env["CHATMAIL_NOPORTCHECK"] = "True" env["CHATMAIL_NOPORTCHECK"] = "True"
@@ -122,7 +153,11 @@ def run_cmd(args, out):
out.check_call(cmd, env=env) out.check_call(cmd, env=env)
if args.website_only: if args.website_only:
out.green("Website deployment completed.") out.green("Website deployment completed.")
elif not args.dns_check_disabled and strict_tls and not remote_data["acme_account_url"]: elif (
not args.dns_check_disabled
and strict_tls
and not remote_data["acme_account_url"]
):
out.red("Deploy completed but letsencrypt not configured") out.red("Deploy completed but letsencrypt not configured")
out.red("Run 'cmdeploy run' again") out.red("Run 'cmdeploy run' again")
else: else:
@@ -139,15 +174,16 @@ def dns_cmd_options(parser):
dest="zonefile", dest="zonefile",
type=pathlib.Path, type=pathlib.Path,
default=None, 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_host_option(parser)
add_ssh_config_option(parser)
def dns_cmd(args, out): def dns_cmd(args, out):
"""Check DNS entries and optionally generate dns zone file.""" """Check DNS entries and optionally generate dns zone file."""
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain 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 tls_cert_mode = args.config.tls_cert_mode
strict_tls = tls_cert_mode == "acme" strict_tls = tls_cert_mode == "acme"
remote_data = dns.get_initial_remote_data(sshexec, args.config.mail_domain) remote_data = dns.get_initial_remote_data(sshexec, args.config.mail_domain)
@@ -178,13 +214,14 @@ def dns_cmd(args, out):
def status_cmd_options(parser): def status_cmd_options(parser):
add_ssh_host_option(parser) add_ssh_host_option(parser)
add_ssh_config_option(parser)
def status_cmd(args, out): def status_cmd(args, out):
"""Display status for online chatmail instance.""" """Display status for online chatmail instance."""
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain 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}") out.green(f"chatmail domain: {args.config.mail_domain}")
if args.config.privacy_mail: if args.config.privacy_mail:
@@ -204,14 +241,18 @@ def test_cmd_options(parser):
help="also run slow tests", help="also run slow tests",
) )
add_ssh_host_option(parser) add_ssh_host_option(parser)
add_ssh_config_option(parser)
def test_cmd(args, out): def test_cmd(args, out):
"""Run local and online tests for chatmail deployment.""" """Run local and online tests for chatmail deployment."""
env = os.environ.copy() env = os.environ.copy()
env["CHATMAIL_INI"] = str(args.inipath.resolve())
if args.ssh_host: if args.ssh_host:
env["CHATMAIL_SSH"] = 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_path = shutil.which("pytest")
pytest_args = [ pytest_args = [
@@ -276,9 +317,7 @@ def bench_cmd(args, out):
def webdev_cmd(args, out): def webdev_cmd(args, out):
"""Run local web development loop for static web pages.""" """Run local web development loop for static web pages."""
from .www import main webdev_main()
main()
# #
@@ -321,6 +360,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): def add_config_option(parser):
parser.add_argument( parser.add_argument(
"--config", "--config",
@@ -330,6 +379,7 @@ def add_config_option(parser):
type=Path, type=Path,
help="path to the chatmail.ini file", help="path to the chatmail.ini file",
) )
parser.add_argument( parser.add_argument(
"--verbose", "--verbose",
"-v", "-v",
@@ -340,15 +390,16 @@ def add_config_option(parser):
) )
def add_subcommand(subparsers, func): def add_subcommand(subparsers, func, add_config=True):
name = func.__name__ name = func.__name__
assert name.endswith("_cmd") assert name.endswith("_cmd")
name = name[:-4] name = name[:-4].replace("_", "-")
doc = func.__doc__.strip() doc = func.__doc__.strip()
help = doc.split("\n")[0].strip(".") help = doc.split("\n")[0].strip(".")
p = subparsers.add_parser(name, description=doc, help=help) p = subparsers.add_parser(name, description=doc, help=help)
p.set_defaults(func=func) p.set_defaults(func=func)
add_config_option(p) if add_config:
add_config_option(p)
return p return p
@@ -362,13 +413,15 @@ def get_parser():
"""Return an ArgumentParser for the 'cmdeploy' CLI""" """Return an ArgumentParser for the 'cmdeploy' CLI"""
parser = argparse.ArgumentParser(description=description.strip()) parser = argparse.ArgumentParser(description=description.strip())
parser.set_defaults(func=None, inipath=None)
subparsers = parser.add_subparsers(title="subcommands") subparsers = parser.add_subparsers(title="subcommands")
# find all subcommands in the module namespace # find all subcommands in the module namespace
glob = globals() glob = globals()
for name, func in glob.items(): for name, func in glob.items():
if name.endswith("_cmd"): if name.endswith("_cmd"):
subparser = add_subcommand(subparsers, func) needs_config = not name.startswith("lxc_")
subparser = add_subcommand(subparsers, func, add_config=needs_config)
addopts = glob.get(name + "_options") addopts = glob.get(name + "_options")
if addopts is not None: if addopts is not None:
addopts(subparser) addopts(subparser)
@@ -376,26 +429,27 @@ def get_parser():
return parser 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"]: if ssh_host in ["localhost", "@local"]:
return LocalExec(verbose, docker=False) return LocalExec(verbose, docker=False)
elif ssh_host == "@docker": elif ssh_host == "@docker":
return LocalExec(verbose, docker=True) return LocalExec(verbose, docker=True)
if verbose: if verbose:
print(f"[ssh] login to {ssh_host}") 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): def main(args=None):
"""Provide main entry point for 'cmdeploy' CLI invocation.""" """Provide main entry point for 'cmdeploy' CLI invocation."""
parser = get_parser() parser = get_parser()
args = parser.parse_args(args=args) args = parser.parse_args(args=args)
if not hasattr(args, "func"): if args.func is None:
return parser.parse_args(["-h"]) return parser.parse_args(["-h"])
out = Out() out = Out()
kwargs = {} kwargs = {}
if args.func.__name__ not in ("init_cmd", "fmt_cmd"):
if args.inipath is not None and args.func.__name__ not in ("init_cmd", "fmt_cmd"):
if not args.inipath.exists(): if not args.inipath.exists():
out.red(f"expecting {args.inipath} to exist, run init first?") out.red(f"expecting {args.inipath} to exist, run init first?")
raise SystemExit(1) raise SystemExit(1)

View File

@@ -18,6 +18,7 @@ from pyinfra.facts.systemd import SystemdEnabled
from pyinfra.operations import apt, files, pip, server, systemd from pyinfra.operations import apt, files, pip, server, systemd
from cmdeploy.cmdeploy import Out from cmdeploy.cmdeploy import Out
from cmdeploy.util import get_version_string
from .acmetool import AcmetoolDeployer from .acmetool import AcmetoolDeployer
from .basedeploy import ( from .basedeploy import (
@@ -271,8 +272,14 @@ class WebsiteDeployer(Deployer):
logger.warning("Web page build failed, skipping website deployment") logger.warning("Web page build failed, skipping website deployment")
return return
# if it is not a hugo page, upload it as is # if it is not a hugo page, upload it as is
files.rsync( # pyinfra files.rsync (experimental) causes problems with ssh-config configuration
f"{www_path}/", "/var/www/html", flags=["-avz", "--chown=www-data"] # the stable files.sync should do
files.sync(
src=str(www_path),
dest="/var/www/html",
user="www-data",
group="www-data",
delete=True,
) )
@@ -532,17 +539,9 @@ class FcgiwrapDeployer(Deployer):
class GithashDeployer(Deployer): class GithashDeployer(Deployer):
def activate(self): 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( files.put(
name="Upload chatmail relay git commit hash", name="Upload chatmail relay git commit hash",
src=StringIO(git_hash + git_diff), src=StringIO(get_version_string()),
dest="/etc/chatmail-version", dest="/etc/chatmail-version",
mode="700", mode="700",
) )
@@ -586,11 +585,17 @@ def deploy_chatmail(config_path: Path, disable_mail: bool, website_only: bool) -
) )
# Check if mtail_address interface is available (if configured) # 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) ipv4_addrs = host.get_fact(hardware.Ipv4Addrs)
all_addresses = [addr for addrs in ipv4_addrs.values() for addr in addrs] all_addresses = [addr for addrs in ipv4_addrs.values() for addr in addrs]
if config.mtail_address not in all_addresses: 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) exit(1)
if not os.environ.get("CHATMAIL_NOPORTCHECK"): if not os.environ.get("CHATMAIL_NOPORTCHECK"):

View File

@@ -1,11 +1,26 @@
import datetime import datetime
import importlib
from jinja2 import Template
from . import remote 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): def get_initial_remote_data(sshexec, mail_domain):
return sshexec.logged( return sshexec.logged(
call=remote.rdns.perform_initial_checks, kwargs=dict(mail_domain=mail_domain) 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: if not sts_id:
remote_data["sts_id"] = datetime.datetime.now().strftime("%Y%m%d%H%M") remote_data["sts_id"] = datetime.datetime.now().strftime("%Y%m%d%H%M")
template = importlib.resources.files(__package__).joinpath("chatmail.zone.j2") d = remote_data["mail_domain"]
content = template.read_text() lines = ["; Required DNS entries"]
zonefile = Template(content).render(**remote_data) if remote_data.get("A"):
lines = [x.strip() for x in zonefile.split("\n") if x.strip()] 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("")
zonefile = "\n".join(lines) lines.append("; Recommended DNS entries")
return zonefile 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("")
return "\n".join(lines)
def check_full_zone(sshexec, remote_data, out, zonefile) -> int: def check_full_zone(sshexec, remote_data, out, zonefile) -> int:

View File

@@ -0,0 +1,469 @@
"""lxc-start/stop/status/test subcommands for testing with local containers."""
import os
import subprocess
import time
from contextlib import contextmanager
from ..util import collapse, get_git_hash, get_version_string, shell
from .incus import Incus, RelayContainer
RELAY_NAMES = ("test0", "test1")
# -------------------------------------------------------------------
# lxc-start
# -------------------------------------------------------------------
def lxc_start_cmd_options(parser):
_add_name_args(
parser,
help_text="User relay name(s) to create (default: test0).",
)
parser.add_argument(
"--ipv4-only",
dest="ipv4_only",
action="store_true",
help="Create an IPv4-only container.",
)
parser.add_argument(
"--run",
action="store_true",
help="Run 'cmdeploy run' on each container after starting it.",
)
def lxc_start_cmd(args, out):
"""Create/Ensure and start LXC relay and DNS containers."""
ix = Incus()
relays = [ix.get_container(n) for n in args.names] or [
ix.get_container(RELAY_NAMES[0])
]
out.green("Ensuring DNS container (ns-localchat) ...")
dns_ct = ix.get_dns_container()
dns_ct.ensure()
dns_ip = dns_ct.ipv4
print(f" DNS container IP: {dns_ip}")
for ct in relays:
out.green(f"Ensuring container {ct.name!r} ({ct.domain}) ...")
ct.ensure()
ip = ct.ipv4
print(" Configuring container hostname ...")
ct.configure_hosts(ip)
print(f" Writing {ct.ini.name} ...")
ct.write_ini(disable_ipv6=args.ipv4_only)
print(f" Config: {ct.ini}")
if args.ipv4_only:
ct.disable_ipv6()
ipv6 = None
else:
output = ct.bash(
"ip -6 addr show scope global -deprecated"
" | grep -oP '(?<=inet6 )[^/]+'",
check=False,
)
ipv6 = output.strip() if output else None
print(f" {_format_addrs(ip, ipv6)}")
out.green(f" Container {ct.name!r} ready: {ct.domain} -> {ip}")
print()
# Reset DNS zones only for the containers we just started
started_cnames = {ct.name for ct in relays}
managed = ix.list_managed()
started = [c for c in managed if c["name"] in started_cnames]
if started:
print(
f"Resetting DNS zones for {len(started)} domain(s) (A + AAAA records) ..."
)
dns_ct.reset_dns_records(dns_ip, started)
for ct in relays:
if ct.name in started_cnames:
print(f" Configuring DNS in {ct.name} ...")
ct.configure_dns(dns_ip)
# Generate the unified SSH config
out.green("Writing ssh-config ...")
ssh_cfg = ix.write_ssh_config()
print(f" {ssh_cfg}")
# Verify SSH via the generated config
for ct in relays:
print(f" Verifying SSH to {ct.name} via ssh-config ...")
if ct.verify_ssh(ssh_cfg):
print(f" SSH OK: ssh -F lxconfigs/ssh-config {ct.domain}")
else:
out.red(f" WARNING: SSH verification failed for {ct.name}")
# Print integration suggestions
ssh_cfg = ix.ssh_config_path
if not ix.check_ssh_include():
out.green(
"\n (Optional) To use containers from any SSH client, add to ~/.ssh/config:"
)
out.green(f" Include {ssh_cfg}")
# Optionally run cmdeploy run on each relay
if args.run:
for ct in relays:
with _section(out, f"cmdeploy run: {ct.sname} ({ct.domain})"):
ret = _run_cmdeploy("run", ct, ix, extra=["--skip-dns-check"])
if ret:
out.red(f"Deploy to {ct.sname} failed (exit {ret})")
return ret
# -------------------------------------------------------------------
# lxc-stop
# -------------------------------------------------------------------
def lxc_stop_cmd_options(parser):
parser.add_argument(
"--destroy",
action="store_true",
help="Delete containers and their config files after stopping.",
)
parser.add_argument(
"--destroy-all",
dest="destroy_all",
action="store_true",
help="Like --destroy, but also remove the ns-localchat DNS container.",
)
_add_name_args(
parser,
help_text="Container name(s) to stop (default: test0 + test1).",
)
def lxc_stop_cmd(args, out):
"""Stop (and optionally destroy) local LXC relay containers."""
ix = Incus()
names = args.names or RELAY_NAMES
destroy = args.destroy or args.destroy_all
for ct in map(ix.get_container, names):
if destroy:
out.green(f"Destroying container {ct.name!r} ...")
ct.destroy()
else:
out.green(f"Stopping container {ct.name!r} ...")
ct.stop(force=True)
if args.destroy_all:
dns_ct = ix.get_dns_container()
out.green(f"Destroying DNS container {dns_ct.name!r} ...")
dns_ct.destroy()
ix.delete_images()
if destroy:
ix.write_ssh_config()
out.green("LXC containers destroyed.")
else:
out.green("LXC containers stopped.")
# -------------------------------------------------------------------
# lxc-test
# -------------------------------------------------------------------
def lxc_test_cmd_options(parser):
parser.add_argument(
"--one",
action="store_true",
help="Only deploy and test against test0 (skip test1).",
)
def lxc_test_cmd(args, out):
"""Run full LXC pipeline: start, deploy, DNS, zone files, and tests.
All commands run directly on the host using
``--ssh-config lxconfigs/ssh-config`` for SSH access.
"""
ix = Incus()
t_total = time.time()
relay_names = list(RELAY_NAMES)
if args.one:
relay_names = relay_names[:1]
local_hash = get_git_hash()
# Per-relay: start, deploy, then snapshot the first relay as a
# reusable image so the second relay launches pre-deployed.
ipv4_only_flags = {RELAY_NAMES[0]: False, RELAY_NAMES[1]: True}
for ct in map(ix.get_container, relay_names):
name = ct.sname
ipv4_only = ipv4_only_flags.get(name, False)
label = "IPv4-only" if ipv4_only else "dual-stack"
with _section(out, f"LXC: lxc-start {name} ({label})"):
args.names = [name]
args.ipv4_only = ipv4_only
args.run = False
ret = lxc_start_cmd(args, out)
if ret:
return ret
status = _deploy_status(ct, local_hash, ix)
if "IN-SYNC" in status:
_section_line(out, f"cmdeploy run: {name}{status}, skipping")
else:
with _section(out, f"cmdeploy run: {name} ({ct.domain})"):
ret = _run_cmdeploy("run", ct, ix, extra=["--skip-dns-check"])
if ret:
out.red(f"Deploy to {name} failed (exit {ret})")
return ret
# Snapshot the first relay so subsequent ones launch pre-deployed
if not ix.find_relay_image():
with _section(out, "LXC: publishing relay image"):
ct.publish_as_relay_image()
for ct in map(ix.get_container, relay_names):
with _section(out, f"cmdeploy dns: {ct.sname} ({ct.domain})"):
ret = _run_cmdeploy("dns", ct, ix, extra=["--zonefile", str(ct.zone)])
if ret:
out.red(f"DNS for {ct.sname} failed (exit {ret})")
return ret
with _section(out, "LXC: PowerDNS zone update"):
dns_ct = ix.get_dns_container()
for ct in map(ix.get_container, relay_names):
if ct.zone.exists():
zone_data = ct.zone.read_text()
print(f" Loading {ct.zone} into PowerDNS ...")
dns_ct.set_dns_records(zone_data)
with _section(out, "cmdeploy test"):
first = ix.get_container(relay_names[0])
env = None
if len(relay_names) > 1:
env = os.environ.copy()
env["CHATMAIL_DOMAIN2"] = ix.get_container(relay_names[1]).domain
ret = _run_cmdeploy("test", first, ix, **({"env": env} if env else {}))
if ret:
out.red(f"Tests failed (exit {ret})")
return ret
elapsed = time.time() - t_total
_section_line(out, f"lxc-test complete ({elapsed:.1f}s)")
return 0
# -------------------------------------------------------------------
# lxc-status
# -------------------------------------------------------------------
def lxc_status_cmd_options(parser):
pass
def lxc_status_cmd(args, out):
"""Show status of local LXC chatmail containers."""
ix = Incus()
containers = ix.list_managed()
if not containers:
out.red("No LXC containers found. Run 'cmdeploy lxc-start' first.")
return 1
local_hash = get_git_hash()
# Get storage pool path for display
storage_path = None
data = ix.run_json(["storage", "show", "default"], check=False)
if data:
storage_path = data.get("config", {}).get("source")
if storage_path:
out.green(f"Containers: ({storage_path})")
else:
out.green("Containers:")
dns_ip = None
for c in containers:
_print_container_status(c, ix, local_hash)
if c["name"] == ix.get_dns_container().name:
dns_ip = c["ip"]
_print_ssh_status(out, ix)
_print_dns_forwarding_status(out, dns_ip)
return 0
def _print_container_status(c, ix, local_hash):
"""Print name/status, domain/IPs, and RAM for one container."""
cname = c["name"]
is_running = c.get("status") == "Running"
ct = ix.get_container(cname)
# First line: name + running/STOPPED + deploy status
if not is_running:
tag = "STOPPED"
elif not isinstance(ct, RelayContainer):
tag = "running"
else:
tag = f"running {_deploy_status(ct, local_hash, ix)}"
print(f" {cname:20s} {tag}")
# Second line: domain, IPv4, IPv6
domain = c.get("domain", "")
ip = c.get("ip") or "?"
ipv6 = c.get("ipv6")
print(f" {domain:20s} {_format_addrs(ip, ipv6)}")
# Third line: RAM (RSS), config
indent = " " * 21
try:
used, total = ct.rss_mib()
except Exception:
ram_str = "RSS ?"
else:
ram_str = f"RSS {used}/{total} MiB ({used * 100 // total}%)"
if isinstance(ct, RelayContainer):
detail = f"{ram_str}, config: {os.path.relpath(ct.ini)}"
else:
detail = ram_str
print(f" {indent}{detail}")
print()
def _print_ssh_status(out, ix):
"""Print SSH integration status."""
print()
ssh_cfg = ix.ssh_config_path
if ix.check_ssh_include():
out.green("SSH: ~/.ssh/config includes lxconfigs/ssh-config ✓")
else:
out.red("SSH: ~/.ssh/config does NOT include lxconfigs/ssh-config")
print(" Add to ~/.ssh/config:")
print(f" Include {ssh_cfg}")
def _print_dns_forwarding_status(out, dns_ip):
"""Print host DNS forwarding status for .localchat."""
if not dns_ip:
out.red("DNS: ns-localchat container not found")
return
try:
rv = shell("resolvectl status incusbr0", timeout=5)
dns_ok = dns_ip in rv.stdout and "localchat" in rv.stdout
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
dns_ok = None
if dns_ok is True:
out.green(f"DNS: .localchat forwarding to {dns_ip}")
elif dns_ok is False:
out.red("DNS: .localchat forwarding NOT configured")
print(" Run:")
print(f" sudo resolvectl dns incusbr0 {dns_ip}")
print(" sudo resolvectl domain incusbr0 ~localchat")
else:
print(" DNS: .localchat forwarding status UNKNOWN")
# -------------------------------------------------------------------
# Internal helpers
# -------------------------------------------------------------------
def _format_addrs(ip, ipv6=None):
parts = [f"IPv4 {ip}"]
if ipv6:
parts.append(f"IPv6 {ipv6}")
return ", ".join(parts)
SECTION_WIDTH = 72
@contextmanager
def _section(out, title):
bar = "\u2501" * (SECTION_WIDTH - len(title) - 5)
out.green(f"\u2501\u2501\u2501 {title} {bar}")
t0 = time.time()
yield
elapsed = time.time() - t0
print(f"{'':>{SECTION_WIDTH - 10}}({elapsed:.1f}s)")
print()
def _section_line(out, title):
bar = "\u2501" * (SECTION_WIDTH - len(title) - 5)
out.green(f"\u2501\u2501\u2501 {title} {bar}")
print()
def _deploy_status(ct, local_hash, ix):
"""Return a human-readable deploy status string.
Compares the full deployed version (hash + diff) against
the local state built by :func:`~cmdeploy.util.get_version_string`.
"""
deployed = ct.deployed_version()
if deployed is None:
return "NOT DEPLOYED"
# A container launched from the relay image has the same
# git hash but a different domain — always redeploy.
deployed_domain = ct.deployed_domain()
if deployed_domain and deployed_domain != ct.domain:
return f"DOMAIN-MISMATCH (deployed: {deployed_domain})"
deployed_lines = deployed.splitlines()
deployed_hash = deployed_lines[0] if deployed_lines else ""
short = deployed_hash[:12]
if not local_hash:
return f"UNKNOWN (deployed: {short})"
local_short = local_hash[:12]
if deployed_hash != local_hash:
return f"STALE (deployed: {short}, local: {local_short})"
# Hash matches — check for uncommitted diffs
local_version = get_version_string()
if deployed != local_version:
return f"DIRTY ({local_short}, undeployed changes)"
return f"IN-SYNC ({short})"
def _add_name_args(parser, help_text=None):
"""Add optional positional NAME arguments."""
parser.add_argument(
"names",
nargs="*",
metavar="NAME",
help=help_text or "Relay name(s) to operate on.",
)
def _run_cmdeploy(subcmd, ct, ix, extra=None, **kwargs):
"""Run ``cmdeploy <subcmd>`` with standard --config/--ssh flags.
*ct* is a Container (uses ``ct.ini`` and ``ct.domain``).
Returns the subprocess exit code.
"""
extra_str = " ".join(extra) if extra else ""
cmd = f"""\
cmdeploy {subcmd}
--config {ct.ini}
--ssh-config {ix.ssh_config_path}
--ssh-host {ct.domain}
{extra_str}
"""
if "cwd" not in kwargs:
kwargs["cwd"] = str(ix.project_root)
cmd = collapse(cmd)
print(f" [$ {cmd}]")
return shell(cmd, capture_output=False, **kwargs).returncode

View File

@@ -0,0 +1,638 @@
"""Core Incus operations for local chatmail LXC containers."""
import json
import subprocess
import textwrap
import time
from pathlib import Path
from ..util import shell
LABEL_KEY = "user.localchat-managed"
SSH_KEY_NAME = "id_localchat"
DOMAIN_SUFFIX = ".localchat"
UPSTREAM_IMAGE = "images:debian/12"
BASE_IMAGE_ALIAS = "localchat-base"
BASE_SETUP_NAME = "localchat-base-setup"
RELAY_IMAGE_ALIAS = "localchat-relay"
DNS_CONTAINER_NAME = "ns-localchat"
DNS_DOMAIN = "ns.localchat"
def _extract_ip(net_data, family="inet"):
"""Extract the first global-scope IP of *family* from network state data.
*net_data* is the ``state.network`` dict from ``incus list --format=json``.
*family* is ``"inet"`` for IPv4 or ``"inet6"`` for IPv6.
Returns the address string, or None.
"""
for iface_name, iface in net_data.items():
if iface_name == "lo":
continue
for addr in iface.get("addresses", []):
if addr["family"] == family and addr["scope"] == "global":
return addr["address"]
return None
class Incus:
"""Gateway for all Incus container operations.
Instantiated once per CLI command and passed around so that
all modules share a single entry point for Incus interactions.
"""
def __init__(self):
self.project_root = Path(__file__).resolve().parent.parent.parent.parent.parent
self.lxconfigs_dir = self.project_root / "lxconfigs"
self.lxconfigs_dir.mkdir(exist_ok=True)
self.ssh_key_path = self.lxconfigs_dir / SSH_KEY_NAME
if not self.ssh_key_path.exists():
shell(
f"ssh-keygen -t ed25519 -f {self.ssh_key_path} -N '' -C localchat",
check=True,
)
self.ssh_config_path = self.lxconfigs_dir / "ssh-config"
def write_ssh_config(self):
"""Write ``lxconfigs/ssh-config`` mapping all containers to their IPs.
Each Host block maps the container name, the domain name, and the
short relay name (e.g. ``_test0``) to the container's IP, using the
shared localchat SSH key. Returns the path to the file.
"""
containers = self.list_managed()
key_path = self.ssh_key_path
lines = ["# Auto-generated by cmdeploy lxc-start — do not edit\n"]
for c in containers:
hosts = [c["name"]]
domain = c.get("domain", "")
if domain and domain != c["name"]:
hosts.append(domain)
short = domain.split(".")[0]
if short and short not in hosts:
hosts.append(short)
lines.append(f"\nHost {' '.join(hosts)}\n")
lines.append(f" Hostname {c['ip']}\n")
lines.append(" User root\n")
lines.append(f" IdentityFile {key_path}\n")
lines.append(" IdentitiesOnly yes\n")
lines.append(" StrictHostKeyChecking accept-new\n")
lines.append(" UserKnownHostsFile /dev/null\n")
lines.append(" LogLevel ERROR\n")
path = self.ssh_config_path
path.write_text("".join(lines))
return path
def check_ssh_include(self):
"""Check if the user's ~/.ssh/config already includes our ssh-config."""
user_ssh_config = Path.home() / ".ssh" / "config"
if not user_ssh_config.exists():
return False
lines = filter(None, map(str.strip, user_ssh_config.open("r")))
return f"Include {self.ssh_config_path}" in lines
def run(self, args, check=True, capture=True, input=None):
"""Run an incus command."""
cmd = ["incus"] + list(args)
kwargs = dict(check=check, text=True, input=input)
if capture:
kwargs["capture_output"] = True
else:
kwargs["stdout"] = None
kwargs["stderr"] = None
return subprocess.run(cmd, **kwargs) # noqa: PLW1510
def run_json(self, args, check=True):
"""Run an incus command with ``--format=json``.
Returns the parsed JSON on success.
When *check* is True raises ``subprocess.CalledProcessError``
on non-zero exit; when False returns *None* instead.
"""
result = self.run(
list(args) + ["--format=json"],
check=check,
)
if result.returncode != 0:
return None
return json.loads(result.stdout)
def run_output(self, args, check=True):
"""Run an incus command and return its stdout.
When *check* is False, returns *None* on non-zero exit
instead of raising.
"""
result = self.run(args, check=check)
if result.returncode != 0:
return None
return result.stdout
def _find_image(self, alias):
"""Return *alias* if an image with that alias exists, else None."""
images = self.run_json(["image", "list"], check=False) or []
for img in images:
for a in img.get("aliases", []):
if a.get("name") == alias:
return alias
return None
def find_relay_image(self):
"""Return the relay image alias if it exists, else None."""
return self._find_image(RELAY_IMAGE_ALIAS)
def delete_images(self):
"""Delete the cached base and relay images."""
for alias in (RELAY_IMAGE_ALIAS, BASE_IMAGE_ALIAS):
self.run(["image", "delete", alias], check=False)
def list_managed(self):
"""Return list of dicts with name, ip, ipv6, domain, status, memory_usage."""
containers = []
for ct in self.run_json(["list"]):
config = ct.get("config", {})
if config.get(LABEL_KEY) != "true":
continue
name = ct["name"]
state = ct.get("state", {})
net = state.get("network") or {}
containers.append(
{
"name": name,
"ip": _extract_ip(net, "inet"),
"ipv6": _extract_ip(net, "inet6"),
"domain": config.get(
"user.localchat-domain", f"{name}{DOMAIN_SUFFIX}"
),
"status": ct.get("status", "Unknown"),
"memory_usage": state.get("memory", {}).get("usage", 0),
}
)
return containers
def ensure_base_image(self):
"""Build and cache a base image with openssh and the SSH key.
The image is published as a local incus image with alias
'localchat-base'. Subsequent container launches use this
image instead of the upstream Debian 12, skipping the
slow apt-get install step.
Returns the image alias.
"""
if self._find_image(BASE_IMAGE_ALIAS):
return BASE_IMAGE_ALIAS
print(" Building base image (one-time setup) ...")
self.run(["delete", BASE_SETUP_NAME, "--force"], check=False)
self.run(["image", "delete", BASE_IMAGE_ALIAS], check=False)
self.run(["launch", UPSTREAM_IMAGE, BASE_SETUP_NAME])
ct = Container(self, BASE_SETUP_NAME)
ct.wait_ready()
key_path = self.ssh_key_path
pub_key = key_path.with_suffix(".pub").read_text().strip()
ct.bash(f"""\
apt-get -o DPkg::Lock::Timeout=60 update
DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server python3
systemctl enable ssh
apt-get clean
mkdir -p /root/.ssh
chmod 700 /root/.ssh
echo '{pub_key}' > /root/.ssh/authorized_keys
chmod 600 /root/.ssh/authorized_keys
""")
self.run(["stop", BASE_SETUP_NAME])
self.run(["publish", BASE_SETUP_NAME, f"--alias={BASE_IMAGE_ALIAS}"])
self.run(["delete", BASE_SETUP_NAME, "--force"])
print(f" Base image '{BASE_IMAGE_ALIAS}' ready.")
return BASE_IMAGE_ALIAS
def get_container(self, name):
"""Return a container handle for the given name.
Accepts both short relay names (``test0``) and full Incus
container names (``test0-localchat``). Returns
``DNSContainer`` for the DNS container and
``RelayContainer`` for everything else.
"""
if name == DNS_CONTAINER_NAME:
return DNSContainer(self)
return RelayContainer(self, name.removesuffix("-localchat"))
def get_dns_container(self):
"""Return a DNSContainer handle."""
return DNSContainer(self)
class Container:
"""Lightweight handle for an Incus container.
Carries the container *name* and provides convenience methods
for running commands, managing lifecycle, and extracting state
so callers don't repeat the name everywhere.
"""
def __init__(self, incus, name, domain=None, memory="100MiB"):
self.incus = incus
self.name = name
self.domain = domain or f"{name}{DOMAIN_SUFFIX}"
self.memory = memory
self.ipv4 = None
self.ipv6 = None
def bash(self, script, check=True):
"""Returns stdout from executing ``bash -ec <script>`` inside this container.
*script* is dedented and stripped so callers can use triple-quoted strings.
When *check* is False, returns *None* on non-zero exit instead of raising.
"""
cmd = ["exec", self.name, "--", "bash", "-ec", textwrap.dedent(script).strip()]
return self.incus.run_output(cmd, check=check)
def run_cmd(self, *args, check=True):
"""Return stdout from running a command directly in the container (no shell).
When *check* is False, returns *None* on non-zero exit instead of raising.
"""
return self.incus.run_output(
["exec", self.name, "--", *args],
check=check,
)
def start(self):
self.incus.run(["start", self.name])
def stop(self, force=False):
cmd = ["stop", self.name]
if force:
cmd.append("--force")
self.incus.run(cmd, check=False)
def launch(self):
"""Launch from the best available image, return the alias used."""
image = self.incus.find_relay_image() or self.incus.ensure_base_image()
print(f" Launching from '{image}' image ...")
cfg = []
cfg += ("-c", f"{LABEL_KEY}=true")
cfg += ("-c", f"user.localchat-domain={self.domain}")
cfg += ("-c", f"limits.memory={self.memory}")
self.incus.run(["launch", image, self.name, *cfg])
return image
def ensure(self):
"""Create/start this container from the cached base image.
On first call, builds the base image (~30s).
Subsequent containers launch in ~2s from the cached image.
Returns ``self`` for chaining.
"""
data = self.incus.run_json(["list", self.name], check=False) or []
existing = [c for c in data if c["name"] == self.name]
if existing:
if existing[0]["status"] != "Running":
self.start()
else:
self.launch()
self.wait_ready()
return self
def destroy(self):
"""Stop, delete, and clean up config files."""
self.stop(force=True)
self.incus.run(["delete", self.name, "--force"], check=False)
def push_file_content(self, dest_path, content):
"""Write *content* to *dest_path* inside the container.
*content* is dedented and stripped so callers can use
indented triple-quoted strings.
"""
content = textwrap.dedent(content).strip() + "\n"
self.incus.run(
["file", "push", "-", f"{self.name}{dest_path}"],
input=content,
)
self.bash(f"chmod 644 {dest_path}")
def wait_ready(self, timeout=60):
"""Wait until the container is running with an IPv4 address.
Sets ``self.ipv4`` and ``self.ipv6`` (may be *None*),
or raises ``TimeoutError``.
"""
deadline = time.time() + timeout
while time.time() < deadline:
data = self.incus.run_json(
["list", self.name],
check=False,
)
if data and data[0].get("status") == "Running":
net = data[0].get("state", {}).get("network", {})
self.ipv4 = _extract_ip(net, "inet")
self.ipv6 = _extract_ip(net, "inet6")
if self.ipv4:
return
time.sleep(1)
raise TimeoutError(
f"Container {self.name!r} did not become ready within {timeout}s"
)
def rss_mib(self):
"""Return ``(used, total)`` memory from container (or None if unobtainable)."""
output = self.bash("free -m", check=False)
if output:
for line in output.splitlines():
if line.startswith("Mem:"):
parts = line.split()
return int(parts[2]), int(parts[1])
class RelayContainer(Container):
"""Container handle for a chatmail relay.
Accepts the short relay name (e.g. ``test0``) and derives
the Incus container name and mail domain automatically.
"""
def __init__(self, incus, name):
super().__init__(
incus,
f"{name}-localchat",
domain=f"_{name}{DOMAIN_SUFFIX}",
memory="500MiB",
)
self.sname = name
self.ini = incus.lxconfigs_dir / f"chatmail-{name}.ini"
self.zone = incus.lxconfigs_dir / f"{name}.zone"
def launch(self):
"""Launch (from a potentially cached image) and clear inherited chatmail-version."""
image = super().launch()
self.bash("rm -f /etc/chatmail-version")
return image
def destroy(self):
"""Stop, delete, and clean up config files."""
super().destroy()
if self.ini.exists():
self.ini.unlink()
def disable_ipv6(self):
"""Disable IPv6 inside the container via sysctl."""
self.bash("""\
sysctl -w net.ipv6.conf.all.disable_ipv6=1
sysctl -w net.ipv6.conf.default.disable_ipv6=1
mkdir -p /etc/sysctl.d
printf 'net.ipv6.conf.all.disable_ipv6=1\\n
net.ipv6.conf.default.disable_ipv6=1\\n'
> /etc/sysctl.d/99-disable-ipv6.conf
""")
def configure_hosts(self, ip):
"""Set hostname and /etc/hosts inside the container."""
self.bash(f"""\
echo '{self.name}' > /etc/hostname
hostname {self.name}
echo '{ip} {self.name} {self.domain}' >> /etc/hosts
""")
def publish_as_relay_image(self):
"""Publish this container as a reusable relay image.
Stops the container, publishes it as 'localchat-relay',
then restarts it.
"""
if self.incus.find_relay_image():
return
print(f" Publishing {self.name!r} as '{RELAY_IMAGE_ALIAS}' image ...")
self.incus.run(
["publish", self.name, f"--alias={RELAY_IMAGE_ALIAS}", "--force"]
)
self.wait_ready()
print(f" Relay image '{RELAY_IMAGE_ALIAS}' ready.")
def deployed_version(self):
"""Read /etc/chatmail-version, or None if absent."""
output = self.bash("cat /etc/chatmail-version", check=False)
return output.strip() if output else None
def deployed_domain(self):
"""Read the domain deployed on the container (postfix myhostname)."""
output = self.bash(
"postconf -h myhostname 2>/dev/null",
check=False,
)
return output.strip() if output else None
def verify_ssh(self, ssh_config):
"""Verify SSH connectivity to this container."""
cmd = f"ssh -F {ssh_config} -o ConnectTimeout=10 root@{self.domain} hostname"
return shell(cmd, timeout=15).returncode == 0
def configure_dns(self, dns_ip):
"""Point this container's resolver at *dns_ip*.
Disables systemd-resolved to free port 53 and writes
a static /etc/resolv.conf. Also configures unbound
(if present) to forward .localchat queries.
"""
self.bash(f"""\
systemctl disable --now systemd-resolved 2>/dev/null || true
rm -f /etc/resolv.conf
echo 'nameserver {dns_ip}' > /etc/resolv.conf
mkdir -p /etc/unbound/unbound.conf.d
printf 'server:\\n domain-insecure: "localchat"\\n\\n
forward-zone:\\n name: "localchat"\\n
forward-addr: {dns_ip}\\n'
> /etc/unbound/unbound.conf.d/localchat-forward.conf
systemctl restart unbound 2>/dev/null || true
""")
def write_ini(self, disable_ipv6=False):
"""Generate a chatmail.ini config file in lxconfigs/."""
from chatmaild.config import write_initial_config
overrides = {
"max_user_send_per_minute": 600,
"max_user_send_burst_size": 100,
"mtail_address": "127.0.0.1",
}
if disable_ipv6:
overrides["disable_ipv6"] = "True"
write_initial_config(self.ini, self.domain, overrides)
return self.ini
class DNSContainer(Container):
"""Specialised container handle for the PowerDNS name server."""
def __init__(self, incus):
super().__init__(incus, DNS_CONTAINER_NAME, domain=DNS_DOMAIN)
def pdnsutil(self, *args, check=True):
"""Run ``pdnsutil <args>`` inside the DNS container."""
return self.run_cmd("pdnsutil", *args, check=check)
def replace_rrset(self, zone, name, rtype, ttl, rdata):
"""Shortcut for ``pdnsutil replace-rrset``."""
self.pdnsutil("replace-rrset", zone, name, rtype, ttl, rdata)
def restart_services(self):
"""Restart pdns and pdns-recursor."""
self.bash("""\
systemctl restart pdns
systemctl restart pdns-recursor || true
""")
def ensure(self):
"""Create the DNS container with PowerDNS if needed.
Calls ``super().ensure()`` to create/start the container
and set up SSH, then installs PowerDNS and configures
the Incus bridge to use this container as DNS.
"""
super().ensure()
self._install_powerdns()
self.incus.run(
["network", "set", "incusbr0", "dns.mode=none"],
check=False,
)
self.incus.run(
["network", "set", "incusbr0", f"raw.dnsmasq=dhcp-option=6,{self.ipv4}"],
check=False,
)
def _install_powerdns(self):
"""Install and configure PowerDNS if not already present."""
if self.run_cmd("which", "pdns_server", check=False) is not None:
return
self.bash("""\
systemctl disable --now systemd-resolved 2>/dev/null || true
rm -f /etc/resolv.conf
echo 'nameserver 9.9.9.9' > /etc/resolv.conf
apt-get -o DPkg::Lock::Timeout=60 update
DEBIAN_FRONTEND=noninteractive apt-get install -y \
pdns-server pdns-backend-sqlite3 sqlite3 pdns-recursor dnsutils
systemctl stop pdns pdns-recursor || true
mkdir -p /var/lib/powerdns
sqlite3 /var/lib/powerdns/pdns.sqlite3 \
</usr/share/doc/pdns-backend-sqlite3/schema.sqlite3.sql
chown -R pdns:pdns /var/lib/powerdns
""")
self.push_file_content(
"/etc/powerdns/pdns.conf",
"""\
launch=gsqlite3
gsqlite3-database=/var/lib/powerdns/pdns.sqlite3
local-address=127.0.0.1
local-port=5353
""",
)
self.push_file_content(
"/etc/powerdns/recursor.conf",
"""\
local-address=0.0.0.0
local-port=53
forward-zones=localchat=127.0.0.1:5353
forward-zones-recurse=.=9.9.9.9;149.112.112.112
allow-from=0.0.0.0/0
dont-query=
dnssec=off
""",
)
self.bash("""\
systemctl start pdns
systemctl start pdns-recursor
echo 'nameserver 127.0.0.1' > /etc/resolv.conf
""")
def reset_dns_records(self, dns_ip, domains):
"""Create DNS zones with initial A records via pdnsutil.
Only sets SOA, NS, and A records as the minimal set
needed for SSH connectivity. Full records (MX, TXT, SRV,
CNAME, DKIM) are added later by ``cmdeploy dns``.
Args:
dns_ip: IP of the DNS container
domains: list of dicts with 'name', 'domain', 'ip'
"""
for d in domains:
domain = d["domain"]
ip = d["ip"]
print(f" {domain} -> {ip}")
# Delete and recreate zone fresh (removes stale records)
self.pdnsutil("delete-zone", domain, check=False)
self.pdnsutil("create-zone", domain, f"ns.{domain}")
serial = str(int(time.time()))
soa = f"ns.{domain} hostmaster.{domain} {serial} 3600 900 604800 300"
self.replace_rrset(domain, ".", "SOA", "3600", soa)
self.replace_rrset(domain, ".", "NS", "3600", f"ns.{domain}.")
self.replace_rrset(domain, ".", "A", "3600", ip)
self.replace_rrset(domain, "ns", "A", "3600", dns_ip)
# AAAA (domain -> container IPv6, if available)
ipv6 = d.get("ipv6")
if ipv6:
self.replace_rrset(domain, ".", "AAAA", "3600", ipv6)
print(f" zone reset: SOA, NS, A, AAAA ({ip}, {ipv6})")
else:
# Remove any stale AAAA record
self.pdnsutil("delete-rrset", domain, ".", "AAAA", check=False)
print(f" zone reset: SOA, NS, A ({ip}, IPv4-only)")
self.restart_services()
def set_dns_records(self, text):
"""Add or overwrite DNS records from standard BIND format.
Uses ``cmdeploy.dns.parse_zone_records`` to parse.
Zones are created automatically from the record names.
"""
from ..dns import parse_zone_records
zones_seen = set()
for name, ttl, rtype, rdata in parse_zone_records(text):
# Derive zone from name: find top-level .localchat domain
name_parts = name.split(".")
zone = name # fallback
for i in range(len(name_parts) - 1):
if name_parts[i + 1 :] == ["localchat"]:
zone = ".".join(name_parts[i:])
break
# Create zone if first time seeing it
if zone not in zones_seen:
self.pdnsutil(
"create-zone",
zone,
f"ns.{zone}",
check=False,
)
zones_seen.add(zone)
# Figure out the record name relative to zone
if name == zone:
relative = "."
elif name.endswith(f".{zone}"):
relative = name[: -(len(zone) + 1)]
else:
relative = name
self.replace_rrset(zone, relative, rtype, ttl, rdata)
if zones_seen:
self.restart_services()

View File

@@ -58,8 +58,8 @@ def get_dkim_entry(mail_domain, pre_command, dkim_selector):
dkim_value = '" "'.join(re.findall(".{1,255}", dkim_value_raw)) dkim_value = '" "'.join(re.findall(".{1,255}", dkim_value_raw))
web_dkim_value = "".join(re.findall(".{1,255}", dkim_value_raw)) web_dkim_value = "".join(re.findall(".{1,255}", dkim_value_raw))
return ( return (
f'{dkim_selector}._domainkey.{mail_domain}. TXT "{dkim_value}"', f'{dkim_selector}._domainkey.{mail_domain}. 3600 IN TXT "{dkim_value}"',
f'{dkim_selector}._domainkey.{mail_domain}. TXT "{web_dkim_value}"', f'{dkim_selector}._domainkey.{mail_domain}. 3600 IN TXT "{web_dkim_value}"',
) )
@@ -94,9 +94,11 @@ def check_zonefile(zonefile, verbose=True):
if not zf_line.strip() or zf_line.startswith(";"): if not zf_line.strip() or zf_line.startswith(";"):
continue continue
print(f"dns-checking {zf_line!r}") if verbose else log_progress("") print(f"dns-checking {zf_line!r}") if verbose else log_progress("")
zf_domain, zf_typ, zf_value = zf_line.split(maxsplit=2) parts = zf_line.split(None, 4)
zf_domain = zf_domain.rstrip(".") zf_domain = parts[0].rstrip(".")
zf_value = zf_value.strip() # parts[1]=TTL, parts[2]=IN, parts[3]=type, parts[4]=rdata
zf_typ = parts[3]
zf_value = parts[4].strip()
query_value = query_dns(zf_typ, zf_domain) query_value = query_dns(zf_typ, zf_domain)
if zf_value != query_value: if zf_value != query_value:
assert zf_typ in ("A", "AAAA", "CNAME", "CAA", "SRV", "MX", "TXT"), zf_line assert zf_typ in ("A", "AAAA", "CNAME", "CAA", "SRV", "MX", "TXT"), zf_line

View File

@@ -12,13 +12,27 @@ def openssl_selfsigned_args(domain, cert_path, key_path, days=36500):
``www.<domain>`` and ``mta-sts.<domain>``. ``www.<domain>`` and ``mta-sts.<domain>``.
""" """
return [ return [
"openssl", "req", "-x509", "openssl",
"-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:P-256", "req",
"-noenc", "-days", str(days), "-x509",
"-keyout", str(key_path), "-newkey",
"-out", str(cert_path), "ec",
"-subj", f"/CN={domain}", "-pkeyopt",
"-addext", "extendedKeyUsage=serverAuth,clientAuth", "ec_paramgen_curve:P-256",
"-noenc",
"-days",
str(days),
"-keyout",
str(key_path),
"-out",
str(cert_path),
"-subj",
f"/CN={domain}",
# Mark as end-entity cert so it cannot be used as a CA to sign others.
"-addext",
"basicConstraints=critical,CA:FALSE",
"-addext",
"extendedKeyUsage=serverAuth,clientAuth",
"-addext", "-addext",
f"subjectAltName=DNS:{domain},DNS:www.{domain},DNS:mta-sts.{domain}", f"subjectAltName=DNS:{domain},DNS:www.{domain},DNS:mta-sts.{domain}",
] ]
@@ -40,7 +54,9 @@ class SelfSignedTlsDeployer(Deployer):
def configure(self): def configure(self):
args = openssl_selfsigned_args( args = openssl_selfsigned_args(
self.mail_domain, self.cert_path, self.key_path, self.mail_domain,
self.cert_path,
self.key_path,
) )
cmd = shlex.join(args) cmd = shlex.join(args)
server.shell( server.shell(

View File

@@ -49,8 +49,13 @@ class SSHExec:
RemoteError = execnet.RemoteError RemoteError = execnet.RemoteError
FuncError = FuncError FuncError = FuncError
def __init__(self, host, verbose=False, python="python3", timeout=60): def __init__(
self.gateway = execnet.makegateway(f"ssh=root@{host}//python={python}") self, host, verbose=False, python="python3", timeout=60, ssh_config=None
):
spec = f"ssh=root@{host}//python={python}"
if ssh_config:
spec += f"//ssh_config={ssh_config}"
self.gateway = execnet.makegateway(spec)
self._remote_cmdloop_channel = bootstrap_remote(self.gateway, remote) self._remote_cmdloop_channel = bootstrap_remote(self.gateway, remote)
self.timeout = timeout self.timeout = timeout
self.verbose = verbose self.verbose = verbose
@@ -113,3 +118,46 @@ class LocalExec:
res = self(call, kwargs, log_callback=remote.rshell.log_progress) res = self(call, kwargs, log_callback=remote.rshell.log_progress)
print_stderr() print_stderr()
return res return res
# pyinfra exposes a ``ssh_config_file`` data key that *should* let
# paramiko parse an SSH config file directly. In practice it silently
# fails to connect (zero hosts / zero operations), so we resolve the
# hostname and identity-file ourselves and pass them via
# ``--data ssh_hostname`` / ``--data ssh_key`` instead.
# Execnet uses ssh natively (and not paramiko) and doesn't have this problem.
def _get_from_ssh_config(host, ssh_config_path, key):
"""Internal helper to parse a value for a specific key from ssh-config."""
current_hosts = []
found_value = None
with open(ssh_config_path) as f:
for raw_line in f:
line = raw_line.strip()
if not line or line.startswith("#"):
continue
parts = line.split(None, 1)
if not parts:
continue
directive = parts[0].lower()
if directive == "host":
if host in current_hosts and found_value:
return found_value
current_hosts = parts[1].split()
found_value = None
elif directive == key.lower():
found_value = parts[1]
if host in current_hosts and found_value:
return found_value
return None
def resolve_host_from_ssh_config(host, ssh_config_path):
"""Resolve a host alias to its IP from an ssh-config file."""
return _get_from_ssh_config(host, ssh_config_path, "Hostname") or host
def resolve_key_from_ssh_config(host, ssh_config_path):
"""Resolve a host alias to its IdentityFile from an ssh-config file."""
return _get_from_ssh_config(host, ssh_config_path, "IdentityFile")

View File

@@ -1,17 +1,18 @@
; Required DNS entries for chatmail servers ; Required DNS entries
zftest.testrun.org. A 135.181.204.127 zftest.testrun.org. 3600 IN A 135.181.204.127
zftest.testrun.org. AAAA 2a01:4f9:c012:52f4::1 zftest.testrun.org. 3600 IN AAAA 2a01:4f9:c012:52f4::1
zftest.testrun.org. MX 10 zftest.testrun.org. zftest.testrun.org. 3600 IN MX 10 zftest.testrun.org.
_mta-sts.zftest.testrun.org. TXT "v=STSv1; id=202403211706" _mta-sts.zftest.testrun.org. 3600 IN TXT "v=STSv1; id=202403211706"
mta-sts.zftest.testrun.org. CNAME zftest.testrun.org. mta-sts.zftest.testrun.org. 3600 IN CNAME zftest.testrun.org.
www.zftest.testrun.org. CNAME zftest.testrun.org. www.zftest.testrun.org. 3600 IN CNAME zftest.testrun.org.
opendkim._domainkey.zftest.testrun.org. TXT "v=DKIM1;k=rsa;p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAoYt82CVUyz2ouaqjX2kB+5J80knAyoOU3MGU5aWppmwUwwTvj/oSTSpkc5JMtVTRmKKr8NUDWAL1Yw7dfGqqPHdHfwwjS3BIvDzYx+hzgtz62RnfNgV+/2MAoNpfX7cAFIHdRzEHNtwugc3RDLquqPoupAE3Y2YRw2T5zG5fILh4vwIcJZL5Uq6B92j8wwJqOex" "33n+vm1NKQ9rxo/UsHAmZlJzpooXcG/4igTBxJyJlamVSRR6N7Nul1v//YJb7J6v2o0iPHW6uE0StzKaPPNC2IVosSRFbD9H2oqppltptFSNPlI0E+t0JBWHem6YK7xcugiO3ImMCaaU8g6Jt/wIDAQAB;s=email;t=s" opendkim._domainkey.zftest.testrun.org. 3600 IN TXT "v=DKIM1;k=rsa;p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAoYt82CVUyz2ouaqjX2kB+5J80knAyoOU3MGU5aWppmwUwwTvj/oSTSpkc5JMtVTRmKKr8NUDWAL1Yw7dfGqqPHdHfwwjS3BIvDzYx+hzgtz62RnfNgV+/2MAoNpfX7cAFIHdRzEHNtwugc3RDLquqPoupAE3Y2YRw2T5zG5fILh4vwIcJZL5Uq6B92j8wwJqOex" "33n+vm1NKQ9rxo/UsHAmZlJzpooXcG/4igTBxJyJlamVSRR6N7Nul1v//YJb7J6v2o0iPHW6uE0StzKaPPNC2IVosSRFbD9H2oqppltptFSNPlI0E+t0JBWHem6YK7xcugiO3ImMCaaU8g6Jt/wIDAQAB;s=email;t=s"
; Recommended DNS entries ; Recommended DNS entries
_submission._tcp.zftest.testrun.org. SRV 0 1 587 zftest.testrun.org. zftest.testrun.org. 3600 IN TXT "v=spf1 a ~all"
_submissions._tcp.zftest.testrun.org. SRV 0 1 465 zftest.testrun.org. _dmarc.zftest.testrun.org. 3600 IN TXT "v=DMARC1;p=reject;adkim=s;aspf=s"
_imap._tcp.zftest.testrun.org. SRV 0 1 143 zftest.testrun.org. zftest.testrun.org. 3600 IN CAA 0 issue "letsencrypt.org;accounturi=https://acme-v02.api.letsencrypt.org/acme/acct/1371472956"
_imaps._tcp.zftest.testrun.org. SRV 0 1 993 zftest.testrun.org. _adsp._domainkey.zftest.testrun.org. 3600 IN TXT "dkim=discardable"
zftest.testrun.org. CAA 0 issue "letsencrypt.org;accounturi=https://acme-v02.api.letsencrypt.org/acme/acct/1371472956" _submission._tcp.zftest.testrun.org. 3600 IN SRV 0 1 587 zftest.testrun.org.
zftest.testrun.org. TXT "v=spf1 a:zftest.testrun.org ~all" _submissions._tcp.zftest.testrun.org. 3600 IN SRV 0 1 465 zftest.testrun.org.
_dmarc.zftest.testrun.org. TXT "v=DMARC1;p=reject;adkim=s;aspf=s" _imap._tcp.zftest.testrun.org. 3600 IN SRV 0 1 143 zftest.testrun.org.
_adsp._domainkey.zftest.testrun.org. TXT "dkim=discardable" _imaps._tcp.zftest.testrun.org. 3600 IN SRV 0 1 993 zftest.testrun.org.

View File

@@ -20,7 +20,7 @@ def test_fastcgi_working(maildomain, chatmail_config):
@pytest.mark.filterwarnings("ignore::urllib3.exceptions.InsecureRequestWarning") @pytest.mark.filterwarnings("ignore::urllib3.exceptions.InsecureRequestWarning")
def test_newemail_configure(maildomain, rpc, chatmail_config): def test_newemail_configure(maildomain, maildomain_ip, rpc, chatmail_config):
"""Test configuring accounts by scanning a QR code works.""" """Test configuring accounts by scanning a QR code works."""
url = f"DCACCOUNT:https://{maildomain}/new" url = f"DCACCOUNT:https://{maildomain}/new"
for i in range(3): for i in range(3):
@@ -30,12 +30,15 @@ def test_newemail_configure(maildomain, rpc, chatmail_config):
# set_config_from_qr, so fetch credentials via requests instead # set_config_from_qr, so fetch credentials via requests instead
res = requests.post(f"https://{maildomain}/new", verify=False) res = requests.post(f"https://{maildomain}/new", verify=False)
data = res.json() data = res.json()
rpc.add_or_update_transport(account_id, { rpc.add_or_update_transport(
"addr": data["email"], account_id,
"password": data["password"], {
"imapServer": maildomain, "addr": data["email"],
"smtpServer": maildomain, "password": data["password"],
"certificateChecks": "acceptInvalidCertificates", "imapServer": maildomain_ip,
}) "smtpServer": maildomain_ip,
"certificateChecks": "acceptInvalidCertificates",
},
)
else: else:
rpc.add_transport_from_qr(account_id, url) rpc.add_transport_from_qr(account_id, url)

View File

@@ -1,4 +1,5 @@
import datetime import datetime
import os
import smtplib import smtplib
import socket import socket
import subprocess import subprocess
@@ -13,7 +14,8 @@ from cmdeploy.cmdeploy import get_sshexec
class TestSSHExecutor: class TestSSHExecutor:
@pytest.fixture(scope="class") @pytest.fixture(scope="class")
def sshexec(self, sshdomain): def sshexec(self, sshdomain):
return get_sshexec(sshdomain) ssh_config = os.environ.get("CHATMAIL_SSH_CONFIG")
return get_sshexec(sshdomain, ssh_config=ssh_config)
def test_ls(self, sshexec): def test_ls(self, sshexec):
out = sshexec(call=remote.rdns.shell, kwargs=dict(command="ls")) out = sshexec(call=remote.rdns.shell, kwargs=dict(command="ls"))
@@ -132,11 +134,10 @@ def test_authenticated_from(cmsetup, maildata):
@pytest.mark.parametrize("from_addr", ["fake@example.org", "fake@testrun.org"]) @pytest.mark.parametrize("from_addr", ["fake@example.org", "fake@testrun.org"])
def test_reject_missing_dkim(cmsetup, maildata, from_addr): def test_reject_missing_dkim(cmsetup, maildata, from_addr):
domain = cmsetup.maildomain domain = cmsetup.maildomain
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
try: try:
sock.connect((domain, 25)) sock = socket.create_connection((domain, 25), timeout=10)
except socket.timeout: sock.close()
except (socket.timeout, OSError):
pytest.skip(f"port 25 not reachable for {domain}") pytest.skip(f"port 25 not reachable for {domain}")
recipient = cmsetup.gen_users(1)[0] recipient = cmsetup.gen_users(1)[0]

View File

@@ -1,4 +1,5 @@
import ipaddress import ipaddress
import os
import re import re
import time import time
@@ -92,7 +93,9 @@ class TestEndToEndDeltaChat:
lp.sec(f"filling remote inbox for {user}") lp.sec(f"filling remote inbox for {user}")
fn = f"7743102289.M843172P2484002.c20,S={quota},W=2398:2," fn = f"7743102289.M843172P2484002.c20,S={quota},W=2398:2,"
path = chatmail_config.mailboxes_dir.joinpath(user, "cur", fn) path = chatmail_config.mailboxes_dir.joinpath(user, "cur", fn)
sshexec = get_sshexec(sshdomain) sshexec = get_sshexec(
sshdomain, ssh_config=os.environ.get("CHATMAIL_SSH_CONFIG")
)
sshexec(call=rshell.write_numbytes, kwargs=dict(path=str(path), num=120)) sshexec(call=rshell.write_numbytes, kwargs=dict(path=str(path), num=120))
res = sshexec(call=rshell.dovecot_recalc_quota, kwargs=dict(user=user)) res = sshexec(call=rshell.dovecot_recalc_quota, kwargs=dict(user=user))
assert res["percent"] >= 100 assert res["percent"] >= 100

View File

@@ -9,6 +9,9 @@ def test_status_cmd(chatmail_config, capsys, request):
if os.getenv("CHATMAIL_SSH"): if os.getenv("CHATMAIL_SSH"):
command.append("--ssh-host") command.append("--ssh-host")
command.append(os.getenv("CHATMAIL_SSH")) command.append(os.getenv("CHATMAIL_SSH"))
if os.getenv("CHATMAIL_SSH_CONFIG"):
command.append("--ssh-config")
command.append(os.getenv("CHATMAIL_SSH_CONFIG"))
assert main(command) == 0 assert main(command) == 0
status_out = capsys.readouterr() status_out = capsys.readouterr()
print(status_out.out) print(status_out.out)

View File

@@ -2,7 +2,9 @@ import imaplib
import itertools import itertools
import os import os
import random import random
import re
import smtplib import smtplib
import socket
import ssl import ssl
import subprocess import subprocess
import time import time
@@ -20,6 +22,64 @@ def pytest_addoption(parser):
) )
def _parse_ssh_config_hosts(path):
"""Parse an OpenSSH config file and return a dict of hostname -> IP."""
mapping = {}
current_names = []
for ln in Path(path).read_text().splitlines():
line = ln.strip()
m = re.match(r"^Host\s+(.+)", line)
if m:
current_names = m.group(1).split()
continue
m = re.match(r"^Hostname\s+(\S+)", line)
if m and current_names:
ip = m.group(1)
for name in current_names:
mapping[name] = ip
current_names = []
return mapping
_original_getaddrinfo = socket.getaddrinfo
def _make_patched_getaddrinfo(host_map):
"""Return a getaddrinfo that resolves hosts in host_map to their IPs."""
def patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
if host in host_map:
ip = host_map[host]
return _original_getaddrinfo(ip, port, family, type, proto, flags)
return _original_getaddrinfo(host, port, family, type, proto, flags)
return patched_getaddrinfo
@pytest.fixture(autouse=True, scope="session")
def _setup_localchat_dns():
"""Monkey-patch socket.getaddrinfo to resolve .localchat via ssh-config."""
ssh_config = os.environ.get("CHATMAIL_SSH_CONFIG")
if not ssh_config or not Path(ssh_config).exists():
yield {}
return
host_map = _parse_ssh_config_hosts(ssh_config)
if not host_map:
yield {}
return
socket.getaddrinfo = _make_patched_getaddrinfo(host_map)
try:
yield host_map
finally:
socket.getaddrinfo = _original_getaddrinfo
@pytest.fixture(scope="session")
def ssh_config_host_map(_setup_localchat_dns):
"""Return the host-name → IP map parsed from ssh-config."""
return _setup_localchat_dns
def pytest_configure(config): def pytest_configure(config):
config._benchresults = {} config._benchresults = {}
config.addinivalue_line( config.addinivalue_line(
@@ -35,6 +95,11 @@ def pytest_runtest_setup(item):
def _get_chatmail_config(): def _get_chatmail_config():
ini = os.environ.get("CHATMAIL_INI")
if ini:
path = Path(ini).resolve()
if path.exists():
return read_config(path), path
current = Path().resolve() current = Path().resolve()
while 1: while 1:
path = current.joinpath("chatmail.ini").resolve() path = current.joinpath("chatmail.ini").resolve()
@@ -65,6 +130,12 @@ def sshdomain(maildomain):
return os.environ.get("CHATMAIL_SSH", maildomain) return os.environ.get("CHATMAIL_SSH", maildomain)
@pytest.fixture(scope="session")
def maildomain_ip(maildomain, ssh_config_host_map):
"""Return the IP for maildomain from ssh-config, or maildomain itself."""
return ssh_config_host_map.get(maildomain, maildomain)
@pytest.fixture @pytest.fixture
def maildomain2(): def maildomain2():
domain = os.environ.get("CHATMAIL_DOMAIN2") domain = os.environ.get("CHATMAIL_DOMAIN2")
@@ -306,23 +377,32 @@ from deltachat_rpc_client import DeltaChat, Rpc
class ChatmailACFactory: class ChatmailACFactory:
"""RPC-based account factory for chatmail testing.""" """RPC-based account factory for chatmail testing."""
def __init__(self, rpc, maildomain, gencreds, chatmail_config): def __init__(
self,
rpc,
maildomain,
maildomain_ip,
gencreds,
chatmail_config,
ssh_config_host_map,
):
self.dc = DeltaChat(rpc) self.dc = DeltaChat(rpc)
self.rpc = rpc self.rpc = rpc
self._maildomain = maildomain self._maildomain = maildomain
self._maildomain_ip = maildomain_ip
self.gencreds = gencreds self.gencreds = gencreds
self.chatmail_config = chatmail_config self.chatmail_config = chatmail_config
self._ssh_config_host_map = ssh_config_host_map
def _make_transport(self, domain): def _make_transport(self, domain):
"""Build a transport config dict for the given domain.""" """Build a transport config dict for the given domain."""
addr, password = self.gencreds(domain) addr, password = self.gencreds(domain)
server = self._ssh_config_host_map.get(domain, domain)
transport = { transport = {
"addr": addr, "addr": addr,
"password": password, "password": password,
# Setting server explicitly skips requesting autoconfig XML, "imapServer": server,
# see https://datatracker.ietf.org/doc/draft-ietf-mailmaint-autoconfig/ "smtpServer": server,
"imapServer": domain,
"smtpServer": domain,
} }
if self.chatmail_config.tls_cert_mode == "self": if self.chatmail_config.tls_cert_mode == "self":
transport["certificateChecks"] = "acceptInvalidCertificates" transport["certificateChecks"] = "acceptInvalidCertificates"
@@ -376,13 +456,17 @@ def rpc(tmp_path_factory):
@pytest.fixture @pytest.fixture
def cmfactory(rpc, gencreds, maildomain, chatmail_config): def cmfactory(
rpc, gencreds, maildomain, maildomain_ip, chatmail_config, ssh_config_host_map
):
"""Return a ChatmailACFactory for creating online Delta Chat accounts.""" """Return a ChatmailACFactory for creating online Delta Chat accounts."""
return ChatmailACFactory( return ChatmailACFactory(
rpc=rpc, rpc=rpc,
maildomain=maildomain, maildomain=maildomain,
maildomain_ip=maildomain_ip,
gencreds=gencreds, gencreds=gencreds,
chatmail_config=chatmail_config, chatmail_config=chatmail_config,
ssh_config_host_map=ssh_config_host_map,
) )
@@ -394,14 +478,21 @@ def remote(sshdomain):
class Remote: class Remote:
def __init__(self, sshdomain): def __init__(self, sshdomain):
self.sshdomain = sshdomain self.sshdomain = sshdomain
self.ssh_config = os.environ.get("CHATMAIL_SSH_CONFIG")
def iter_output(self, logcmd="", ready=None): def iter_output(self, logcmd="", ready=None):
getjournal = "journalctl -f" if not logcmd else logcmd getjournal = "journalctl -f" if not logcmd else logcmd
print(self.sshdomain) print(self.sshdomain)
match self.sshdomain: match self.sshdomain:
case "@local": command = [] case "@local":
case "localhost": command = [] command = []
case _: command = ["ssh", f"root@{self.sshdomain}"] case "localhost":
command = []
case _:
command = ["ssh"]
if self.ssh_config:
command.extend(["-F", self.ssh_config])
command.append(f"root@{self.sshdomain}")
[command.append(arg) for arg in getjournal.split()] [command.append(arg) for arg in getjournal.split()]
self.popen = subprocess.Popen( self.popen = subprocess.Popen(
command, command,

View File

@@ -23,7 +23,10 @@ class TestCmdline:
run = parser.parse_args(["run"]) run = parser.parse_args(["run"])
assert init and run assert init and run
def test_init_not_overwrite(self, capsys): def test_init_not_overwrite(self, tmp_path, capsys, monkeypatch):
monkeypatch.delenv("CHATMAIL_INI", raising=False)
monkeypatch.chdir(tmp_path)
assert main(["init", "chat.example.org"]) == 0 assert main(["init", "chat.example.org"]) == 0
capsys.readouterr() capsys.readouterr()

View File

@@ -3,7 +3,7 @@ from copy import deepcopy
import pytest import pytest
from cmdeploy import remote from cmdeploy import remote
from cmdeploy.dns import check_full_zone, check_initial_remote_data from cmdeploy.dns import check_full_zone, check_initial_remote_data, parse_zone_records
@pytest.fixture @pytest.fixture
@@ -126,17 +126,11 @@ class TestPerformInitialChecks:
def parse_zonefile_into_dict(zonefile, mockdns_base, only_required=False): def parse_zonefile_into_dict(zonefile, mockdns_base, only_required=False):
for zf_line in zonefile.split("\n"): if only_required:
if zf_line.startswith("#"): # Only take records before the "; Recommended" section
if "Recommended" in zf_line and only_required: zonefile = zonefile.split("; Recommended")[0]
return for name, ttl, rtype, rdata in parse_zone_records(zonefile):
continue mockdns_base.setdefault(rtype, {})[name] = rdata
if not zf_line.strip():
continue
zf_domain, zf_typ, zf_value = zf_line.split(maxsplit=2)
zf_domain = zf_domain.rstrip(".")
zf_value = zf_value.strip()
mockdns_base.setdefault(zf_typ, {})[zf_domain] = zf_value
class MockSSHExec: class MockSSHExec:

View File

@@ -0,0 +1,174 @@
"""Tests for cmdeploy lxc-* subcommands."""
import shutil
import subprocess
import sys
import pytest
from cmdeploy.lxc import cli
from cmdeploy.lxc.incus import Incus
pytestmark = pytest.mark.skipif(
not shutil.which("incus") or not shutil.which("lxc"),
reason="incus/lxc not installed",
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def ix():
return Incus()
@pytest.fixture(scope="session")
def lxc_setup():
ix = Incus()
ix.get_dns_container().ensure()
return ix.list_managed()
@pytest.fixture(scope="session")
def relay_container(lxc_setup):
test_names = {f"{n}-localchat" for n in cli.RELAY_NAMES}
relays = [c for c in lxc_setup if c["name"] in test_names and c.get("ip")]
if not relays:
pytest.skip("no test relay containers running")
return relays[0]
@pytest.fixture
def cmdeploy():
def run(*args):
return subprocess.run(
[sys.executable, "-m", "cmdeploy.cmdeploy", *args],
capture_output=True,
text=True,
check=False,
)
return run
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"subcmd, expected, absent",
[
(None, ["lxc-start", "lxc-stop", "lxc-test", "lxc-status"], ["lxc-destroy"]),
("lxc-start", ["--ipv4-only", "--run"], ["--config"]),
("lxc-stop", ["--destroy", "--destroy-all"], ["--config"]),
("lxc-test", ["--one"], ["--config"]),
("lxc-status", [], ["--config"]),
("run", ["--ssh-config"], ["--lxc"]),
("dns", ["--ssh-config"], []),
("test", ["--ssh-config"], []),
("status", ["--ssh-config"], []),
],
)
def test_help_options(cmdeploy, subcmd, expected, absent):
args = [subcmd, "--help"] if subcmd else ["--help"]
result = cmdeploy(*args)
output = result.stdout + result.stderr
assert result.returncode == 0
for flag in expected:
assert flag in output
for flag in absent:
assert flag not in output
class TestSSHConfig:
def test_lxconfigs(self, ix, lxc_setup):
d = ix.lxconfigs_dir
assert d.name == "lxconfigs"
assert d.exists()
path = ix.ssh_config_path
assert path.name == "ssh-config"
assert path.parent.name == "lxconfigs"
def test_write_ssh_config(self, ix, lxc_setup):
path = ix.write_ssh_config()
assert path.exists()
text = path.read_text()
for c in lxc_setup:
if c.get("ip"):
assert c["name"] in text
assert f"Hostname {c['ip']}" in text
assert "User root" in text
assert "IdentityFile" in text
assert "StrictHostKeyChecking accept-new" in text
def test_dns(ix, relay_container):
def dig(qname, qtype):
ct = ix.get_dns_container()
return ct.bash(f"dig @127.0.0.1 {qname} {qtype} +short").strip()
domain = relay_container["domain"]
assert dig(domain, "A") == relay_container["ip"]
assert domain in dig(domain, "MX")
assert "587" in dig(f"_submission._tcp.{domain}", "SRV")
class TestLxcStatus:
def test_cli_lxc_status_help(self, cmdeploy):
result = cmdeploy("lxc-status", "--help")
assert result.returncode == 0
assert "status" in result.stdout.lower()
def test_shows_containers(self, lxc_setup, capsys):
class QuietOut:
def red(self, msg, **kw):
pass
def green(self, msg, **kw):
pass
ret = cli.lxc_status_cmd(None, QuietOut())
assert ret == 0
captured = capsys.readouterr().out
assert "ns-localchat" in captured
assert "running" in captured
def test_deploy_freshness(self, ix, monkeypatch):
ct = ix.get_container("x")
monkeypatch.setattr(
"cmdeploy.lxc.incus.RelayContainer.deployed_version",
lambda _self: "abc123def456",
)
monkeypatch.setattr(
"cmdeploy.lxc.incus.RelayContainer.deployed_domain",
lambda _self: ct.domain,
)
monkeypatch.setattr(
"cmdeploy.lxc.cli.get_version_string",
lambda: "abc123def456",
)
assert "IN-SYNC" in cli._deploy_status(ct, "abc123def456", ix)
assert "STALE" in cli._deploy_status(ct, "other_hash_here", ix)
# Hash matches but local has uncommitted changes
monkeypatch.setattr(
"cmdeploy.lxc.cli.get_version_string",
lambda: "abc123def456\ndiff --git a/foo",
)
assert "DIRTY" in cli._deploy_status(ct, "abc123def456", ix)
monkeypatch.setattr(
"cmdeploy.lxc.incus.RelayContainer.deployed_version",
lambda _self: None,
)
assert "NOT DEPLOYED" in cli._deploy_status(ct, "abc123", ix)

View File

@@ -0,0 +1,63 @@
"""Shared utility functions for cmdeploy."""
import subprocess
import textwrap
from pathlib import Path
def _project_root():
"""Return the project root directory."""
return Path(__file__).resolve().parent.parent.parent.parent
def collapse(text):
"""Dedent, join lines, and strip a (triple-quoted) string.
Handy for writing shell commands across multiple lines::
cmd = collapse(f\"""
cmdeploy run
--config {ct.ini}
--ssh-host {ct.domain}
\""")
"""
return textwrap.dedent(text).replace("\n", " ").strip()
def shell(cmd, check=False, **kwargs):
"""Run a shell command string with sensible defaults.
*cmd* is passed through :func:`collapse` first, so callers
can use triple-quoted f-strings freely.
Captures stdout/stderr by default; pass ``capture_output=False``
to stream output to the terminal instead.
"""
if "capture_output" not in kwargs and "stdout" not in kwargs:
kwargs["capture_output"] = True
return subprocess.run(collapse(cmd), shell=True, text=True, check=check, **kwargs)
def get_git_hash():
"""Return the local HEAD commit hash, or None."""
result = shell(
"git rev-parse HEAD",
cwd=str(_project_root()),
)
if result.returncode == 0:
return result.stdout.strip()
return None
def get_version_string():
"""Return ``git_hash\\ngit_diff`` for the local working tree.
Used by :class:`~cmdeploy.deployers.GithashDeployer` to write
``/etc/chatmail-version`` and by ``lxc-status`` to compare
the deployed state against the local checkout.
"""
git_hash = get_git_hash() or "unknown"
try:
git_diff = shell("git diff", cwd=str(_project_root())).stdout
except Exception:
git_diff = ""
return git_hash + "\n" + git_diff

View File

@@ -15,7 +15,7 @@ author = 'chatmail collective'
extensions = [ extensions = [
#'sphinx.ext.autodoc', #'sphinx.ext.autodoc',
#'sphinx.ext.viewdoc', #'sphinx.ext.viewcode',
'sphinxcontrib.mermaid', 'sphinxcontrib.mermaid',
] ]

View File

@@ -16,5 +16,6 @@ Contributions and feedback welcome through the https://github.com/chatmail/relay
proxy proxy
migrate migrate
overview overview
lxc
related related
faq faq

312
doc/source/lxc.rst Normal file
View File

@@ -0,0 +1,312 @@
Local testing with LXC/Incus
============================
.. warning::
cmdeploy LXC support is geared towards local testing and CI, only.
Do not base production setups on it.
The ``cmdeploy`` tool includes support for running
chatmail relays inside local
`Incus <https://linuxcontainers.org/incus/>`_ LXC containers.
This is useful for development, testing, and CI
without requiring a remote server.
LXC system containers behave like lightweight virtual machines.
They share the host's kernel but run their own init system
(systemd), package manager, and network stack,
so the cmdeploy deployment scripts work exactly
as they would on a real Debian server or cloud VPS.
Incus is
`well supported
<https://linuxcontainers.org/incus/docs/main/installing/>`_
on Debian, Ubuntu, Arch, Fedora,
and other major distributions.
Prerequisites
-------------
Install `Incus <https://linuxcontainers.org/incus/>`_
(LXC container manager).
See the `official installation guide
<https://linuxcontainers.org/incus/docs/main/installing/>`_
for full details, or use one of the shortcuts below:
**Debian 12+ / Ubuntu 24.04+**
Incus is in the default repositories::
sudo apt install incus
**Older Debian / Ubuntu**: Use the
`Zabbly package repository
<https://github.com/zabbly/incus>`_::
curl -fsSL https://pkgs.zabbly.com/get/incus-stable | sudo bash
**Arch Linux**::
sudo pacman -S incus
After installing, initialise and grant yourself access::
sudo incus admin init --minimal
sudo usermod -aG incus-admin $USER
.. warning::
You **must log out and back in** (or run ``newgrp incus-admin``)
after adding yourself to the group.
Without this, all ``cmdeploy lxc-*`` commands
will fail with permission errors.
Verify the installation works by running ``incus list``,
which should print an empty table without errors.
Quick start
-----------
::
cd relay
scripts/initenv.sh # bootstrap venv
cmdeploy lxc-test # create containers, deploy, test
The ``lxc-test`` command executes each ``cmdeploy`` subprocess command
so you can copy-paste and run them individually.
No host DNS delegation or ``~/.ssh/config`` changes are needed
because lxc-test passes ssh-related CLI options to "cmdeploy run" and "cmdeploy test" commands.
CLI reference
--------------
``lxc-start [--ipv4-only] [--run] [NAME ...]``
Create and start containers.
Without arguments, creates ``test0-localchat`` and ``ns-localchat`` (DNS).
Pass one or more ``NAME`` arguments to create user relay containers instead
(e.g. ``cmdeploy lxc-start myrelay``).
Use ``--ipv4-only`` to set ``disable_ipv6 = True`` in the generated ``chatmail.ini``,
producing an IPv4-only relay.
Use ``--run`` to automatically run ``cmdeploy run`` on each container after starting it.
Generates ``lxconfigs/ssh-config``.
It reuses existing containers and resets DNS zones to minimal records.
``lxc-stop [--destroy] [--destroy-all] [NAME ...]``
Stop relay containers.
Without arguments, stops ``test0-localchat``
and ``test1-localchat``.
Pass ``NAME`` to stop specific containers.
Use ``--destroy`` to also delete the containers
and their config files.
Use ``--destroy-all`` to additionally destroy
the ``ns-localchat`` DNS container **and** remove
the cached ``localchat-base`` and ``localchat-relay``
images, giving a fully clean slate
for the next ``lxc-test``.
User containers are **never** destroyed
unless named explicitly.
``lxc-test [--one]``
Idempotent full pipeline:
1. ``lxc-start``: create ``test0`` + ``test1``
containers, minimal DNS
2. ``cmdeploy run``: deploy chatmail services on each relay
3. publish ``localchat-relay`` image after first successful deploy
4. ``cmdeploy dns --zonefile``: generate standard
BIND-format zone files, load full DNS records
5. ``cmdeploy test``: run full test suite
with ``-n4 -x``
By default creates, deploys, and tests both ``test0`` and ``test1``
for dual-domain federation testing (sets ``CHATMAIL_DOMAIN2=_test1.localchat``).
test0 runs dual-stack (IPv4 + IPv6) while test1 runs IPv4-only (``disable_ipv6 = True``).
Pass ``--one`` to only deploy and test against ``test0``
(skips ``test1``, does not set ``CHATMAIL_DOMAIN2``).
``lxc-status``
Show live status of all LXC containers (including the DNS container),
deploy freshness (comparing ``/etc/chatmail-version``
against local ``git rev-parse HEAD`` and ``git diff``),
SSH config inclusion, and host DNS forwarding for ``.localchat``.
Reports **IN-SYNC**, **DIRTY** (hash matches but uncommitted changes exist),
**STALE** (different commit), or **NOT DEPLOYED**.
Container types
-----------------
**Test relay containers** (``test0-localchat``, ``test1-localchat``)
Created automatically by ``lxc-test``.
**test0** has IPv4 and IPv6 configured,
**test1** is IPv4-only (``disable_ipv6 = True``).
**User relay containers** (``<name>-localchat``)
Created by ``cmdeploy lxc-start <name>``
where ``<name>`` does not start with ``test``.
These are personal development instances,
never touched by ``lxc-stop --destroy`` unless named explicitly.
**DNS container** (``ns-localchat``)
Singleton container running PowerDNS.
Created automatically when any relay is started.
.. _lxc-ssh-config:
SSH configuration
-----------------
``cmdeploy lxc-start`` generates ``lxconfigs/ssh-config``,
a standard OpenSSH config file mapping every container name,
its domain, and a short alias to the container's IP address::
Host test0-localchat _test0.localchat _test0
Hostname 10.204.0.42
User root
IdentityFile /path/to/relay/lxconfigs/id_localchat
IdentitiesOnly yes
StrictHostKeyChecking accept-new
UserKnownHostsFile /dev/null
LogLevel ERROR
All ``cmdeploy`` commands (``run``, ``dns``, ``status``, ``test``)
accept ``--ssh-config lxconfigs/ssh-config`` to use this file.
``lxc-test`` passes it automatically.
**Using containers from the host shell:**
To make ``ssh _test0`` work from any terminal, add one line to ``~/.ssh/config``::
Include /absolute/path/to/relay/lxconfigs/ssh-config
.. _lxc-dns-setup:
.. _localchat-tld:
``.localchat`` DNS and name resolution
---------------------------------------
All LXC-managed chatmail domains use the ``.localchat`` pseudo-TLD
(e.g. ``_test0.localchat``, ``_test1.localchat``),
a non-delegated suffix that exists only within the local PowerDNS infrastructure.
A dedicated DNS container (``ns-localchat``)
is created so that local test relays interact
with DNS similar to a regular public Internet setup.
On first start, ``cmdeploy lxc-start`` creates this container
running two `PowerDNS <https://www.powerdns.com/>`_ services:
* **pdns-server** (authoritative) serves ``.localchat``
zones from a local SQLite database.
* **pdns-recursor** (recursive) listens on the Incus
bridge so all containers can use it.
Forwards ``.localchat`` queries to the local
authoritative server and everything else to Quad9 (``9.9.9.9``).
After the DNS container is up, ``lxc-start`` configures the Incus bridge
to advertise its IP via DHCP and disables Incus's own DNS.
DNS records are then created in two phases matching the "cmdeploy run" deployment flow:
1. **``lxc-start``** resets each relay zone to
**SOA, NS, and A** records (plus **AAAA** for dual-stack containers).
If host DNS resolution is configured, users can
afterwards run ``cmdeploy run --config lxconfigs/chatmail-test0.ini
--ssh-config lxconfigs/ssh-config --ssh-host _test0.localchat``.
LXC subcommands do not depend on host DNS resolution
and resolve addresses via ``lxconfigs/ssh-config``.
2. **``cmdeploy dns --zonefile``** generates a standard
BIND-format zone file (MX, TXT/SPF, TXT/DMARC,
TXT/MTA-STS, SRV, CNAME, DKIM) and loads it
into PowerDNS.
This two-phase approach prevents premature configuration of mail records
before the relay is actually deployed and running.
Once ``cmdeploy run`` deploys `Unbound <https://nlnetlabs.nl/projects/unbound/>`_
inside a relay container, Unbound has a configuration plugin snippet
that forwards all ``.localchat`` queries to the PowerDNS recursor,
and lets all other queries go through normal recursive resolution.
State outside the repository
-----------------------------
All generated configuration by lxc subcommands live in ``lxconfigs/``
(git-ignored), including the SSH key pair (``id_localchat``),
per-container ``chatmail-*.ini`` files, zone files, and ``ssh-config``.
The only state *outside* the repository is the Incus containers and images themselves
(managed via the ``incus`` CLI, labelled with ``user.localchat-managed=true``).
Two cached images are published to the local Incus image store:
* ``localchat-base``: Debian 12 with openssh-server and Python (built on first run)
* ``localchat-relay``: fully deployed relay snapshot,
published after the first successful ``cmdeploy run``.
Subsequent relay containers launch from this image
so the deploy step is mostly no-ops (roughly 3× faster than a fresh deploy).
Relay containers are limited to **500 MiB RAM** and the DNS container to **100 MiB**.
.. _lxc-tls:
TLS handling and underscore domains
------------------------------------
Container domains start with ``_`` (e.g. ``_test0.localchat``).
As described in :doc:`getting_started` ("Running a relay with self-signed certificates"),
underscore domains automatically use self-signed TLS
and ``smtp_tls_security_level = encrypt``.
This permits cross-relay federation between LXC containers
without any external certificate authority.
Delta Chat clients connecting to these relays
must be configured with
``certificateChecks = acceptInvalidCertificates``
(the test fixtures handle this automatically).
`PR #7926 on chatmail-core <https://github.com/chatmail/core/pull/7926>`_
is meant to make this special setting unnecessary for chatmail clients
that are connecting to underscore domains.
Known limitations
------------------
The LXC environment differs from a production
deployment in several ways:
**No ACME / Let's Encrypt**:
Self-signed TLS only (see :ref:`lxc-tls`);
ACME code paths are never exercised locally.
**No inbound connections from the internet**:
Containers sit on a private Incus bridge and are not port-forwarded.
Only the host and other containers on the same bridge can reach them.
**Local federation only**:
Cross-relay mail delivery (e.g. test0 → test1) works between containers on the same host,
but these relays are invisible to any external mail server.
**DNS is local only**:
The ``.localchat`` pseudo-TLD is not resolvable from the wider internet
(see :ref:`lxc-dns-setup`).
**IPv6 is ULA-only**:
Containers receive IPv6 addresses from the ``fd42:...`` ULA range on the Incus bridge.
These are not globally routable, but are sufficient for testing IPv6 service binding
(Postfix, Dovecot, Nginx) and DNS AAAA records inside the local environment.
test1 runs with ``disable_ipv6 = True`` to exercise the IPv4-only deployment path.
**TURN server does not start**:
``chatmail-turn`` discovers its listen addresses by enumerating globally routable IPs but
LXC containers only have private RFC 1918 addresses (``10.x.x.x``),
so the address list is empty and the server exits immediately.
`PR #11 on chatmail-turn <https://github.com/chatmail/chatmail-turn/pull/11>`_
is meant to fix this.