Compare commits

..

5 Commits

Author SHA1 Message Date
link2xt
77727e259e Use tox -c option 2023-10-22 11:49:25 +00:00
link2xt
732fdb3dab Setup deltachat dependency in init.sh 2023-10-22 11:47:40 +00:00
holger krekel
fe648f4784 run tests via scripts 2023-10-21 12:51:30 +02:00
holger krekel
d43e046c5d try to run all offline tests in CI 2023-10-21 12:49:05 +02:00
holger krekel
3716f2e429 fix init.sh and test.sh
use tox for chatmaild non-online tests
2023-10-21 12:22:21 +02:00
10 changed files with 122 additions and 107 deletions

View File

@@ -5,14 +5,27 @@ on:
push:
jobs:
lint:
name: Lint
tox:
name: chatmail tests
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Lint chatmaild
- name: run chatmaild tests
working-directory: chatmaild
run: pipx run tox
- name: Lint deploy-chatmail
- name: run deploy-chatmail offline tests
working-directory: deploy-chatmail
run: pipx run tox
- name: run deploy-chatmail offline tests
working-directory: deploy-chatmail
run: pipx run tox
scripts:
name: chatmail script invocations
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: run init.sh
run: ./scripts/init.sh
- name: run test.sh
run: ./scripts/test.sh

View File

@@ -20,7 +20,7 @@ addopts = "-v -ra --strict-markers"
legacy_tox_ini = """
[tox]
isolated_build = true
envlist = lint
envlist = lint,py
[testenv:lint]
skipdist = True
@@ -31,4 +31,10 @@ deps =
commands =
black --quiet --check --diff src/
ruff src/
[testenv]
passenv = CHATMAIL_DOMAIN
deps = pytest
pdbpp
commands = pytest -v -rsXx {posargs: ../tests/chatmaild}
"""

View File

