power-mailinabox/management/mailconfig.py

#!/usr/bin/python3

import subprocess, shutil, os, sqlite3, re
import utils

def validate_email(email, strict):
	# There are a lot of characters permitted in email addresses, but
	# Dovecot's sqlite driver seems to get confused if there are any
	# unusual characters in the address. Bah. Also note that since
	# the mailbox path name is based on the email address, the address
	# shouldn't be absurdly long and must not have a forward slash.

	if len(email) > 255: return False

	if strict:
		# For Dovecot's benefit, only allow basic characters.
		ATEXT = r'[\w\-]'
	else:
		# Based on RFC 2822 and https://github.com/SyrusAkbary/validate_email/blob/master/validate_email.py,
		# these characters are permitted in email address.
		ATEXT = r'[\w!#$%&\'\*\+\-/=\?\^`\{\|\}~]' # see 3.2.4

	DOT_ATOM_TEXT = ATEXT + r'+(?:\.' + ATEXT + r'+)*'      # see 3.2.4
	DOT_ATOM_TEXT2 = ATEXT + r'+(?:\.' + ATEXT + r'+)+'      # as above, but with a "+" since the host part must be under some TLD
	ADDR_SPEC = '^%s@%s$' % (DOT_ATOM_TEXT, DOT_ATOM_TEXT2) # see 3.4.1

	return re.match(ADDR_SPEC, email)

def open_database(env, with_connection=False):
	conn = sqlite3.connect(env["STORAGE_ROOT"] + "/mail/users.sqlite")
	if not with_connection:
		return conn.cursor()
	else:
		return conn, conn.cursor()

def get_mail_users(env):
	c = open_database(env)
	c.execute('SELECT email FROM users')
	return [row[0] for row in c.fetchall()]

def get_mail_aliases(env):
	c = open_database(env)
	c.execute('SELECT source, destination FROM aliases')
	return [(row[0], row[1]) for row in c.fetchall()]

def get_mail_domains(env):
	def get_domain(emailaddr):
		return emailaddr.split('@', 1)[1]
	return set([get_domain(addr) for addr in get_mail_users(env)] + [get_domain(addr1) for addr1, addr2 in get_mail_aliases(env)])

def add_mail_user(email, pw, env):
	if not validate_email(email, True):
		return ("Invalid email address.", 400)

	# get the database
	conn, c = open_database(env, with_connection=True)

	# hash the password
	pw = utils.shell('check_output', ["/usr/bin/doveadm", "pw", "-s", "SHA512-CRYPT", "-p", pw]).strip()

	# add the user to the database
	try:
		c.execute("INSERT INTO users (email, password) VALUES (?, ?)", (email, pw))
	except sqlite3.IntegrityError:
		return ("User already exists.", 400)
		
	# write databasebefore next step
	conn.commit()

	# Create the user's INBOX and Spam folders and subscribe them.

	# Check if the mailboxes exist before creating them. When creating a user that had previously
	# been deleted, the mailboxes will still exist because they are still on disk.
	try:
		existing_mboxes = utils.shell('check_output', ["doveadm", "mailbox", "list", "-u", email, "-8"], capture_stderr=True).split("\n")
	except subprocess.CalledProcessError as e:
		c.execute("DELETE FROM users WHERE email=?", (email,))
		conn.commit()
		return ("Failed to initialize the user: " + e.output.decode("utf8"), 400)

	if "INBOX" not in existing_mboxes: utils.shell('check_call', ["doveadm", "mailbox", "create", "-u", email, "-s", "INBOX"])
	if "Spam" not in existing_mboxes: utils.shell('check_call', ["doveadm", "mailbox", "create", "-u", email, "-s", "Spam"])

	# Create the user's sieve script to move spam into the Spam folder, and make it owned by mail.
	maildirstat = os.stat(env["STORAGE_ROOT"] + "/mail/mailboxes")
	(em_user, em_domain) = email.split("@", 1)
	user_mail_dir = env["STORAGE_ROOT"] + ("/mail/mailboxes/%s/%s" % (em_domain, em_user))
	if not os.path.exists(user_mail_dir):
		os.makedirs(user_mail_dir)
		os.chown(user_mail_dir, maildirstat.st_uid, maildirstat.st_gid)
	shutil.copyfile(utils.CONF_DIR + "/dovecot_sieve.txt", user_mail_dir + "/.dovecot.sieve")
	os.chown(user_mail_dir + "/.dovecot.sieve", maildirstat.st_uid, maildirstat.st_gid)

	# Update DNS/web in case any new domains are added.
	return kick(env, "mail user added")

def set_mail_password(email, pw, env):
	# hash the password
	pw = utils.shell('check_output', ["/usr/bin/doveadm", "pw", "-s", "SHA512-CRYPT", "-p", pw]).strip()

	# update the database
	conn, c = open_database(env, with_connection=True)
	c.execute("UPDATE users SET password=? WHERE email=?", (pw, email))
	if c.rowcount != 1:
		return ("That's not a user (%s)." % email, 400)
	conn.commit()
	return "OK"

def remove_mail_user(email, env):
	conn, c = open_database(env, with_connection=True)
	c.execute("DELETE FROM users WHERE email=?", (email,))
	if c.rowcount != 1:
		return ("That's not a user (%s)." % email, 400)
	conn.commit()

	# Update DNS/web in case any domains are removed.
	return kick(env, "mail user removed")

def add_mail_alias(source, destination, env):
	if not validate_email(source, False):
		return ("Invalid email address.", 400)

	conn, c = open_database(env, with_connection=True)
	try:
		c.execute("INSERT INTO aliases (source, destination) VALUES (?, ?)", (source, destination))
	except sqlite3.IntegrityError:
		return ("Alias already exists (%s)." % source, 400)
	conn.commit()

	# Update DNS/web in case any new domains are added.
	return kick(env, "alias added")

def remove_mail_alias(source, env):
	conn, c = open_database(env, with_connection=True)
	c.execute("DELETE FROM aliases WHERE source=?", (source,))
	if c.rowcount != 1:
		return ("That's not an alias (%s)." % source, 400)
	conn.commit()

	# Update DNS and nginx in case any domains are removed.
	return kick(env, "alias removed")

def kick(env, mail_result):
	# Update DNS and nginx in case any domains are added/removed.
	from dns_update import do_dns_update
	from web_update import do_web_update
	results = [
		do_dns_update(env),
		mail_result + "\n",
		do_web_update(env),
	]
	return "".join(s for s in results if s != "")

if __name__ == "__main__":
	import sys
	if len(sys.argv) > 2 and sys.argv[1] == "validate-email":
		# Validate that we can create a Dovecot account for a given string.
		if validate_email(sys.argv[2], True):
			sys.exit(0)
		else:
			sys.exit(1)