Compare commits

..

1 Commits

Author SHA1 Message Date
holger krekel
3561ef20e0 feat: expose metadata "maxsmtprecipients" value
also add metadata tests and make metadata lookup method more readable by using structural match/case syntax
2026-05-09 16:10:09 +02:00
26 changed files with 132 additions and 288 deletions

View File

@@ -1,37 +0,0 @@
name: No-DNS
on:
# Triggers when a PR is merged into main or a direct push occurs
push:
branches: [ "main" ]
# Triggers for any PR (and its subsequent commits) targeting the main branch
pull_request:
branches: [ "main" ]
# Newest push wins: Prevents multiple runs from clashing and wasting runner efforts
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
no-dns:
name: LXC deploy and test
uses: chatmail/cmlxc/.github/workflows/lxc-test.yml@v0.14.1
with:
cmlxc_commands: |
cmlxc init
# single cmdeploy relay test
cmlxc -v deploy-cmdeploy --source ./repo --type ipv4 cm0
cmlxc -v test-cmdeploy cm0
# cross cmdeploy relay test (two ipv4 relays)
cmlxc -v deploy-cmdeploy --source ./repo --ipv4-only --type ipv4 cm1
cmlxc -v test-cmdeploy cm0 cm1
# cross cmdeploy/madmail relay tests
cmlxc -v deploy-madmail mad0
cmlxc -v test-cmdeploy cm0 mad0
cmlxc -v test-mini mad0 cm0
cmlxc -v test-mini cm0 mad0

View File

@@ -1,4 +1,4 @@
name: CI
name: Run unit-tests and container-based deploy+test verification
on:
# Triggers when a PR is merged into main or a direct push occurs
@@ -57,8 +57,9 @@ jobs:
lxc-test:
name: LXC deploy and test
uses: chatmail/cmlxc/.github/workflows/lxc-test.yml@v0.14.1
uses: chatmail/cmlxc/.github/workflows/lxc-test.yml@v0.13.5
with:
cmlxc_version: v0.13.5
cmlxc_commands: |
cmlxc init
# single cmdeploy relay test
@@ -75,4 +76,3 @@ jobs:
cmlxc -v test-cmdeploy cm0 mad0
cmlxc -v test-mini cm0 mad0
cmlxc -v test-mini mad0 cm0

View File

@@ -10,7 +10,6 @@ dependencies = [
"filelock",
"requests",
"crypt-r >= 3.13.1 ; python_version >= '3.11'",
"domain-validator",
]
[tool.setuptools]

View File

@@ -1,8 +1,6 @@
import ipaddress
from pathlib import Path
import iniconfig
from domain_validator import DomainValidator
from chatmaild.user import User
@@ -21,19 +19,7 @@ def read_config(inipath):
class Config:
def __init__(self, inipath, params):
self._inipath = inipath
raw_domain = params["mail_domain"]
self.mail_domain_bare = raw_domain
if is_valid_ipv4(raw_domain):
self.ipv4_relay = raw_domain
self.mail_domain = f"[{raw_domain}]"
self.postfix_myhostname = ipaddress.IPv4Address(raw_domain).reverse_pointer
else:
DomainValidator().validate_domain_re(raw_domain)
self.ipv4_relay = None
self.mail_domain = raw_domain
self.postfix_myhostname = raw_domain
self.mail_domain = params["mail_domain"]
self.max_user_send_per_minute = int(params.get("max_user_send_per_minute", 60))
self.max_user_send_burst_size = int(params.get("max_user_send_burst_size", 10))
self.max_mailbox_size = params["max_mailbox_size"]
@@ -67,7 +53,7 @@ class Config:
self.imap_rawlog = params.get("imap_rawlog", "false").lower() == "true"
self.imap_compress = params.get("imap_compress", "false").lower() == "true"
if "iroh_relay" not in params:
self.iroh_relay = "https://" + raw_domain
self.iroh_relay = "https://" + params["mail_domain"]
self.enable_iroh_relay = True
else:
self.iroh_relay = params["iroh_relay"].strip()
@@ -93,17 +79,17 @@ class Config:
)
self.tls_cert_mode = "external"
self.tls_cert_path, self.tls_key_path = parts
elif raw_domain.startswith("_") or self.ipv4_relay:
elif self.mail_domain.startswith("_"):
self.tls_cert_mode = "self"
self.tls_cert_path = "/etc/ssl/certs/mailserver.pem"
self.tls_key_path = "/etc/ssl/private/mailserver.key"
else:
self.tls_cert_mode = "acme"
self.tls_cert_path = f"/var/lib/acme/live/{raw_domain}/fullchain"
self.tls_key_path = f"/var/lib/acme/live/{raw_domain}/privkey"
self.tls_cert_path = f"/var/lib/acme/live/{self.mail_domain}/fullchain"
self.tls_key_path = f"/var/lib/acme/live/{self.mail_domain}/privkey"
# deprecated option
mbdir = params.get("mailboxes_dir", f"/home/vmail/mail/{raw_domain}")
mbdir = params.get("mailboxes_dir", f"/home/vmail/mail/{self.mail_domain}")
self.mailboxes_dir = Path(mbdir.strip())
# old unused option (except for first migration from sqlite to maildir store)
@@ -189,19 +175,3 @@ def get_default_config_content(mail_domain, **overrides):
lines.append(line)
content = "\n".join(lines)
return content
def is_valid_ipv4(address: str) -> bool:
"""Check if a mail_domain is an IPv4 address."""
try:
ipaddress.IPv4Address(address)
return True
except ValueError:
return False
def format_mail_domain(raw_domain: str) -> str:
if is_valid_ipv4(raw_domain):
return f"[{raw_domain}]"
DomainValidator().validate_domain_re(raw_domain)
return raw_domain

