mirror of
https://github.com/chatmail/relay.git
synced 2026-05-10 16:04:37 +00:00
introduce "mailboxes_dir" config ini option to avoid hardcoding /home/vmail/mail/....
in source code and to improve testability.
This commit is contained in:
@@ -3,17 +3,15 @@ from pathlib import Path
|
||||
import iniconfig
|
||||
|
||||
|
||||
def read_config(inipath, mail_basedir=None):
|
||||
def read_config(inipath):
|
||||
assert Path(inipath).exists(), inipath
|
||||
cfg = iniconfig.IniConfig(inipath)
|
||||
params = cfg.sections["params"]
|
||||
if mail_basedir is None:
|
||||
mail_basedir = Path(f"/home/vmail/mail/{params['mail_domain']}")
|
||||
return Config(inipath, params=params, mail_basedir=mail_basedir)
|
||||
return Config(inipath, params=params)
|
||||
|
||||
|
||||
class Config:
|
||||
def __init__(self, inipath, params, mail_basedir: Path):
|
||||
def __init__(self, inipath, params):
|
||||
self._inipath = inipath
|
||||
self.mail_domain = params["mail_domain"]
|
||||
self.max_user_send_per_minute = int(params["max_user_send_per_minute"])
|
||||
@@ -25,6 +23,7 @@ class Config:
|
||||
self.password_min_length = int(params["password_min_length"])
|
||||
self.passthrough_senders = params["passthrough_senders"].split()
|
||||
self.passthrough_recipients = params["passthrough_recipients"].split()
|
||||
self.mailboxes_dir = params["mailboxes_dir"].strip().rstrip("/")
|
||||
self.filtermail_smtp_port = int(params["filtermail_smtp_port"])
|
||||
self.postfix_reinject_port = int(params["postfix_reinject_port"])
|
||||
self.iroh_relay = params.get("iroh_relay")
|
||||
@@ -32,26 +31,40 @@ class Config:
|
||||
self.privacy_mail = params.get("privacy_mail")
|
||||
self.privacy_pdo = params.get("privacy_pdo")
|
||||
self.privacy_supervisor = params.get("privacy_supervisor")
|
||||
self.mail_basedir = mail_basedir
|
||||
|
||||
def _getbytefile(self):
|
||||
return open(self._inipath, "rb")
|
||||
|
||||
def get_user_maildir(self, addr):
|
||||
if addr and addr != "." and "/" not in addr:
|
||||
res = self.mail_basedir.joinpath(addr).resolve()
|
||||
if res.is_relative_to(self.mail_basedir):
|
||||
return res
|
||||
res = Path(self.mailboxes_dir).joinpath(addr).resolve()
|
||||
if res.is_relative_to(self.mailboxes_dir):
|
||||
return str(res)
|
||||
raise ValueError(f"invalid address {addr!r}")
|
||||
|
||||
|
||||
def write_initial_config(inipath, mail_domain):
|
||||
def write_initial_config(inipath, mail_domain, **overrides):
|
||||
"""Write out default config file, using the specified config value overrides."""
|
||||
from importlib.resources import files
|
||||
|
||||
inidir = files(__package__).joinpath("ini")
|
||||
content = (
|
||||
inidir.joinpath("chatmail.ini.f").read_text().format(mail_domain=mail_domain)
|
||||
)
|
||||
source_inipath = inidir.joinpath("chatmail.ini.f")
|
||||
content = source_inipath.read_text().format(mail_domain=mail_domain)
|
||||
|
||||
# apply config overrides
|
||||
newlines = []
|
||||
for line in content.split("\n"):
|
||||
newline = line.strip()
|
||||
if newline and newline[0] not in "#[":
|
||||
name, value = newline.split(" =", maxsplit=1)
|
||||
value = overrides.get(name.strip(), value.strip())
|
||||
newline = f"{name.strip()} = {value.strip()}"
|
||||
newlines.append(newline)
|
||||
|
||||
content = "\n".join(newlines)
|
||||
|
||||
# apply testrun privacy overrides
|
||||
|
||||
if mail_domain.endswith(".testrun.org"):
|
||||
override_inipath = inidir.joinpath("override-testrun.ini")
|
||||
privacy = iniconfig.IniConfig(override_inipath)["privacy"]
|
||||
|
||||
@@ -68,7 +68,7 @@ def is_allowed_to_create(config: Config, user, cleartext_password) -> bool:
|
||||
def get_user_data(db, config: Config, user):
|
||||
if user == f"echo@{config.mail_domain}":
|
||||
return dict(
|
||||
home=f"/home/vmail/mail/{config.mail_domain}/echo@{config.mail_domain}",
|
||||
home=config.get_user_maildir(user),
|
||||
uid="vmail",
|
||||
gid="vmail",
|
||||
)
|
||||
@@ -76,7 +76,7 @@ def get_user_data(db, config: Config, user):
|
||||
with db.read_connection() as conn:
|
||||
result = conn.get_user(user)
|
||||
if result:
|
||||
result["home"] = f"/home/vmail/mail/{config.mail_domain}/{user}"
|
||||
result["home"] = config.get_user_maildir(user)
|
||||
result["uid"] = "vmail"
|
||||
result["gid"] = "vmail"
|
||||
return result
|
||||
@@ -96,7 +96,7 @@ def lookup_passdb(db, config: Config, user, cleartext_password, last_login=None)
|
||||
return None
|
||||
|
||||
return dict(
|
||||
home=f"/home/vmail/mail/{config.mail_domain}/echo@{config.mail_domain}",
|
||||
home=config.get_user_maildir(user),
|
||||
uid="vmail",
|
||||
gid="vmail",
|
||||
password=encrypt_password(password),
|
||||
@@ -114,7 +114,7 @@ def lookup_passdb(db, config: Config, user, cleartext_password, last_login=None)
|
||||
"UPDATE users SET last_login=? WHERE addr=?", (last_login, user)
|
||||
)
|
||||
|
||||
userdata["home"] = f"/home/vmail/mail/{config.mail_domain}/{user}"
|
||||
userdata["home"] = config.get_user_maildir(user)
|
||||
userdata["uid"] = "vmail"
|
||||
userdata["gid"] = "vmail"
|
||||
return userdata
|
||||
@@ -127,7 +127,7 @@ def lookup_passdb(db, config: Config, user, cleartext_password, last_login=None)
|
||||
conn.execute(q, (user, encrypted_password, last_login))
|
||||
print(f"Created address: {user}", file=sys.stderr)
|
||||
return dict(
|
||||
home=f"/home/vmail/mail/{config.mail_domain}/{user}",
|
||||
home=config.get_user_maildir(user),
|
||||
uid="vmail",
|
||||
gid="vmail",
|
||||
password=encrypted_password,
|
||||
|
||||
@@ -42,6 +42,9 @@ passthrough_recipients = xstore@testrun.org groupsbot@hispanilandia.net
|
||||
# Deployment Details
|
||||
#
|
||||
|
||||
# Directory where user mailboxes are stored
|
||||
mailboxes_dir = /home/vmail/mail/{mail_domain}
|
||||
|
||||
# where the filtermail SMTP service listens
|
||||
filtermail_smtp_port = 10080
|
||||
|
||||
@@ -63,4 +66,3 @@ privacy_pdo =
|
||||
|
||||
# postal address of the privacy supervisor
|
||||
privacy_supervisor =
|
||||
|
||||
|
||||
@@ -16,10 +16,11 @@ def make_config(tmp_path):
|
||||
inipath = tmp_path.joinpath("chatmail.ini")
|
||||
|
||||
def make_conf(mail_domain):
|
||||
write_initial_config(inipath, mail_domain=mail_domain)
|
||||
basedir = tmp_path.joinpath(f"vmail/{mail_domain}")
|
||||
basedir.mkdir(parents=True, exist_ok=True)
|
||||
return read_config(inipath, mail_basedir=basedir)
|
||||
overrides = dict(mailboxes_dir=str(basedir))
|
||||
write_initial_config(inipath, mail_domain=mail_domain, **overrides)
|
||||
return read_config(inipath)
|
||||
|
||||
return make_conf
|
||||
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from chatmaild.config import read_config
|
||||
|
||||
@@ -35,11 +37,12 @@ def test_read_config_testrun(make_config):
|
||||
|
||||
def test_get_user_maildir(make_config):
|
||||
config = make_config("something.testrun.org")
|
||||
assert config.mail_basedir.name == "something.testrun.org"
|
||||
mailboxes_dir = Path(config.mailboxes_dir)
|
||||
assert mailboxes_dir.name == "something.testrun.org"
|
||||
assert config.mail_domain == "something.testrun.org"
|
||||
path = config.get_user_maildir("user1@something.testrun.org")
|
||||
path = Path(config.get_user_maildir("user1@something.testrun.org"))
|
||||
assert not path.exists()
|
||||
assert path == config.mail_basedir.joinpath("user1@something.testrun.org")
|
||||
assert path == mailboxes_dir.joinpath("user1@something.testrun.org")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
config.get_user_maildir("")
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from chatmaild.delete_inactive_users import delete_inactive_users
|
||||
from chatmaild.doveauth import lookup_passdb
|
||||
@@ -8,9 +9,12 @@ def test_remove_stale_users(db, example_config):
|
||||
new = time.time()
|
||||
old = new - (example_config.delete_inactive_users_after * 86400) - 1
|
||||
|
||||
def get_user_path(addr):
|
||||
return Path(example_config.get_user_maildir(addr))
|
||||
|
||||
def create_user(addr, last_login):
|
||||
lookup_passdb(db, example_config, addr, "q9mr3faue", last_login=last_login)
|
||||
md = example_config.get_user_maildir(addr)
|
||||
md = get_user_path(addr)
|
||||
md.mkdir(parents=True)
|
||||
md.joinpath("cur").mkdir()
|
||||
md.joinpath("cur", "something").mkdir()
|
||||
@@ -33,19 +37,19 @@ def test_remove_stale_users(db, example_config):
|
||||
# check pre and post-conditions for delete_inactive_users()
|
||||
|
||||
for addr in to_remove:
|
||||
assert example_config.get_user_maildir(addr).exists()
|
||||
assert get_user_path(addr).exists()
|
||||
|
||||
delete_inactive_users(db, example_config)
|
||||
|
||||
for p in example_config.mail_basedir.iterdir():
|
||||
for p in Path(example_config.mailboxes_dir).iterdir():
|
||||
assert not p.name.startswith("old")
|
||||
|
||||
for addr in to_remove:
|
||||
assert not get_user_path(addr).exists()
|
||||
with db.read_connection() as conn:
|
||||
assert not conn.get_user(addr)
|
||||
assert not example_config.get_user_maildir(addr).exists()
|
||||
|
||||
for addr in remain:
|
||||
assert example_config.get_user_maildir(addr).exists()
|
||||
assert get_user_path(addr).exists()
|
||||
with db.read_connection() as conn:
|
||||
assert conn.get_user(addr)
|
||||
|
||||
@@ -114,10 +114,7 @@ def test_handle_dovecot_request(db, example_config):
|
||||
assert res
|
||||
assert res[0] == "O" and res.endswith("\n")
|
||||
userdata = json.loads(res[1:].strip())
|
||||
assert (
|
||||
userdata["home"]
|
||||
== "/home/vmail/mail/chat.example.org/some42123@chat.example.org"
|
||||
)
|
||||
assert userdata["home"].endswith("chat.example.org/some42123@chat.example.org")
|
||||
assert userdata["uid"] == userdata["gid"] == "vmail"
|
||||
assert userdata["password"].startswith("{SHA512-CRYPT}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user