mirror of
https://github.com/chatmail/relay.git
synced 2026-05-20 12:58:04 +00:00
let config.get_user_maildir return a Path
This commit is contained in:
@@ -40,7 +40,7 @@ class Config:
|
|||||||
if addr and addr != "." and "/" not in addr:
|
if addr and addr != "." and "/" not in addr:
|
||||||
res = self.mailboxes_dir.joinpath(addr).resolve()
|
res = self.mailboxes_dir.joinpath(addr).resolve()
|
||||||
if res.is_relative_to(self.mailboxes_dir):
|
if res.is_relative_to(self.mailboxes_dir):
|
||||||
return str(res)
|
return res
|
||||||
raise ValueError(f"invalid address {addr!r}")
|
raise ValueError(f"invalid address {addr!r}")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -68,7 +68,7 @@ def is_allowed_to_create(config: Config, user, cleartext_password) -> bool:
|
|||||||
def get_user_data(db, config: Config, user):
|
def get_user_data(db, config: Config, user):
|
||||||
if user == f"echo@{config.mail_domain}":
|
if user == f"echo@{config.mail_domain}":
|
||||||
return dict(
|
return dict(
|
||||||
home=config.get_user_maildir(user),
|
home=str(config.get_user_maildir(user)),
|
||||||
uid="vmail",
|
uid="vmail",
|
||||||
gid="vmail",
|
gid="vmail",
|
||||||
)
|
)
|
||||||
@@ -76,7 +76,7 @@ def get_user_data(db, config: Config, user):
|
|||||||
with db.read_connection() as conn:
|
with db.read_connection() as conn:
|
||||||
result = conn.get_user(user)
|
result = conn.get_user(user)
|
||||||
if result:
|
if result:
|
||||||
result["home"] = config.get_user_maildir(user)
|
result["home"] = str(config.get_user_maildir(user))
|
||||||
result["uid"] = "vmail"
|
result["uid"] = "vmail"
|
||||||
result["gid"] = "vmail"
|
result["gid"] = "vmail"
|
||||||
return result
|
return result
|
||||||
@@ -96,7 +96,7 @@ def lookup_passdb(db, config: Config, user, cleartext_password, last_login=None)
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
return dict(
|
return dict(
|
||||||
home=config.get_user_maildir(user),
|
home=str(config.get_user_maildir(user)),
|
||||||
uid="vmail",
|
uid="vmail",
|
||||||
gid="vmail",
|
gid="vmail",
|
||||||
password=encrypt_password(password),
|
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)
|
"UPDATE users SET last_login=? WHERE addr=?", (last_login, user)
|
||||||
)
|
)
|
||||||
|
|
||||||
userdata["home"] = config.get_user_maildir(user)
|
userdata["home"] = str(config.get_user_maildir(user))
|
||||||
userdata["uid"] = "vmail"
|
userdata["uid"] = "vmail"
|
||||||
userdata["gid"] = "vmail"
|
userdata["gid"] = "vmail"
|
||||||
return userdata
|
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))
|
conn.execute(q, (user, encrypted_password, last_login))
|
||||||
print(f"Created address: {user}", file=sys.stderr)
|
print(f"Created address: {user}", file=sys.stderr)
|
||||||
return dict(
|
return dict(
|
||||||
home=config.get_user_maildir(user),
|
home=str(config.get_user_maildir(user)),
|
||||||
uid="vmail",
|
uid="vmail",
|
||||||
gid="vmail",
|
gid="vmail",
|
||||||
password=encrypted_password,
|
password=encrypted_password,
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
import time
|
import time
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from chatmaild.delete_inactive_users import delete_inactive_users
|
from chatmaild.delete_inactive_users import delete_inactive_users
|
||||||
from chatmaild.doveauth import lookup_passdb
|
from chatmaild.doveauth import lookup_passdb
|
||||||
@@ -9,12 +8,9 @@ def test_remove_stale_users(db, example_config):
|
|||||||
new = time.time()
|
new = time.time()
|
||||||
old = new - (example_config.delete_inactive_users_after * 86400) - 1
|
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):
|
def create_user(addr, last_login):
|
||||||
lookup_passdb(db, example_config, addr, "q9mr3faue", last_login=last_login)
|
lookup_passdb(db, example_config, addr, "q9mr3faue", last_login=last_login)
|
||||||
md = get_user_path(addr)
|
md = example_config.get_user_maildir(addr)
|
||||||
md.mkdir(parents=True)
|
md.mkdir(parents=True)
|
||||||
md.joinpath("cur").mkdir()
|
md.joinpath("cur").mkdir()
|
||||||
md.joinpath("cur", "something").mkdir()
|
md.joinpath("cur", "something").mkdir()
|
||||||
@@ -37,7 +33,7 @@ def test_remove_stale_users(db, example_config):
|
|||||||
# check pre and post-conditions for delete_inactive_users()
|
# check pre and post-conditions for delete_inactive_users()
|
||||||
|
|
||||||
for addr in to_remove:
|
for addr in to_remove:
|
||||||
assert get_user_path(addr).exists()
|
assert example_config.get_user_maildir(addr).exists()
|
||||||
|
|
||||||
delete_inactive_users(db, example_config)
|
delete_inactive_users(db, example_config)
|
||||||
|
|
||||||
@@ -45,11 +41,11 @@ def test_remove_stale_users(db, example_config):
|
|||||||
assert not p.name.startswith("old")
|
assert not p.name.startswith("old")
|
||||||
|
|
||||||
for addr in to_remove:
|
for addr in to_remove:
|
||||||
assert not get_user_path(addr).exists()
|
assert not example_config.get_user_maildir(addr).exists()
|
||||||
with db.read_connection() as conn:
|
with db.read_connection() as conn:
|
||||||
assert not conn.get_user(addr)
|
assert not conn.get_user(addr)
|
||||||
|
|
||||||
for addr in remain:
|
for addr in remain:
|
||||||
assert get_user_path(addr).exists()
|
assert example_config.get_user_maildir(addr).exists()
|
||||||
with db.read_connection() as conn:
|
with db.read_connection() as conn:
|
||||||
assert conn.get_user(addr)
|
assert conn.get_user(addr)
|
||||||
|
|||||||
Reference in New Issue
Block a user