View File

@@ -88,29 +88,27 @@ class MetadataDictProxy(DictProxy):
def handle_lookup(self, parts):
# Lpriv/43f5f508a7ea0366dff30200c15250e3/devicetoken\tlkj123poi@c2.testrun.org
keyparts = parts[0].split("/", 2)
if keyparts[0] == "priv":
keyname = keyparts[2]
addr = parts[1]
if keyname == self.metadata.DEVICETOKEN_KEY:
match parts[0].split("/", 2):
case ["priv", _, keyname] if keyname == self.metadata.DEVICETOKEN_KEY:
addr = parts[1]
res = " ".join(self.metadata.get_tokens_for_addr(addr))
return f"O{res}\n"
elif keyparts[0] == "shared":
keyname = keyparts[2]
if (
keyname == "vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay"
and self.iroh_relay
):
# Handle `GETMETADATA "" /shared/vendor/deltachat/irohrelay`
return f"O{self.iroh_relay}\n"
elif keyname == "vendor/vendor.dovecot/pvt/server/vendor/deltachat/turn":
try:
res = turn_credentials()
except Exception:
logging.exception("failed to get TURN credentials")
return "N\n"
port = 3478
return f"O{self.turn_hostname}:{port}:{res}\n"
case ["shared", _, keyname]:
prefix = "vendor/vendor.dovecot/pvt/server/vendor/deltachat/"
if keyname.startswith(prefix):
match keyname[len(prefix) :]:
case "irohrelay" if self.iroh_relay:
return f"O{self.iroh_relay}\n"
case "turn":
try:
res = turn_credentials()
except Exception:
logging.exception("failed to get TURN credentials")
return "N\n"
return f"O{self.turn_hostname}:3478:{res}\n"
case "maxsmtprecipients":
# postfix default (see "postconf smtpd_recipient_limit")
return "O1000\n"
logging.warning(f"lookup ignored: {parts!r}")
return "N\n"
@@ -120,12 +118,13 @@ class MetadataDictProxy(DictProxy):
# https://github.com/dovecot/core/blob/main/src/lib-storage/mailbox-attribute.h
keyname = parts[1].split("/")
value = parts[2] if len(parts) > 2 else ""
if keyname[0] == "priv" and keyname[2] == self.metadata.DEVICETOKEN_KEY:
self.metadata.add_token_to_addr(addr, value)
return True
elif keyname[0] == "priv" and keyname[2] == "messagenew":
self.notifier.new_message_for_addr(addr, self.metadata)
return True
match keyname:
case ["priv", _, key] if key == self.metadata.DEVICETOKEN_KEY:
self.metadata.add_token_to_addr(addr, value)
return True
case ["priv", _, "messagenew"]:
self.notifier.new_message_for_addr(addr, self.metadata)
return True
return False

View File