@@ -21,68 +21,66 @@ def encrypt_password(password: str):
return "{SHA512-CRYPT}" + passhash
class DictProxy:
def __init__(self, db, mail_domain):
self.db = db
self.mail_domain = mail_domain
def create_user(self, user, password):
if os.path.exists(NOCREATE_FILE):
logging.warning(f"Didn't create account: {NOCREATE_FILE} exists.")
return
with self.db.write_transaction() as conn:
conn.create_user(user, password)
return dict(home=f"/home/vmail/{user}", uid="vmail", gid="vmail", password=password)
def get_user_data(self, user):
with self.db.read_connection() as conn:
result = conn.get_user(user)
if result:
result["uid"] = "vmail"
result["gid"] = "vmail"
return result
def create_user(db, user, password):
if os.path.exists(NOCREATE_FILE):
logging.warning(
f"Didn't create account: {NOCREATE_FILE} exists. Delete the file to enable account creation."
)
return
with db.write_transaction() as conn:
conn.create_user(user, password)
return dict(home=f"/home/vmail/{user}", uid="vmail", gid="vmail", password=password)
def lookup_userdb(self, user):
return self.get_user_data(user)
def get_user_data(db, user):
with db.read_connection() as conn:
result = conn.get_user(user)
if result:
result["uid"] = "vmail"
result["gid"] = "vmail"
return result
def lookup_passdb(self, user, password):
userdata = self.get_user_data(user)
if not userdata:
return self.create_user(user, encrypt_password(password))
userdata["password"] = userdata["password"].strip()
return userdata
def lookup_userdb(db, user):
return get_user_data(db, user)
def handle_dovecot_request(self, msg):
print(f"received msg: {msg!r}", file=sys.stderr)
short_command = msg[0]
if short_command == "L": # LOOKUP
parts = msg[1:].split("\t")
keyname, user = parts[:2]
namespace, type, *args = keyname.split("/")
reply_command = "F"
res = ""
if namespace == "shared":
if type == "userdb":
if user.endswith(f"@{self.mail_domain}"):
res = lookup_userdb(db, user)
if res:
reply_command = "O"
else:
reply_command = "N"
elif type == "passdb":
if user.endswith(f"@{self.mail_domain}"):
res = lookup_passdb(db, user, password=args[0])
if res:
reply_command = "O"
else:
reply_command = "N"
print(f"res: {res!r}", file=sys.stderr)
json_res = json.dumps(res) if res else ""
return f"{reply_command}{json_res}\n"
return None
def lookup_passdb(db, user, password):
userdata = get_user_data(db, user)
if not userdata:
return create_user(db, user, encrypt_password(password))
userdata["password"] = userdata["password"].strip()
return userdata
def handle_dovecot_request(msg, db, mail_domain):
print(f"received msg: {msg!r}", file=sys.stderr)
short_command = msg[0]
if short_command == "L": # LOOKUP
parts = msg[1:].split("\t")
keyname, user = parts[:2]
namespace, type, *args = keyname.split("/")
reply_command = "F"
res = ""
if namespace == "shared":
if type == "userdb":
if user.endswith(f"@{mail_domain}"):
res = lookup_userdb(db, user)
if res:
reply_command = "O"
else:
reply_command = "N"
elif type == "passdb":
if user.endswith(f"@{mail_domain}"):
res = lookup_passdb(db, user, password=args[0])
if res:
reply_command = "O"
else:
reply_command = "N"
print(f"res: {res!r}", file=sys.stderr)
json_res = json.dumps(res) if res else ""
return f"{reply_command}{json_res}\n"
return None
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
@@ -92,18 +90,17 @@ class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
def main():
socket = sys.argv[1]
passwd_entry = pwd.getpwnam(sys.argv[2])
db = Database(sys.argv[3])
with open("/etc/mailname", "r") as fp:
mail_domain = fp.read().strip()
db = Database(sys.argv[3])
dictproxy = DictProxy(db, mail_domain)
class Handler(StreamRequestHandler):
def handle(self):
while True:
msg = self.rfile.readline().strip().decode()
if not msg:
break
res = dictproxy.handle_dovecot_request(msg)
res = handle_dovecot_request(msg, db, mail_domain)
if res:
print(f"sending result: {res!r}", file=sys.stderr)
self.wfile.write(res.encode("ascii"))

View File

@@ -11,3 +11,4 @@ conn.login(f"imapcapa", "pass")
status, res = conn.capability()
for capa in sorted(res[0].decode().split()):
print(capa)

View File

@@ -1,13 +1,8 @@
#!/bin/sh
set -e
python3 -m venv deploy-chatmail/venv
deploy-chatmail/venv/bin/pip install pyinfra pytest
deploy-chatmail/venv/bin/pip install -e deploy-chatmail
deploy-chatmail/venv/bin/pip install -e chatmaild
python3 -m venv venv
pip=venv/bin/pip
python3 -m venv chatmaild/venv
chatmaild/venv/bin/pip install --upgrade pytest build 'setuptools>=68'
chatmaild/venv/bin/pip install -e chatmaild
python3 -m venv online-tests/venv
online-tests/venv/bin/pip install pytest pytest-timeout pdbpp deltachat
$pip install pyinfra pytest build 'setuptools>=68' tox deltachat
$pip install -e deploy-chatmail
$pip install -e chatmaild

View File

@@ -5,7 +5,7 @@ import imaplib
domain = os.environ.get("CHATMAIL_DOMAIN", "c3.testrun.org")
NUM_CONNECTIONS = 10
NUM_CONNECTIONS=10
conns = []
@@ -16,7 +16,7 @@ for i in range(NUM_CONNECTIONS):
conns.append(conn)
tlsdone = time.time()
duration = tlsdone - start
duration = tlsdone-start
print(f"{duration}: TLS connections opening TLS connections")
for i, conn in enumerate(conns):

View File

