pEp MixMailer prototype. See aditional documentation at https://mixmailer_docs.codeberg.page/ (source: https://gitea.pep.foundation/pEp.foundation/mixmailer_docs). See slides at gitea.pep.foundation/pEp.foundation/mixmailer_slides
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

182 lines
5.2 KiB

"""Common functions to remailer and client."""
import email
import getpass
import logging
import os
import random
import re
import smtplib
import sys
import urllib3
from mixnet import defaults, exceptions, settings
logger = logging.getLogger(__name__)
def guess_home():
"""Return the home of the user that is running the script."""
user = os.path.join("~", getpass.getuser())
home = os.environ.get("HOME", os.path.expanduser(user))
# When this is run by postfix pipe, the output has to be in stderr, to be
# shown in the log.
print("User home", home, file=sys.stderr)
logger.debug("User home: {}".format(home))
return home
def set_gnupg_home(gnupg_home):
"""Create gnupg directory and set GNUPGHOME env variable.
To be able to have different keyrings for the tests.
Since pEp-2.1.0rc2, it does not use gnupg and it's enough to set the home,
to have different keys.db per test.
"""
os.makedirs(gnupg_home, mode=0o0700, exist_ok=True)
os.environ["GNUPGHOME"] = gnupg_home
print("GNUPGHOME: %s", gnupg_home)
def set_env(home=None, create_dir=False, create_gnupg_dir=False):
"""Set HOME environment variable to set where pEp database will live.
And optionally the GNUPGPHOME. Since pEp-2.1.0rc2, gnupg is not used
anymore.
"""
if not home:
home = guess_home()
if create_dir:
os.makedirs(home, mode=0o0700, exist_ok=True)
os.environ["HOME"] = home
if create_gnupg_dir:
set_gnupg_home(os.path.join(home, ".gnupg"))
return home
def log_path(home, create_dir=True):
file_path = os.path.join(home, defaults.LOG_PATH)
if create_dir:
print("Creating path ", file_path)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
return file_path
def import_keys(keyring_path_list):
import pEp
for keyring_path in keyring_path_list:
keyring_data = keyring_path.read()
try:
pEp.import_key(keyring_data)
except RuntimeError as e:
logger.debug(e)
# Because the key was already in the keyring?
# With pEp-2.1.0rc2, when the keyring_data is not a key, it raises
# PEP_UNKNOWN_ERROR
if str(e) == "PEP_NO_KEY_IMPORTED" or "PEP_UNKNOWN_ERROR":
raise exceptions.PepNoKeyImported
raise (e)
logger.info(
"Imported key %s",
getattr("keyring_path", "name", None) or keyring_path,
)
def set_identity(name_addr, fpr=""):
import pEp
name, address = email.utils.parseaddr(name_addr)
logger.debug("name: %s, address: %s, fp: %s", name, address, fpr)
if not re.match(defaults.BASIC_EMAIL_RE, address):
raise exceptions.NoEmailAddress
# XXX: because set_own_key would complain when name is not assigned here
# or username later:
# with `ValueError: username needed )`, add one by default
name = name or "root"
ident = pEp.Identity(address, name, name_addr, fpr, 0, "")
logger.info("Created identity %s", name_addr)
# XXX: RuntimeError: username must be at least 5 characters
# ident.username = name
ident.user_id = name_addr
return ident
def set_my_identity(name_addr, fpr=""):
import pEp
me = set_identity(name_addr, fpr)
if fpr:
logger.info("Setting my own identity to: %s, %s.", me, fpr)
try:
pEp.set_own_key(me, fpr)
except RuntimeError as e:
# With pEp-2.1.0rc2, if the key has not been imported, it raises
# PEP_KEY_UNSUITABLE
if str(e) == "PEP_KEY_NOT_FOUND" or "PEP_KEY_UNSUITABLE":
raise exceptions.PepKeyNotFound
raise (e)
# XXX: why is changing user_id?
logger.info("Set my own identity to: %s, %s", me, fpr)
else:
logger.info("Setting my own identity to: %s.", me)
pEp.myself(me)
logger.info("Set my own identity to: %s", me)
return me
def create_incoming_msg(mime_text):
import pEp
msg = pEp.Message(mime_text)
msg.dir = defaults.INCOMING_MSG
return msg
def create_outgoing_msg(me, to, subject, body):
import pEp
msg = pEp.Message(defaults.OUTGOING_MSG, me)
msg.to = [to]
msg.shortmsg = subject
msg.longmsg = body
return msg
def send_email(mail_str, host="localhost", port=25):
mail = email.message_from_string(mail_str)
logger.info(
"Sending email from: %s, to: %s, via %s:%s",
mail["From"],
mail["To"],
host,
port,
)
s = smtplib.SMTP(host, port)
s.send_message(mail)
s.quit()
def parse_host_port(host_port):
u = urllib3.util.url.parse_url("//" + host_port)
return (u.host or "localhost", u.port or 25)
def dummy_delay(
min_secs=defaults.MIN_DELAY_SECS, max_secs=defaults.MAX_DELAY_SECS
):
"""Return a random float number of seconds between min_secs and max_secs.
.. warning:: this is a proof of concept for the mixnet "mixing".
It does not take into account previous delays nor use a distribution
function for it.
Also, when it's run in a ``postfix`` pipe, it'll timeout processing the
Email.
"""
if settings.TEST_NET:
max_secs = 3
return random.uniform(min_secs, max_secs)