@@ -2,6 +2,7 @@
"""CGI script for creating new accounts."""
import ipaddress
import json
import secrets
import string
@@ -14,6 +15,16 @@ ALPHANUMERIC = string.ascii_lowercase + string.digits
ALPHANUMERIC_PUNCT = string.ascii_letters + string.digits + string.punctuation
def wrap_ip(host):
if host.startswith("[") and host.endswith("]"):
return host
try:
ipaddress.ip_address(host)
return f"[{host}]"
except ValueError:
return host
def create_newemail_dict(config: Config):
user = "".join(
secrets.choice(ALPHANUMERIC) for _ in range(config.username_max_length)
@@ -22,22 +33,16 @@ def create_newemail_dict(config: Config):
secrets.choice(ALPHANUMERIC_PUNCT)
for _ in range(config.password_min_length + 3)
)
return dict(email=f"{user}@{config.mail_domain}", password=f"{password}")
return dict(email=f"{user}@{wrap_ip(config.mail_domain)}", password=f"{password}")
def create_dclogin_url(config, email, password):
def create_dclogin_url(email, password):
"""Build a dclogin: URL with credentials and self-signed cert acceptance.
Uses ic=3 (AcceptInvalidCertificates) so chatmail clients
can connect to servers with self-signed TLS certificates.
"""
if config.ipv4_relay:
imap_host = "&ih=" + config.ipv4_relay
smtp_host = "&sh=" + config.ipv4_relay
else:
imap_host = ""
smtp_host = ""
return f"dclogin:{quote(email, safe='@[]')}?p={quote(password, safe='')}&v=1{imap_host}{smtp_host}&ic=3"
return f"dclogin:{quote(email, safe='@')}?p={quote(password, safe='')}&v=1&ic=3"
def print_new_account():
@@ -46,9 +51,7 @@ def print_new_account():
result = dict(email=creds["email"], password=creds["password"])
if config.tls_cert_mode == "self":
result["dclogin_url"] = create_dclogin_url(
config, creds["email"], creds["password"]
)
result["dclogin_url"] = create_dclogin_url(creds["email"], creds["password"])
print("Content-Type: application/json")
print("")

View File

@@ -31,11 +31,6 @@ def example_config(make_config):
return make_config("chat.example.org")
@pytest.fixture
def ipv4_config(make_config):
return make_config("1.3.3.7")
@pytest.fixture
def maildomain(example_config):
return example_config.mail_domain

View File

@@ -1,13 +1,6 @@
from contextlib import nullcontext as does_not_raise
import pytest
from chatmaild.config import (
format_mail_domain,
is_valid_ipv4,
parse_size_mb,
read_config,
)
from chatmaild.config import parse_size_mb, read_config
def test_read_config_basic(example_config):
@@ -20,12 +13,6 @@ def test_read_config_basic(example_config):
example_config = read_config(inipath)
assert example_config.max_user_send_per_minute == 37
assert example_config.mail_domain == "chat.example.org"
assert example_config.ipv4_relay is None
def test_read_config_ipv4(ipv4_config):
assert ipv4_config.ipv4_relay == "1.3.3.7"
assert ipv4_config.mail_domain == "[1.3.3.7]"
def test_read_config_basic_using_defaults(tmp_path, maildomain):
@@ -148,31 +135,3 @@ def test_max_mailbox_size_mb(make_config):
config = make_config("chat.example.org")
assert config.max_mailbox_size == "500M"
assert config.max_mailbox_size_mb == 500
@pytest.mark.parametrize(
["input", "result"],
[
("example.org", False),
("1.3.3.7", True),
("fe::1", False),
("ad.1e.dag.adf", False),
("12394142", False),
],
)
def test_is_valid_ipv4(input, result):
assert result == is_valid_ipv4(input)
@pytest.mark.parametrize(
["input", "result", "exception"],
[
("example.org", "example.org", does_not_raise()),
("1.3.3.7", "[1.3.3.7]", does_not_raise()),
("fe::1", None, pytest.raises(ValueError)),
("12394142", None, pytest.raises(ValueError)),
],
)
def test_format_mail_domain(input, result, exception):
with exception:
assert result == format_mail_domain(input)

View File

