Compare commits
12 Commits
Author | SHA1 | Date | |
---|---|---|---|
fbb2cf38ae | |||
4df53ce247 | |||
eb5c71968a | |||
f3e80c43ae | |||
829be8413b | |||
121a02b8ae | |||
e18f1c7a96 | |||
cc5ad89977 | |||
a1fd586dbd | |||
8fe42e9163 | |||
937992a72e | |||
e1a78d560d |
13
Makefile
13
Makefile
@ -1,8 +1,10 @@
|
||||
# Needs python3 >= 3.9, sed, git for build
|
||||
# Needs python3 >= 3.9, sed, git for build, docker for tests
|
||||
build: clean
|
||||
python3 -m pip install -r requirements.txt --no-compile --target build
|
||||
cp -r mail4one/ build/
|
||||
sed -i "s/DEVELOMENT/$(shell scripts/get_version.sh)/" build/mail4one/version.py
|
||||
find build -name "*.pyi" -o -name "py.typed" | xargs -I typefile rm typefile
|
||||
rm -rf build/bin
|
||||
rm -rf build/mail4one/__pycache__
|
||||
rm -rf build/*.dist-info
|
||||
python3 -m zipapp \
|
||||
@ -18,17 +20,19 @@ clean:
|
||||
docker-tests:
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.11-alpine sh scripts/runtests.sh
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.10-alpine sh scripts/runtests.sh
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.12 sh scripts/runtests.sh
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.11 sh scripts/runtests.sh
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.10 sh scripts/runtests.sh
|
||||
docker run --pull=always -v `pwd`:/app -w /app --rm python:3.9 sh scripts/runtests.sh
|
||||
|
||||
# ============================================================================
|
||||
# Below targets for devs. Need pipenv, black installed
|
||||
|
||||
requirements.txt: Pipfile.lock
|
||||
pipenv requirements > requirements.txt
|
||||
|
||||
format:
|
||||
black mail4one/*py
|
||||
black mail4one/*py tests/*py
|
||||
|
||||
build-dev: requirements.txt build
|
||||
|
||||
@ -38,6 +42,11 @@ setup:
|
||||
cleanup:
|
||||
pipenv --rm
|
||||
|
||||
update:
|
||||
rm requirements.txt Pipfile.lock
|
||||
pipenv update
|
||||
pipenv requirements > requirements.txt
|
||||
|
||||
shell:
|
||||
MYPYPATH=`pipenv --venv`/lib/python3.11/site-packages pipenv shell
|
||||
|
||||
|
20
Pipfile.lock
generated
20
Pipfile.lock
generated
@ -18,27 +18,28 @@
|
||||
"default": {
|
||||
"aiosmtpd": {
|
||||
"hashes": [
|
||||
"sha256:f821fe424b703b2ea391dc2df11d89d2afd728af27393e13cf1a3530f19fdc5e",
|
||||
"sha256:f9243b7dfe00aaf567da8728d891752426b51392174a34d2cf5c18053b63dcbc"
|
||||
"sha256:78d7b14f859ad0e6de252b47f9cf1ca6f1c82a8b0f10a9e39bec7e915a6aa5fe",
|
||||
"sha256:a196922f1903e54c4d37c53415b7613056d39e2b1e8249f324b9ee7a439be0f1"
|
||||
],
|
||||
"index": "pypi",
|
||||
"version": "==1.4.4.post2"
|
||||
"markers": "python_version >= '3.8'",
|
||||
"version": "==1.4.5"
|
||||
},
|
||||
"atpublic": {
|
||||
"hashes": [
|
||||
"sha256:0f40433219e124edf115c6c363808ca6f0e1cfa7d160d86b2fb94793086d1294",
|
||||
"sha256:80057c55641253b86dcb68b524f82328172371b6547d4c7462a9127fbfbbabfc"
|
||||
"sha256:d1c8cd931af7461f6d18bc6063383e8654d9e9ef19d58ee6dc01e8515bbf55df",
|
||||
"sha256:df90de1162b1a941ee486f484691dc7c33123ee638ea5d6ca604061306e0fdde"
|
||||
],
|
||||
"markers": "python_version >= '3.8'",
|
||||
"version": "==4.0"
|
||||
"version": "==4.1.0"
|
||||
},
|
||||
"attrs": {
|
||||
"hashes": [
|
||||
"sha256:1f28b4522cdc2fb4256ac1a020c78acf9cba2c6b461ccd2c126f3aa8e8335d04",
|
||||
"sha256:6279836d581513a26f1bf235f9acd333bc9115683f14f7e8fae46c98fc50e015"
|
||||
"sha256:935dc3b529c262f6cf76e50877d35a4bd3c1de194fd41f47a2b7ae8f19971f30",
|
||||
"sha256:99b87a485a5820b23b879f04c2305b44b951b502fd64be915879d77a7e8fc6f1"
|
||||
],
|
||||
"markers": "python_version >= '3.7'",
|
||||
"version": "==23.1.0"
|
||||
"version": "==23.2.0"
|
||||
},
|
||||
"python-jata": {
|
||||
"hashes": [
|
||||
@ -46,6 +47,7 @@
|
||||
"sha256:ff4cd7ca75c9a8306b69ef6e878c296a5602f3279c6f9a82b6105b8eba764760"
|
||||
],
|
||||
"index": "pypi",
|
||||
"markers": "python_version >= '3.8'",
|
||||
"version": "==1.2"
|
||||
}
|
||||
},
|
||||
|
@ -1,6 +1,6 @@
|
||||
# Mail4one
|
||||
|
||||
Personal mail server for a single user or a small family. Written in pure python with minimal dependencies.
|
||||
Personal mail server for a single user or a small family. Written in pure python with [minimal dependencies](Pipfile).
|
||||
Designed for dynamic alias based workflow where a different alias is used for each purpose.
|
||||
|
||||
# Getting started
|
||||
@ -11,7 +11,7 @@ Designed for dynamic alias based workflow where a different alias is used for ea
|
||||
1. [Build](#building-from-source) / Download latest release - `mail4one.pyz`
|
||||
1. Generate `config.json` from [config.sample](deploy_configs/config.sample)
|
||||
1. Run `./mail4one.pyz -c config.json`
|
||||
1. [Optional] Setup systemd service and TLS certificates. See [deploy_configs](deploy_configs/) for examples
|
||||
1. Setup systemd service and TLS certificates. See [deploy_configs](deploy_configs/) for examples
|
||||
|
||||
# Sending email
|
||||
|
||||
@ -23,7 +23,7 @@ Mail4one only takes care of receiving and serving email. For sending email, use
|
||||
|
||||
Most of them have generous free tier which is more than enough for personal use.
|
||||
|
||||
Sending email is tricky. Even if everything is correctly setup (DMARC, DKIM, SPF), popular email vendors like google, microsoft may mark emails sent from your IP as spam for no reason.
|
||||
Sending email is tricky. Even if everything is correctly setup (DMARC, DKIM, SPF), popular email vendors like google, microsoft may mark emails sent from your IP as spam for no reason. Hence using a dedicated service is the only reliable way to send emails.
|
||||
|
||||
# Community
|
||||
|
||||
@ -62,7 +62,7 @@ This should generate `mail4one.pyz` in current folder. This is a [executable pyt
|
||||
* Write dedicated documentation
|
||||
* Test with more email clients ([Thunderbird](https://www.thunderbird.net/) and [k9mail](https://k9mail.app/) are tested now)
|
||||
* IMAP support
|
||||
* Web UI for editing config
|
||||
* Web UI for editing config ([WIP](https://github.com/mail4one/mail4one/tree/webform))
|
||||
* Support email submission from client to forward to other senders or direct delivery
|
||||
* Optional SPAM filtering
|
||||
* Optional DMARC,SPF,DKIM verification
|
||||
|
115
deploy_configs/README.md
Normal file
115
deploy_configs/README.md
Normal file
@ -0,0 +1,115 @@
|
||||
# Deployment command line example
|
||||
|
||||
Example terminal session for deploying. ssh to your VPS and follow along. Minor differences may be required. e.g. if you are already root, skip `sudo`. If curl is missing, use wget.
|
||||
|
||||
## Check python version
|
||||
|
||||
Python version should be a supported (as of now 3.9 and above)
|
||||
```sh
|
||||
python3 -V
|
||||
```
|
||||
|
||||
## Choose release
|
||||
```sh
|
||||
RELEASE=v1.0
|
||||
```
|
||||
|
||||
## Download App
|
||||
```sh
|
||||
curl -OL "https://gitea.balki.me/balki/mail4one/releases/download/$RELEASE/mail4one.pyz"
|
||||
chmod 555 mail4one.pyz
|
||||
```
|
||||
|
||||
## Download sample configurations
|
||||
```sh
|
||||
curl -OL "https://gitea.balki.me/balki/mail4one/raw/tag/$RELEASE/deploy_configs/mail4one.service"
|
||||
curl -OL "https://gitea.balki.me/balki/mail4one/raw/tag/$RELEASE/deploy_configs/mail4one.conf"
|
||||
curl -OL "https://gitea.balki.me/balki/mail4one/raw/tag/$RELEASE/deploy_configs/mail4one_cert_copy.sh"
|
||||
```
|
||||
|
||||
## Generate Password hash
|
||||
|
||||
This can be done in any machine. Do this once for each user. Every time a new hash is generated as a random salt is used. Even if you are using the same password for multiple clients, it is recommended to generate different hashes for each.
|
||||
|
||||
```sh
|
||||
./mail4one.pyz -g
|
||||
./mail4one.pyz -g <password> # also works but the password is saved in the shell commandline history
|
||||
```
|
||||
|
||||
## Generate config.json
|
||||
|
||||
Edit [config.sample](config.sample) in your local machine and convert to config.json (See [here](./config.sample#L5) for some tools).
|
||||
|
||||
Then copy the config.json to your vps
|
||||
```sh
|
||||
scp config.json user@vps:~/
|
||||
# or run below in vps terminal
|
||||
cat > config.json
|
||||
<paste json config from clibboard
|
||||
<Ctrl + D>
|
||||
|
||||
# move to /etc
|
||||
|
||||
# This should show number of lines in your config
|
||||
wc -l config.json
|
||||
sudo mv config.json /etc/mail4one/config.json
|
||||
```
|
||||
|
||||
## Create mail4one user
|
||||
|
||||
```sh
|
||||
sudo mkdir -p /etc/sysusers.d/
|
||||
sudo cp mail4one.conf /etc/sysusers.d/
|
||||
sudo systemctl restart systemd-sysusers
|
||||
# This should show the new user created
|
||||
id mail4one
|
||||
```
|
||||
## Copy app
|
||||
```sh
|
||||
sudo cp mail4one.pyz /usr/local/bin/mail4one
|
||||
# This should show executable permissions and should be owned by root
|
||||
ls -l /usr/local/bin/mail4one
|
||||
```
|
||||
|
||||
## Setup mail4one.service
|
||||
```sh
|
||||
sudo cp mail4one.service /etc/systemd/system/mail4one.service
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now mail4one.service
|
||||
systemctl status mail4one
|
||||
```
|
||||
Above command should fail as the TLS certificates don't exist yet.
|
||||
|
||||
## Setup TLS certificates
|
||||
Install [certbot](https://certbot.eff.org/) and run below command. Follow instructions to create TLS certificates. Usually you want certificate for domain name like `mail.mydomain.com`
|
||||
```sh
|
||||
sudo certbot certonly
|
||||
|
||||
# **Edit** mail4one_cert_copy.sh to update your domain name
|
||||
sudo cp mail4one_cert_copy.sh /etc/letsencrypt/renewal-hooks/deploy/
|
||||
sudo chmod +x /etc/letsencrypt/renewal-hooks/deploy/mail4one_cert_copy.sh
|
||||
|
||||
# This will create and copy the certificates to the right path with correct permissions and ownership
|
||||
sudo certbot certonly -d mail.mydomain.com --run-deploy-hooks --dry-run
|
||||
```
|
||||
## Restart service and check logs
|
||||
```sh
|
||||
sudo systemctl restart mail4one.service
|
||||
systemctl status mail4one.service
|
||||
cat /var/log/mail4one/mail4one.log
|
||||
```
|
||||
|
||||
## Testing dns and firewall
|
||||
In vps
|
||||
```sh
|
||||
mkdir test_dir
|
||||
touch test_dir/{a,b,c}
|
||||
cd test_dir
|
||||
python3 -m http.server 25
|
||||
```
|
||||
In local machine or a browser
|
||||
You should see file listing a, b, c. Repeat for port 465, 995 to make sure firewall rules and dns is working
|
||||
```sh
|
||||
curl http://mail.mydomain.com:25
|
||||
```
|
||||
If not working, refer to VPS settings and OS firewall settings.
|
@ -1,17 +1,17 @@
|
||||
# NOTE: Sample config is provided in yaml format for easy editing
|
||||
# mail4one needs a json config, Please convert the config to json before passing to app
|
||||
# This is to avoid yaml depependency in the app
|
||||
# This is to avoid yaml dependency in the app
|
||||
#
|
||||
# Some tools to convert to json:
|
||||
# If you have go in your system (https://go.dev/)
|
||||
# If you have `go` in your system (https://go.dev/)
|
||||
# go run github.com/mikefarah/yq/v4@latest -oj -P . config.sample > config.json
|
||||
#
|
||||
# If you have pipx in your system (https://pypa.github.io/pipx/)
|
||||
# If you have `pipx` in your system (https://pypa.github.io/pipx/)
|
||||
# pipx run yq . config.sample > config.json
|
||||
#
|
||||
# or a browser:
|
||||
# https://onlineyamltools.com/convert-yaml-to-json
|
||||
#
|
||||
|
||||
default_tls: # Will be used by both pop and smtp servers
|
||||
# If using certbot(https://certbot.eff.org/),
|
||||
# the following files will be here /etc/letsencrypt/live/<domain name>
|
||||
|
@ -2,19 +2,26 @@
|
||||
|
||||
# certbot deploy hook to copy certificates to mail4one when renewed.
|
||||
# Initial setup, Install certbot(https://certbot.eff.org/) and run `certbot certonly` as root
|
||||
# Doc: https://eff-certbot.readthedocs.io/en/latest/using.html#renewing-certificates
|
||||
#
|
||||
# This file is supposed to be copied to /etc/letsencrypt/renewal-hooks/deploy/
|
||||
# Change the mail domain to the one on MX record
|
||||
|
||||
set -x
|
||||
|
||||
set -eu
|
||||
|
||||
if [ "$RENEWED_DOMAINS" = "mail.mydomain.com" ]
|
||||
then
|
||||
mkdir -p /var/lib/mail4one/certs
|
||||
chmod 750 /var/lib/mail4one/certs
|
||||
chown mail4one:mail4one /var/lib/mail4one/certs
|
||||
cp "$RENEWED_LINEAGE/fullchain.pem" /var/lib/mail4one/certs/
|
||||
cp "$RENEWED_LINEAGE/privkey.pem" /var/lib/mail4one/certs/
|
||||
systemctl restart mail4one.service
|
||||
app=mail4one
|
||||
appuser=$app
|
||||
certpath="/var/lib/$app/certs"
|
||||
|
||||
mkdir -p "$certpath"
|
||||
chmod 750 "$certpath"
|
||||
|
||||
chown $appuser:$appuser "$certpath"
|
||||
install -o "$appuser" -g "$appuser" -m 444 "$RENEWED_LINEAGE/fullchain.pem" -t "$certpath"
|
||||
install -o "$appuser" -g "$appuser" -m 400 "$RENEWED_LINEAGE/privkey.pem" -t "$certpath"
|
||||
|
||||
systemctl restart $app.service
|
||||
echo "$(date) Renewed and deployed certificates for $app" >> /var/log/cert-renew.log
|
||||
fi
|
||||
|
@ -29,7 +29,7 @@ from .poputils import (
|
||||
end,
|
||||
Request,
|
||||
MailEntry,
|
||||
get_mail,
|
||||
get_mail_fp,
|
||||
get_mails_list,
|
||||
MailList,
|
||||
)
|
||||
@ -116,8 +116,8 @@ async def expect_cmd(*commands: Command) -> Request:
|
||||
return req
|
||||
|
||||
|
||||
def write(data) -> None:
|
||||
logger.debug(f"Server: {data}")
|
||||
def write(data: bytes) -> None:
|
||||
logger.debug(f"Server: {data!r}")
|
||||
state().writer.write(data)
|
||||
|
||||
|
||||
@ -217,7 +217,12 @@ def trans_command_retr(mails: MailList, req: Request) -> None:
|
||||
entry = mails.get(req.arg1)
|
||||
if entry:
|
||||
write(ok("Contents follow"))
|
||||
write(get_mail(entry))
|
||||
with get_mail_fp(entry) as fp:
|
||||
for line in fp:
|
||||
if line.startswith(b"."):
|
||||
write(b".") # prepend dot
|
||||
write(line)
|
||||
# write(get_mail(entry)) # no prepend dot
|
||||
write(end())
|
||||
mails.delete(req.arg1)
|
||||
else:
|
||||
@ -386,13 +391,14 @@ def debug_main():
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
import sys
|
||||
from .pwhash import gen_pwhash
|
||||
|
||||
_, mails_path, port, password = sys.argv
|
||||
_, mails_path, mbox = sys.argv
|
||||
|
||||
mails_path = Path(mails_path)
|
||||
port = int(port)
|
||||
users = [User(username="dummy", password_hash=gen_pwhash("dummy"), mbox=mbox)]
|
||||
|
||||
asyncio.run(a_main(mails_path, port, password_hash=password_hash))
|
||||
asyncio.run(a_main("127.0.0.1", 1101, mails_path, users=users))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
@ -2,6 +2,7 @@ import os
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
class ClientError(Exception):
|
||||
@ -124,6 +125,12 @@ def set_nid(entries: list[MailEntry]):
|
||||
entry.nid = i
|
||||
|
||||
|
||||
@contextmanager
|
||||
def get_mail_fp(entry: MailEntry):
|
||||
with open(entry.path, mode="rb") as fp:
|
||||
yield fp
|
||||
|
||||
|
||||
def get_mail(entry: MailEntry) -> bytes:
|
||||
with open(entry.path, mode="rb") as fp:
|
||||
return fp.read()
|
||||
|
@ -1,5 +1,5 @@
|
||||
-i https://pypi.org/simple
|
||||
aiosmtpd==1.4.4.post2
|
||||
atpublic==4.0 ; python_version >= '3.8'
|
||||
attrs==23.1.0 ; python_version >= '3.7'
|
||||
python-jata==1.2
|
||||
aiosmtpd==1.4.5; python_version >= '3.8'
|
||||
atpublic==4.1.0; python_version >= '3.8'
|
||||
attrs==23.2.0; python_version >= '3.7'
|
||||
python-jata==1.2; python_version >= '3.8'
|
||||
|
@ -8,7 +8,7 @@ then
|
||||
tag_val=$(git describe --dirty=DIRTY --exact-match)
|
||||
case "$tag_val" in
|
||||
*DIRTY)
|
||||
echo "git=$commit-changes"
|
||||
echo "git-$commit-changes"
|
||||
;;
|
||||
v*) # Only consider tags starting with v
|
||||
echo "$tag_val"
|
||||
|
@ -67,10 +67,11 @@ class TestConfig(unittest.TestCase):
|
||||
def test_get_mboxes(self) -> None:
|
||||
cfg = config.Config(TEST_CONFIG)
|
||||
rules = config.parse_checkers(cfg)
|
||||
self.assertEqual(config.get_mboxes("foo@bar.com", rules), ['spam'])
|
||||
self.assertEqual(config.get_mboxes("foo@mydomain.com", rules), ['all'])
|
||||
self.assertEqual(config.get_mboxes("first.last@mydomain.com", rules),
|
||||
['important', 'all'])
|
||||
self.assertEqual(config.get_mboxes("foo@bar.com", rules), ["spam"])
|
||||
self.assertEqual(config.get_mboxes("foo@mydomain.com", rules), ["all"])
|
||||
self.assertEqual(
|
||||
config.get_mboxes("first.last@mydomain.com", rules), ["important", "all"]
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
@ -4,20 +4,29 @@ import logging
|
||||
import tempfile
|
||||
import time
|
||||
import os
|
||||
import poplib
|
||||
from mail4one.pop3 import create_pop_server
|
||||
from mail4one.config import User
|
||||
from pathlib import Path
|
||||
|
||||
TEST_HASH = "".join(c for c in """
|
||||
TEST_HASH = "".join(
|
||||
"""
|
||||
AFTY5EVN7AX47ZL7UMH3BETYWFBTAV3XHR73CEFAJBPN2NIHPWD
|
||||
ZHV2UQSMSPHSQQ2A2BFQBNC77VL7F2UKATQNJZGYLCSU6C43UQD
|
||||
AQXWXSWNGAEPGIMG2F3QDKBXL3MRHY6K2BPID64ZR6LABLPVSF
|
||||
""" if not c.isspace())
|
||||
""".split()
|
||||
)
|
||||
|
||||
TEST_USER = 'foobar'
|
||||
TEST_MBOX = 'foobar_mails'
|
||||
TEST_USER = "foobar"
|
||||
TEST_MBOX = "foobar_mails"
|
||||
|
||||
USERS = [User(username=TEST_USER, password_hash=TEST_HASH, mbox=TEST_MBOX)]
|
||||
TEST_USER2 = "foo2"
|
||||
TEST_MBOX2 = "foo2mails"
|
||||
|
||||
USERS = [
|
||||
User(username=TEST_USER, password_hash=TEST_HASH, mbox=TEST_MBOX),
|
||||
User(username=TEST_USER2, password_hash=TEST_HASH, mbox=TEST_MBOX2),
|
||||
]
|
||||
|
||||
MAILS_PATH: Path
|
||||
|
||||
@ -38,7 +47,8 @@ Hello bro\r
|
||||
IlzVOJqu9Zp7twFAtzcV\r
|
||||
yQVk36B0mGU2gtWxXLr\r
|
||||
PeF0RtbI0mAuVPLQDHCi\r
|
||||
\r\n"""
|
||||
\r
|
||||
"""
|
||||
|
||||
|
||||
def setUpModule() -> None:
|
||||
@ -47,13 +57,21 @@ def setUpModule() -> None:
|
||||
td = tempfile.TemporaryDirectory(prefix="m41.pop.")
|
||||
unittest.addModuleCleanup(td.cleanup)
|
||||
MAILS_PATH = Path(td.name)
|
||||
os.mkdir(MAILS_PATH / TEST_MBOX)
|
||||
for md in ('new', 'cur', 'tmp'):
|
||||
os.mkdir(MAILS_PATH / TEST_MBOX / md)
|
||||
with open(MAILS_PATH / TEST_MBOX/ 'new/msg1.eml', 'wb') as f:
|
||||
for mbox in (TEST_MBOX, TEST_MBOX2):
|
||||
os.mkdir(MAILS_PATH / mbox)
|
||||
for md in ("new", "cur", "tmp"):
|
||||
os.mkdir(MAILS_PATH / mbox / md)
|
||||
with open(MAILS_PATH / TEST_MBOX / "new/msg1.eml", "wb") as f:
|
||||
f.write(TESTMAIL)
|
||||
with open(MAILS_PATH / TEST_MBOX/ 'new/msg2.eml', 'wb') as f:
|
||||
with open(MAILS_PATH / TEST_MBOX / "new/msg2.eml", "wb") as f:
|
||||
f.write(TESTMAIL)
|
||||
with open(MAILS_PATH / TEST_MBOX2 / "new/msg1.eml", "wb") as f:
|
||||
f.write(TESTMAIL)
|
||||
f.write(b"More lines to follow\r\n")
|
||||
f.write(b".Line starts with a dot\r\n")
|
||||
f.write(b"some more lines\r\n")
|
||||
f.write(b".\r\n")
|
||||
f.write(b"Previous line just has a dot\r\n")
|
||||
logging.debug(MAILS_PATH)
|
||||
|
||||
|
||||
@ -65,13 +83,11 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
async def asyncSetUp(self) -> None:
|
||||
logging.debug("at asyncSetUp")
|
||||
pop_server = await create_pop_server(host='127.0.0.1',
|
||||
port=7995,
|
||||
mails_path=MAILS_PATH,
|
||||
users=USERS)
|
||||
pop_server = await create_pop_server(
|
||||
host="127.0.0.1", port=7995, mails_path=MAILS_PATH, users=USERS
|
||||
)
|
||||
self.task = asyncio.create_task(pop_server.serve_forever())
|
||||
self.reader, self.writer = await asyncio.open_connection(
|
||||
'127.0.0.1', 7995)
|
||||
self.reader, self.writer = await asyncio.open_connection("127.0.0.1", 7995)
|
||||
|
||||
async def test_QUIT(self) -> None:
|
||||
dialog = """
|
||||
@ -115,8 +131,8 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
|
||||
await self.dialog_checker(dialog)
|
||||
|
||||
async def test_dupe_AUTH(self) -> None:
|
||||
r1, w1 = await asyncio.open_connection('127.0.0.1', 7995)
|
||||
r2, w2 = await asyncio.open_connection('127.0.0.1', 7995)
|
||||
r1, w1 = await asyncio.open_connection("127.0.0.1", 7995)
|
||||
r2, w2 = await asyncio.open_connection("127.0.0.1", 7995)
|
||||
dialog = """
|
||||
S: +OK Server Ready
|
||||
C: USER foobar
|
||||
@ -197,6 +213,22 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
|
||||
"""
|
||||
await self.dialog_checker(dialog)
|
||||
|
||||
async def test_poplib(self) -> None:
|
||||
def run_poplib():
|
||||
pc = poplib.POP3("127.0.0.1", 7995)
|
||||
try:
|
||||
self.assertEqual(b"+OK Server Ready", pc.getwelcome())
|
||||
self.assertEqual(b"+OK Welcome", pc.user("foo2"))
|
||||
self.assertEqual(b"+OK Login successful", pc.pass_("helloworld"))
|
||||
_, eml, oc = pc.retr(1)
|
||||
self.assertIn(b"Previous line just has a dot", eml)
|
||||
self.assertIn(b".Line starts with a dot", eml)
|
||||
self.assertIn(b".", eml)
|
||||
finally:
|
||||
pc.quit()
|
||||
|
||||
await asyncio.to_thread(run_poplib)
|
||||
|
||||
async def asyncTearDown(self) -> None:
|
||||
logging.debug("at teardown")
|
||||
self.writer.close()
|
||||
@ -206,9 +238,9 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
|
||||
async def dialog_checker(self, dialog: str) -> None:
|
||||
await self.dialog_checker_impl(self.reader, self.writer, dialog)
|
||||
|
||||
async def dialog_checker_impl(self, reader: asyncio.StreamReader,
|
||||
writer: asyncio.StreamWriter,
|
||||
dialog: str) -> None:
|
||||
async def dialog_checker_impl(
|
||||
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, dialog: str
|
||||
) -> None:
|
||||
for line in dialog.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
@ -222,5 +254,5 @@ class TestPop3(unittest.IsolatedAsyncioTestCase):
|
||||
self.assertEqual(data, resp)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
@ -10,27 +10,31 @@ class TestPWHash(unittest.TestCase):
|
||||
pwinfo = parse_hash(pwhash)
|
||||
self.assertEqual(len(pwinfo.salt), SALT_LEN)
|
||||
self.assertEqual(len(pwinfo.scrypt_hash), KEY_LEN)
|
||||
self.assertTrue(check_pass(password, pwinfo),
|
||||
"check pass with correct password")
|
||||
self.assertFalse(check_pass("foobar", pwinfo),
|
||||
"check pass with wrong password")
|
||||
self.assertTrue(
|
||||
check_pass(password, pwinfo), "check pass with correct password"
|
||||
)
|
||||
self.assertFalse(check_pass("foobar", pwinfo), "check pass with wrong password")
|
||||
|
||||
def test_hardcoded_hash(self):
|
||||
test_hash = "".join(c for c in """
|
||||
test_hash = "".join(
|
||||
c
|
||||
for c in """
|
||||
AFTY5EVN7AX47ZL7UMH3BETYWFBTAV3XHR73CEFAJBPN2NIHPWD
|
||||
ZHV2UQSMSPHSQQ2A2BFQBNC77VL7F2UKATQNJZGYLCSU6C43UQD
|
||||
AQXWXSWNGAEPGIMG2F3QDKBXL3MRHY6K2BPID64ZR6LABLPVSF
|
||||
""" if not c.isspace())
|
||||
"""
|
||||
if not c.isspace()
|
||||
)
|
||||
pwinfo = parse_hash(test_hash)
|
||||
self.assertTrue(check_pass("helloworld", pwinfo),
|
||||
"check pass with correct password")
|
||||
self.assertFalse(check_pass("foobar", pwinfo),
|
||||
"check pass with wrong password")
|
||||
self.assertTrue(
|
||||
check_pass("helloworld", pwinfo), "check pass with correct password"
|
||||
)
|
||||
self.assertFalse(check_pass("foobar", pwinfo), "check pass with wrong password")
|
||||
|
||||
def test_invalid_hash(self):
|
||||
with self.assertRaises(Exception):
|
||||
parse_hash("sdlfkjdsklfjdsk")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
@ -10,7 +10,7 @@ from pathlib import Path
|
||||
|
||||
from mail4one.smtp import create_smtp_server
|
||||
|
||||
TEST_MBOX = 'foobar_mails'
|
||||
TEST_MBOX = "foobar_mails"
|
||||
MAILS_PATH: Path
|
||||
|
||||
|
||||
@ -21,7 +21,7 @@ def setUpModule() -> None:
|
||||
unittest.addModuleCleanup(td.cleanup)
|
||||
MAILS_PATH = Path(td.name)
|
||||
os.mkdir(MAILS_PATH / TEST_MBOX)
|
||||
for md in ('new', 'cur', 'tmp'):
|
||||
for md in ("new", "cur", "tmp"):
|
||||
os.mkdir(MAILS_PATH / TEST_MBOX / md)
|
||||
|
||||
|
||||
@ -32,7 +32,8 @@ class TestSMTP(unittest.IsolatedAsyncioTestCase):
|
||||
host="127.0.0.1",
|
||||
port=7996,
|
||||
mails_path=MAILS_PATH,
|
||||
mbox_finder=lambda addr: [TEST_MBOX])
|
||||
mbox_finder=lambda addr: [TEST_MBOX],
|
||||
)
|
||||
self.task = asyncio.create_task(smtp_server.serve_forever())
|
||||
|
||||
async def test_send_mail(self) -> None:
|
||||
@ -45,8 +46,9 @@ class TestSMTP(unittest.IsolatedAsyncioTestCase):
|
||||
msg = b"".join(l.strip() + b"\r\n" for l in msg.splitlines())
|
||||
|
||||
def send_mail():
|
||||
with contextlib.closing(smtplib.SMTP(host="127.0.0.1",
|
||||
port=7996)) as client:
|
||||
with contextlib.closing(
|
||||
smtplib.SMTP(host="127.0.0.1", port=7996)
|
||||
) as client:
|
||||
client.sendmail("foo@sender.com", "foo@bar.com", msg)
|
||||
_, local_port = client.sock.getsockname()
|
||||
return local_port
|
||||
@ -62,7 +64,7 @@ class TestSMTP(unittest.IsolatedAsyncioTestCase):
|
||||
Byee
|
||||
"""
|
||||
expected = "".join(l.strip() + "\r\n" for l in expected.splitlines())
|
||||
mails = list((MAILS_PATH / TEST_MBOX / 'new').glob("*"))
|
||||
mails = list((MAILS_PATH / TEST_MBOX / "new").glob("*"))
|
||||
self.assertEqual(len(mails), 1)
|
||||
self.assertEqual(mails[0].read_bytes(), expected.encode())
|
||||
|
||||
|
Reference in New Issue
Block a user