@@ -1,3 +1,4 @@
#!/bin/bash
chatmaild/venv/bin/pytest chatmaild/ $@
online-tests/venv/bin/pytest online-tests/ -vrx --durations=5 $@
tox -c chatmaild
tox -c deploy-chatmail
venv/bin/pytest tests/online -vrx --durations=5 $@

View File

@@ -1,9 +0,0 @@
import pytest
from chatmaild.database import Database
@pytest.fixture()
def db(tmpdir):
db_path = tmpdir / "passdb.sqlite"
print("database path:", db_path)
return Database(db_path)

View File

@@ -3,38 +3,43 @@ import os
import pytest
import chatmaild.dictproxy
from chatmaild.dictproxy import DictProxy
from chatmaild.database import DBError
from chatmaild.dictproxy import get_user_data, lookup_passdb
from chatmaild.database import Database, DBError
@pytest.fixture
def dictproxy(db, maildomain):
return DictProxy(db, maildomain)
@pytest.fixture()
def db(tmpdir):
db_path = tmpdir / "passdb.sqlite"
print("database path:", db_path)
return Database(db_path)
def test_basic(dictproxy, tmpdir, monkeypatch):
monkeypatch.setattr(
chatmaild.dictproxy, "NOCREATE_FILE", tmpdir.join("nocreate").strpath
)
dictproxy.lookup_passdb("link2xt@c1.testrun.org", "asdf")
assert dictproxy.get_user_data("link2xt@c1.testrun.org")
def test_basic(db):
chatmaild.dictproxy.NOCREATE_FILE = "/tmp/nocreate"
if os.path.exists(chatmaild.dictproxy.NOCREATE_FILE):
os.remove(chatmaild.dictproxy.NOCREATE_FILE)
lookup_passdb(db, "link2xt@c1.testrun.org", "asdf")
data = get_user_data(db, "link2xt@c1.testrun.org")
assert data
def test_dont_overwrite_password_on_wrong_login(dictproxy):
def test_dont_overwrite_password_on_wrong_login(db):
"""Test that logging in with a different password doesn't create a new user"""
res = dictproxy.lookup_passdb("newuser1@something.org", "kajdlkajsldk12l3kj1983")
res = lookup_passdb(db, "newuser1@something.org", "kajdlkajsldk12l3kj1983")
assert res["password"]
res2 = dictproxy.lookup_passdb("newuser1@something.org", "kajdlqweqwe")
res2 = lookup_passdb(db, "newuser1@something.org", "kajdlqweqwe")
# this function always returns a password hash, which is actually compared by dovecot.
assert res["password"] == res2["password"]
def test_nocreate_file(dictproxy, tmpdir, monkeypatch):
nocreate = tmpdir.join("nocreate")
monkeypatch.setattr(chatmaild.dictproxy, "NOCREATE_FILE", str(nocreate))
nocreate.write("")
dictproxy.lookup_passdb("newuser1@something.org", "kajdlqweqwe")
assert not dictproxy.get_user_data("newuser1@something.org")
def test_nocreate_file(db):
chatmaild.dictproxy.NOCREATE_FILE = "/tmp/nocreate"
with open(chatmaild.dictproxy.NOCREATE_FILE, "w+") as f:
f.write("")
assert os.path.exists(chatmaild.dictproxy.NOCREATE_FILE)
lookup_passdb(db, "newuser1@something.org", "kajdlqweqwe")
assert not get_user_data(db, "newuser1@something.org")
os.remove(chatmaild.dictproxy.NOCREATE_FILE)
def test_db_version(db):

View File

@@ -2,6 +2,12 @@ from chatmaild.filtermail import check_encrypted, check_DATA, SendRateLimiter
import pytest
@pytest.fixture
def maildomain():
# let's not depend on a real chatmail instance for the offline tests below
return "chatmail.example.org"
def test_reject_forged_from(maildata, gencreds):
class env:
mail_from = gencreds()[0]