@@ -360,15 +360,8 @@ def test_turn_credentials_success(notifier, metadata, monkeypatch):
def test_iroh_relay(dictproxy):
rfile = io.BytesIO(
b"\n".join(
[
b"H",
b"Lshared/0123/vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay\tuser@example.org",
]
)
)
wfile = io.BytesIO()
key = b"Lshared/0123/vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay\tuser@example.org"
rfile, wfile = io.BytesIO(b"H\n" + key), io.BytesIO()
dictproxy.iroh_relay = "https://example.org/"
dictproxy.loop_forever(rfile, wfile)
assert wfile.getvalue() == b"Ohttps://example.org/\n"
@@ -383,3 +376,23 @@ def test_legacy_token_migration(metadata, testaddr):
tokens = mdict[metadata.DEVICETOKEN_KEY]
assert isinstance(tokens, dict)
assert "oldtoken1" in tokens and "oldtoken2" in tokens
@pytest.mark.parametrize(
"suffix, expected",
[
(b"vendor/deltachat/maxsmtprecipients", b"O1000\n"),
(b"wrong/prefix/key", b"N\n"),
(b"vendor/deltachat/unknown", b"N\n"),
],
ids=["maxsmtprecipients", "prefix_mismatch", "unknown_name"],
)
def test_shared_lookup(dictproxy, suffix, expected):
key = (
b"Lshared/0123/vendor/vendor.dovecot/pvt/server/"
+ suffix
+ b"\tuser@example.org"
)
rfile, wfile = io.BytesIO(b"H\n" + key), io.BytesIO()
dictproxy.loop_forever(rfile, wfile)
assert wfile.getvalue() == expected

View File

@@ -19,35 +19,24 @@ def test_create_newemail_dict(example_config):
assert ac1["password"] != ac2["password"]
def test_create_newemail_dict_ip(ipv4_config):
ac = create_newemail_dict(ipv4_config)
assert ac["email"].endswith("@[1.3.3.7]")
def test_create_newemail_dict_ip(make_config):
config = make_config("1.2.3.4")
ac = create_newemail_dict(config)
assert ac["email"].endswith("@[1.2.3.4]")
def test_create_dclogin_url(example_config):
addr = "user@example.org"
password = "p@ss w+rd"
url = create_dclogin_url(example_config, addr, password)
def test_create_dclogin_url():
url = create_dclogin_url("user@example.org", "p@ss w+rd")
assert url.startswith("dclogin:")
assert "v=1" in url
assert "ic=3" in url
assert addr in url
assert "user@example.org" in url
# password special chars must be encoded
assert "p%40ss" in url
assert "w%2Brd" in url
def test_create_dclogin_url_ipv4(ipv4_config):
addr = "user@[1.3.3.7]"
password = "p@ss w+rd"
url = create_dclogin_url(ipv4_config, addr, password)
assert url.startswith("dclogin:")
assert "v=1" in url
assert "ic=3" in url
assert addr in url
def test_print_new_account(capsys, monkeypatch, maildomain, tmpdir, example_config):
monkeypatch.setattr(chatmaild.newemail, "CONFIG_PATH", str(example_config._inipath))
print_new_account()

View File

@@ -87,12 +87,10 @@ def run_cmd_options(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_bare
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
sshexec = get_sshexec(ssh_host)
require_iroh = args.config.enable_iroh_relay
strict_tls = args.config.tls_cert_mode == "acme"
if args.config.ipv4_relay:
args.dns_check_disabled = True
if not args.dns_check_disabled:
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):
@@ -121,8 +119,6 @@ def run_cmd(args, out):
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("Run 'cmdeploy run' again")
elif args.config.ipv4_relay:
out.green("Deploy completed.")
else:
out.green("Deploy completed, call `cmdeploy dns` next.")
return 0
@@ -144,10 +140,6 @@ def dns_cmd_options(parser):
def dns_cmd(args, out):
"""Check DNS entries and optionally generate dns zone file."""
if args.config.ipv4_relay:
ipv4 = args.config.ipv4_relay
print(f"[WARNING] {ipv4} is not a domain, skipping DNS checks.")
return 0
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
sshexec = get_sshexec(ssh_host, verbose=args.verbose)
tls_cert_mode = args.config.tls_cert_mode
@@ -185,7 +177,7 @@ def status_cmd_options(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_bare
ssh_host = args.ssh_host if args.ssh_host else args.config.mail_domain
sshexec = get_sshexec(ssh_host, verbose=args.verbose)
out.green(f"chatmail domain: {args.config.mail_domain}")

View File

@@ -370,7 +370,7 @@ class ChatmailVenvDeployer(Deployer):
def configure(self):
_configure_remote_venv_with_chatmaild(self, self.config)
configure_remote_units(self, self.config.mail_domain_bare, self.units)
configure_remote_units(self, self.config.mail_domain, self.units)
def activate(self):
activate_remote_units(self, self.units)
@@ -469,7 +469,7 @@ def deploy_chatmail(config_path: Path, disable_mail: bool, website_only: bool) -
"""
config = read_config(config_path)
check_config(config)
bare_host = config.mail_domain_bare
mail_domain = config.mail_domain
if website_only:
Deployment().perform_stages([WebsiteDeployer(config)])
@@ -526,7 +526,7 @@ def deploy_chatmail(config_path: Path, disable_mail: bool, website_only: bool) -
)
exit(1)
tls_deployer = get_tls_deployer(config, bare_host)
tls_deployer = get_tls_deployer(config, mail_domain)
all_deployers = [
ChatmailDeployer(config),
@@ -534,13 +534,13 @@ def deploy_chatmail(config_path: Path, disable_mail: bool, website_only: bool) -
FiltermailDeployer(),
JournaldDeployer(),
UnboundDeployer(config),
TurnDeployer(bare_host),
TurnDeployer(mail_domain),
IrohDeployer(config.enable_iroh_relay),
tls_deployer,
WebsiteDeployer(config),
ChatmailVenvDeployer(config),
MtastsDeployer(),
*([] if config.ipv4_relay else [OpendkimDeployer(bare_host)]),
OpendkimDeployer(mail_domain),
# Dovecot should be started before Postfix
# because it creates authentication socket
# required by Postfix.

View File

@@ -68,7 +68,7 @@ class DovecotDeployer(Deployer):
)
def configure(self):
configure_remote_units(self, self.config.mail_domain_bare, self.units)
configure_remote_units(self, self.config.mail_domain, self.units)
_configure_dovecot(self, self.config)
def activate(self):

View File

@@ -7,7 +7,6 @@ listen = 0.0.0.0
protocols = imap lmtp
auth_mechanisms = plain
auth_username_chars = abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ01234567890.-_@[]
{% if debug == true %}
auth_verbose = yes

View File

@@ -1,3 +1,4 @@
from pyinfra import host
from pyinfra.facts.files import File
@@ -20,8 +21,8 @@ class ExternalTlsDeployer(Deployer):
def configure(self):
# Verify cert and key exist on the remote host using pyinfra facts.
for path in (self.cert_path, self.key_path):
if host.get_fact(File, path=path) is None:
raise Exception(f"External TLS file not found on server: {path}")
if host.get_fact(File, path=path) is None:
raise Exception(f"External TLS file not found on server: {path}")
self.ensure_systemd_unit(
"external/tls-cert-reload.path.j2",
@@ -39,3 +40,5 @@ class ExternalTlsDeployer(Deployer):
running=True,
enabled=True,
)

View File

@@ -54,16 +54,14 @@ smtpd_tls_exclude_ciphers = aNULL, RC4, MD5, DES
tls_preempt_cipherlist = yes
smtpd_relay_restrictions = permit_mynetworks permit_sasl_authenticated defer_unauth_destination
myhostname = {{ config.postfix_myhostname }}
myhostname = {{ config.mail_domain }}
alias_maps = hash:/etc/aliases
alias_database = hash:/etc/aliases
# When postfix receives mail for $mydestination,
# it hands it over to dovecot via $local_transport.
mydestination = {{ config.mail_domain }}
local_transport = lmtp:unix:private/dovecot-lmtp
# postfix doesn't check whether local users exist or not:
local_recipient_maps =
# Postfix does not deliver mail for any domain by itself.
# Primary domain is listed in `virtual_mailbox_domains` instead
# and handed over to Dovecot.
mydestination =
relayhost =
{% if disable_ipv6 %}
@@ -81,6 +79,8 @@ inet_protocols = ipv4
inet_protocols = all
{% endif %}
virtual_transport = lmtp:unix:private/dovecot-lmtp
virtual_mailbox_domains = {{ config.mail_domain }}
lmtp_header_checks = regexp:/etc/postfix/lmtp_header_cleanup
mua_client_restrictions = permit_sasl_authenticated, reject

View File

@@ -80,9 +80,8 @@ filter unix - n n - - lmtp
127.0.0.1:{{ config.postfix_reinject_port }} inet n - n - 100 smtpd
-o syslog_name=postfix/reinject
-o milter_macro_daemon_name=ORIGINATING
-o smtpd_milters=unix:opendkim/opendkim.sock
-o cleanup_service_name=authclean
{% if not config.ipv4_relay %} -o smtpd_milters=unix:opendkim/opendkim.sock
{% endif %}
# Local SMTP server for reinjecting incoming filtered mail
127.0.0.1:{{ config.postfix_reinject_port_incoming }} inet n - n - 100 smtpd

View File

@@ -12,7 +12,7 @@ def test_init(tmp_path, maildomain):
inipath = tmp_path.joinpath("chatmail.ini")
main(["init", "--config", str(inipath), maildomain])
config = read_config(inipath)
assert config.mail_domain_bare == maildomain
assert config.mail_domain == maildomain
def test_capabilities(imap):
@@ -89,11 +89,12 @@ def test_concurrent_logins_same_account(
assert login_results.get()
def test_no_vrfy(cmfactory, chatmail_config, maildomain):
def test_no_vrfy(cmfactory, chatmail_config):
ac = cmfactory.get_online_account()
addr = ac.get_config("addr")
domain = chatmail_config.mail_domain
s = smtplib.SMTP(maildomain)
s = smtplib.SMTP(domain)
s.starttls()
s.putcmd("vrfy", f"wrongaddress@{chatmail_config.mail_domain}")

View File

@@ -5,7 +5,6 @@ import subprocess
import time
import pytest
from chatmaild.config import is_valid_ipv4
from cmdeploy import remote
from cmdeploy.cmdeploy import get_sshexec
@@ -22,8 +21,6 @@ class TestSSHExecutor:
assert out == out2
def test_perform_initial(self, sshexec, maildomain):
if is_valid_ipv4(maildomain):
pytest.skip(f"{maildomain} is not a domain")
res = sshexec(
remote.rdns.perform_initial_checks, kwargs=dict(mail_domain=maildomain)
)

View File

@@ -15,7 +15,7 @@ def imap_mailbox(cmfactory, ssl_context):
(ac1,) = cmfactory.get_online_accounts(1)
user = ac1.get_config("addr")
password = ac1.get_config("mail_pw")
host = user.split("@")[1].strip("[").strip("]")
host = user.split("@")[1]
mailbox = imap_tools.MailBox(host, ssl_context=ssl_context)
mailbox.login(user, password)
mailbox.dc_ac = ac1
@@ -178,7 +178,7 @@ def test_hide_senders_ip_address(cmfactory, ssl_context):
chat.send_text("testing submission header cleanup")
user2.wait_for_incoming_msg()
addr = user2.get_config("addr")
host = addr.split("@")[1].strip("[").strip("]")
host = addr.split("@")[1]
pw = user2.get_config("mail_pw")
mailbox = imap_tools.MailBox(host, ssl_context=ssl_context)
mailbox.login(addr, pw)

View File

@@ -1,4 +1,5 @@
import imaplib
import ipaddress
import itertools
import os
import random
@@ -9,11 +10,19 @@ import time
from pathlib import Path
import pytest
from chatmaild.config import format_mail_domain, is_valid_ipv4, read_config
from chatmaild.config import read_config
conftestdir = Path(__file__).parent
def _is_ip(domain):
try:
ipaddress.ip_address(domain)
return True
except ValueError:
return False
def pytest_configure(config):
config._benchresults = {}
config.addinivalue_line(
@@ -49,12 +58,7 @@ def chatmail_config(pytestconfig):
@pytest.fixture(scope="session")
def maildomain(chatmail_config):
return chatmail_config.mail_domain_bare
@pytest.fixture(scope="session")
def maildomain_deliverable(maildomain):
return format_mail_domain(maildomain)
return chatmail_config.mail_domain
@pytest.fixture(scope="session")
@@ -274,6 +278,7 @@ def gencreds(chatmail_config):
def gen(domain=None):
domain = domain if domain else chatmail_config.mail_domain
addr_domain = f"[{domain}]" if _is_ip(domain) else domain
while 1:
num = next(count)
alphanumeric = "abcdefghijklmnopqrstuvwxyz1234567890"
@@ -287,7 +292,7 @@ def gencreds(chatmail_config):
password = "".join(
random.choices(alphanumeric, k=chatmail_config.password_min_length)
)
yield f"{user}@{domain}", f"{password}"
yield f"{user}@{addr_domain}", f"{password}"
return lambda domain=None: next(gen(domain))
@@ -312,8 +317,7 @@ class ChatmailACFactory:
def _make_transport(self, domain):
"""Build a transport config dict for the given domain."""
domain_deliverable = format_mail_domain(domain)
addr, password = self.gencreds(domain_deliverable)
addr, password = self.gencreds(domain)
transport = {
"addr": addr,
"password": password,
@@ -322,7 +326,7 @@ class ChatmailACFactory:
"imapServer": domain,
"smtpServer": domain,
}
if domain.startswith("_") or is_valid_ipv4(domain):
if self.chatmail_config.tls_cert_mode == "self":
transport["certificateChecks"] = "acceptInvalidCertificates"
return transport
@@ -337,9 +341,8 @@ class ChatmailACFactory:
accounts = []
for _ in range(num):
account = self.dc.add_account()
domain_deliverable = format_mail_domain(domain)
addr, password = self.gencreds(domain_deliverable)
if is_valid_ipv4(domain):
addr, password = self.gencreds(domain)
if _is_ip(domain):
# Use DCLOGIN scheme with explicit server hosts,
# matching how madmail presents its addresses to users.
qr = (
@@ -413,10 +416,10 @@ class Remote:
def iter_output(self, logcmd="", ready=None):
getjournal = "journalctl -f" if not logcmd else logcmd
print(self.sshdomain)
if self.sshdomain in ("@local", "localhost"):
command = []
else:
command = ["ssh", f"root@{self.sshdomain}"]
match self.sshdomain:
case "@local": command = []
case "localhost": command = []
case _: command = ["ssh", f"root@{self.sshdomain}"]
[command.append(arg) for arg in getjournal.split()]
popen = subprocess.Popen(
command,

View File

@@ -39,14 +39,6 @@ class TestCmdline:
out, err = capsys.readouterr()
assert "deleting config file" in out.lower()
def test_dns_skip_on_ip(self, capsys, tmp_path, monkeypatch):
monkeypatch.delenv("CHATMAIL_INI", raising=False)
inipath = tmp_path / "chatmail.ini"
assert main(["init", "--config", str(inipath), "1.3.3.7"]) == 0
assert main(["dns", "--config", str(inipath)]) == 0
out, err = capsys.readouterr()
assert out == "[WARNING] 1.3.3.7 is not a domain, skipping DNS checks.\n"
def test_www_folder(example_config, tmp_path):
reporoot = importlib.resources.files(__package__).joinpath("../../../../").resolve()

View File

@@ -23,7 +23,8 @@ def make_host(*fact_pairs):
if cls not in facts:
registered = ", ".join(c.__name__ for c in facts)
raise LookupError(
f"unexpected get_fact({cls.__name__}); only registered: {registered}"
f"unexpected get_fact({cls.__name__}); "
f"only registered: {registered}"
)
return facts[cls]

View File

@@ -14,6 +14,8 @@ Minimal requirements and prerequisites
You will need the following:
- Control over a domain through a DNS provider of your choice.
- A Debian 12 **deployment server** with reachable SMTP/SUBMISSIONS/IMAPS/HTTPS ports.
IPv6 is encouraged if available. Chatmail relay servers only require
1GB RAM, one CPU, and perhaps 10GB storage for a few thousand active
@@ -26,11 +28,6 @@ You will need the following:
(An ed25519 private key is required due to an `upstream bug in
paramiko <https://github.com/paramiko/paramiko/issues/2191>`_)
- Control over a domain through a DNS provider of your choice
(there is experimental support for :ref:`DNS-less relays <iponly>`).
.. _setup:
Setup with ``scripts/cmdeploy``
-------------------------------------

View File

@@ -19,4 +19,3 @@ Contributions and feedback welcome through the https://github.com/chatmail/relay
reverse_dns
related
faq
iponly

View File

@@ -1,29 +0,0 @@
.. _iponly:
Hosting without DNS records
===========================
.. note::
This option is experimental and might change without notice.
In case you don't have a domain,
for example in a local network,
you can run a chatmail relay with only an IPv4 address as well.
To deploy a relay without a domain,
run ``cmdeploy init`` with only the IPv4 address
during the :ref:`installation steps <setup>`,
for example ``cmdeploy init 13.12.23.42``.
Drawbacks
---------
- your transport encryption will only use self-signed TLS certificates,
which are vulnerable against MITM attacks.
the chatmail core's end-to-end encryption should suffice in most scenarios though.
- your messages will not be DKIM-signed;
experimentally, most chatmail relays accept non-DKIM-signed messages from IPv4-only relays,
but some relays might not accept messages from yours.